From d91fb1ba69f80dbfa597a67cb20b1fdf57bc4c65 Mon Sep 17 00:00:00 2001 From: Jan Oberhauser Date: Mon, 24 Oct 2022 23:39:15 +0200 Subject: [PATCH] :zap: Add support for mode "latest" --- packages/cli/src/Interfaces.ts | 14 +- .../ProcessedDataManagers/NativeDatabase.ts | 127 +++++++++-- .../src/databases/entities/ProcessedData.ts | 7 +- .../NativeDatabase.test.ts | 214 ++++++++++++++++-- packages/core/src/Interfaces.ts | 30 +-- packages/core/src/NodeExecuteFunctions.ts | 54 +++-- packages/core/src/ProcessedDataManager.ts | 20 +- .../CheckProcessed/CheckProcessed.node.ts | 35 ++- .../nodes/RssFeedRead/RssFeedRead.node.ts | 1 + 9 files changed, 405 insertions(+), 97 deletions(-) diff --git a/packages/cli/src/Interfaces.ts b/packages/cli/src/Interfaces.ts index b8a8ab58a5f..9bbc3b483c4 100644 --- a/packages/cli/src/Interfaces.ts +++ b/packages/cli/src/Interfaces.ts @@ -21,7 +21,7 @@ import { WorkflowExecuteMode, } from 'n8n-workflow'; -import { WorkflowExecute } from 'n8n-core'; +import { ProcessedDataMode, ProcessedDataItemTypes, WorkflowExecute } from 'n8n-core'; // eslint-disable-next-line import/no-extraneous-dependencies import PCancelable from 'p-cancelable'; @@ -101,10 +101,20 @@ export interface IWebhookDb { // ProcessedData // ---------------------------------- +export interface IProcessedDataLatest { + mode: ProcessedDataMode; + data: ProcessedDataItemTypes; +} + +export interface IProcessedDataEntries { + mode: ProcessedDataMode; + data: ProcessedDataItemTypes[]; +} + export interface IProcessedDataDb { createdAt: Date; context: string; - value: IDataObject; + value: IProcessedDataEntries | IProcessedDataLatest; workflowId: string; } diff --git a/packages/cli/src/ProcessedDataManagers/NativeDatabase.ts b/packages/cli/src/ProcessedDataManagers/NativeDatabase.ts index 7f6240fca38..c1837b3edb2 100644 --- a/packages/cli/src/ProcessedDataManagers/NativeDatabase.ts +++ b/packages/cli/src/ProcessedDataManagers/NativeDatabase.ts @@ -1,22 +1,21 @@ -/* eslint-disable @typescript-eslint/no-unsafe-call */ -/* eslint-disable @typescript-eslint/no-unsafe-member-access */ import { ICheckProcessedContextData, ICheckProcessedOptions, ICheckProcessedOutput, - IProcessedDataContext, IProcessedDataManager, + ProcessedDataContext, + ProcessedDataItemTypes, } from 'n8n-core'; import { In, Not } from 'typeorm'; import { createHash } from 'crypto'; -// eslint-disable-next-line import/no-cycle import { DatabaseType, Db, ExternalHooks, GenericHelpers, IExternalHooksFileData, + IProcessedDataEntries, IWorkflowBase, } from '..'; @@ -65,8 +64,12 @@ export class ProcessedDataManagerNativeDatabase implements IProcessedDataManager externalHooks.loadHooks(hookFileData); } + private static sortEntries(items: ProcessedDataItemTypes[]) { + return [...items].sort(); + } + private static createContext( - context: IProcessedDataContext, + context: ProcessedDataContext, contextData: ICheckProcessedContextData, ): string { if (context === 'node') { @@ -81,21 +84,21 @@ export class ProcessedDataManagerNativeDatabase implements IProcessedDataManager return ''; } - private static createValueHash(value: string): string { - return createHash('md5').update(value).digest('base64'); + private static createValueHash(value: ProcessedDataItemTypes): string { + return createHash('md5').update(value.toString()).digest('base64'); } async checkProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, + options: ICheckProcessedOptions, ): Promise { const returnData: ICheckProcessedOutput = { new: [], processed: [], }; - // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment const processedData = await Db.collections.ProcessedData.findOne({ where: { workflowId: contextData.workflow.id as string, @@ -103,18 +106,37 @@ export class ProcessedDataManagerNativeDatabase implements IProcessedDataManager }, }); + if (processedData && processedData.value.mode !== options.mode) { + throw new Error( + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + `Mode is not compatible. Data got originally used with mode "${processedData?.value.mode}" but gets queried with mode "${options.mode}"`, + ); + } + if (!processedData) { // If there is nothing it the database all items are new returnData.new = items; return returnData; } + if (options.mode === 'latest') { + const incomingItems = ProcessedDataManagerNativeDatabase.sortEntries(items); + incomingItems.forEach((item) => { + if (item <= processedData.value.data) { + returnData.processed.push(item); + } else { + returnData.new.push(item); + } + }); + return returnData; + } + const hashedItems = items.map((item) => ProcessedDataManagerNativeDatabase.createValueHash(item), ); hashedItems.forEach((item, index) => { - if ((processedData.value.data as string[]).find((data) => data === item)) { + if ((processedData.value.data as string[]).find((entry) => entry === item)) { returnData.processed.push(items[index]); } else { returnData.new.push(items[index]); @@ -125,8 +147,8 @@ export class ProcessedDataManagerNativeDatabase implements IProcessedDataManager } async checkProcessedAndRecord( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, options: ICheckProcessedOptions, ): Promise { @@ -143,6 +165,58 @@ export class ProcessedDataManagerNativeDatabase implements IProcessedDataManager }, }); + if (processedData && processedData.value.mode !== options.mode) { + throw new Error( + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + `Mode is not compatible. Data got originally used with mode "${processedData?.value.mode}" but gets queried with mode "${options.mode}"`, + ); + } + + if (options.mode === 'latest') { + const incomingItems = ProcessedDataManagerNativeDatabase.sortEntries(items); + + if (!processedData) { + // All items are new so add new entries + await Db.collections.ProcessedData.insert({ + workflowId: contextData.workflow.id.toString(), + context: dbContext, + value: { + mode: options.mode, + data: incomingItems.pop(), + }, + }); + + return { + new: items, + processed: [], + }; + } + + const returnData: ICheckProcessedOutput = { + new: [], + processed: [], + }; + + let largestValue = processedData.value.data; + + incomingItems.forEach((item) => { + if (item <= processedData.value.data) { + returnData.processed.push(item); + } else { + returnData.new.push(item); + if (item > largestValue) { + largestValue = item; + } + } + }); + + processedData.value.data = largestValue; + + await Db.collections.ProcessedData.save(processedData); + + return returnData; + } + const hashedItems = items.map((item) => ProcessedDataManagerNativeDatabase.createValueHash(item), ); @@ -156,6 +230,7 @@ export class ProcessedDataManagerNativeDatabase implements IProcessedDataManager workflowId: contextData.workflow.id.toString(), context: dbContext, value: { + mode: options.mode, data: hashedItems, }, }); @@ -171,20 +246,19 @@ export class ProcessedDataManagerNativeDatabase implements IProcessedDataManager processed: [], }; + const processedDataValue = processedData.value as IProcessedDataEntries; + hashedItems.forEach((item, index) => { - if ((processedData.value.data as string[]).find((data) => data === item)) { + if (processedDataValue.data.find((entry) => entry === item)) { returnData.processed.push(items[index]); } else { returnData.new.push(items[index]); - (processedData.value.data as string[]).push(item); + processedDataValue.data.push(item); } }); if (options.maxEntries) { - (processedData.value.data as string[]).splice( - 0, - (processedData.value.data as string[]).length - options.maxEntries, - ); + processedDataValue.data.splice(0, processedDataValue.data.length - options.maxEntries); } await Db.collections.ProcessedData.save(processedData); @@ -193,10 +267,15 @@ export class ProcessedDataManagerNativeDatabase implements IProcessedDataManager } async removeProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, + options: ICheckProcessedOptions, ): Promise { + if (options.mode === 'latest') { + throw new Error('Removing processed data is not possible in mode "latest"'); + } + const processedData = await Db.collections.ProcessedData.findOne({ where: { workflowId: contextData.workflow.id as string, @@ -212,11 +291,13 @@ export class ProcessedDataManagerNativeDatabase implements IProcessedDataManager ProcessedDataManagerNativeDatabase.createValueHash(item), ); + const processedDataValue = processedData.value as IProcessedDataEntries; + let index; hashedItems.forEach((item) => { - index = (processedData.value.data as string[]).findIndex((value) => value === item); + index = processedDataValue.data.findIndex((value) => value === item); if (index !== -1) { - (processedData.value.data as string[]).splice(index, 1); + processedDataValue.data.splice(index, 1); } }); diff --git a/packages/cli/src/databases/entities/ProcessedData.ts b/packages/cli/src/databases/entities/ProcessedData.ts index 176fb8357cc..feaae0e4e55 100644 --- a/packages/cli/src/databases/entities/ProcessedData.ts +++ b/packages/cli/src/databases/entities/ProcessedData.ts @@ -2,10 +2,9 @@ import { Column, CreateDateColumn, Entity, Index, PrimaryColumn } from 'typeorm' import * as config from '../../../config'; // eslint-disable-next-line import/no-cycle -import { DatabaseType, IProcessedDataDb } from '../..'; +import { DatabaseType, IProcessedDataDb, IProcessedDataEntries, IProcessedDataLatest } from '../..'; import { jsonColumnType } from './AbstractEntity'; import { objectRetriever } from '../utils/transformers'; -import { IDataObject } from 'n8n-workflow'; function getTimestampSyntax() { const dbType = config.getEnv('database.type'); @@ -21,7 +20,7 @@ function getTimestampSyntax() { } @Entity() -@Index(['workflowId', 'context', 'value'], { unique: true }) +@Index(['workflowId', 'context'], { unique: true }) export class ProcessedData implements IProcessedDataDb { // @Column('varchar') @PrimaryColumn('varchar') @@ -36,7 +35,7 @@ export class ProcessedData implements IProcessedDataDb { nullable: true, transformer: objectRetriever, }) - value: IDataObject; + value: IProcessedDataEntries | IProcessedDataLatest; @CreateDateColumn({ precision: 3, default: () => getTimestampSyntax() }) createdAt: Date; diff --git a/packages/cli/test/integration/ProcessedDataManagers/NativeDatabase.test.ts b/packages/cli/test/integration/ProcessedDataManagers/NativeDatabase.test.ts index 0a9384f3438..88712bcbb1f 100644 --- a/packages/cli/test/integration/ProcessedDataManagers/NativeDatabase.test.ts +++ b/packages/cli/test/integration/ProcessedDataManagers/NativeDatabase.test.ts @@ -1,4 +1,9 @@ -import { ICheckProcessedContextData, ICheckProcessedOutput, ProcessedDataManager } from 'n8n-core'; +import { + ICheckProcessedContextData, + ICheckProcessedOutput, + ProcessedDataManager, + ProcessedDataItemTypes, +} from 'n8n-core'; import { INode, Workflow } from 'n8n-workflow'; import * as Helpers from '../../unit/Helpers'; @@ -55,7 +60,7 @@ afterAll(async () => { await testDb.terminate(testDbName); }); -test('ProcessedData: NativeDatabase should record and check data correctly', async () => { +test('ProcessedData (mode: entries): NativeDatabase should record and check data correctly', async () => { const context = 'node'; const contextData: ICheckProcessedContextData = { workflow, @@ -68,7 +73,7 @@ test('ProcessedData: NativeDatabase should record and check data correctly', asy ['a', 'b'], context, contextData, - {}, + { mode: 'entries' }, ); // No data exists yet so has to be new @@ -78,7 +83,7 @@ test('ProcessedData: NativeDatabase should record and check data correctly', asy ['a', 'b'], context, contextData, - {}, + { mode: 'entries' }, ); // 'a' & 'b' got only checked before, so still has to be new @@ -88,7 +93,7 @@ test('ProcessedData: NativeDatabase should record and check data correctly', asy ['a', 'b', 'c'], context, contextData, - {}, + { mode: 'entries' }, ); // 'a' & 'b' got recorded before, 'c' never @@ -98,26 +103,28 @@ test('ProcessedData: NativeDatabase should record and check data correctly', asy ['a', 'b', 'c', 'd'], context, contextData, - {}, + { mode: 'entries' }, ); // 'a' & 'b' got recorded before, 'c' only checked bfeore and 'd' has never been seen expect(processedData).toEqual({ new: ['c', 'd'], processed: ['a', 'b'] }); - await ProcessedDataManager.getInstance().removeProcessed(['b', 'd'], context, contextData); + await ProcessedDataManager.getInstance().removeProcessed(['b', 'd'], context, contextData, { + mode: 'entries', + }); processedData = await ProcessedDataManager.getInstance().checkProcessed( ['a', 'b', 'c', 'd'], context, contextData, - {}, + { mode: 'entries' }, ); // 'b' & 'd' got removed from the database so they should be new, 'a' & 'b' should still be known expect(processedData).toEqual({ new: ['b', 'd'], processed: ['a', 'c'] }); }); -test('ProcessedData: NativeDatabase different contexts should not interfere with each other', async () => { +test('ProcessedData (mode: entries): NativeDatabase different contexts should not interfere with each other', async () => { const contextData: ICheckProcessedContextData = { workflow, node, @@ -130,7 +137,7 @@ test('ProcessedData: NativeDatabase different contexts should not interfere with ['a', 'b'], 'node', contextData, - {}, + { mode: 'entries' }, ); // No data exists yet for context "node" so has to be new @@ -141,31 +148,35 @@ test('ProcessedData: NativeDatabase different contexts should not interfere with ['a', 'b', 'c'], 'workflow', contextData, - {}, + { mode: 'entries' }, ); // No data exists yet for context 'worklow' so has to be new expect(processedData).toEqual({ new: ['a', 'b', 'c'], processed: [] }); - await ProcessedDataManager.getInstance().removeProcessed(['a'], 'node', contextData); + await ProcessedDataManager.getInstance().removeProcessed(['a'], 'node', contextData, { + mode: 'entries', + }); processedData = await ProcessedDataManager.getInstance().checkProcessed( ['a', 'b', 'c'], 'node', contextData, - {}, + { mode: 'entries' }, ); // 'a' got removed for the context 'node' and 'c' never got saved, so only 'b' should be known expect(processedData).toEqual({ new: ['a', 'c'], processed: ['b'] }); - await ProcessedDataManager.getInstance().removeProcessed(['b'], 'workflow', contextData); + await ProcessedDataManager.getInstance().removeProcessed(['b'], 'workflow', contextData, { + mode: 'entries', + }); processedData = await ProcessedDataManager.getInstance().checkProcessed( ['a', 'b', 'c', 'd'], 'workflow', contextData, - {}, + { mode: 'entries' }, ); // 'b' got removed for the context 'workflow' and 'd' never got saved for that reason new @@ -173,7 +184,7 @@ test('ProcessedData: NativeDatabase different contexts should not interfere with expect(processedData).toEqual({ new: ['b', 'd'], processed: ['a', 'c'] }); }); -test('ProcessedData: NativeDatabase check maxEntries', async () => { +test('ProcessedData (mode: entries): NativeDatabase check maxEntries', async () => { const contextData: ICheckProcessedContextData = { workflow, node, @@ -185,7 +196,7 @@ test('ProcessedData: NativeDatabase check maxEntries', async () => { ['0', '1', '2', '3'], 'node', contextData, - { maxEntries: 5 }, + { mode: 'entries', maxEntries: 5 }, ); // All data should be new @@ -196,22 +207,185 @@ test('ProcessedData: NativeDatabase check maxEntries', async () => { ['4', '5', '6'], 'node', contextData, - { maxEntries: 5 }, + { mode: 'entries', maxEntries: 5 }, ); // All given data should be new expect(processedData).toEqual({ new: ['4', '5', '6'], processed: [] }); // This should not make a difference, removing an item which does not exist - await ProcessedDataManager.getInstance().removeProcessed(['a'], 'node', contextData); + await ProcessedDataManager.getInstance().removeProcessed(['a'], 'node', contextData, { + mode: 'entries', + }); processedData = await ProcessedDataManager.getInstance().checkProcessed( ['0', '1', '2', '3', '4', '5', '6', '7'], 'node', contextData, - { maxEntries: 5 }, + { mode: 'entries', maxEntries: 5 }, ); // '7' should be new and '0' and '1' also because they got been pruned as max 5 get saved expect(processedData).toEqual({ new: ['0', '1', '7'], processed: ['2', '3', '4', '5', '6'] }); }); + +describe('ProcessedData (mode: latest): NativeDatabase should record and check data correctly', () => { + const tests: Array<{ + description: string; + data: Array<{ + operation: 'checkProcessed' | 'checkProcessedAndRecord'; + input: ProcessedDataItemTypes[]; + output: ICheckProcessedOutput; + }>; + }> = [ + { + description: 'dates', + data: [ + { + operation: 'checkProcessed', + input: [new Date('2022-01-02').toISOString(), new Date('2022-01-03').toISOString()], + output: { + new: [new Date('2022-01-02').toISOString(), new Date('2022-01-03').toISOString()], + processed: [], + }, + }, + { + operation: 'checkProcessedAndRecord', + input: [new Date('2022-01-02').toISOString(), new Date('2022-01-03').toISOString()], + output: { + new: [new Date('2022-01-02').toISOString(), new Date('2022-01-03').toISOString()], + processed: [], + }, + }, + { + operation: 'checkProcessed', + input: [ + new Date('2022-01-02').toISOString(), + new Date('2022-01-03').toISOString(), + new Date('2022-01-04').toISOString(), + ], + output: { + new: [new Date('2022-01-04').toISOString()], + processed: [new Date('2022-01-02').toISOString(), new Date('2022-01-03').toISOString()], + }, + }, + { + operation: 'checkProcessedAndRecord', + input: [ + new Date('2022-01-02').toISOString(), + new Date('2022-01-03').toISOString(), + new Date('2022-01-04').toISOString(), + new Date('2022-01-05').toISOString(), + ], + output: { + new: [new Date('2022-01-04').toISOString(), new Date('2022-01-05').toISOString()], + processed: [new Date('2022-01-02').toISOString(), new Date('2022-01-03').toISOString()], + }, + }, + { + operation: 'checkProcessed', + input: [ + new Date('2022-01-01').toISOString(), + new Date('2022-01-02').toISOString(), + new Date('2022-01-03').toISOString(), + new Date('2022-01-04').toISOString(), + new Date('2022-01-06').toISOString(), + new Date('2022-01-07').toISOString(), + ], + output: { + new: [new Date('2022-01-06').toISOString(), new Date('2022-01-07').toISOString()], + processed: [ + new Date('2022-01-01').toISOString(), + new Date('2022-01-02').toISOString(), + new Date('2022-01-03').toISOString(), + new Date('2022-01-04').toISOString(), + ], + }, + }, + ], + }, + { + description: 'letters', + data: [ + { + operation: 'checkProcessed', + input: ['b', 'c'], + output: { new: ['b', 'c'], processed: [] }, + }, + { + operation: 'checkProcessedAndRecord', + input: ['b', 'c'], + output: { new: ['b', 'c'], processed: [] }, + }, + { + operation: 'checkProcessed', + input: ['b', 'c', 'd'], + output: { new: ['d'], processed: ['b', 'c'] }, + }, + { + operation: 'checkProcessedAndRecord', + input: ['b', 'c', 'd', 'e'], + output: { new: ['d', 'e'], processed: ['b', 'c'] }, + }, + { + operation: 'checkProcessed', + input: ['a', 'b', 'c', 'd', 'f', 'g'], + output: { new: ['f', 'g'], processed: ['a', 'b', 'c', 'd'] }, + }, + ], + }, + { + description: 'numbers', + data: [ + { + operation: 'checkProcessed', + input: [2, 3], + output: { new: [2, 3], processed: [] }, + }, + { + operation: 'checkProcessedAndRecord', + input: [2, 3], + output: { new: [2, 3], processed: [] }, + }, + { + operation: 'checkProcessed', + input: [2, 3, 4], + output: { new: [4], processed: [2, 3] }, + }, + { + operation: 'checkProcessedAndRecord', + input: [2, 3, 4, 5], + output: { new: [4, 5], processed: [2, 3] }, + }, + { + operation: 'checkProcessed', + input: [1, 2, 3, 4, 6, 7], + output: { new: [6, 7], processed: [1, 2, 3, 4] }, + }, + ], + }, + ]; + + for (const testData of tests) { + test(testData.description, async () => { + const context = 'node'; + const contextData: ICheckProcessedContextData = { + workflow, + node, + }; + + let processedData: ICheckProcessedOutput; + + for (const data of testData.data) { + processedData = await ProcessedDataManager.getInstance()[data.operation]( + data.input, + context, + contextData, + { mode: 'latest' }, + ); + + expect(processedData).toEqual(data.output); + } + }); + } +}); diff --git a/packages/core/src/Interfaces.ts b/packages/core/src/Interfaces.ts index f24a7ad4a8f..a12015731ce 100644 --- a/packages/core/src/Interfaces.ts +++ b/packages/core/src/Interfaces.ts @@ -33,19 +33,19 @@ export interface IProcessMessage { export interface IExecuteFunctions extends IExecuteFunctionsBase { helpers: { - checkProcessed(items: string[], context: IProcessedDataContext): Promise; + checkProcessed(items: string[], context: ProcessedDataContext): Promise; checkProcessedAndRecord( items: string[], - context: IProcessedDataContext, + context: ProcessedDataContext, options: ICheckProcessedOptions, ): Promise; checkProcessedItemsAndRecord( propertyName: string, items: IDataObject[], - context: IProcessedDataContext, + context: ProcessedDataContext, options: ICheckProcessedOptions, ): Promise; - removeProcessed(items: string[], context: IProcessedDataContext): Promise; + removeProcessed(items: string[], context: ProcessedDataContext): Promise; httpRequest(requestOptions: IHttpRequestOptions): Promise; // tslint:disable-line:no-any prepareBinaryData( binaryData: Buffer, @@ -332,7 +332,9 @@ export interface IBinaryDataManager { persistBinaryDataForExecutionId(executionId: string): Promise; } -export type IProcessedDataContext = 'node' | 'workflow'; +export type ProcessedDataContext = 'node' | 'workflow'; +export type ProcessedDataMode = 'entries' | 'latest'; +export type ProcessedDataItemTypes = string | number; export interface IProcessedDataConfig { availableModes: string; @@ -342,23 +344,24 @@ export interface IProcessedDataConfig { export interface IProcessedDataManager { init(): Promise; checkProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, options: ICheckProcessedOptions, ): Promise; checkProcessedAndRecord( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, options: ICheckProcessedOptions, ): Promise; removeProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, + options: ICheckProcessedOptions, ): Promise; } @@ -367,8 +370,8 @@ export interface IProcessedDataManagers { } export interface ICheckProcessedOutput { - new: string[]; - processed: string[]; + new: ProcessedDataItemTypes[]; + processed: ProcessedDataItemTypes[]; } export interface ICheckProcessedOutputItems { @@ -385,5 +388,6 @@ export interface ICheckProcessedContextData { } export interface ICheckProcessedOptions { + mode: ProcessedDataMode; maxEntries?: number; } diff --git a/packages/core/src/NodeExecuteFunctions.ts b/packages/core/src/NodeExecuteFunctions.ts index 9713ac39be6..3b7b093ece5 100644 --- a/packages/core/src/NodeExecuteFunctions.ts +++ b/packages/core/src/NodeExecuteFunctions.ts @@ -100,12 +100,13 @@ import { ICredentialTestFunctions, IHookFunctions, ILoadOptionsFunctions, - IProcessedDataContext, IResponseError, IWorkflowSettings, ProcessedDataManager, PLACEHOLDER_EMPTY_EXECUTION_ID, ICheckProcessedOptions, + ProcessedDataContext, + ProcessedDataItemTypes, } from '.'; import { extractValue } from './ExtractValue'; import { getClientCredentialsToken } from './OAuth2Helper'; @@ -825,8 +826,8 @@ export async function getBinaryDataBuffer( } export async function checkProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, options: ICheckProcessedOptions, ): Promise { @@ -834,8 +835,8 @@ export async function checkProcessed( } export async function checkProcessedAndRecord( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, options: ICheckProcessedOptions, ): Promise { @@ -850,7 +851,7 @@ export async function checkProcessedAndRecord( export async function checkProcessedItemsAndRecord( key: string, items: IDataObject[], - context: IProcessedDataContext, + context: ProcessedDataContext, contextData: ICheckProcessedContextData, options: ICheckProcessedOptions, ): Promise { @@ -864,11 +865,12 @@ export async function checkProcessedItemsAndRecord( } export async function removeProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, + options: ICheckProcessedOptions, ): Promise { - return ProcessedDataManager.getInstance().removeProcessed(items, context, contextData); + return ProcessedDataManager.getInstance().removeProcessed(items, context, contextData, options); } /** @@ -2418,15 +2420,15 @@ export function getExecuteFunctions( return getBinaryDataBuffer.call(this, inputData, itemIndex, propertyName, inputIndex); }, async checkProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, options: ICheckProcessedOptions, ): Promise { return checkProcessed(items, context, { node, workflow }, options); }, async checkProcessedAndRecord( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, options: ICheckProcessedOptions, ): Promise { return checkProcessedAndRecord(items, context, { node, workflow }, options); @@ -2434,7 +2436,7 @@ export function getExecuteFunctions( async checkProcessedItemsAndRecord( propertyName: string, items: IDataObject[], - context: IProcessedDataContext, + context: ProcessedDataContext, options: ICheckProcessedOptions, ): Promise { return checkProcessedItemsAndRecord( @@ -2445,8 +2447,12 @@ export function getExecuteFunctions( options, ); }, - async removeProcessed(items: string[], context: IProcessedDataContext): Promise { - return removeProcessed(items, context, { node, workflow }); + async removeProcessed( + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, + options: ICheckProcessedOptions, + ): Promise { + return removeProcessed(items, context, { node, workflow }, options); }, request: proxyRequestToAxios, async requestOAuth2( @@ -2674,21 +2680,25 @@ export function getExecuteSingleFunctions( ); }, async checkProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, options: ICheckProcessedOptions, ): Promise { return checkProcessed(items, context, { node, workflow }, options); }, async checkProcessedAndRecord( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, options: ICheckProcessedOptions, ): Promise { return checkProcessedAndRecord(items, context, { node, workflow }, options); }, - async removeProcessed(items: string[], context: IProcessedDataContext): Promise { - return removeProcessed(items, context, { node, workflow }); + async removeProcessed( + items: string[], + context: ProcessedDataContext, + options: ICheckProcessedOptions, + ): Promise { + return removeProcessed(items, context, { node, workflow }, options); }, request: proxyRequestToAxios, async requestOAuth2( diff --git a/packages/core/src/ProcessedDataManager.ts b/packages/core/src/ProcessedDataManager.ts index 3b4dc7884ed..e5b0ea13b13 100644 --- a/packages/core/src/ProcessedDataManager.ts +++ b/packages/core/src/ProcessedDataManager.ts @@ -6,8 +6,9 @@ import { ICheckProcessedOutput, ICheckProcessedOutputItems, IProcessedDataConfig, - IProcessedDataContext, IProcessedDataManagers, + ProcessedDataContext, + ProcessedDataItemTypes, } from './Interfaces'; export class ProcessedDataManager { @@ -44,8 +45,8 @@ export class ProcessedDataManager { } async checkProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, options: ICheckProcessedOptions, ): Promise { @@ -59,7 +60,7 @@ export class ProcessedDataManager { async checkProcessedItemsAndRecord( propertyName: string, items: IDataObject[], - context: IProcessedDataContext, + context: ProcessedDataContext, contextData: ICheckProcessedContextData, options: ICheckProcessedOptions, ): Promise { @@ -88,8 +89,8 @@ export class ProcessedDataManager { } async checkProcessedAndRecord( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, options: ICheckProcessedOptions, ): Promise { @@ -101,12 +102,13 @@ export class ProcessedDataManager { } async removeProcessed( - items: string[], - context: IProcessedDataContext, + items: ProcessedDataItemTypes[], + context: ProcessedDataContext, contextData: ICheckProcessedContextData, + options: ICheckProcessedOptions, ): Promise { if (this.managers[this.mode]) { - return this.managers[this.mode].removeProcessed(items, context, contextData); + return this.managers[this.mode].removeProcessed(items, context, contextData, options); } throw new Error(`There is no manager for the defined mode "${this.mode}"`); diff --git a/packages/nodes-base/nodes/CheckProcessed/CheckProcessed.node.ts b/packages/nodes-base/nodes/CheckProcessed/CheckProcessed.node.ts index 8157f60df08..fb5f38d68e6 100644 --- a/packages/nodes-base/nodes/CheckProcessed/CheckProcessed.node.ts +++ b/packages/nodes-base/nodes/CheckProcessed/CheckProcessed.node.ts @@ -1,4 +1,9 @@ -import { ICheckProcessedOutput, IExecuteFunctions, IProcessedDataContext } from 'n8n-core'; +import { + ICheckProcessedOutput, + IExecuteFunctions, + ProcessedDataContext, + ProcessedDataMode, +} from 'n8n-core'; import { INodeExecutionData, INodeType, @@ -45,6 +50,23 @@ export class CheckProcessed implements INodeType { ], default: 'filterOut', }, + { + displayName: 'Check Mode', + name: 'checkMode', + type: 'options', + noDataExpression: true, + options: [ + { + name: 'Entries', + value: 'entries', + }, + { + name: 'Latest', + value: 'latest', + }, + ], + default: 'latest', + }, { displayName: 'Check Value', name: 'checkValue', @@ -90,6 +112,7 @@ export class CheckProcessed implements INodeType { displayOptions: { show: { mode: ['add', 'filterOut'], + checkMode: ['entries'], }, }, default: 50, @@ -101,7 +124,8 @@ export class CheckProcessed implements INodeType { async execute(this: IExecuteFunctions): Promise { const items = this.getInputData(); const mode = this.getNodeParameter('mode', 0) as string; - const context = this.getNodeParameter('context', 0, 'workflow') as IProcessedDataContext; + const checkMode = this.getNodeParameter('checkMode', 0) as ProcessedDataMode; + const context = this.getNodeParameter('context', 0, 'workflow') as ProcessedDataContext; if (!['node', 'workflow'].includes(context)) { throw new NodeOperationError( @@ -132,7 +156,10 @@ export class CheckProcessed implements INodeType { if (mode === 'add') { const maxEntries = this.getNodeParameter('maxEntries', 0) as number; - await this.helpers.checkProcessedAndRecord(Object.keys(itemMapping), context, { maxEntries }); + await this.helpers.checkProcessedAndRecord(Object.keys(itemMapping), context, { + mode: checkMode, + maxEntries, + }); return [items]; } else if (mode === 'remove') { await this.helpers.removeProcessed(Object.keys(itemMapping), context); @@ -147,7 +174,7 @@ export class CheckProcessed implements INodeType { itemsProcessed = await this.helpers.checkProcessedAndRecord( Object.keys(itemMapping), context, - { maxEntries }, + { mode: checkMode, maxEntries }, ); } else { itemsProcessed = await this.helpers.checkProcessed(Object.keys(itemMapping), context); diff --git a/packages/nodes-base/nodes/RssFeedRead/RssFeedRead.node.ts b/packages/nodes-base/nodes/RssFeedRead/RssFeedRead.node.ts index 61716cef694..62c4a3c84d5 100644 --- a/packages/nodes-base/nodes/RssFeedRead/RssFeedRead.node.ts +++ b/packages/nodes-base/nodes/RssFeedRead/RssFeedRead.node.ts @@ -93,6 +93,7 @@ export class RssFeedRead implements INodeType { returnData = ( await this.helpers.checkProcessedItemsAndRecord('title', returnData, 'node', { + mode: 'entries', maxEntries: 100, }) ).new;