feat(NODE-6329): client bulk write happy path (#4206)

Author: durran

src/cmap/command_monitoring_events.ts

@@ -7,7 +7,12 @@ import {
   LEGACY_HELLO_COMMAND_CAMEL_CASE
 } from '../constants';
 import { calculateDurationInMs, deepCopy } from '../utils';
-import { OpMsgRequest, type OpQueryRequest, type WriteProtocolMessageType } from './commands';
+import {
+  DocumentSequence,
+  OpMsgRequest,
+  type OpQueryRequest,
+  type WriteProtocolMessageType
+} from './commands';
 import type { Connection } from './connection';
 
 /**
@@ -249,7 +254,16 @@ const OP_QUERY_KEYS = [
 /** Extract the actual command from the query, possibly up-converting if it's a legacy format */
 function extractCommand(command: WriteProtocolMessageType): Document {
   if (command instanceof OpMsgRequest) {
-    return deepCopy(command.command);
+    const cmd = deepCopy(command.command);
+    // For OP_MSG with payload type 1 we need to pull the documents
+    // array out of the document sequence for monitoring.
+    if (cmd.ops instanceof DocumentSequence) {
+      cmd.ops = cmd.ops.documents;
+    }
+    if (cmd.nsInfo instanceof DocumentSequence) {
+      cmd.nsInfo = cmd.nsInfo.documents;
+    }
+    return cmd;
   }
 
   if (command.query?.$query) {

src/cmap/commands.ts

@@ -544,10 +544,10 @@ export class OpMsgRequest {
     for (const [key, value] of Object.entries(document)) {
       if (value instanceof DocumentSequence) {
         // Document sequences starts with type 1 at the first byte.
-        const buffer = Buffer.allocUnsafe(1 + 4 + key.length);
+        const buffer = Buffer.allocUnsafe(1 + 4 + key.length + 1);
         buffer[0] = 1;
-        // Third part is the field name at offset 5.
-        encodeUTF8Into(buffer, key, 5);
+        // Third part is the field name at offset 5 with trailing null byte.
+        encodeUTF8Into(buffer, `${key}\0`, 5);
         chunks.push(buffer);
         // Fourth part are the documents' bytes.
         let docsLength = 0;
@@ -557,7 +557,7 @@ export class OpMsgRequest {
           chunks.push(docBson);
         }
         // Second part of the sequence is the length at offset 1;
-        buffer.writeInt32LE(key.length + docsLength, 1);
+        buffer.writeInt32LE(4 + key.length + 1 + docsLength, 1);
         // Why are we removing the field from the command? This is because it needs to be
         // removed in the OP_MSG request first section, and DocumentSequence is not a
         // BSON type and is specific to the MongoDB wire protocol so there's nothing