feat(NODE-4059): ChangeStreamDocument not fully typed to specification

Author: nbbeeken

src/change_stream.ts

@@ -2,7 +2,7 @@ import { BSONSerializeOptions, Document, resolveBSONOptions } from './bson';
 import type { AnyBulkWriteOperation, BulkWriteOptions, BulkWriteResult } from './bulk/common';
 import { OrderedBulkOperation } from './bulk/ordered';
 import { UnorderedBulkOperation } from './bulk/unordered';
-import { ChangeStream, ChangeStreamOptions } from './change_stream';
+import { ChangeStream, ChangeStreamDocument, ChangeStreamOptions } from './change_stream';
 import { AggregationCursor } from './cursor/aggregation_cursor';
 import { FindCursor } from './cursor/find_cursor';
 import type { Db } from './db';
@@ -1422,17 +1422,17 @@ export class Collection<TSchema extends Document = Document> {
    * @param pipeline - An array of {@link https://docs.mongodb.com/manual/reference/operator/aggregation-pipeline/|aggregation pipeline stages} through which to pass change stream documents. This allows for filtering (using $match) and manipulating the change stream documents.
    * @param options - Optional settings for the command
    */
-  watch<TLocal extends Document = TSchema>(
-    pipeline: Document[] = [],
-    options: ChangeStreamOptions = {}
-  ): ChangeStream<TLocal> {
+  watch<
+    TLocal extends Document = TSchema,
+    TChange extends ChangeStreamDocument<TLocal> = ChangeStreamDocument<TLocal>
+  >(pipeline: Document[] = [], options: ChangeStreamOptions = {}): ChangeStream<TLocal, TChange> {
     // Allow optionally not specifying a pipeline
     if (!Array.isArray(pipeline)) {
       options = pipeline;
       pipeline = [];
     }
 
-    return new ChangeStream<TLocal>(this, pipeline, resolveOptions(this, options));
+    return new ChangeStream<TLocal, TChange>(this, pipeline, resolveOptions(this, options));
   }
 
   /**

src/collection.ts

@@ -1,6 +1,6 @@
 import { Admin } from './admin';
 import { BSONSerializeOptions, Document, resolveBSONOptions } from './bson';
-import { ChangeStream, ChangeStreamOptions } from './change_stream';
+import { ChangeStream, ChangeStreamDocument, ChangeStreamOptions } from './change_stream';
 import { Collection, CollectionOptions } from './collection';
 import * as CONSTANTS from './constants';
 import { AggregationCursor } from './cursor/aggregation_cursor';
@@ -722,17 +722,17 @@ export class Db {
    * @param pipeline - An array of {@link https://docs.mongodb.com/manual/reference/operator/aggregation-pipeline/|aggregation pipeline stages} through which to pass change stream documents. This allows for filtering (using $match) and manipulating the change stream documents.
    * @param options - Optional settings for the command
    */
-  watch<TSchema extends Document = Document>(
-    pipeline: Document[] = [],
-    options: ChangeStreamOptions = {}
-  ): ChangeStream<TSchema> {
+  watch<
+    TSchema extends Document = Document,
+    TChange extends ChangeStreamDocument<TSchema> = ChangeStreamDocument<TSchema>
+  >(pipeline: Document[] = [], options: ChangeStreamOptions = {}): ChangeStream<TSchema, TChange> {
     // Allow optionally not specifying a pipeline
     if (!Array.isArray(pipeline)) {
       options = pipeline;
       pipeline = [];
     }
 
-    return new ChangeStream<TSchema>(this, pipeline, resolveOptions(this, options));
+    return new ChangeStream<TSchema, TChange>(this, pipeline, resolveOptions(this, options));
   }
 
   /** Return the db logger */

src/db.ts

@@ -172,9 +172,20 @@ export type {
   ChangeStreamAggregateRawResult,
   ChangeStreamCursor,
   ChangeStreamCursorOptions,
+  ChangeStreamDeleteDocument,
   ChangeStreamDocument,
+  ChangeStreamDocumentCommon,
+  ChangeStreamDocumentKey,
+  ChangeStreamDropDatabaseDocument,
+  ChangeStreamDropDocument,
   ChangeStreamEvents,
+  ChangeStreamInsertDocument,
+  ChangeStreamInvalidateDocument,
+  ChangeStreamNameSpace,
   ChangeStreamOptions,
+  ChangeStreamRenameDocument,
+  ChangeStreamReplaceDocument,
+  ChangeStreamUpdateDocument,
   OperationTime,
   PipeOptions,
   ResumeOptions,

src/index.ts

@@ -2,7 +2,7 @@ import type { TcpNetConnectOpts } from 'net';
 import type { ConnectionOptions as TLSConnectionOptions, TLSSocketOptions } from 'tls';
 
 import { BSONSerializeOptions, Document, resolveBSONOptions } from './bson';
-import { ChangeStream, ChangeStreamOptions } from './change_stream';
+import { ChangeStream, ChangeStreamDocument, ChangeStreamOptions } from './change_stream';
 import type { AuthMechanismProperties, MongoCredentials } from './cmap/auth/mongo_credentials';
 import type { AuthMechanism } from './cmap/auth/providers';
 import type { LEGAL_TCP_SOCKET_OPTIONS, LEGAL_TLS_SOCKET_OPTIONS } from './cmap/connect';
@@ -593,17 +593,17 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> {
    * @param pipeline - An array of {@link https://docs.mongodb.com/manual/reference/operator/aggregation-pipeline/|aggregation pipeline stages} through which to pass change stream documents. This allows for filtering (using $match) and manipulating the change stream documents.
    * @param options - Optional settings for the command
    */
-  watch<TSchema extends Document = Document>(
-    pipeline: Document[] = [],
-    options: ChangeStreamOptions = {}
-  ): ChangeStream<TSchema> {
+  watch<
+    TSchema extends Document = Document,
+    TChange extends ChangeStreamDocument<TSchema> = ChangeStreamDocument<TSchema>
+  >(pipeline: Document[] = [], options: ChangeStreamOptions = {}): ChangeStream<TSchema, TChange> {
     // Allow optionally not specifying a pipeline
     if (!Array.isArray(pipeline)) {
       options = pipeline;
       pipeline = [];
     }
 
-    return new ChangeStream<TSchema>(this, pipeline, resolveOptions(this, options));
+    return new ChangeStream<TSchema, TChange>(this, pipeline, resolveOptions(this, options));
   }
 
   /** Return the mongo client logger */