opusdev/vector-similarity-api

1import type { Readable } from 'stream';2 3import type { Binary, Document, Timestamp } from './bson';4import { Collection } from './collection';5import { CHANGE, CLOSE, END, ERROR, INIT, MORE, RESPONSE, RESUME_TOKEN_CHANGED } from './constants';6import { type CursorStreamOptions, CursorTimeoutContext } from './cursor/abstract_cursor';7import { ChangeStreamCursor, type ChangeStreamCursorOptions } from './cursor/change_stream_cursor';8import { Db } from './db';9import {10  type AnyError,11  isResumableError,12  MongoAPIError,13  MongoChangeStreamError,14  MongoOperationTimeoutError,15  MongoRuntimeError16} from './error';17import { MongoClient } from './mongo_client';18import { type InferIdType, TypedEventEmitter } from './mongo_types';19import type { AggregateOptions } from './operations/aggregate';20import type { CollationOptions, OperationParent } from './operations/command';21import type { ReadPreference } from './read_preference';22import { type AsyncDisposable, configureResourceManagement } from './resource_management';23import type { ServerSessionId } from './sessions';24import { CSOTTimeoutContext, type TimeoutContext } from './timeout';25import { filterOptions, getTopology, type MongoDBNamespace, squashError } from './utils';26 27const CHANGE_STREAM_OPTIONS = [28  'resumeAfter',29  'startAfter',30  'startAtOperationTime',31  'fullDocument',32  'fullDocumentBeforeChange',33  'showExpandedEvents'34] as const;35 36const CHANGE_DOMAIN_TYPES = {37  COLLECTION: Symbol('Collection'),38  DATABASE: Symbol('Database'),39  CLUSTER: Symbol('Cluster')40};41 42const CHANGE_STREAM_EVENTS = [RESUME_TOKEN_CHANGED, END, CLOSE] as const;43 44const NO_RESUME_TOKEN_ERROR =45  'A change stream document has been received that lacks a resume token (_id).';46const CHANGESTREAM_CLOSED_ERROR = 'ChangeStream is closed';47 48/**49 * @public50 * @deprecated Please use the ChangeStreamCursorOptions type instead.51 */52export interface ResumeOptions {53  startAtOperationTime?: Timestamp;54  batchSize?: number;55  maxAwaitTimeMS?: number;56  collation?: CollationOptions;57  readPreference?: ReadPreference;58  resumeAfter?: ResumeToken;59  startAfter?: ResumeToken;60  fullDocument?: string;61}62 63/**64 * Represents the logical starting point for a new ChangeStream or resuming a ChangeStream on the server.65 * @see https://www.mongodb.com/docs/manual/changeStreams/#std-label-change-stream-resume66 * @public67 */68export type ResumeToken = unknown;69 70/**71 * Represents a specific point in time on a server. Can be retrieved by using `db.command()`72 * @public73 * @see https://www.mongodb.com/docs/manual/reference/method/db.runCommand/#response74 */75export type OperationTime = Timestamp;76 77/**78 * Options that can be passed to a ChangeStream. Note that startAfter, resumeAfter, and startAtOperationTime are all mutually exclusive, and the server will error if more than one is specified.79 * @public80 */81export interface ChangeStreamOptions extends Omit<AggregateOptions, 'writeConcern'> {82  /**83   * Allowed values: 'updateLookup', 'whenAvailable', 'required'.84   *85   * When set to 'updateLookup', the change notification for partial updates86   * will include both a delta describing the changes to the document as well87   * as a copy of the entire document that was changed from some time after88   * the change occurred.89   *90   * When set to 'whenAvailable', configures the change stream to return the91   * post-image of the modified document for replace and update change events92   * if the post-image for this event is available.93   *94   * When set to 'required', the same behavior as 'whenAvailable' except that95   * an error is raised if the post-image is not available.96   */97  fullDocument?: string;98 99  /**100   * Allowed values: 'whenAvailable', 'required', 'off'.101   *102   * The default is to not send a value, which is equivalent to 'off'.103   *104   * When set to 'whenAvailable', configures the change stream to return the105   * pre-image of the modified document for replace, update, and delete change106   * events if it is available.107   *108   * When set to 'required', the same behavior as 'whenAvailable' except that109   * an error is raised if the pre-image is not available.110   */111  fullDocumentBeforeChange?: string;112  /** The maximum amount of time for the server to wait on new documents to satisfy a change stream query. */113  maxAwaitTimeMS?: number;114  /**115   * Allows you to start a changeStream after a specified event.116   * @see https://www.mongodb.com/docs/manual/changeStreams/#resumeafter-for-change-streams117   */118  resumeAfter?: ResumeToken;119  /**120   * Similar to resumeAfter, but will allow you to start after an invalidated event.121   * @see https://www.mongodb.com/docs/manual/changeStreams/#startafter-for-change-streams122   */123  startAfter?: ResumeToken;124  /** Will start the changeStream after the specified operationTime. */125  startAtOperationTime?: OperationTime;126  /**127   * The number of documents to return per batch.128   * @see https://www.mongodb.com/docs/manual/reference/command/aggregate129   */130  batchSize?: number;131 132  /**133   * When enabled, configures the change stream to include extra change events.134   *135   * - createIndexes136   * - dropIndexes137   * - modify138   * - create139   * - shardCollection140   * - reshardCollection141   * - refineCollectionShardKey142   */143  showExpandedEvents?: boolean;144}145 146/** @public */147export interface ChangeStreamNameSpace {148  db: string;149  coll: string;150}151 152/** @public */153export interface ChangeStreamDocumentKey<TSchema extends Document = Document> {154  /**155   * For unsharded collections this contains a single field `_id`.156   * For sharded collections, this will contain all the components of the shard key157   */158  documentKey: { _id: InferIdType<TSchema>; [shardKey: string]: any };159}160 161/** @public */162export interface ChangeStreamSplitEvent {163  /** Which fragment of the change this is. */164  fragment: number;165  /** The total number of fragments. */166  of: number;167}168 169/** @public */170export interface ChangeStreamDocumentCommon {171  /**172   * The id functions as an opaque token for use when resuming an interrupted173   * change stream.174   */175  _id: ResumeToken;176  /**177   * The timestamp from the oplog entry associated with the event.178   * For events that happened as part of a multi-document transaction, the associated change stream179   * notifications will have the same clusterTime value, namely the time when the transaction was committed.180   * On a sharded cluster, events that occur on different shards can have the same clusterTime but be181   * associated with different transactions or even not be associated with any transaction.182   * To identify events for a single transaction, you can use the combination of lsid and txnNumber in the change stream event document.183   */184  clusterTime?: Timestamp;185 186  /**187   * The transaction number.188   * Only present if the operation is part of a multi-document transaction.189   *190   * **NOTE:** txnNumber can be a Long if promoteLongs is set to false191   */192  txnNumber?: number;193 194  /**195   * The identifier for the session associated with the transaction.196   * Only present if the operation is part of a multi-document transaction.197   */198  lsid?: ServerSessionId;199 200  /**201   * When the change stream's backing aggregation pipeline contains the $changeStreamSplitLargeEvent202   * stage, events larger than 16MB will be split into multiple events and contain the203   * following information about which fragment the current event is.204   */205  splitEvent?: ChangeStreamSplitEvent;206}207 208/** @public */209export interface ChangeStreamDocumentWallTime {210  /**211   * The server date and time of the database operation.212   * wallTime differs from clusterTime in that clusterTime is a timestamp taken from the oplog entry associated with the database operation event.213   * @sinceServerVersion 6.0.0214   */215  wallTime?: Date;216}217 218/** @public */219export interface ChangeStreamDocumentCollectionUUID {220  /**221   * The UUID (Binary subtype 4) of the collection that the operation was performed on.222   *223   * Only present when the `showExpandedEvents` flag is enabled.224   *225   * **NOTE:** collectionUUID will be converted to a NodeJS Buffer if the promoteBuffers226   *    flag is enabled.227   *228   * @sinceServerVersion 6.1.0229   */230  collectionUUID: Binary;231}232 233/** @public */234export interface ChangeStreamDocumentOperationDescription {235  /**236   * An description of the operation.237   *238   * Only present when the `showExpandedEvents` flag is enabled.239   *240   * @sinceServerVersion 6.1.0241   */242  operationDescription?: Document;243}244 245/**246 * @public247 * @see https://www.mongodb.com/docs/manual/reference/change-events/#insert-event248 */249export interface ChangeStreamInsertDocument<TSchema extends Document = Document>250  extends ChangeStreamDocumentCommon,251    ChangeStreamDocumentKey<TSchema>,252    ChangeStreamDocumentCollectionUUID,253    ChangeStreamDocumentWallTime {254  /** Describes the type of operation represented in this change notification */255  operationType: 'insert';256  /** This key will contain the document being inserted */257  fullDocument: TSchema;258  /** Namespace the insert event occurred on */259  ns: ChangeStreamNameSpace;260}261 262/**263 * @public264 * @see https://www.mongodb.com/docs/manual/reference/change-events/#update-event265 */266export interface ChangeStreamUpdateDocument<TSchema extends Document = Document>267  extends ChangeStreamDocumentCommon,268    ChangeStreamDocumentKey<TSchema>,269    ChangeStreamDocumentCollectionUUID,270    ChangeStreamDocumentWallTime {271  /** Describes the type of operation represented in this change notification */272  operationType: 'update';273  /**274   * This is only set if `fullDocument` is set to `'updateLookup'`275   * Contains the point-in-time post-image of the modified document if the276   * post-image is available and either 'required' or 'whenAvailable' was277   * specified for the 'fullDocument' option when creating the change stream.278   */279  fullDocument?: TSchema;280  /** Contains a description of updated and removed fields in this operation */281  updateDescription: UpdateDescription<TSchema>;282  /** Namespace the update event occurred on */283  ns: ChangeStreamNameSpace;284  /**285   * Contains the pre-image of the modified or deleted document if the286   * pre-image is available for the change event and either 'required' or287   * 'whenAvailable' was specified for the 'fullDocumentBeforeChange' option288   * when creating the change stream. If 'whenAvailable' was specified but the289   * pre-image is unavailable, this will be explicitly set to null.290   */291  fullDocumentBeforeChange?: TSchema;292}293 294/**295 * @public296 * @see https://www.mongodb.com/docs/manual/reference/change-events/#replace-event297 */298export interface ChangeStreamReplaceDocument<TSchema extends Document = Document>299  extends ChangeStreamDocumentCommon,300    ChangeStreamDocumentKey<TSchema>,301    ChangeStreamDocumentWallTime {302  /** Describes the type of operation represented in this change notification */303  operationType: 'replace';304  /** The fullDocument of a replace event represents the document after the insert of the replacement document */305  fullDocument: TSchema;306  /** Namespace the replace event occurred on */307  ns: ChangeStreamNameSpace;308  /**309   * Contains the pre-image of the modified or deleted document if the310   * pre-image is available for the change event and either 'required' or311   * 'whenAvailable' was specified for the 'fullDocumentBeforeChange' option312   * when creating the change stream. If 'whenAvailable' was specified but the313   * pre-image is unavailable, this will be explicitly set to null.314   */315  fullDocumentBeforeChange?: TSchema;316}317 318/**319 * @public320 * @see https://www.mongodb.com/docs/manual/reference/change-events/#delete-event321 */322export interface ChangeStreamDeleteDocument<TSchema extends Document = Document>323  extends ChangeStreamDocumentCommon,324    ChangeStreamDocumentKey<TSchema>,325    ChangeStreamDocumentCollectionUUID,326    ChangeStreamDocumentWallTime {327  /** Describes the type of operation represented in this change notification */328  operationType: 'delete';329  /** Namespace the delete event occurred on */330  ns: ChangeStreamNameSpace;331  /**332   * Contains the pre-image of the modified or deleted document if the333   * pre-image is available for the change event and either 'required' or334   * 'whenAvailable' was specified for the 'fullDocumentBeforeChange' option335   * when creating the change stream. If 'whenAvailable' was specified but the336   * pre-image is unavailable, this will be explicitly set to null.337   */338  fullDocumentBeforeChange?: TSchema;339}340 341/**342 * @public343 * @see https://www.mongodb.com/docs/manual/reference/change-events/#drop-event344 */345export interface ChangeStreamDropDocument346  extends ChangeStreamDocumentCommon,347    ChangeStreamDocumentCollectionUUID,348    ChangeStreamDocumentWallTime {349  /** Describes the type of operation represented in this change notification */350  operationType: 'drop';351  /** Namespace the drop event occurred on */352  ns: ChangeStreamNameSpace;353}354 355/**356 * @public357 * @see https://www.mongodb.com/docs/manual/reference/change-events/#rename-event358 */359export interface ChangeStreamRenameDocument360  extends ChangeStreamDocumentCommon,361    ChangeStreamDocumentCollectionUUID,362    ChangeStreamDocumentWallTime {363  /** Describes the type of operation represented in this change notification */364  operationType: 'rename';365  /** The new name for the `ns.coll` collection */366  to: { db: string; coll: string };367  /** The "from" namespace that the rename occurred on */368  ns: ChangeStreamNameSpace;369}370 371/**372 * @public373 * @see https://www.mongodb.com/docs/manual/reference/change-events/#dropdatabase-event374 */375export interface ChangeStreamDropDatabaseDocument376  extends ChangeStreamDocumentCommon,377    ChangeStreamDocumentWallTime {378  /** Describes the type of operation represented in this change notification */379  operationType: 'dropDatabase';380  /** The database dropped */381  ns: { db: string };382}383 384/**385 * @public386 * @see https://www.mongodb.com/docs/manual/reference/change-events/#invalidate-event387 */388export interface ChangeStreamInvalidateDocument389  extends ChangeStreamDocumentCommon,390    ChangeStreamDocumentWallTime {391  /** Describes the type of operation represented in this change notification */392  operationType: 'invalidate';393}394 395/**396 * Only present when the `showExpandedEvents` flag is enabled.397 * @public398 * @see https://www.mongodb.com/docs/manual/reference/change-events/createIndexes/#mongodb-data-createIndexes399 */400export interface ChangeStreamCreateIndexDocument401  extends ChangeStreamDocumentCommon,402    ChangeStreamDocumentCollectionUUID,403    ChangeStreamDocumentOperationDescription,404    ChangeStreamDocumentWallTime {405  /** Describes the type of operation represented in this change notification */406  operationType: 'createIndexes';407}408 409/**410 * Only present when the `showExpandedEvents` flag is enabled.411 * @public412 * @see https://www.mongodb.com/docs/manual/reference/change-events/dropIndexes/#mongodb-data-dropIndexes413 */414export interface ChangeStreamDropIndexDocument415  extends ChangeStreamDocumentCommon,416    ChangeStreamDocumentCollectionUUID,417    ChangeStreamDocumentOperationDescription,418    ChangeStreamDocumentWallTime {419  /** Describes the type of operation represented in this change notification */420  operationType: 'dropIndexes';421}422 423/**424 * Only present when the `showExpandedEvents` flag is enabled.425 * @public426 * @see https://www.mongodb.com/docs/manual/reference/change-events/modify/#mongodb-data-modify427 */428export interface ChangeStreamCollModDocument429  extends ChangeStreamDocumentCommon,430    ChangeStreamDocumentCollectionUUID,431    ChangeStreamDocumentWallTime {432  /** Describes the type of operation represented in this change notification */433  operationType: 'modify';434}435 436/**437 * @public438 * @see https://www.mongodb.com/docs/manual/reference/change-events/create/#mongodb-data-create439 */440export interface ChangeStreamCreateDocument441  extends ChangeStreamDocumentCommon,442    ChangeStreamDocumentCollectionUUID,443    ChangeStreamDocumentWallTime {444  /** Describes the type of operation represented in this change notification */445  operationType: 'create';446 447  /**448   * The type of the newly created object.449   *450   * @sinceServerVersion 8.1.0451   */452  nsType?: 'collection' | 'timeseries' | 'view';453}454 455/**456 * @public457 * @see https://www.mongodb.com/docs/manual/reference/change-events/shardCollection/#mongodb-data-shardCollection458 */459export interface ChangeStreamShardCollectionDocument460  extends ChangeStreamDocumentCommon,461    ChangeStreamDocumentCollectionUUID,462    ChangeStreamDocumentOperationDescription,463    ChangeStreamDocumentWallTime {464  /** Describes the type of operation represented in this change notification */465  operationType: 'shardCollection';466}467 468/**469 * @public470 * @see https://www.mongodb.com/docs/manual/reference/change-events/reshardCollection/#mongodb-data-reshardCollection471 */472export interface ChangeStreamReshardCollectionDocument473  extends ChangeStreamDocumentCommon,474    ChangeStreamDocumentCollectionUUID,475    ChangeStreamDocumentOperationDescription {476  /** Describes the type of operation represented in this change notification */477  operationType: 'reshardCollection';478}479 480/**481 * @public482 * @see https://www.mongodb.com/docs/manual/reference/change-events/refineCollectionShardKey/#mongodb-data-refineCollectionShardKey483 */484export interface ChangeStreamRefineCollectionShardKeyDocument485  extends ChangeStreamDocumentCommon,486    ChangeStreamDocumentCollectionUUID,487    ChangeStreamDocumentOperationDescription {488  /** Describes the type of operation represented in this change notification */489  operationType: 'refineCollectionShardKey';490}491 492/** @public */493export type ChangeStreamDocument<TSchema extends Document = Document> =494  | ChangeStreamInsertDocument<TSchema>495  | ChangeStreamUpdateDocument<TSchema>496  | ChangeStreamReplaceDocument<TSchema>497  | ChangeStreamDeleteDocument<TSchema>498  | ChangeStreamDropDocument499  | ChangeStreamRenameDocument500  | ChangeStreamDropDatabaseDocument501  | ChangeStreamInvalidateDocument502  | ChangeStreamCreateIndexDocument503  | ChangeStreamCreateDocument504  | ChangeStreamCollModDocument505  | ChangeStreamDropIndexDocument506  | ChangeStreamShardCollectionDocument507  | ChangeStreamReshardCollectionDocument508  | ChangeStreamRefineCollectionShardKeyDocument;509 510/** @public */511export interface UpdateDescription<TSchema extends Document = Document> {512  /**513   * A document containing key:value pairs of names of the fields that were514   * changed, and the new value for those fields.515   */516  updatedFields?: Partial<TSchema>;517 518  /**519   * An array of field names that were removed from the document.520   */521  removedFields?: string[];522 523  /**524   * An array of documents which record array truncations performed with pipeline-based updates using one or more of the following stages:525   * - $addFields526   * - $set527   * - $replaceRoot528   * - $replaceWith529   */530  truncatedArrays?: Array<{531    /** The name of the truncated field. */532    field: string;533    /** The number of elements in the truncated array. */534    newSize: number;535  }>;536 537  /**538   * A document containing additional information about any ambiguous update paths from the update event.  The document539   * maps the full ambiguous update path to an array containing the actual resolved components of the path.  For example,540   * given a document shaped like `{ a: { '0': 0 } }`, and an update of `{ $inc: 'a.0' }`, disambiguated paths would look like541   * the following:542   *543   * ```544   *   {545   *     'a.0': ['a', '0']546   *   }547   * ```548   *549   * This field is only present when there are ambiguous paths that are updated as a part of the update event.550   *551   * On \<8.2.0 servers, this field is only present when `showExpandedEvents` is set to true.552   * is enabled for the change stream.553   *554   * On 8.2.0+ servers, this field is present for update events regardless of whether `showExpandedEvents` is enabled.555   * @sinceServerVersion 6.1.0556   */557  disambiguatedPaths?: Document;558}559 560/** @public */561export type ChangeStreamEvents<562  TSchema extends Document = Document,563  TChange extends Document = ChangeStreamDocument<TSchema>564> = {565  resumeTokenChanged(token: ResumeToken): void;566  init(response: any): void;567  more(response?: any): void;568  response(): void;569  end(): void;570  error(error: Error): void;571  change(change: TChange): void;572  /**573   * @remarks Note that the `close` event is currently emitted whenever the internal `ChangeStreamCursor`574   * instance is closed, which can occur multiple times for a given `ChangeStream` instance.575   *576   * TODO(NODE-6434): address this issue in NODE-6434577   */578  close(): void;579};580 581/**582 * Creates a new Change Stream instance. Normally created using {@link Collection#watch|Collection.watch()}.583 * @public584 */585export class ChangeStream<586    TSchema extends Document = Document,587    TChange extends Document = ChangeStreamDocument<TSchema>588  >589  extends TypedEventEmitter<ChangeStreamEvents<TSchema, TChange>>590  implements AsyncDisposable591{592  /**593   * @beta594   * @experimental595   * An alias for {@link ChangeStream.close|ChangeStream.close()}.596   */597  declare [Symbol.asyncDispose]: () => Promise<void>;598  /** @internal */599  async asyncDispose() {600    await this.close();601  }602 603  pipeline: Document[];604  /**605   * @remarks WriteConcern can still be present on the options because606   * we inherit options from the client/db/collection.  The607   * key must be present on the options in order to delete it.608   * This allows typescript to delete the key but will609   * not allow a writeConcern to be assigned as a property on options.610   */611  options: ChangeStreamOptions & { writeConcern?: never };612  parent: MongoClient | Db | Collection;613  namespace: MongoDBNamespace;614  type: symbol;615  /** @internal */616  private cursor: ChangeStreamCursor<TSchema, TChange>;617  streamOptions?: CursorStreamOptions;618  /** @internal */619  private cursorStream?: Readable & AsyncIterable<TChange>;620  /** @internal */621  private isClosed: boolean;622  /** @internal */623  private mode: false | 'iterator' | 'emitter';624 625  /** @event */626  static readonly RESPONSE = RESPONSE;627  /** @event */628  static readonly MORE = MORE;629  /** @event */630  static readonly INIT = INIT;631  /** @event */632  static readonly CLOSE = CLOSE;633  /**634   * Fired for each new matching change in the specified namespace. Attaching a `change`635   * event listener to a Change Stream will switch the stream into flowing mode. Data will636   * then be passed as soon as it is available.637   * @event638   */639  static readonly CHANGE = CHANGE;640  /** @event */641  static readonly END = END;642  /** @event */643  static readonly ERROR = ERROR;644  /**645   * Emitted each time the change stream stores a new resume token.646   * @event647   */648  static readonly RESUME_TOKEN_CHANGED = RESUME_TOKEN_CHANGED;649 650  private timeoutContext?: TimeoutContext;651  /**652   * Note that this property is here to uniquely identify a ChangeStream instance as the owner of653   * the {@link CursorTimeoutContext} instance (see {@link ChangeStream._createChangeStreamCursor}) to ensure654   * that {@link AbstractCursor.close} does not mutate the timeoutContext.655   */656  private contextOwner: symbol;657  /**658   * @internal659   *660   * @param parent - The parent object that created this change stream661   * @param pipeline - An array of {@link https://www.mongodb.com/docs/manual/reference/operator/aggregation-pipeline/|aggregation pipeline stages} through which to pass change stream documents662   */663  constructor(664    parent: OperationParent,665    pipeline: Document[] = [],666    options: ChangeStreamOptions = {}667  ) {668    super();669 670    this.pipeline = pipeline;671    this.options = { ...options };672    let serverSelectionTimeoutMS: number;673    delete this.options.writeConcern;674 675    if (parent instanceof Collection) {676      this.type = CHANGE_DOMAIN_TYPES.COLLECTION;677      serverSelectionTimeoutMS = parent.s.db.client.options.serverSelectionTimeoutMS;678    } else if (parent instanceof Db) {679      this.type = CHANGE_DOMAIN_TYPES.DATABASE;680      serverSelectionTimeoutMS = parent.client.options.serverSelectionTimeoutMS;681    } else if (parent instanceof MongoClient) {682      this.type = CHANGE_DOMAIN_TYPES.CLUSTER;683      serverSelectionTimeoutMS = parent.options.serverSelectionTimeoutMS;684    } else {685      throw new MongoChangeStreamError(686        'Parent provided to ChangeStream constructor must be an instance of Collection, Db, or MongoClient'687      );688    }689 690    this.contextOwner = Symbol();691    this.parent = parent;692    this.namespace = parent.s.namespace;693    if (!this.options.readPreference && parent.readPreference) {694      this.options.readPreference = parent.readPreference;695    }696 697    // Create contained Change Stream cursor698    this.cursor = this._createChangeStreamCursor(options);699 700    this.isClosed = false;701    this.mode = false;702 703    // Listen for any `change` listeners being added to ChangeStream704    this.on('newListener', eventName => {705      if (eventName === 'change' && this.cursor && this.listenerCount('change') === 0) {706        this._streamEvents(this.cursor);707      }708    });709 710    this.on('removeListener', eventName => {711      if (eventName === 'change' && this.listenerCount('change') === 0 && this.cursor) {712        this.cursorStream?.removeAllListeners('data');713      }714    });715 716    if (this.options.timeoutMS != null) {717      this.timeoutContext = new CSOTTimeoutContext({718        timeoutMS: this.options.timeoutMS,719        serverSelectionTimeoutMS720      });721    }722  }723 724  /** The cached resume token that is used to resume after the most recently returned change. */725  get resumeToken(): ResumeToken {726    return this.cursor?.resumeToken;727  }728 729  /** Check if there is any document still available in the Change Stream */730  async hasNext(): Promise<boolean> {731    this._setIsIterator();732    // Change streams must resume indefinitely while each resume event succeeds.733    // This loop continues until either a change event is received or until a resume attempt734    // fails.735 736    this.timeoutContext?.refresh();737    try {738      while (true) {739        try {740          const hasNext = await this.cursor.hasNext();741          return hasNext;742        } catch (error) {743          try {744            await this._processErrorIteratorMode(error, this.cursor.id != null);745          } catch (error) {746            if (error instanceof MongoOperationTimeoutError && this.cursor.id == null) {747              throw error;748            }749            try {750              await this.close();751            } catch (error) {752              squashError(error);753            }754            throw error;755          }756        }757      }758    } finally {759      this.timeoutContext?.clear();760    }761  }762 763  /** Get the next available document from the Change Stream. */764  async next(): Promise<TChange> {765    this._setIsIterator();766    // Change streams must resume indefinitely while each resume event succeeds.767    // This loop continues until either a change event is received or until a resume attempt768    // fails.769    this.timeoutContext?.refresh();770 771    try {772      while (true) {773        try {774          const change = await this.cursor.next();775          const processedChange = this._processChange(change ?? null);776          return processedChange;777        } catch (error) {778          try {779            await this._processErrorIteratorMode(error, this.cursor.id != null);780          } catch (error) {781            if (error instanceof MongoOperationTimeoutError && this.cursor.id == null) {782              throw error;783            }784            try {785              await this.close();786            } catch (error) {787              squashError(error);788            }789            throw error;790          }791        }792      }793    } finally {794      this.timeoutContext?.clear();795    }796  }797 798  /**799   * Try to get the next available document from the Change Stream's cursor or `null` if an empty batch is returned800   */801  async tryNext(): Promise<TChange | null> {802    this._setIsIterator();803    // Change streams must resume indefinitely while each resume event succeeds.804    // This loop continues until either a change event is received or until a resume attempt805    // fails.806    this.timeoutContext?.refresh();807 808    try {809      while (true) {810        try {811          const change = await this.cursor.tryNext();812          if (!change) {813            return null;814          }815          const processedChange = this._processChange(change);816          return processedChange;817        } catch (error) {818          try {819            await this._processErrorIteratorMode(error, this.cursor.id != null);820          } catch (error) {821            if (error instanceof MongoOperationTimeoutError && this.cursor.id == null) throw error;822            try {823              await this.close();824            } catch (error) {825              squashError(error);826            }827            throw error;828          }829        }830      }831    } finally {832      this.timeoutContext?.clear();833    }834  }835 836  async *[Symbol.asyncIterator](): AsyncGenerator<TChange, void, void> {837    if (this.closed) {838      return;839    }840 841    try {842      // Change streams run indefinitely as long as errors are resumable843      // So the only loop breaking condition is if `next()` throws844      while (true) {845        yield await this.next();846      }847    } finally {848      try {849        await this.close();850      } catch (error) {851        squashError(error);852      }853    }854  }855 856  /** Is the cursor closed */857  public get closed(): boolean {858    return this.isClosed || this.cursor.closed;859  }860 861  /**862   * Frees the internal resources used by the change stream.863   */864  async close(): Promise<void> {865    this.timeoutContext?.clear();866    this.timeoutContext = undefined;867    this.isClosed = true;868 869    const cursor = this.cursor;870    try {871      await cursor.close();872    } finally {873      this._endStream();874    }875  }876 877  /**878   * Return a modified Readable stream including a possible transform method.879   *880   * NOTE: When using a Stream to process change stream events, the stream will881   * NOT automatically resume in the case a resumable error is encountered.882   *883   * @throws MongoChangeStreamError if the underlying cursor or the change stream is closed884   */885  stream(options?: CursorStreamOptions): Readable & AsyncIterable<TChange> {886    if (this.closed) {887      throw new MongoChangeStreamError(CHANGESTREAM_CLOSED_ERROR);888    }889 890    this.streamOptions = options;891    return this.cursor.stream(options);892  }893 894  /** @internal */895  private _setIsEmitter(): void {896    if (this.mode === 'iterator') {897      // TODO(NODE-3485): Replace with MongoChangeStreamModeError898      throw new MongoAPIError(899        'ChangeStream cannot be used as an EventEmitter after being used as an iterator'900      );901    }902    this.mode = 'emitter';903  }904 905  /** @internal */906  private _setIsIterator(): void {907    if (this.mode === 'emitter') {908      // TODO(NODE-3485): Replace with MongoChangeStreamModeError909      throw new MongoAPIError(910        'ChangeStream cannot be used as an iterator after being used as an EventEmitter'911      );912    }913    this.mode = 'iterator';914  }915 916  /**917   * Create a new change stream cursor based on self's configuration918   * @internal919   */920  private _createChangeStreamCursor(921    options: ChangeStreamOptions | ChangeStreamCursorOptions922  ): ChangeStreamCursor<TSchema, TChange> {923    const changeStreamStageOptions = filterOptions(options, CHANGE_STREAM_OPTIONS);924    if (this.type === CHANGE_DOMAIN_TYPES.CLUSTER) {925      changeStreamStageOptions.allChangesForCluster = true;926    }927    const pipeline = [{ $changeStream: changeStreamStageOptions }, ...this.pipeline];928 929    const client: MongoClient | null =930      this.type === CHANGE_DOMAIN_TYPES.CLUSTER931        ? (this.parent as MongoClient)932        : this.type === CHANGE_DOMAIN_TYPES.DATABASE933          ? (this.parent as Db).client934          : this.type === CHANGE_DOMAIN_TYPES.COLLECTION935            ? (this.parent as Collection).client936            : null;937 938    if (client == null) {939      // This should never happen because of the assertion in the constructor940      throw new MongoRuntimeError(941        `Changestream type should only be one of cluster, database, collection. Found ${this.type.toString()}`942      );943    }944 945    const changeStreamCursor = new ChangeStreamCursor<TSchema, TChange>(946      client,947      this.namespace,948      pipeline,949      {950        ...options,951        timeoutContext: this.timeoutContext952          ? new CursorTimeoutContext(this.timeoutContext, this.contextOwner)953          : undefined954      }955    );956 957    for (const event of CHANGE_STREAM_EVENTS) {958      changeStreamCursor.on(event, e => this.emit(event, e));959    }960 961    if (this.listenerCount(ChangeStream.CHANGE) > 0) {962      this._streamEvents(changeStreamCursor);963    }964 965    return changeStreamCursor;966  }967 968  /** @internal */969  private _closeEmitterModeWithError(error: AnyError): void {970    this.emit(ChangeStream.ERROR, error);971 972    this.close().then(undefined, squashError);973  }974 975  /** @internal */976  private _streamEvents(cursor: ChangeStreamCursor<TSchema, TChange>): void {977    this._setIsEmitter();978    const stream = this.cursorStream ?? cursor.stream();979    this.cursorStream = stream;980    stream.on('data', change => {981      try {982        const processedChange = this._processChange(change);983        this.emit(ChangeStream.CHANGE, processedChange);984      } catch (error) {985        this.emit(ChangeStream.ERROR, error);986      }987      this.timeoutContext?.refresh();988    });989    stream.on('error', error => this._processErrorStreamMode(error, this.cursor.id != null));990  }991 992  /** @internal */993  private _endStream(): void {994    this.cursorStream?.removeAllListeners('data');995    this.cursorStream?.removeAllListeners('close');996    this.cursorStream?.removeAllListeners('end');997    this.cursorStream?.destroy();998    this.cursorStream = undefined;999  }1000 1001  /** @internal */1002  private _processChange(change: TChange | null): TChange {1003    if (this.isClosed) {1004      // TODO(NODE-3485): Replace with MongoChangeStreamClosedError1005      throw new MongoAPIError(CHANGESTREAM_CLOSED_ERROR);1006    }1007 1008    // a null change means the cursor has been notified, implicitly closing the change stream1009    if (change == null) {1010      // TODO(NODE-3485): Replace with MongoChangeStreamClosedError1011      throw new MongoRuntimeError(CHANGESTREAM_CLOSED_ERROR);1012    }1013 1014    if (change && !change._id) {1015      throw new MongoChangeStreamError(NO_RESUME_TOKEN_ERROR);1016    }1017 1018    // cache the resume token1019    this.cursor.cacheResumeToken(change._id);1020 1021    // wipe the startAtOperationTime if there was one so that there won't be a conflict1022    // between resumeToken and startAtOperationTime if we need to reconnect the cursor1023    this.options.startAtOperationTime = undefined;1024 1025    return change;1026  }1027 1028  /** @internal */1029  private _processErrorStreamMode(changeStreamError: AnyError, cursorInitialized: boolean) {1030    // If the change stream has been closed explicitly, do not process error.1031    if (this.isClosed) return;1032 1033    if (1034      cursorInitialized &&1035      (isResumableError(changeStreamError, this.cursor.maxWireVersion) ||1036        changeStreamError instanceof MongoOperationTimeoutError)1037    ) {1038      this._endStream();1039 1040      this.cursor1041        .close()1042        .then(1043          () => this._resume(changeStreamError),1044          e => {1045            squashError(e);1046            return this._resume(changeStreamError);1047          }1048        )1049        .then(1050          () => {1051            if (changeStreamError instanceof MongoOperationTimeoutError)1052              this.emit(ChangeStream.ERROR, changeStreamError);1053          },1054          () => this._closeEmitterModeWithError(changeStreamError)1055        );1056    } else {1057      this._closeEmitterModeWithError(changeStreamError);1058    }1059  }1060 1061  /** @internal */1062  private async _processErrorIteratorMode(changeStreamError: AnyError, cursorInitialized: boolean) {1063    if (this.isClosed) {1064      // TODO(NODE-3485): Replace with MongoChangeStreamClosedError1065      throw new MongoAPIError(CHANGESTREAM_CLOSED_ERROR);1066    }1067 1068    if (1069      cursorInitialized &&1070      (isResumableError(changeStreamError, this.cursor.maxWireVersion) ||1071        changeStreamError instanceof MongoOperationTimeoutError)1072    ) {1073      try {1074        await this.cursor.close();1075      } catch (error) {1076        squashError(error);1077      }1078 1079      await this._resume(changeStreamError);1080 1081      if (changeStreamError instanceof MongoOperationTimeoutError) throw changeStreamError;1082    } else {1083      try {1084        await this.close();1085      } catch (error) {1086        squashError(error);1087      }1088 1089      throw changeStreamError;1090    }1091  }1092 1093  private async _resume(changeStreamError: AnyError) {1094    this.timeoutContext?.refresh();1095    const topology = getTopology(this.parent);1096    try {1097      await topology.selectServer(this.cursor.readPreference, {1098        operationName: 'reconnect topology in change stream',1099        timeoutContext: this.timeoutContext1100      });1101      this.cursor = this._createChangeStreamCursor(this.cursor.resumeOptions);1102    } catch {1103      // if the topology can't reconnect, close the stream1104      await this.close();1105      throw changeStreamError;1106    }1107  }1108}1109 1110configureResourceManagement(ChangeStream.prototype);1111