209 lines
8.6 KiB
TypeScript
209 lines
8.6 KiB
TypeScript
// The document requests and the index job's orchestration: idempotent (collection, file) pairs,
|
|
// the flag gate, the content-type and size admission, the row's progress states, and which
|
|
// failures are permanent.
|
|
|
|
import type { TenantScope } from "@conusai-kit/core"
|
|
import { createMemoryDataSource } from "@conusai-kit/data"
|
|
import { DEFAULT_FILES_POLICY, files } from "@conusai-kit/files"
|
|
import { fakeObjectStore } from "@conusai-kit/testing"
|
|
import { describe, expect, it } from "vitest"
|
|
import { collections, documents, KNOWLEDGE_ENTITIES } from "../contract/index.js"
|
|
import { listDocuments, requestIndex, requestRemove } from "../core/documents.js"
|
|
import { indexDocument, removeDocument } from "../core/index-job.js"
|
|
import type { KnowledgeDeps } from "../core/ports.js"
|
|
|
|
const USER = "01900000-0000-7000-0000-000000000009"
|
|
const scope: TenantScope = {
|
|
level: "project",
|
|
actor: { userId: USER },
|
|
orgId: "01900000-0000-7000-0000-000000000001",
|
|
workspaceId: "01900000-0000-7000-0000-000000000002",
|
|
projectId: "01900000-0000-7000-0000-000000000003",
|
|
}
|
|
|
|
const fileRow = {
|
|
status: "stored" as const,
|
|
storageKey: "k/seed",
|
|
filename: "a.txt",
|
|
contentType: "text/plain",
|
|
size: 12,
|
|
sha256: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
|
|
uploadedBy: USER,
|
|
storedAt: "2026-10-23T00:00:00.000Z",
|
|
uploadId: null,
|
|
declaredSize: null,
|
|
partSize: null,
|
|
deletingAt: null,
|
|
}
|
|
|
|
const makeDeps = (options: { flag?: boolean; search?: KnowledgeDeps["search"] } = {}) => {
|
|
const data = createMemoryDataSource({ entities: [...KNOWLEDGE_ENTITIES, files] })
|
|
const store = fakeObjectStore()
|
|
const search = options.search ?? {
|
|
readiness: async () => ({ state: "ready" as const, reasons: [] }),
|
|
index: async () => undefined,
|
|
search: async () => [],
|
|
remove: async () => undefined,
|
|
}
|
|
const deps: KnowledgeDeps = {
|
|
data,
|
|
files: { data, store, policy: DEFAULT_FILES_POLICY },
|
|
search,
|
|
flags: { enabled: async () => options.flag ?? true },
|
|
}
|
|
void deps
|
|
return { data, deps, store }
|
|
}
|
|
|
|
const seed = async (data: ReturnType<typeof createMemoryDataSource>, contentType = "text/plain") => {
|
|
await data.forScope(scope).of(collections).insert({ name: "contracts", createdBy: USER })
|
|
const row = await data
|
|
.forScope(scope)
|
|
.of(files)
|
|
.insert({ ...fileRow, ...(contentType === "text/plain" ? {} : { contentType }) })
|
|
return row.id
|
|
}
|
|
|
|
/** Seeds the file row AND the object the row names, the way a completed upload leaves both. */
|
|
const seedWithBytes = async (
|
|
data: ReturnType<typeof createMemoryDataSource>,
|
|
store: ReturnType<typeof fakeObjectStore>,
|
|
bytes: Uint8Array,
|
|
) => {
|
|
const fileId = await seed(data)
|
|
await store.put(scope, { key: fileRow.storageKey, body: bytes, contentType: "text/plain" })
|
|
return fileId
|
|
}
|
|
|
|
describe("requesting a document's indexing", () => {
|
|
it("creates the row, publishes the event, and a repeat request changes nothing", async () => {
|
|
const { data, deps } = makeDeps()
|
|
const fileId = await seed(data)
|
|
const first = await requestIndex(deps, scope, { collectionName: "contracts", fileId })
|
|
expect(first.status).toBe("queued")
|
|
const second = await requestIndex(deps, scope, { collectionName: "contracts", fileId: first.fileId })
|
|
expect(second.documentId).toBe(first.documentId)
|
|
expect(await listDocuments(deps, scope, "contracts")).toEqual({
|
|
documents: [expect.objectContaining({ documentId: first.documentId, status: "queued" })],
|
|
})
|
|
// One outbox event: the subscriber's enqueue is the exactly-once path.
|
|
const outbox = data.outbox.pending(10)
|
|
expect((await outbox).filter((record) => record.name === "knowledge.documentIndexRequested")).toHaveLength(1)
|
|
})
|
|
|
|
it("re-queues an indexed pair instead of duplicating it", async () => {
|
|
const { data, deps } = makeDeps()
|
|
const fileId = await seed(data)
|
|
const first = await requestIndex(deps, scope, { collectionName: "contracts", fileId })
|
|
await data.forScope(scope).of(documents).update(first.documentId, { status: "indexed" }, 1)
|
|
const again = await requestIndex(deps, scope, { collectionName: "contracts", fileId })
|
|
expect(again.documentId).toBe(first.documentId)
|
|
expect(again.status).toBe("queued")
|
|
})
|
|
|
|
it("refuses when the flag is off, and admits text files only, within the size cap", async () => {
|
|
const { data, deps } = makeDeps({ flag: false })
|
|
const fileId = await seed(data)
|
|
const off = await requestIndex(deps, scope, { collectionName: "contracts", fileId }).catch((e: unknown) => e)
|
|
expect((off as { status?: number }).status).toBe(403)
|
|
|
|
const refusing = makeDeps()
|
|
const pdfFile = await seed(refusing.data, "application/pdf")
|
|
const pdf = await requestIndex(refusing.deps, scope, { collectionName: "contracts", fileId: pdfFile }).catch(
|
|
(e: unknown) => e,
|
|
)
|
|
expect((pdf as { status?: number }).status).toBe(422)
|
|
|
|
const big = makeDeps()
|
|
const bigFile = await seed(big.data)
|
|
await big.data.forScope(scope).of(files).update(bigFile, { size: 2_000_000 }, 1)
|
|
const tooBig = await requestIndex(big.deps, scope, { collectionName: "contracts", fileId: bigFile }).catch(
|
|
(e: unknown) => e,
|
|
)
|
|
expect((tooBig as { status?: number }).status).toBe(422)
|
|
})
|
|
})
|
|
|
|
describe("the index job's orchestration", () => {
|
|
it("moves the row queued -> indexing -> indexed and calls the search service once", async () => {
|
|
const indexed: string[] = []
|
|
const { data, deps, store } = makeDeps({
|
|
search: {
|
|
readiness: async () => ({ state: "ready", reasons: [] }),
|
|
index: async (_scope, input) => {
|
|
indexed.push(input.fileId)
|
|
},
|
|
search: async () => [],
|
|
remove: async () => undefined,
|
|
},
|
|
})
|
|
const fileId = await seedWithBytes(data, store, new TextEncoder().encode("hello knowledge"))
|
|
const requested = await requestIndex(deps, scope, { collectionName: "contracts", fileId })
|
|
const done = await indexDocument(deps, scope, requested.documentId)
|
|
expect(done.status).toBe("indexed")
|
|
expect(indexed).toEqual([fileId])
|
|
// A redelivered job finds the row indexed and does the external work again never.
|
|
const again = await indexDocument(deps, scope, requested.documentId)
|
|
expect(again.status).toBe("indexed")
|
|
expect(indexed).toEqual([fileId])
|
|
})
|
|
|
|
it("records a permanent failure on the row; a transient one throws for the queue to retry", async () => {
|
|
const permanent = makeDeps({
|
|
search: {
|
|
readiness: async () => ({ state: "ready", reasons: [] }),
|
|
index: async () => {
|
|
throw Object.assign(new Error("not text"), { permanent: "not_text" })
|
|
},
|
|
search: async () => [],
|
|
remove: async () => undefined,
|
|
},
|
|
})
|
|
const permanentFile = await seedWithBytes(
|
|
permanent.data,
|
|
permanent.store,
|
|
new TextEncoder().encode("not text at all"),
|
|
)
|
|
const requested = await requestIndex(permanent.deps, scope, { collectionName: "contracts", fileId: permanentFile })
|
|
const failed = await indexDocument(permanent.deps, scope, requested.documentId)
|
|
expect(failed.status).toBe("failed")
|
|
expect(failed.error).toBe("not_text")
|
|
|
|
const transient = makeDeps({
|
|
search: {
|
|
readiness: async () => ({ state: "ready", reasons: [] }),
|
|
index: async () => {
|
|
throw new Error("the store hiccuped")
|
|
},
|
|
search: async () => [],
|
|
remove: async () => undefined,
|
|
},
|
|
})
|
|
const transientFile = await seedWithBytes(transient.data, transient.store, new TextEncoder().encode("hello"))
|
|
const requested2 = await requestIndex(transient.deps, scope, { collectionName: "contracts", fileId: transientFile })
|
|
await expect(indexDocument(transient.deps, scope, requested2.documentId)).rejects.toThrow("hiccuped")
|
|
})
|
|
|
|
it("removal deletes the row and asks the service to drop the vectors", async () => {
|
|
const removed: string[] = []
|
|
const { data, deps } = makeDeps({
|
|
search: {
|
|
readiness: async () => ({ state: "ready", reasons: [] }),
|
|
index: async () => undefined,
|
|
search: async () => [],
|
|
remove: async (_scope, input) => {
|
|
removed.push(input.fileId)
|
|
},
|
|
},
|
|
})
|
|
const fileId = await seed(data)
|
|
const requested = await requestIndex(deps, scope, { collectionName: "contracts", fileId })
|
|
const cancelled = await requestRemove(deps, scope, { collectionName: "contracts", fileId })
|
|
expect(cancelled.status).toBe("cancelled")
|
|
// The remove job's body: delete the vectors, then the row.
|
|
await removeDocument(deps, scope, requested.documentId)
|
|
expect(removed).toEqual([fileId])
|
|
expect((await listDocuments(deps, scope, "contracts")).documents).toEqual([])
|
|
})
|
|
})
|