feat(NODE-6136): parse cursor responses on demand

Author: nbbeeken

src/bson.ts

@@ -27,6 +27,7 @@ export {
   UUID
 } from 'bson';
 
+/** @internal */
 export type BSONElement = BSON.OnDemand['BSONElement'];
 
 export function parseToElementsToArray(bytes: Uint8Array, offset?: number): BSONElement[] {

src/cmap/wire_protocol/responses.ts

@@ -10,7 +10,7 @@ import {
 } from '../../bson';
 import { MongoUnexpectedServerResponseError } from '../../error';
 import { type ClusterTime } from '../../sdam/common';
-import { type MongoDBNamespace, ns } from '../../utils';
+import { ns } from '../../utils';
 import { OnDemandDocument } from './on_demand/document';
 
 // eslint-disable-next-line no-restricted-syntax
@@ -171,13 +171,29 @@ export class MongoDBResponse extends OnDemandDocument {
   }
 }
 
+// Here's a litle blast from the past.
+// OLD style method definition so that I can override get without redefining ALL the fancy TS :/
+Object.defineProperty(MongoDBResponse.prototype, 'get', {
+  value: function get(name: any, as: any, required: any) {
+    try {
+      return OnDemandDocument.prototype.get.call(this, name, as, required);
+    } catch (cause) {
+      throw new MongoUnexpectedServerResponseError(cause.message, { cause });
+    }
+  }
+});
+
 /** @internal */
 export class CursorResponse extends MongoDBResponse {
   /**
    * This supports a feature of the FindCursor.
    * It is an optimization to avoid an extra getMore when the limit has been reached
    */
-  static emptyGetMore = { id: new Long(0), length: 0, shift: () => null };
+  static emptyGetMore: CursorResponse = {
+    id: new Long(0),
+    length: 0,
+    shift: () => null
+  } as unknown as CursorResponse;
 
   static override is(value: unknown): value is CursorResponse {
     return value instanceof CursorResponse || value === CursorResponse.emptyGetMore;
@@ -186,25 +202,25 @@ export class CursorResponse extends MongoDBResponse {
   private _batch: OnDemandDocument | null = null;
   private iterated = 0;
 
-  get cursor() {
+  private get cursor() {
     return this.get('cursor', BSONType.object, true);
   }
 
-  get id(): Long {
+  public get id(): Long {
     return Long.fromBigInt(this.cursor.get('id', BSONType.long, true));
   }
 
-  get ns() {
+  public get ns() {
     const namespace = this.cursor.get('ns', BSONType.string);
     if (namespace != null) return ns(namespace);
     return null;
   }
 
-  get length() {
+  public get length() {
     return Math.max(this.batchSize - this.iterated, 0);
   }
 
-  get batch() {
+  private get batch() {
     if (this._batch != null) return this._batch;
     const cursor = this.cursor;
     if (cursor.has('firstBatch')) this._batch = cursor.get('firstBatch', BSONType.array, true);
@@ -213,11 +229,21 @@ export class CursorResponse extends MongoDBResponse {
     return this._batch;
   }
 
-  get batchSize() {
+  public get batchSize() {
     return this.batch?.size();
   }
 
-  shift(options?: BSONSerializeOptions): any {
+  public get postBatchResumeToken() {
+    return (
+      this.cursor.get('postBatchResumeToken', BSONType.object)?.toObject({
+        promoteValues: false,
+        promoteLongs: false,
+        promoteBuffers: false
+      }) ?? null
+    );
+  }
+
+  public shift(options?: BSONSerializeOptions): any {
     if (this.iterated >= this.batchSize) {
       return null;
     }
@@ -232,15 +258,40 @@ export class CursorResponse extends MongoDBResponse {
     }
   }
 
-  clear() {
+  public clear() {
     this.iterated = this.batchSize;
   }
+}
+
+/**
+ * Explain responses have nothing to do with cursor responses
+ * This class serves to temporarily avoid refactoring how cursors handle
+ * explain responses which is to detect that the response is not cursor-like and return the explain
+ * result as the "first and only" document in the "batch" and end the "cursor"
+ */
+export class ExplainedCursorResponse extends CursorResponse {
+  isExplain = true;
+
+  override get id(): Long {
+    return Long.fromBigInt(0n);
+  }
+
+  override get batchSize() {
+    return 0;
+  }
+
+  override get ns() {
+    return null;
+  }
 
-  pushMany() {
-    throw new Error('pushMany Unsupported method');
+  _length = 1;
+  override get length(): number {
+    return this._length;
   }
 
-  push() {
-    throw new Error('push Unsupported method');
+  override shift(options?: BSONSerializeOptions | undefined) {
+    if (this._length === 0) return null;
+    this._length -= 1;
+    return this.toObject(options);
   }
 }

src/cursor/abstract_cursor.ts

@@ -13,15 +13,15 @@ import {
   MongoTailableCursorError
 } from '../error';
 import type { MongoClient } from '../mongo_client';
-import { type TODO_NODE_3286, TypedEventEmitter } from '../mongo_types';
+import { TypedEventEmitter } from '../mongo_types';
 import { executeOperation, type ExecutionResult } from '../operations/execute_operation';
 import { GetMoreOperation } from '../operations/get_more';
 import { KillCursorsOperation } from '../operations/kill_cursors';
 import { ReadConcern, type ReadConcernLike } from '../read_concern';
 import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
 import type { Server } from '../sdam/server';
 import { ClientSession, maybeClearPinnedConnection } from '../sessions';
-import { List, type MongoDBNamespace, ns, squashError } from '../utils';
+import { type MongoDBNamespace, squashError } from '../utils';
 
 /** @internal */
 const kId = Symbol('id');
@@ -145,13 +145,7 @@ export abstract class AbstractCursor<
   /** @internal */
   [kNamespace]: MongoDBNamespace;
   /** @internal */
-  [kDocuments]: {
-    length: number;
-    shift(bsonOptions?: any): TSchema | null;
-    clear(): void;
-    pushMany(many: Iterable<TSchema>): void;
-    push(item: TSchema): void;
-  };
+  [kDocuments]: CursorResponse = { length: 0 } as unknown as CursorResponse;
   /** @internal */
   [kClient]: MongoClient;
   /** @internal */
@@ -182,7 +176,6 @@ export abstract class AbstractCursor<
     this[kClient] = client;
     this[kNamespace] = namespace;
     this[kId] = null;
-    this[kDocuments] = new List();
     this[kInitialized] = false;
     this[kClosed] = false;
     this[kKilled] = false;
@@ -637,13 +630,12 @@ export abstract class AbstractCursor<
   protected abstract _initialize(session: ClientSession | undefined): Promise<ExecutionResult>;
 
   /** @internal */
-  async getMore(batchSize: number, useCursorResponse = false): Promise<Document | null> {
+  async getMore(batchSize: number): Promise<CursorResponse> {
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
     const getMoreOperation = new GetMoreOperation(this[kNamespace], this[kId]!, this[kServer]!, {
       ...this[kOptions],
       session: this[kSession],
-      batchSize,
-      useCursorResponse
+      batchSize
     });
 
     return await executeOperation(this[kClient], getMoreOperation);
@@ -661,37 +653,13 @@ export abstract class AbstractCursor<
       const state = await this._initialize(this[kSession]);
       const response = state.response;
       this[kServer] = state.server;
-      if (CursorResponse.is(response)) {
-        this[kId] = response.id;
-        if (response.ns) this[kNamespace] = response.ns;
-        this[kDocuments] = response;
-      } else if (response.cursor) {
-        // TODO(NODE-2674): Preserve int64 sent from MongoDB
-        this[kId] =
-          typeof response.cursor.id === 'number'
-            ? Long.fromNumber(response.cursor.id)
-            : typeof response.cursor.id === 'bigint'
-            ? Long.fromBigInt(response.cursor.id)
-            : response.cursor.id;
-
-        if (response.cursor.ns) {
-          this[kNamespace] = ns(response.cursor.ns);
-        }
-
-        this[kDocuments].pushMany(response.cursor.firstBatch);
-      }
 
-      // When server responses return without a cursor document, we close this cursor
-      // and return the raw server response. This is often the case for explain commands
-      // for example
-      if (this[kId] == null) {
-        this[kId] = Long.ZERO;
-        // TODO(NODE-3286): ExecutionResult needs to accept a generic parameter
-        this[kDocuments].push(state.response as TODO_NODE_3286);
-      }
+      if (!CursorResponse.is(response)) throw new Error('ah');
 
-      // the cursor is now initialized, even if it is dead
-      this[kInitialized] = true;
+      this[kId] = response.id;
+      this[kNamespace] = response.ns ?? this[kNamespace];
+      this[kDocuments] = response;
+      this[kInitialized] = true; // the cursor is now initialized, even if it is dead
     } catch (error) {
       // the cursor is now initialized, even if an error occurred
       this[kInitialized] = true;
@@ -802,20 +770,8 @@ async function next<T>(
 
     try {
       const response = await cursor.getMore(batchSize);
-      if (CursorResponse.is(response)) {
-        cursor[kId] = response.id;
-        cursor[kDocuments] = response;
-      } else if (response) {
-        const cursorId =
-          typeof response.cursor.id === 'number'
-            ? Long.fromNumber(response.cursor.id)
-            : typeof response.cursor.id === 'bigint'
-            ? Long.fromBigInt(response.cursor.id)
-            : response.cursor.id;
-
-        cursor[kDocuments].pushMany(response.cursor.nextBatch);
-        cursor[kId] = cursorId;
-      }
+      cursor[kId] = response.id;
+      cursor[kDocuments] = response;
     } catch (error) {
       try {
         await cleanupCursor(cursor, { error, needsToEmitClosed: true });

src/cursor/aggregation_cursor.ts

@@ -76,14 +76,16 @@ export class AggregationCursor<TSchema = any> extends AbstractCursor<TSchema> {
 
   /** Execute the explain for the cursor */
   async explain(verbosity?: ExplainVerbosityLike): Promise<Document> {
-    return await executeOperation(
-      this.client,
-      new AggregateOperation(this.namespace, this[kPipeline], {
-        ...this[kOptions], // NOTE: order matters here, we may need to refine this
-        ...this.cursorOptions,
-        explain: verbosity ?? true
-      })
-    );
+    return (
+      await executeOperation(
+        this.client,
+        new AggregateOperation(this.namespace, this[kPipeline], {
+          ...this[kOptions], // NOTE: order matters here, we may need to refine this
+          ...this.cursorOptions,
+          explain: verbosity ?? true
+        })
+      )
+    ).shift(this[kOptions]);
   }
 
   /** Add a stage to the aggregation pipeline

src/cursor/change_stream_cursor.ts

@@ -1,14 +1,14 @@
-import type { Document, Long, Timestamp } from '../bson';
+import type { Document } from '../bson';
 import {
   ChangeStream,
   type ChangeStreamDocument,
   type ChangeStreamEvents,
   type OperationTime,
   type ResumeToken
 } from '../change_stream';
+import { type CursorResponse } from '../cmap/wire_protocol/responses';
 import { INIT, RESPONSE } from '../constants';
 import type { MongoClient } from '../mongo_client';
-import type { TODO_NODE_3286 } from '../mongo_types';
 import { AggregateOperation } from '../operations/aggregate';
 import type { CollationOptions } from '../operations/command';
 import { executeOperation, type ExecutionResult } from '../operations/execute_operation';
@@ -26,25 +26,13 @@ export interface ChangeStreamCursorOptions extends AbstractCursorOptions {
   fullDocument?: string;
 }
 
-/** @internal */
-export type ChangeStreamAggregateRawResult<TChange> = {
-  $clusterTime: { clusterTime: Timestamp };
-  cursor: {
-    postBatchResumeToken: ResumeToken;
-    ns: string;
-    id: number | Long;
-  } & ({ firstBatch: TChange[] } | { nextBatch: TChange[] });
-  ok: 1;
-  operationTime: Timestamp;
-};
-
 /** @internal */
 export class ChangeStreamCursor<
   TSchema extends Document = Document,
   TChange extends Document = ChangeStreamDocument<TSchema>
 > extends AbstractCursor<TChange, ChangeStreamEvents> {
   _resumeToken: ResumeToken;
-  startAtOperationTime?: OperationTime;
+  startAtOperationTime: OperationTime | null = null;
   hasReceived?: boolean;
   resumeAfter: ResumeToken;
   startAfter: ResumeToken;
@@ -71,7 +59,7 @@ export class ChangeStreamCursor<
     this.pipeline = pipeline;
     this.options = options;
     this._resumeToken = null;
-    this.startAtOperationTime = options.startAtOperationTime;
+    this.startAtOperationTime = options.startAtOperationTime ?? null;
 
     if (options.startAfter) {
       this.resumeToken = options.startAfter;
@@ -120,15 +108,13 @@ export class ChangeStreamCursor<
     this.hasReceived = true;
   }
 
-  _processBatch(response: ChangeStreamAggregateRawResult<TChange>): void {
-    const cursor = response.cursor;
-    if (cursor.postBatchResumeToken) {
-      this.postBatchResumeToken = response.cursor.postBatchResumeToken;
+  _processBatch(response: CursorResponse): void {
+    const { postBatchResumeToken } = response;
+    if (postBatchResumeToken) {
+      this.postBatchResumeToken = postBatchResumeToken;
 
-      const batch =
-        'firstBatch' in response.cursor ? response.cursor.firstBatch : response.cursor.nextBatch;
-      if (batch.length === 0) {
-        this.resumeToken = cursor.postBatchResumeToken;
+      if (response.batchSize === 0) {
+        this.resumeToken = postBatchResumeToken;
       }
     }
   }
@@ -146,10 +132,7 @@ export class ChangeStreamCursor<
       session
     });
 
-    const response = await executeOperation<
-      TODO_NODE_3286,
-      ChangeStreamAggregateRawResult<TChange>
-    >(session.client, aggregateOperation);
+    const response = await executeOperation(session.client, aggregateOperation);
 
     const server = aggregateOperation.server;
     this.maxWireVersion = maxWireVersion(server);
@@ -172,11 +155,12 @@ export class ChangeStreamCursor<
     return { server, session, response };
   }
 
-  override async getMore(batchSize: number): Promise<Document | null> {
+  override async getMore(batchSize: number): Promise<CursorResponse> {
     const response = await super.getMore(batchSize);
 
     this.maxWireVersion = maxWireVersion(this.server);
-    this._processBatch(response as ChangeStreamAggregateRawResult<TChange>);
+    // as ChangeStreamAggregateRawResult<TChange>
+    this._processBatch(response);
 
     this.emit(ChangeStream.MORE, response);
     this.emit(ChangeStream.RESPONSE);