template-app/test/support/documents-app.ts

// SPDX-License-Identifier: LicenseRef-ConusAI-Template-1.0
// The whole app of this template, wired the way apps/web and apps/worker wire it, on the in-memory
// data source and the fakes of fakes.ts: the registry of every module, one caller per actor, the
// outbox relay's behaviour and the worker's job runner.
//
// `deliver()` stands in for the outbox relay (kit-jobs-pgboss): it drains the outbox once, runs each
// record's subscribers in a child container with the record's organization and a system actor, and
// marks it published -- exactly once per subscriber, never more forgiving than the real relay.
// `runJobs()` stands in for the worker: it runs every queued job in a child container carrying the
// scope the job was enqueued in, with a system actor. What neither does is retry, so a test of a
// retry runs the handler again itself.

import {
  DOCUMENT_EXCLUDED_PAGE_CLASSES,
  DOCUMENT_MODEL,
  DOCUMENT_OCR,
  DOCUMENTS_SCOPE,
  MODEL_GATEWAY,
  type PageClass,
} from "@app/modules/documents"
import { type Actor, buildRegistry, Container, createCaller, type TransportCaller } from "@conusai-kit/core"
import { systemActor } from "@conusai-kit/core/worker"
import {
  createMemoryDataSource,
  DATA_SOURCE,
  entitiesOf,
  JOB_QUEUE,
  OBJECT_STORE,
  SYSTEM_DATA,
} from "@conusai-kit/data"
import { FILES_AUDIT, FILES_REGISTRY, FILES_SCOPE, FILES_USAGE } from "@conusai-kit/files"
import { createModelGateway } from "@conusai-kit/platform"
import { PROPOSALS_AUDIT, PROPOSALS_REGISTRY, PROPOSALS_SCOPE } from "@conusai-kit/proposals"
import { createTenancyHooks, TENANCY, users } from "@conusai-kit/tenancy"
import {
  fakeJobQueue,
  fakeObjectStore,
  recordingAudit,
  recordingUsage,
  uploadViaProcedures,
} from "@conusai-kit/testing"
import type { FakeChatResponder } from "@conusai-kit/testing/fake-platform"
import { appModules } from "../../modules/index.js"
import { deterministicModel } from "./fake-model.js"
import { type FakePlatform, startOcr } from "./ocr.js"

const started: FakePlatform[] = []

/** Stops every fake platform the test file started: call it from `afterEach`. */
export async function closeApps(): Promise<void> {
  await Promise.all(started.splice(0).map((platform) => platform.close()))
}

export interface SeededProject {
  readonly owner: Actor
  readonly call: TransportCaller
  readonly orgSlug: string
  readonly workspaceSlug: string
  readonly projectSlug: string
  readonly orgId: string
  readonly workspaceId: string
  readonly projectId: string
  /** The three slugs a project-scoped procedure's input carries. */
  readonly slugs: { org: string; workspace: string; project: string }
}

/** `document` is what the fake platform's OCR reads every file as: a `conusai.ocr.document/v1`. */
export async function makeDocumentsApp(options: {
  document: Record<string, unknown>
  /** What the model answers; by default the deterministic reader of fake-model.ts. */
  responder?: FakeChatResponder
  /** The page classes the deployment leaves out; by default the module's own (drawings and empty pages). */
  excludedPageClasses?: readonly PageClass[]
}) {
  const registry = buildRegistry([...appModules])
  const entities = entitiesOf(registry, "postgres")
  const data = createMemoryDataSource({ entities })
  const store = fakeObjectStore()
  const queue = fakeJobQueue()
  const audit = recordingAudit()
  const usage = recordingUsage()
  const { platform, ocr } = await startOcr(options.document)
  started.push(platform)

  const root = new Container()
  root.value(DATA_SOURCE, data)
  root.value(SYSTEM_DATA, data)
  root.value(OBJECT_STORE, store)
  root.value(JOB_QUEUE, queue)
  root.value(FILES_REGISTRY, registry)
  root.value(FILES_AUDIT, audit)
  root.value(FILES_USAGE, usage)
  root.value(PROPOSALS_REGISTRY, registry)
  root.value(PROPOSALS_AUDIT, audit)
  root.value(DOCUMENT_OCR, ocr)
  // The model is the real gateway client over HTTP to the fake platform, whose reply is scripted.
  const modelRequests: Array<Record<string, unknown>> = []
  const script = (responder: FakeChatResponder | undefined): void =>
    platform.ai.respond(
      responder &&
        ((request) => {
          modelRequests.push(request)
          return responder(request)
        }),
    )
  script(options.responder ?? deterministicModel)
  root.value(
    MODEL_GATEWAY,
    createModelGateway({ CONUSAI_API_URL: `${platform.url}/api`, CONUSAI_API_TOKEN: "deployment-token" }),
  )
  root.value(DOCUMENT_MODEL, "fake-model")
  if (options.excludedPageClasses) root.value(DOCUMENT_EXCLUDED_PAGE_CLASSES, options.excludedPageClasses)

  const caller = (actor: Actor): TransportCaller => {
    const tenancy = createTenancyHooks({ data, registry })
    const container = root.child()
    container.value(TENANCY, tenancy)
    return createCaller({ registry, container, tenancy }, { actor })
  }

  const signUp = async (email: string): Promise<Actor> => {
    const user = await data.global(users).insert({
      issuer: "https://id.example.test",
      subject: email,
      email,
      displayName: email.split("@")[0] ?? email,
      locale: "en",
    })
    return { userId: user.id }
  }

  const delivered: Array<{ name: string }> = []
  const deliver = async (): Promise<number> => {
    const pending = await data.outbox.pending(500)
    let ran = 0
    for (const record of pending) {
      const entry = registry.events[record.name]
      if (entry) {
        for (const subscriber of entry.subscribers) {
          const container = root.child()
          const scope = { level: "org" as const, actor: systemActor("app.relay"), orgId: record.orgId }
          container.value(FILES_SCOPE, scope)
          container.value(PROPOSALS_SCOPE, scope)
          await subscriber.handler(
            { name: record.name, payload: record.payload },
            { container, correlationId: record.id },
          )
          delivered.push({ name: record.name })
          ran += 1
        }
      }
      await data.outbox.markPublished([record.id])
    }
    return ran
  }

  /** The jobs not yet run, in the order they were enqueued. */
  let consumed = 0
  const runJobs = async (): Promise<number> => {
    let ran = 0
    while (consumed < queue.jobs.length) {
      const job = queue.jobs[consumed] as (typeof queue.jobs)[number]
      consumed += 1
      const registered = registry.jobs[job.job]
      if (!registered) throw new Error(`no module declares the job ${job.job}`)
      const container = root.child()
      const tenancy = createTenancyHooks({ data, registry })
      container.value(TENANCY, tenancy)
      container.value(DOCUMENTS_SCOPE, { ...job.scope, actor: systemActor(`${job.job}`) })
      await registered.def.handler(registered.def.payload.parse(job.payload), {
        container,
        correlationId: job.id,
        attempt: 1,
      })
      ran += 1
    }
    return ran
  }

  async function seedProject(slug: string, email: string): Promise<SeededProject> {
    const owner = await signUp(email)
    const call = caller(owner)
    const org = (await call.callByName("tenancy.org.create", { slug, name: slug })) as { orgId: string; slug: string }
    const workspace = (await call.callByName("tenancy.workspace.create", {
      org: slug,
      slug: `ws-${slug}`,
      name: `Workspace ${slug}`,
      visibility: "org",
    })) as { workspaceId: string; slug: string }
    const project = (await call.callByName("tenancy.project.create", {
      org: slug,
      workspace: workspace.slug,
      slug: `p-${slug}`,
      name: `Project ${slug}`,
      moduleKey: "documents",
    })) as { projectId: string; slug: string }
    return {
      owner,
      call,
      orgSlug: org.slug,
      workspaceSlug: workspace.slug,
      projectSlug: project.slug,
      orgId: org.orgId,
      workspaceId: workspace.workspaceId,
      projectId: project.projectId,
      slugs: { org: org.slug, workspace: workspace.slug, project: project.slug },
    }
  }

  /** A signed-up user who joined the project's organization with a role. */
  async function addMember(project: SeededProject, email: string, role: "viewer" | "member" | "admin") {
    const actor = await signUp(email)
    const invite = (await project.call.callByName("tenancy.member.invite", {
      org: project.orgSlug,
      email,
      role,
    })) as { token: string }
    const call = caller(actor)
    await call.callByName("tenancy.invitation.accept", { org: project.orgSlug, token: invite.token })
    return { actor, call }
  }

  return {
    registry,
    data,
    entities,
    store,
    queue,
    audit,
    usage,
    /** The OCR port the worker uses, observed, and the fake platform behind it. */
    ocr,
    platform,
    /** Every request the model was sent, as the gateway received it, and a way to change its answers. */
    model: { requests: modelRequests, respond: script },
    caller,
    signUp,
    deliver,
    runJobs,
    delivered,
    seedProject,
    addMember,
  }
}

export type DocumentsApp = Awaited<ReturnType<typeof makeDocumentsApp>>

/** Some text as the bytes of a file, for an upload. */
export const bytesOf = (text: string): Uint8Array => new TextEncoder().encode(text)

/** Uploads a file (a PDF unless told otherwise) as its project's owner and carries it through the relay and the worker. */
export async function uploadAndProcess(
  app: DocumentsApp,
  project: SeededProject,
  filename = "report.pdf",
  contentType = "application/pdf",
) {
  const file = await uploadViaProcedures(project.call, project.slugs, {
    filename,
    contentType,
    // The files module checks the first bytes against the type the upload declares, so a picture
    // starts with the PNG signature and a document with the PDF one.
    bytes:
      contentType === "image/png"
        ? new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, ...bytesOf(`fixture ${filename}`)])
        : bytesOf(`%PDF-1.4 fixture ${filename}`),
  })
  await app.deliver()
  await app.runJobs()
  return file
}