command op too

Author: baileympearson

src/operations/command.ts

@@ -1,3 +1,4 @@
+import { type Connection } from '..';
 import type { BSONSerializeOptions, Document } from '../bson';
 import { type MongoDBResponseConstructor } from '../cmap/wire_protocol/responses';
 import { MongoInvalidArgumentError } from '../error';
@@ -9,14 +10,14 @@ import {
 } from '../explain';
 import { ReadConcern } from '../read_concern';
 import type { ReadPreference } from '../read_preference';
-import type { Server } from '../sdam/server';
+import type { Server, ServerCommandOptions } from '../sdam/server';
 import { MIN_SECONDARY_WRITE_WIRE_VERSION } from '../sdam/server_selection';
 import type { ClientSession } from '../sessions';
 import { type TimeoutContext } from '../timeout';
 import { commandSupportsReadConcern, maxWireVersion, MongoDBNamespace } from '../utils';
 import { WriteConcern, type WriteConcernOptions } from '../write_concern';
 import type { ReadConcernLike } from './../read_concern';
-import { AbstractOperation, Aspect, type OperationOptions } from './operation';
+import { AbstractOperation, Aspect, ModernizedOperation, type OperationOptions } from './operation';
 
 /** @public */
 export interface CollationOptions {
@@ -183,3 +184,93 @@ export abstract class CommandOperation<T> extends AbstractOperation<T> {
     return await server.command(this.ns, cmd, options, responseType);
   }
 }
+
+export abstract class ModernizedCommandOperation<T> extends ModernizedOperation<T> {
+  override options: CommandOperationOptions;
+  readConcern?: ReadConcern;
+  writeConcern?: WriteConcern;
+  explain?: Explain;
+
+  constructor(parent?: OperationParent, options?: CommandOperationOptions) {
+    super(options);
+    this.options = options ?? {};
+
+    // NOTE: this was explicitly added for the add/remove user operations, it's likely
+    //       something we'd want to reconsider. Perhaps those commands can use `Admin`
+    //       as a parent?
+    const dbNameOverride = options?.dbName || options?.authdb;
+    if (dbNameOverride) {
+      this.ns = new MongoDBNamespace(dbNameOverride, '$cmd');
+    } else {
+      this.ns = parent
+        ? parent.s.namespace.withCollection('$cmd')
+        : new MongoDBNamespace('admin', '$cmd');
+    }
+
+    this.readConcern = ReadConcern.fromOptions(options);
+    this.writeConcern = WriteConcern.fromOptions(options);
+
+    if (this.hasAspect(Aspect.EXPLAINABLE)) {
+      this.explain = Explain.fromOptions(options);
+      if (this.explain) validateExplainTimeoutOptions(this.options, this.explain);
+    } else if (options?.explain != null) {
+      throw new MongoInvalidArgumentError(`Option "explain" is not supported on this command`);
+    }
+  }
+
+  override get canRetryWrite(): boolean {
+    if (this.hasAspect(Aspect.EXPLAINABLE)) {
+      return this.explain == null;
+    }
+    return super.canRetryWrite;
+  }
+
+  abstract buildCommandDocument(connection: Connection, session?: ClientSession): Document;
+
+  override buildOptions(timeoutContext: TimeoutContext): ServerCommandOptions {
+    return {
+      ...this.options,
+      ...this.bsonOptions,
+      timeoutContext,
+      readPreference: this.readPreference,
+      session: this.session
+    };
+  }
+
+  override buildCommand(connection: Connection, session?: ClientSession): Document {
+    const command = this.buildCommandDocument(connection, session);
+
+    const serverWireVersion = maxWireVersion(connection);
+    const inTransaction = this.session && this.session.inTransaction();
+
+    if (this.readConcern && commandSupportsReadConcern(command) && !inTransaction) {
+      Object.assign(command, { readConcern: this.readConcern });
+    }
+
+    if (this.trySecondaryWrite && serverWireVersion < MIN_SECONDARY_WRITE_WIRE_VERSION) {
+      command.omitReadPreference = true;
+    }
+
+    if (this.writeConcern && this.hasAspect(Aspect.WRITE_OPERATION) && !inTransaction) {
+      WriteConcern.apply(command, this.writeConcern);
+    }
+
+    if (
+      this.options.collation &&
+      typeof this.options.collation === 'object' &&
+      !this.hasAspect(Aspect.SKIP_COLLATION)
+    ) {
+      Object.assign(command, { collation: this.options.collation });
+    }
+
+    if (typeof this.options.maxTimeMS === 'number') {
+      command.maxTimeMS = this.options.maxTimeMS;
+    }
+
+    if (this.hasAspect(Aspect.EXPLAINABLE) && this.explain) {
+      return decorateWithExplain(command, this.explain);
+    }
+
+    return command;
+  }
+}

src/operations/execute_operation.ts

@@ -279,6 +279,8 @@ async function tryOperation<
       }
     }
 
+    operation.server = server;
+
     try {
       // If tries > 0 and we are command batching we need to reset the batch.
       if (tries > 0 && operation.hasAspect(Aspect.COMMAND_BATCHING)) {

src/operations/insert.ts

@@ -1,18 +1,21 @@
+import { type Connection } from '..';
 import type { Document } from '../bson';
 import type { BulkWriteOptions } from '../bulk/common';
+import { MongoDBResponse } from '../cmap/wire_protocol/responses';
 import type { Collection } from '../collection';
 import { MongoServerError } from '../error';
 import type { InferIdType } from '../mongo_types';
 import type { Server } from '../sdam/server';
 import type { ClientSession } from '../sessions';
 import { type TimeoutContext } from '../timeout';
 import { maybeAddIdToDocuments, type MongoDBNamespace } from '../utils';
-import { CommandOperation, type CommandOperationOptions } from './command';
+import { type CommandOperationOptions, ModernizedCommandOperation } from './command';
 import { Aspect, defineAspects } from './operation';
-
 /** @internal */
-export class InsertOperation extends CommandOperation<Document> {
+export class InsertOperation extends ModernizedCommandOperation<Document> {
+  override SERVER_COMMAND_RESPONSE_TYPE = MongoDBResponse;
   override options: BulkWriteOptions;
+
   documents: Document[];
 
   constructor(ns: MongoDBNamespace, documents: Document[], options: BulkWriteOptions) {
@@ -26,11 +29,7 @@ export class InsertOperation extends CommandOperation<Document> {
     return 'insert' as const;
   }
 
-  override async execute(
-    server: Server,
-    session: ClientSession | undefined,
-    timeoutContext: TimeoutContext
-  ): Promise<Document> {
+  override buildCommandDocument(_connection: Connection, _session?: ClientSession): Document {
     const options = this.options ?? {};
     const ordered = typeof options.ordered === 'boolean' ? options.ordered : true;
     const command: Document = {
@@ -49,7 +48,7 @@ export class InsertOperation extends CommandOperation<Document> {
       command.comment = options.comment;
     }
 
-    return await super.executeCommand(server, session, command, timeoutContext);
+    return command;
   }
 }
 
@@ -91,6 +90,20 @@ export class InsertOneOperation extends InsertOperation {
       insertedId: this.documents[0]._id
     };
   }
+
+  override handleOk(response: InstanceType<typeof this.SERVER_COMMAND_RESPONSE_TYPE>): Document {
+    const res = super.handleOk(response);
+    if (res.code) throw new MongoServerError(res);
+    if (res.writeErrors) {
+      // This should be a WriteError but we can't change it now because of error hierarchy
+      throw new MongoServerError(res.writeErrors[0]);
+    }
+
+    return {
+      acknowledged: this.writeConcern?.w !== 0,
+      insertedId: this.documents[0]._id
+    };
+  }
 }
 
 /** @public */

src/operations/search_indexes/drop.ts

@@ -1,12 +1,12 @@
 import { type Connection, type MongoError } from '../..';
 import type { Document } from '../../bson';
+import { MongoDBResponse } from '../../cmap/wire_protocol/responses';
 import type { Collection } from '../../collection';
 import { MONGODB_ERROR_CODES, MongoServerError } from '../../error';
 import type { ServerCommandOptions } from '../../sdam/server';
 import type { ClientSession } from '../../sessions';
 import { type TimeoutContext } from '../../timeout';
 import { ModernizedOperation } from '../operation';
-import { MongoDBResponse } from '../../cmap/wire_protocol/responses'
 
 /** @internal */
 export class DropSearchIndexOperation extends ModernizedOperation<void> {

test/integration/retryable-writes/non-server-retryable_writes.test.ts

@@ -32,7 +32,7 @@ describe('Non Server Retryable Writes', function () {
     'returns the original error with a PoolRequstedRetry label after encountering a WriteConcernError',
     { requires: { topology: 'replicaset' } },
     async () => {
-      const serverCommandStub = sinon.stub(Server.prototype, 'command');
+      const serverCommandStub = sinon.stub(Server.prototype, 'modernCommand');
       serverCommandStub.onCall(0).rejects(new PoolClearedError('error'));
       serverCommandStub.onCall(1).returns(
         Promise.reject(