github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-19 16:55:02 +02:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity Graph

⚡ Add support for mode "latest"

Browse Source
This commit is contained in:
Jan Oberhauser 2022-10-24 23:39:15 +02:00
parent 4c3c5b258e
commit d91fb1ba69
9 changed files with 405 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
packages/cli/src/Interfaces.ts
View File

@ -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;
}

127
packages/cli/src/ProcessedDataManagers/NativeDatabase.ts
View File

@ -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<ICheckProcessedOutput> {
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<ICheckProcessedOutput> {
@ -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<void> {
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);
}
});

7
packages/cli/src/databases/entities/ProcessedData.ts
View File

@ -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;

214
packages/cli/test/integration/ProcessedDataManagers/NativeDatabase.test.ts
View File

@ -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);
}
});
}
});

30
packages/core/src/Interfaces.ts
View File

@ -33,19 +33,19 @@ export interface IProcessMessage {
export interface IExecuteFunctions extends IExecuteFunctionsBase {
helpers: {
checkProcessed(items: string[], context: IProcessedDataContext): Promise<ICheckProcessedOutput>;
checkProcessed(items: string[], context: ProcessedDataContext): Promise<ICheckProcessedOutput>;
checkProcessedAndRecord(
items: string[],
context: IProcessedDataContext,
context: ProcessedDataContext,
options: ICheckProcessedOptions,
): Promise<ICheckProcessedOutput>;
checkProcessedItemsAndRecord(
propertyName: string,
items: IDataObject[],
context: IProcessedDataContext,
context: ProcessedDataContext,
options: ICheckProcessedOptions,
): Promise<ICheckProcessedOutputItems>;
removeProcessed(items: string[], context: IProcessedDataContext): Promise<void>;
removeProcessed(items: string[], context: ProcessedDataContext): Promise<void>;
httpRequest(requestOptions: IHttpRequestOptions): Promise<any>; // tslint:disable-line:no-any
prepareBinaryData(
binaryData: Buffer,
@ -332,7 +332,9 @@ export interface IBinaryDataManager {
persistBinaryDataForExecutionId(executionId: string): Promise<void>;
}
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<void>;
checkProcessed(
items: string[],
context: IProcessedDataContext,
items: ProcessedDataItemTypes[],
context: ProcessedDataContext,
contextData: ICheckProcessedContextData,
options: ICheckProcessedOptions,
): Promise<ICheckProcessedOutput>;
checkProcessedAndRecord(
items: string[],
context: IProcessedDataContext,
items: ProcessedDataItemTypes[],
context: ProcessedDataContext,
contextData: ICheckProcessedContextData,
options: ICheckProcessedOptions,
): Promise<ICheckProcessedOutput>;
removeProcessed(
items: string[],
context: IProcessedDataContext,
items: ProcessedDataItemTypes[],
context: ProcessedDataContext,
contextData: ICheckProcessedContextData,
options: ICheckProcessedOptions,
): Promise<void>;
}
@ -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;
}

54
packages/core/src/NodeExecuteFunctions.ts
View File

@ -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<ICheckProcessedOutput> {
@ -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<ICheckProcessedOutput> {
@ -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<ICheckProcessedOutputItems> {
@ -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<void> {
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<ICheckProcessedOutput> {
return checkProcessed(items, context, { node, workflow }, options);
},
async checkProcessedAndRecord(
items: string[],
context: IProcessedDataContext,
items: ProcessedDataItemTypes[],
context: ProcessedDataContext,
options: ICheckProcessedOptions,
): Promise<ICheckProcessedOutput> {
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<ICheckProcessedOutputItems> {
return checkProcessedItemsAndRecord(
@ -2445,8 +2447,12 @@ export function getExecuteFunctions(
options,
);
},
async removeProcessed(items: string[], context: IProcessedDataContext): Promise<void> {
return removeProcessed(items, context, { node, workflow });
async removeProcessed(
items: ProcessedDataItemTypes[],
context: ProcessedDataContext,
options: ICheckProcessedOptions,
): Promise<void> {
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<ICheckProcessedOutput> {
return checkProcessed(items, context, { node, workflow }, options);
},
async checkProcessedAndRecord(
items: string[],
context: IProcessedDataContext,
items: ProcessedDataItemTypes[],
context: ProcessedDataContext,
options: ICheckProcessedOptions,
): Promise<ICheckProcessedOutput> {
return checkProcessedAndRecord(items, context, { node, workflow }, options);
},
async removeProcessed(items: string[], context: IProcessedDataContext): Promise<void> {
return removeProcessed(items, context, { node, workflow });
async removeProcessed(
items: string[],
context: ProcessedDataContext,
options: ICheckProcessedOptions,
): Promise<void> {
return removeProcessed(items, context, { node, workflow }, options);
},
request: proxyRequestToAxios,
async requestOAuth2(

20
packages/core/src/ProcessedDataManager.ts
View File

@ -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<ICheckProcessedOutput> {
@ -59,7 +60,7 @@ export class ProcessedDataManager {
async checkProcessedItemsAndRecord(
propertyName: string,
items: IDataObject[],
context: IProcessedDataContext,
context: ProcessedDataContext,
contextData: ICheckProcessedContextData,
options: ICheckProcessedOptions,
): Promise<ICheckProcessedOutputItems> {
@ -88,8 +89,8 @@ export class ProcessedDataManager {
}
async checkProcessedAndRecord(
items: string[],
context: IProcessedDataContext,
items: ProcessedDataItemTypes[],
context: ProcessedDataContext,
contextData: ICheckProcessedContextData,
options: ICheckProcessedOptions,
): Promise<ICheckProcessedOutput> {
@ -101,12 +102,13 @@ export class ProcessedDataManager {
}
async removeProcessed(
items: string[],
context: IProcessedDataContext,
items: ProcessedDataItemTypes[],
context: ProcessedDataContext,
contextData: ICheckProcessedContextData,
options: ICheckProcessedOptions,
): Promise<void> {
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}"`);

35
packages/nodes-base/nodes/CheckProcessed/CheckProcessed.node.ts
View File

@ -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<INodeExecutionData[][]> {
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);

1
packages/nodes-base/nodes/RssFeedRead/RssFeedRead.node.ts
View File

@ -93,6 +93,7 @@ export class RssFeedRead implements INodeType {
returnData = (
await this.helpers.checkProcessedItemsAndRecord('title', returnData, 'node', {
mode: 'entries',
maxEntries: 100,
})
).new;