// 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 { 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 /** 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> = [] 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 => { 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 => { 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 => { 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 { 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> /** 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 }