feat(MongoDB Node): Batch update and find-and-update writes with bulkWrite (#37035)

packages/nodes-base/nodes/MongoDb/MongoDb.node.ts

@@ -1,11 +1,13 @@
 import type {
+	AnyBulkWriteOperation,
+	Db,
 	FindOneAndReplaceOptions,
 	FindOneAndUpdateOptions,
 	UpdateOptions,
 	Sort,
 	MongoClient,
 } from 'mongodb';
-import { ObjectId } from 'mongodb';
+import { MongoBulkWriteError, ObjectId } from 'mongodb';
 import { NodeConnectionTypes, NodeOperationError, UserError } from 'n8n-workflow';
 import type {
 	IExecuteFunctions,
@@ -35,13 +37,130 @@ import type { IMongoParametricCredentials } from './mongoDb.types';
 import { nodeProperties } from './MongoDbProperties';
 import { generatePairedItemData } from '../../utils/utilities';
 
+interface BulkUpdateEntry {
+	op: AnyBulkWriteOperation;
+	item: IDataObject;
+	originalIndex: number;
+}
+
+/**
+ * Batches update/findOneAndUpdate items into one `bulkWrite` per collection.
+ * Both operations already discard the driver result and echo the prepared
+ * item, so the per-item output contract is unchanged.
+ */
+async function executeBulkUpdate(
+	ctx: IExecuteFunctions,
+	mdb: Db,
+	items: INodeExecutionData[],
+	itemsLength: number,
+	sanitizeErrorMessage: (error: unknown) => string,
+): Promise<INodeExecutionData[]> {
+	const continueOnFail = ctx.continueOnFail();
+	const returnData: INodeExecutionData[] = [];
+	const groups = new Map<string, BulkUpdateEntry[]>();
+
+	for (let i = 0; i < itemsLength; i++) {
+		try {
+			const fields = prepareFields(ctx.getNodeParameter('fields', i) as string);
+			const useDotNotation = Boolean(ctx.getNodeParameter('options.useDotNotation', i, false));
+			const dateFields = prepareFields(ctx.getNodeParameter('options.dateFields', i, '') as string);
+			const updateKey = ((ctx.getNodeParameter('updateKey', i) as string) || '').trim();
+			const upsert = Boolean(ctx.getNodeParameter('upsert', i));
+
+			const [item] = prepareItems({
+				items: [items[i]],
+				fields,
+				updateKey,
+				useDotNotation,
+				dateFields,
+				isUpdate: true,
+				node: ctx.getNode(),
+			});
+
+			if (!item) {
+				throw new NodeOperationError(ctx.getNode(), 'Item is missing the updateKey field', {
+					itemIndex: i,
+				});
+			}
+
+			const filter: IDataObject = { [updateKey]: item[updateKey] };
+			if (updateKey === '_id') {
+				filter[updateKey] = new ObjectId(item[updateKey] as string);
+				delete item._id;
+			}
+
+			const collection = ctx.getNodeParameter('collection', i) as string;
+			const group = groups.get(collection) ?? [];
+			groups.set(collection, group);
+			group.push({
+				op: { updateOne: { filter, update: { $set: item }, ...(upsert ? { upsert: true } : {}) } },
+				item,
+				originalIndex: i,
+			});
+		} catch (error) {
+			if (!continueOnFail) throw error;
+			returnData.push({
+				json: { error: sanitizeErrorMessage(error) },
+				pairedItem: { item: i },
+			});
+		}
+	}
+
+	for (const [collection, entries] of groups) {
+		try {
+			// Ordered stops at the first failure within a collection (the pre-1.5 per-item
+			// behaviour). Across interleaved collections it does not: groups run one at a
+			// time, so a later group's failure can't stop an already-run group — matches
+			// Insert, and only observable when `collection` is a per-item expression.
+			// Continue-on-fail flips to unordered: attempt every item independently.
+			await mdb.collection(collection).bulkWrite(
+				entries.map((entry) => entry.op),
+				{ ordered: !continueOnFail },
+			);
+			for (const entry of entries) {
+				returnData.push({ json: entry.item, pairedItem: { item: entry.originalIndex } });
+			}
+		} catch (error) {
+			if (!continueOnFail) throw error;
+
+			// writeErrors carry the op's position within this bulkWrite call.
+			// The driver types this as OneOrMore<WriteError>, so normalise to an array.
+			const failedOpIndexes = new Map<number, string>();
+			if (error instanceof MongoBulkWriteError) {
+				const writeErrors = [error.writeErrors].flat();
+				for (const writeError of writeErrors) {
+					failedOpIndexes.set(writeError.index, writeError.errmsg ?? error.message);
+				}
+			}
+
+			for (const [opIndex, entry] of entries.entries()) {
+				const failure = failedOpIndexes.get(opIndex);
+				// No per-op verdicts (e.g. a connection failure): treat the whole group as failed
+				if (failure !== undefined || failedOpIndexes.size === 0) {
+					returnData.push({
+						json: { error: sanitizeErrorMessage(failure ?? error) },
+						pairedItem: { item: entry.originalIndex },
+					});
+				} else {
+					returnData.push({ json: entry.item, pairedItem: { item: entry.originalIndex } });
+				}
+			}
+		}
+	}
+
+	returnData.sort(
+		(a, b) => (a.pairedItem as { item: number }).item - (b.pairedItem as { item: number }).item,
+	);
+	return returnData;
+}
+
 export class MongoDb implements INodeType {
 	description: INodeTypeDescription = {
 		displayName: 'MongoDB',
 		name: 'mongoDb',
 		icon: 'file:mongodb.svg',
 		group: ['input'],
-		version: [1, 1.1, 1.2, 1.3, 1.4],
+		version: [1, 1.1, 1.2, 1.3, 1.4, 1.5],
 		description: 'Find, insert and update documents in MongoDB',
 		defaults: {
 			name: 'MongoDB',
@@ -373,7 +492,11 @@ export class MongoDb implements INodeType {
 
 			if (operation === 'findOneAndUpdate') {
 				fallbackPairedItems = fallbackPairedItems ?? generatePairedItemData(items.length);
-				if (nodeVersion >= 1.3) {
+				if (nodeVersion >= 1.5) {
+					returnData = returnData.concat(
+						await executeBulkUpdate(this, mdb, items, itemsLength, sanitizeErrorMessage),
+					);
+				} else if (nodeVersion >= 1.3) {
 					for (let i = 0; i < itemsLength; i++) {
 						const fields = prepareFields(this.getNodeParameter('fields', i) as string);
 						const useDotNotation = this.getNodeParameter(
@@ -612,7 +735,11 @@ export class MongoDb implements INodeType {
 
 			if (operation === 'update') {
 				fallbackPairedItems = fallbackPairedItems ?? generatePairedItemData(items.length);
-				if (nodeVersion >= 1.3) {
+				if (nodeVersion >= 1.5) {
+					returnData = returnData.concat(
+						await executeBulkUpdate(this, mdb, items, itemsLength, sanitizeErrorMessage),
+					);
+				} else if (nodeVersion >= 1.3) {
 					for (let i = 0; i < itemsLength; i++) {
 						const fields = prepareFields(this.getNodeParameter('fields', i) as string);
 						const useDotNotation = this.getNodeParameter(

packages/nodes-base/nodes/MongoDb/test/MongoDB.test.ts

@@ -1,6 +1,6 @@
 import { NodeTestHarness } from '@nodes-testing/node-test-harness';
 import { mockDeep } from 'vitest-mock-extended';
-import { Collection, Db, MongoClient, ObjectId } from 'mongodb';
+import { Collection, Db, MongoBulkWriteError, MongoClient, ObjectId } from 'mongodb';
 import { constructExecutionMetaData, returnJsonArray } from 'n8n-core';
 import type {
 	IExecuteFunctions,
@@ -176,6 +176,271 @@ function searchIndexOperationResult(indexName: string) {
 describe('MongoDB CRUD Node', () => {
 	const testHarness = new NodeTestHarness();
 
+	describe('document operations in version 1.5', () => {
+		let collectionSpy: MockInstance;
+		const node = new MongoDb();
+
+		function bulkWriteError(
+			writeErrors: Array<{ index: number; errmsg: string }>,
+			message = 'bulk write failed',
+		) {
+			const error = Object.create(MongoBulkWriteError.prototype) as MongoBulkWriteError;
+			Object.assign(error, { message, writeErrors });
+			return error;
+		}
+
+		function mockBulkExecuteFunctions(
+			operation: string,
+			{
+				continueOnFail = false,
+				params = {},
+			}: {
+				continueOnFail?: boolean;
+				params?: Record<
+					string,
+					NodeParameterValueType | ((itemIndex: number) => NodeParameterValueType)
+				>;
+			} = {},
+		) {
+			const executeFunctions = mockExecuteFunctions(1.5, operation);
+			executeFunctions.continueOnFail.mockReturnValue(continueOnFail);
+			const merged = new Map<
+				string,
+				NodeParameterValueType | ((itemIndex: number) => NodeParameterValueType)
+			>([
+				['operation', operation],
+				['collection', 'users'],
+				['fields', 'id,value'],
+				['updateKey', 'id'],
+				['upsert', false],
+				['options.useDotNotation', false],
+				['options.dateFields', ''],
+				...Object.entries(params),
+			]);
+			executeFunctions.getNodeParameter.mockImplementation(
+				(parameterName: string, itemIndex = 0, fallbackValue?: NodeParameterValueType) => {
+					if (!merged.has(parameterName)) return fallbackValue as never;
+					const value = merged.get(parameterName);
+					return (typeof value === 'function' ? value(itemIndex) : value) as never;
+				},
+			);
+			return executeFunctions;
+		}
+
+		beforeEach(() => {
+			collectionSpy = vi.spyOn(Db.prototype, 'collection');
+		});
+
+		afterEach(() => {
+			collectionSpy.mockRestore();
+			vi.clearAllMocks();
+		});
+
+		it.each(['update', 'findOneAndUpdate'])(
+			'batches %s items into a single ordered bulkWrite per collection',
+			async (operation) => {
+				const updateOneSpy = vi.spyOn(Collection.prototype, 'updateOne');
+				const findOneAndUpdateSpy = vi.spyOn(Collection.prototype, 'findOneAndUpdate');
+				const bulkWriteSpy = vi
+					.spyOn(Collection.prototype, 'bulkWrite')
+					.mockResolvedValue({} as never);
+
+				const [items] = await node.execute.call(mockBulkExecuteFunctions(operation));
+
+				expect(bulkWriteSpy).toHaveBeenCalledTimes(1);
+				expect(bulkWriteSpy).toHaveBeenCalledWith(
+					[
+						{ updateOne: { filter: { id: '1' }, update: { $set: { id: '1', value: 'first' } } } },
+						{ updateOne: { filter: { id: '2' }, update: { $set: { id: '2', value: 'second' } } } },
+						{ updateOne: { filter: { id: '3' }, update: { $set: { id: '3', value: 'third' } } } },
+					],
+					{ ordered: true },
+				);
+				expect(updateOneSpy).not.toHaveBeenCalled();
+				expect(findOneAndUpdateSpy).not.toHaveBeenCalled();
+				expect(items).toEqual([
+					{ json: { id: '1', value: 'first' }, pairedItem: { item: 0 } },
+					{ json: { id: '2', value: 'second' }, pairedItem: { item: 1 } },
+					{ json: { id: '3', value: 'third' }, pairedItem: { item: 2 } },
+				]);
+			},
+		);
+
+		it('resolves the collection per item and issues one bulkWrite per group', async () => {
+			const bulkWriteSpy = vi
+				.spyOn(Collection.prototype, 'bulkWrite')
+				.mockResolvedValue({} as never);
+
+			await node.execute.call(mockExecuteFunctions(1.5, 'update'));
+
+			expect(collectionNames(collectionSpy)).toEqual([
+				'collection-1',
+				'collection-2',
+				'collection-3',
+			]);
+			expect(bulkWriteSpy).toHaveBeenCalledTimes(3);
+		});
+
+		it('restores input order when grouping interleaves collections', async () => {
+			const bulkWriteSpy = vi
+				.spyOn(Collection.prototype, 'bulkWrite')
+				.mockResolvedValue({} as never);
+			const executeFunctions = mockBulkExecuteFunctions('update', {
+				params: { collection: (itemIndex: number) => ['a', 'b', 'a'][itemIndex] },
+			});
+
+			const [items] = await node.execute.call(executeFunctions);
+
+			expect(bulkWriteSpy).toHaveBeenCalledTimes(2);
+			expect(items.map((item) => item.pairedItem)).toEqual([{ item: 0 }, { item: 1 }, { item: 2 }]);
+		});
+
+		// The string case pins the pre-1.5 truthy coercion for expression-driven values
+		it.each([true, 'true'])(
+			'sends upsert per operation when the parameter is truthy (%j)',
+			async (upsert) => {
+				const bulkWriteSpy = vi
+					.spyOn(Collection.prototype, 'bulkWrite')
+					.mockResolvedValue({} as never);
+
+				await node.execute.call(mockBulkExecuteFunctions('update', { params: { upsert } }));
+
+				expect(bulkWriteSpy).toHaveBeenCalledWith(
+					[
+						{
+							updateOne: {
+								filter: { id: '1' },
+								update: { $set: { id: '1', value: 'first' } },
+								upsert: true,
+							},
+						},
+						{
+							updateOne: {
+								filter: { id: '2' },
+								update: { $set: { id: '2', value: 'second' } },
+								upsert: true,
+							},
+						},
+						{
+							updateOne: {
+								filter: { id: '3' },
+								update: { $set: { id: '3', value: 'third' } },
+								upsert: true,
+							},
+						},
+					],
+					{ ordered: true },
+				);
+			},
+		);
+
+		it('filters by ObjectId and strips _id from the update when the update key is _id', async () => {
+			const bulkWriteSpy = vi
+				.spyOn(Collection.prototype, 'bulkWrite')
+				.mockResolvedValue({} as never);
+			const documentId = '662a2b1a2f8b9c0d1e2f3a4b';
+			const executeFunctions = mockBulkExecuteFunctions('update', {
+				params: { updateKey: '_id', fields: '_id,value' },
+			});
+			executeFunctions.getInputData.mockReturnValue([
+				{ json: { _id: documentId, value: 'renamed' } },
+			]);
+
+			const [items] = await node.execute.call(executeFunctions);
+
+			expect(bulkWriteSpy).toHaveBeenCalledWith(
+				[
+					{
+						updateOne: {
+							filter: { _id: new ObjectId(documentId) },
+							update: { $set: { value: 'renamed' } },
+						},
+					},
+				],
+				{ ordered: true },
+			);
+			expect(items).toEqual([{ json: { value: 'renamed' }, pairedItem: { item: 0 } }]);
+		});
+
+		it('uses an unordered bulkWrite and maps write errors to items when continue-on-fail is on', async () => {
+			const bulkWriteSpy = vi
+				.spyOn(Collection.prototype, 'bulkWrite')
+				.mockRejectedValue(bulkWriteError([{ index: 1, errmsg: 'E11000 duplicate key' }]));
+
+			const [items] = await node.execute.call(
+				mockBulkExecuteFunctions('update', { continueOnFail: true }),
+			);
+
+			expect(bulkWriteSpy).toHaveBeenCalledWith(expect.any(Array), { ordered: false });
+			expect(items).toEqual([
+				{ json: { id: '1', value: 'first' }, pairedItem: { item: 0 } },
+				{ json: { error: 'E11000 duplicate key' }, pairedItem: { item: 1 } },
+				{ json: { id: '3', value: 'third' }, pairedItem: { item: 2 } },
+			]);
+		});
+
+		it('maps write errors by op position when a prepare failure shifts the indexes', async () => {
+			const bulkWriteSpy = vi
+				.spyOn(Collection.prototype, 'bulkWrite')
+				.mockRejectedValue(bulkWriteError([{ index: 1, errmsg: 'E11000 duplicate key' }]));
+			const executeFunctions = mockBulkExecuteFunctions('update', { continueOnFail: true });
+			executeFunctions.getInputData.mockReturnValue([
+				inputItems[0],
+				{ json: { value: 'missing-key' } },
+				inputItems[2],
+			]);
+
+			const [items] = await node.execute.call(executeFunctions);
+
+			// Item 1 never reached the bulkWrite, so write-error index 1 is original item 2
+			expect(bulkWriteSpy).toHaveBeenCalledWith(
+				[
+					{ updateOne: { filter: { id: '1' }, update: { $set: { id: '1', value: 'first' } } } },
+					{ updateOne: { filter: { id: '3' }, update: { $set: { id: '3', value: 'third' } } } },
+				],
+				{ ordered: false },
+			);
+			expect(items).toEqual([
+				{ json: { id: '1', value: 'first' }, pairedItem: { item: 0 } },
+				{ json: { error: 'Item is missing the updateKey field' }, pairedItem: { item: 1 } },
+				{ json: { error: 'E11000 duplicate key' }, pairedItem: { item: 2 } },
+			]);
+		});
+
+		it('fails the whole group when the error carries no per-operation verdicts', async () => {
+			vi.spyOn(Collection.prototype, 'bulkWrite').mockRejectedValue(new Error('connection lost'));
+
+			const [items] = await node.execute.call(
+				mockBulkExecuteFunctions('update', { continueOnFail: true }),
+			);
+
+			expect(items).toEqual([
+				{ json: { error: 'connection lost' }, pairedItem: { item: 0 } },
+				{ json: { error: 'connection lost' }, pairedItem: { item: 1 } },
+				{ json: { error: 'connection lost' }, pairedItem: { item: 2 } },
+			]);
+		});
+
+		it('throws the bulk failure when continue-on-fail is off', async () => {
+			vi.spyOn(Collection.prototype, 'bulkWrite').mockRejectedValue(new Error('boom'));
+
+			await expect(node.execute.call(mockBulkExecuteFunctions('update'))).rejects.toThrow('boom');
+		});
+
+		it.each([
+			['update', 'updateOne'],
+			['findOneAndUpdate', 'findOneAndUpdate'],
+		] as const)('keeps per-item %s calls in version 1.4', async (operation, driverMethod) => {
+			const driverSpy = vi.spyOn(Collection.prototype, driverMethod).mockResolvedValue({} as never);
+			const bulkWriteSpy = vi.spyOn(Collection.prototype, 'bulkWrite');
+
+			await node.execute.call(mockExecuteFunctions(1.4, operation));
+
+			expect(driverSpy).toHaveBeenCalledTimes(3);
+			expect(bulkWriteSpy).not.toHaveBeenCalled();
+		});
+	});
+
 	describe('document operations in version 1.3', () => {
 		let collectionSpy: MockInstance;
 		const node = new MongoDb();