mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-19 16:55:02 +02:00
refactor(core): Replace agent knowledge base binary data storage with blob storage (no-changelog)
Moves agent knowledge file bytes off BinaryDataService (N8N_DEFAULT_BINARY_DATA_MODE) onto the @n8n/blob-storage ByteStore backends already used for agent execution logs, governed by N8N_EXECUTION_DATA_STORAGE_MODE. Falls back to the local filesystem when that mode is 'database', since no db-backed byte store exists. Existing knowledge-file rows are deleted by the migration rather than migrated, as the feature is at low rollout. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
b5da0744c4
commit
d292ca54ea
|
|
@ -441,12 +441,12 @@ erDiagram
|
|||
}
|
||||
"public.agent_files" {
|
||||
varchar_36_ agentId FK
|
||||
text binaryDataId
|
||||
timestamp_3__with_time_zone createdAt
|
||||
varchar_255_ fileName
|
||||
integer fileSizeBytes
|
||||
varchar_16_ id
|
||||
varchar_255_ mimeType
|
||||
varchar_2_ storedAt
|
||||
timestamp_3__with_time_zone updatedAt
|
||||
}
|
||||
"public.agent_history" {
|
||||
|
|
|
|||
|
|
@ -5,27 +5,28 @@
|
|||
| Name | Type | Default | Nullable | Children | Parents | Comment |
|
||||
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
|
||||
| agentId | varchar(36) | | false | | [public.agents](public.agents.md) | Agent that owns this uploaded file |
|
||||
| binaryDataId | text | | false | | | Opaque BinaryDataService reference (mode-prefixed, e.g. "filesystem-v2:\<uuid\>"); not an FK to binary_data, which only has rows in DB storage mode |
|
||||
| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
|
||||
| fileName | varchar(255) | | false | | | |
|
||||
| fileSizeBytes | integer | | false | | | Uploaded file size in bytes |
|
||||
| id | varchar(16) | | false | | | Application-generated n8n nano ID |
|
||||
| mimeType | varchar(255) | | false | | | |
|
||||
| storedAt | varchar(2) | 'fs'::character varying | false | | | |
|
||||
| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
|
||||
|
||||
## Constraints
|
||||
|
||||
| Name | Type | Definition |
|
||||
| ---- | ---- | ---------- |
|
||||
| CHK_agent_files_storedAt | CHECK | CHECK ((("storedAt")::text = ANY ((ARRAY['fs'::character varying, 's3'::character varying, 'az'::character varying])::text[]))) |
|
||||
| FK_aca4514cb500494b64356c2e164 | FOREIGN KEY | FOREIGN KEY ("agentId") REFERENCES agents(id) ON DELETE CASCADE |
|
||||
| PK_692920e59217af7d124cd95106f | PRIMARY KEY | PRIMARY KEY (id) |
|
||||
| agent_files_agentId_not_null | n | NOT NULL "agentId" |
|
||||
| agent_files_binaryDataId_not_null | n | NOT NULL "binaryDataId" |
|
||||
| agent_files_createdAt_not_null | n | NOT NULL "createdAt" |
|
||||
| agent_files_fileName_not_null | n | NOT NULL "fileName" |
|
||||
| agent_files_fileSizeBytes_not_null | n | NOT NULL "fileSizeBytes" |
|
||||
| agent_files_id_not_null | n | NOT NULL id |
|
||||
| agent_files_mimeType_not_null | n | NOT NULL "mimeType" |
|
||||
| agent_files_storedAt_not_null | n | NOT NULL "storedAt" |
|
||||
| agent_files_updatedAt_not_null | n | NOT NULL "updatedAt" |
|
||||
|
||||
## Indexes
|
||||
|
|
@ -33,7 +34,6 @@
|
|||
| Name | Definition |
|
||||
| ---- | ---------- |
|
||||
| IDX_45dafc48fe2ce95eac30fc8ffd | CREATE INDEX "IDX_45dafc48fe2ce95eac30fc8ffd" ON public.agent_files USING btree ("agentId", "createdAt") |
|
||||
| IDX_agent_files_agentId_binaryDataId | CREATE UNIQUE INDEX "IDX_agent_files_agentId_binaryDataId" ON public.agent_files USING btree ("agentId", "binaryDataId") |
|
||||
| IDX_agent_files_agentId_fileName | CREATE UNIQUE INDEX "IDX_agent_files_agentId_fileName" ON public.agent_files USING btree ("agentId", "fileName") |
|
||||
| PK_692920e59217af7d124cd95106f | CREATE UNIQUE INDEX "PK_692920e59217af7d124cd95106f" ON public.agent_files USING btree (id) |
|
||||
|
||||
|
|
@ -46,12 +46,12 @@ erDiagram
|
|||
|
||||
"public.agent_files" {
|
||||
varchar_36_ agentId FK
|
||||
text binaryDataId
|
||||
timestamp_3__with_time_zone createdAt
|
||||
varchar_255_ fileName
|
||||
integer fileSizeBytes
|
||||
varchar_16_ id
|
||||
varchar_255_ mimeType
|
||||
varchar_2_ storedAt
|
||||
timestamp_3__with_time_zone updatedAt
|
||||
}
|
||||
"public.agents" {
|
||||
|
|
|
|||
|
|
@ -136,12 +136,12 @@ erDiagram
|
|||
}
|
||||
"public.agent_files" {
|
||||
varchar_36_ agentId FK
|
||||
text binaryDataId
|
||||
timestamp_3__with_time_zone createdAt
|
||||
varchar_255_ fileName
|
||||
integer fileSizeBytes
|
||||
varchar_16_ id
|
||||
varchar_255_ mimeType
|
||||
varchar_2_ storedAt
|
||||
timestamp_3__with_time_zone updatedAt
|
||||
}
|
||||
"public.agent_task_definition" {
|
||||
|
|
|
|||
|
|
@ -428,12 +428,12 @@ erDiagram
|
|||
}
|
||||
"agent_files" {
|
||||
varchar_36_ agentId FK
|
||||
TEXT binaryDataId
|
||||
datetime_3_ createdAt
|
||||
varchar_255_ fileName
|
||||
INTEGER fileSizeBytes
|
||||
varchar_16_ id PK
|
||||
varchar_255_ mimeType
|
||||
VARCHAR_2_ storedAt
|
||||
datetime_3_ updatedAt
|
||||
}
|
||||
"agent_history" {
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@
|
|||
<summary><strong>Table Definition</strong></summary>
|
||||
|
||||
```sql
|
||||
CREATE TABLE "agent_files" ("id" varchar(16) PRIMARY KEY NOT NULL, "agentId" varchar(36) NOT NULL, "binaryDataId" text NOT NULL, "fileName" varchar(255) NOT NULL, "mimeType" varchar(255) NOT NULL, "fileSizeBytes" integer NOT NULL, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "FK_aca4514cb500494b64356c2e164" FOREIGN KEY ("agentId") REFERENCES "agents" ("id") ON DELETE CASCADE)
|
||||
CREATE TABLE "agent_files" ("id" varchar(16) PRIMARY KEY NOT NULL, "agentId" varchar(36) NOT NULL, "fileName" varchar(255) NOT NULL, "mimeType" varchar(255) NOT NULL, "fileSizeBytes" integer NOT NULL, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "storedAt" VARCHAR(2) NOT NULL DEFAULT 'fs' CONSTRAINT "CHK_agent_files_storedAt" CHECK("storedAt" IN ('fs', 's3', 'az')), CONSTRAINT "FK_aca4514cb500494b64356c2e164" FOREIGN KEY ("agentId") REFERENCES "agents" ("id") ON DELETE CASCADE ON UPDATE NO ACTION)
|
||||
```
|
||||
|
||||
</details>
|
||||
|
|
@ -16,18 +16,19 @@ CREATE TABLE "agent_files" ("id" varchar(16) PRIMARY KEY NOT NULL, "agentId" var
|
|||
| Name | Type | Default | Nullable | Children | Parents | Comment |
|
||||
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
|
||||
| agentId | varchar(36) | | false | | [agents](agents.md) | |
|
||||
| binaryDataId | TEXT | | false | | | |
|
||||
| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
|
||||
| fileName | varchar(255) | | false | | | |
|
||||
| fileSizeBytes | INTEGER | | false | | | |
|
||||
| id | varchar(16) | | false | | | |
|
||||
| mimeType | varchar(255) | | false | | | |
|
||||
| storedAt | VARCHAR(2) | 'fs' | false | | | |
|
||||
| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
|
||||
|
||||
## Constraints
|
||||
|
||||
| Name | Type | Definition |
|
||||
| ---- | ---- | ---------- |
|
||||
| - | CHECK | CHECK("storedAt" IN ('fs', 's3', 'az')) |
|
||||
| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (agentId) REFERENCES agents (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
|
||||
| id | PRIMARY KEY | PRIMARY KEY (id) |
|
||||
| sqlite_autoindex_agent_files_1 | PRIMARY KEY | PRIMARY KEY (id) |
|
||||
|
|
@ -37,7 +38,6 @@ CREATE TABLE "agent_files" ("id" varchar(16) PRIMARY KEY NOT NULL, "agentId" var
|
|||
| Name | Definition |
|
||||
| ---- | ---------- |
|
||||
| IDX_45dafc48fe2ce95eac30fc8ffd | CREATE INDEX "IDX_45dafc48fe2ce95eac30fc8ffd" ON "agent_files" ("agentId", "createdAt") |
|
||||
| IDX_agent_files_agentId_binaryDataId | CREATE UNIQUE INDEX "IDX_agent_files_agentId_binaryDataId" ON "agent_files" ("agentId", "binaryDataId") |
|
||||
| IDX_agent_files_agentId_fileName | CREATE UNIQUE INDEX "IDX_agent_files_agentId_fileName" ON "agent_files" ("agentId", "fileName") |
|
||||
| sqlite_autoindex_agent_files_1 | PRIMARY KEY (id) |
|
||||
|
||||
|
|
@ -50,12 +50,12 @@ erDiagram
|
|||
|
||||
"agent_files" {
|
||||
varchar_36_ agentId FK
|
||||
TEXT binaryDataId
|
||||
datetime_3_ createdAt
|
||||
varchar_255_ fileName
|
||||
INTEGER fileSizeBytes
|
||||
varchar_16_ id PK
|
||||
varchar_255_ mimeType
|
||||
VARCHAR_2_ storedAt
|
||||
datetime_3_ updatedAt
|
||||
}
|
||||
"agents" {
|
||||
|
|
|
|||
|
|
@ -140,12 +140,12 @@ erDiagram
|
|||
}
|
||||
"agent_files" {
|
||||
varchar_36_ agentId FK
|
||||
TEXT binaryDataId
|
||||
datetime_3_ createdAt
|
||||
varchar_255_ fileName
|
||||
INTEGER fileSizeBytes
|
||||
varchar_16_ id PK
|
||||
varchar_255_ mimeType
|
||||
VARCHAR_2_ storedAt
|
||||
datetime_3_ updatedAt
|
||||
}
|
||||
"agent_task_definition" {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,55 @@
|
|||
import type { MigrationContext, ReversibleMigration } from '../migration-types';
|
||||
|
||||
/**
|
||||
* Replaces BinaryDataService's opaque `binaryDataId` with a blob-storage
|
||||
* `storedAt` location. Existing knowledge-file rows are disposable (feature at
|
||||
* low rollout) and are deleted rather than migrated — the bytes they point at
|
||||
* are no longer readable through the new store.
|
||||
*/
|
||||
export class ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603 implements ReversibleMigration {
|
||||
async up({
|
||||
escape,
|
||||
runQuery,
|
||||
tablePrefix,
|
||||
schemaBuilder: { dropIndex, dropColumns },
|
||||
}: MigrationContext) {
|
||||
const agentFiles = escape.tableName('agent_files');
|
||||
const binaryData = escape.tableName('binary_data');
|
||||
const sourceType = escape.columnName('sourceType');
|
||||
const storedAt = escape.columnName('storedAt');
|
||||
|
||||
await runQuery(`DELETE FROM ${agentFiles}`);
|
||||
await runQuery(`DELETE FROM ${binaryData} WHERE ${sourceType} = 'agent_file'`);
|
||||
|
||||
await dropIndex('agent_files', ['agentId', 'binaryDataId'], { skipIfMissing: true });
|
||||
await dropColumns('agent_files', ['binaryDataId'], { recreatesOnSqlite: true });
|
||||
|
||||
await runQuery(
|
||||
`ALTER TABLE ${agentFiles} ADD COLUMN ${storedAt} VARCHAR(2) NOT NULL DEFAULT 'fs' ` +
|
||||
`CONSTRAINT "CHK_${tablePrefix}agent_files_storedAt" CHECK(${storedAt} IN ('fs', 's3', 'az'))`,
|
||||
);
|
||||
}
|
||||
|
||||
async down({
|
||||
escape,
|
||||
runQuery,
|
||||
schemaBuilder: { addColumns, createIndex, column },
|
||||
}: MigrationContext) {
|
||||
const agentFiles = escape.tableName('agent_files');
|
||||
const storedAt = escape.columnName('storedAt');
|
||||
|
||||
await runQuery(`DELETE FROM ${agentFiles}`);
|
||||
await runQuery(`ALTER TABLE ${agentFiles} DROP COLUMN ${storedAt}`);
|
||||
|
||||
await addColumns(
|
||||
'agent_files',
|
||||
[
|
||||
column('binaryDataId').text.notNull.comment(
|
||||
'Opaque BinaryDataService reference (mode-prefixed); not an FK to binary_data',
|
||||
),
|
||||
],
|
||||
{ recreatesOnSqlite: true },
|
||||
);
|
||||
await createIndex('agent_files', ['agentId', 'binaryDataId'], true);
|
||||
}
|
||||
}
|
||||
|
|
@ -230,6 +230,7 @@ import { AddStoredAtToAgentExecution1784815940110 } from '../common/178481594011
|
|||
import { AddInstanceCredentials1784815940111 } from '../common/1784815940111-AddInstanceCredentials';
|
||||
import { CreateAgentEvalTables1784815940112 } from '../common/1784815940112-CreateAgentEvalTables';
|
||||
import type { Migration } from '../migration-types';
|
||||
import { ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603 } from '../common/1784900752603-ReplaceAgentFileBinaryDataIdWithStoredAt';
|
||||
|
||||
export const postgresMigrations: Migration[] = [
|
||||
InitialMigration1587669153312,
|
||||
|
|
@ -463,4 +464,5 @@ export const postgresMigrations: Migration[] = [
|
|||
AddStoredAtToAgentExecution1784815940110,
|
||||
AddInstanceCredentials1784815940111,
|
||||
CreateAgentEvalTables1784815940112,
|
||||
ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603,
|
||||
];
|
||||
|
|
|
|||
|
|
@ -221,6 +221,7 @@ import { CreateWorkflowReviewRequestTables1784000000052 } from '../common/178400
|
|||
import { AddStoredAtToAgentExecution1784815940110 } from '../common/1784815940110-AddStoredAtToAgentExecution';
|
||||
import { AddInstanceCredentials1784815940111 } from '../common/1784815940111-AddInstanceCredentials';
|
||||
import { CreateAgentEvalTables1784815940112 } from '../common/1784815940112-CreateAgentEvalTables';
|
||||
import { ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603 } from '../common/1784900752603-ReplaceAgentFileBinaryDataIdWithStoredAt';
|
||||
|
||||
const sqliteMigrations: Migration[] = [
|
||||
InitialMigration1588102412422,
|
||||
|
|
@ -445,6 +446,7 @@ const sqliteMigrations: Migration[] = [
|
|||
AddStoredAtToAgentExecution1784815940110,
|
||||
AddInstanceCredentials1784815940111,
|
||||
CreateAgentEvalTables1784815940112,
|
||||
ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603,
|
||||
];
|
||||
|
||||
export { sqliteMigrations };
|
||||
|
|
|
|||
|
|
@ -18,7 +18,8 @@ import {
|
|||
const mobyDickFile: AgentKnowledgeFileReference = {
|
||||
file: 'moby-dick.txt',
|
||||
fileId: 'file-1',
|
||||
binaryDataId: 'filesystem-v2:agents/agent-1/knowledge-files/file-1/binary_data/uuid',
|
||||
agentId: 'agent-1',
|
||||
storedAt: 'fs',
|
||||
displayName: 'moby-dick.txt',
|
||||
mimeType: 'text/plain',
|
||||
fileSizeBytes: 123,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,104 @@
|
|||
import type { Logger } from '@n8n/backend-common';
|
||||
import { FsByteStore } from '@n8n/blob-storage';
|
||||
import { UnexpectedError } from 'n8n-workflow';
|
||||
import type { StorageConfig } from 'n8n-core';
|
||||
import { mkdtemp, rm } from 'node:fs/promises';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import { AgentKnowledgeFileStore } from '../agent-knowledge-file-store';
|
||||
|
||||
vi.unmock('node:fs');
|
||||
vi.unmock('node:fs/promises');
|
||||
|
||||
describe('AgentKnowledgeFileStore', () => {
|
||||
let storagePath: string;
|
||||
let storageConfig: StorageConfig;
|
||||
let logger: ReturnType<typeof mock<Logger>>;
|
||||
let store: AgentKnowledgeFileStore;
|
||||
|
||||
beforeAll(async () => {
|
||||
storagePath = await mkdtemp(join(tmpdir(), 'n8n-agent-knowledge-file-store-'));
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
for (const entry of await (await import('node:fs/promises')).readdir(storagePath)) {
|
||||
await rm(join(storagePath, entry), { recursive: true, force: true });
|
||||
}
|
||||
storageConfig = { storagePath, modeTag: 'db' } as StorageConfig;
|
||||
logger = mock<Logger>();
|
||||
const fsByteStore = new FsByteStore({
|
||||
storagePath,
|
||||
reportError: () => {},
|
||||
});
|
||||
store = new AgentKnowledgeFileStore(fsByteStore as never, storageConfig, logger);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await rm(storagePath, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it('falls back to fs when execution data storage mode is database and round-trips bytes', async () => {
|
||||
const body = Buffer.from('knowledge-bytes', 'utf-8');
|
||||
const ref = { agentId: 'agent-1', fileId: 'file-1' };
|
||||
|
||||
const storedAt = await store.write(ref, body, {
|
||||
fileName: 'notes.txt',
|
||||
mimeType: 'text/plain',
|
||||
});
|
||||
|
||||
expect(storedAt).toBe('fs');
|
||||
expect(logger.warn).toHaveBeenCalledWith(
|
||||
"Execution data storage mode is 'database'; agent knowledge files will be stored on the local filesystem",
|
||||
);
|
||||
await expect(store.readAsBuffer({ ...ref, storedAt: 'fs' })).resolves.toEqual(body);
|
||||
});
|
||||
|
||||
it('writes to a registered s3 location when modeTag is s3', async () => {
|
||||
storageConfig = { storagePath, modeTag: 's3' } as StorageConfig;
|
||||
const fsByteStore = new FsByteStore({ storagePath, reportError: () => {} });
|
||||
store = new AgentKnowledgeFileStore(fsByteStore as never, storageConfig, logger);
|
||||
|
||||
const s3Path = join(storagePath, 's3-root');
|
||||
const s3Store = new FsByteStore({ storagePath: s3Path, reportError: () => {} });
|
||||
store.registerByteStore('s3', s3Store);
|
||||
|
||||
const body = Buffer.from('s3-bytes', 'utf-8');
|
||||
const ref = { agentId: 'agent-1', fileId: 'file-2' };
|
||||
|
||||
const storedAt = await store.write(ref, body, { mimeType: 'text/plain' });
|
||||
|
||||
expect(storedAt).toBe('s3');
|
||||
await expect(store.readAsBuffer({ ...ref, storedAt: 's3' })).resolves.toEqual(body);
|
||||
});
|
||||
|
||||
it('returns null for a missing blob and throws for an unregistered storedAt', async () => {
|
||||
await expect(
|
||||
store.readAsBuffer({ agentId: 'agent-1', fileId: 'missing', storedAt: 'fs' }),
|
||||
).resolves.toBeNull();
|
||||
|
||||
await expect(
|
||||
store.readAsBuffer({ agentId: 'agent-1', fileId: 'file-1', storedAt: 's3' }),
|
||||
).rejects.toThrow(UnexpectedError);
|
||||
});
|
||||
|
||||
it('deletes registered blobs and skips unregistered locations without throwing', async () => {
|
||||
const kept = { agentId: 'agent-1', fileId: 'keep' };
|
||||
const body = Buffer.from('delete-me', 'utf-8');
|
||||
await store.write(kept, body, { mimeType: 'text/plain' });
|
||||
|
||||
await expect(
|
||||
store.delete([
|
||||
{ ...kept, storedAt: 'fs' },
|
||||
{ agentId: 'agent-1', fileId: 'orphan', storedAt: 's3' },
|
||||
]),
|
||||
).resolves.toBeUndefined();
|
||||
|
||||
await expect(store.readAsBuffer({ ...kept, storedAt: 'fs' })).resolves.toBeNull();
|
||||
expect(logger.warn).toHaveBeenCalledWith(
|
||||
'Skipped deleting agent knowledge files for unconfigured storage',
|
||||
expect.objectContaining({ storedAt: 's3', count: 1 }),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
@ -3,10 +3,11 @@ import type { Logger } from '@n8n/backend-common';
|
|||
import type { AgentsConfig } from '@n8n/config';
|
||||
import type { AiAssistantClient } from '@n8n_io/ai-assistant-sdk';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
import type { BinaryDataService, InstanceSettings } from 'n8n-core';
|
||||
import type { InstanceSettings } from 'n8n-core';
|
||||
|
||||
import type { AiService } from '../../../services/ai.service';
|
||||
|
||||
import type { AgentKnowledgeFileStore } from '../agent-knowledge-file-store';
|
||||
import type { Agent } from '../entities/agent.entity';
|
||||
import type { AgentFile } from '../entities/agent-file.entity';
|
||||
import { KNOWLEDGE_MIRROR_FILES_DIR } from '../agent-knowledge-storage';
|
||||
|
|
@ -102,10 +103,10 @@ function makePublishedAgentRepository(): ReturnType<typeof mock<AgentRepository>
|
|||
return repository;
|
||||
}
|
||||
|
||||
function makeBinaryDataService(): Mocked<BinaryDataService> {
|
||||
const service = mock<BinaryDataService>();
|
||||
service.getAsBuffer.mockResolvedValue(Buffer.from('mock file content'));
|
||||
return service;
|
||||
function makeKnowledgeFileStore(): Mocked<AgentKnowledgeFileStore> {
|
||||
const store = mock<AgentKnowledgeFileStore>();
|
||||
store.readAsBuffer.mockResolvedValue(Buffer.from('mock file content'));
|
||||
return store;
|
||||
}
|
||||
|
||||
function makeService(
|
||||
|
|
@ -115,7 +116,7 @@ function makeService(
|
|||
instanceSettings: InstanceSettings = mock<InstanceSettings>({ instanceId }),
|
||||
agentFileRepository: AgentFileRepository = mock<AgentFileRepository>(),
|
||||
agentRepository: AgentRepository = makePublishedAgentRepository(),
|
||||
binaryDataService: BinaryDataService = makeBinaryDataService(),
|
||||
agentKnowledgeFileStore: AgentKnowledgeFileStore = makeKnowledgeFileStore(),
|
||||
): AgentKnowledgeSandboxService {
|
||||
return new AgentKnowledgeSandboxService(
|
||||
{
|
||||
|
|
@ -134,7 +135,7 @@ function makeService(
|
|||
instanceSettings,
|
||||
agentFileRepository,
|
||||
agentRepository,
|
||||
binaryDataService,
|
||||
agentKnowledgeFileStore,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -144,7 +145,7 @@ function makeAgentFile(overrides: Partial<AgentFile> = {}): AgentFile {
|
|||
return {
|
||||
id,
|
||||
agentId,
|
||||
binaryDataId: `filesystem-v2:agents/${agentId}/knowledge-files/${id}/binary_data/uuid`,
|
||||
storedAt: 'fs',
|
||||
fileName,
|
||||
mimeType: 'text/plain',
|
||||
fileSizeBytes: 100,
|
||||
|
|
@ -547,7 +548,7 @@ describe('AgentKnowledgeSandboxService', () => {
|
|||
]);
|
||||
const agentRepository = makePublishedAgentRepository();
|
||||
agentRepository.existsBy.mockResolvedValue(true);
|
||||
const binaryDataService = makeBinaryDataService();
|
||||
const agentKnowledgeFileStore = makeKnowledgeFileStore();
|
||||
const service = makeService(
|
||||
{},
|
||||
mock<Logger>(),
|
||||
|
|
@ -555,30 +556,30 @@ describe('AgentKnowledgeSandboxService', () => {
|
|||
mock<InstanceSettings>({ instanceId }),
|
||||
agentFileRepository,
|
||||
agentRepository,
|
||||
binaryDataService,
|
||||
agentKnowledgeFileStore,
|
||||
);
|
||||
|
||||
await service.searchKnowledge(projectId, agentId, { pattern: 'foo' });
|
||||
let commands = sandbox.process.executeCommand.mock.calls.map(([command]) => command);
|
||||
expect(commands.filter(isManifestReadCommand)).toHaveLength(1);
|
||||
expect(commands.filter(isMirrorSyncCommand)).toHaveLength(1);
|
||||
expect(binaryDataService.getAsBuffer).toHaveBeenCalledTimes(2);
|
||||
expect(agentKnowledgeFileStore.readAsBuffer).toHaveBeenCalledTimes(2);
|
||||
expect(sandbox.fs.uploadFiles).toHaveBeenCalledTimes(1);
|
||||
manifestState = 'doc1.txt\ndoc2.txt\n';
|
||||
|
||||
sandbox.process.executeCommand.mockClear();
|
||||
binaryDataService.getAsBuffer.mockClear();
|
||||
agentKnowledgeFileStore.readAsBuffer.mockClear();
|
||||
sandbox.fs.uploadFiles.mockClear();
|
||||
await service.searchKnowledge(projectId, agentId, { pattern: 'bar' });
|
||||
commands = sandbox.process.executeCommand.mock.calls.map(([command]) => command);
|
||||
expect(commands.filter(isManifestReadCommand)).toHaveLength(0);
|
||||
expect(commands.filter(isMirrorSyncCommand)).toHaveLength(0);
|
||||
expect(commands).toHaveLength(1);
|
||||
expect(binaryDataService.getAsBuffer).not.toHaveBeenCalled();
|
||||
expect(agentKnowledgeFileStore.readAsBuffer).not.toHaveBeenCalled();
|
||||
expect(sandbox.fs.uploadFiles).not.toHaveBeenCalled();
|
||||
|
||||
sandbox.process.executeCommand.mockClear();
|
||||
binaryDataService.getAsBuffer.mockClear();
|
||||
agentKnowledgeFileStore.readAsBuffer.mockClear();
|
||||
sandbox.fs.uploadFiles.mockClear();
|
||||
agentFileRepository.findByAgentId.mockResolvedValue([
|
||||
makeMirrorFile('file-1', 'doc1.txt'),
|
||||
|
|
@ -593,7 +594,7 @@ describe('AgentKnowledgeSandboxService', () => {
|
|||
// Only the newly-added name should be fetched and staged for move —
|
||||
// the manifest rewrite (which always lists every expected name) is
|
||||
// a separate, later part of the finalize command.
|
||||
expect(binaryDataService.getAsBuffer).toHaveBeenCalledTimes(1);
|
||||
expect(agentKnowledgeFileStore.readAsBuffer).toHaveBeenCalledTimes(1);
|
||||
expect(syncCommands[0]).toContain('.tmp-doc3.txt');
|
||||
expect(syncCommands[0]).not.toContain('.tmp-doc1.txt');
|
||||
expect(syncCommands[0]).not.toContain('.tmp-doc2.txt');
|
||||
|
|
@ -614,8 +615,10 @@ describe('AgentKnowledgeSandboxService', () => {
|
|||
]);
|
||||
const agentRepository = makePublishedAgentRepository();
|
||||
agentRepository.existsBy.mockResolvedValue(true);
|
||||
const binaryDataService = makeBinaryDataService();
|
||||
binaryDataService.getAsBuffer.mockResolvedValue(Buffer.alloc(MIRROR_UPLOAD_BATCH_BYTES));
|
||||
const agentKnowledgeFileStore = makeKnowledgeFileStore();
|
||||
agentKnowledgeFileStore.readAsBuffer.mockResolvedValue(
|
||||
Buffer.alloc(MIRROR_UPLOAD_BATCH_BYTES),
|
||||
);
|
||||
const service = makeService(
|
||||
{},
|
||||
mock<Logger>(),
|
||||
|
|
@ -623,7 +626,7 @@ describe('AgentKnowledgeSandboxService', () => {
|
|||
mock<InstanceSettings>({ instanceId }),
|
||||
agentFileRepository,
|
||||
agentRepository,
|
||||
binaryDataService,
|
||||
agentKnowledgeFileStore,
|
||||
);
|
||||
|
||||
await service.searchKnowledge(projectId, agentId, { pattern: 'foo' });
|
||||
|
|
@ -691,7 +694,7 @@ describe('AgentKnowledgeSandboxService', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('skips a file that fails to load from binary data storage and retries it next sync', async () => {
|
||||
it('skips a file that fails to load from the knowledge file store and retries it next sync', async () => {
|
||||
const sandbox = makeSandbox('started');
|
||||
sandbox.process.executeCommand.mockResolvedValue({
|
||||
exitCode: 0,
|
||||
|
|
@ -702,8 +705,8 @@ describe('AgentKnowledgeSandboxService', () => {
|
|||
agentFileRepository.findByAgentId.mockResolvedValue([makeMirrorFile('file-1', 'doc1.txt')]);
|
||||
const agentRepository = makePublishedAgentRepository();
|
||||
agentRepository.existsBy.mockResolvedValue(true);
|
||||
const binaryDataService = makeBinaryDataService();
|
||||
binaryDataService.getAsBuffer.mockRejectedValueOnce(new Error('missing on disk'));
|
||||
const agentKnowledgeFileStore = makeKnowledgeFileStore();
|
||||
agentKnowledgeFileStore.readAsBuffer.mockRejectedValueOnce(new Error('missing on disk'));
|
||||
const logger = mock<Logger>();
|
||||
const service = makeService(
|
||||
{},
|
||||
|
|
@ -712,7 +715,7 @@ describe('AgentKnowledgeSandboxService', () => {
|
|||
mock<InstanceSettings>({ instanceId }),
|
||||
agentFileRepository,
|
||||
agentRepository,
|
||||
binaryDataService,
|
||||
agentKnowledgeFileStore,
|
||||
);
|
||||
|
||||
await expect(
|
||||
|
|
@ -727,12 +730,12 @@ describe('AgentKnowledgeSandboxService', () => {
|
|||
|
||||
// The failed file was left out of the cached manifest, so the next
|
||||
// sync's expected hash mismatches and it retries — this time
|
||||
// `getAsBuffer` succeeds (the mock's default resolves).
|
||||
// `readAsBuffer` succeeds (the mock's default resolves).
|
||||
await expect(
|
||||
service.searchKnowledge(projectId, agentId, { pattern: 'bar' }),
|
||||
).resolves.toBeDefined();
|
||||
|
||||
expect(binaryDataService.getAsBuffer).toHaveBeenCalledTimes(2);
|
||||
expect(agentKnowledgeFileStore.readAsBuffer).toHaveBeenCalledTimes(2);
|
||||
expect(sandbox.fs.uploadFiles).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -7,8 +7,8 @@ import { Readable } from 'node:stream';
|
|||
|
||||
import { MAX_AGENT_KNOWLEDGE_BASE_SIZE_BYTES } from '@n8n/api-types';
|
||||
import type { Logger } from '@n8n/backend-common';
|
||||
import type { BinaryDataConfig, BinaryDataService } from 'n8n-core';
|
||||
|
||||
import type { AgentKnowledgeFileStore } from '../agent-knowledge-file-store';
|
||||
import { AgentKnowledgeService } from '../agent-knowledge.service';
|
||||
import type { AgentKnowledgeSandboxService } from '../agent-knowledge-sandbox.service';
|
||||
import type { AgentFile } from '../entities/agent-file.entity';
|
||||
|
|
@ -28,10 +28,9 @@ vi.mock('@n8n/ai-utilities', () => ({
|
|||
}));
|
||||
|
||||
/**
|
||||
* Real `BinaryDataService` managers fully consume the stream before
|
||||
* resolving/rejecting; mock implementations must too, or an unconsumed
|
||||
* `createReadStream` lazily opens after the test's temp-file cleanup runs
|
||||
* and throws an unhandled ENOENT.
|
||||
* The knowledge file store writes streams by consuming them; mock
|
||||
* implementations must too, or an unconsumed `createReadStream` lazily opens
|
||||
* after the test's temp-file cleanup runs and throws an unhandled ENOENT.
|
||||
*/
|
||||
async function drainIfStream(content: unknown): Promise<void> {
|
||||
if (content instanceof Readable) {
|
||||
|
|
@ -65,7 +64,7 @@ function makeAgentFile(overrides: Partial<AgentFile> = {}): AgentFile {
|
|||
return {
|
||||
id: 'file-1',
|
||||
agentId,
|
||||
binaryDataId: 'filesystem-v2:agents/agent-1/knowledge-files/file-1/binary_data/uuid',
|
||||
storedAt: 'fs',
|
||||
fileName: 'first.txt',
|
||||
mimeType: 'text/plain',
|
||||
fileSizeBytes: 4,
|
||||
|
|
@ -84,7 +83,7 @@ class InMemoryAgentFileRepository {
|
|||
return {
|
||||
id: input.id ?? 'generated-id',
|
||||
agentId: input.agentId ?? agentId,
|
||||
binaryDataId: input.binaryDataId ?? '',
|
||||
storedAt: input.storedAt ?? 'fs',
|
||||
fileName: input.fileName ?? '',
|
||||
mimeType: input.mimeType ?? '',
|
||||
fileSizeBytes: input.fileSizeBytes ?? 0,
|
||||
|
|
@ -134,39 +133,33 @@ describe('AgentKnowledgeService', () => {
|
|||
let agentRepository: Mocked<AgentRepository>;
|
||||
let agentFileRepository: InMemoryAgentFileRepository;
|
||||
let agentKnowledgeSandboxService: Mocked<AgentKnowledgeSandboxService>;
|
||||
let binaryDataService: Mocked<BinaryDataService>;
|
||||
let binaryDataConfig: BinaryDataConfig;
|
||||
let agentKnowledgeFileStore: Mocked<AgentKnowledgeFileStore>;
|
||||
let logger: Mocked<Logger>;
|
||||
let service: AgentKnowledgeService;
|
||||
let storeSequence: number;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
storeSequence = 0;
|
||||
agentRepository = mock<AgentRepository>();
|
||||
agentFileRepository = new InMemoryAgentFileRepository();
|
||||
agentKnowledgeSandboxService = mock<AgentKnowledgeSandboxService>();
|
||||
binaryDataService = mock<BinaryDataService>();
|
||||
binaryDataService.store.mockImplementation(async (_location, content, binaryData) => {
|
||||
storeSequence += 1;
|
||||
agentKnowledgeFileStore = mock<AgentKnowledgeFileStore>();
|
||||
agentKnowledgeFileStore.write.mockImplementation(async (_ref, content) => {
|
||||
await drainIfStream(content);
|
||||
return { ...binaryData, id: `filesystem-v2:mock-${storeSequence}` };
|
||||
return 'fs';
|
||||
});
|
||||
binaryDataService.deleteManyByBinaryDataId.mockResolvedValue(undefined);
|
||||
binaryDataConfig = { mode: 'filesystem' } as BinaryDataConfig;
|
||||
agentKnowledgeFileStore.delete.mockResolvedValue(undefined);
|
||||
logger = mock<Logger>();
|
||||
service = new AgentKnowledgeService(
|
||||
agentRepository,
|
||||
agentFileRepository as unknown as AgentFileRepository,
|
||||
agentKnowledgeSandboxService,
|
||||
binaryDataService,
|
||||
binaryDataConfig,
|
||||
agentKnowledgeFileStore,
|
||||
logger,
|
||||
);
|
||||
loadMock.mockResolvedValue([{ pageContent: 'extracted pdf text' }]);
|
||||
});
|
||||
|
||||
it('stores text and PDF files via BinaryDataService, creates DB rows, and cleans temp files', async () => {
|
||||
it('stores text and PDF files via the knowledge file store, creates DB rows, and cleans temp files', async () => {
|
||||
agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never);
|
||||
const tempDirectory = await mkdtemp(path.join(tmpdir(), 'agent-knowledge-upload-'));
|
||||
const textFilePath = path.join(tempDirectory, 'notes.txt');
|
||||
|
|
@ -206,16 +199,9 @@ describe('AgentKnowledgeService', () => {
|
|||
}),
|
||||
]);
|
||||
|
||||
expect(binaryDataService.store).toHaveBeenCalledTimes(2);
|
||||
const [textCall, pdfCall] = binaryDataService.store.mock.calls;
|
||||
expect(textCall[0]).toEqual(
|
||||
expect.objectContaining({
|
||||
type: 'custom',
|
||||
pathSegments: ['agents', agentId, 'knowledge-files', expect.any(String)],
|
||||
sourceType: 'agent_file',
|
||||
sourceId: expect.any(String),
|
||||
}),
|
||||
);
|
||||
expect(agentKnowledgeFileStore.write).toHaveBeenCalledTimes(2);
|
||||
const [textCall, pdfCall] = agentKnowledgeFileStore.write.mock.calls;
|
||||
expect(textCall[0]).toEqual({ agentId, fileId: expect.any(String) });
|
||||
expect(textCall[1]).toBeInstanceOf(Readable);
|
||||
expect(textCall[2]).toEqual(
|
||||
expect.objectContaining({ mimeType: 'text/plain', fileName: 'notes.txt' }),
|
||||
|
|
@ -226,8 +212,8 @@ describe('AgentKnowledgeService', () => {
|
|||
);
|
||||
|
||||
const [storedTextFile, storedPdfFile] = agentFileRepository.all();
|
||||
expect(storedTextFile.binaryDataId).toBe('filesystem-v2:mock-1');
|
||||
expect(storedPdfFile.binaryDataId).toBe('filesystem-v2:mock-2');
|
||||
expect(storedTextFile.storedAt).toBe('fs');
|
||||
expect(storedPdfFile.storedAt).toBe('fs');
|
||||
|
||||
await expect(access(textFilePath)).rejects.toThrow();
|
||||
await expect(access(pdfFilePath)).rejects.toThrow();
|
||||
|
|
@ -241,7 +227,7 @@ describe('AgentKnowledgeService', () => {
|
|||
).rejects.toThrow('Invalid knowledge file name');
|
||||
|
||||
expect(agentFileRepository.all()).toEqual([]);
|
||||
expect(binaryDataService.store).not.toHaveBeenCalled();
|
||||
expect(agentKnowledgeFileStore.write).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('cleans up already-stored files when a later upload in the batch fails', async () => {
|
||||
|
|
@ -251,13 +237,13 @@ describe('AgentKnowledgeService', () => {
|
|||
const secondPath = path.join(tempDirectory, 'second.txt');
|
||||
await writeFile(firstPath, 'hello');
|
||||
await writeFile(secondPath, 'world');
|
||||
binaryDataService.store
|
||||
agentKnowledgeFileStore.write
|
||||
.mockReset()
|
||||
.mockImplementationOnce(async (_location, content, binaryData) => {
|
||||
.mockImplementationOnce(async (_ref, content) => {
|
||||
await drainIfStream(content);
|
||||
return { ...binaryData, id: 'filesystem-v2:mock-1' };
|
||||
return 'fs';
|
||||
})
|
||||
.mockImplementationOnce(async (_location, content) => {
|
||||
.mockImplementationOnce(async (_ref, content) => {
|
||||
await drainIfStream(content);
|
||||
throw new Error('store failed');
|
||||
});
|
||||
|
|
@ -275,48 +261,11 @@ describe('AgentKnowledgeService', () => {
|
|||
).rejects.toThrow('store failed');
|
||||
|
||||
expect(agentFileRepository.all()).toEqual([]);
|
||||
expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledWith([
|
||||
'filesystem-v2:mock-1',
|
||||
expect(agentKnowledgeFileStore.delete).toHaveBeenCalledWith([
|
||||
expect.objectContaining({ agentId, storedAt: 'fs' }),
|
||||
]);
|
||||
});
|
||||
|
||||
it('rejects uploads when the binary data service has no persisted storage mode', async () => {
|
||||
agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never);
|
||||
const tempDirectory = await mkdtemp(path.join(tmpdir(), 'agent-knowledge-upload-'));
|
||||
const filePath = path.join(tempDirectory, 'notes.txt');
|
||||
await writeFile(filePath, 'hello');
|
||||
binaryDataService.store.mockImplementationOnce(async (_location, content, binaryData) => {
|
||||
await drainIfStream(content);
|
||||
return { ...binaryData, id: undefined };
|
||||
});
|
||||
|
||||
await expect(
|
||||
service.uploadFiles(agentId, projectId, [
|
||||
makeMulterFile({ originalname: 'notes.txt', path: filePath, size: 5, buffer: undefined }),
|
||||
]),
|
||||
).rejects.toThrow('Agent knowledge base requires a persisted binary data storage mode');
|
||||
expect(agentFileRepository.all()).toEqual([]);
|
||||
});
|
||||
|
||||
it('rejects uploads up front when binary data mode is in-memory', async () => {
|
||||
agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never);
|
||||
binaryDataConfig = { mode: 'default' } as BinaryDataConfig;
|
||||
service = new AgentKnowledgeService(
|
||||
agentRepository,
|
||||
agentFileRepository as unknown as AgentFileRepository,
|
||||
agentKnowledgeSandboxService,
|
||||
binaryDataService,
|
||||
binaryDataConfig,
|
||||
logger,
|
||||
);
|
||||
|
||||
await expect(
|
||||
service.uploadFiles(agentId, projectId, [makeMulterFile({ originalname: 'notes.txt' })]),
|
||||
).rejects.toThrow('Agent knowledge base requires a persisted binary data storage mode');
|
||||
expect(binaryDataService.store).not.toHaveBeenCalled();
|
||||
expect(agentFileRepository.all()).toEqual([]);
|
||||
});
|
||||
|
||||
it('allows uploads that bring the knowledge base exactly to the size limit', async () => {
|
||||
agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never);
|
||||
await agentFileRepository.save(
|
||||
|
|
@ -347,7 +296,7 @@ describe('AgentKnowledgeService', () => {
|
|||
}),
|
||||
]);
|
||||
expect(agentFileRepository.all()).toHaveLength(2);
|
||||
expect(binaryDataService.store).toHaveBeenCalledTimes(1);
|
||||
expect(agentKnowledgeFileStore.write).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('rejects uploads that would exceed the knowledge base size limit', async () => {
|
||||
|
|
@ -369,7 +318,7 @@ describe('AgentKnowledgeService', () => {
|
|||
]),
|
||||
).rejects.toThrow('Knowledge base limit reached');
|
||||
expect(agentFileRepository.all()).toHaveLength(1);
|
||||
expect(binaryDataService.store).not.toHaveBeenCalled();
|
||||
expect(agentKnowledgeFileStore.write).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects uploads when existing knowledge files already exceed the size limit', async () => {
|
||||
|
|
@ -391,48 +340,42 @@ describe('AgentKnowledgeService', () => {
|
|||
]),
|
||||
).rejects.toThrow('Knowledge base limit reached');
|
||||
expect(agentFileRepository.all()).toHaveLength(1);
|
||||
expect(binaryDataService.store).not.toHaveBeenCalled();
|
||||
expect(agentKnowledgeFileStore.write).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('deletes the DB row and its binary data', async () => {
|
||||
it('deletes the DB row and its blob', async () => {
|
||||
agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never);
|
||||
await agentFileRepository.save(
|
||||
makeAgentFile({ id: 'file-1', binaryDataId: 'filesystem-v2:file-1.txt' }),
|
||||
);
|
||||
await agentFileRepository.save(makeAgentFile({ id: 'file-1', storedAt: 'fs' }));
|
||||
|
||||
await expect(service.deleteFile(agentId, projectId, 'file-1')).resolves.toBeUndefined();
|
||||
|
||||
expect(agentFileRepository.all()).toEqual([]);
|
||||
expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledWith([
|
||||
'filesystem-v2:file-1.txt',
|
||||
expect(agentKnowledgeFileStore.delete).toHaveBeenCalledWith([
|
||||
{ agentId, fileId: 'file-1', storedAt: 'fs' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('logs binary data deletion failures without restoring the DB row', async () => {
|
||||
it('logs blob deletion failures without restoring the DB row', async () => {
|
||||
agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never);
|
||||
await agentFileRepository.save(
|
||||
makeAgentFile({ id: 'file-1', binaryDataId: 'filesystem-v2:file-1.txt' }),
|
||||
);
|
||||
binaryDataService.deleteManyByBinaryDataId.mockRejectedValueOnce(new Error('delete failed'));
|
||||
await agentFileRepository.save(makeAgentFile({ id: 'file-1', storedAt: 'fs' }));
|
||||
agentKnowledgeFileStore.delete.mockRejectedValueOnce(new Error('delete failed'));
|
||||
|
||||
await expect(service.deleteFile(agentId, projectId, 'file-1')).resolves.toBeUndefined();
|
||||
|
||||
expect(agentFileRepository.all()).toEqual([]);
|
||||
expect(logger.warn).toHaveBeenCalledWith('Failed to delete knowledge file binary data', {
|
||||
expect(logger.warn).toHaveBeenCalledWith('Failed to delete knowledge file blob', {
|
||||
agentId,
|
||||
fileId: 'file-1',
|
||||
error: 'delete failed',
|
||||
});
|
||||
});
|
||||
|
||||
it('deletes all agent files and their binary data', async () => {
|
||||
await agentFileRepository.save(
|
||||
makeAgentFile({ id: 'file-1', binaryDataId: 'filesystem-v2:file-1.txt' }),
|
||||
);
|
||||
it('deletes all agent files and their blobs', async () => {
|
||||
await agentFileRepository.save(makeAgentFile({ id: 'file-1', storedAt: 'fs' }));
|
||||
await agentFileRepository.save(
|
||||
makeAgentFile({
|
||||
id: 'file-2',
|
||||
binaryDataId: 'filesystem-v2:file-2.md',
|
||||
storedAt: 's3',
|
||||
fileName: 'guide.md',
|
||||
mimeType: 'text/markdown',
|
||||
}),
|
||||
|
|
@ -441,11 +384,10 @@ describe('AgentKnowledgeService', () => {
|
|||
await expect(service.deleteAllFilesForAgent(projectId, agentId)).resolves.toBeUndefined();
|
||||
|
||||
expect(agentFileRepository.all()).toEqual([]);
|
||||
expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledWith([
|
||||
'filesystem-v2:file-1.txt',
|
||||
'filesystem-v2:file-2.md',
|
||||
expect(agentKnowledgeFileStore.delete).toHaveBeenCalledWith([
|
||||
{ agentId, fileId: 'file-1', storedAt: 'fs' },
|
||||
{ agentId, fileId: 'file-2', storedAt: 's3' },
|
||||
]);
|
||||
expect(binaryDataService.deleteMany).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
describe('published-only gating', () => {
|
||||
|
|
@ -460,7 +402,7 @@ describe('AgentKnowledgeService', () => {
|
|||
'Knowledge base is only available for published agents. Publish the agent first.',
|
||||
);
|
||||
expect(agentFileRepository.all()).toEqual([]);
|
||||
expect(binaryDataService.store).not.toHaveBeenCalled();
|
||||
expect(agentKnowledgeFileStore.write).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects warmup for an unpublished agent', async () => {
|
||||
|
|
|
|||
121
packages/cli/src/modules/agents/agent-knowledge-file-store.ts
Normal file
121
packages/cli/src/modules/agents/agent-knowledge-file-store.ts
Normal file
|
|
@ -0,0 +1,121 @@
|
|||
import { Logger } from '@n8n/backend-common';
|
||||
import {
|
||||
FsByteStore,
|
||||
type ByteStore,
|
||||
type PreWriteBlobMetadata,
|
||||
type StorageLocation,
|
||||
} from '@n8n/blob-storage';
|
||||
import { Service } from '@n8n/di';
|
||||
import { ErrorReporter, StorageConfig } from 'n8n-core';
|
||||
import { UnexpectedError } from 'n8n-workflow';
|
||||
import type { Readable } from 'node:stream';
|
||||
|
||||
export type AgentKnowledgeFileRef = { agentId: string; fileId: string };
|
||||
export type StoredAgentKnowledgeFileRef = AgentKnowledgeFileRef & { storedAt: StorageLocation };
|
||||
|
||||
@Service()
|
||||
export class AgentKnowledgeFileFsByteStore extends FsByteStore {
|
||||
constructor(storageConfig: StorageConfig, errorReporter: ErrorReporter) {
|
||||
super({
|
||||
storagePath: storageConfig.storagePath,
|
||||
reportError: (error) => errorReporter.error(error),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stores agent knowledge file bytes via pluggable backends. The `fs` backend is
|
||||
* always available; `s3` and `az` are registered at module init when configured.
|
||||
* When execution data storage mode is `database`, writes fall back to `fs`.
|
||||
*/
|
||||
@Service()
|
||||
export class AgentKnowledgeFileStore {
|
||||
private readonly byteStores = new Map<StorageLocation, ByteStore>();
|
||||
private warnedAboutDbFallback = false;
|
||||
|
||||
constructor(
|
||||
fsByteStore: AgentKnowledgeFileFsByteStore,
|
||||
private readonly storageConfig: StorageConfig,
|
||||
private readonly logger: Logger,
|
||||
) {
|
||||
this.byteStores.set('fs', fsByteStore);
|
||||
}
|
||||
|
||||
registerByteStore(loc: StorageLocation, store: ByteStore) {
|
||||
this.byteStores.set(loc, store);
|
||||
}
|
||||
|
||||
hasLocation(loc: StorageLocation) {
|
||||
return this.byteStores.has(loc);
|
||||
}
|
||||
|
||||
get writeLocation(): StorageLocation {
|
||||
if (this.storageConfig.modeTag === 'db') {
|
||||
if (!this.warnedAboutDbFallback) {
|
||||
this.warnedAboutDbFallback = true;
|
||||
this.logger.warn(
|
||||
"Execution data storage mode is 'database'; agent knowledge files will be stored on the local filesystem",
|
||||
);
|
||||
}
|
||||
return 'fs';
|
||||
}
|
||||
return this.storageConfig.modeTag;
|
||||
}
|
||||
|
||||
async write(
|
||||
ref: AgentKnowledgeFileRef,
|
||||
body: Buffer | Readable,
|
||||
metadata: PreWriteBlobMetadata,
|
||||
): Promise<StorageLocation> {
|
||||
const loc = this.writeLocation;
|
||||
await this.getByteStore(loc).write(this.key(ref), body, metadata);
|
||||
return loc;
|
||||
}
|
||||
|
||||
async readAsBuffer(ref: StoredAgentKnowledgeFileRef): Promise<Buffer | null> {
|
||||
return await this.getByteStore(ref.storedAt).read(this.key(ref));
|
||||
}
|
||||
|
||||
async delete(refs: StoredAgentKnowledgeFileRef[]): Promise<void> {
|
||||
if (refs.length === 0) return;
|
||||
|
||||
const groups = new Map<StorageLocation, StoredAgentKnowledgeFileRef[]>();
|
||||
for (const ref of refs) {
|
||||
const group = groups.get(ref.storedAt) ?? [];
|
||||
group.push(ref);
|
||||
groups.set(ref.storedAt, group);
|
||||
}
|
||||
|
||||
await Promise.all(
|
||||
[...groups].map(async ([loc, group]) => {
|
||||
const store = this.byteStores.get(loc);
|
||||
if (!store) {
|
||||
this.logger.warn('Skipped deleting agent knowledge files for unconfigured storage', {
|
||||
storedAt: loc,
|
||||
count: group.length,
|
||||
});
|
||||
return;
|
||||
}
|
||||
await store.delete(group.map((ref) => this.key(ref)));
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
private key(ref: AgentKnowledgeFileRef) {
|
||||
return [
|
||||
'agents',
|
||||
encodeURIComponent(ref.agentId),
|
||||
'knowledge-files',
|
||||
encodeURIComponent(ref.fileId),
|
||||
'content',
|
||||
].join('/');
|
||||
}
|
||||
|
||||
private getByteStore(loc: StorageLocation): ByteStore {
|
||||
const store = this.byteStores.get(loc);
|
||||
if (!store) {
|
||||
throw new UnexpectedError(`Knowledge file store for location "${loc}" is not configured.`);
|
||||
}
|
||||
return store;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +1,8 @@
|
|||
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
|
||||
import type { StorageLocation } from '@n8n/blob-storage';
|
||||
import { z } from 'zod';
|
||||
|
||||
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
|
||||
|
||||
import { hasControlCharacter } from './agent-knowledge-storage';
|
||||
|
||||
const MAX_SEARCH_PATTERN_LENGTH = 500;
|
||||
|
|
@ -186,7 +188,8 @@ export type ReadKnowledgeRequest = z.infer<typeof readKnowledgeInputSchema>;
|
|||
export interface AgentKnowledgeFileReference {
|
||||
file: string;
|
||||
fileId: string;
|
||||
binaryDataId: string;
|
||||
agentId: string;
|
||||
storedAt: StorageLocation;
|
||||
displayName: string;
|
||||
mimeType: string;
|
||||
fileSizeBytes: number;
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import { Logger } from '@n8n/backend-common';
|
|||
import { AgentsConfig } from '@n8n/config';
|
||||
import { Time } from '@n8n/constants';
|
||||
import { Service } from '@n8n/di';
|
||||
import { BinaryDataService, InstanceSettings } from 'n8n-core';
|
||||
import { InstanceSettings } from 'n8n-core';
|
||||
import { OperationalError } from 'n8n-workflow';
|
||||
import { nanoid } from 'nanoid';
|
||||
import { createHash } from 'node:crypto';
|
||||
|
|
@ -30,6 +30,7 @@ import {
|
|||
parseRipgrepFilesOutput,
|
||||
parseRipgrepOutput,
|
||||
} from './agent-knowledge-commands';
|
||||
import { AgentKnowledgeFileStore } from './agent-knowledge-file-store';
|
||||
import { isAgentKnowledgeBaseEnabled } from './agent-knowledge-gate';
|
||||
import {
|
||||
assertValidKnowledgeFilePath,
|
||||
|
|
@ -216,7 +217,7 @@ export class AgentKnowledgeSandboxService {
|
|||
private readonly instanceSettings: InstanceSettings,
|
||||
private readonly agentFileRepository: AgentFileRepository,
|
||||
private readonly agentRepository: AgentRepository,
|
||||
private readonly binaryDataService: BinaryDataService,
|
||||
private readonly agentKnowledgeFileStore: AgentKnowledgeFileStore,
|
||||
) {}
|
||||
|
||||
async warmSandbox(projectId: string, agentId: string): Promise<void> {
|
||||
|
|
@ -497,8 +498,8 @@ export class AgentKnowledgeSandboxService {
|
|||
|
||||
const filesByName = new Map(files.map((file) => [file.file, file]));
|
||||
const copiedNames = await this.uploadMirrorFiles(sandbox, toCopy, filesByName, sandboxName);
|
||||
// Files that failed to load from BinaryDataService are left out of the
|
||||
// manifest, so the next sync attempt retries them as `toCopy` again.
|
||||
// Files that failed to load from the knowledge file store are left out of
|
||||
// the manifest, so the next sync attempt retries them as `toCopy` again.
|
||||
const finalManifestNames = expectedNames.filter(
|
||||
(name) => copiedNames.has(name) || presentSet.has(name),
|
||||
);
|
||||
|
|
@ -521,11 +522,11 @@ export class AgentKnowledgeSandboxService {
|
|||
}
|
||||
|
||||
/**
|
||||
* Fetches each `names` entry from BinaryDataService and uploads it to the
|
||||
* sandbox mirror under a `.tmp-` prefix; `buildMirrorFinalizeCommand` moves
|
||||
* it into place so a concurrent search never sees a partially-written file.
|
||||
* Uploads flush in `MIRROR_UPLOAD_BATCH_BYTES`-sized batches so the whole
|
||||
* knowledge base (up to 1.5 GB) is never held in memory at once.
|
||||
* Fetches each `names` entry from the knowledge file store and uploads it to
|
||||
* the sandbox mirror under a `.tmp-` prefix; `buildMirrorFinalizeCommand`
|
||||
* moves it into place so a concurrent search never sees a partially-written
|
||||
* file. Uploads flush in `MIRROR_UPLOAD_BATCH_BYTES`-sized batches so the
|
||||
* whole knowledge base (up to 1.5 GB) is never held in memory at once.
|
||||
* Returns the subset of `names` that were fetched and uploaded successfully.
|
||||
*/
|
||||
private async uploadMirrorFiles(
|
||||
|
|
@ -557,11 +558,19 @@ export class AgentKnowledgeSandboxService {
|
|||
if (!file) continue;
|
||||
|
||||
try {
|
||||
const buffer = await this.binaryDataService.getAsBuffer({
|
||||
id: file.binaryDataId,
|
||||
data: '',
|
||||
mimeType: file.mimeType,
|
||||
const buffer = await this.agentKnowledgeFileStore.readAsBuffer({
|
||||
agentId: file.agentId,
|
||||
fileId: file.fileId,
|
||||
storedAt: file.storedAt,
|
||||
});
|
||||
if (!buffer) {
|
||||
this.logger.warn('Failed to load agent knowledge file for mirror sync', {
|
||||
sandboxName,
|
||||
file: name,
|
||||
error: 'not found',
|
||||
});
|
||||
continue;
|
||||
}
|
||||
batch.push({ source: buffer, destination: `${KNOWLEDGE_MIRROR_FILES_DIR}/.tmp-${name}` });
|
||||
batchNames.push(name);
|
||||
batchBytes += buffer.length;
|
||||
|
|
@ -639,7 +648,8 @@ export class AgentKnowledgeSandboxService {
|
|||
return files.map((file) => ({
|
||||
file: storageFileNameForOriginalFileName(file.fileName),
|
||||
fileId: file.id,
|
||||
binaryDataId: file.binaryDataId,
|
||||
agentId: file.agentId,
|
||||
storedAt: file.storedAt,
|
||||
displayName: file.fileName,
|
||||
mimeType: file.mimeType,
|
||||
fileSizeBytes: file.fileSizeBytes,
|
||||
|
|
|
|||
|
|
@ -1,6 +1,4 @@
|
|||
import type { AgentFileDto } from '@n8n/api-types';
|
||||
import type { SourceType } from '@n8n/db';
|
||||
import { FileLocation } from 'n8n-core';
|
||||
import path from 'node:path';
|
||||
|
||||
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
|
||||
|
|
@ -8,15 +6,11 @@ import { BadRequestError } from '@/errors/response-errors/bad-request.error';
|
|||
import type { AgentFile } from './entities/agent-file.entity';
|
||||
|
||||
// Local sandbox disk mirroring the DB-backed knowledge files, so repeated
|
||||
// reads/searches avoid re-fetching from BinaryDataService each time.
|
||||
// reads/searches avoid re-fetching from the knowledge file store each time.
|
||||
export const KNOWLEDGE_MIRROR_DIR = '/home/daytona/knowledge-mirror';
|
||||
export const KNOWLEDGE_MIRROR_FILES_DIR = `${KNOWLEDGE_MIRROR_DIR}/files`;
|
||||
export const KNOWLEDGE_MIRROR_MANIFEST = `${KNOWLEDGE_MIRROR_DIR}/manifest`;
|
||||
|
||||
// Typed against `SourceType` so a drift from the `binary_data` schema enum
|
||||
// (see `packages/@n8n/db/src/entities/binary-data-file.ts`) is a compile error.
|
||||
const AGENT_FILE_SOURCE_TYPE: SourceType = 'agent_file';
|
||||
|
||||
export function hasControlCharacter(value: string): boolean {
|
||||
for (const character of value) {
|
||||
if (character.charCodeAt(0) < 32) {
|
||||
|
|
@ -33,15 +27,6 @@ function sanitizePathCharacter(character: string): string {
|
|||
return character;
|
||||
}
|
||||
|
||||
/** One directory per file, so deleting a single file never touches others. */
|
||||
export function buildKnowledgeFileLocation(agentId: string, fileId: string) {
|
||||
return FileLocation.ofCustom({
|
||||
pathSegments: ['agents', agentId, 'knowledge-files', fileId],
|
||||
sourceType: AGENT_FILE_SOURCE_TYPE,
|
||||
sourceId: fileId,
|
||||
});
|
||||
}
|
||||
|
||||
export function assertKnowledgePathSegment(segment: string, label: string): void {
|
||||
if (
|
||||
!segment ||
|
||||
|
|
|
|||
|
|
@ -5,11 +5,10 @@ import {
|
|||
} from '@n8n/api-types';
|
||||
import { N8nPdfLoader } from '@n8n/ai-utilities';
|
||||
import { Logger } from '@n8n/backend-common';
|
||||
import type { StorageLocation } from '@n8n/blob-storage';
|
||||
import { Service } from '@n8n/di';
|
||||
import { QueryFailedError } from '@n8n/typeorm';
|
||||
import { generateNanoId } from '@n8n/utils/generate-nano-id';
|
||||
import { BinaryDataConfig, BinaryDataService } from 'n8n-core';
|
||||
import { OperationalError, type IBinaryData } from 'n8n-workflow';
|
||||
import { createReadStream } from 'node:fs';
|
||||
import { unlink } from 'node:fs/promises';
|
||||
import path from 'node:path';
|
||||
|
|
@ -17,11 +16,8 @@ import type { Readable } from 'node:stream';
|
|||
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
|
||||
import { NotFoundError } from '@/errors/response-errors/not-found.error';
|
||||
|
||||
import {
|
||||
buildKnowledgeFileLocation,
|
||||
storageFileNameForOriginalFileName,
|
||||
toAgentFileDto,
|
||||
} from './agent-knowledge-storage';
|
||||
import { AgentKnowledgeFileStore } from './agent-knowledge-file-store';
|
||||
import { storageFileNameForOriginalFileName, toAgentFileDto } from './agent-knowledge-storage';
|
||||
import { AgentKnowledgeSandboxService } from './agent-knowledge-sandbox.service';
|
||||
import type { AgentFile } from './entities/agent-file.entity';
|
||||
import type { Agent } from './entities/agent.entity';
|
||||
|
|
@ -52,8 +48,7 @@ export class AgentKnowledgeService {
|
|||
private readonly agentRepository: AgentRepository,
|
||||
private readonly agentFileRepository: AgentFileRepository,
|
||||
private readonly agentKnowledgeSandboxService: AgentKnowledgeSandboxService,
|
||||
private readonly binaryDataService: BinaryDataService,
|
||||
private readonly binaryDataConfig: BinaryDataConfig,
|
||||
private readonly agentKnowledgeFileStore: AgentKnowledgeFileStore,
|
||||
private readonly logger: Logger,
|
||||
) {}
|
||||
|
||||
|
|
@ -65,11 +60,6 @@ export class AgentKnowledgeService {
|
|||
try {
|
||||
const agent = await this.ensureAgentBelongsToProject(agentId, projectId);
|
||||
this.assertAgentPublished(agent);
|
||||
if (this.binaryDataConfig.mode === 'default') {
|
||||
throw new OperationalError(
|
||||
'Agent knowledge base requires a persisted binary data storage mode',
|
||||
);
|
||||
}
|
||||
this.validateUploadMetadata(files);
|
||||
await this.validateUploadBatch(agentId, files);
|
||||
|
||||
|
|
@ -113,13 +103,15 @@ export class AgentKnowledgeService {
|
|||
}
|
||||
|
||||
await this.agentFileRepository.delete({ id: fileId, agentId });
|
||||
await this.binaryDataService.deleteManyByBinaryDataId([file.binaryDataId]).catch((error) => {
|
||||
this.logger.warn('Failed to delete knowledge file binary data', {
|
||||
agentId,
|
||||
fileId: file.id,
|
||||
error: error instanceof Error ? error.message : error,
|
||||
await this.agentKnowledgeFileStore
|
||||
.delete([{ agentId, fileId: file.id, storedAt: file.storedAt }])
|
||||
.catch((error) => {
|
||||
this.logger.warn('Failed to delete knowledge file blob', {
|
||||
agentId,
|
||||
fileId: file.id,
|
||||
error: error instanceof Error ? error.message : error,
|
||||
});
|
||||
});
|
||||
});
|
||||
this.agentKnowledgeSandboxService.invalidateMirror(projectId, agentId);
|
||||
this.agentKnowledgeSandboxService.prewarmMirrorInBackground(projectId, agentId);
|
||||
}
|
||||
|
|
@ -128,10 +120,16 @@ export class AgentKnowledgeService {
|
|||
const files = await this.agentFileRepository.findByAgentId(agentId);
|
||||
await this.agentFileRepository.delete({ agentId });
|
||||
if (files.length > 0) {
|
||||
await this.binaryDataService
|
||||
.deleteManyByBinaryDataId(files.map((file) => file.binaryDataId))
|
||||
await this.agentKnowledgeFileStore
|
||||
.delete(
|
||||
files.map((file) => ({
|
||||
agentId,
|
||||
fileId: file.id,
|
||||
storedAt: file.storedAt,
|
||||
})),
|
||||
)
|
||||
.catch((error) => {
|
||||
this.logger.warn('Failed to delete knowledge files binary data', {
|
||||
this.logger.warn('Failed to delete knowledge file blobs', {
|
||||
agentId,
|
||||
error: error instanceof Error ? error.message : error,
|
||||
});
|
||||
|
|
@ -145,32 +143,21 @@ export class AgentKnowledgeService {
|
|||
await this.agentKnowledgeSandboxService.destroySandbox(projectId, agentId);
|
||||
}
|
||||
|
||||
/** Stores the file's bytes via BinaryDataService, then reserves its DB row. */
|
||||
/** Stores the file's bytes via AgentKnowledgeFileStore, then reserves its DB row. */
|
||||
private async storeAgentFile(agentId: string, file: Express.Multer.File): Promise<AgentFile> {
|
||||
const fileId = generateNanoId();
|
||||
const storageFileName = storageFileNameForOriginalFileName(file.originalname);
|
||||
const content = await this.prepareUploadContent(file);
|
||||
|
||||
const binaryData: IBinaryData = {
|
||||
data: '',
|
||||
mimeType: file.mimetype,
|
||||
const storedAt = await this.agentKnowledgeFileStore.write({ agentId, fileId }, content, {
|
||||
fileName: storageFileName,
|
||||
};
|
||||
const stored = await this.binaryDataService.store(
|
||||
buildKnowledgeFileLocation(agentId, fileId),
|
||||
content,
|
||||
binaryData,
|
||||
);
|
||||
if (!stored.id) {
|
||||
throw new OperationalError(
|
||||
'Agent knowledge base requires a persisted binary data storage mode',
|
||||
);
|
||||
}
|
||||
mimeType: file.mimetype,
|
||||
});
|
||||
|
||||
try {
|
||||
return await this.saveAgentFile(agentId, fileId, file, stored.id);
|
||||
return await this.saveAgentFile(agentId, fileId, file, storedAt);
|
||||
} catch (error) {
|
||||
await this.binaryDataService.deleteManyByBinaryDataId([stored.id]).catch(() => {});
|
||||
await this.agentKnowledgeFileStore.delete([{ agentId, fileId, storedAt }]).catch(() => {});
|
||||
if (isUniqueConstraintError(error)) {
|
||||
throw this.duplicateFileNameError(file.originalname);
|
||||
}
|
||||
|
|
@ -182,12 +169,12 @@ export class AgentKnowledgeService {
|
|||
agentId: string,
|
||||
fileId: string,
|
||||
file: Express.Multer.File,
|
||||
binaryDataId: string,
|
||||
storedAt: StorageLocation,
|
||||
): Promise<AgentFile> {
|
||||
const agentFile = this.agentFileRepository.create({
|
||||
id: fileId,
|
||||
agentId,
|
||||
binaryDataId,
|
||||
storedAt,
|
||||
fileName: file.originalname,
|
||||
mimeType: file.mimetype,
|
||||
fileSizeBytes: file.size,
|
||||
|
|
@ -212,7 +199,9 @@ export class AgentKnowledgeService {
|
|||
|
||||
private async cleanupUploadedFiles(files: AgentFile[]): Promise<void> {
|
||||
for (const file of files) {
|
||||
await this.binaryDataService.deleteManyByBinaryDataId([file.binaryDataId]).catch(() => {});
|
||||
await this.agentKnowledgeFileStore
|
||||
.delete([{ agentId: file.agentId, fileId: file.id, storedAt: file.storedAt }])
|
||||
.catch(() => {});
|
||||
await this.agentFileRepository.delete({ id: file.id, agentId: file.agentId }).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -34,31 +34,32 @@ export class AgentsModule implements ModuleInterface {
|
|||
const { AgentExecutionService } = await import('./agent-execution.service.js');
|
||||
Container.get(AgentExecutionService);
|
||||
|
||||
// Register blob backends for agent execution logs. The fs backend is always
|
||||
// available; s3/az reuse the clients base-command already initialized
|
||||
// (initBinaryDataService runs before module init in all commands) and are
|
||||
// only registered when that init succeeded — mirroring ExecutionDataJsonStore,
|
||||
// so a configured-but-unreachable backend degrades instead of throwing.
|
||||
// Register blob backends for agent execution logs and knowledge files.
|
||||
// The fs backend is always available; s3/az reuse the clients base-command
|
||||
// already initialized (initBinaryDataService runs before module init in
|
||||
// all commands) and are only registered when that init succeeded —
|
||||
// mirroring ExecutionDataJsonStore, so a configured-but-unreachable
|
||||
// backend degrades instead of throwing.
|
||||
const { AzureByteStore, S3ByteStore } = await import('@n8n/blob-storage');
|
||||
const { AgentExecutionLogStore } = await import('./execution-log/agent-execution-log-store.js');
|
||||
const { AgentKnowledgeFileStore } = await import('./agent-knowledge-file-store.js');
|
||||
const { ExecutionDataJsonStore } = await import(
|
||||
'@/executions/execution-data/execution-data-json-store.js'
|
||||
);
|
||||
const agentExecutionLogStore = Container.get(AgentExecutionLogStore);
|
||||
const agentKnowledgeFileStore = Container.get(AgentKnowledgeFileStore);
|
||||
const executionDataJsonStore = Container.get(ExecutionDataJsonStore);
|
||||
if (executionDataJsonStore.hasLocation('s3')) {
|
||||
const { ObjectStoreService } = await import('@n8n/blob-storage/object-store');
|
||||
agentExecutionLogStore.registerByteStore(
|
||||
's3',
|
||||
new S3ByteStore(Container.get(ObjectStoreService)),
|
||||
);
|
||||
const s3Store = new S3ByteStore(Container.get(ObjectStoreService));
|
||||
agentExecutionLogStore.registerByteStore('s3', s3Store);
|
||||
agentKnowledgeFileStore.registerByteStore('s3', s3Store);
|
||||
}
|
||||
if (executionDataJsonStore.hasLocation('az')) {
|
||||
const { AzureBlobService } = await import('@n8n/blob-storage/azure-blob');
|
||||
agentExecutionLogStore.registerByteStore(
|
||||
'az',
|
||||
new AzureByteStore(Container.get(AzureBlobService)),
|
||||
);
|
||||
const azStore = new AzureByteStore(Container.get(AzureBlobService));
|
||||
agentExecutionLogStore.registerByteStore('az', azStore);
|
||||
agentKnowledgeFileStore.registerByteStore('az', azStore);
|
||||
}
|
||||
|
||||
const { AgentRuntimeCacheService } = await import('./agent-runtime-cache.service.js');
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import type { StorageLocation } from '@n8n/blob-storage';
|
||||
import { WithTimestampsAndStringId } from '@n8n/db';
|
||||
import { Column, Entity, Index, JoinColumn, ManyToOne, type Relation } from '@n8n/typeorm';
|
||||
|
||||
|
|
@ -6,7 +7,6 @@ import { Agent } from './agent.entity';
|
|||
@Entity({ name: 'agent_files' })
|
||||
@Index(['agentId', 'createdAt'])
|
||||
@Index(['agentId', 'fileName'], { unique: true })
|
||||
@Index(['agentId', 'binaryDataId'], { unique: true })
|
||||
export class AgentFile extends WithTimestampsAndStringId {
|
||||
@ManyToOne(() => Agent, { onDelete: 'CASCADE' })
|
||||
@JoinColumn({ name: 'agentId' })
|
||||
|
|
@ -16,17 +16,16 @@ export class AgentFile extends WithTimestampsAndStringId {
|
|||
agentId: string;
|
||||
|
||||
/**
|
||||
* BinaryDataService id (e.g. `filesystem-v2:agents/<agentId>/knowledge-files/<fileId>/binary_data/<uuid>`).
|
||||
* Not a DB FK — see `BinaryDataService` in `n8n-core` for how the bytes are resolved from this id.
|
||||
* Blob-storage location for the file bytes: 'fs', 's3', or 'az'. Never 'db'.
|
||||
* Bytes are resolved via AgentKnowledgeFileStore.
|
||||
*/
|
||||
@Column({ type: 'text' })
|
||||
binaryDataId: string;
|
||||
@Column({ type: 'varchar', length: 2, nullable: false, default: 'fs' })
|
||||
storedAt: StorageLocation;
|
||||
|
||||
// fileName/mimeType/fileSizeBytes are intentionally denormalized rather than
|
||||
// joined from binary_data: (1) binaryDataId is an opaque storage reference,
|
||||
// not an FK; (2) we keep the original user-facing values, which differ from
|
||||
// the stored binary for converted uploads (a PDF is stored as extracted
|
||||
// text with a different byte size).
|
||||
// joined from storage metadata: we keep the original user-facing values,
|
||||
// which differ from the stored bytes for converted uploads (a PDF is stored
|
||||
// as extracted text with a different byte size).
|
||||
@Column({ type: 'varchar', length: 255 })
|
||||
fileName: string;
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,144 @@
|
|||
import {
|
||||
createTestMigrationContext,
|
||||
initDbUpToMigration,
|
||||
runSingleMigration,
|
||||
type TestMigrationContext,
|
||||
} from '@n8n/backend-test-utils';
|
||||
import { DbConnection } from '@n8n/db';
|
||||
import { Container } from '@n8n/di';
|
||||
import { DataSource } from '@n8n/typeorm';
|
||||
|
||||
const MIGRATION_NAME = 'ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603';
|
||||
|
||||
async function columnNames(context: TestMigrationContext, table: string): Promise<string[]> {
|
||||
if (context.isSqlite) {
|
||||
const rows = await context.runQuery<Array<{ name: string }>>(
|
||||
`PRAGMA table_info(${context.escape.tableName(table)})`,
|
||||
);
|
||||
return rows.map((row) => row.name);
|
||||
}
|
||||
const fullName = `${context.tablePrefix}${table}`;
|
||||
const rows = await context.runQuery<Array<{ column_name: string }>>(
|
||||
`SELECT column_name FROM information_schema.columns WHERE table_name = :name`,
|
||||
{ name: fullName },
|
||||
);
|
||||
return rows.map((row) => row.column_name);
|
||||
}
|
||||
|
||||
describe('ReplaceAgentFileBinaryDataIdWithStoredAt Migration', () => {
|
||||
let dataSource: DataSource;
|
||||
|
||||
beforeAll(async () => {
|
||||
const dbConnection = Container.get(DbConnection);
|
||||
await dbConnection.init();
|
||||
|
||||
dataSource = Container.get(DataSource);
|
||||
|
||||
const context = createTestMigrationContext(dataSource);
|
||||
await context.queryRunner.clearDatabase();
|
||||
await context.queryRunner.release();
|
||||
|
||||
await initDbUpToMigration(MIGRATION_NAME);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
const dbConnection = Container.get(DbConnection);
|
||||
await dbConnection.close();
|
||||
});
|
||||
|
||||
async function insertProject(context: TestMigrationContext, id: string): Promise<void> {
|
||||
const table = context.escape.tableName('project');
|
||||
const now = new Date();
|
||||
await context.runQuery(
|
||||
`INSERT INTO ${table} ("id", "name", "type", "customTelemetryTags", "createdAt", "updatedAt") VALUES (:id, :name, :type, :tags, :createdAt, :updatedAt)`,
|
||||
{ id, name: 'Test Project', type: 'team', tags: '[]', createdAt: now, updatedAt: now },
|
||||
);
|
||||
}
|
||||
|
||||
async function insertAgent(
|
||||
context: TestMigrationContext,
|
||||
data: { id: string; projectId: string },
|
||||
): Promise<void> {
|
||||
const table = context.escape.tableName('agents');
|
||||
const now = new Date();
|
||||
await context.runQuery(
|
||||
`INSERT INTO ${table} ("id", "name", "projectId", "integrations", "tools", "skills", "createdAt", "updatedAt") VALUES (:id, :name, :projectId, :integrations, :tools, :skills, :createdAt, :updatedAt)`,
|
||||
{
|
||||
id: data.id,
|
||||
name: 'Test Agent',
|
||||
projectId: data.projectId,
|
||||
integrations: '[]',
|
||||
tools: '{}',
|
||||
skills: '{}',
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
it('deletes existing knowledge files and replaces binaryDataId with storedAt', async () => {
|
||||
const seedContext = createTestMigrationContext(dataSource);
|
||||
try {
|
||||
await insertProject(seedContext, 'project-1');
|
||||
await insertAgent(seedContext, { id: 'agent-1', projectId: 'project-1' });
|
||||
|
||||
const agentFiles = seedContext.escape.tableName('agent_files');
|
||||
const now = new Date();
|
||||
await seedContext.runQuery(
|
||||
`INSERT INTO ${agentFiles} ("id", "agentId", "binaryDataId", "fileName", "mimeType", "fileSizeBytes", "createdAt", "updatedAt") VALUES (:id, :agentId, :binaryDataId, :fileName, :mimeType, :fileSizeBytes, :createdAt, :updatedAt)`,
|
||||
{
|
||||
id: 'file-1',
|
||||
agentId: 'agent-1',
|
||||
binaryDataId: 'filesystem-v2:agents/agent-1/knowledge-files/file-1/binary_data/uuid',
|
||||
fileName: 'notes.txt',
|
||||
mimeType: 'text/plain',
|
||||
fileSizeBytes: 11,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
},
|
||||
);
|
||||
|
||||
const binaryData = seedContext.escape.tableName('binary_data');
|
||||
await seedContext.runQuery(
|
||||
`INSERT INTO ${binaryData} ("fileId", "sourceType", "sourceId", "data", "mimeType", "fileName", "fileSize", "createdAt", "updatedAt") VALUES (:fileId, :sourceType, :sourceId, :data, :mimeType, :fileName, :fileSize, :createdAt, :updatedAt)`,
|
||||
{
|
||||
fileId: '11111111-1111-1111-1111-111111111111',
|
||||
sourceType: 'agent_file',
|
||||
sourceId: 'file-1',
|
||||
data: Buffer.from('notes'),
|
||||
mimeType: 'text/plain',
|
||||
fileName: 'notes.txt',
|
||||
fileSize: 5,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
},
|
||||
);
|
||||
} finally {
|
||||
await seedContext.queryRunner.release();
|
||||
}
|
||||
|
||||
await runSingleMigration(MIGRATION_NAME);
|
||||
|
||||
const assertContext = createTestMigrationContext(dataSource);
|
||||
try {
|
||||
const agentFiles = assertContext.escape.tableName('agent_files');
|
||||
const binaryData = assertContext.escape.tableName('binary_data');
|
||||
|
||||
const remainingFiles = await assertContext.runQuery<Array<{ id: string }>>(
|
||||
`SELECT "id" FROM ${agentFiles}`,
|
||||
);
|
||||
expect(remainingFiles).toEqual([]);
|
||||
|
||||
const remainingBinary = await assertContext.runQuery<Array<{ fileId: string }>>(
|
||||
`SELECT "fileId" FROM ${binaryData} WHERE "sourceType" = 'agent_file'`,
|
||||
);
|
||||
expect(remainingBinary).toEqual([]);
|
||||
|
||||
const columns = await columnNames(assertContext, 'agent_files');
|
||||
expect(columns).toContain('storedAt');
|
||||
expect(columns).not.toContain('binaryDataId');
|
||||
} finally {
|
||||
await assertContext.queryRunner.release();
|
||||
}
|
||||
});
|
||||
});
|
||||
Loading…
Reference in New Issue
Block a user