template-app/modules/knowledge/core/documents.ts

// The documents: request an index (a web write that records the row and publishes the outbox
// event the worker's subscriber turns into the index job), request a removal the same way, and
// list a collection's documents with their progress states.

import { forbidden, notFound, validationFailed } from "@conusai-kit/core"
import { getFile } from "@conusai-kit/files"
import {
  type DocumentView,
  documentIndexRequested,
  documentRemoveRequested,
  documents,
  INDEXABLE_CONTENT_TYPES,
  MAX_INDEX_BYTES,
} from "../contract/index.js"
import { requireCollection } from "./collections.js"
import { type KnowledgeDeps, type KnowledgeScope, knowledgeEnabled } from "./ports.js"

const PAGE = 200

export const documentView = (row: {
  id: string
  collectionName: string
  fileId: string
  filename: string
  status: DocumentView["status"]
  error: string | null
  indexedAt: string | null
  requestedBy: string
}): DocumentView => ({
  documentId: row.id,
  collectionName: row.collectionName,
  fileId: row.fileId,
  filename: row.filename,
  status: row.status,
  error: row.error,
  indexedAt: row.indexedAt,
  requestedBy: row.requestedBy,
})

export async function listDocuments(
  deps: KnowledgeDeps,
  scope: KnowledgeScope,
  collectionName: string,
): Promise<{ documents: DocumentView[] }> {
  const found = []
  let cursor: string | undefined
  for (;;) {
    const page = await deps.data
      .forScope(scope)
      .of(documents)
      .list({
        where: [{ field: "collectionName", op: "eq", value: collectionName }],
        order: [{ field: "createdAt", direction: "desc" }],
        limit: PAGE,
        ...(cursor === undefined ? {} : { cursor }),
      })
    found.push(...page.items)
    if (!page.hasMore) break
    cursor = page.cursor
  }
  return { documents: found.map((row) => documentView(row)) }
}

/**
 * Asks for a document's indexing. The write is three things in one transaction: the durable row
 * (the UI's progress states and the job's idempotency key), the outbox event the worker's
 * subscriber enqueues the index job from, and nothing else — chunking, embeddings and the vector
 * store all happen in the worker, never in a request (AGENTS.md: long work is a job).
 *
 * Idempotent by (collection, file): asking again for a pair that is queued or indexing answers
 * the same row and publishes nothing; asking for an indexed or failed pair re-queues the one row,
 * so a retry can never duplicate a document.
 */
export async function requestIndex(
  deps: KnowledgeDeps,
  scope: KnowledgeScope,
  input: { collectionName: string; fileId: string },
): Promise<DocumentView> {
  // The tenant-wall reads come before the flag check: another tenant's ids answer NotFound, never
  // a policy refusal, so ids do not leak existence behind a disabled flag.
  await requireCollection(deps, scope, input.collectionName)
  const file = await getFile(deps.files, scope, input.fileId)
  if (!(INDEXABLE_CONTENT_TYPES as readonly string[]).includes(file.contentType)) {
    throw validationFailed(
      [{ path: "fileId", message: `${file.contentType} cannot be indexed; index a text file` }],
      "this file type is not indexable",
    )
  }
  if (file.size > MAX_INDEX_BYTES) {
    throw validationFailed(
      [{ path: "fileId", message: `the file is larger than ${MAX_INDEX_BYTES} bytes` }],
      "the file is too large to index",
    )
  }
  if (!(await knowledgeEnabled(deps, scope))) throw forbidden("knowledge is switched off for this project")

  return deps.data.unitOfWork.run(scope, async (tx) => {
    const existing = (
      await tx.of(documents).list({
        where: [
          { field: "collectionName", op: "eq", value: input.collectionName },
          { field: "fileId", op: "eq", value: input.fileId },
        ],
        limit: 1,
      })
    ).items[0]
    if (existing && (existing.status === "queued" || existing.status === "indexing")) {
      return documentView(existing)
    }
    const row =
      existing !== undefined
        ? await tx
            .of(documents)
            .update(
              existing.id,
              { status: "queued", error: null, indexedAt: null, requestedBy: scope.actor.userId },
              existing.version,
            )
        : await tx.of(documents).insert({
            collectionName: input.collectionName,
            fileId: input.fileId,
            filename: file.filename,
            status: "queued",
            error: null,
            indexedAt: null,
            requestedBy: scope.actor.userId,
          })
    await tx.publish(documentIndexRequested, {
      orgId: scope.orgId,
      workspaceId: scope.workspaceId,
      projectId: scope.projectId,
      documentId: row.id,
      collectionName: input.collectionName,
      fileId: input.fileId,
    })
    return documentView(row)
  })
}

/** Asks for a document's removal: the row and the event in one transaction, the job in the worker. */
export async function requestRemove(
  deps: KnowledgeDeps,
  scope: KnowledgeScope,
  input: { collectionName: string; fileId: string },
): Promise<DocumentView> {
  const current = (
    await deps.data
      .forScope(scope)
      .of(documents)
      .list({
        where: [
          { field: "collectionName", op: "eq", value: input.collectionName },
          { field: "fileId", op: "eq", value: input.fileId },
        ],
        limit: 1,
      })
  ).items[0]
  // The tenant-wall read comes first, as in requestIndex: another tenant's ids answer NotFound.
  if (!current) throw notFound("no such document")
  if (current.status === "cancelled") return documentView(current)
  if (!(await knowledgeEnabled(deps, scope))) throw forbidden("knowledge is switched off for this project")
  return deps.data.unitOfWork.run(scope, async (tx) => {
    const row = (
      await tx.of(documents).list({
        where: [
          { field: "collectionName", op: "eq", value: input.collectionName },
          { field: "fileId", op: "eq", value: input.fileId },
        ],
        limit: 1,
      })
    ).items[0]
    if (!row) throw notFound("no such document")
    if (row.status === "cancelled") return documentView(row)
    const next = await tx.of(documents).update(row.id, { status: "cancelled", error: null }, row.version)
    await tx.publish(documentRemoveRequested, {
      orgId: scope.orgId,
      workspaceId: scope.workspaceId,
      projectId: scope.projectId,
      documentId: row.id,
      collectionName: input.collectionName,
      fileId: input.fileId,
    })
    return documentView(next)
  })
}

/** The document row a job resumes by; NotFound when another tenant's id was named. */
export async function documentById(deps: KnowledgeDeps, scope: KnowledgeScope, documentId: string) {
  const row = await deps.data.forScope(scope).of(documents).get(documentId)
  if (!row) throw notFound("no such document")
  return row
}