refactor(core): Move blob storage primitives to @n8n/blob-storage (#34579)

packages/@n8n/blob-storage/src/__tests__/azure-byte-store.test.ts

@@ -1,9 +1,9 @@
 /* eslint-disable @typescript-eslint/unbound-method */
 
-import type { AzureBlobService } from '@n8n/blob-storage';
 import type { Readable } from 'node:stream';
 import { mock } from 'vitest-mock-extended';
 
+import type { AzureBlobService } from '../azure-blob/azure-blob.service.ee';
 import { AzureByteStore } from '../azure-byte-store.ee';
 import { azureNotFoundError, azureThrottledError, body } from './mocks';
 

packages/@n8n/blob-storage/src/__tests__/fs-byte-store.integration.test.ts

@@ -1,8 +1,3 @@
-/* eslint-disable @typescript-eslint/unbound-method */
-
-import { mockInstance } from '@n8n/backend-test-utils';
-import { Container } from '@n8n/di';
-import { ErrorReporter, StorageConfig } from 'n8n-core';
 import { UnexpectedError } from 'n8n-workflow';
 import fs, { mkdtemp, rm } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
@@ -10,23 +5,19 @@ import { join } from 'node:path';
 
 import { FsByteStore } from '../fs-byte-store';
 
-vi.unmock('node:fs/promises');
-
 let store: FsByteStore;
 let storagePath: string;
-let errorReporter: ErrorReporter;
+const reportError = vi.fn();
 
 const body = Buffer.from('hello-bytes', 'utf-8');
 
 beforeAll(async () => {
 	storagePath = await mkdtemp(join(tmpdir(), 'n8n-fs-byte-store-test-'));
-	mockInstance(StorageConfig, { storagePath });
-	errorReporter = mockInstance(ErrorReporter);
-	store = Container.get(FsByteStore);
+	store = new FsByteStore({ storagePath, reportError });
 });
 
 beforeEach(async () => {
-	vi.mocked(errorReporter.error).mockClear();
+	reportError.mockClear();
 	for (const entry of await fs.readdir(storagePath)) {
 		await rm(join(storagePath, entry), { recursive: true, force: true });
 	}
@@ -43,7 +34,7 @@ afterAll(async () => {
 describe('init', () => {
 	it('creates the storage dir if absent', async () => {
 		const customPath = join(storagePath, 'custom-init');
-		const custom = new FsByteStore({ storagePath: customPath } as StorageConfig, errorReporter);
+		const custom = new FsByteStore({ storagePath: customPath, reportError });
 
 		await custom.init();
 

packages/@n8n/blob-storage/src/__tests__/s3-byte-store.test.ts

@@ -1,9 +1,9 @@
 /* eslint-disable @typescript-eslint/unbound-method */
 
-import type { ObjectStoreService } from '@n8n/blob-storage';
 import type { Readable } from 'node:stream';
 import { mock } from 'vitest-mock-extended';
 
+import type { ObjectStoreService } from '../object-store/object-store.service.ee';
 import { S3ByteStore } from '../s3-byte-store.ee';
 import { body, s3NotFoundError, s3NoSuchBucketError, s3ThrottledError } from './mocks';
 

packages/@n8n/blob-storage/src/azure-byte-store.ee.ts

@@ -1,7 +1,7 @@
 import { ensureError } from '@n8n/utils/errors/ensure-error';
 import chunk from 'lodash/chunk';
-import type { AzureBlobService } from '@n8n/blob-storage';
 
+import type { AzureBlobService } from './azure-blob/azure-blob.service.ee';
 import type { ByteStore, ByteStoreKey } from './types';
 
 const MAX_DELETE_CONCURRENCY = 50;

packages/@n8n/blob-storage/src/fs-byte-store.ts

@@ -1,6 +1,4 @@
 import { assertDir } from '@n8n/backend-common';
-import { Service } from '@n8n/di';
-import { ErrorReporter, StorageConfig } from 'n8n-core';
 import { UnexpectedError } from 'n8n-workflow';
 import { randomUUID } from 'node:crypto';
 import fs from 'node:fs/promises';
@@ -8,19 +6,23 @@ import path from 'node:path';
 
 import type { ByteStore, ByteStoreKey } from './types';
 
-@Service()
+export type FsByteStoreOptions = {
+	/** Root dir all keys resolve under. */
+	storagePath: string;
+
+	/** Reports a non-fatal error, e.g. a failed temp-file cleanup. */
+	reportError: (error: unknown) => void;
+};
+
 export class FsByteStore implements ByteStore {
-	constructor(
-		private readonly config: StorageConfig,
-		private readonly reporter: ErrorReporter,
-	) {}
+	constructor(private readonly options: FsByteStoreOptions) {}
 
 	async init() {
-		await assertDir(this.config.storagePath);
+		await assertDir(this.options.storagePath);
 	}
 
 	async write(key: ByteStoreKey, body: Buffer) {
-		const writePath = this.resolvePath(key);
+		const writePath = this.getAbsolutePath(key);
 		await assertDir(path.dirname(writePath));
 		const tempPath = `${writePath}.tmp.${process.pid}.${randomUUID()}`;
 
@@ -33,13 +35,13 @@ export class FsByteStore implements ByteStore {
 			return body.length;
 		} finally {
 			if (!success) {
-				await fs.rm(tempPath, { force: true }).catch((error) => this.reporter.error(error));
+				await fs.rm(tempPath, { force: true }).catch((error) => this.options.reportError(error));
 			}
 		}
 	}
 
 	async read(key: ByteStoreKey) {
-		const readPath = this.resolvePath(key);
+		const readPath = this.getAbsolutePath(key);
 		try {
 			return await fs.readFile(readPath);
 		} catch (error) {
@@ -49,29 +51,15 @@ export class FsByteStore implements ByteStore {
 	}
 
 	async delete(keys: ByteStoreKey[]) {
-		const deletePaths = keys.map((key) => this.resolvePath(key));
+		const deletePaths = keys.map((key) => this.getAbsolutePath(key));
 		await Promise.all(deletePaths.map(async (p) => await fs.rm(p, { force: true })));
 		const dirs = [...new Set(deletePaths.map((p) => path.dirname(p)))];
 		await Promise.all(dirs.map(async (dir) => await this.removeEmptyAncestors(dir)));
 	}
 
-	// private methods
-
-	private async removeEmptyAncestors(dir: string) {
-		const storagePath = path.resolve(this.config.storagePath);
-		let current = dir;
-		while (current.startsWith(storagePath) && current !== storagePath) {
-			try {
-				await fs.rmdir(current);
-			} catch {
-				return; // non-empty (ENOTEMPTY) or already gone (ENOENT)
-			}
-			current = path.dirname(current);
-		}
-	}
-
-	private resolvePath(key: ByteStoreKey) {
-		const storagePath = path.resolve(this.config.storagePath);
+	/** Absolute filesystem path for `key`, guarded against escaping the storage root. */
+	getAbsolutePath(key: ByteStoreKey) {
+		const storagePath = path.resolve(this.options.storagePath);
 		const resolvedPath = path.resolve(storagePath, key);
 		const relativePath = path.relative(storagePath, resolvedPath);
 
@@ -84,6 +72,21 @@ export class FsByteStore implements ByteStore {
 		return resolvedPath;
 	}
 
+	// private methods
+
+	private async removeEmptyAncestors(dir: string) {
+		const storagePath = path.resolve(this.options.storagePath);
+		let current = dir;
+		while (current.startsWith(storagePath) && current !== storagePath) {
+			try {
+				await fs.rmdir(current);
+			} catch {
+				return; // non-empty (ENOTEMPTY) or already gone (ENOENT)
+			}
+			current = path.dirname(current);
+		}
+	}
+
 	private isFileNotFound(error: unknown): error is NodeJS.ErrnoException {
 		return (
 			error !== null && typeof error === 'object' && 'code' in error && error.code === 'ENOENT'

packages/@n8n/blob-storage/src/index.ts

@@ -1,4 +1,18 @@
-export type { BlobMetadata, PreWriteBlobMetadata } from './types';
+export type {
+	BlobMetadata,
+	PreWriteBlobMetadata,
+	ByteStore,
+	ByteStoreKey,
+	StorageLocation,
+	JsonEntry,
+	Stored,
+	JsonStoreOptions,
+} from './types';
+export { FsByteStore, type FsByteStoreOptions } from './fs-byte-store';
+export { S3ByteStore } from './s3-byte-store.ee';
+export { AzureByteStore } from './azure-byte-store.ee';
+export { JsonStore } from './json-store';
+export { SkippedEntryDeletionError } from './skipped-entry-deletion.error';
 export { createFixedSizeChunker } from './stream-utils';
 export { ObjectStoreConfig } from './object-store/object-store.config';
 export type { MetadataResponseHeaders } from './object-store/types';

packages/@n8n/blob-storage/src/s3-byte-store.ee.ts

@@ -1,6 +1,6 @@
 import { ensureError } from '@n8n/utils/errors/ensure-error';
-import type { ObjectStoreService } from '@n8n/blob-storage';
 
+import type { ObjectStoreService } from './object-store/object-store.service.ee';
 import type { ByteStore, ByteStoreKey } from './types';
 
 export class S3ByteStore implements ByteStore {

packages/@n8n/blob-storage/src/types.ts

@@ -7,3 +7,50 @@ export type BlobMetadata = {
 
 /** Caller-supplied metadata for a blob write, before its size is known. */
 export type PreWriteBlobMetadata = Omit<BlobMetadata, 'fileSize'>;
+
+export type StorageLocation = 'fs' | 's3' | 'az';
+
+export type ByteStoreKey = string;
+
+/** A raw byte store addressed by an opaque key. */
+export interface ByteStore {
+	init?(): Promise<void>;
+
+	/** Returns the number of bytes written. */
+	write(key: ByteStoreKey, body: Buffer, contentType?: string): Promise<number>;
+
+	/** Returns `null` when no object exists for `key`. */
+	read(key: ByteStoreKey): Promise<Buffer | null>;
+
+	delete(keys: ByteStoreKey[]): Promise<void>;
+}
+
+/** A stored JSON entry. */
+export type JsonEntry<Payload extends object> = Payload & { version: number };
+
+/** A stored JSON entry identifier. */
+export type Stored<Ref> = Ref & { storedAt: StorageLocation };
+
+/** Per-domain configuration for a {@link JsonStore}. */
+export interface JsonStoreOptions<Ref> {
+	/** Byte store per location. `fs` is always present. Externals e.g. `s3` and `az` are registered when configured. */
+	byteStores: Partial<Record<StorageLocation, ByteStore>>;
+
+	/** Schema version stamped on write and verified on read. */
+	version: number;
+
+	/** Maps a ref to the byte-store key (path) where the entry is stored at. */
+	key: (ref: Ref) => string;
+
+	/** Identity used to key `readMany` results. */
+	getId: (ref: Ref) => string;
+
+	/** Wraps a write failure as a domain error. */
+	createWriteError: (ref: Ref, cause: unknown) => Error;
+
+	/** Wraps an unparseable or version-mismatched entry as a domain error. */
+	createCorruptedError: (ref: Ref, cause: unknown) => Error;
+
+	/** Reports a non-fatal error, e.g. a skipped delete. */
+	reportError: (error: unknown) => void;
+}

packages/cli/src/blob-storage/types.ts

@@ -1,46 +0,0 @@
-export type StorageLocation = 'fs' | 's3' | 'az';
-
-export type ByteStoreKey = string;
-
-/** A raw byte store addressed by an opaque key. */
-export interface ByteStore {
-	init?(): Promise<void>;
-
-	/** Returns the number of bytes written. */
-	write(key: ByteStoreKey, body: Buffer, contentType?: string): Promise<number>;
-
-	/** Returns `null` when no object exists for `key`. */
-	read(key: ByteStoreKey): Promise<Buffer | null>;
-
-	delete(keys: ByteStoreKey[]): Promise<void>;
-}
-
-/** A stored JSON entry. */
-export type JsonEntry<Payload extends object> = Payload & { version: number };
-
-/** A stored JSON entry identifier. */
-export type Stored<Ref> = Ref & { storedAt: StorageLocation };
-
-/** Per-domain configuration for a {@link JsonStore}. */
-export interface JsonStoreOptions<Ref> {
-	/** Byte store per location. `fs` is always present. Externals e.g. `s3` and `az` are registered when configured. */
-	byteStores: Partial<Record<StorageLocation, ByteStore>>;
-
-	/** Schema version stamped on write and verified on read. */
-	version: number;
-
-	/** Maps a ref to the byte-store key (path) where the entry is stored at. */
-	key: (ref: Ref) => string;
-
-	/** Identity used to key `readMany` results. */
-	getId: (ref: Ref) => string;
-
-	/** Wraps a write failure as a domain error. */
-	createWriteError: (ref: Ref, cause: unknown) => Error;
-
-	/** Wraps an unparseable or version-mismatched entry as a domain error. */
-	createCorruptedError: (ref: Ref, cause: unknown) => Error;
-
-	/** Reports a non-fatal error, e.g. a skipped delete. */
-	reportError: (error: unknown) => void;
-}

packages/cli/src/commands/base-command.ts

@@ -21,7 +21,7 @@ import {
 	ExecutionContextHookRegistry,
 	StorageConfig,
 } from 'n8n-core';
-import { AzureBlobConfig, ObjectStoreConfig } from '@n8n/blob-storage';
+import { AzureBlobConfig, AzureByteStore, ObjectStoreConfig, S3ByteStore } from '@n8n/blob-storage';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
 import { Expression, sleep, UnexpectedError } from 'n8n-workflow';
 
@@ -377,7 +377,6 @@ export abstract class BaseCommand<F = never> {
 		const objectStoreService = Container.get(ObjectStoreService);
 		await objectStoreService.init();
 
-		const { S3ByteStore } = await import('@/blob-storage/s3-byte-store.ee.js');
 		Container.get(ExecutionDataJsonStore).registerByteStore(
 			's3',
 			new S3ByteStore(objectStoreService),
@@ -393,7 +392,6 @@ export abstract class BaseCommand<F = never> {
 		const azureBlobService = Container.get(AzureBlobService);
 		await azureBlobService.init();
 
-		const { AzureByteStore } = await import('@/blob-storage/azure-byte-store.ee.js');
 		Container.get(ExecutionDataJsonStore).registerByteStore(
 			'az',
 			new AzureByteStore(azureBlobService),

packages/cli/src/executions/execution-data/__tests__/execution-data-json-store.test.ts

@@ -4,7 +4,7 @@
 import type { ErrorReporter } from 'n8n-core';
 import { mock } from 'vitest-mock-extended';
 
-import type { FsByteStore } from '@/blob-storage/fs-byte-store';
+import type { ExecutionDataFsByteStore } from '../execution-data-json-store';
 
 import { CorruptedExecutionDataError } from '../corrupted-execution-data.error';
 import { ExecutionDataJsonStore } from '../execution-data-json-store';
@@ -12,12 +12,12 @@ import { ExecutionDataWriteError } from '../execution-data-write.error';
 import { createExecutionRef } from '../types';
 import { executionId, payload, ref, workflowId } from './mocks';
 
-let fsByteStore: ReturnType<typeof mock<FsByteStore>>;
+let fsByteStore: ReturnType<typeof mock<ExecutionDataFsByteStore>>;
 let errorReporter: ReturnType<typeof mock<ErrorReporter>>;
 let store: ExecutionDataJsonStore;
 
 beforeEach(() => {
-	fsByteStore = mock<FsByteStore>();
+	fsByteStore = mock<ExecutionDataFsByteStore>();
 	errorReporter = mock<ErrorReporter>();
 	store = new ExecutionDataJsonStore(fsByteStore, errorReporter);
 });

packages/cli/src/executions/execution-data/execution-data-json-store.ts

@@ -1,21 +1,29 @@
+import { FsByteStore, JsonStore } from '@n8n/blob-storage';
 import { Service } from '@n8n/di';
-import { ErrorReporter } from 'n8n-core';
-
-import { FsByteStore } from '@/blob-storage/fs-byte-store';
-import { JsonStore } from '@/blob-storage/json-store';
+import { ErrorReporter, StorageConfig } from 'n8n-core';
 
 import { EXECUTION_DATA_BUNDLE_FILENAME, EXECUTION_DATA_BUNDLE_VERSION } from './constants';
 import { CorruptedExecutionDataError } from './corrupted-execution-data.error';
 import { ExecutionDataWriteError } from './execution-data-write.error';
 import type { ExecutionDataPayload, ExecutionRef } from './types';
 
+@Service()
+export class ExecutionDataFsByteStore extends FsByteStore {
+	constructor(storageConfig: StorageConfig, errorReporter: ErrorReporter) {
+		super({
+			storagePath: storageConfig.storagePath,
+			reportError: (error) => errorReporter.error(error),
+		});
+	}
+}
+
 /**
  * Stores execution data bundles as JSON blobs. The `fs` backend is always
  * available, but `s3` and `az` are registered at startup only if configured.
  */
 @Service()
 export class ExecutionDataJsonStore extends JsonStore<ExecutionRef, ExecutionDataPayload> {
-	constructor(fsByteStore: FsByteStore, errorReporter: ErrorReporter) {
+	constructor(fsByteStore: ExecutionDataFsByteStore, errorReporter: ErrorReporter) {
 		super({
 			byteStores: { fs: fsByteStore },
 			version: EXECUTION_DATA_BUNDLE_VERSION,