diff --git a/packages/@n8n/blob-storage/src/__tests__/azure-byte-store.test.ts b/packages/@n8n/blob-storage/src/__tests__/azure-byte-store.test.ts index 9102a88c8a6..0e671a92e29 100644 --- a/packages/@n8n/blob-storage/src/__tests__/azure-byte-store.test.ts +++ b/packages/@n8n/blob-storage/src/__tests__/azure-byte-store.test.ts @@ -1,6 +1,6 @@ /* eslint-disable @typescript-eslint/unbound-method */ -import type { Readable } from 'node:stream'; +import { Readable } from 'node:stream'; import { mock } from 'vitest-mock-extended'; import type { AzureBlobService } from '../azure-blob/azure-blob.service.ee'; @@ -17,7 +17,7 @@ beforeEach(() => { describe('write', () => { it('should put the bytes at the key and return the byte count', async () => { - const bytes = await store.write('a/b/blob.json', body, 'application/json'); + const bytes = await store.write('a/b/blob.json', body, { mimeType: 'application/json' }); expect(azureBlob.put).toHaveBeenCalledWith('a/b/blob.json', body, { mimeType: 'application/json', }); @@ -30,6 +30,95 @@ describe('write', () => { mimeType: 'application/octet-stream', }); }); + + it('should forward native metadata', async () => { + await store.write('a/b/blob.bin', body, { fileName: 'file.txt', mimeType: 'text/plain' }); + expect(azureBlob.put).toHaveBeenCalledWith('a/b/blob.bin', body, { + fileName: 'file.txt', + mimeType: 'text/plain', + }); + }); + + it('should buffer a stream before uploading and return the byte count', async () => { + const bytes = await store.write('a/b/blob.bin', Readable.from(body)); + expect(azureBlob.put).toHaveBeenCalledWith('a/b/blob.bin', body, { + mimeType: 'application/octet-stream', + }); + expect(bytes).toBe(body.length); + }); +}); + +describe('readStream', () => { + it('should return the stored stream, forwarding chunkSize', async () => { + const stream = Readable.from(body); + azureBlob.get.mockResolvedValueOnce(stream); + + expect(await store.readStream('a/b/blob.json', { chunkSize: 5 })).toBe(stream); + expect(azureBlob.get).toHaveBeenCalledWith('a/b/blob.json', { mode: 'stream', chunkSize: 5 }); + }); + + it('should return `null` when the blob is missing', async () => { + azureBlob.get.mockRejectedValueOnce(azureNotFoundError); + expect(await store.readStream('a/b/blob.json')).toBeNull(); + }); + + it('should rethrow a systemic error', async () => { + azureBlob.get.mockRejectedValueOnce(azureThrottledError); + await expect(store.readStream('a/b/blob.json')).rejects.toBe(azureThrottledError); + }); + + it('should reject a fractional chunkSize before downloading', async () => { + await expect(store.readStream('a/b/blob.json', { chunkSize: 1.5 })).rejects.toThrow( + 'positive integer', + ); + expect(azureBlob.get).not.toHaveBeenCalled(); + }); +}); + +describe('getMetadata', () => { + it('should return the native metadata', async () => { + const metadata = { fileSize: 11, mimeType: 'text/plain', fileName: 'file.txt' }; + azureBlob.getMetadata.mockResolvedValueOnce(metadata); + + expect(await store.getMetadata('a/b/blob.json')).toEqual(metadata); + }); + + it('should return `null` when the blob is missing', async () => { + azureBlob.getMetadata.mockRejectedValueOnce(azureNotFoundError); + expect(await store.getMetadata('a/b/blob.json')).toBeNull(); + }); +}); + +describe('copy', () => { + it('should copy the bytes and preserved metadata to the target key', async () => { + azureBlob.get.mockResolvedValueOnce(body as unknown as Readable); + const metadata = { fileSize: 11, mimeType: 'text/plain', fileName: 'file.txt' }; + azureBlob.getMetadata.mockResolvedValueOnce(metadata); + + await store.copy('a/src.bin', 'a/dst.bin'); + + expect(azureBlob.get).toHaveBeenCalledWith('a/src.bin', { mode: 'buffer' }); + expect(azureBlob.put).toHaveBeenCalledWith('a/dst.bin', body, metadata); + }); +}); + +describe('rename', () => { + it('should copy to the new key and delete the old one', async () => { + azureBlob.get.mockResolvedValueOnce(body as unknown as Readable); + azureBlob.getMetadata.mockResolvedValueOnce({ fileSize: 11 }); + + await store.rename('a/old.bin', 'a/new.bin'); + + expect(azureBlob.put).toHaveBeenCalledWith('a/new.bin', body, { fileSize: 11 }); + expect(azureBlob.delete).toHaveBeenCalledWith('a/old.bin'); + }); + + it('should be a no-op when old and new keys match', async () => { + await store.rename('a/same.bin', 'a/same.bin'); + + expect(azureBlob.put).not.toHaveBeenCalled(); + expect(azureBlob.delete).not.toHaveBeenCalled(); + }); }); describe('read', () => { diff --git a/packages/@n8n/blob-storage/src/__tests__/fs-byte-store.integration.test.ts b/packages/@n8n/blob-storage/src/__tests__/fs-byte-store.integration.test.ts index 0528dc0ac5b..f6492c73582 100644 --- a/packages/@n8n/blob-storage/src/__tests__/fs-byte-store.integration.test.ts +++ b/packages/@n8n/blob-storage/src/__tests__/fs-byte-store.integration.test.ts @@ -2,6 +2,7 @@ import { UnexpectedError } from 'n8n-workflow'; import fs, { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { Readable } from 'node:stream'; import { FsByteStore } from '../fs-byte-store'; @@ -126,6 +127,100 @@ describe('delete', () => { }); }); +describe('write (stream)', () => { + it('writes a stream atomically and returns the byte count', async () => { + const bytes = await store.write('a/streamed.bin', Readable.from(body)); + + expect(bytes).toBe(body.length); + expect((await store.read('a/streamed.bin'))!.equals(body)).toBe(true); + const files = await fs.readdir(join(storagePath, 'a')); + expect(files.filter((f) => f.includes('.tmp.'))).toHaveLength(0); + }); +}); + +describe('readStream', () => { + it('returns a stream of the stored bytes', async () => { + await store.write('a/blob.json', body); + + const stream = await store.readStream('a/blob.json'); + + const chunks: Buffer[] = []; + for await (const chunk of stream!) chunks.push(Buffer.from(chunk as Buffer)); + expect(Buffer.concat(chunks).equals(body)).toBe(true); + }); + + it('returns null for a missing key', async () => { + expect(await store.readStream('a/missing.json')).toBeNull(); + }); + + it.each([0, -1, 1.5])('rejects invalid chunkSize %s', async (chunkSize) => { + await store.write('a/blob.json', body); + + await expect(store.readStream('a/blob.json', { chunkSize })).rejects.toThrow(UnexpectedError); + }); +}); + +describe('copy', () => { + it('copies the bytes to the target key, creating parent dirs', async () => { + await store.write('a/src.bin', body); + + await store.copy('a/src.bin', 'b/c/dst.bin'); + + expect((await store.read('b/c/dst.bin'))!.equals(body)).toBe(true); + expect((await store.read('a/src.bin'))!.equals(body)).toBe(true); + }); +}); + +describe('rename', () => { + it('moves the bytes to the new key, creating parent dirs', async () => { + await store.write('a/old.bin', body); + + await store.rename('a/old.bin', 'b/new.bin'); + + expect(await store.read('a/old.bin')).toBeNull(); + expect((await store.read('b/new.bin'))!.equals(body)).toBe(true); + }); + + it('keeps the bytes intact when old and new keys match', async () => { + await store.write('a/same.bin', body); + + await store.rename('a/same.bin', 'a/same.bin'); + + expect((await store.read('a/same.bin'))!.equals(body)).toBe(true); + }); +}); + +describe('deletePrefix', () => { + it('removes everything under the prefix and prunes empty ancestors', async () => { + await store.write('wf/exec/binary_data/one.bin', body); + await store.write('wf/exec/binary_data/two.bin', body); + + await store.deletePrefix('wf/exec/binary_data'); + + expect(await store.read('wf/exec/binary_data/one.bin')).toBeNull(); + await expect(fs.stat(join(storagePath, 'wf'))).rejects.toThrow(); + }); + + it('leaves siblings outside the prefix intact', async () => { + await store.write('wf/exec/binary_data/one.bin', body); + await store.write('wf/exec/execution_data/bundle.json', body); + + await store.deletePrefix('wf/exec/binary_data'); + + expect(await store.read('wf/exec/execution_data/bundle.json')).not.toBeNull(); + expect((await fs.stat(join(storagePath, 'wf/exec'))).isDirectory()).toBe(true); + }); + + it('does not throw for a missing prefix', async () => { + await expect(store.deletePrefix('missing/prefix')).resolves.toBeUndefined(); + }); + + it('rejects a prefix that escapes the storage root', async () => { + await expect(store.deletePrefix('../escape')).rejects.toThrow(UnexpectedError); + await expect(store.deletePrefix('.')).rejects.toThrow(UnexpectedError); + }); +}); + describe('path traversal guard', () => { it('rejects a key that escapes the storage root', async () => { await expect(store.write('../escape.json', body)).rejects.toThrow(UnexpectedError); diff --git a/packages/@n8n/blob-storage/src/__tests__/json-store.test.ts b/packages/@n8n/blob-storage/src/__tests__/json-store.test.ts index d2cdb5663de..4a750230005 100644 --- a/packages/@n8n/blob-storage/src/__tests__/json-store.test.ts +++ b/packages/@n8n/blob-storage/src/__tests__/json-store.test.ts @@ -2,6 +2,8 @@ /* eslint-disable n8n-local-rules/no-uncaught-json-parse */ /* eslint-disable @typescript-eslint/no-unsafe-assignment */ +import { Readable } from 'node:stream'; + import { JsonStore } from '../json-store'; import type { ByteStore, ByteStoreKey, StorageLocation } from '../types'; @@ -14,7 +16,8 @@ class InMemoryByteStore implements ByteStore { /** Force a non-ENOENT (systemic) failure on read for these keys. */ readonly failingKeys = new Set(); - async write(key: ByteStoreKey, body: Buffer) { + async write(key: ByteStoreKey, body: Buffer | Readable) { + if (!Buffer.isBuffer(body)) throw new Error('InMemoryByteStore supports Buffer writes only'); this.objects.set(key, body); return body.length; } @@ -24,6 +27,21 @@ class InMemoryByteStore implements ByteStore { return this.objects.get(key) ?? null; } + async readStream(key: ByteStoreKey): Promise { + const buffer = this.objects.get(key); + return buffer ? Readable.from(buffer) : null; + } + + async copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey): Promise { + const buffer = this.objects.get(sourceKey); + if (buffer) this.objects.set(targetKey, buffer); + } + + async rename(oldKey: ByteStoreKey, newKey: ByteStoreKey): Promise { + await this.copy(oldKey, newKey); + this.objects.delete(oldKey); + } + async delete(keys: ByteStoreKey[]): Promise { for (const key of keys) this.objects.delete(key); } diff --git a/packages/@n8n/blob-storage/src/__tests__/mocks.ts b/packages/@n8n/blob-storage/src/__tests__/mocks.ts index 964d87d9114..95b70e1829c 100644 --- a/packages/@n8n/blob-storage/src/__tests__/mocks.ts +++ b/packages/@n8n/blob-storage/src/__tests__/mocks.ts @@ -18,6 +18,11 @@ export const s3ThrottledError = wrapProviderError( Object.assign(new Error('slow down'), { name: 'ThrottlingException' }), ); +/** S3 HEAD 404s carry no error code, only a generic `NotFound` name. */ +export const s3HeadNotFoundError = wrapProviderError( + Object.assign(new Error('missing'), { name: 'NotFound' }), +); + export const azureNotFoundError = wrapProviderError( Object.assign(new Error('missing'), { code: 'BlobNotFound' }), ); diff --git a/packages/@n8n/blob-storage/src/__tests__/s3-byte-store.test.ts b/packages/@n8n/blob-storage/src/__tests__/s3-byte-store.test.ts index b2ac133c716..bbdfc4fcf4b 100644 --- a/packages/@n8n/blob-storage/src/__tests__/s3-byte-store.test.ts +++ b/packages/@n8n/blob-storage/src/__tests__/s3-byte-store.test.ts @@ -1,11 +1,17 @@ /* eslint-disable @typescript-eslint/unbound-method */ -import type { Readable } from 'node:stream'; +import { 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'; +import { + body, + s3HeadNotFoundError, + s3NotFoundError, + s3NoSuchBucketError, + s3ThrottledError, +} from './mocks'; let objectStoreService: ReturnType>; let store: S3ByteStore; @@ -17,7 +23,7 @@ beforeEach(() => { describe('write', () => { it('should put the bytes at the key and return the byte count', async () => { - const bytes = await store.write('a/b/blob.json', body, 'application/json'); + const bytes = await store.write('a/b/blob.json', body, { mimeType: 'application/json' }); expect(objectStoreService.put).toHaveBeenCalledWith('a/b/blob.json', body, { mimeType: 'application/json', }); @@ -30,6 +36,110 @@ describe('write', () => { mimeType: 'application/octet-stream', }); }); + + it('should forward native metadata', async () => { + await store.write('a/b/blob.bin', body, { fileName: 'file.txt', mimeType: 'text/plain' }); + expect(objectStoreService.put).toHaveBeenCalledWith('a/b/blob.bin', body, { + fileName: 'file.txt', + mimeType: 'text/plain', + }); + }); + + it('should buffer a stream before uploading and return the byte count', async () => { + const bytes = await store.write('a/b/blob.bin', Readable.from(body)); + expect(objectStoreService.put).toHaveBeenCalledWith('a/b/blob.bin', body, { + mimeType: 'application/octet-stream', + }); + expect(bytes).toBe(body.length); + }); +}); + +describe('readStream', () => { + it('should return the stored stream, forwarding chunkSize', async () => { + const stream = Readable.from(body); + objectStoreService.get.mockResolvedValueOnce(stream); + + expect(await store.readStream('a/b/blob.json', { chunkSize: 5 })).toBe(stream); + expect(objectStoreService.get).toHaveBeenCalledWith('a/b/blob.json', { + mode: 'stream', + chunkSize: 5, + }); + }); + + it('should return `null` when the object is missing', async () => { + objectStoreService.get.mockRejectedValueOnce(s3NotFoundError); + expect(await store.readStream('a/b/blob.json')).toBeNull(); + }); + + it('should rethrow a systemic error', async () => { + objectStoreService.get.mockRejectedValueOnce(s3ThrottledError); + await expect(store.readStream('a/b/blob.json')).rejects.toBe(s3ThrottledError); + }); + + it('should reject a fractional chunkSize before downloading', async () => { + await expect(store.readStream('a/b/blob.json', { chunkSize: 1.5 })).rejects.toThrow( + 'positive integer', + ); + expect(objectStoreService.get).not.toHaveBeenCalled(); + }); +}); + +describe('getMetadata', () => { + it('should normalize native metadata', async () => { + objectStoreService.getMetadata.mockResolvedValueOnce({ + 'content-length': '11', + 'content-type': 'text/plain', + 'x-amz-meta-filename': 'file.txt', + }); + + expect(await store.getMetadata('a/b/blob.json')).toEqual({ + fileSize: 11, + mimeType: 'text/plain', + fileName: 'file.txt', + }); + }); + + it('should return `null` when the object is missing (HEAD 404)', async () => { + objectStoreService.getMetadata.mockRejectedValueOnce(s3HeadNotFoundError); + expect(await store.getMetadata('a/b/blob.json')).toBeNull(); + }); +}); + +describe('copy', () => { + it('should copy the bytes and preserved metadata to the target key', async () => { + objectStoreService.get.mockResolvedValueOnce(body as unknown as Readable); + objectStoreService.getMetadata.mockResolvedValueOnce({ + 'content-type': 'text/plain', + 'x-amz-meta-filename': 'file.txt', + }); + + await store.copy('a/src.bin', 'a/dst.bin'); + + expect(objectStoreService.get).toHaveBeenCalledWith('a/src.bin', { mode: 'buffer' }); + expect(objectStoreService.put).toHaveBeenCalledWith('a/dst.bin', body, { + mimeType: 'text/plain', + fileName: 'file.txt', + }); + }); +}); + +describe('rename', () => { + it('should copy to the new key and delete the old one', async () => { + objectStoreService.get.mockResolvedValueOnce(body as unknown as Readable); + objectStoreService.getMetadata.mockResolvedValueOnce({}); + + await store.rename('a/old.bin', 'a/new.bin'); + + expect(objectStoreService.put).toHaveBeenCalledWith('a/new.bin', body, {}); + expect(objectStoreService.deleteOne).toHaveBeenCalledWith('a/old.bin'); + }); + + it('should be a no-op when old and new keys match', async () => { + await store.rename('a/same.bin', 'a/same.bin'); + + expect(objectStoreService.put).not.toHaveBeenCalled(); + expect(objectStoreService.deleteOne).not.toHaveBeenCalled(); + }); }); describe('read', () => { diff --git a/packages/@n8n/blob-storage/src/__tests__/stream-utils.test.ts b/packages/@n8n/blob-storage/src/__tests__/stream-utils.test.ts index f7fdbdb2d40..8f5c35f48b5 100644 --- a/packages/@n8n/blob-storage/src/__tests__/stream-utils.test.ts +++ b/packages/@n8n/blob-storage/src/__tests__/stream-utils.test.ts @@ -52,7 +52,7 @@ describe('stream-utils', () => { expect(chunks).toEqual([]); }); - it.each([0, -1])('should throw when chunkSize is %s', (chunkSize) => { + it.each([0, -1, 1.5])('should throw when chunkSize is %s', (chunkSize) => { expect(() => createFixedSizeChunker(chunkSize)).toThrow(UnexpectedError); }); }); diff --git a/packages/@n8n/blob-storage/src/azure-byte-store.ee.ts b/packages/@n8n/blob-storage/src/azure-byte-store.ee.ts index 7490f722552..dd682f98b01 100644 --- a/packages/@n8n/blob-storage/src/azure-byte-store.ee.ts +++ b/packages/@n8n/blob-storage/src/azure-byte-store.ee.ts @@ -1,17 +1,24 @@ +import { binaryToBuffer } from '@n8n/backend-network'; import { ensureError } from '@n8n/utils/errors/ensure-error'; import chunk from 'lodash/chunk'; +import type { Readable } from 'node:stream'; import type { AzureBlobService } from './azure-blob/azure-blob.service.ee'; -import type { ByteStore, ByteStoreKey } from './types'; +import { assertChunkSize } from './stream-utils'; +import type { BlobMetadata, ByteStore, ByteStoreKey, PreWriteBlobMetadata } from './types'; const MAX_DELETE_CONCURRENCY = 50; export class AzureByteStore implements ByteStore { constructor(private readonly azureBlob: AzureBlobService) {} - async write(key: ByteStoreKey, body: Buffer, contentType?: string) { - await this.azureBlob.put(key, body, { mimeType: contentType ?? 'application/octet-stream' }); - return body.length; + async write(key: ByteStoreKey, body: Buffer | Readable, metadata: PreWriteBlobMetadata = {}) { + const buffer = Buffer.isBuffer(body) ? body : await binaryToBuffer(body); + await this.azureBlob.put(key, buffer, { + ...metadata, + mimeType: metadata.mimeType ?? 'application/octet-stream', + }); + return buffer.length; } async read(key: ByteStoreKey): Promise { @@ -23,12 +30,45 @@ export class AzureByteStore implements ByteStore { } } + async readStream(key: ByteStoreKey, { chunkSize }: { chunkSize?: number } = {}) { + if (chunkSize !== undefined) assertChunkSize(chunkSize); + try { + return await this.azureBlob.get(key, { mode: 'stream', chunkSize }); + } catch (error) { + if (this.isNotFound(error)) return null; + throw error; + } + } + + async copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey) { + const buffer = await this.azureBlob.get(sourceKey, { mode: 'buffer' }); + const metadata = await this.azureBlob.getMetadata(sourceKey); + await this.azureBlob.put(targetKey, buffer, metadata); + } + + async rename(oldKey: ByteStoreKey, newKey: ByteStoreKey) { + if (oldKey === newKey) return; + await this.copy(oldKey, newKey); + await this.azureBlob.delete(oldKey); + } + async delete(keys: ByteStoreKey[]): Promise { for (const batch of chunk(keys, MAX_DELETE_CONCURRENCY)) { await Promise.all(batch.map(async (key) => await this.azureBlob.delete(key))); } } + async getMetadata(key: ByteStoreKey): Promise { + try { + return await this.azureBlob.getMetadata(key); + } catch (error) { + if (this.isNotFound(error)) return null; + throw error; + } + } + + // private methods + private isNotFound(error: unknown) { const original = ensureError(error).cause ?? error; if (typeof original !== 'object' || original === null) return false; diff --git a/packages/@n8n/blob-storage/src/fs-byte-store.ts b/packages/@n8n/blob-storage/src/fs-byte-store.ts index af6c58626da..1594ceeace8 100644 --- a/packages/@n8n/blob-storage/src/fs-byte-store.ts +++ b/packages/@n8n/blob-storage/src/fs-byte-store.ts @@ -3,7 +3,9 @@ import { UnexpectedError } from 'n8n-workflow'; import { randomUUID } from 'node:crypto'; import fs from 'node:fs/promises'; import path from 'node:path'; +import type { Readable } from 'node:stream'; +import { assertChunkSize } from './stream-utils'; import type { ByteStore, ByteStoreKey } from './types'; export type FsByteStoreOptions = { @@ -21,7 +23,7 @@ export class FsByteStore implements ByteStore { await assertDir(this.options.storagePath); } - async write(key: ByteStoreKey, body: Buffer) { + async write(key: ByteStoreKey, body: Buffer | Readable) { const writePath = this.getAbsolutePath(key); await assertDir(path.dirname(writePath)); const tempPath = `${writePath}.tmp.${process.pid}.${randomUUID()}`; @@ -30,9 +32,10 @@ export class FsByteStore implements ByteStore { try { await fs.writeFile(tempPath, body); + const bytesWritten = Buffer.isBuffer(body) ? body.length : (await fs.stat(tempPath)).size; await fs.rename(tempPath, writePath); success = true; - return body.length; + return bytesWritten; } finally { if (!success) { await fs.rm(tempPath, { force: true }).catch((error) => this.options.reportError(error)); @@ -50,6 +53,43 @@ export class FsByteStore implements ByteStore { } } + async readStream(key: ByteStoreKey, { chunkSize }: { chunkSize?: number } = {}) { + if (chunkSize !== undefined) assertChunkSize(chunkSize); + const readPath = this.getAbsolutePath(key); + let fileHandle; + try { + fileHandle = await fs.open(readPath, 'r'); + } catch (error) { + if (this.isFileNotFound(error)) return null; + throw error; + } + try { + return fileHandle.createReadStream(chunkSize ? { highWaterMark: chunkSize } : {}); + } catch (error) { + await fileHandle.close().catch(() => {}); + throw error; + } + } + + async copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey) { + const targetPath = this.getAbsolutePath(targetKey); + await assertDir(path.dirname(targetPath)); + await fs.copyFile(this.getAbsolutePath(sourceKey), targetPath); + } + + async rename(oldKey: ByteStoreKey, newKey: ByteStoreKey) { + if (oldKey === newKey) return; + const newPath = this.getAbsolutePath(newKey); + await assertDir(path.dirname(newPath)); + await fs.rename(this.getAbsolutePath(oldKey), newPath); + } + + async deletePrefix(prefix: string) { + const dir = this.getAbsolutePath(prefix); + await fs.rm(dir, { recursive: true, force: true }); + await this.removeEmptyAncestors(path.dirname(dir)); + } + async delete(keys: ByteStoreKey[]) { const deletePaths = keys.map((key) => this.getAbsolutePath(key)); await Promise.all(deletePaths.map(async (p) => await fs.rm(p, { force: true }))); diff --git a/packages/@n8n/blob-storage/src/json-store.ts b/packages/@n8n/blob-storage/src/json-store.ts index d5d9469b158..1d910ef377b 100644 --- a/packages/@n8n/blob-storage/src/json-store.ts +++ b/packages/@n8n/blob-storage/src/json-store.ts @@ -39,7 +39,7 @@ export class JsonStore { jsonStringify({ ...payload, version: this.options.version }), 'utf-8', ); - return await store.write(this.options.key(ref), body, 'application/json'); + return await store.write(this.options.key(ref), body, { mimeType: 'application/json' }); } catch (error) { throw this.options.createWriteError(ref, error); } diff --git a/packages/@n8n/blob-storage/src/s3-byte-store.ee.ts b/packages/@n8n/blob-storage/src/s3-byte-store.ee.ts index 97b46ff3b39..7c61b227fb1 100644 --- a/packages/@n8n/blob-storage/src/s3-byte-store.ee.ts +++ b/packages/@n8n/blob-storage/src/s3-byte-store.ee.ts @@ -1,36 +1,97 @@ +import { binaryToBuffer } from '@n8n/backend-network'; import { ensureError } from '@n8n/utils/errors/ensure-error'; +import type { Readable } from 'node:stream'; import type { ObjectStoreService } from './object-store/object-store.service.ee'; -import type { ByteStore, ByteStoreKey } from './types'; +import { assertChunkSize } from './stream-utils'; +import type { BlobMetadata, ByteStore, ByteStoreKey, PreWriteBlobMetadata } from './types'; export class S3ByteStore implements ByteStore { constructor(private readonly objectStore: ObjectStoreService) {} - async write(key: ByteStoreKey, body: Buffer, contentType?: string) { - await this.objectStore.put(key, body, { mimeType: contentType ?? 'application/octet-stream' }); - return body.length; + async write(key: ByteStoreKey, body: Buffer | Readable, metadata: PreWriteBlobMetadata = {}) { + const buffer = Buffer.isBuffer(body) ? body : await binaryToBuffer(body); + await this.objectStore.put(key, buffer, { + ...metadata, + mimeType: metadata.mimeType ?? 'application/octet-stream', + }); + return buffer.length; } async read(key: ByteStoreKey) { try { return await this.objectStore.get(key, { mode: 'buffer' }); } catch (error) { - if (this.isNotFound(error)) return null; + if (this.isNoSuchKey(error)) return null; throw error; } } + async readStream(key: ByteStoreKey, { chunkSize }: { chunkSize?: number } = {}) { + if (chunkSize !== undefined) assertChunkSize(chunkSize); + try { + return await this.objectStore.get(key, { mode: 'stream', chunkSize }); + } catch (error) { + if (this.isNoSuchKey(error)) return null; + throw error; + } + } + + async copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey) { + const buffer = await this.objectStore.get(sourceKey, { mode: 'buffer' }); + await this.objectStore.put(targetKey, buffer, await this.readNativeMetadata(sourceKey)); + } + + async rename(oldKey: ByteStoreKey, newKey: ByteStoreKey) { + if (oldKey === newKey) return; + await this.copy(oldKey, newKey); + await this.objectStore.deleteOne(oldKey); + } + async delete(keys: ByteStoreKey[]): Promise { if (keys.length === 0) return; await this.objectStore.deleteByKeys(keys); } + async getMetadata(key: ByteStoreKey): Promise { + try { + const headers = await this.objectStore.getMetadata(key); + const metadata: BlobMetadata = { fileSize: Number(headers['content-length'] ?? 0) }; + if (headers['content-type']) metadata.mimeType = headers['content-type']; + if (headers['x-amz-meta-filename']) metadata.fileName = headers['x-amz-meta-filename']; + return metadata; + } catch (error) { + if (this.isHeadNotFound(error)) return null; + throw error; + } + } + // private methods - private isNotFound(error: unknown) { + private async readNativeMetadata(key: ByteStoreKey): Promise { + const headers = await this.objectStore.getMetadata(key); + const metadata: PreWriteBlobMetadata = {}; + if (headers['content-type']) metadata.mimeType = headers['content-type']; + if (headers['x-amz-meta-filename']) metadata.fileName = headers['x-amz-meta-filename']; + return metadata; + } + + /** Missing object on a GET. Bucket-level failures surface as `NoSuchBucket` and rethrow. */ + private isNoSuchKey(error: unknown) { + return this.errorName(error) === 'NoSuchKey'; + } + + /** + * Missing object on a HEAD, which carries no error code, so a 404 cannot be told apart + * from a missing bucket. Acceptable: bucket existence is verified at startup (`init`). + */ + private isHeadNotFound(error: unknown) { + return this.errorName(error) === 'NotFound'; + } + + private errorName(error: unknown) { const original = ensureError(error).cause ?? error; - if (typeof original !== 'object' || original === null) return false; - const name = 'name' in original ? original.name : undefined; - return name === 'NoSuchKey' || name === 'NotFound'; + if (typeof original !== 'object' || original === null) return undefined; + return 'name' in original ? original.name : undefined; } } diff --git a/packages/@n8n/blob-storage/src/stream-utils.ts b/packages/@n8n/blob-storage/src/stream-utils.ts index bb3008c447e..e01b0d51010 100644 --- a/packages/@n8n/blob-storage/src/stream-utils.ts +++ b/packages/@n8n/blob-storage/src/stream-utils.ts @@ -1,6 +1,13 @@ import { UnexpectedError } from 'n8n-workflow'; import { Transform } from 'node:stream'; +/** Throws unless `chunkSize` is a positive integer. */ +export function assertChunkSize(chunkSize: number): void { + if (!Number.isInteger(chunkSize) || chunkSize <= 0) { + throw new UnexpectedError(`chunkSize must be a positive integer, got ${chunkSize}`); + } +} + /** * A `Transform` that re-emits its input as chunks of exactly `chunkSize` bytes, with a possibly smaller final chunk. * `chunkSize` must be a positive integer: values `<= 0` throws an `UnexpectedError`. @@ -14,9 +21,7 @@ import { Transform } from 'node:stream'; * `.pipe()` does neither. */ export function createFixedSizeChunker(chunkSize: number): Transform { - if (chunkSize <= 0) { - throw new UnexpectedError(`createFixedSizeChunker requires chunkSize > 0, got ${chunkSize}`); - } + assertChunkSize(chunkSize); const queue: Buffer[] = []; let queued = 0; diff --git a/packages/@n8n/blob-storage/src/types.ts b/packages/@n8n/blob-storage/src/types.ts index 6b82645cb90..4a532bfd038 100644 --- a/packages/@n8n/blob-storage/src/types.ts +++ b/packages/@n8n/blob-storage/src/types.ts @@ -1,3 +1,5 @@ +import type { Readable } from 'node:stream'; + /** Metadata stored alongside a blob. */ export type BlobMetadata = { fileName?: string; @@ -16,13 +18,32 @@ export type ByteStoreKey = string; export interface ByteStore { init?(): Promise; - /** Returns the number of bytes written. */ - write(key: ByteStoreKey, body: Buffer, contentType?: string): Promise; + /** Returns the number of bytes written. `metadata` is stored natively where supported, `fs` ignores it. */ + write( + key: ByteStoreKey, + body: Buffer | Readable, + metadata?: PreWriteBlobMetadata, + ): Promise; /** Returns `null` when no object exists for `key`. */ read(key: ByteStoreKey): Promise; + /** Returns `null` when no object exists. `chunkSize` (positive integer) fixes the emitted chunk size (final chunk may be smaller). */ + readStream(key: ByteStoreKey, opts?: { chunkSize?: number }): Promise; + + /** Preserves native metadata where supported. */ + copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey): Promise; + + /** Preserves native metadata where supported. Same-key rename is a no-op, even for a missing source. */ + rename(oldKey: ByteStoreKey, newKey: ByteStoreKey): Promise; + delete(keys: ByteStoreKey[]): Promise; + + /** `null` when no object exists. Absent on `fs`, which keeps sidecar entries at the domain layer. */ + getMetadata?(key: ByteStoreKey): Promise; + + /** Recursively deletes under `prefix`. Absent where the backend can't list or delegates bulk deletion (e.g. lifecycle policies). */ + deletePrefix?(prefix: string): Promise; } /** A stored JSON entry. */ diff --git a/packages/cli/src/executions/execution-data/__tests__/execution-data-json-store.test.ts b/packages/cli/src/executions/execution-data/__tests__/execution-data-json-store.test.ts index 073537d9c28..2ee3a966003 100644 --- a/packages/cli/src/executions/execution-data/__tests__/execution-data-json-store.test.ts +++ b/packages/cli/src/executions/execution-data/__tests__/execution-data-json-store.test.ts @@ -33,7 +33,7 @@ it('should write the version-1 bundle at the stable execution-data key', async ( expect(fsByteStore.write).toHaveBeenCalledWith( `workflows/${workflowId}/executions/${executionId}/execution_data/bundle.json`, expect.any(Buffer), - 'application/json', + { mimeType: 'application/json' }, ); const [_, body] = fsByteStore.write.mock.calls[0]; expect(JSON.parse(body.toString('utf-8'))).toEqual({ ...payload, version: 1 });