refactor(core): Generalize binary data manager interface (no-changelog)

packages/cli/src/ActiveExecutions.ts

@@ -12,6 +12,7 @@ import { createDeferredPromise, LoggerProxy } from 'n8n-workflow';
 import type { ChildProcess } from 'child_process';
 import type PCancelable from 'p-cancelable';
 import type {
+	ExecutionPayload,
 	IExecutingWorkflowData,
 	IExecutionDb,
 	IExecutionsCurrentSummary,
@@ -38,7 +39,7 @@ export class ActiveExecutions {
 		if (executionId === undefined) {
 			// Is a new execution so save in DB
 
-			const fullExecutionData: IExecutionDb = {
+			const fullExecutionData: ExecutionPayload = {
 				data: executionData.executionData!,
 				mode: executionData.executionMode,
 				finished: false,

packages/cli/src/GenericHelpers.ts

@@ -11,7 +11,7 @@ import { Container } from 'typedi';
 import { Like } from 'typeorm';
 import config from '@/config';
 import * as Db from '@/Db';
-import type { ICredentialsDb, IExecutionDb, IWorkflowDb } from '@/Interfaces';
+import type { ExecutionPayload, ICredentialsDb, IWorkflowDb } from '@/Interfaces';
 import * as ResponseHelper from '@/ResponseHelper';
 import type { WorkflowEntity } from '@db/entities/WorkflowEntity';
 import type { CredentialsEntity } from '@db/entities/CredentialsEntity';
@@ -178,7 +178,7 @@ export async function createErrorExecution(
 		},
 	};
 
-	const fullExecutionData: IExecutionDb = {
+	const fullExecutionData: ExecutionPayload = {
 		data: executionData,
 		mode,
 		finished: false,

packages/cli/src/Interfaces.ts

@@ -171,7 +171,7 @@ export type ICredentialsDecryptedResponse = ICredentialsDecryptedDb;
 export type SaveExecutionDataType = 'all' | 'none';
 
 export interface IExecutionBase {
-	id?: string;
+	id: string;
 	mode: WorkflowExecuteMode;
 	startedAt: Date;
 	stoppedAt?: Date; // empty value means execution is still running
@@ -189,6 +189,11 @@ export interface IExecutionDb extends IExecutionBase {
 	workflowData?: IWorkflowBase;
 }
 
+/**
+ * Payload for creating or updating an execution.
+ */
+export type ExecutionPayload = Omit<IExecutionDb, 'id'>;
+
 export interface IExecutionPushResponse {
 	executionId?: string;
 	waitingForWebhook?: boolean;

packages/cli/src/PublicApi/v1/handlers/executions/executions.handler.ts

@@ -1,20 +1,14 @@
 import type express from 'express';
 
-import { BinaryDataManager } from 'n8n-core';
-
-import {
-	getExecutions,
-	getExecutionInWorkflows,
-	deleteExecution,
-	getExecutionsCount,
-} from './executions.service';
+import { getExecutions, getExecutionInWorkflows, getExecutionsCount } from './executions.service';
 import { ActiveExecutions } from '@/ActiveExecutions';
 import { authorize, validCursor } from '../../shared/middlewares/global.middleware';
 import type { ExecutionRequest } from '../../../types';
 import { getSharedWorkflowIds } from '../workflows/workflows.service';
 import { encodeNextCursor } from '../../shared/services/pagination.service';
 import { Container } from 'typedi';
 import { InternalHooks } from '@/InternalHooks';
+import { ExecutionRepository } from '@/databases/repositories';
 
 export = {
 	deleteExecution: [
@@ -37,9 +31,7 @@ export = {
 				return res.status(404).json({ message: 'Not Found' });
 			}
 
-			await BinaryDataManager.getInstance().deleteBinaryDataByExecutionIds([execution.id!]);
-
-			await deleteExecution(execution);
+			await Container.get(ExecutionRepository).softDelete(execution.id);
 
 			execution.id = id;
 
@@ -111,7 +103,7 @@ export = {
 
 			const executions = await getExecutions(filters);
 
-			const newLastId = !executions.length ? '0' : (executions.slice(-1)[0].id as string);
+			const newLastId = !executions.length ? '0' : executions.slice(-1)[0].id;
 
 			filters.lastId = newLastId;
 

packages/cli/src/PublicApi/v1/handlers/executions/executions.service.ts

@@ -1,4 +1,4 @@
-import type { DeleteResult, FindOptionsWhere } from 'typeorm';
+import type { FindOptionsWhere } from 'typeorm';
 import { In, Not, Raw, LessThan } from 'typeorm';
 import { Container } from 'typedi';
 import type { ExecutionStatus } from 'n8n-workflow';
@@ -109,7 +109,3 @@ export async function getExecutionInWorkflows(
 		unflattenData: true,
 	});
 }
-
-export async function deleteExecution(execution: IExecutionBase): Promise<DeleteResult> {
-	return Container.get(ExecutionRepository).deleteExecution(execution.id as string);
-}

packages/cli/src/Server.ts

@@ -26,7 +26,7 @@ import type { RequestOptions } from 'oauth-1.0a';
 import clientOAuth1 from 'oauth-1.0a';
 
 import {
-	BinaryDataManager,
+	BinaryDataService,
 	Credentials,
 	LoadMappingOptions,
 	LoadNodeParameterOptions,
@@ -202,6 +202,8 @@ export class Server extends AbstractServer {
 
 	push: Push;
 
+	binaryDataService: BinaryDataService;
+
 	constructor() {
 		super('main');
 
@@ -358,12 +360,13 @@ export class Server extends AbstractServer {
 		this.endpointPresetCredentials = config.getEnv('credentials.overwrite.endpoint');
 
 		this.push = Container.get(Push);
+		this.binaryDataService = Container.get(BinaryDataService);
 
 		await super.start();
 		LoggerProxy.debug(`Server ID: ${this.uniqueInstanceId}`);
 
 		const cpus = os.cpus();
-		const binaryDataConfig = config.getEnv('binaryDataManager');
+		const binaryDataConfig = config.getEnv('binaryDataService');
 		const diagnosticInfo: IDiagnosticInfo = {
 			databaseType: config.getEnv('database.type'),
 			disableProductionWebhooksOnMainProcess: config.getEnv(
@@ -396,7 +399,6 @@ export class Server extends AbstractServer {
 				),
 				executions_data_prune: config.getEnv('executions.pruneData'),
 				executions_data_max_age: config.getEnv('executions.pruneDataMaxAge'),
-				executions_data_prune_timeout: config.getEnv('executions.pruneDataTimeout'),
 			},
 			deploymentType: config.getEnv('deployment.type'),
 			binaryDataMode: binaryDataConfig.mode,
@@ -1425,13 +1427,12 @@ export class Server extends AbstractServer {
 			async (req: BinaryDataRequest, res: express.Response): Promise<void> => {
 				// TODO UM: check if this needs permission check for UM
 				const identifier = req.params.path;
-				const binaryDataManager = BinaryDataManager.getInstance();
 				try {
-					const binaryPath = binaryDataManager.getBinaryPath(identifier);
+					const binaryPath = this.binaryDataService.getPath(identifier);
 					let { mode, fileName, mimeType } = req.query;
 					if (!fileName || !mimeType) {
 						try {
-							const metadata = await binaryDataManager.getBinaryMetadata(identifier);
+							const metadata = await this.binaryDataService.getMetadata(identifier);
 							fileName = metadata.fileName;
 							mimeType = metadata.mimeType;
 							res.setHeader('Content-Length', metadata.fileSize);

packages/cli/src/WebhookHelpers.ts

@@ -14,7 +14,7 @@ import stream from 'stream';
 import { promisify } from 'util';
 import formidable from 'formidable';
 
-import { BinaryDataManager, NodeExecuteFunctions } from 'n8n-core';
+import { BinaryDataService, NodeExecuteFunctions } from 'n8n-core';
 
 import type {
 	IBinaryData,
@@ -514,7 +514,7 @@ export async function executeWebhook(
 					const binaryData = (response.body as IDataObject)?.binaryData as IBinaryData;
 					if (binaryData?.id) {
 						res.header(response.headers);
-						const stream = BinaryDataManager.getInstance().getBinaryStream(binaryData.id);
+						const stream = Container.get(BinaryDataService).getAsStream(binaryData.id);
 						void pipeline(stream, res).then(() =>
 							responseCallback(null, { noWebhookResponse: true }),
 						);
@@ -734,7 +734,7 @@ export async function executeWebhook(
 								// Send the webhook response manually
 								res.setHeader('Content-Type', binaryData.mimeType);
 								if (binaryData.id) {
-									const stream = BinaryDataManager.getInstance().getBinaryStream(binaryData.id);
+									const stream = Container.get(BinaryDataService).getAsStream(binaryData.id);
 									await pipeline(stream, res);
 								} else {
 									res.end(Buffer.from(binaryData.data, BINARY_ENCODING));

packages/cli/src/WorkflowExecuteAdditionalData.ts

@@ -9,7 +9,7 @@
 /* eslint-disable @typescript-eslint/no-unused-vars */
 
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
-import { BinaryDataManager, UserSettings, WorkflowExecute } from 'n8n-core';
+import { UserSettings, WorkflowExecute } from 'n8n-core';
 
 import type {
 	IDataObject,
@@ -37,21 +37,16 @@ import {
 } from 'n8n-workflow';
 
 import { Container } from 'typedi';
-import type { FindOptionsWhere } from 'typeorm';
-import { LessThanOrEqual, In } from 'typeorm';
-import { DateUtils } from 'typeorm/util/DateUtils';
 import config from '@/config';
-import * as Db from '@/Db';
 import { ActiveExecutions } from '@/ActiveExecutions';
 import { CredentialsHelper } from '@/CredentialsHelper';
 import { ExternalHooks } from '@/ExternalHooks';
 import type {
-	IExecutionDb,
-	IExecutionFlattedDb,
 	IPushDataExecutionFinished,
 	IWorkflowExecuteProcess,
 	IWorkflowExecutionDataProcess,
 	IWorkflowErrorData,
+	ExecutionPayload,
 } from '@/Interfaces';
 import { NodeTypes } from '@/NodeTypes';
 import { Push } from '@/push';
@@ -184,77 +179,6 @@ export function executeErrorWorkflow(
 	}
 }
 
-/**
- * Prunes Saved Execution which are older than configured.
- * Throttled to be executed just once in configured timeframe.
- * TODO: Consider moving this whole function to the repository or at least the queries
- */
-let throttling = false;
-async function pruneExecutionData(this: WorkflowHooks): Promise<void> {
-	if (!throttling) {
-		Logger.verbose('Pruning execution data from database');
-
-		throttling = true;
-		const timeout = config.getEnv('executions.pruneDataTimeout'); // in seconds
-		const maxAge = config.getEnv('executions.pruneDataMaxAge'); // in h
-		const maxCount = config.getEnv('executions.pruneDataMaxCount');
-		const date = new Date(); // today
-		date.setHours(date.getHours() - maxAge);
-
-		// date reformatting needed - see https://github.com/typeorm/typeorm/issues/2286
-
-		const utcDate = DateUtils.mixedDateToUtcDatetimeString(date);
-
-		const toPrune: Array<FindOptionsWhere<IExecutionFlattedDb>> = [
-			{ stoppedAt: LessThanOrEqual(utcDate) },
-		];
-
-		if (maxCount > 0) {
-			const executions = await Db.collections.Execution.find({
-				select: ['id'],
-				skip: maxCount,
-				take: 1,
-				order: { id: 'DESC' },
-			});
-
-			if (executions[0]) {
-				toPrune.push({ id: LessThanOrEqual(executions[0].id) });
-			}
-		}
-
-		try {
-			setTimeout(() => {
-				throttling = false;
-			}, timeout * 1000);
-			let executionIds: Array<IExecutionFlattedDb['id']>;
-			do {
-				executionIds = (
-					await Db.collections.Execution.find({
-						select: ['id'],
-						where: toPrune,
-						take: 100,
-					})
-				).map(({ id }) => id);
-				await Db.collections.Execution.delete({ id: In(executionIds) });
-				// Mark binary data for deletion for all executions
-				await BinaryDataManager.getInstance().markDataForDeletionByExecutionIds(executionIds);
-			} while (executionIds.length > 0);
-		} catch (error) {
-			ErrorReporter.error(error);
-			throttling = false;
-			Logger.error(
-				`Failed pruning execution data from database for execution ID ${this.executionId} (hookFunctionsSave)`,
-				{
-					...error,
-					executionId: this.executionId,
-					sessionId: this.sessionId,
-					workflowId: this.workflowData.id,
-				},
-			);
-		}
-	}
-}
-
 /**
  * Returns hook functions to push data to Editor-UI
  *
@@ -522,11 +446,6 @@ function hookFunctionsSave(parentProcessMode?: string): IWorkflowExecuteHooks {
 					workflowId: this.workflowData.id,
 				});
 
-				// Prune old execution data
-				if (config.getEnv('executions.pruneData')) {
-					await pruneExecutionData.call(this);
-				}
-
 				const isManualMode = [this.mode, parentProcessMode].includes('manual');
 
 				try {
@@ -554,8 +473,7 @@ function hookFunctionsSave(parentProcessMode?: string): IWorkflowExecuteHooks {
 					}
 
 					if (isManualMode && !saveManualExecutions && !fullRunData.waitTill) {
-						// Data is always saved, so we remove from database
-						await Container.get(ExecutionRepository).deleteExecution(this.executionId, true);
+						await Container.get(ExecutionRepository).softDelete(this.executionId);
 
 						return;
 					}
@@ -586,8 +504,7 @@ function hookFunctionsSave(parentProcessMode?: string): IWorkflowExecuteHooks {
 								this.executionId,
 								this.retryOf,
 							);
-							// Data is always saved, so we remove from database
-							await Container.get(ExecutionRepository).deleteExecution(this.executionId);
+							await Container.get(ExecutionRepository).softDelete(this.executionId);
 
 							return;
 						}
@@ -682,11 +599,6 @@ function hookFunctionsSaveWorker(): IWorkflowExecuteHooks {
 					workflowId: this.workflowData.id,
 				});
 				try {
-					// Prune old execution data
-					if (config.getEnv('executions.pruneData')) {
-						await pruneExecutionData.call(this);
-					}
-
 					if (isWorkflowIdValid(this.workflowData.id) && newStaticData) {
 						// Workflow is saved so update in database
 						try {
@@ -973,7 +885,7 @@ async function executeWorkflow(
 		// Therefore, database might not contain finished errors.
 		// Force an update to db as there should be no harm doing this
 
-		const fullExecutionData: IExecutionDb = {
+		const fullExecutionData: ExecutionPayload = {
 			data: fullRunData.data,
 			mode: fullRunData.mode,
 			finished: fullRunData.finished ? fullRunData.finished : false,

packages/cli/src/WorkflowHelpers.ts

@@ -23,7 +23,11 @@ import {
 } from 'n8n-workflow';
 import { v4 as uuid } from 'uuid';
 import * as Db from '@/Db';
-import type { IExecutionDb, IWorkflowErrorData, IWorkflowExecutionDataProcess } from '@/Interfaces';
+import type {
+	ExecutionPayload,
+	IWorkflowErrorData,
+	IWorkflowExecutionDataProcess,
+} from '@/Interfaces';
 import { NodeTypes } from '@/NodeTypes';
 // eslint-disable-next-line import/no-cycle
 import { WorkflowRunner } from '@/WorkflowRunner';
@@ -186,7 +190,7 @@ export async function executeErrorWorkflow(
 					initialNode,
 				);
 
-				const fullExecutionData: IExecutionDb = {
+				const fullExecutionData: ExecutionPayload = {
 					data: fakeExecution.data,
 					mode: fakeExecution.mode,
 					finished: false,

packages/cli/src/WorkflowRunner.ts

@@ -629,7 +629,7 @@ export class WorkflowRunner {
 						(workflowDidSucceed && saveDataSuccessExecution === 'none') ||
 						(!workflowDidSucceed && saveDataErrorExecution === 'none')
 					) {
-						await Container.get(ExecutionRepository).deleteExecution(executionId);
+						await Container.get(ExecutionRepository).softDelete(executionId);
 					}
 					// eslint-disable-next-line id-denylist
 				} catch (err) {