diff --git a/packages/nodes-base/nodes/MongoDb/MongoDb.node.ts b/packages/nodes-base/nodes/MongoDb/MongoDb.node.ts index ed059d3d9d9..9d87f75c9ca 100644 --- a/packages/nodes-base/nodes/MongoDb/MongoDb.node.ts +++ b/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 { + const continueOnFail = ctx.continueOnFail(); + const returnData: INodeExecutionData[] = []; + const groups = new Map(); + + 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, so normalise to an array. + const failedOpIndexes = new Map(); + 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( diff --git a/packages/nodes-base/nodes/MongoDb/test/MongoDB.test.ts b/packages/nodes-base/nodes/MongoDb/test/MongoDB.test.ts index 41f6a71f4d1..df5c6ccdf45 100644 --- a/packages/nodes-base/nodes/MongoDb/test/MongoDB.test.ts +++ b/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();