113 lines
4.3 KiB
TypeScript
113 lines
4.3 KiB
TypeScript
// The index job's body: what the worker runs for one document. Read the file's bytes, chunk the
|
|
// text, embed the chunks through the search service, and record the outcome on the durable row.
|
|
// Transient failures (the vector store hiccups) throw, and the queue retries the job; the row the
|
|
// retry resumes from makes the retry safe. Permanent failures are recorded on the row with a
|
|
// stable code, so a person sees why a document failed and a retried request does not pay twice.
|
|
|
|
import { internal } from "@conusai-kit/core"
|
|
import { downloadFile } from "@conusai-kit/files"
|
|
import { type DocumentView, documents } from "../contract/index.js"
|
|
import { documentById } from "./documents.js"
|
|
import type { KnowledgeDeps, KnowledgeScope } from "./ports.js"
|
|
|
|
export async function indexDocument(
|
|
deps: KnowledgeDeps,
|
|
scope: KnowledgeScope,
|
|
documentId: string,
|
|
options: { signal?: AbortSignal } = {},
|
|
): Promise<DocumentView> {
|
|
const row = await documentById(deps, scope, documentId)
|
|
if (row.status === "indexing" || row.status === "indexed") {
|
|
// A redelivered job or a crashed retry: the row says where the work stands.
|
|
return {
|
|
documentId: row.id,
|
|
collectionName: row.collectionName,
|
|
fileId: row.fileId,
|
|
filename: row.filename,
|
|
status: row.status,
|
|
error: row.error,
|
|
indexedAt: row.indexedAt,
|
|
requestedBy: row.requestedBy,
|
|
}
|
|
}
|
|
const service = deps.search
|
|
if (!service) throw internal("the knowledge search service is not bound in this process")
|
|
const indexing = await deps.data
|
|
.forScope(scope)
|
|
.of(documents)
|
|
.update(row.id, { status: "indexing", error: null }, row.version)
|
|
|
|
try {
|
|
const bytes = await readFileBytes(deps, scope, row.fileId)
|
|
const text = decodeText(bytes)
|
|
await service.index(scope, {
|
|
collectionName: row.collectionName,
|
|
fileId: row.fileId,
|
|
filename: row.filename,
|
|
bytes: new TextEncoder().encode(text),
|
|
})
|
|
const done = await deps.data
|
|
.forScope(scope)
|
|
.of(documents)
|
|
.update(indexing.id, { status: "indexed", indexedAt: new Date().toISOString(), error: null }, indexing.version)
|
|
return {
|
|
documentId: done.id,
|
|
collectionName: done.collectionName,
|
|
fileId: done.fileId,
|
|
filename: done.filename,
|
|
status: done.status as DocumentView["status"],
|
|
error: done.error,
|
|
indexedAt: done.indexedAt,
|
|
requestedBy: done.requestedBy,
|
|
}
|
|
} catch (error) {
|
|
if (options.signal?.aborted) throw error
|
|
const permanent = permanentFailure(error)
|
|
if (permanent === null) throw error
|
|
const failed = await deps.data
|
|
.forScope(scope)
|
|
.of(documents)
|
|
.update(indexing.id, { status: "failed", error: permanent }, indexing.version)
|
|
return {
|
|
documentId: failed.id,
|
|
collectionName: failed.collectionName,
|
|
fileId: failed.fileId,
|
|
filename: failed.filename,
|
|
status: "failed",
|
|
error: failed.error,
|
|
indexedAt: failed.indexedAt,
|
|
requestedBy: failed.requestedBy,
|
|
}
|
|
}
|
|
}
|
|
|
|
export async function removeDocument(deps: KnowledgeDeps, scope: KnowledgeScope, documentId: string): Promise<void> {
|
|
const row = await documentById(deps, scope, documentId)
|
|
const service = deps.search
|
|
if (!service) throw internal("the knowledge search service is not bound in this process")
|
|
await service.remove(scope, { collectionName: row.collectionName, fileId: row.fileId })
|
|
await deps.data.forScope(scope).of(documents).delete(row.id)
|
|
}
|
|
|
|
async function readFileBytes(deps: KnowledgeDeps, scope: KnowledgeScope, fileId: string): Promise<Uint8Array> {
|
|
const response = await downloadFile(deps.files, scope, fileId)
|
|
return new Uint8Array(await response.arrayBuffer())
|
|
}
|
|
|
|
function decodeText(bytes: Uint8Array): string {
|
|
try {
|
|
return new TextDecoder("utf-8", { fatal: true }).decode(bytes)
|
|
} catch {
|
|
throw Object.assign(new Error("the file is not valid UTF-8 text"), { permanent: "not_text" })
|
|
}
|
|
}
|
|
|
|
/** Maps an indexing failure to the stable code recorded on the row; null when it is transient. */
|
|
function permanentFailure(error: unknown): string | null {
|
|
const code = (error as { permanent?: unknown }).permanent
|
|
if (code === "not_text" || code === "not_indexable" || code === "too_large" || code === "vector_store_unavailable") {
|
|
return String(code)
|
|
}
|
|
return null
|
|
}
|