template-app/convex/documents.ts

// SPDX-License-Identifier: LicenseRef-ConusAI-Template-1.0
// Convex transactions for the documents OCR action. External OCR/model/Blob work runs in the Node
// action; these functions perform one scoped row transition each, and proposal creation itself is
// still one mutation containing the proposal plus its outbox record.

import type { TenantScope } from "@conusai-kit/core"
import { systemActor } from "@conusai-kit/core/worker"
import { createConvexDataSource } from "@conusai-kit/data-convex"
import { executorFor } from "@conusai-kit/data-convex/runtime"
import { files } from "@conusai-kit/files"
import { createProposal, DEFAULT_PROPOSALS_POLICY, type ProposalView, proposals } from "@conusai-kit/proposals"
import { createTenancyHooks } from "@conusai-kit/tenancy"
import { ConvexError, type Value, v } from "convex/values"
import { documents } from "../modules/dist/documents/contract/entities.js"
import { internalMutation, internalQuery } from "./_generated/server"
import { convexEntities, registry } from "./lib/runtime"

const scopeArgs = {
  orgId: v.string(),
  workspaceId: v.string(),
  projectId: v.string(),
}
const withFile = { ...scopeArgs, fileId: v.string() }

type ProjectScope = TenantScope & { workspaceId: string; projectId: string }

const systemScope = (ids: { orgId: string; workspaceId: string; projectId: string }): ProjectScope => ({
  level: "project",
  actor: systemActor("documents.ocr"),
  ...ids,
})

const dataFor = (ctx: { db: unknown }) =>
  createConvexDataSource({
    entities: convexEntities,
    executor: executorFor(ctx as unknown as Parameters<typeof executorFor>[0]),
  })

const problem = (message: string): never => {
  throw new ConvexError({ code: "internal", message } as unknown as Value)
}

export const fileInfo = internalQuery({
  args: withFile,
  handler: async (ctx, { fileId, ...ids }) => {
    const row = await dataFor(ctx).forScope(systemScope(ids)).of(files).get(fileId)
    if (row?.status !== "stored" || row.size === null || row.sha256 === null || row.storedAt === null) {
      return null
    }
    return {
      fileId: row.id,
      filename: row.filename,
      contentType: row.contentType,
      size: row.size,
      sha256: row.sha256,
      uploadedBy: row.uploadedBy,
      createdAt: row.createdAt,
      storedAt: row.storedAt,
    }
  },
})

export const ensureProcessing = internalMutation({
  args: { ...withFile, storageKey: v.string(), sha256: v.string() },
  handler: async (ctx, { fileId, storageKey, sha256, ...ids }) => {
    const scope = systemScope(ids)
    const data = dataFor(ctx)
    const file = await data.forScope(scope).of(files).get(fileId)
    if (
      file?.status !== "stored" ||
      file.storageKey !== storageKey ||
      file.sha256 !== sha256 ||
      file.size === null ||
      file.storedAt === null
    ) {
      return null
    }
    const repo = data.forScope(scope).of(documents)
    const existing = (await repo.list({ where: [{ field: "fileId", op: "eq", value: fileId }], limit: 1 })).items[0]
    if (existing) return existing.status === "processing" ? existing : null
    try {
      return await repo.insert({
        fileId,
        filename: file.filename,
        status: "processing",
        ocrJobId: null,
        pageCount: null,
        proposedCount: null,
        failureCode: null,
        skippedPages: null,
        finishedAt: null,
      } as never)
    } catch (error) {
      const raced = (await repo.list({ where: [{ field: "fileId", op: "eq", value: fileId }], limit: 1 })).items[0]
      if (raced) return raced.status === "processing" ? raced : null
      throw error
    }
  },
})

export const setOcrJob = internalMutation({
  args: { ...scopeArgs, documentId: v.string(), expectedVersion: v.number(), jobId: v.number() },
  handler: async (ctx, { documentId, expectedVersion, jobId, ...ids }) => {
    const repo = dataFor(ctx).forScope(systemScope(ids)).of(documents)
    const current = await repo.get(documentId)
    if (current?.status !== "processing") return problem("document processing row is no longer active")
    if (current.ocrJobId !== null) return current
    return repo.update(documentId, { ocrJobId: jobId } as never, expectedVersion)
  },
})

export const markFailed = internalMutation({
  args: {
    ...scopeArgs,
    documentId: v.string(),
    expectedVersion: v.number(),
    code: v.union(
      v.literal("file_missing"),
      v.literal("file_changed"),
      v.literal("ocr_failed"),
      v.literal("extraction_refused"),
      v.literal("extraction_truncated"),
      v.literal("extraction_invalid"),
    ),
    finishedAt: v.string(),
  },
  handler: async (ctx, { documentId, expectedVersion, code, finishedAt, ...ids }) => {
    const repo = dataFor(ctx).forScope(systemScope(ids)).of(documents)
    const current = await repo.get(documentId)
    if (current?.status !== "processing") return null
    return repo.update(documentId, { status: "failed", failureCode: code, finishedAt } as never, expectedVersion)
  },
})

export const lineProposals = internalQuery({
  args: { ...scopeArgs, documentId: v.string() },
  handler: async (ctx, { documentId, ...ids }) => {
    const repo = dataFor(ctx).forScope(systemScope(ids)).of(proposals)
    const proposalsForDocument: ProposalView[] = []
    let cursor: string | undefined
    do {
      const page = await repo.list({
        where: [
          { field: "targetKey", op: "eq", value: "documents.line" },
          { field: "targetRecordId", op: "eq", value: documentId },
        ],
        limit: 200,
        ...(cursor === undefined ? {} : { cursor }),
      })
      proposalsForDocument.push(...(page.items as unknown as ProposalView[]))
      cursor = page.hasMore ? page.cursor : undefined
    } while (cursor !== undefined)
    return proposalsForDocument.map((proposal) => ({ lineKey: (proposal.change as { lineKey?: string }).lineKey }))
  },
})

export const addLineProposal = internalMutation({
  args: {
    ...scopeArgs,
    documentId: v.string(),
    input: v.any(),
    idempotencyKey: v.string(),
  },
  handler: async (ctx, { documentId, input, idempotencyKey, ...ids }) => {
    const scope = systemScope(ids)
    const data = dataFor(ctx)
    const key = (input.change as { lineKey?: string }).lineKey
    let cursor: string | undefined
    do {
      const page = await data
        .forScope(scope)
        .of(proposals)
        .list({
          where: [
            { field: "targetKey", op: "eq", value: "documents.line" },
            { field: "targetRecordId", op: "eq", value: documentId },
          ],
          limit: 200,
          ...(cursor === undefined ? {} : { cursor }),
        })
      if (page.items.some((proposal) => (proposal.change as { lineKey?: string }).lineKey === key)) return null
      cursor = page.hasMore ? page.cursor : undefined
    } while (cursor !== undefined)
    const tenancy = createTenancyHooks({ data, registry })
    return createProposal(
      { data, registry, tenancy, policy: DEFAULT_PROPOSALS_POLICY },
      scope,
      {
        targetKey: "documents.line",
        targetRecordId: documentId,
        ...(input as Record<string, unknown>),
      } as never,
      idempotencyKey,
    )
  },
})

export const finish = internalMutation({
  args: {
    ...scopeArgs,
    documentId: v.string(),
    expectedVersion: v.number(),
    pageCount: v.number(),
    proposedCount: v.number(),
    skippedPages: v.array(v.any()),
    finishedAt: v.string(),
  },
  handler: async (ctx, { documentId, expectedVersion, ...result }) => {
    const { orgId, workspaceId, projectId, ...fields } = result
    const repo = dataFor(ctx).forScope(systemScope({ orgId, workspaceId, projectId })).of(documents)
    const current = await repo.get(documentId)
    if (current?.status !== "processing") return null
    return repo.update(documentId, { ...fields, status: "extracted", failureCode: null } as never, expectedVersion)
  },
})