mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-18 05:59:42 +02:00
Implement new built-in node to store file in project
This commit is contained in:
parent
01d996e940
commit
f4fe65fde9
|
|
@ -0,0 +1,158 @@
|
|||
import { Logger } from '@n8n/backend-common';
|
||||
import { Service } from '@n8n/di';
|
||||
import {
|
||||
NodeOperationError,
|
||||
type INode,
|
||||
type IProjectFileWriteService,
|
||||
type ProjectFileNodeInput,
|
||||
type ProjectFileNodeOutput,
|
||||
type ProjectFileProxyProvider,
|
||||
type Workflow,
|
||||
} from 'n8n-workflow';
|
||||
|
||||
import { ForbiddenError } from '@/errors/response-errors/forbidden.error';
|
||||
import { SourceControlPreferencesService } from '@/modules/source-control.ee/source-control-preferences.service.ee';
|
||||
import { OwnershipService } from '@/services/ownership.service';
|
||||
|
||||
import { ProjectFileConcurrentModificationError } from './errors/project-file-concurrent-modification.error';
|
||||
import { ProjectFileNameConflictError } from './errors/project-file-name-conflict.error';
|
||||
import { ProjectFileQuotaExceededError } from './errors/project-file-quota-exceeded.error';
|
||||
import { ProjectFileTooLargeError } from './errors/project-file-too-large.error';
|
||||
import type { ProjectFile } from './project-file.entity';
|
||||
import { ProjectFileService } from './project-file.service';
|
||||
import type { ProjectFileActor } from './project-files.types';
|
||||
|
||||
const ALLOWED_NODES = ['n8n-nodes-base.projectFile'] as const;
|
||||
|
||||
type AllowedNode = (typeof ALLOWED_NODES)[number];
|
||||
|
||||
export function isAllowedNode(s: string): s is AllowedNode {
|
||||
return ALLOWED_NODES.includes(s as AllowedNode);
|
||||
}
|
||||
|
||||
/**
|
||||
* Bridges the project-files module to the node that writes files from inside a
|
||||
* workflow. Published into workflow context by `ProjectFilesModule.context()`.
|
||||
*
|
||||
* The project is always the one that owns the executing workflow — there is no
|
||||
* project parameter on the node, so a workflow cannot write to another project's
|
||||
* files, and no per-execution scope check is needed.
|
||||
*/
|
||||
@Service()
|
||||
export class ProjectFileProxyService implements ProjectFileProxyProvider {
|
||||
constructor(
|
||||
private readonly projectFileService: ProjectFileService,
|
||||
private readonly ownershipService: OwnershipService,
|
||||
private readonly sourceControlPreferencesService: SourceControlPreferencesService,
|
||||
private readonly logger: Logger,
|
||||
) {
|
||||
this.logger = this.logger.scoped('project-files');
|
||||
}
|
||||
|
||||
async getProjectFileProxy(workflow: Workflow, node: INode): Promise<IProjectFileWriteService> {
|
||||
if (!isAllowedNode(node.type)) {
|
||||
throw new Error('This proxy is only available for the Project File node');
|
||||
}
|
||||
|
||||
const projectId = await this.resolveProjectId(workflow, node);
|
||||
const actor: ProjectFileActor = { type: 'workflow', workflowId: workflow.id };
|
||||
|
||||
return {
|
||||
addFile: async (file: ProjectFileNodeInput, options): Promise<ProjectFileNodeOutput> => {
|
||||
this.checkInstanceWriteAccess();
|
||||
|
||||
try {
|
||||
const { file: stored, overwritten } = await this.projectFileService.store(
|
||||
projectId,
|
||||
actor,
|
||||
file,
|
||||
{ overwrite: options?.overwrite ?? true },
|
||||
);
|
||||
|
||||
return { ...toNodeOutput(stored), overwritten };
|
||||
} catch (error) {
|
||||
throw this.toNodeError(node, error);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private checkInstanceWriteAccess(): void {
|
||||
const preferences = this.sourceControlPreferencesService.getPreferences();
|
||||
|
||||
if (preferences.branchReadOnly) {
|
||||
throw new ForbiddenError(
|
||||
'Cannot modify project files on a protected instance. This instance is in read-only mode.',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* An unsaved workflow has no owner row, so this is also what stops an
|
||||
* unpersisted workflow id from reaching the `createdByWorkflowId` FK.
|
||||
*/
|
||||
private async resolveProjectId(workflow: Workflow, node: INode): Promise<string> {
|
||||
try {
|
||||
const project = await this.ownershipService.getWorkflowProjectCached(workflow.id);
|
||||
|
||||
return project.id;
|
||||
} catch (error) {
|
||||
this.logger.warn('Could not resolve the project owning the workflow', {
|
||||
workflowId: workflow.id,
|
||||
error,
|
||||
});
|
||||
|
||||
throw new NodeOperationError(node, 'Could not find the project this workflow belongs to', {
|
||||
description: 'Save the workflow before writing files to its project.',
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/** Module errors carry good messages; a node also needs context and a next step. */
|
||||
private toNodeError(node: INode, error: unknown): unknown {
|
||||
if (error instanceof ProjectFileNameConflictError) {
|
||||
return new NodeOperationError(node, error.message, {
|
||||
description:
|
||||
"Turn on 'Replace Existing File' to overwrite it, or give the file a unique name.",
|
||||
});
|
||||
}
|
||||
|
||||
if (error instanceof ProjectFileQuotaExceededError) {
|
||||
return new NodeOperationError(node, error.message, {
|
||||
description:
|
||||
"Delete files from the project's Files tab, or raise the storage limit for this instance.",
|
||||
});
|
||||
}
|
||||
|
||||
if (error instanceof ProjectFileTooLargeError) {
|
||||
return new NodeOperationError(node, error.message, {
|
||||
description:
|
||||
'Raise N8N_PROJECT_FILES_MAX_FILE_SIZE_BYTES to store files this large, or shrink the file.',
|
||||
});
|
||||
}
|
||||
|
||||
if (error instanceof ProjectFileConcurrentModificationError) {
|
||||
return new NodeOperationError(node, error.message, {
|
||||
description: 'Another write to this file landed first. Retry the node.',
|
||||
});
|
||||
}
|
||||
|
||||
return error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Node output is visible in the NDV and persisted in execution data, so the
|
||||
* binary data reference is dropped here — it must never leave the server.
|
||||
*/
|
||||
function toNodeOutput(file: ProjectFile): Omit<ProjectFileNodeOutput, 'overwritten'> {
|
||||
return {
|
||||
id: file.id,
|
||||
name: file.name,
|
||||
mimeType: file.mimeType,
|
||||
fileSizeBytes: file.fileSizeBytes,
|
||||
projectId: file.projectId,
|
||||
createdAt: file.createdAt.toISOString(),
|
||||
updatedAt: file.updatedAt.toISOString(),
|
||||
};
|
||||
}
|
||||
|
|
@ -204,7 +204,13 @@ export class ProjectFileService {
|
|||
file.source.type === 'path'
|
||||
? // Takes a path and streams internally, so a large file never lands in memory.
|
||||
await this.binaryDataService.copyBinaryFile(location, binaryData, file.source.path)
|
||||
: await this.binaryDataService.store(location, file.source.buffer, binaryData);
|
||||
: await this.binaryDataService.store(
|
||||
location,
|
||||
// `store` accepts either, so a stream is passed straight through
|
||||
// instead of being buffered first.
|
||||
file.source.type === 'stream' ? file.source.stream : file.source.buffer,
|
||||
binaryData,
|
||||
);
|
||||
|
||||
if (!stored.id) {
|
||||
// The in-memory `default` binary mode returns the bytes inline and no id,
|
||||
|
|
|
|||
|
|
@ -43,4 +43,10 @@ export class ProjectFilesModule implements ModuleInterface {
|
|||
|
||||
return [ProjectFile];
|
||||
}
|
||||
|
||||
async context() {
|
||||
const { ProjectFileProxyService } = await import('./project-file-proxy.service.js');
|
||||
|
||||
return { projectFileProxyProvider: Container.get(ProjectFileProxyService) };
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import type { AuthenticatedRequest } from '@n8n/db';
|
||||
import type { Readable } from 'node:stream';
|
||||
|
||||
/**
|
||||
* Who performed an operation on a project file.
|
||||
|
|
@ -14,7 +15,10 @@ export type ProjectFileActor =
|
|||
/** Where the bytes of an incoming file come from. */
|
||||
export type ProjectFileSource =
|
||||
/** A temp file on disk, e.g. staged by multer. Streamed, never fully buffered. */
|
||||
{ type: 'path'; path: string } | { type: 'buffer'; buffer: Buffer };
|
||||
| { type: 'path'; path: string }
|
||||
| { type: 'buffer'; buffer: Buffer }
|
||||
/** An open read stream, e.g. a persisted execution binary. Never fully buffered. */
|
||||
| { type: 'stream'; stream: Readable };
|
||||
|
||||
export type ProjectFileListOptions = {
|
||||
take?: number;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,244 @@
|
|||
/**
|
||||
* Integration test for the proxy the "Add file to project" node writes through.
|
||||
*
|
||||
* Runs against a real database and a real filesystem-backed `BinaryDataService`,
|
||||
* so the workflow attribution and the resolved project are read back from actual
|
||||
* rows rather than asserted on mock calls.
|
||||
*/
|
||||
|
||||
import type { Logger } from '@n8n/backend-common';
|
||||
import { createTeamProject, createWorkflow, testDb, testModules } from '@n8n/backend-test-utils';
|
||||
import { ProjectFilesConfig } from '@n8n/config';
|
||||
import type { Project, User } from '@n8n/db';
|
||||
import { Container } from '@n8n/di';
|
||||
import { BinaryDataService, type BinaryDataConfig, type ErrorReporter } from 'n8n-core';
|
||||
import { NodeOperationError, type INode, type Workflow } from 'n8n-workflow';
|
||||
import { mkdtemp, rm } from 'node:fs/promises';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { Readable, type Readable as ReadableType } from 'node:stream';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import { ForbiddenError } from '@/errors/response-errors/forbidden.error';
|
||||
import { ProjectFileProxyService } from '@/modules/project-files/project-file-proxy.service';
|
||||
import { ProjectFileRepository } from '@/modules/project-files/project-file.repository';
|
||||
import { ProjectFileService } from '@/modules/project-files/project-file.service';
|
||||
import { SourceControlPreferencesService } from '@/modules/source-control.ee/source-control-preferences.service.ee';
|
||||
import type { SourceControlPreferences } from '@/modules/source-control.ee/types/source-control-preferences';
|
||||
|
||||
import { createOwner } from '../shared/db/users';
|
||||
|
||||
const NODE: INode = {
|
||||
id: 'node-id',
|
||||
name: 'Add file to project',
|
||||
type: 'n8n-nodes-base.projectFile',
|
||||
typeVersion: 1,
|
||||
position: [0, 0],
|
||||
parameters: {},
|
||||
};
|
||||
|
||||
const sourceControl = { getPreferences: vi.fn<() => SourceControlPreferences>() };
|
||||
|
||||
async function readStream(stream: ReadableType): Promise<Buffer> {
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of stream) chunks.push(chunk as Buffer);
|
||||
return Buffer.concat(chunks);
|
||||
}
|
||||
|
||||
describe('ProjectFileProxyService', () => {
|
||||
let storagePath: string;
|
||||
let proxyService: ProjectFileProxyService;
|
||||
let fileService: ProjectFileService;
|
||||
let repository: ProjectFileRepository;
|
||||
let config: ProjectFilesConfig;
|
||||
let owner: User;
|
||||
let project: Project;
|
||||
let workflowId: string;
|
||||
|
||||
/** The proxy only ever reads `workflow.id`. */
|
||||
function workflowRef(id: string) {
|
||||
return mock<Workflow>({ id });
|
||||
}
|
||||
|
||||
async function addFile(
|
||||
content: string,
|
||||
options?: { overwrite?: boolean },
|
||||
file?: { name?: string; source?: 'buffer' | 'stream' },
|
||||
) {
|
||||
const proxy = await proxyService.getProjectFileProxy(workflowRef(workflowId), NODE);
|
||||
const buffer = Buffer.from(content);
|
||||
|
||||
return await proxy.addFile(
|
||||
{
|
||||
name: file?.name ?? 'report.csv',
|
||||
mimeType: 'text/csv',
|
||||
sizeBytes: buffer.length,
|
||||
source:
|
||||
file?.source === 'stream'
|
||||
? { type: 'stream', stream: Readable.from(buffer) }
|
||||
: { type: 'buffer', buffer },
|
||||
},
|
||||
options,
|
||||
);
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
await testModules.loadModules(['project-files']);
|
||||
await testDb.init();
|
||||
|
||||
storagePath = await mkdtemp(join(tmpdir(), 'n8n-project-file-proxy-test-'));
|
||||
const binaryDataService = new BinaryDataService(
|
||||
mock<BinaryDataConfig>({
|
||||
mode: 'filesystem',
|
||||
availableModes: ['filesystem'],
|
||||
localStoragePath: storagePath,
|
||||
}),
|
||||
mock<ErrorReporter>(),
|
||||
mock<Logger>(),
|
||||
);
|
||||
await binaryDataService.init();
|
||||
Container.set(BinaryDataService, binaryDataService);
|
||||
|
||||
// Stubbed rather than constructed: the real service reaches for the instance's
|
||||
// ssh/git folders, and `branchReadOnly` is the only preference used here.
|
||||
Container.set(
|
||||
SourceControlPreferencesService,
|
||||
sourceControl as unknown as SourceControlPreferencesService,
|
||||
);
|
||||
|
||||
proxyService = Container.get(ProjectFileProxyService);
|
||||
fileService = Container.get(ProjectFileService);
|
||||
repository = Container.get(ProjectFileRepository);
|
||||
config = Container.get(ProjectFilesConfig);
|
||||
|
||||
owner = await createOwner();
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await rm(storagePath, { recursive: true, force: true });
|
||||
await testDb.terminate();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await testDb.truncate(['ProjectFile', 'WorkflowEntity']);
|
||||
config.maxFileSize = 100 * 1024 * 1024;
|
||||
config.projectMaxSize = 2 * 1024 * 1024 * 1024;
|
||||
config.personalTotalMaxSize = 1024 * 1024 * 1024;
|
||||
sourceControl.getPreferences.mockReturnValue(
|
||||
mock<SourceControlPreferences>({ branchReadOnly: false }),
|
||||
);
|
||||
|
||||
project = await createTeamProject(undefined, owner);
|
||||
const workflow = await createWorkflow({}, project);
|
||||
workflowId = workflow.id;
|
||||
});
|
||||
|
||||
describe('addFile', () => {
|
||||
it('stores a buffer and attributes the file to the workflow', async () => {
|
||||
const result = await addFile('a,b\n1,2\n');
|
||||
|
||||
expect(result).toMatchObject({
|
||||
name: 'report.csv',
|
||||
mimeType: 'text/csv',
|
||||
fileSizeBytes: 8,
|
||||
projectId: project.id,
|
||||
overwritten: false,
|
||||
});
|
||||
|
||||
// The binary data reference must never reach the node's output.
|
||||
expect(result).not.toHaveProperty('binaryDataId');
|
||||
|
||||
const [rows] = await repository.findManyByProjectId(project.id, {}, {});
|
||||
expect(rows[0]).toMatchObject({
|
||||
createdByWorkflowId: workflowId,
|
||||
updatedByWorkflowId: workflowId,
|
||||
createdById: null,
|
||||
updatedById: null,
|
||||
});
|
||||
});
|
||||
|
||||
it('stores a stream without buffering it first', async () => {
|
||||
const result = await addFile('streamed,content\n', undefined, { source: 'stream' });
|
||||
|
||||
const { file, stream } = await fileService.getAsStream(project.id, result.id);
|
||||
expect(await readStream(stream)).toEqual(Buffer.from('streamed,content\n'));
|
||||
expect(file.fileSizeBytes).toBe(17);
|
||||
});
|
||||
|
||||
it('replaces the content of an existing file by default', async () => {
|
||||
const first = await addFile('first\n');
|
||||
const second = await addFile('second-and-longer\n');
|
||||
|
||||
expect(second.id).toBe(first.id);
|
||||
expect(second.overwritten).toBe(true);
|
||||
|
||||
const [rows, count] = await repository.findManyByProjectId(project.id, {}, {});
|
||||
expect(count).toBe(1);
|
||||
expect(rows[0].fileSizeBytes).toBe(18);
|
||||
|
||||
const { stream } = await fileService.getAsStream(project.id, second.id);
|
||||
expect(await readStream(stream)).toEqual(Buffer.from('second-and-longer\n'));
|
||||
});
|
||||
|
||||
it('surfaces a name conflict as a node error naming the fix', async () => {
|
||||
await addFile('first\n');
|
||||
|
||||
await expect(addFile('second\n', { overwrite: false })).rejects.toThrow(NodeOperationError);
|
||||
await expect(addFile('second\n', { overwrite: false })).rejects.toThrow(
|
||||
/already exists in this project/,
|
||||
);
|
||||
});
|
||||
|
||||
it('surfaces an over-quota write as a node error', async () => {
|
||||
config.projectMaxSize = 4;
|
||||
|
||||
await expect(addFile('more than four bytes\n')).rejects.toThrow(NodeOperationError);
|
||||
await expect(addFile('more than four bytes\n')).rejects.toThrow(
|
||||
/Storage limit for this project reached/,
|
||||
);
|
||||
});
|
||||
|
||||
it('surfaces an oversized file as a node error', async () => {
|
||||
config.maxFileSize = 4;
|
||||
|
||||
await expect(addFile('more than four bytes\n')).rejects.toThrow(NodeOperationError);
|
||||
await expect(addFile('more than four bytes\n')).rejects.toThrow(/exceeds the maximum/);
|
||||
});
|
||||
|
||||
it('refuses to write on a read-only instance', async () => {
|
||||
sourceControl.getPreferences.mockReturnValue(
|
||||
mock<SourceControlPreferences>({ branchReadOnly: true }),
|
||||
);
|
||||
|
||||
await expect(addFile('blocked\n')).rejects.toThrow(ForbiddenError);
|
||||
});
|
||||
|
||||
it('writes to the project owning the workflow, not another project', async () => {
|
||||
const otherProject = await createTeamProject(undefined, owner);
|
||||
|
||||
const result = await addFile('a,b\n1,2\n');
|
||||
|
||||
expect(result.projectId).toBe(project.id);
|
||||
|
||||
const [, otherCount] = await repository.findManyByProjectId(otherProject.id, {}, {});
|
||||
expect(otherCount).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('getProjectFileProxy', () => {
|
||||
it('rejects a node type that is not the project file node', async () => {
|
||||
await expect(
|
||||
proxyService.getProjectFileProxy(workflowRef(workflowId), {
|
||||
...NODE,
|
||||
type: 'n8n-nodes-base.httpRequest',
|
||||
}),
|
||||
).rejects.toThrow('This proxy is only available for the Project File node');
|
||||
});
|
||||
|
||||
it('asks the user to save an unsaved workflow', async () => {
|
||||
await expect(
|
||||
proxyService.getProjectFileProxy(workflowRef('unsaved-workflow-id'), NODE),
|
||||
).rejects.toThrow(/Could not find the project this workflow belongs to/);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -7,6 +7,7 @@ import type {
|
|||
INode,
|
||||
IWorkflowSettings,
|
||||
OauthJweProxyProvider,
|
||||
ProjectFileProxyProvider,
|
||||
} from 'n8n-workflow';
|
||||
|
||||
import type { ExecutionLifecycleHooks } from './execution-lifecycle-hooks';
|
||||
|
|
@ -49,6 +50,7 @@ declare module 'n8n-workflow' {
|
|||
*/
|
||||
evalLlmMockHandler?: EvalLlmMockHandler;
|
||||
'data-table'?: { dataTableProxyProvider: DataTableProxyProvider };
|
||||
'project-files'?: { projectFileProxyProvider: ProjectFileProxyProvider };
|
||||
'dynamic-credentials'?: { credentialCheckProxy: DynamicCredentialCheckProxyProvider };
|
||||
'oauth-jwe'?: { oauthJweProxyProvider: OauthJweProxyProvider };
|
||||
// Project ID is currently only added on the additionalData if the user
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ import { getDeduplicationHelperFunctions } from './utils/deduplication-helper-fu
|
|||
import { getFileSystemHelperFunctions } from './utils/file-system-helper-functions';
|
||||
import { getInputConnectionData } from './utils/get-input-connection-data';
|
||||
import { normalizeItems } from './utils/normalize-items';
|
||||
import { getProjectFileHelperFunctions } from './utils/project-file-helper-functions';
|
||||
import { getRequestHelperFunctions } from './utils/request-helper-functions';
|
||||
import { returnJsonArray } from './utils/return-json-array';
|
||||
import { getSSHTunnelFunctions } from './utils/ssh-tunnel-helper-functions';
|
||||
|
|
@ -93,6 +94,7 @@ export class ExecuteContext extends BaseExecuteContext implements IExecuteFuncti
|
|||
),
|
||||
...getBinaryHelperFunctions(additionalData, workflow.id),
|
||||
...getDataTableHelperFunctions(additionalData, workflow, node),
|
||||
...getProjectFileHelperFunctions(additionalData, workflow, node),
|
||||
...getCredentialCheckHelperFunctions(additionalData),
|
||||
...getSSHTunnelFunctions(),
|
||||
...getFileSystemHelperFunctions(node),
|
||||
|
|
|
|||
|
|
@ -0,0 +1,20 @@
|
|||
import type {
|
||||
INode,
|
||||
IWorkflowExecuteAdditionalData,
|
||||
ProjectFileProxyFunctions,
|
||||
Workflow,
|
||||
} from 'n8n-workflow';
|
||||
|
||||
export function getProjectFileHelperFunctions(
|
||||
additionalData: IWorkflowExecuteAdditionalData,
|
||||
workflow: Workflow,
|
||||
node: INode,
|
||||
): Partial<ProjectFileProxyFunctions> {
|
||||
const projectFileProxyProvider = additionalData['project-files']?.projectFileProxyProvider;
|
||||
if (!projectFileProxyProvider) return {};
|
||||
|
||||
return {
|
||||
getProjectFileProxy: async () =>
|
||||
await projectFileProxyProvider.getProjectFileProxy(workflow, node),
|
||||
};
|
||||
}
|
||||
28
packages/nodes-base/nodes/ProjectFile/ProjectFile.node.json
Normal file
28
packages/nodes-base/nodes/ProjectFile/ProjectFile.node.json
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
{
|
||||
"node": "n8n-nodes-base.projectFile",
|
||||
"nodeVersion": "1.0",
|
||||
"codexVersion": "1.0",
|
||||
"details": "Add file to project",
|
||||
"categories": ["Core Nodes"],
|
||||
"resources": {
|
||||
"primaryDocumentation": [
|
||||
{
|
||||
"url": "https://docs.n8n.io/integrations/builtin/core-nodes/n8n-nodes-base.projectfile/"
|
||||
}
|
||||
]
|
||||
},
|
||||
"alias": [
|
||||
"file",
|
||||
"files",
|
||||
"project file",
|
||||
"project files",
|
||||
"save file",
|
||||
"store file",
|
||||
"upload",
|
||||
"attachment",
|
||||
"storage"
|
||||
],
|
||||
"subcategories": {
|
||||
"Core Nodes": ["Files"]
|
||||
}
|
||||
}
|
||||
123
packages/nodes-base/nodes/ProjectFile/ProjectFile.node.ts
Normal file
123
packages/nodes-base/nodes/ProjectFile/ProjectFile.node.ts
Normal file
|
|
@ -0,0 +1,123 @@
|
|||
import type {
|
||||
IExecuteFunctions,
|
||||
INodeExecutionData,
|
||||
INodeType,
|
||||
INodeTypeDescription,
|
||||
ProjectFileNodeInput,
|
||||
} from 'n8n-workflow';
|
||||
import { NodeConnectionTypes, NodeOperationError } from 'n8n-workflow';
|
||||
|
||||
export class ProjectFile implements INodeType {
|
||||
description: INodeTypeDescription = {
|
||||
displayName: 'Add file to project',
|
||||
name: 'projectFile',
|
||||
icon: 'fa:file-import',
|
||||
iconColor: 'orange-red',
|
||||
group: ['output'],
|
||||
version: 1,
|
||||
subtitle: '={{$parameter["fileName"]}}',
|
||||
description: 'Save a file to this project so other workflows can use it',
|
||||
defaults: {
|
||||
name: 'Add file to project',
|
||||
},
|
||||
inputs: [NodeConnectionTypes.Main],
|
||||
outputs: [NodeConnectionTypes.Main],
|
||||
properties: [
|
||||
{
|
||||
displayName: 'Input Binary Field',
|
||||
name: 'binaryPropertyName',
|
||||
type: 'string',
|
||||
default: 'data',
|
||||
required: true,
|
||||
hint: 'The name of the input binary field containing the file to save',
|
||||
},
|
||||
{
|
||||
displayName: 'File Name',
|
||||
name: 'fileName',
|
||||
type: 'string',
|
||||
default: '={{ $binary[$parameter.binaryPropertyName].fileName }}',
|
||||
required: true,
|
||||
placeholder: 'e.g. report.csv',
|
||||
description:
|
||||
'Name to store the file under. Names are unique per project, so reusing one replaces that file.',
|
||||
},
|
||||
{
|
||||
displayName: 'Replace Existing File',
|
||||
name: 'overwrite',
|
||||
type: 'boolean',
|
||||
default: true,
|
||||
description:
|
||||
'Whether to replace the file when one with the same name already exists. When turned off, the node fails instead.',
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
|
||||
const items = this.getInputData();
|
||||
const returnData: INodeExecutionData[] = [];
|
||||
|
||||
const getProjectFileProxy = this.helpers.getProjectFileProxy;
|
||||
|
||||
if (getProjectFileProxy === undefined) {
|
||||
throw new NodeOperationError(
|
||||
this.getNode(),
|
||||
'Project files are not available on this instance',
|
||||
{
|
||||
description:
|
||||
'The project-files module is disabled. Remove it from N8N_DISABLED_MODULES to use this node.',
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
const proxy = await getProjectFileProxy();
|
||||
|
||||
for (let i = 0; i < items.length; i++) {
|
||||
try {
|
||||
const binaryPropertyName = this.getNodeParameter('binaryPropertyName', i) as string;
|
||||
const fileName = this.getNodeParameter('fileName', i) as string;
|
||||
const overwrite = this.getNodeParameter('overwrite', i) as boolean;
|
||||
|
||||
const binaryData = this.helpers.assertBinaryData(i, binaryPropertyName);
|
||||
|
||||
// A persisted binary is streamed straight through; only binaries still
|
||||
// held in memory have to be buffered.
|
||||
const source: ProjectFileNodeInput['source'] = binaryData.id
|
||||
? { type: 'stream', stream: await this.helpers.getBinaryStream(binaryData.id) }
|
||||
: {
|
||||
type: 'buffer',
|
||||
buffer: await this.helpers.getBinaryDataBuffer(i, binaryPropertyName),
|
||||
};
|
||||
|
||||
const file = await proxy.addFile(
|
||||
{
|
||||
name: fileName,
|
||||
mimeType: binaryData.mimeType,
|
||||
sizeBytes: binaryData.bytes ?? 0,
|
||||
source,
|
||||
},
|
||||
{ overwrite },
|
||||
);
|
||||
|
||||
returnData.push({
|
||||
json: { ...file },
|
||||
// Passed through so a second node can store the same file under
|
||||
// another name.
|
||||
binary: items[i].binary,
|
||||
pairedItem: { item: i },
|
||||
});
|
||||
} catch (error) {
|
||||
if (this.continueOnFail()) {
|
||||
returnData.push({
|
||||
json: { error: error.message },
|
||||
pairedItem: { item: i },
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
return [returnData];
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,186 @@
|
|||
import type {
|
||||
IBinaryData,
|
||||
IExecuteFunctions,
|
||||
INode,
|
||||
INodeExecutionData,
|
||||
IProjectFileWriteService,
|
||||
ProjectFileNodeOutput,
|
||||
} from 'n8n-workflow';
|
||||
import { Readable } from 'node:stream';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import { ProjectFile } from '../ProjectFile.node';
|
||||
|
||||
const node: INode = {
|
||||
id: 'test-node',
|
||||
name: 'Add file to project',
|
||||
type: 'n8n-nodes-base.projectFile',
|
||||
typeVersion: 1,
|
||||
position: [0, 0],
|
||||
parameters: {},
|
||||
};
|
||||
|
||||
const storedFile: ProjectFileNodeOutput = {
|
||||
id: 'file-id',
|
||||
name: 'report.csv',
|
||||
mimeType: 'text/csv',
|
||||
fileSizeBytes: 12,
|
||||
projectId: 'project-id',
|
||||
createdAt: '2026-08-12T00:00:00.000Z',
|
||||
updatedAt: '2026-08-12T00:00:00.000Z',
|
||||
overwritten: false,
|
||||
};
|
||||
|
||||
type SetupOptions = {
|
||||
items?: INodeExecutionData[];
|
||||
binaryData?: Partial<IBinaryData>;
|
||||
parameters?: Record<string, unknown>;
|
||||
proxyAvailable?: boolean;
|
||||
continueOnFail?: boolean;
|
||||
addFile?: IProjectFileWriteService['addFile'];
|
||||
};
|
||||
|
||||
function setup({
|
||||
items = [{ json: {}, binary: { data: mock<IBinaryData>() } }],
|
||||
binaryData = {},
|
||||
parameters = {},
|
||||
proxyAvailable = true,
|
||||
continueOnFail = false,
|
||||
addFile = vi.fn().mockResolvedValue(storedFile),
|
||||
}: SetupOptions = {}) {
|
||||
const context = mock<IExecuteFunctions>();
|
||||
const proxy = mock<IProjectFileWriteService>({ addFile });
|
||||
|
||||
const resolvedParameters: Record<string, unknown> = {
|
||||
binaryPropertyName: 'data',
|
||||
fileName: 'report.csv',
|
||||
overwrite: true,
|
||||
...parameters,
|
||||
};
|
||||
|
||||
context.getNode.mockReturnValue(node);
|
||||
context.getInputData.mockReturnValue(items);
|
||||
context.continueOnFail.mockReturnValue(continueOnFail);
|
||||
context.getNodeParameter.mockImplementation(
|
||||
(name: string) => resolvedParameters[name] as ReturnType<IExecuteFunctions['getNodeParameter']>,
|
||||
);
|
||||
|
||||
const helpers = {
|
||||
assertBinaryData: vi.fn().mockReturnValue({
|
||||
mimeType: 'text/csv',
|
||||
fileName: 'report.csv',
|
||||
bytes: 12,
|
||||
...binaryData,
|
||||
}),
|
||||
getBinaryStream: vi.fn().mockResolvedValue(Readable.from('a,b\n1,2\n')),
|
||||
getBinaryDataBuffer: vi.fn().mockResolvedValue(Buffer.from('a,b\n1,2\n')),
|
||||
getProjectFileProxy: proxyAvailable ? vi.fn().mockResolvedValue(proxy) : undefined,
|
||||
};
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
context.helpers = helpers as any;
|
||||
|
||||
return { context, helpers, proxy };
|
||||
}
|
||||
|
||||
describe('Add file to project node', () => {
|
||||
it('streams a persisted binary instead of buffering it', async () => {
|
||||
const { context, helpers, proxy } = setup({ binaryData: { id: 'filesystem-v2:abc' } });
|
||||
|
||||
await new ProjectFile().execute.call(context);
|
||||
|
||||
expect(helpers.getBinaryStream).toHaveBeenCalledWith('filesystem-v2:abc');
|
||||
expect(helpers.getBinaryDataBuffer).not.toHaveBeenCalled();
|
||||
expect(proxy.addFile).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
name: 'report.csv',
|
||||
mimeType: 'text/csv',
|
||||
sizeBytes: 12,
|
||||
source: expect.objectContaining({ type: 'stream' }),
|
||||
}),
|
||||
{ overwrite: true },
|
||||
);
|
||||
});
|
||||
|
||||
it('buffers a binary that is still held in memory', async () => {
|
||||
const { context, helpers, proxy } = setup({ binaryData: { id: undefined } });
|
||||
|
||||
await new ProjectFile().execute.call(context);
|
||||
|
||||
expect(helpers.getBinaryStream).not.toHaveBeenCalled();
|
||||
expect(helpers.getBinaryDataBuffer).toHaveBeenCalledWith(0, 'data');
|
||||
expect(proxy.addFile).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ source: { type: 'buffer', buffer: expect.any(Buffer) } }),
|
||||
{ overwrite: true },
|
||||
);
|
||||
});
|
||||
|
||||
it('forwards the file name and overwrite parameters', async () => {
|
||||
const { context, proxy } = setup({
|
||||
parameters: { fileName: 'rates-latest.csv', overwrite: false },
|
||||
});
|
||||
|
||||
await new ProjectFile().execute.call(context);
|
||||
|
||||
expect(proxy.addFile).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ name: 'rates-latest.csv' }),
|
||||
{ overwrite: false },
|
||||
);
|
||||
});
|
||||
|
||||
it('returns the stored metadata and passes the input binary through', async () => {
|
||||
const binary = { data: mock<IBinaryData>() };
|
||||
const { context } = setup({ items: [{ json: { unrelated: true }, binary }] });
|
||||
|
||||
const result = await new ProjectFile().execute.call(context);
|
||||
|
||||
expect(result).toEqual([[{ json: { ...storedFile }, binary, pairedItem: { item: 0 } }]]);
|
||||
});
|
||||
|
||||
it('processes every input item', async () => {
|
||||
const { context, proxy } = setup({
|
||||
items: [{ json: {} }, { json: {} }, { json: {} }],
|
||||
});
|
||||
|
||||
const result = await new ProjectFile().execute.call(context);
|
||||
|
||||
expect(proxy.addFile).toHaveBeenCalledTimes(3);
|
||||
expect(result[0]).toHaveLength(3);
|
||||
expect(result[0].map((item) => item.pairedItem)).toEqual([
|
||||
{ item: 0 },
|
||||
{ item: 1 },
|
||||
{ item: 2 },
|
||||
]);
|
||||
});
|
||||
|
||||
it('throws when the project-files module is disabled', async () => {
|
||||
const { context } = setup({ proxyAvailable: false });
|
||||
|
||||
await expect(new ProjectFile().execute.call(context)).rejects.toThrow(
|
||||
'Project files are not available on this instance',
|
||||
);
|
||||
});
|
||||
|
||||
it('rethrows a store failure', async () => {
|
||||
const { context } = setup({
|
||||
addFile: vi.fn().mockRejectedValue(new Error("A file named 'report.csv' already exists")),
|
||||
});
|
||||
|
||||
await expect(new ProjectFile().execute.call(context)).rejects.toThrow(
|
||||
"A file named 'report.csv' already exists",
|
||||
);
|
||||
});
|
||||
|
||||
it('collects the error on the item when continueOnFail is on', async () => {
|
||||
const { context } = setup({
|
||||
continueOnFail: true,
|
||||
addFile: vi.fn().mockRejectedValue(new Error('Storage limit reached')),
|
||||
});
|
||||
|
||||
const result = await new ProjectFile().execute.call(context);
|
||||
|
||||
expect(result).toEqual([
|
||||
[{ json: { error: 'Storage limit reached' }, pairedItem: { item: 0 } }],
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
|
@ -754,6 +754,7 @@
|
|||
"dist/nodes/PostHog/PostHog.node.js",
|
||||
"dist/nodes/Postmark/PostmarkTrigger.node.js",
|
||||
"dist/nodes/ProfitWell/ProfitWell.node.js",
|
||||
"dist/nodes/ProjectFile/ProjectFile.node.js",
|
||||
"dist/nodes/Pushbullet/Pushbullet.node.js",
|
||||
"dist/nodes/Pushcut/Pushcut.node.js",
|
||||
"dist/nodes/Pushcut/PushcutTrigger.node.js",
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ export * from './constants';
|
|||
export * from './common';
|
||||
export * from './cron';
|
||||
export * from './data-table.types';
|
||||
export type * from './project-file.types';
|
||||
export * from './execution-context';
|
||||
export * from './execution-context-establishment-hooks';
|
||||
export * from './redaction-channels';
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ import type {
|
|||
} from './data-table.types';
|
||||
import type { ExecutionCancelledError } from './errors';
|
||||
import type { ExpressionError } from './errors/expression.error';
|
||||
import type { IProjectFileWriteService } from './project-file.types';
|
||||
import type { NodeApiError } from './errors/node-api.error';
|
||||
import type { NodeOperationError } from './errors/node-operation.error';
|
||||
import type { WorkflowActivationError } from './errors/workflow-activation.error';
|
||||
|
|
@ -1152,6 +1153,15 @@ export type DataTableProxyFunctions = {
|
|||
getDataTableProxy?(dataTableId: string): Promise<IDataTableProjectService>;
|
||||
};
|
||||
|
||||
export type ProjectFileProxyProvider = {
|
||||
getProjectFileProxy(workflow: Workflow, node: INode): Promise<IProjectFileWriteService>;
|
||||
};
|
||||
|
||||
export type ProjectFileProxyFunctions = {
|
||||
// Optional to account for situations where the project-files module is disabled
|
||||
getProjectFileProxy?(): Promise<IProjectFileWriteService>;
|
||||
};
|
||||
|
||||
export type CredentialCheckStatus = {
|
||||
credentialId: string;
|
||||
credentialName: string;
|
||||
|
|
@ -1275,6 +1285,7 @@ export type IExecuteFunctions = ExecuteFunctions.GetNodeParameterFn &
|
|||
FileSystemHelperFunctions &
|
||||
SSHTunnelFunctions &
|
||||
DataTableProxyFunctions &
|
||||
ProjectFileProxyFunctions &
|
||||
CredentialCheckProxyFunctions & {
|
||||
normalizeItems(items: INodeExecutionData | INodeExecutionData[]): INodeExecutionData[];
|
||||
constructExecutionMetaData(
|
||||
|
|
|
|||
40
packages/workflow/src/project-file.types.ts
Normal file
40
packages/workflow/src/project-file.types.ts
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
import type { Readable } from 'node:stream';
|
||||
|
||||
/**
|
||||
* A file handed to the project-files module by a node.
|
||||
*
|
||||
* The `path` source used by the multipart upload route is deliberately absent —
|
||||
* only the REST layer stages temp files on disk.
|
||||
*/
|
||||
export type ProjectFileNodeInput = {
|
||||
name: string;
|
||||
mimeType: string;
|
||||
/** Declared size, used for a cheap quota rejection before any bytes are written. */
|
||||
sizeBytes: number;
|
||||
source: { type: 'buffer'; buffer: Buffer } | { type: 'stream'; stream: Readable };
|
||||
};
|
||||
|
||||
/**
|
||||
* Metadata of a stored file, as returned to the node.
|
||||
*
|
||||
* Never carries `binaryDataId`: node output is visible in the NDV and persisted
|
||||
* in execution data, and `GET /rest/binary-data?id=` performs no ownership check.
|
||||
*/
|
||||
export type ProjectFileNodeOutput = {
|
||||
id: string;
|
||||
name: string;
|
||||
mimeType: string;
|
||||
fileSizeBytes: number;
|
||||
projectId: string;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
/** True when this replaced the content of an existing file of the same name. */
|
||||
overwritten: boolean;
|
||||
};
|
||||
|
||||
export type IProjectFileWriteService = {
|
||||
addFile(
|
||||
file: ProjectFileNodeInput,
|
||||
options?: { overwrite?: boolean },
|
||||
): Promise<ProjectFileNodeOutput>;
|
||||
};
|
||||
Loading…
Reference in New Issue
Block a user