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

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(
@@ -1424,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/WorkflowRunnerProcess.ts

@@ -11,7 +11,7 @@ setDefaultResultOrder('ipv4first');
 
 import { Container } from 'typedi';
 import type { IProcessMessage } from 'n8n-core';
-import { BinaryDataManager, UserSettings, WorkflowExecute } from 'n8n-core';
+import { BinaryDataService, UserSettings, WorkflowExecute } from 'n8n-core';
 
 import type {
 	ExecutionError,
@@ -123,8 +123,8 @@ class WorkflowRunnerProcess {
 		await Container.get(PostHogClient).init(instanceId);
 		await Container.get(InternalHooks).init(instanceId);
 
-		const binaryDataConfig = config.getEnv('binaryDataManager');
-		await BinaryDataManager.init(binaryDataConfig);
+		const binaryDataConfig = config.getEnv('binaryDataService');
+		await Container.get(BinaryDataService).init(binaryDataConfig);
 
 		const license = Container.get(License);
 		await license.init(instanceId);

packages/cli/src/commands/BaseCommand.ts

@@ -3,7 +3,7 @@ import { ExitError } from '@oclif/errors';
 import { Container } from 'typedi';
 import { LoggerProxy, ErrorReporterProxy as ErrorReporter, sleep } from 'n8n-workflow';
 import type { IUserSettings } from 'n8n-core';
-import { BinaryDataManager, UserSettings } from 'n8n-core';
+import { BinaryDataService, UserSettings } from 'n8n-core';
 import type { AbstractServer } from '@/AbstractServer';
 import { getLogger } from '@/Logger';
 import config from '@/config';
@@ -103,9 +103,9 @@ export abstract class BaseCommand extends Command {
 		process.exit(1);
 	}
 
-	async initBinaryManager() {
-		const binaryDataConfig = config.getEnv('binaryDataManager');
-		await BinaryDataManager.init(binaryDataConfig, true);
+	async initBinaryDataService() {
+		const binaryDataConfig = config.getEnv('binaryDataService');
+		await Container.get(BinaryDataService).init(binaryDataConfig);
 	}
 
 	async initExternalHooks() {

packages/cli/src/commands/execute.ts

@@ -33,7 +33,7 @@ export class Execute extends BaseCommand {
 
 	async init() {
 		await super.init();
-		await this.initBinaryManager();
+		await this.initBinaryDataService();
 		await this.initExternalHooks();
 	}
 

packages/cli/src/commands/executeBatch.ts

@@ -180,7 +180,7 @@ export class ExecuteBatch extends BaseCommand {
 
 	async init() {
 		await super.init();
-		await this.initBinaryManager();
+		await this.initBinaryDataService();
 		await this.initExternalHooks();
 	}
 

packages/cli/src/commands/start.ts

@@ -201,7 +201,7 @@ export class Start extends BaseCommand {
 		this.activeWorkflowRunner = Container.get(ActiveWorkflowRunner);
 
 		await this.initLicense('main');
-		await this.initBinaryManager();
+		await this.initBinaryDataService();
 		await this.initExternalHooks();
 		await this.initExternalSecrets();
 

packages/cli/src/commands/webhook.ts

@@ -78,7 +78,7 @@ export class Webhook extends BaseCommand {
 		await super.init();
 
 		await this.initLicense('webhook');
-		await this.initBinaryManager();
+		await this.initBinaryDataService();
 		await this.initExternalHooks();
 		await this.initExternalSecrets();
 	}

packages/cli/src/commands/worker.ts

@@ -257,7 +257,7 @@ export class Worker extends BaseCommand {
 		this.logger.debug('Starting n8n worker...');
 
 		await this.initLicense('worker');
-		await this.initBinaryManager();
+		await this.initBinaryDataService();
 		await this.initExternalHooks();
 		await this.initExternalSecrets();
 		await this.initEventBus();

packages/cli/src/config/schema.ts

@@ -2,28 +2,21 @@ import path from 'path';
 import convict from 'convict';
 import { UserSettings } from 'n8n-core';
 import { jsonParse } from 'n8n-workflow';
+import { ensureStringArray } from './utils';
 
 convict.addFormat({
-	name: 'nodes-list',
-	// @ts-ignore
-	validate(values: string[], { env }: { env: string }): void {
-		try {
-			if (!Array.isArray(values)) {
-				throw new Error();
-			}
+	name: 'json-string-array',
+	coerce: (rawStr: string) =>
+		jsonParse<string[]>(rawStr, {
+			errorMessage: `Expected this value "${rawStr}" to be valid JSON`,
+		}),
+	validate: ensureStringArray,
+});
 
-			for (const value of values) {
-				if (typeof value !== 'string') {
-					throw new Error();
-				}
-			}
-		} catch (error) {
-			throw new TypeError(`${env} is not a valid Array of strings.`);
-		}
-	},
-	coerce(rawValue: string): string[] {
-		return jsonParse(rawValue, { errorMessage: 'nodes-list needs to be valid JSON' });
-	},
+convict.addFormat({
+	name: 'comma-separated-list',
+	coerce: (rawStr: string) => rawStr.split(','),
+	validate: ensureStringArray,
 });
 
 export const schema = {
@@ -788,13 +781,13 @@ export const schema = {
 	nodes: {
 		include: {
 			doc: 'Nodes to load',
-			format: 'nodes-list',
+			format: 'json-string-array',
 			default: undefined,
 			env: 'NODES_INCLUDE',
 		},
 		exclude: {
 			doc: 'Nodes not to load',
-			format: 'nodes-list',
+			format: 'json-string-array',
 			default: undefined,
 			env: 'NODES_EXCLUDE',
 		},
@@ -900,9 +893,9 @@ export const schema = {
 		},
 	},
 
-	binaryDataManager: {
+	binaryDataService: {
 		availableModes: {
-			format: String,
+			format: 'comma-separated-list',
 			default: 'filesystem',
 			env: 'N8N_AVAILABLE_BINARY_DATA_MODES',
 			doc: 'Available modes of binary data storage, as comma separated strings',

packages/cli/src/config/types.ts

@@ -1,7 +1,7 @@
 /* eslint-disable @typescript-eslint/naming-convention */
 /* eslint-disable @typescript-eslint/no-unused-vars */
 
-import type { IBinaryDataConfig } from 'n8n-core';
+import type { BinaryData } from 'n8n-core';
 import type { schema } from './schema';
 
 // -----------------------------------
@@ -76,7 +76,7 @@ type ToReturnType<T extends ConfigOptionPath> = T extends NumericPath
 
 type ExceptionPaths = {
 	'queue.bull.redis': object;
-	binaryDataManager: IBinaryDataConfig;
+	binaryDataService: BinaryData.Config;
 	'nodes.exclude': string[] | undefined;
 	'nodes.include': string[] | undefined;
 	'userManagement.isInstanceOwnerSetUp': boolean;

packages/cli/src/config/utils.ts

@@ -0,0 +1,17 @@
+import type { SchemaObj } from 'convict';
+
+class NotStringArrayError extends Error {
+	constructor(env: string) {
+		super(`${env} is not a string array.`);
+	}
+}
+
+export const ensureStringArray = (values: string[], { env }: SchemaObj<string>) => {
+	if (!env) throw new Error(`Missing env: ${env}`);
+
+	if (!Array.isArray(values)) throw new NotStringArrayError(env);
+
+	for (const value of values) {
+		if (typeof value !== 'string') throw new NotStringArrayError(env);
+	}
+};

packages/cli/src/databases/repositories/execution.repository.ts

@@ -10,7 +10,7 @@ import type {
 import { parse, stringify } from 'flatted';
 import { LoggerProxy as Logger } from 'n8n-workflow';
 import type { IExecutionsSummary, IRunExecutionData } from 'n8n-workflow';
-import { BinaryDataManager } from 'n8n-core';
+import { BinaryDataService } from 'n8n-core';
 import type {
 	ExecutionPayload,
 	IExecutionBase,
@@ -89,6 +89,7 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
 	constructor(
 		dataSource: DataSource,
 		private readonly executionDataRepository: ExecutionDataRepository,
+		private readonly binaryDataService: BinaryDataService,
 	) {
 		super(ExecutionEntity, dataSource.manager);
 
@@ -520,8 +521,7 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
 			})
 		).map(({ id }) => id);
 
-		const binaryDataManager = BinaryDataManager.getInstance();
-		await binaryDataManager.deleteBinaryDataByExecutionIds(executionIds);
+		await this.binaryDataService.deleteManyByExecutionIds(executionIds);
 
 		this.logger.debug(`Hard-deleting ${executionIds.length} executions from database`, {
 			executionIds,

packages/cli/test/integration/commands/worker.cmd.test.ts

@@ -5,7 +5,7 @@ import { LoggerProxy } from 'n8n-workflow';
 import { Telemetry } from '@/telemetry';
 import { getLogger } from '@/Logger';
 import { ExternalSecretsManager } from '@/ExternalSecrets/ExternalSecretsManager.ee';
-import { BinaryDataManager } from 'n8n-core';
+import { BinaryDataService } from 'n8n-core';
 import { CacheService } from '@/services/cache.service';
 import { RedisServicePubSubPublisher } from '@/services/redis/RedisServicePubSubPublisher';
 import { RedisServicePubSubSubscriber } from '@/services/redis/RedisServicePubSubSubscriber';
@@ -26,7 +26,7 @@ beforeAll(async () => {
 	mockInstance(InternalHooks);
 	mockInstance(CacheService);
 	mockInstance(ExternalSecretsManager);
-	mockInstance(BinaryDataManager);
+	mockInstance(BinaryDataService);
 	mockInstance(MessageEventBus);
 	mockInstance(LoadNodesAndCredentials);
 	mockInstance(CredentialTypes);
@@ -41,7 +41,7 @@ test('worker initializes all its components', async () => {
 
 	jest.spyOn(worker, 'init');
 	jest.spyOn(worker, 'initLicense').mockImplementation(async () => {});
-	jest.spyOn(worker, 'initBinaryManager').mockImplementation(async () => {});
+	jest.spyOn(worker, 'initBinaryDataService').mockImplementation(async () => {});
 	jest.spyOn(worker, 'initExternalHooks').mockImplementation(async () => {});
 	jest.spyOn(worker, 'initExternalSecrets').mockImplementation(async () => {});
 	jest.spyOn(worker, 'initEventBus').mockImplementation(async () => {});
@@ -64,7 +64,7 @@ test('worker initializes all its components', async () => {
 	expect(worker.uniqueInstanceId).toContain('worker');
 	expect(worker.uniqueInstanceId.length).toBeGreaterThan(15);
 	expect(worker.initLicense).toHaveBeenCalled();
-	expect(worker.initBinaryManager).toHaveBeenCalled();
+	expect(worker.initBinaryDataService).toHaveBeenCalled();
 	expect(worker.initExternalHooks).toHaveBeenCalled();
 	expect(worker.initExternalSecrets).toHaveBeenCalled();
 	expect(worker.initEventBus).toHaveBeenCalled();

packages/cli/test/integration/publicApi/executions.test.ts

@@ -27,8 +27,8 @@ beforeAll(async () => {
 	user1 = await testDb.createUser({ globalRole: globalUserRole, apiKey: randomApiKey() });
 	user2 = await testDb.createUser({ globalRole: globalUserRole, apiKey: randomApiKey() });
 
-	// TODO: mock BinaryDataManager instead
-	await utils.initBinaryManager();
+	// TODO: mock BinaryDataService instead
+	await utils.initBinaryDataService();
 	await utils.initNodeTypes();
 
 	workflowRunner = await utils.initActiveWorkflowRunner();

packages/cli/test/integration/shared/utils/index.ts

@@ -1,7 +1,7 @@
 import { Container } from 'typedi';
 import { randomBytes } from 'crypto';
 import { existsSync } from 'fs';
-import { BinaryDataManager, UserSettings } from 'n8n-core';
+import { BinaryDataService, UserSettings } from 'n8n-core';
 import type { INode } from 'n8n-workflow';
 import { GithubApi } from 'n8n-nodes-base/credentials/GithubApi.credentials';
 import { Ftp } from 'n8n-nodes-base/credentials/Ftp.credentials';
@@ -72,11 +72,14 @@ export async function initNodeTypes() {
 }
 
 /**
- * Initialize a BinaryManager for test runs.
+ * Initialize a BinaryDataService for test runs.
  */
-export async function initBinaryManager() {
-	const binaryDataConfig = config.getEnv('binaryDataManager');
-	await BinaryDataManager.init(binaryDataConfig);
+export async function initBinaryDataService() {
+	const binaryDataService = new BinaryDataService();
+
+	await binaryDataService.init(config.getEnv('binaryDataService'));
+
+	Container.set(BinaryDataService, binaryDataService);
 }
 
 /**

packages/core/package.json

@@ -47,8 +47,8 @@
     "n8n-nodes-base": "workspace:*"
   },
   "dependencies": {
-    "axios": "^0.21.1",
     "@n8n/client-oauth2": "workspace:*",
+    "axios": "^0.21.1",
     "concat-stream": "^2.0.0",
     "cron": "~1.7.2",
     "crypto-js": "~4.1.1",
@@ -63,6 +63,7 @@
     "p-cancelable": "^2.0.0",
     "pretty-bytes": "^5.6.0",
     "qs": "^6.10.1",
+    "typedi": "^0.10.0",
     "uuid": "^8.3.2"
   }
 }