fix(NODE-5213): ChangeStream.tryNext() should return TChange type (#3649)

Author: alecgibson

src/change_stream.ts

@@ -700,7 +700,7 @@ export class ChangeStream<
  /**
  * Try to get the next available document from the Change Stream's cursor or `null` if an empty batch is returned
  */
- async tryNext(): Promise<Document | null> {
+ async tryNext(): Promise<TChange | null> {
  this._setIsIterator();
  // Change streams must resume indefinitely while each resume event succeeds.
  // This loop continues until either a change event is received or until a resume attempt

test/types/change_stream.test-d.ts

@@ -15,20 +15,18 @@ import type {
  ChangeStreamInvalidateDocument,
  ChangeStreamNameSpace,
  ChangeStreamOptions,
+ ChangeStreamRefineCollectionShardKeyDocument,
  ChangeStreamRenameDocument,
  ChangeStreamReplaceDocument,
+ ChangeStreamReshardCollectionDocument,
+ ChangeStreamShardCollectionDocument,
  ChangeStreamUpdateDocument,
  Collection,
  Document,
  ResumeToken,
  ServerSessionId,
  Timestamp,
  UpdateDescription
-} from '../../src';
-import type {
- ChangeStreamRefineCollectionShardKeyDocument,
- ChangeStreamReshardCollectionDocument,
- ChangeStreamShardCollectionDocument
 } from '../mongodb';
 
 declare const changeStreamOptions: ChangeStreamOptions;
@@ -181,8 +179,8 @@ switch (change.operationType) {
 // New fields can be added with $addFields, but you have to use TChange to type it
 expectError(change.randomKeyAlwaysAccessibleBecauseOfPipelineFlexibilty);
 
-declare const collection: Collection<Schema>;
-const pipelineChangeStream = collection.watch<
+declare const collectionWithSchema: Collection<Schema>;
+const pipelineChangeStream = collectionWithSchema.watch<
  Schema,
  ChangeStreamInsertDocument<Schema> & { comment: string }
 >([{ $addFields: { comment: 'big changes' } }, { $match: { operationType: 'insert' } }]);
@@ -193,28 +191,52 @@ pipelineChangeStream.on('change', change => {
  expectType<Schema>(change.fullDocument);
 });
 
-collection.watch().on('change', change => expectType<ChangeStreamDocument<Schema>>(change));
+collectionWithSchema
+ .watch()
+ .on('change', change => expectType<ChangeStreamDocument<Schema>>(change));
 
 // Just overriding the schema provides a typed changestream OF that schema
-collection
+collectionWithSchema
  .watch<Document>()
  .on('change', change => expectType<ChangeStreamDocument<Document>>(change));
 
-// both schema and Tchange can be made as flexible as possible (Document)
-collection.watch<Document, Document>().on('change', change => expectType<Document>(change));
+// both schema and TChange can be made as flexible as possible (Document)
+collectionWithSchema
+ .watch<Document, Document>()
+ .on('change', change => expectType<Document>(change));
 
 // first argument does not stop you from making second more generic
-collection.watch<{ a: number }, Document>().on('change', change => expectType<Document>(change));
+collectionWithSchema
+ .watch<{ a: number }, Document>()
+ .on('change', change => expectType<Document>(change));
 
 // Arguments must be objects
-expectError(collection.watch<Document, number>());
-expectError(collection.watch<number, number>());
+expectError(collectionWithSchema.watch<Document, number>());
+expectError(collectionWithSchema.watch<number, number>());
 
 // First argument no longer relates to second
-collection
+collectionWithSchema
  .watch<{ a: number }, { b: boolean }>()
  .on('change', change => expectType<{ b: boolean }>(change));
 
 expectType<AsyncGenerator<ChangeStreamDocument<Schema>, void, void>>(
- collection.watch()[Symbol.asyncIterator]()
+ collectionWithSchema.watch()[Symbol.asyncIterator]()
+);
+
+// Change type returned to user is equivalent across next/tryNext/on/once/addListener
+const changeStream = collectionWithSchema.watch();
+expectType<ChangeStreamDocument<Schema> | null>(await changeStream.tryNext());
+expectType<ChangeStreamDocument<Schema>>(await changeStream.next());
+changeStream.on('change', change => expectType<ChangeStreamDocument<Schema>>(change));
+changeStream.once('change', change => expectType<ChangeStreamDocument<Schema>>(change));
+changeStream.addListener('change', change => expectType<ChangeStreamDocument<Schema>>(change));
+
+declare const noSchemaCollection: Collection;
+const changeStreamNoSchema = noSchemaCollection.watch();
+expectType<ChangeStreamDocument<Document> | null>(await changeStreamNoSchema.tryNext());
+expectType<ChangeStreamDocument<Document>>(await changeStreamNoSchema.next());
+changeStreamNoSchema.on('change', change => expectType<ChangeStreamDocument<Document>>(change));
+changeStreamNoSchema.once('change', change => expectType<ChangeStreamDocument<Document>>(change));
+changeStreamNoSchema.addListener('change', change =>
+ expectType<ChangeStreamDocument<Document>>(change)
 );