diff --git a/CONTEXT.md b/CONTEXT.md index d8d43d4..80a0498 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -57,6 +57,13 @@ The Source Repository is a stable facade over smaller persistence modules. It composes Source row lifecycle, Demo Source persistence, and Source Parse Result artifact metadata without exposing those internal modules to route services. +## Source Library Localization + +Source Library Localization is the workflow that turns Knowhere-owned library +documents into Notebook Source rows for a Workspace. Listing and chat should +localize missing Knowhere documents before chunks, archive, selection, or +retrieval flows act on them. + ## Source Upload A Source Upload is the workflow that turns either a browser `File` or a diff --git a/drizzle/0009_true_lila_cheney.sql b/drizzle/0009_true_lila_cheney.sql new file mode 100644 index 0000000..20c2ccf --- /dev/null +++ b/drizzle/0009_true_lila_cheney.sql @@ -0,0 +1 @@ +CREATE UNIQUE INDEX "sources_workspace_document_idx" ON "sources" USING btree ("workspace_id","knowhere_document_id") WHERE knowhere_document_id IS NOT NULL AND deleted_at IS NULL; \ No newline at end of file diff --git a/drizzle/meta/0009_snapshot.json b/drizzle/meta/0009_snapshot.json new file mode 100644 index 0000000..65ae979 --- /dev/null +++ b/drizzle/meta/0009_snapshot.json @@ -0,0 +1,716 @@ +{ + "id": "c59ee2f8-6b59-4a2c-98a3-d71a67388d47", + "prevId": "eefd2e20-20fc-4fa0-afbd-7f6d46a37095", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.chat_messages": { + "name": "chat_messages", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "thread_id": { + "name": "thread_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "role": { + "name": "role", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "content": { + "name": "content", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "citations": { + "name": "citations", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "artifacts": { + "name": "artifacts", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "chat_messages_thread_created_idx": { + "name": "chat_messages_thread_created_idx", + "columns": [ + { + "expression": "thread_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "chat_messages_thread_id_chat_threads_id_fk": { + "name": "chat_messages_thread_id_chat_threads_id_fk", + "tableFrom": "chat_messages", + "tableTo": "chat_threads", + "columnsFrom": [ + "thread_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.chat_threads": { + "name": "chat_threads", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "demo_key": { + "name": "demo_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "chat_threads_workspace_updated_idx": { + "name": "chat_threads_workspace_updated_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "updated_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "deleted_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "chat_threads_workspace_demo_key_idx": { + "name": "chat_threads_workspace_demo_key_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "demo_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "chat_threads_workspace_id_workspaces_id_fk": { + "name": "chat_threads_workspace_id_workspaces_id_fk", + "tableFrom": "chat_threads", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.demo_source_visibilities": { + "name": "demo_source_visibilities", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "demo_source_id": { + "name": "demo_source_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "hidden_at": { + "name": "hidden_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "demo_source_visibilities_workspace_source_idx": { + "name": "demo_source_visibilities_workspace_source_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "demo_source_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "demo_source_visibilities_workspace_idx": { + "name": "demo_source_visibilities_workspace_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "demo_source_visibilities_workspace_id_workspaces_id_fk": { + "name": "demo_source_visibilities_workspace_id_workspaces_id_fk", + "tableFrom": "demo_source_visibilities", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.source_parse_results": { + "name": "source_parse_results", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "source_id": { + "name": "source_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "result_blob_url": { + "name": "result_blob_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "asset_urls": { + "name": "asset_urls", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "source_parse_results_source_id_idx": { + "name": "source_parse_results_source_id_idx", + "columns": [ + { + "expression": "source_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "source_parse_results_source_id_sources_id_fk": { + "name": "source_parse_results_source_id_sources_id_fk", + "tableFrom": "source_parse_results", + "tableTo": "sources", + "columnsFrom": [ + "source_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "source_parse_results_source_id_unique": { + "name": "source_parse_results_source_id_unique", + "nullsNotDistinct": false, + "columns": [ + "source_id" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.sources": { + "name": "sources", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "mime_type": { + "name": "mime_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "size_bytes": { + "name": "size_bytes", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "failure_reason": { + "name": "failure_reason", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "knowhere_job_id": { + "name": "knowhere_job_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "knowhere_document_id": { + "name": "knowhere_document_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "staged_blob_pathname": { + "name": "staged_blob_pathname", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "staged_blob_url": { + "name": "staged_blob_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "original_blob_pathname": { + "name": "original_blob_pathname", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "original_blob_url": { + "name": "original_blob_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "demo_key": { + "name": "demo_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "sources_workspace_created_idx": { + "name": "sources_workspace_created_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "deleted_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "sources_workspace_status_idx": { + "name": "sources_workspace_status_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "sources_workspace_demo_key_idx": { + "name": "sources_workspace_demo_key_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "demo_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "sources_workspace_document_idx": { + "name": "sources_workspace_document_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "knowhere_document_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "knowhere_document_id IS NOT NULL AND deleted_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "sources_workspace_id_workspaces_id_fk": { + "name": "sources_workspace_id_workspaces_id_fk", + "tableFrom": "sources", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.workspaces": { + "name": "workspaces", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "namespace": { + "name": "namespace", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "workspaces_user_id_idx": { + "name": "workspaces_user_id_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "workspaces_user_id_unique": { + "name": "workspaces_user_id_unique", + "nullsNotDistinct": false, + "columns": [ + "user_id" + ] + }, + "workspaces_namespace_unique": { + "name": "workspaces_namespace_unique", + "nullsNotDistinct": false, + "columns": [ + "namespace" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index 5823c91..568b77f 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -64,6 +64,13 @@ "when": 1781257344080, "tag": "0008_chat_message_artifacts", "breakpoints": true + }, + { + "idx": 9, + "version": "7", + "when": 1782472980935, + "tag": "0009_true_lila_cheney", + "breakpoints": true } ] } \ No newline at end of file diff --git a/src/app/api/sources/[sourceId]/chunks/route.test.ts b/src/app/api/sources/[sourceId]/chunks/route.test.ts index 4b7ebf7..6074abf 100644 --- a/src/app/api/sources/[sourceId]/chunks/route.test.ts +++ b/src/app/api/sources/[sourceId]/chunks/route.test.ts @@ -501,4 +501,38 @@ describe("GET /api/sources/[sourceId]/chunks", () => { includeAssetUrls: true, }) }) + + it("does not load chunks from unlocalized remote source ids", async () => { + mocks.getCurrentUser.mockResolvedValue({ + id: "user_1", + email: null, + name: null, + }) + mocks.ensureWorkspace.mockResolvedValue({ + id: "workspace_1", + userId: "user_1", + namespace: "notebook-workspace_1", + createdAt: new Date("2026-05-10T00:00:00.000Z"), + }) + mocks.fetchDemoChunkPage.mockRejectedValue(new Error("not a demo")) + + const response = await GET( + new NextRequest( + "http://localhost:3001/api/sources/knowhere-doc:default:doc_remote/chunks?page=1&pageSize=1", + ), + { + params: Promise.resolve({ + sourceId: "knowhere-doc:default:doc_remote", + }), + }, + ) + + await expect(response.json()).resolves.toEqual({ + message: "Source not found.", + }) + expect(response.status).toBe(404) + expect(mocks.findSourceInWorkspace).not.toHaveBeenCalled() + expect(mocks.ensureApiKeyForWorkspace).not.toHaveBeenCalled() + expect(mocks.makeKnowhereClient).not.toHaveBeenCalled() + }) }) diff --git a/src/app/api/sources/[sourceId]/route.test.ts b/src/app/api/sources/[sourceId]/route.test.ts index 6e49d81..b82d620 100644 --- a/src/app/api/sources/[sourceId]/route.test.ts +++ b/src/app/api/sources/[sourceId]/route.test.ts @@ -109,6 +109,40 @@ describe("PATCH /api/sources/[sourceId]", () => { ); }); + it("rejects archive requests for unlocalized remote source ids", async () => { + mocks.requireUser.mockResolvedValue({ id: "user_1" }); + mocks.ensureWorkspace.mockResolvedValue({ id: "workspace_1" }); + mocks.findSourceInWorkspace.mockResolvedValue(null); + mocks.fetchDemoCatalog.mockResolvedValue({ sources: [] }); + + const response = await PATCH( + new NextRequest( + "http://localhost:3001/api/sources/knowhere-doc:default:doc_remote", + { + method: "PATCH", + body: JSON.stringify({ archived: true }), + }, + ), + { + params: Promise.resolve({ + sourceId: "knowhere-doc:default:doc_remote", + }), + }, + ); + + await expect(response.json()).resolves.toEqual({ + message: "Source not found.", + }); + expect(response.status).toBe(404); + expect(mocks.findSourceInWorkspace).toHaveBeenCalledWith( + "workspace_1", + "knowhere-doc:default:doc_remote", + ); + expect(mocks.archive).not.toHaveBeenCalled(); + expect(mocks.softDeleteSource).not.toHaveBeenCalled(); + expect(mocks.deleteBlob).not.toHaveBeenCalled(); + }); + it("does not fail an already-soft-deleted source when original Blob cleanup fails", async () => { mocks.requireUser.mockResolvedValue({ id: "user_1" }); mocks.ensureWorkspace.mockResolvedValue({ id: "workspace_1" }); diff --git a/src/domains/chat/contracts.ts b/src/domains/chat/contracts.ts index b349b97..4c8d224 100644 --- a/src/domains/chat/contracts.ts +++ b/src/domains/chat/contracts.ts @@ -62,6 +62,7 @@ export type GenerateAnswer = (input: { export type AnswerQuestionInput = { question: string namespace: string + namespaces?: readonly string[] sources: readonly Source[] excludedSourceIds: readonly string[] retrieval: RetrievalClient diff --git a/src/domains/chat/index.test.ts b/src/domains/chat/index.test.ts index 7b11a42..0c8cfd6 100644 --- a/src/domains/chat/index.test.ts +++ b/src/domains/chat/index.test.ts @@ -92,6 +92,137 @@ describe("answerQuestionWithRetrieval", () => { }); }); + it("does not carry no-evidence metadata from default into a successful legacy namespace result", async () => { + const legacyResult = makeRetrievalResult({ + source: { + documentId: "doc_legacy", + sourceFileName: "legacy.pdf", + sectionPath: "Overview", + }, + }); + const retrieval = { + query: vi + .fn() + .mockResolvedValueOnce({ + results: [], + evidenceText: null, + referencedChunks: [], + namespace: "default", + query: "legacy document answer", + routerUsed: "workflow_single_step", + answerText: null, + stopReason: "not_found", + failureReason: "No relevant evidence found.", + }) + .mockResolvedValueOnce({ + results: [legacyResult], + evidenceText: "Legacy namespace evidence", + referencedChunks: [], + namespace: "notebook-legacy", + query: "legacy document answer", + routerUsed: "workflow_single_step", + answerText: null, + stopReason: "answer_done", + failureReason: null, + }), + }; + const generateAnswer = vi.fn(async ({ searchSources }) => { + const response = await searchSources({ query: "legacy document answer" }); + expect(response).toMatchObject({ + namespace: "default,notebook-legacy", + stopReason: "answer_done", + failureReason: null, + results: [legacyResult], + evidenceText: "Legacy namespace evidence", + }); + return makeHarnessRunResult("The legacy answer is grounded."); + }); + + const answer = await Effect.runPromise( + answerQuestionWithRetrieval({ + question: "What does the legacy document say?", + namespace: "notebook-legacy", + namespaces: ["default", "notebook-legacy"], + sources: [ + makeSource({ + title: "legacy.pdf", + knowhereDocumentId: "doc_legacy", + }), + ], + excludedSourceIds: [], + retrieval, + generateAnswer, + messages: [], + }), + ); + + expect(retrieval.query).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ namespace: "default" }), + ); + expect(retrieval.query).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ namespace: "notebook-legacy" }), + ); + expect(answer).toEqual({ + answer: "The legacy answer is grounded.", + citations: [legacyResult], + artifacts: [], + }); + }); + + it("does not hide a failed namespace query behind an empty namespace result", async () => { + const retrievalError = new Error("Legacy namespace query failed."); + const retrieval = { + query: vi + .fn() + .mockResolvedValueOnce({ + results: [], + evidenceText: null, + referencedChunks: [], + namespace: "default", + query: "legacy document answer", + routerUsed: "workflow_single_step", + answerText: null, + stopReason: "not_found", + failureReason: "No relevant evidence found.", + }) + .mockRejectedValueOnce(retrievalError), + }; + const generateAnswer = vi.fn(async ({ searchSources }) => { + await searchSources({ query: "legacy document answer" }); + return makeHarnessRunResult("This should not be used."); + }); + + await expect( + Effect.runPromise( + answerQuestionWithRetrieval({ + question: "What does the legacy document say?", + namespace: "notebook-legacy", + namespaces: ["default", "notebook-legacy"], + sources: [ + makeSource({ + title: "legacy.pdf", + knowhereDocumentId: "doc_legacy", + }), + ], + excludedSourceIds: [], + retrieval, + generateAnswer, + messages: [], + }), + ), + ).rejects.toThrow(); + expect(retrieval.query).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ namespace: "default" }), + ); + expect(retrieval.query).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ namespace: "notebook-legacy" }), + ); + }); + it("logs bounded Knowhere query response chunks", async () => { const result = makeRetrievalResult({ chunkType: "image", diff --git a/src/domains/chat/index.ts b/src/domains/chat/index.ts index 9dc0f6c..2d7245d 100644 --- a/src/domains/chat/index.ts +++ b/src/domains/chat/index.ts @@ -123,50 +123,68 @@ export const answerQuestionWithRetrieval = ( ): Promise => { const startedAt = Date.now() const retrievalPlan = toAgenticRetrievalPlan(queryInput) - const retrievalQueryParams = buildRetrievalQueryParams({ - input: queryInput, - fallbackQuestion: question, - namespace: input.namespace, - sources: input.sources, - excludedSourceIds: input.excludedSourceIds, - }) - logger.info("chat-agent: searchSources start", { - query: retrievalQueryParams.query, - topK: retrievalQueryParams.topK, - dataType: retrievalQueryParams.dataType ?? null, - signalPathCount: retrievalQueryParams.signalPaths?.length ?? 0, - filterMode: retrievalQueryParams.filterMode ?? null, - threshold: retrievalQueryParams.threshold ?? null, - targetContent: retrievalPlan.targetContent, - purpose: retrievalPlan.purpose, - }) - - try { - const response = await input.retrieval.query(retrievalQueryParams) - retrievalResponses.push(response) - logger.info("chat-agent: searchSources ok", { - query: response.query, - durationMs: Date.now() - startedAt, - resultCount: response.results.length, - referencedChunkCount: response.referencedChunks.length, - stopReason: response.stopReason ?? null, - failureReason: response.failureReason ?? null, - targetContent: retrievalPlan.targetContent, - }) - logger.info("chat-agent: knowhere query response", { - durationMs: Date.now() - startedAt, - response: formatKnowhereQueryResponseForLog(response), + const namespaces = getRetrievalNamespaces(input) + const queryResponses: RetrievalQueryResponse[] = [] + const queryFailures: unknown[] = [] + + for (const namespace of namespaces) { + const retrievalQueryParams = buildRetrievalQueryParams({ + input: queryInput, + fallbackQuestion: question, + namespace, + sources: input.sources, + excludedSourceIds: input.excludedSourceIds, }) - return { ...response, retrievalPlan } - } catch (error) { - logger.error("chat-agent: searchSources failed", { + logger.info("chat-agent: searchSources start", { + namespace, query: retrievalQueryParams.query, - durationMs: Date.now() - startedAt, - error: error instanceof Error ? error.message : String(error), + topK: retrievalQueryParams.topK, + dataType: retrievalQueryParams.dataType ?? null, + signalPathCount: retrievalQueryParams.signalPaths?.length ?? 0, + filterMode: retrievalQueryParams.filterMode ?? null, + threshold: retrievalQueryParams.threshold ?? null, targetContent: retrievalPlan.targetContent, + purpose: retrievalPlan.purpose, }) - throw error + + try { + const response = await input.retrieval.query(retrievalQueryParams) + retrievalResponses.push(response) + queryResponses.push(response) + logger.info("chat-agent: searchSources ok", { + namespace, + query: response.query, + durationMs: Date.now() - startedAt, + resultCount: response.results.length, + referencedChunkCount: response.referencedChunks.length, + stopReason: response.stopReason ?? null, + failureReason: response.failureReason ?? null, + targetContent: retrievalPlan.targetContent, + }) + logger.info("chat-agent: knowhere query response", { + durationMs: Date.now() - startedAt, + response: formatKnowhereQueryResponseForLog(response), + }) + } catch (error) { + queryFailures.push(error) + logger.error("chat-agent: searchSources failed", { + namespace, + query: retrievalQueryParams.query, + durationMs: Date.now() - startedAt, + error: formatUnknownError(error), + targetContent: retrievalPlan.targetContent, + }) + } + } + + if (queryResponses.length === 0) throw queryFailures[0] + if ( + queryFailures.length > 0 && + !queryResponses.some(hasRetrievalEvidence) + ) { + throw queryFailures[0] } + return mergeRetrievalResponses(queryResponses, retrievalPlan) } const generatedAnswer = yield* Effect.tryPromise(() => @@ -578,6 +596,92 @@ function formatUnknownError(error: unknown): string { return String(error) } +function getRetrievalNamespaces(input: AnswerQuestionInput): readonly string[] { + const candidates = + input.namespaces && input.namespaces.length > 0 + ? input.namespaces + : [input.namespace] + const namespaces: string[] = [] + + for (const namespace of candidates) { + if (namespaces.includes(namespace)) continue + namespaces.push(namespace) + } + + return namespaces +} + +function mergeRetrievalResponses( + responses: readonly RetrievalQueryResponse[], + retrievalPlan: AgenticRetrievalPlan, +): AgenticRetrievalResponse { + const [first] = responses + if (!first) { + throw new Error("No retrieval responses to merge.") + } + + const statusResponses = getRetrievalStatusResponses(responses) + const results = responses.flatMap((response) => response.results) + const referencedChunks = responses.flatMap( + (response) => response.referencedChunks, + ) + const evidenceTexts = responses + .map((response) => response.evidenceText) + .filter((value): value is string => Boolean(value)) + const answerTexts = responses + .map((response) => response.answerText) + .filter((value): value is string => Boolean(value)) + + return { + ...first, + namespace: responses.map((response) => response.namespace).join(","), + routerUsed: joinResponseText(responses.map((response) => response.routerUsed)) ?? "", + answerText: answerTexts.length > 0 ? answerTexts.join("\n\n") : null, + evidenceText: evidenceTexts.length > 0 ? evidenceTexts.join("\n\n") : null, + stopReason: joinResponseText( + statusResponses.map((response) => response.stopReason), + ), + failureReason: joinResponseText( + statusResponses.map((response) => response.failureReason), + ), + decisionTrace: statusResponses.flatMap( + (response) => response.decisionTrace ?? [], + ), + results, + referencedChunks, + retrievalPlan, + } +} + +function getRetrievalStatusResponses( + responses: readonly RetrievalQueryResponse[], +): readonly RetrievalQueryResponse[] { + const responsesWithEvidence = responses.filter(hasRetrievalEvidence) + return responsesWithEvidence.length > 0 ? responsesWithEvidence : responses +} + +function hasRetrievalEvidence(response: RetrievalQueryResponse): boolean { + return ( + response.results.length > 0 || + response.referencedChunks.length > 0 || + Boolean(response.evidenceText?.trim()) || + Boolean(response.answerText?.trim()) + ) +} + +function joinResponseText( + values: readonly (string | null | undefined)[], +): string | null { + const uniqueValues: string[] = [] + for (const value of values) { + const normalized = value?.trim() + if (!normalized || uniqueValues.includes(normalized)) continue + uniqueValues.push(normalized) + } + + return uniqueValues.length > 0 ? uniqueValues.join(",") : null +} + function buildRetrievalQueryParams(input: { readonly input: AgenticRetrievalQuery readonly fallbackQuestion: string diff --git a/src/domains/chat/route-answer.ts b/src/domains/chat/route-answer.ts index 03ddb73..2006be6 100644 --- a/src/domains/chat/route-answer.ts +++ b/src/domains/chat/route-answer.ts @@ -12,6 +12,7 @@ import { } from "@/domains/chat/service" import { chatTurnPersistence } from "@/domains/chat/chat-turn-persistence" import { reconcileSourcesForWorkspace } from "@/domains/sources/reconcile" +import { localizeRemoteLibrarySources } from "@/domains/sources/remote-library" import { sourceService } from "@/domains/sources/service" import { notebookRequestContext } from "@/domains/workspace/request-context" import { isAuthError } from "@/integrations/dashboard/api-key-service" @@ -59,6 +60,13 @@ const answerChatEffect = (input: AnswerChatInput) => const sources = yield* Effect.tryPromise(() => reconcileSourcesForWorkspace(workspace, client), ) + const compatibleSources = yield* localizeRemoteLibrarySources({ + workspace, + client, + localSources: sources, + localizeDocument: (document) => + sourceService.localizeRemoteDocument(workspace.id, document), + }) const loadSourceAssetUrls = (source: (typeof sources)[number]) => sourceService.getParseAssetUrls(workspace.id, source.id) @@ -66,7 +74,7 @@ const answerChatEffect = (input: AnswerChatInput) => yield* Effect.tryPromise(() => handleChatTurn({ workspace, - sources, + sources: compatibleSources, question: body.value.question, threadId: body.value.threadId, excludedSourceIds: body.value.excludedSourceIds, @@ -76,7 +84,7 @@ const answerChatEffect = (input: AnswerChatInput) => hardenMediaAssetUrls: ({ results, artifacts }) => hardenChatMediaAssetUrls({ workspaceId: workspace.id, - sources, + sources: compatibleSources, results, artifacts, loadSourceAssetUrls, diff --git a/src/domains/chat/service.ts b/src/domains/chat/service.ts index dea33e8..5361d5b 100644 --- a/src/domains/chat/service.ts +++ b/src/domains/chat/service.ts @@ -9,6 +9,7 @@ import { } from "." import { toChatMessageView } from "./view" import type { ChatMessage, ChatThread, Source, Workspace } from "@/infrastructure/db/schema" +import { getCompatibleNamespaces } from "@/domains/sources/namespace" import type { ChatArtifactView, ChatCitationView, @@ -120,6 +121,7 @@ export const handleChatTurnEffect = (input: ChatTurnInput) => const answer = yield* answerQuestionWithRetrieval({ question: input.question, namespace: input.workspace.namespace, + namespaces: getCompatibleNamespaces(input.workspace), sources: readySources, excludedSourceIds: input.excludedSourceIds, retrieval: input.retrieval, diff --git a/src/domains/sources/knowhere-upload.ts b/src/domains/sources/knowhere-upload.ts index 411b880..63e00ca 100644 --- a/src/domains/sources/knowhere-upload.ts +++ b/src/domains/sources/knowhere-upload.ts @@ -7,9 +7,13 @@ import { type SourceBlobUploadInput, validateSourceBlobUploadInput, } from "./blob-upload" -import type { UploadSourceDependencies } from "./source-upload-contracts" +import type { + UploadJobResult, + UploadSourceDependencies, +} from "./source-upload-contracts" import { validateUploadFile } from "./validation" import { TempFile, tempFileLayer } from "@/lib/temp-files" +import { getUploadNamespace } from "./namespace" /** * Upload a browser file to Knowhere for parsing. @@ -45,20 +49,26 @@ export const uploadSourceToKnowhereEffect = ( deps.knowhere.jobs.create({ sourceType: "file", fileName: validation.title, - namespace: workspace.namespace, + namespace: getUploadNamespace(), + documentMetadata: createNotebookDocumentMetadata({ + title: validation.title, + mimeType: validation.mimeType, + sizeBytes: file.size, + }), }), ) yield* Effect.tryPromise(() => deps.knowhere.jobs.upload(job, { file: path }), ) + const documentId = yield* tryGetPlannedDocumentIdEffect(job, deps) - return yield* Effect.promise(() => - deps.repository.markSourceParsing( - workspace.id, - source.id, - job.jobId, - ), - ) + return yield* markSourceParsingEffect({ + workspace, + source, + jobId: job.jobId, + documentId, + deps, + }) }), ).pipe( Effect.provide(tempFileLayer), @@ -105,17 +115,23 @@ export const uploadSourceBlobToKnowhereEffect = ( sourceType: "url", sourceUrl: input.url, fileName: validation.title, - namespace: workspace.namespace, + namespace: getUploadNamespace(), + documentMetadata: createNotebookDocumentMetadata({ + title: validation.title, + mimeType: validation.mimeType, + sizeBytes: input.sizeBytes, + }), }), ) + const documentId = yield* tryGetPlannedDocumentIdEffect(job, deps) - return yield* Effect.promise(() => - deps.repository.markSourceParsing( - workspace.id, - source.id, - job.jobId, - ), - ) + return yield* markSourceParsingEffect({ + workspace, + source, + jobId: job.jobId, + documentId, + deps, + }) }).pipe( Effect.catchAll(() => Effect.gen(function* () { @@ -150,3 +166,62 @@ export async function uploadSourceBlobToKnowhere( uploadSourceBlobToKnowhereEffect(workspace, input, deps), ) } + +const tryGetPlannedDocumentIdEffect = ( + job: UploadJobResult, + deps: UploadSourceDependencies, +) => { + const plannedDocumentId = getDocumentId(job) + if (plannedDocumentId !== null) { + return Effect.succeed(plannedDocumentId) + } + + return Effect.gen(function* () { + const currentJob = yield* Effect.tryPromise(() => + deps.knowhere.jobs.get(job.jobId), + ) + return getDocumentId(currentJob) + }).pipe(Effect.catchAll(() => Effect.succeed(null))) +} + +const markSourceParsingEffect = (input: { + readonly workspace: Workspace + readonly source: Source + readonly jobId: string + readonly documentId: string | null + readonly deps: UploadSourceDependencies +}) => + Effect.promise(() => + input.documentId + ? input.deps.repository.markSourceParsing( + input.workspace.id, + input.source.id, + input.jobId, + input.documentId, + ) + : input.deps.repository.markSourceParsing( + input.workspace.id, + input.source.id, + input.jobId, + ), + ) + +function createNotebookDocumentMetadata(input: { + readonly title: string + readonly mimeType: string + readonly sizeBytes: number +}): Readonly> { + return { + createdByClient: "notebook", + sourceFileName: input.title, + title: input.title, + mimeType: input.mimeType, + sizeBytes: input.sizeBytes, + } +} + +function getDocumentId(job: UploadJobResult): string | null { + return typeof job.documentId === "string" && job.documentId.length > 0 + ? job.documentId + : null +} diff --git a/src/domains/sources/namespace.ts b/src/domains/sources/namespace.ts new file mode 100644 index 0000000..30ffc82 --- /dev/null +++ b/src/domains/sources/namespace.ts @@ -0,0 +1,22 @@ +export const sharedLibraryNamespace = "default" + +type SourceNamespace = { + readonly namespace: string +} + +export function getUploadNamespace(): string { + return sharedLibraryNamespace +} + +export function getCompatibleNamespaces( + workspace: SourceNamespace, +): readonly string[] { + const namespaces = [sharedLibraryNamespace] + if ( + workspace.namespace && + workspace.namespace !== sharedLibraryNamespace + ) { + namespaces.push(workspace.namespace) + } + return namespaces +} diff --git a/src/domains/sources/remote-library.ts b/src/domains/sources/remote-library.ts new file mode 100644 index 0000000..5c867bf --- /dev/null +++ b/src/domains/sources/remote-library.ts @@ -0,0 +1,255 @@ +import { Effect } from "effect" + +import type { Source } from "@/infrastructure/db/schema" +import type { SourceStatus } from "./types" +import { getCompatibleNamespaces, sharedLibraryNamespace } from "./namespace" + +type RemoteDocument = { + readonly documentId: string + readonly namespace: string + readonly status: SourceStatus + readonly title?: string + readonly mimeType?: string + readonly sizeBytes?: number + readonly sourceFileName?: string | null + readonly documentMetadata?: Record +} + +type RemoteDocumentCandidate = RemoteDocument | { + readonly documentId?: string | null + readonly namespace?: string | null + readonly status?: string | null + readonly sourceFileName?: string | null + readonly documentMetadata?: Record +} + +type RemoteDocumentListResponse = { + readonly documents: readonly RemoteDocumentCandidate[] +} + +type RemoteDocumentClient = { + readonly documents?: { + readonly list?: (params?: { + readonly namespace?: string + }) => Promise + } +} + +type RemoteLibraryWorkspace = { + readonly namespace: string +} + +type RemoteLibraryProjectionInput = { + readonly workspace: RemoteLibraryWorkspace + readonly client: RemoteDocumentClient + readonly localSources: readonly Source[] +} + +type RemoteLibraryLocalizationInput = RemoteLibraryProjectionInput & { + readonly localizeDocument: ( + document: RemoteDocument, + ) => Promise +} + +type RemoteDocumentRaw = { + readonly document_id?: unknown + readonly documentId?: unknown + readonly namespace?: unknown + readonly status?: unknown + readonly source_file_name?: unknown + readonly sourceFileName?: unknown + readonly document_metadata?: unknown + readonly documentMetadata?: unknown +} + +export type RemoteLibraryDocument = RemoteDocument + +export function listRemoteLibraryDocuments( + input: RemoteLibraryProjectionInput, +): Effect.Effect { + return Effect.gen(function* () { + const seenDocumentIds = new Set() + const documents: RemoteLibraryDocument[] = [] + + for (const namespace of getCompatibleNamespaces(input.workspace)) { + if (!input.client.documents?.list) continue + const response = yield* Effect.tryPromise(() => + input.client.documents?.list?.({ + namespace, + }) ?? + Promise.resolve({ documents: [] }), + ).pipe( + Effect.catchAll(() => + Effect.succeed({ + documents: [], + } satisfies RemoteDocumentListResponse), + ), + ) + + for (const rawDocument of response.documents ?? []) { + const document = normalizeRemoteDocument(rawDocument) + if (!document) continue + if (seenDocumentIds.has(document.documentId)) continue + + seenDocumentIds.add(document.documentId) + documents.push(document) + } + } + + return documents + }) +} + +export function localizeRemoteLibrarySources( + input: RemoteLibraryLocalizationInput, +): Effect.Effect { + return Effect.gen(function* () { + const remoteDocuments = yield* listRemoteLibraryDocuments(input) + if (remoteDocuments.length === 0) return input.localSources + + const localizedSources = yield* Effect.all( + remoteDocuments.map((document) => + Effect.tryPromise(() => input.localizeDocument(document)).pipe( + Effect.catchAll(() => Effect.succeed(null)), + ), + ), + { concurrency: 4 }, + ) + + return mergeLocalizedSources({ + localSources: input.localSources, + localizedSources: localizedSources.filter( + (source): source is Source => source !== null, + ), + }) + }) +} + +function normalizeRemoteDocument( + value: RemoteDocumentCandidate | RemoteDocumentRaw, +): RemoteDocument | null { + const raw = value as RemoteDocumentRaw + const documentId = getString(raw.documentId ?? raw.document_id) + if (!documentId) return null + + const namespace = + getString(raw.namespace) ?? sharedLibraryNamespace + const sourceFileName = getString( + raw.sourceFileName ?? raw.source_file_name, + ) + const status = getString(raw.status) ?? "ready" + if (status === "archived") return null + + const documentMetadata = getRecord( + raw.documentMetadata ?? raw.document_metadata, + ) + + const title = getRemoteDocumentTitle({ + documentId, + sourceFileName, + documentMetadata, + }) + const mimeType = getRemoteDocumentMimeType({ documentMetadata }) + const sizeBytes = getRemoteDocumentSizeBytes({ documentMetadata }) + + return { + documentId, + namespace, + status: getRemoteSourceStatus(status), + ...(title ? { title } : {}), + ...(mimeType ? { mimeType } : {}), + ...(sizeBytes !== undefined ? { sizeBytes } : {}), + sourceFileName, + documentMetadata, + } +} + +function getRemoteSourceStatus(status: string): SourceStatus { + if (isActiveDocumentStatus(status)) return "ready" + if (status === "failed") return "failed" + return "parsing" +} + +function isActiveDocumentStatus(status: string): boolean { + return status === "active" || status === "ready" || status === "done" +} + +function getRemoteDocumentTitle(document: Pick< + RemoteDocument, + "documentId" | "documentMetadata" | "sourceFileName" +>): string | undefined { + return ( + getString(document.documentMetadata?.["source_file_name"]) ?? + getString(document.documentMetadata?.["title"]) ?? + document.sourceFileName ?? + undefined + ) +} + +function getRemoteDocumentMimeType(document: Pick< + RemoteDocument, + "documentMetadata" +>): string | undefined { + return ( + getString(document.documentMetadata?.["mime_type"]) ?? + getString(document.documentMetadata?.["mimeType"]) + ) +} + +function getRemoteDocumentSizeBytes(document: Pick< + RemoteDocument, + "documentMetadata" +>): number | undefined { + const value = + getNumber(document.documentMetadata?.["size_bytes"]) ?? + getNumber(document.documentMetadata?.["sizeBytes"]) + return value +} + +function mergeLocalizedSources(input: { + readonly localSources: readonly Source[] + readonly localizedSources: readonly Source[] +}): readonly Source[] { + const localizedSourceByDocumentId = new Map( + input.localizedSources.flatMap((source): readonly [string, Source][] => { + const documentId = source.knowhereDocumentId + return documentId ? [[documentId, source]] : [] + }), + ) + const mergedDocumentIds = new Set() + const mergedSources: Source[] = [] + + for (const source of input.localSources) { + const documentId = source.knowhereDocumentId + if (documentId) { + const localizedSource = localizedSourceByDocumentId.get(documentId) + mergedSources.push(localizedSource ?? source) + mergedDocumentIds.add(documentId) + continue + } + + mergedSources.push(source) + } + + for (const source of input.localizedSources) { + const documentId = source.knowhereDocumentId + if (documentId && mergedDocumentIds.has(documentId)) continue + if (documentId) mergedDocumentIds.add(documentId) + mergedSources.push(source) + } + + return mergedSources +} + +function getString(value: unknown): string | undefined { + return typeof value === "string" && value.length > 0 ? value : undefined +} + +function getNumber(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) ? value : undefined +} + +function getRecord(value: unknown): Record { + if (!value || typeof value !== "object" || Array.isArray(value)) return {} + return value as Record +} diff --git a/src/domains/sources/repository.ts b/src/domains/sources/repository.ts index 52eb1c9..62accd6 100644 --- a/src/domains/sources/repository.ts +++ b/src/domains/sources/repository.ts @@ -8,6 +8,7 @@ type SourceRepository = { readonly findInWorkspaceEffect: typeof sourceRowRepository.findInWorkspaceEffect readonly listForWorkspaceEffect: typeof sourceRowRepository.listForWorkspaceEffect readonly createUploadingEffect: typeof sourceRowRepository.createUploadingEffect + readonly localizeRemoteDocumentEffect: typeof sourceRowRepository.localizeRemoteDocumentEffect readonly listHiddenDemoSourceIdsEffect: typeof demoSourceRepository.listHiddenDemoSourceIdsEffect readonly hideDemoSourceEffect: typeof demoSourceRepository.hideDemoSourceEffect readonly upsertMaterializedDemoSourceEffect: typeof demoSourceRepository.upsertMaterializedDemoSourceEffect @@ -24,6 +25,8 @@ export const sourceRepository: SourceRepository = { findInWorkspaceEffect: sourceRowRepository.findInWorkspaceEffect, listForWorkspaceEffect: sourceRowRepository.listForWorkspaceEffect, createUploadingEffect: sourceRowRepository.createUploadingEffect, + localizeRemoteDocumentEffect: + sourceRowRepository.localizeRemoteDocumentEffect, listHiddenDemoSourceIdsEffect: demoSourceRepository.listHiddenDemoSourceIdsEffect, hideDemoSourceEffect: demoSourceRepository.hideDemoSourceEffect, upsertMaterializedDemoSourceEffect: diff --git a/src/domains/sources/route-archive.ts b/src/domains/sources/route-archive.ts index 2b570b6..897dee3 100644 --- a/src/domains/sources/route-archive.ts +++ b/src/domains/sources/route-archive.ts @@ -46,6 +46,7 @@ const archiveSourceEffect = ( const workspace = yield* Effect.tryPromise(() => deps.ensureWorkspace(user.id), ) + const source = yield* Effect.tryPromise(() => deps.sourceService.findInWorkspace(workspace.id, input.sourceId), ) diff --git a/src/domains/sources/route-dependencies.ts b/src/domains/sources/route-dependencies.ts index 0af5abb..724411e 100644 --- a/src/domains/sources/route-dependencies.ts +++ b/src/domains/sources/route-dependencies.ts @@ -48,6 +48,7 @@ const defaultDependencies: SourceRouteServiceDependencies = { getParseAssetUrls: defaultSourceService.getParseAssetUrls, hideDemoSource: defaultSourceService.hideDemoSource, listHiddenDemoSourceIds: defaultSourceService.listHiddenDemoSourceIds, + localizeRemoteDocument: defaultSourceService.localizeRemoteDocument, softDelete: defaultSourceService.softDelete, upsertMaterializedDemoSource: defaultSourceService.upsertMaterializedDemoSource, diff --git a/src/domains/sources/route-listing.ts b/src/domains/sources/route-listing.ts index 4350f9e..d21ac90 100644 --- a/src/domains/sources/route-listing.ts +++ b/src/domains/sources/route-listing.ts @@ -11,6 +11,7 @@ import { logger } from "@/lib/logger" import { knowhereDemoApi } from "@/integrations/knowhere-demo" import { toSourceView } from "./view" import { startBackgroundReconciliation } from "./background-reconcile" +import { localizeRemoteLibrarySources } from "./remote-library" import type { JsonRouteResult, ListSourcesBody, @@ -33,8 +34,11 @@ type RouteListingDependencies = Pick< > readonly sourceService: Pick< SourceRouteServiceDependencies["sourceService"], - "listHiddenDemoSourceIds" + "listHiddenDemoSourceIds" | "localizeRemoteDocument" > + readonly reconcileSourcesForWorkspace: SourceRouteServiceDependencies[ + "reconcileSourcesForWorkspace" + ] } type RouteListing = { @@ -73,23 +77,31 @@ const listSourcesEffect = ( const workspace = yield* Effect.tryPromise(() => deps.ensureWorkspace(user.id), ) - const sources = yield* Effect.tryPromise(() => + const listedSources = yield* Effect.tryPromise(() => deps.listSourcesForWorkspace(workspace.id), ) - const demoSourceResolution = resolveWorkspaceDemoSources(sources, catalog) - const sourcesNeedingKnowhereChunkCount = - getWorkspaceSourcesNeedingKnowhereChunkCount( - demoSourceResolution.workspaceSources, - ) - const materializedDemoSourceOptions = - getMaterializedDemoSourceViewOptionsBySourceId( - demoSourceResolution.workspaceSources, - catalog, - ) const apiKey = yield* Effect.tryPromise(() => deps.ensureApiKeyForWorkspace(workspace.id, input.cookieHeader), ) const client = deps.makeKnowhereClient(apiKey) + const sources = yield* Effect.tryPromise(() => { + if (!hasParsingSources(listedSources)) { + return Promise.resolve(listedSources) + } + return deps.reconcileSourcesForWorkspace(workspace, client) + }) + const demoSourceResolution = resolveWorkspaceDemoSources(sources, catalog) + const workspaceSources = yield* localizeRemoteLibrarySources({ + workspace, + client, + localSources: demoSourceResolution.workspaceSources, + localizeDocument: (document) => + deps.sourceService.localizeRemoteDocument(workspace.id, document), + }) + const sourcesNeedingKnowhereChunkCount = + getWorkspaceSourcesNeedingKnowhereChunkCount(workspaceSources) + const materializedDemoSourceOptions = + getMaterializedDemoSourceViewOptionsBySourceId(workspaceSources, catalog) const parsingSources = sources.filter( (source) => source.status === "parsing" && source.knowhereJobId, ) @@ -129,7 +141,7 @@ const listSourcesEffect = ( return routeResult.ok({ sources: [ ...visibleDemoSources, - ...demoSourceResolution.workspaceSources.map((source) => + ...workspaceSources.map((source) => toSourceView( source, materializedDemoSourceOptions.get(source.id) ?? @@ -141,3 +153,14 @@ const listSourcesEffect = ( }) export { createRouteListing } + +function hasParsingSources( + sources: readonly { + readonly status: string + readonly knowhereJobId: string | null + }[], +): boolean { + return sources.some( + (source) => source.status === "parsing" && source.knowhereJobId, + ) +} diff --git a/src/domains/sources/route-service.test.ts b/src/domains/sources/route-service.test.ts index b7ca91f..4ac60cc 100644 --- a/src/domains/sources/route-service.test.ts +++ b/src/domains/sources/route-service.test.ts @@ -34,6 +34,8 @@ const source: Source = { deletedAt: null, }; +const localizeNoRemoteDocuments = vi.fn(async () => source); + describe("source route service", () => { it("reconciles authenticated sources through the listing route workflow", async () => { const knowhereClient = { @@ -51,6 +53,7 @@ describe("source route service", () => { }, jobs: { create: vi.fn(), + get: vi.fn(), upload: vi.fn(), }, }; @@ -59,6 +62,7 @@ describe("source route service", () => { Effect.succeed(new Map([[source.id, { chunkCount: 8 }]])), ); const listSourcesForWorkspace = vi.fn(async () => [source]); + const reconcileSourcesForWorkspace = vi.fn(async () => [source]); const listHiddenDemoSourceIds = vi.fn(async () => []); const listing = createRouteListing({ demoApi: { @@ -74,7 +78,11 @@ describe("source route service", () => { getSourceViewOptionsBySourceId, makeKnowhereClient: vi.fn(() => knowhereClient), listSourcesForWorkspace, - sourceService: { listHiddenDemoSourceIds }, + reconcileSourcesForWorkspace, + sourceService: { + listHiddenDemoSourceIds, + localizeRemoteDocument: localizeNoRemoteDocuments, + }, }); const result = await listing.listSources({ cookieHeader: "session=abc" }); @@ -100,9 +108,284 @@ describe("source route service", () => { "session=abc", ); expect(listSourcesForWorkspace).toHaveBeenCalledWith(workspace.id); + expect(reconcileSourcesForWorkspace).toHaveBeenCalledWith( + workspace, + knowhereClient, + ); expect(listHiddenDemoSourceIds).toHaveBeenCalledWith(workspace.id); }); + it("localizes shared default and legacy namespace documents into workspace sources", async () => { + const localReadySource: Source = { + ...source, + id: "source_ready", + status: "ready", + knowhereJobId: null, + knowhereDocumentId: "doc_local", + }; + const listDocuments = vi + .fn() + .mockResolvedValueOnce({ + documents: [ + { + documentId: "doc_default", + namespace: "default", + status: "active", + sourceFileName: "cli.pdf", + documentMetadata: { + mimeType: "application/pdf", + }, + }, + { + documentId: "doc_local", + namespace: "default", + status: "active", + sourceFileName: "local-duplicate.pdf", + }, + ], + }) + .mockResolvedValueOnce({ + documents: [ + { + documentId: "doc_legacy", + namespace: workspace.namespace, + status: "active", + sourceFileName: "legacy.pdf", + }, + ], + }); + const knowhereClient = { + documents: { + archive: vi.fn(async () => undefined), + list: listDocuments, + listChunks: vi.fn(async () => ({ + chunks: [], + pagination: { + page: 1, + pageSize: 1, + total: 0, + totalPages: 0, + }, + })), + }, + jobs: { + create: vi.fn(), + get: vi.fn(), + upload: vi.fn(), + }, + }; + const localizedDefaultSource: Source = { + ...source, + id: "00000000-0000-0000-0000-000000000101", + title: "cli.pdf", + mimeType: "application/pdf", + status: "ready", + knowhereJobId: null, + knowhereDocumentId: "doc_default", + }; + const refreshedLocalSource: Source = { + ...localReadySource, + title: "local-duplicate.pdf", + mimeType: "application/octet-stream", + status: "ready", + knowhereJobId: null, + knowhereDocumentId: "doc_local", + }; + const localizedLegacySource: Source = { + ...source, + id: "00000000-0000-0000-0000-000000000102", + title: "legacy.pdf", + mimeType: "application/octet-stream", + status: "ready", + knowhereJobId: null, + knowhereDocumentId: "doc_legacy", + }; + const localizeRemoteDocument = vi + .fn() + .mockResolvedValueOnce(localizedDefaultSource) + .mockResolvedValueOnce(refreshedLocalSource) + .mockResolvedValueOnce(localizedLegacySource); + const listing = createRouteListing({ + demoApi: { + fetchCatalog: vi.fn(async () => emptyDemoCatalog), + }, + ensureApiKeyForWorkspace: vi.fn(async () => "jwt_123"), + ensureWorkspace: vi.fn(async () => workspace), + getCurrentUser: vi.fn(async () => ({ + id: "user_1", + email: null, + name: null, + })), + getSourceViewOptionsBySourceId: vi.fn(() => Effect.succeed(new Map())), + makeKnowhereClient: vi.fn(() => knowhereClient), + listSourcesForWorkspace: vi.fn(async () => [localReadySource]), + reconcileSourcesForWorkspace: vi.fn(async () => [localReadySource]), + sourceService: { + listHiddenDemoSourceIds: vi.fn(async () => []), + localizeRemoteDocument, + }, + }); + + const result = await listing.listSources({ cookieHeader: "session=abc" }); + + expect(listDocuments).toHaveBeenNthCalledWith(1, { + namespace: "default", + }); + expect(listDocuments).toHaveBeenNthCalledWith(2, { + namespace: workspace.namespace, + }); + expect(localizeRemoteDocument).toHaveBeenNthCalledWith( + 1, + workspace.id, + { + documentId: "doc_default", + namespace: "default", + status: "ready", + title: "cli.pdf", + mimeType: "application/pdf", + sourceFileName: "cli.pdf", + documentMetadata: { + mimeType: "application/pdf", + }, + }, + ); + expect(localizeRemoteDocument).toHaveBeenNthCalledWith( + 2, + workspace.id, + { + documentId: "doc_local", + namespace: "default", + status: "ready", + title: "local-duplicate.pdf", + sourceFileName: "local-duplicate.pdf", + documentMetadata: {}, + }, + ); + expect(localizeRemoteDocument).toHaveBeenNthCalledWith( + 3, + workspace.id, + { + documentId: "doc_legacy", + namespace: workspace.namespace, + status: "ready", + title: "legacy.pdf", + sourceFileName: "legacy.pdf", + documentMetadata: {}, + }, + ); + expect(result.body.sources).toEqual([ + expect.objectContaining({ + id: "source_ready", + documentId: "doc_local", + title: "local-duplicate.pdf", + status: "ready", + }), + expect.objectContaining({ + id: "00000000-0000-0000-0000-000000000101", + kind: "workspace", + title: "cli.pdf", + mimeType: "application/pdf", + status: "ready", + documentId: "doc_default", + }), + expect.objectContaining({ + id: "00000000-0000-0000-0000-000000000102", + kind: "workspace", + title: "legacy.pdf", + mimeType: "application/octet-stream", + status: "ready", + documentId: "doc_legacy", + }), + ]); + }); + + it("reconciles parsing sources before localizing matching Knowhere documents", async () => { + const reconciledSource: Source = { + ...source, + id: "source_1", + title: "uploaded.pdf", + status: "ready", + knowhereJobId: null, + knowhereDocumentId: "doc_uploaded", + }; + const listDocuments = vi + .fn() + .mockResolvedValueOnce({ + documents: [ + { + documentId: "doc_uploaded", + namespace: "default", + status: "active", + sourceFileName: "uploaded.pdf", + }, + ], + }) + .mockResolvedValueOnce({ documents: [] }); + const knowhereClient = { + documents: { + archive: vi.fn(async () => undefined), + list: listDocuments, + listChunks: vi.fn(async () => ({ + chunks: [], + pagination: { + page: 1, + pageSize: 1, + total: 0, + totalPages: 0, + }, + })), + }, + jobs: { + create: vi.fn(), + get: vi.fn(), + upload: vi.fn(), + }, + }; + const reconcileSourcesForWorkspace = vi.fn(async () => [reconciledSource]); + const localizeRemoteDocument = vi.fn(async () => reconciledSource); + const listing = createRouteListing({ + demoApi: { + fetchCatalog: vi.fn(async () => emptyDemoCatalog), + }, + ensureApiKeyForWorkspace: vi.fn(async () => "jwt_123"), + ensureWorkspace: vi.fn(async () => workspace), + getCurrentUser: vi.fn(async () => ({ + id: "user_1", + email: null, + name: null, + })), + getSourceViewOptionsBySourceId: vi.fn(() => Effect.succeed(new Map())), + makeKnowhereClient: vi.fn(() => knowhereClient), + listSourcesForWorkspace: vi.fn(async () => [source]), + reconcileSourcesForWorkspace, + sourceService: { + listHiddenDemoSourceIds: vi.fn(async () => []), + localizeRemoteDocument, + }, + }); + + const result = await listing.listSources({ cookieHeader: "session=abc" }); + + expect(reconcileSourcesForWorkspace).toHaveBeenCalledWith( + workspace, + knowhereClient, + ); + expect(localizeRemoteDocument).toHaveBeenCalledWith( + workspace.id, + expect.objectContaining({ + documentId: "doc_uploaded", + }), + ); + expect(result.body.sources).toEqual([ + expect.objectContaining({ + id: "source_1", + documentId: "doc_uploaded", + title: "uploaded.pdf", + status: "ready", + }), + ]); + }); + it("lists authenticated workspace sources when the demo catalog is unavailable", async () => { const legacyFakeSource: Source = { ...source, @@ -127,6 +410,7 @@ describe("source route service", () => { }, jobs: { create: vi.fn(), + get: vi.fn(), upload: vi.fn(), }, }; @@ -149,7 +433,14 @@ describe("source route service", () => { getSourceViewOptionsBySourceId, makeKnowhereClient: vi.fn(() => knowhereClient), listSourcesForWorkspace: vi.fn(async () => [legacyFakeSource, source]), - sourceService: { listHiddenDemoSourceIds: vi.fn(async () => []) }, + reconcileSourcesForWorkspace: vi.fn(async () => [ + legacyFakeSource, + source, + ]), + sourceService: { + listHiddenDemoSourceIds: vi.fn(async () => []), + localizeRemoteDocument: localizeNoRemoteDocuments, + }, }); const result = await listing.listSources({ cookieHeader: "session=abc" }); @@ -200,6 +491,7 @@ describe("source route service", () => { }, jobs: { create: vi.fn(), + get: vi.fn(), upload: vi.fn(), }, }; @@ -218,7 +510,11 @@ describe("source route service", () => { getSourceViewOptionsBySourceId, makeKnowhereClient: vi.fn(() => knowhereClient), listSourcesForWorkspace: vi.fn(async () => [legacyFakeSource]), - sourceService: { listHiddenDemoSourceIds: vi.fn(async () => []) }, + reconcileSourcesForWorkspace: vi.fn(async () => [legacyFakeSource]), + sourceService: { + listHiddenDemoSourceIds: vi.fn(async () => []), + localizeRemoteDocument: localizeNoRemoteDocuments, + }, }); const result = await listing.listSources({ cookieHeader: "session=abc" }); @@ -277,6 +573,7 @@ describe("source route service", () => { }, jobs: { create: vi.fn(), + get: vi.fn(), upload: vi.fn(), }, }; @@ -295,7 +592,11 @@ describe("source route service", () => { getSourceViewOptionsBySourceId, makeKnowhereClient: vi.fn(() => knowhereClient), listSourcesForWorkspace: vi.fn(async () => [nonReadyLegacySource]), - sourceService: { listHiddenDemoSourceIds: vi.fn(async () => []) }, + reconcileSourcesForWorkspace: vi.fn(async () => [nonReadyLegacySource]), + sourceService: { + listHiddenDemoSourceIds: vi.fn(async () => []), + localizeRemoteDocument: localizeNoRemoteDocuments, + }, }); const result = await listing.listSources({ cookieHeader: "session=abc" }); @@ -344,6 +645,7 @@ describe("source route service", () => { }, jobs: { create: vi.fn(), + get: vi.fn(), upload: vi.fn(), }, }; @@ -362,7 +664,11 @@ describe("source route service", () => { getSourceViewOptionsBySourceId, makeKnowhereClient: vi.fn(() => knowhereClient), listSourcesForWorkspace: vi.fn(async () => [materializedSource]), - sourceService: { listHiddenDemoSourceIds: vi.fn(async () => []) }, + reconcileSourcesForWorkspace: vi.fn(async () => [materializedSource]), + sourceService: { + listHiddenDemoSourceIds: vi.fn(async () => []), + localizeRemoteDocument: localizeNoRemoteDocuments, + }, }); const result = await listing.listSources({ cookieHeader: "session=abc" }); @@ -437,6 +743,7 @@ describe("source route service", () => { const knowhereClient = { jobs: { create: vi.fn(async () => knowhereJob), + get: vi.fn(), upload: vi.fn(async () => undefined), }, documents: { diff --git a/src/domains/sources/route-types.ts b/src/domains/sources/route-types.ts index fd6bb9f..c56335b 100644 --- a/src/domains/sources/route-types.ts +++ b/src/domains/sources/route-types.ts @@ -6,7 +6,7 @@ import type { loadChunksForSource, } from "@/domains/chunks" import type { ParsedChunkView } from "@/domains/chunks/types" -import type { SourceView } from "@/domains/sources/types" +import type { SourceStatus, SourceView } from "@/domains/sources/types" import type { AuthUser } from "@/infrastructure/auth" import type { Source, Workspace } from "@/infrastructure/db/schema" import type { @@ -21,6 +21,17 @@ import type { UploadKnowhereClient } from "./upload" type SourceRouteKnowhereClient = UploadKnowhereClient & ChunkKnowhereClient & { readonly documents: ChunkKnowhereClient["documents"] & { + list?(params?: { + readonly namespace?: string + }): Promise<{ + readonly documents: readonly { + readonly documentId: string + readonly namespace: string + readonly status: string + readonly sourceFileName?: string | null + readonly documentMetadata?: Record + }[] + }> archive(documentId: string): Promise } } @@ -136,6 +147,16 @@ type SourceWorkflowService = { demoSourceId: string, ) => Promise readonly listHiddenDemoSourceIds: (workspaceId: string) => Promise + readonly localizeRemoteDocument: ( + workspaceId: string, + input: { + readonly documentId: string + readonly title?: string + readonly mimeType?: string + readonly sizeBytes?: number + readonly status: SourceStatus + }, + ) => Promise readonly upsertMaterializedDemoSource: ( workspaceId: string, input: { diff --git a/src/domains/sources/service.ts b/src/domains/sources/service.ts index a93f103..b515300 100644 --- a/src/domains/sources/service.ts +++ b/src/domains/sources/service.ts @@ -21,6 +21,10 @@ type SourceService = { sourceId: string, ) => Promise>> readonly listForWorkspace: (workspaceId: string) => Promise + readonly localizeRemoteDocument: ( + workspaceId: string, + input: Parameters[1], + ) => Promise readonly listHiddenDemoSourceIds: (workspaceId: string) => Promise readonly hideDemoSource: ( workspaceId: string, @@ -79,6 +83,7 @@ export const sourceService: SourceService = { hideDemoSource: sourceWorkflowRuntime.hideDemoSource, listHiddenDemoSourceIds: sourceWorkflowRuntime.listHiddenDemoSourceIds, listForWorkspace: sourceWorkflowRuntime.listForWorkspace, + localizeRemoteDocument: sourceWorkflowRuntime.localizeRemoteDocument, softDelete: sourceWorkflowRuntime.softDelete, upsertMaterializedDemoSource: sourceWorkflowRuntime.upsertMaterializedDemoSource, diff --git a/src/domains/sources/source-row-repository.test.ts b/src/domains/sources/source-row-repository.test.ts index aaa2cce..4b2833f 100644 --- a/src/domains/sources/source-row-repository.test.ts +++ b/src/domains/sources/source-row-repository.test.ts @@ -38,11 +38,35 @@ describe("sourceRowRepository", () => { await expect( Effect.runPromise( sourceRowRepository - .softDeleteEffect("workspace_1", "demo-tsla-q4-2025") - .pipe(Effect.provideService(DbClient, db)), + .softDeleteEffect("workspace_1", "demo-tsla-q4-2025") + .pipe(Effect.provideService(DbClient, db)), ), ).resolves.toBe(false) }) + + it("preserves active parsing job state when localizing an existing document", async () => { + const conflictSet = await captureLocalizeConflictSet({ + title: "remote.pdf", + mimeType: "application/pdf", + sizeBytes: 12, + }) + + expect(getSqlText(conflictSet.status)).toContain("CASE WHEN") + expect(getSqlText(conflictSet.status)).toContain("knowhere_job_id") + expect(getSqlText(conflictSet.knowhereJobId)).toContain("CASE WHEN") + expect(getSqlText(conflictSet.knowhereJobId)).toContain( + "knowhere_job_id", + ) + expect(getSqlText(conflictSet.failureReason)).toContain("CASE WHEN") + }) + + it("preserves local metadata when remote document metadata is missing", async () => { + const conflictSet = await captureLocalizeConflictSet({}) + + expect(getSqlText(conflictSet.title)).toBe("title") + expect(getSqlText(conflictSet.mimeType)).toBe("mime_type") + expect(getSqlText(conflictSet.sizeBytes)).toBe("size_bytes") + }) }) function makeThrowingDb(): Db { @@ -52,3 +76,76 @@ function makeThrowingDb(): Db { }, } as unknown as Db } + +async function captureLocalizeConflictSet(input: { + readonly title?: string + readonly mimeType?: string + readonly sizeBytes?: number +}): Promise> { + let conflictSet: Record | null = null + const returningSource = { + id: "source_1", + workspaceId: "workspace_1", + title: "remote.pdf", + mimeType: "application/pdf", + sizeBytes: 12, + status: "ready", + failureReason: null, + knowhereJobId: null, + knowhereDocumentId: "doc_1", + stagedBlobPathname: null, + stagedBlobUrl: null, + originalBlobPathname: null, + originalBlobUrl: null, + demoKey: null, + createdAt: new Date("2026-06-26T00:00:00Z"), + updatedAt: new Date("2026-06-26T00:00:00Z"), + deletedAt: null, + } + const db = { + insert: () => ({ + values: () => ({ + onConflictDoUpdate: (config: { readonly set: Record }) => { + conflictSet = config.set + return { + returning: async () => [returningSource], + } + }, + }), + }), + } as unknown as Db + + await sourceRowRepository.localizeRemoteDocumentWithDb(db, "workspace_1", { + documentId: "doc_1", + status: "ready", + ...input, + }) + + if (!conflictSet) { + throw new Error("localizeRemoteDocumentWithDb did not configure conflict set") + } + + return conflictSet +} + +function getSqlText(value: unknown): string { + const chunks = (value as { readonly queryChunks?: readonly unknown[] }) + .queryChunks + return (chunks ?? []).map(getSqlChunkText).join("") +} + +function getSqlChunkText(value: unknown): string { + if (typeof value === "string") return value + if (!value || typeof value !== "object") return "" + + const sqlChunk = value as { readonly queryChunks?: readonly unknown[] } + if (Array.isArray(sqlChunk.queryChunks)) { + return sqlChunk.queryChunks.map(getSqlChunkText).join("") + } + + const chunk = value as { readonly value?: readonly string[] } + if (Array.isArray(chunk.value)) return chunk.value.join("") + + const column = value as { readonly name?: unknown } + return typeof column.name === "string" ? column.name : "" +} diff --git a/src/domains/sources/source-row-repository.ts b/src/domains/sources/source-row-repository.ts index f6621fc..bfb613f 100644 --- a/src/domains/sources/source-row-repository.ts +++ b/src/domains/sources/source-row-repository.ts @@ -6,6 +6,7 @@ import { Effect } from "effect" import { DbClient, type Db } from "@/infrastructure/db" import { sources, type Source } from "@/infrastructure/db/schema" import { logger } from "@/lib/logger" +import type { SourceStatus } from "./types" type CreateUploadingSourceInput = { readonly title: string @@ -34,6 +35,14 @@ type SourceUpdate = Partial< > > +type LocalizeRemoteDocumentInput = { + readonly documentId: string + readonly title?: string + readonly mimeType?: string + readonly sizeBytes?: number + readonly status: SourceStatus +} + type SourceRowRepository = { readonly findInWorkspaceEffect: ( workspaceId: string, @@ -46,10 +55,15 @@ type SourceRowRepository = { workspaceId: string, input: CreateUploadingSourceInput, ) => Effect.Effect + readonly localizeRemoteDocumentEffect: ( + workspaceId: string, + input: LocalizeRemoteDocumentInput, + ) => Effect.Effect readonly markParsingEffect: ( workspaceId: string, sourceId: string, jobId: string, + documentId?: string, ) => Effect.Effect readonly markReadyEffect: ( workspaceId: string, @@ -83,6 +97,11 @@ type SourceRowRepository = { values: SourceUpdate, requiredStatus?: string, ) => Promise + readonly localizeRemoteDocumentWithDb: ( + db: Db, + workspaceId: string, + input: LocalizeRemoteDocumentInput, + ) => Promise readonly requireSource: (source: Source | null, message: string) => Source } @@ -145,14 +164,25 @@ const createUploadingEffect: SourceRowRepository["createUploadingEffect"] = ( return source }) +const localizeRemoteDocumentEffect: SourceRowRepository["localizeRemoteDocumentEffect"] = + (workspaceId: string, input: LocalizeRemoteDocumentInput) => + Effect.gen(function* () { + const db = yield* DbClient + return yield* Effect.promise(() => + localizeRemoteDocumentWithDb(db, workspaceId, input), + ) + }) + const markParsingEffect: SourceRowRepository["markParsingEffect"] = ( workspaceId: string, sourceId: string, jobId: string, + documentId?: string, ) => updateInWorkspaceEffect(workspaceId, sourceId, { status: "parsing", knowhereJobId: jobId, + knowhereDocumentId: documentId, failureReason: null, }) @@ -289,6 +319,63 @@ async function updateInWorkspaceWithDb( return source ?? null } +async function localizeRemoteDocumentWithDb( + db: Db, + workspaceId: string, + input: LocalizeRemoteDocumentInput, +): Promise { + const hasActiveLocalParsingJob = sql`${sources.status} = 'parsing' AND ${sources.knowhereJobId} IS NOT NULL` + const values = { + workspaceId, + title: input.title ?? input.documentId, + mimeType: input.mimeType ?? "application/octet-stream", + sizeBytes: input.sizeBytes ?? 0, + status: input.status, + failureReason: + input.status === "failed" ? "Knowhere document failed." : null, + knowhereJobId: null, + knowhereDocumentId: input.documentId, + stagedBlobPathname: null, + stagedBlobUrl: null, + originalBlobPathname: null, + originalBlobUrl: null, + demoKey: null, + } + + const [source] = await db + .insert(sources) + .values(values) + .onConflictDoUpdate({ + target: [sources.workspaceId, sources.knowhereDocumentId], + targetWhere: sql`knowhere_document_id IS NOT NULL AND deleted_at IS NULL`, + set: { + title: + input.title === undefined + ? sql`${sources.title}` + : values.title, + mimeType: + input.mimeType === undefined + ? sql`${sources.mimeType}` + : values.mimeType, + sizeBytes: + input.sizeBytes === undefined + ? sql`${sources.sizeBytes}` + : values.sizeBytes, + status: sql`CASE WHEN ${hasActiveLocalParsingJob} THEN ${sources.status} ELSE ${values.status} END`, + failureReason: sql`CASE WHEN ${hasActiveLocalParsingJob} THEN ${sources.failureReason} ELSE ${values.failureReason} END`, + knowhereJobId: sql`CASE WHEN ${hasActiveLocalParsingJob} THEN ${sources.knowhereJobId} ELSE ${values.knowhereJobId} END`, + updatedAt: sql`now()`, + }, + }) + .returning() + + if (!source) { + throw new Error("localizeRemoteDocument: upsert did not return a row.") + } + + return source +} + const WORKSPACE_SOURCE_ID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/iu @@ -305,6 +392,7 @@ export const sourceRowRepository: SourceRowRepository = { findInWorkspaceEffect, listForWorkspaceEffect, createUploadingEffect, + localizeRemoteDocumentEffect, markParsingEffect, markReadyEffect, markFailedEffect, @@ -313,5 +401,6 @@ export const sourceRowRepository: SourceRowRepository = { isWorkspaceSourceId, findInWorkspaceWithDb, updateInWorkspaceWithDb, + localizeRemoteDocumentWithDb, requireSource, } diff --git a/src/domains/sources/source-upload-contracts.ts b/src/domains/sources/source-upload-contracts.ts index 1684927..fd0fa13 100644 --- a/src/domains/sources/source-upload-contracts.ts +++ b/src/domains/sources/source-upload-contracts.ts @@ -1,7 +1,17 @@ -import type { Job } from "@ontos-ai/knowhere-sdk" +import type Knowhere from "@ontos-ai/knowhere-sdk" import type { Source } from "@/infrastructure/db/schema" +type KnowhereJobCreateInput = Parameters[0] & { + readonly documentMetadata?: Readonly> +} + +export type UploadJobResult = Awaited< + ReturnType +> & { + readonly documentId?: string | null +} + export type UploadSourceRepository = { createUploadingSource( workspaceId: string, @@ -19,6 +29,7 @@ export type UploadSourceRepository = { workspaceId: string, sourceId: string, jobId: string, + documentId?: string, ): Promise markSourceFailed( workspaceId: string, @@ -29,21 +40,12 @@ export type UploadSourceRepository = { export type UploadKnowhereClient = { jobs: { - create( - input: - | { - sourceType: "file" - fileName: string - namespace: string - } - | { - sourceType: "url" - sourceUrl: string - fileName: string - namespace: string - }, - ): Promise - upload(job: string | Job, input: { file: string }): Promise + create(input: KnowhereJobCreateInput): Promise + get(jobId: string): Promise + upload( + job: Parameters[0], + input: { file: string }, + ): Promise } } diff --git a/src/domains/sources/types.ts b/src/domains/sources/types.ts index 16a5b90..94f97b5 100644 --- a/src/domains/sources/types.ts +++ b/src/domains/sources/types.ts @@ -35,6 +35,7 @@ export type SourceView = { readonly id: string readonly kind?: SourceKind readonly demoSourceId?: string + readonly namespace?: string readonly title: string /** Browser-provided content type for preview routing. */ readonly mimeType: string diff --git a/src/domains/sources/upload.test.ts b/src/domains/sources/upload.test.ts index 443dc42..9632731 100644 --- a/src/domains/sources/upload.test.ts +++ b/src/domains/sources/upload.test.ts @@ -48,6 +48,7 @@ describe("uploadSourceToKnowhere", () => { knowhere: { jobs: { create: vi.fn(), + get: vi.fn(), upload: vi.fn(), }, }, @@ -70,6 +71,7 @@ describe("uploadSourceToKnowhere", () => { const parsingSource = makeSource({ status: "parsing", knowhereJobId: "job_123", + knowhereDocumentId: "doc_123", }); const deps = { repository: { @@ -85,6 +87,9 @@ describe("uploadSourceToKnowhere", () => { sourceType: "file", createdAt: new Date("2026-05-06T00:00:00Z"), }), + get: vi.fn().mockResolvedValue({ + documentId: "doc_123", + }), upload: vi.fn().mockResolvedValue(undefined), }, }, @@ -104,13 +109,22 @@ describe("uploadSourceToKnowhere", () => { expect(deps.knowhere.jobs.create).toHaveBeenCalledWith({ sourceType: "file", fileName: "notes.pdf", - namespace: workspace.namespace, + namespace: "default", + documentMetadata: { + createdByClient: "notebook", + sourceFileName: "notes.pdf", + title: "notes.pdf", + mimeType: "application/pdf", + sizeBytes: file.size, + }, }); expect(deps.knowhere.jobs.upload).toHaveBeenCalled(); + expect(deps.knowhere.jobs.get).toHaveBeenCalledWith("job_123"); expect(deps.repository.markSourceParsing).toHaveBeenCalledWith( workspace.id, uploadingSource.id, "job_123", + "doc_123", ); expect(result).toMatchObject({ id: "source_1", @@ -139,6 +153,7 @@ describe("uploadSourceToKnowhere", () => { sourceType: "file", createdAt: new Date("2026-05-06T00:00:00Z"), }), + get: vi.fn(), upload: vi.fn().mockRejectedValue(new Error("network")), }, }, @@ -157,6 +172,95 @@ describe("uploadSourceToKnowhere", () => { uploadingSource.id, "Knowhere upload failed.", ); + expect(deps.knowhere.jobs.get).not.toHaveBeenCalled(); + }); + + it("keeps a queued file upload parsing when Knowhere has not published a document id yet", async () => { + const uploadingSource = makeSource(); + const parsingSource = makeSource({ + status: "parsing", + knowhereJobId: "job_123", + knowhereDocumentId: null, + }); + const deps = { + repository: { + createUploadingSource: vi.fn().mockResolvedValue(uploadingSource), + markSourceParsing: vi.fn().mockResolvedValue(parsingSource), + markSourceFailed: vi.fn(), + }, + knowhere: { + jobs: { + create: vi.fn().mockResolvedValue({ + jobId: "job_123", + status: "waiting-file", + sourceType: "file", + createdAt: new Date("2026-05-06T00:00:00Z"), + }), + get: vi.fn().mockResolvedValue({ + status: "running", + }), + upload: vi.fn().mockResolvedValue(undefined), + }, + }, + }; + const file = new File(["hello"], "notes.pdf", { type: "application/pdf" }); + + const result = await uploadSourceToKnowhere(workspace, file, deps); + + expect(deps.repository.markSourceParsing).toHaveBeenCalledWith( + workspace.id, + uploadingSource.id, + "job_123", + ); + expect(deps.repository.markSourceFailed).not.toHaveBeenCalled(); + expect(result).toMatchObject({ + id: "source_1", + status: "parsing", + knowhereDocumentId: null, + }); + }); + + it("uses the planned document id from the SDK job creation response", async () => { + const uploadingSource = makeSource(); + const parsingSource = makeSource({ + status: "parsing", + knowhereJobId: "job_123", + knowhereDocumentId: "doc_planned", + }); + const deps = { + repository: { + createUploadingSource: vi.fn().mockResolvedValue(uploadingSource), + markSourceParsing: vi.fn().mockResolvedValue(parsingSource), + markSourceFailed: vi.fn(), + }, + knowhere: { + jobs: { + create: vi.fn().mockResolvedValue({ + jobId: "job_123", + status: "waiting-file", + sourceType: "file", + documentId: "doc_planned", + createdAt: new Date("2026-05-06T00:00:00Z"), + }), + get: vi.fn(), + upload: vi.fn().mockResolvedValue(undefined), + }, + }, + }; + + await uploadSourceToKnowhere( + workspace, + new File(["hello"], "notes.pdf", { type: "application/pdf" }), + deps, + ); + + expect(deps.knowhere.jobs.get).not.toHaveBeenCalled(); + expect(deps.repository.markSourceParsing).toHaveBeenCalledWith( + workspace.id, + uploadingSource.id, + "job_123", + "doc_planned", + ); }); it("creates a URL parse job from a client-uploaded public Blob", async () => { @@ -165,6 +269,7 @@ describe("uploadSourceToKnowhere", () => { title: "large.pdf", status: "parsing", knowhereJobId: "job_123", + knowhereDocumentId: "doc_123", sizeBytes: 5, }); const deps = { @@ -181,6 +286,9 @@ describe("uploadSourceToKnowhere", () => { sourceType: "url", createdAt: new Date("2026-05-06T00:00:00Z"), }), + get: vi.fn().mockResolvedValue({ + documentId: "doc_123", + }), upload: vi.fn().mockResolvedValue(undefined), }, }, @@ -212,9 +320,23 @@ describe("uploadSourceToKnowhere", () => { sourceType: "url", sourceUrl: "https://store.public.blob.vercel-storage.com/source-uploads/upload_1/document.pdf", fileName: "large.pdf", - namespace: workspace.namespace, + namespace: "default", + documentMetadata: { + createdByClient: "notebook", + sourceFileName: "large.pdf", + title: "large.pdf", + mimeType: "application/pdf", + sizeBytes: 5, + }, }); expect(deps.knowhere.jobs.upload).not.toHaveBeenCalled(); + expect(deps.knowhere.jobs.get).toHaveBeenCalledWith("job_123"); + expect(deps.repository.markSourceParsing).toHaveBeenCalledWith( + workspace.id, + uploadingSource.id, + "job_123", + "doc_123", + ); expect(result).toMatchObject({ id: "source_1", title: "large.pdf", @@ -222,6 +344,62 @@ describe("uploadSourceToKnowhere", () => { }); }); + it("keeps a URL parse job parsing when Knowhere has not published a document id yet", async () => { + const uploadingSource = makeSource({ title: "large.pdf", sizeBytes: 5 }); + const parsingSource = makeSource({ + title: "large.pdf", + status: "parsing", + knowhereJobId: "job_123", + knowhereDocumentId: null, + sizeBytes: 5, + }); + const deps = { + repository: { + createUploadingSource: vi.fn().mockResolvedValue(uploadingSource), + markSourceParsing: vi.fn().mockResolvedValue(parsingSource), + markSourceFailed: vi.fn(), + }, + knowhere: { + jobs: { + create: vi.fn().mockResolvedValue({ + jobId: "job_123", + status: "pending", + sourceType: "url", + createdAt: new Date("2026-05-06T00:00:00Z"), + }), + get: vi.fn().mockResolvedValue({ + status: "running", + }), + upload: vi.fn(), + }, + }, + }; + + const result = await uploadSourceBlobToKnowhere( + workspace, + { + pathname: "source-uploads/upload_1/document.pdf", + url: "https://store.public.blob.vercel-storage.com/source-uploads/upload_1/document.pdf", + fileName: "large.pdf", + mimeType: "application/pdf", + sizeBytes: 5, + }, + deps, + ); + + expect(deps.repository.markSourceParsing).toHaveBeenCalledWith( + workspace.id, + uploadingSource.id, + "job_123", + ); + expect(deps.repository.markSourceFailed).not.toHaveBeenCalled(); + expect(result).toMatchObject({ + id: "source_1", + status: "parsing", + knowhereDocumentId: null, + }); + }); + it("keeps the original public Blob and returns a failed source when URL job creation fails", async () => { const uploadingSource = makeSource({ title: "large.pdf", sizeBytes: 5 }); const failedSource = makeSource({ @@ -241,6 +419,7 @@ describe("uploadSourceToKnowhere", () => { knowhere: { jobs: { create: vi.fn().mockRejectedValue(new Error("network")), + get: vi.fn(), upload: vi.fn(), }, }, diff --git a/src/domains/sources/workflow-runtime.ts b/src/domains/sources/workflow-runtime.ts index fff2202..e2c6b5f 100644 --- a/src/domains/sources/workflow-runtime.ts +++ b/src/domains/sources/workflow-runtime.ts @@ -9,6 +9,10 @@ type CreateUploadingSourceInput = Parameters< typeof sourceRepository.createUploadingEffect >[1] +type LocalizeRemoteDocumentInput = Parameters< + typeof sourceRepository.localizeRemoteDocumentEffect +>[1] + type SaveSourceParseResultInput = Parameters< typeof sourceRepository.saveParseResultEffect >[2] @@ -26,6 +30,7 @@ type UploadRepositoryRuntime = { workspaceId: string, sourceId: string, jobId: string, + documentId?: string, ) => Promise readonly markFailed: ( workspaceId: string, @@ -52,6 +57,10 @@ type SourceWorkflowRuntime = UploadRepositoryRuntime & { sourceId: string, ) => Promise>> readonly listForWorkspace: (workspaceId: string) => Promise + readonly localizeRemoteDocument: ( + workspaceId: string, + input: LocalizeRemoteDocumentInput, + ) => Promise readonly listHiddenDemoSourceIds: (workspaceId: string) => Promise readonly hideDemoSource: ( workspaceId: string, @@ -90,6 +99,12 @@ const listForWorkspace: SourceWorkflowRuntime["listForWorkspace"] = ( ) => databaseRuntime.runPromise(sourceRepository.listForWorkspaceEffect(workspaceId)) +const localizeRemoteDocument: SourceWorkflowRuntime["localizeRemoteDocument"] = + (workspaceId: string, input: LocalizeRemoteDocumentInput) => + databaseRuntime.runPromise( + sourceRepository.localizeRemoteDocumentEffect(workspaceId, input), + ) + const listHiddenDemoSourceIds: SourceWorkflowRuntime["listHiddenDemoSourceIds"] = (workspaceId: string) => databaseRuntime.runPromise( @@ -116,9 +131,15 @@ const markParsing: SourceWorkflowRuntime["markParsing"] = ( workspaceId: string, sourceId: string, jobId: string, + documentId?: string, ) => databaseRuntime.runPromise( - sourceRepository.markParsingEffect(workspaceId, sourceId, jobId), + sourceRepository.markParsingEffect( + workspaceId, + sourceId, + jobId, + documentId, + ), ) const markReady: SourceWorkflowRuntime["markReady"] = ( @@ -188,9 +209,10 @@ function createUploadRepository( workspaceId: string, sourceId: string, jobId: string, + documentId?: string, ) => requireSource( - await runtime.markParsing(workspaceId, sourceId, jobId), + await runtime.markParsing(workspaceId, sourceId, jobId, documentId), "Source disappeared before parsing.", ), markSourceFailed: async ( @@ -219,6 +241,7 @@ export const sourceWorkflowRuntime: SourceWorkflowRuntime = { hideDemoSource, listForWorkspace, listHiddenDemoSourceIds, + localizeRemoteDocument, markFailed, markParsing, markReady, diff --git a/src/domains/workspace/initial-state.test.ts b/src/domains/workspace/initial-state.test.ts index 6be3c8e..6361411 100644 --- a/src/domains/workspace/initial-state.test.ts +++ b/src/domains/workspace/initial-state.test.ts @@ -366,6 +366,7 @@ describe("loadWorkspaceShellInitialState", () => { const state = await loadWorkspaceShellInitialState(deps) expect(listSourcesForWorkspace).toHaveBeenCalledWith(workspace.id) + expect(deps.reconcileSourcesForWorkspace).not.toHaveBeenCalled() expect(state.sources).toEqual([ expect.objectContaining({ id: "demo-tsla-q4-2025", @@ -382,6 +383,95 @@ describe("loadWorkspaceShellInitialState", () => { ]) }) + it("reconciles parsing sources before localizing matching Knowhere documents", async () => { + const workspace = makeWorkspace() + const parsingSource = makeSource(workspace.id, { + status: "parsing", + knowhereJobId: "job_123", + knowhereDocumentId: null, + }) + const reconciledSource = makeSource(workspace.id, { + id: parsingSource.id, + title: "uploaded.pdf", + status: "ready", + knowhereJobId: null, + knowhereDocumentId: "doc_uploaded", + }) + const client = { + documents: { + list: vi + .fn() + .mockResolvedValueOnce({ + documents: [ + { + documentId: "doc_uploaded", + namespace: "default", + status: "active", + sourceFileName: "uploaded.pdf", + }, + ], + }) + .mockResolvedValueOnce({ documents: [] }), + listChunks: vi.fn(async () => ({ + chunks: [], + pagination: { + page: 1, + pageSize: 1, + total: 0, + totalPages: 0, + }, + })), + }, + jobs: { + get: vi.fn(), + load: vi.fn(), + }, + } as unknown as InitialStateClient + const reconcileSourcesForWorkspace = vi.fn(async () => [reconciledSource]) + const localizeRemoteDocument = vi.fn(async () => reconciledSource) + const deps = createDependencies({ + getClientForWorkspace: vi.fn(async () => ({ client, apiKey: "sk_test" })), + getOptionalAuthenticated: vi.fn(async () => ({ + user: { + id: "user_1", + email: "ada@example.com", + name: "Ada", + }, + workspace, + })), + listSourcesForWorkspace: vi.fn(async () => [parsingSource]), + reconcileSourcesForWorkspace, + localizeRemoteDocument, + }) + + const state = await loadWorkspaceShellInitialState(deps) + + expect(reconcileSourcesForWorkspace).toHaveBeenCalledWith( + workspace, + client, + ) + expect(localizeRemoteDocument).toHaveBeenCalledWith( + workspace.id, + expect.objectContaining({ + documentId: "doc_uploaded", + }), + ) + expect(state.sources).toEqual([ + expect.objectContaining({ + id: "demo-tsla-q4-2025", + kind: "demo", + }), + { + id: parsingSource.id, + kind: "workspace", + title: "uploaded.pdf", + mimeType: "application/pdf", + status: "ready", + documentId: "doc_uploaded", + }, + ]) + }) + it("adds operation context when initial state loading fails", async () => { const deps = createDependencies({ listSourcesForWorkspace: vi.fn(async () => { @@ -443,6 +533,18 @@ function createDependencies( listHiddenDemoSourceIds: vi.fn(async () => []), listMessages: vi.fn(async () => []), listSourcesForWorkspace: vi.fn(async () => []), + reconcileSourcesForWorkspace: vi.fn(async () => []), + localizeRemoteDocument: vi.fn(async (workspaceId, input) => + makeSource(workspaceId, { + id: `source_${input.documentId}`, + title: input.title, + mimeType: input.mimeType, + sizeBytes: input.sizeBytes, + status: input.status, + knowhereJobId: null, + knowhereDocumentId: input.documentId, + }), + ), sourceViewOptionsBySourceId: vi.fn(() => Effect.succeed(new Map())), ...overrides, } diff --git a/src/domains/workspace/initial-state.ts b/src/domains/workspace/initial-state.ts index a407b71..1a73f83 100644 --- a/src/domains/workspace/initial-state.ts +++ b/src/domains/workspace/initial-state.ts @@ -13,6 +13,8 @@ import { import { chatThreadService } from "@/domains/chat/thread-service" import { toChatMessageView, toChatThreadView } from "@/domains/chat/view" import { sourceViewOptionsBySourceId as getSourceViewOptionsBySourceId } from "@/domains/sources/counts" +import { localizeRemoteLibrarySources } from "@/domains/sources/remote-library" +import { reconcileSourcesForWorkspace as reconcileDefaultSourcesForWorkspace } from "@/domains/sources/reconcile" import { sourceService } from "@/domains/sources/service" import { startBackgroundReconciliation } from "@/domains/sources/background-reconcile" import { sourceWorkflowRuntime } from "@/domains/sources/workflow-runtime" @@ -88,7 +90,22 @@ async function getDemoChunksForSource( } type WorkspaceShellInitialStateClient = - Parameters[1] + Parameters[1] & + Parameters[1] & { + readonly documents: { + readonly list: (params?: { + readonly namespace?: string + }) => Promise<{ + readonly documents: readonly { + readonly documentId: string + readonly namespace: string + readonly status: string + readonly sourceFileName?: string | null + readonly documentMetadata?: Record + }[] + }> + } + } type WorkspaceShellInitialStateDependencies = { readonly fetchDemoCatalog: () => Promise @@ -121,6 +138,14 @@ type WorkspaceShellInitialStateDependencies = { readonly listSourcesForWorkspace: ( workspaceId: string, ) => Promise + readonly reconcileSourcesForWorkspace: ( + workspace: Workspace, + client: WorkspaceShellInitialStateClient, + ) => Promise + readonly localizeRemoteDocument: ( + workspaceId: string, + input: Parameters[1], + ) => Promise readonly sourceViewOptionsBySourceId: ( sources: readonly Source[], client: WorkspaceShellInitialStateClient, @@ -137,6 +162,8 @@ const defaultDependencies: WorkspaceShellInitialStateDependencies = { listHiddenDemoSourceIds: sourceService.listHiddenDemoSourceIds, listMessages: chatThreadService.listMessages, listSourcesForWorkspace: sourceWorkflowRuntime.listForWorkspace, + reconcileSourcesForWorkspace: reconcileDefaultSourcesForWorkspace, + localizeRemoteDocument: sourceService.localizeRemoteDocument, sourceViewOptionsBySourceId: getSourceViewOptionsBySourceId, } @@ -214,17 +241,13 @@ export const loadWorkspaceShellInitialStateEffect = ( }, () => knowhereDemoApi.fetchOptionalCatalog(deps.fetchDemoCatalog), ) - const sources = yield* effectOperation.tryPromise( + const listedSources = yield* effectOperation.tryPromise( { context: workspaceInitialStateContext, operation: "listSourcesForWorkspace", }, () => deps.listSourcesForWorkspace(workspace.id), ) - const demoSourceResolution = resolveWorkspaceDemoSources( - sources, - demoCatalog, - ) const hiddenDemoSourceIds = new Set( yield* effectOperation.tryPromise( { @@ -234,15 +257,6 @@ export const loadWorkspaceShellInitialStateEffect = ( () => deps.listHiddenDemoSourceIds(workspace.id), ), ) - const visibleDemoCatalogSources = demoCatalog.sources - .filter( - (source) => - !demoSourceResolution.materializedDemoSourceIds.has( - source.demoSourceId, - ), - ) - .filter((source) => !hiddenDemoSourceIds.has(source.demoSourceId)) - const demoSources = visibleDemoCatalogSources.map(demoView.toSourceView) const listedChatThreads = yield* effectOperation.tryPromise( { context: workspaceInitialStateContext, @@ -278,15 +292,6 @@ export const loadWorkspaceShellInitialStateEffect = ( const chatMessages = activeChatMessages ? activeChatMessages.map((message) => toChatMessageView(message)) : [] - const sourcesNeedingKnowhereChunkCount = - getWorkspaceSourcesNeedingKnowhereChunkCount( - demoSourceResolution.workspaceSources, - ) - const materializedDemoSourceOptions = - getMaterializedDemoSourceViewOptionsBySourceId( - demoSourceResolution.workspaceSources, - demoCatalog, - ) const { client, apiKey } = yield* effectOperation.tryPromise( { context: workspaceInitialStateContext, @@ -294,6 +299,49 @@ export const loadWorkspaceShellInitialStateEffect = ( }, () => deps.getClientForWorkspace(workspace), ) + const sources = yield* effectOperation.tryPromise( + { + context: workspaceInitialStateContext, + operation: "reconcileSourcesForWorkspace", + }, + () => + hasParsingSources(listedSources) + ? deps.reconcileSourcesForWorkspace(workspace, client) + : Promise.resolve(listedSources), + ) + const demoSourceResolution = resolveWorkspaceDemoSources( + sources, + demoCatalog, + ) + const visibleDemoCatalogSources = demoCatalog.sources + .filter( + (source) => + !demoSourceResolution.materializedDemoSourceIds.has( + source.demoSourceId, + ), + ) + .filter((source) => !hiddenDemoSourceIds.has(source.demoSourceId)) + const demoSources = visibleDemoCatalogSources.map(demoView.toSourceView) + const workspaceSources = yield* effectOperation.addContext( + { + context: workspaceInitialStateContext, + operation: "localizeRemoteLibrarySources", + }, + localizeRemoteLibrarySources({ + workspace, + client, + localSources: demoSourceResolution.workspaceSources, + localizeDocument: (document) => + deps.localizeRemoteDocument(workspace.id, document), + }), + ) + const sourcesNeedingKnowhereChunkCount = + getWorkspaceSourcesNeedingKnowhereChunkCount(workspaceSources) + const materializedDemoSourceOptions = + getMaterializedDemoSourceViewOptionsBySourceId( + workspaceSources, + demoCatalog, + ) for (const source of sources) { if (source.status === "parsing" && source.knowhereJobId) { yield* Effect.fork( @@ -331,7 +379,7 @@ export const loadWorkspaceShellInitialStateEffect = ( dashboardUrl: resolveDashboardUrl(), sources: [ ...demoSources, - ...demoSourceResolution.workspaceSources.map((source) => + ...workspaceSources.map((source) => toSourceView( source, materializedDemoSourceOptions.get(source.id) ?? @@ -364,6 +412,17 @@ function resolveDashboardUrl(): string | undefined { return process.env.DASHBOARD_ORIGIN } +function hasParsingSources( + sources: readonly { + readonly status: string + readonly knowhereJobId: string | null + }[], +): boolean { + return sources.some( + (source) => source.status === "parsing" && source.knowhereJobId, + ) +} + function toOfficialLibrarySourceViews( catalog: DemoCatalog, ): OfficialLibrarySourceView[] { diff --git a/src/infrastructure/db/schema.ts b/src/infrastructure/db/schema.ts index 585bfbb..720a303 100644 --- a/src/infrastructure/db/schema.ts +++ b/src/infrastructure/db/schema.ts @@ -116,6 +116,9 @@ export const sources = pgTable( .where(sql`deleted_at IS NULL`), index("sources_workspace_status_idx").on(t.workspaceId, t.status), uniqueIndex("sources_workspace_demo_key_idx").on(t.workspaceId, t.demoKey), + uniqueIndex("sources_workspace_document_idx") + .on(t.workspaceId, t.knowhereDocumentId) + .where(sql`knowhere_document_id IS NOT NULL AND deleted_at IS NULL`), ], ); diff --git a/src/integrations/knowhere.test.ts b/src/integrations/knowhere.test.ts index bbc268c..74b8c25 100644 --- a/src/integrations/knowhere.test.ts +++ b/src/integrations/knowhere.test.ts @@ -2,16 +2,19 @@ import { afterEach, describe, expect, it, vi } from "vitest"; const constructorSpy = vi.fn(); const postSpy = vi.fn(); +const listSpy = vi.fn(); vi.mock("@ontos-ai/knowhere-sdk", () => ({ default: class FakeKnowhere { readonly jobs: FakeJobs; + readonly documents: FakeDocuments; constructor(options: unknown) { constructorSpy(options); this.jobs = new FakeJobs({ post: postSpy, }); + this.documents = new FakeDocuments(); } }, })); @@ -28,6 +31,12 @@ class FakeJobs { } } +class FakeDocuments { + async list(input: unknown): Promise { + return listSpy(input); + } +} + describe("makeKnowhereClient", () => { const originalBaseURL = process.env.KNOWHERE_BASE_URL; @@ -35,6 +44,7 @@ describe("makeKnowhereClient", () => { vi.resetModules(); constructorSpy.mockReset(); postSpy.mockReset(); + listSpy.mockReset(); restoreEnv("KNOWHERE_BASE_URL", originalBaseURL); });