mirror of
https://github.com/n8n-io/n8n.git
synced 2025-03-05 20:50:17 -08:00
feat(core): Remove storeMetadata
and getSize
from binary data manager interface (no-changelog) (#7195)
Depends on: #7164 | Story: [PAY-838](https://linear.app/n8n/issue/PAY-838/introduce-object-store-service-for-binary-data) This PR removes `storeMetadata` and `getSize` from the binary data manager interface, as these are specific to filesystem mode. Also this disambiguates identifiers: ``` binaryDataId filesystem:289b4aac51e-dac6-4167-b793-6d5c415e2b47 {mode}:{fileId} fileId - FS 289b4aac51e-dac6-4167-b793-6d5c415e2b47 {executionId}{uuid} fileId - S3 /workflows/{workflowId}/executions/{executionId}/binary_data/b4aac51e-dac6-4167-b793-6d5c415e2b47 ``` Note: The object store changes originally in this PR were extracted out into the final PR. --------- Co-authored-by: कारतोफ्फेलस्क्रिप्ट™ <aditya@netroy.in>
This commit is contained in:
parent
6d6e2488c6
commit
dcc9cc13ed
|
@ -5,12 +5,13 @@ import { Service } from 'typedi';
|
|||
import { BINARY_ENCODING, LoggerProxy as Logger, IBinaryData } from 'n8n-workflow';
|
||||
|
||||
import { FileSystemManager } from './FileSystem.manager';
|
||||
import { InvalidBinaryDataManagerError, InvalidBinaryDataModeError, areValidModes } from './utils';
|
||||
import { UnknownBinaryDataManager, InvalidBinaryDataMode } from './errors';
|
||||
import { LogCatch } from '../decorators/LogCatch.decorator';
|
||||
import { areValidModes } from './utils';
|
||||
|
||||
import type { Readable } from 'stream';
|
||||
import type { BinaryData } from './types';
|
||||
import type { INodeExecutionData } from 'n8n-workflow';
|
||||
import { LogCatch } from '../decorators/LogCatch.decorator';
|
||||
|
||||
@Service()
|
||||
export class BinaryDataService {
|
||||
|
@ -21,7 +22,7 @@ export class BinaryDataService {
|
|||
private managers: Record<string, BinaryData.Manager> = {};
|
||||
|
||||
async init(config: BinaryData.Config) {
|
||||
if (!areValidModes(config.availableModes)) throw new InvalidBinaryDataModeError();
|
||||
if (!areValidModes(config.availableModes)) throw new InvalidBinaryDataMode();
|
||||
|
||||
this.availableModes = config.availableModes;
|
||||
this.mode = config.mode;
|
||||
|
@ -45,47 +46,39 @@ export class BinaryDataService {
|
|||
return binaryData;
|
||||
}
|
||||
|
||||
const identifier = await manager.copyByPath(path, executionId);
|
||||
binaryData.id = this.createIdentifier(identifier);
|
||||
binaryData.data = this.mode; // clear binary data from memory
|
||||
|
||||
const fileSize = await manager.getSize(identifier);
|
||||
binaryData.fileSize = prettyBytes(fileSize);
|
||||
|
||||
await manager.storeMetadata(identifier, {
|
||||
const { fileId, fileSize } = await manager.copyByFilePath(path, executionId, {
|
||||
fileName: binaryData.fileName,
|
||||
mimeType: binaryData.mimeType,
|
||||
fileSize,
|
||||
});
|
||||
|
||||
binaryData.id = this.createBinaryDataId(fileId);
|
||||
binaryData.fileSize = prettyBytes(fileSize);
|
||||
binaryData.data = this.mode; // clear binary data from memory
|
||||
|
||||
return binaryData;
|
||||
}
|
||||
|
||||
@LogCatch((error) => Logger.error('Failed to write binary data file', { error }))
|
||||
async store(binaryData: IBinaryData, input: Buffer | Readable, executionId: string) {
|
||||
async store(binaryData: IBinaryData, bufferOrStream: Buffer | Readable, executionId: string) {
|
||||
const manager = this.managers[this.mode];
|
||||
|
||||
if (!manager) {
|
||||
const buffer = await this.binaryToBuffer(input);
|
||||
const buffer = await this.binaryToBuffer(bufferOrStream);
|
||||
binaryData.data = buffer.toString(BINARY_ENCODING);
|
||||
binaryData.fileSize = prettyBytes(buffer.length);
|
||||
|
||||
return binaryData;
|
||||
}
|
||||
|
||||
const identifier = await manager.store(input, executionId);
|
||||
binaryData.id = this.createIdentifier(identifier);
|
||||
binaryData.data = this.mode; // clear binary data from memory
|
||||
|
||||
const fileSize = await manager.getSize(identifier);
|
||||
binaryData.fileSize = prettyBytes(fileSize);
|
||||
|
||||
await manager.storeMetadata(identifier, {
|
||||
const { fileId, fileSize } = await manager.store(bufferOrStream, executionId, {
|
||||
fileName: binaryData.fileName,
|
||||
mimeType: binaryData.mimeType,
|
||||
fileSize,
|
||||
});
|
||||
|
||||
binaryData.id = this.createBinaryDataId(fileId);
|
||||
binaryData.fileSize = prettyBytes(fileSize);
|
||||
binaryData.data = this.mode; // clear binary data from memory
|
||||
|
||||
return binaryData;
|
||||
}
|
||||
|
||||
|
@ -96,34 +89,32 @@ export class BinaryDataService {
|
|||
});
|
||||
}
|
||||
|
||||
getAsStream(identifier: string, chunkSize?: number) {
|
||||
const { mode, id } = this.splitBinaryModeFileId(identifier);
|
||||
getAsStream(binaryDataId: string, chunkSize?: number) {
|
||||
const [mode, fileId] = binaryDataId.split(':');
|
||||
|
||||
return this.getManager(mode).getStream(id, chunkSize);
|
||||
return this.getManager(mode).getAsStream(fileId, chunkSize);
|
||||
}
|
||||
|
||||
async getBinaryDataBuffer(binaryData: IBinaryData) {
|
||||
if (binaryData.id) return this.retrieveBinaryDataByIdentifier(binaryData.id);
|
||||
async getAsBuffer(binaryData: IBinaryData) {
|
||||
if (binaryData.id) {
|
||||
const [mode, fileId] = binaryData.id.split(':');
|
||||
|
||||
return this.getManager(mode).getAsBuffer(fileId);
|
||||
}
|
||||
|
||||
return Buffer.from(binaryData.data, BINARY_ENCODING);
|
||||
}
|
||||
|
||||
async retrieveBinaryDataByIdentifier(identifier: string) {
|
||||
const { mode, id } = this.splitBinaryModeFileId(identifier);
|
||||
getPath(binaryDataId: string) {
|
||||
const [mode, fileId] = binaryDataId.split(':');
|
||||
|
||||
return this.getManager(mode).getBuffer(id);
|
||||
return this.getManager(mode).getPath(fileId);
|
||||
}
|
||||
|
||||
getPath(identifier: string) {
|
||||
const { mode, id } = this.splitBinaryModeFileId(identifier);
|
||||
async getMetadata(binaryDataId: string) {
|
||||
const [mode, fileId] = binaryDataId.split(':');
|
||||
|
||||
return this.getManager(mode).getPath(id);
|
||||
}
|
||||
|
||||
async getMetadata(identifier: string) {
|
||||
const { mode, id } = this.splitBinaryModeFileId(identifier);
|
||||
|
||||
return this.getManager(mode).getMetadata(id);
|
||||
return this.getManager(mode).getMetadata(fileId);
|
||||
}
|
||||
|
||||
async deleteManyByExecutionIds(executionIds: string[]) {
|
||||
|
@ -167,14 +158,11 @@ export class BinaryDataService {
|
|||
// private methods
|
||||
// ----------------------------------
|
||||
|
||||
private createIdentifier(filename: string) {
|
||||
return `${this.mode}:${filename}`;
|
||||
}
|
||||
|
||||
private splitBinaryModeFileId(fileId: string) {
|
||||
const [mode, id] = fileId.split(':');
|
||||
|
||||
return { mode, id };
|
||||
/**
|
||||
* Create an identifier `${mode}:{fileId}` for `IBinaryData['id']`.
|
||||
*/
|
||||
private createBinaryDataId(fileId: string) {
|
||||
return `${this.mode}:${fileId}`;
|
||||
}
|
||||
|
||||
private async duplicateBinaryDataInExecData(
|
||||
|
@ -195,12 +183,12 @@ export class BinaryDataService {
|
|||
return { key, newId: undefined };
|
||||
}
|
||||
|
||||
return manager
|
||||
?.copyByIdentifier(this.splitBinaryModeFileId(binaryDataId).id, executionId)
|
||||
.then((filename) => ({
|
||||
newId: this.createIdentifier(filename),
|
||||
key,
|
||||
}));
|
||||
const [_mode, fileId] = binaryDataId.split(':');
|
||||
|
||||
return manager?.copyByFileId(fileId, executionId).then((newFileId) => ({
|
||||
newId: this.createBinaryDataId(newFileId),
|
||||
key,
|
||||
}));
|
||||
});
|
||||
|
||||
return Promise.all(bdPromises).then((b) => {
|
||||
|
@ -222,6 +210,6 @@ export class BinaryDataService {
|
|||
|
||||
if (manager) return manager;
|
||||
|
||||
throw new InvalidBinaryDataManagerError(mode);
|
||||
throw new UnknownBinaryDataManager(mode);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -5,9 +5,9 @@ import { v4 as uuid } from 'uuid';
|
|||
import { jsonParse } from 'n8n-workflow';
|
||||
|
||||
import { FileNotFoundError } from '../errors';
|
||||
import { ensureDirExists } from './utils';
|
||||
|
||||
import type { Readable } from 'stream';
|
||||
import type { BinaryMetadata } from 'n8n-workflow';
|
||||
import type { BinaryData } from './types';
|
||||
|
||||
const EXECUTION_ID_EXTRACTOR =
|
||||
|
@ -17,15 +17,15 @@ export class FileSystemManager implements BinaryData.Manager {
|
|||
constructor(private storagePath: string) {}
|
||||
|
||||
async init() {
|
||||
await this.ensureDirExists(this.storagePath);
|
||||
await ensureDirExists(this.storagePath);
|
||||
}
|
||||
|
||||
getPath(identifier: string) {
|
||||
return this.resolvePath(identifier);
|
||||
getPath(fileId: string) {
|
||||
return this.resolvePath(fileId);
|
||||
}
|
||||
|
||||
async getSize(identifier: string) {
|
||||
const filePath = this.getPath(identifier);
|
||||
async getSize(fileId: string) {
|
||||
const filePath = this.getPath(fileId);
|
||||
|
||||
try {
|
||||
const stats = await fs.stat(filePath);
|
||||
|
@ -35,14 +35,14 @@ export class FileSystemManager implements BinaryData.Manager {
|
|||
}
|
||||
}
|
||||
|
||||
getStream(identifier: string, chunkSize?: number) {
|
||||
const filePath = this.getPath(identifier);
|
||||
getAsStream(fileId: string, chunkSize?: number) {
|
||||
const filePath = this.getPath(fileId);
|
||||
|
||||
return createReadStream(filePath, { highWaterMark: chunkSize });
|
||||
}
|
||||
|
||||
async getBuffer(identifier: string) {
|
||||
const filePath = this.getPath(identifier);
|
||||
async getAsBuffer(fileId: string) {
|
||||
const filePath = this.getPath(fileId);
|
||||
|
||||
try {
|
||||
return await fs.readFile(filePath);
|
||||
|
@ -51,29 +51,31 @@ export class FileSystemManager implements BinaryData.Manager {
|
|||
}
|
||||
}
|
||||
|
||||
async storeMetadata(identifier: string, metadata: BinaryMetadata) {
|
||||
const filePath = this.resolvePath(`${identifier}.metadata`);
|
||||
|
||||
await fs.writeFile(filePath, JSON.stringify(metadata), { encoding: 'utf-8' });
|
||||
}
|
||||
|
||||
async getMetadata(identifier: string): Promise<BinaryMetadata> {
|
||||
const filePath = this.resolvePath(`${identifier}.metadata`);
|
||||
async getMetadata(fileId: string): Promise<BinaryData.Metadata> {
|
||||
const filePath = this.resolvePath(`${fileId}.metadata`);
|
||||
|
||||
return jsonParse(await fs.readFile(filePath, { encoding: 'utf-8' }));
|
||||
}
|
||||
|
||||
async store(binaryData: Buffer | Readable, executionId: string) {
|
||||
const identifier = this.createIdentifier(executionId);
|
||||
const filePath = this.getPath(identifier);
|
||||
async store(
|
||||
binaryData: Buffer | Readable,
|
||||
executionId: string,
|
||||
{ mimeType, fileName }: BinaryData.PreWriteMetadata,
|
||||
) {
|
||||
const fileId = this.createFileId(executionId);
|
||||
const filePath = this.getPath(fileId);
|
||||
|
||||
await fs.writeFile(filePath, binaryData);
|
||||
|
||||
return identifier;
|
||||
const fileSize = await this.getSize(fileId);
|
||||
|
||||
await this.storeMetadata(fileId, { mimeType, fileName, fileSize });
|
||||
|
||||
return { fileId, fileSize };
|
||||
}
|
||||
|
||||
async deleteOne(identifier: string) {
|
||||
const filePath = this.getPath(identifier);
|
||||
async deleteOne(fileId: string) {
|
||||
const filePath = this.getPath(fileId);
|
||||
|
||||
return fs.rm(filePath);
|
||||
}
|
||||
|
@ -98,35 +100,35 @@ export class FileSystemManager implements BinaryData.Manager {
|
|||
return deletedIds;
|
||||
}
|
||||
|
||||
async copyByPath(filePath: string, executionId: string) {
|
||||
const identifier = this.createIdentifier(executionId);
|
||||
async copyByFilePath(
|
||||
filePath: string,
|
||||
executionId: string,
|
||||
{ mimeType, fileName }: BinaryData.PreWriteMetadata,
|
||||
) {
|
||||
const newFileId = this.createFileId(executionId);
|
||||
|
||||
await fs.cp(filePath, this.getPath(identifier));
|
||||
await fs.cp(filePath, this.getPath(newFileId));
|
||||
|
||||
return identifier;
|
||||
const fileSize = await this.getSize(newFileId);
|
||||
|
||||
await this.storeMetadata(newFileId, { mimeType, fileName, fileSize });
|
||||
|
||||
return { fileId: newFileId, fileSize };
|
||||
}
|
||||
|
||||
async copyByIdentifier(identifier: string, executionId: string) {
|
||||
const newIdentifier = this.createIdentifier(executionId);
|
||||
async copyByFileId(fileId: string, executionId: string) {
|
||||
const newFileId = this.createFileId(executionId);
|
||||
|
||||
await fs.copyFile(this.resolvePath(identifier), this.resolvePath(newIdentifier));
|
||||
await fs.copyFile(this.resolvePath(fileId), this.resolvePath(newFileId));
|
||||
|
||||
return newIdentifier;
|
||||
return newFileId;
|
||||
}
|
||||
|
||||
// ----------------------------------
|
||||
// private methods
|
||||
// ----------------------------------
|
||||
|
||||
private async ensureDirExists(dir: string) {
|
||||
try {
|
||||
await fs.access(dir);
|
||||
} catch {
|
||||
await fs.mkdir(dir, { recursive: true });
|
||||
}
|
||||
}
|
||||
|
||||
private createIdentifier(executionId: string) {
|
||||
private createFileId(executionId: string) {
|
||||
return [executionId, uuid()].join('');
|
||||
}
|
||||
|
||||
|
@ -139,4 +141,10 @@ export class FileSystemManager implements BinaryData.Manager {
|
|||
|
||||
return returnPath;
|
||||
}
|
||||
|
||||
private async storeMetadata(fileId: string, metadata: BinaryData.Metadata) {
|
||||
const filePath = this.resolvePath(`${fileId}.metadata`);
|
||||
|
||||
await fs.writeFile(filePath, JSON.stringify(metadata), { encoding: 'utf-8' });
|
||||
}
|
||||
}
|
||||
|
|
13
packages/core/src/BinaryData/errors.ts
Normal file
13
packages/core/src/BinaryData/errors.ts
Normal file
|
@ -0,0 +1,13 @@
|
|||
import { BINARY_DATA_MODES } from './utils';
|
||||
|
||||
export class InvalidBinaryDataMode extends Error {
|
||||
constructor() {
|
||||
super(`Invalid binary data mode. Valid modes: ${BINARY_DATA_MODES.join(', ')}`);
|
||||
}
|
||||
}
|
||||
|
||||
export class UnknownBinaryDataManager extends Error {
|
||||
constructor(mode: string) {
|
||||
super(`No binary data manager found for: ${mode}`);
|
||||
}
|
||||
}
|
|
@ -1,5 +1,4 @@
|
|||
import type { Readable } from 'stream';
|
||||
import type { BinaryMetadata } from 'n8n-workflow';
|
||||
import type { BINARY_DATA_MODES } from './utils';
|
||||
|
||||
export namespace BinaryData {
|
||||
|
@ -11,31 +10,39 @@ export namespace BinaryData {
|
|||
localStoragePath: string;
|
||||
};
|
||||
|
||||
export type Metadata = {
|
||||
fileName?: string;
|
||||
mimeType?: string;
|
||||
fileSize: number;
|
||||
};
|
||||
|
||||
export type PreWriteMetadata = Omit<Metadata, 'fileSize'>;
|
||||
|
||||
export interface Manager {
|
||||
init(): Promise<void>;
|
||||
|
||||
store(binaryData: Buffer | Readable, executionId: string): Promise<string>;
|
||||
getPath(identifier: string): string;
|
||||
store(
|
||||
binaryData: Buffer | Readable,
|
||||
executionId: string,
|
||||
preStoreMetadata: PreWriteMetadata,
|
||||
): Promise<{ fileId: string; fileSize: number }>;
|
||||
|
||||
// @TODO: Refactor to use identifier
|
||||
getSize(path: string): Promise<number>;
|
||||
|
||||
getBuffer(identifier: string): Promise<Buffer>;
|
||||
getStream(identifier: string, chunkSize?: number): Readable;
|
||||
|
||||
// @TODO: Refactor out - not needed for object storage
|
||||
storeMetadata(identifier: string, metadata: BinaryMetadata): Promise<void>;
|
||||
|
||||
// @TODO: Refactor out - not needed for object storage
|
||||
getMetadata(identifier: string): Promise<BinaryMetadata>;
|
||||
getPath(fileId: string): string;
|
||||
getAsBuffer(fileId: string): Promise<Buffer>;
|
||||
getAsStream(fileId: string, chunkSize?: number): Readable;
|
||||
getMetadata(fileId: string): Promise<Metadata>;
|
||||
|
||||
// @TODO: Refactor to also use `workflowId` to support full path-like identifier:
|
||||
// `workflows/{workflowId}/executions/{executionId}/binary_data/{fileId}`
|
||||
copyByPath(path: string, executionId: string): Promise<string>;
|
||||
copyByFilePath(
|
||||
path: string,
|
||||
executionId: string,
|
||||
metadata: PreWriteMetadata,
|
||||
): Promise<{ fileId: string; fileSize: number }>;
|
||||
|
||||
copyByIdentifier(identifier: string, prefix: string): Promise<string>;
|
||||
copyByFileId(fileId: string, prefix: string): Promise<string>;
|
||||
|
||||
deleteOne(identifier: string): Promise<void>;
|
||||
deleteOne(fileId: string): Promise<void>;
|
||||
|
||||
// @TODO: Refactor to also receive `workflowId` to support full path-like identifier:
|
||||
// `workflows/{workflowId}/executions/{executionId}/binary_data/{fileId}`
|
||||
|
|
|
@ -1,3 +1,4 @@
|
|||
import fs from 'fs/promises';
|
||||
import type { BinaryData } from './types';
|
||||
|
||||
/**
|
||||
|
@ -12,14 +13,10 @@ export function areValidModes(modes: string[]): modes is BinaryData.Mode[] {
|
|||
return modes.every((m) => BINARY_DATA_MODES.includes(m as BinaryData.Mode));
|
||||
}
|
||||
|
||||
export class InvalidBinaryDataModeError extends Error {
|
||||
constructor() {
|
||||
super(`Invalid binary data mode. Valid modes: ${BINARY_DATA_MODES.join(', ')}`);
|
||||
}
|
||||
}
|
||||
|
||||
export class InvalidBinaryDataManagerError extends Error {
|
||||
constructor(mode: string) {
|
||||
super('No binary data manager found for mode: ' + mode);
|
||||
export async function ensureDirExists(dir: string) {
|
||||
try {
|
||||
await fs.access(dir);
|
||||
} catch {
|
||||
await fs.mkdir(dir, { recursive: true });
|
||||
}
|
||||
}
|
||||
|
|
|
@ -39,7 +39,6 @@ import pick from 'lodash/pick';
|
|||
import { extension, lookup } from 'mime-types';
|
||||
import type {
|
||||
BinaryHelperFunctions,
|
||||
BinaryMetadata,
|
||||
FieldType,
|
||||
FileSystemHelperFunctions,
|
||||
FunctionsBase,
|
||||
|
@ -140,6 +139,7 @@ import {
|
|||
import { getSecretsProxy } from './Secrets';
|
||||
import { getUserN8nFolderPath } from './UserSettings';
|
||||
import Container from 'typedi';
|
||||
import type { BinaryData } from './BinaryData/types';
|
||||
|
||||
axios.defaults.timeout = 300000;
|
||||
// Prevent axios from adding x-form-www-urlencoded headers by default
|
||||
|
@ -947,7 +947,7 @@ export function getBinaryPath(binaryDataId: string): string {
|
|||
/**
|
||||
* Returns binary file metadata
|
||||
*/
|
||||
export async function getBinaryMetadata(binaryDataId: string): Promise<BinaryMetadata> {
|
||||
export async function getBinaryMetadata(binaryDataId: string): Promise<BinaryData.Metadata> {
|
||||
return Container.get(BinaryDataService).getMetadata(binaryDataId);
|
||||
}
|
||||
|
||||
|
@ -992,7 +992,7 @@ export async function getBinaryDataBuffer(
|
|||
inputIndex: number,
|
||||
): Promise<Buffer> {
|
||||
const binaryData = inputData.main[inputIndex]![itemIndex]!.binary![propertyName]!;
|
||||
return Container.get(BinaryDataService).getBinaryDataBuffer(binaryData);
|
||||
return Container.get(BinaryDataService).getAsBuffer(binaryData);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
@ -48,12 +48,6 @@ export interface IBinaryData {
|
|||
id?: string;
|
||||
}
|
||||
|
||||
export interface BinaryMetadata {
|
||||
fileName?: string;
|
||||
mimeType?: string;
|
||||
fileSize: number;
|
||||
}
|
||||
|
||||
// All properties in this interface except for
|
||||
// "includeCredentialsOnRefreshOnBody" will get
|
||||
// removed once we add the OAuth2 hooks to the
|
||||
|
@ -694,7 +688,11 @@ export interface BinaryHelperFunctions {
|
|||
binaryToBuffer(body: Buffer | Readable): Promise<Buffer>;
|
||||
getBinaryPath(binaryDataId: string): string;
|
||||
getBinaryStream(binaryDataId: string, chunkSize?: number): Readable;
|
||||
getBinaryMetadata(binaryDataId: string): Promise<BinaryMetadata>;
|
||||
getBinaryMetadata(binaryDataId: string): Promise<{
|
||||
fileName?: string;
|
||||
mimeType?: string;
|
||||
fileSize: number;
|
||||
}>;
|
||||
}
|
||||
|
||||
export interface NodeHelperFunctions {
|
||||
|
|
Loading…
Reference in a new issue