Documents, jobs and the AI service

The kit ships the machinery an AI feature needs around the model: getting files in, running work that outlives a request, metering it against the plan, searching what was uploaded and showing progress. All of it works on a fresh clone with no accounts. Decision records: docs/DECISIONS.md D27 (documents, jobs, the service) and D29 (search, OCR, storage quotas).

Documents

Documents belong to a workspace and are uploaded straight to storage:

1. documents.createUpload (POST /api/v1/documents) validates name, type and size, inserts a pending row and returns a presigned PUT target (url, headers, expiresAt).
2. The browser PUTs the bytes to that URL. No session is needed: the signature is the authorization.
3. documents.complete requires the exact declared size, checks the workspace quota and commits a record of the selected source generation. It then publishes those bytes to a unique, server-only final key and marks the document ready. A retry uses the same source generation and destination; it never selects replacement bytes from a reused upload URL.

Browser upload URLs address only staging/<organization>/<upload>. Downloads and jobs read objects/<organization>/<document>/<generation>. Reusing a valid upload URL can replace staging bytes, but cannot modify a completed document. Local finalization publishes a complete directory containing the file and its verification marker atomically. S3/R2 finalization verifies a SHA-256 digest and conditionally writes the captured bytes to the final key. It adds server transfer during completion; uploads and downloads remain direct between browser and bucket. Completion measures and copies with no database transaction open, and holds the organization’s lock, which every quota write of the organization takes, only to check the quota and write the rows, so a slow bucket holds up no other upload, job or workspace of the organization.

A missing or incorrectly sized source returns 412 PRECONDITION_FAILED; an oversized object returns 413 PAYLOAD_TOO_LARGE. Either way a refused object is deleted before the answer, so upload the file again before you retry. A timed-out finalization remains pending and can be retried. If its fixed source generation has disappeared before publication, delete that pending document and start a new upload. An already-ready completion is idempotent.

The PUT is bounded on the local driver: the API’s own route requires a Content-Length (411 length_required), refuses a declared size over MAX_UPLOAD_BYTES (413 too_large), counts the bytes as they stream, and aborts a chunked body that lied about its length. It returns no ETag. On S3 and R2 the URL is signed for the declared Content-Length, which the browser sends for the file, so a provider that checks the signature refuses any other length (403) and stores nothing. Step 3 measures the object all the same, for a provider that does not.

Allowed types are the contract’s DOCUMENT_CONTENT_TYPES: PDF, plain text, Markdown, CSV, HTML, JSON, and PNG, JPEG, WebP and TIFF images. GIF, HEIC and other image types are refused at upload. See OCR for what happens to images afterwards. The cap on one file is MAX_UPLOAD_BYTES (25 MiB by default). The cap on a workspace is its storage quota. Uploads that never complete are purged after a day.

documents.downloadUrl returns a short-lived presigned GET. documents.text returns the text an extract job pulled out of the file. documents.list is paged (limit default 50, max 200, and an opaque cursor) and returns { items, nextCursor, storage }, where storage is the whole workspace’s usage against the quota. The Documents page pages it with a Load more button and uploads through XMLHttpRequest, which reports progress, so every file shows its own percentage.

A member deletes the documents they uploaded; owners and admins (document: deleteAny) delete anyone’s, and the Documents page offers Delete only where mayDelete allows it. A document in a workspace scoped to a team reaches only that team and the organization’s owners and admins (Teams are access boundaries).

Storage drivers

@repo/storage picks the driver from env:

S3_BUCKET	Driver	Where the bytes live	URLs
unset	local	STORAGE_DIR on the API’s disk (./data/uploads, the uploads volume in compose)	Signed by the API and served by PUT/GET /uploads/<key>
set	s3	An S3-compatible bucket; Bun signs browser URLs and the AWS SDK handles conditional storage operations	Presigned by the bucket. The bucket’s CORS rules must allow the web origin (PUT, GET, header Content-Type)

The web app has one code path for both. The local driver suits development and single-box self-hosting. Validate the provider gate below before moving to a bucket.

Storage quota

Every plan has limits.storageBytes per workspace (Free 100 MB, Pro 5 GB, Team 50 GB by default). The other limits are per organization. Usage is the sum of document.size over the workspace’s pending and ready rows, plus the workspace’s stored files, so an upload in flight already counts and a product’s attachments share the cap. The check and the insert run in one transaction holding pg_advisory_xact_lock(hashtext(organizationId)), so two parallel uploads cannot both squeeze under the cap. documents.createUpload checks used + size ≤ limit with the same assertWithinLimit as workspaces, seats and credits. Over it the API answers LIMIT_REACHED with { kind: 'storageBytes', limit, plan }. While billing is off nothing is capped and storage.limitBytes is null.

documents.list and workspaces.get both return storage: { usedBytes, limitBytes }. The Documents page and the workspace’s settings show the meter. Deleting a document releases its quota immediately. Its database deletion also records a durable storage_deletion task, including when a workspace or organization cascade removes it. The worker retries object deletion every minute, with backoff after failures, and only acknowledges successful deletion after the upload URL expires. Keep a worker running until this queue drains. Object storage can therefore lag behind the UI.

An upload already in progress at URL expiry can still finish later, particularly on S3. The worker therefore also sweeps old staging objects and unreferenced final generations, using durable cursors that eventually wrap through the whole namespace. These sweeps use object modification time and a minimum age of 24 hours. Legacy client-writable keys remain in a permanent retry registry so a late write cannot erase the last cleanup locator. Keep a worker running even when the immediate deletion queue is empty. Object-store lifecycle expiration for staging/ is additional protection during a worker outage.

Upgrading an existing deployment

Migration 0009_immutable_uploads adds the promotion records and cleanup cursors. Apply it with db:migrate, which also installs its triggers: they are SQL in the migration file, not part of the schema. Stop all old API and worker replicas before starting the new application. The shipped Compose deployment scripts drain them after images are available.

The worker migrates legacy ready documents in batches. It preserves a row if the old source is missing or cannot be verified, and reports the problem for operator action. A successfully migrated file gets a new final key; its extracted text and search index must be rebuilt from those verified bytes. Legacy pending uploads must be restarted. Image-only rollback to a release that does not understand the new storage layout is unsupported; recovery requires a compatible image or restoring the matching pre-upgrade database and uploads together.

Validate a configured cloud provider using the gated storage integration suite before deploying this protocol there. Passing local-disk or simulated-provider tests does not establish live AWS S3 or R2 behavior. The provider test setup is documented with the storage package.

Jobs

Postgres is the queue. A job row is claimed with UPDATE … WHERE id = (SELECT … ORDER BY priority, run_at, created_at FOR UPDATE SKIP LOCKED LIMIT 1), which is safe with any number of workers and needs no broker.

Kind	Input	Credits	What happens
document.extract	documentId	1	bytes → services/ai /v1/extract → text stored on the document
document.summarize	documentId, maxSentences	5	reuses the stored text (extracts if missing) → /v1/summarize
document.index	documentId	2	reuses the stored text (extracts if missing) → chunks → /v1/embed in batches → document_chunk rows for search
text.summarize	text, maxSentences	5	/v1/summarize
text.embed	texts (≤ 16)	1	/v1/embed → vectors in the job result (not stored; document.index is what stores)

Lifecycle: queued → running → succeeded | failed, or canceled (only while queued). Failures retry with backoff (10 s, 30 s, 90 s) up to maxAttempts unless the input is at fault (missing document, unsupported file, 4xx from the service). A running job is leased: its worker renews the lease every 20 seconds until twice the kind’s timeout has passed. A job whose lease has run out, because its worker died or because it is still running at twice its timeout, is taken back by the minute sweep: queued again, or failed when it has no attempt left. A machine that sleeps does not take back its own jobs: the sweep renews its own process’s leases before it takes any back, and on macOS, where it can tell that its process slept, it takes nothing back after a sleep until the other workers on the machine have had 40 seconds to renew theirs. A worker whose database or sweeping peers stay awake elsewhere while its machine sleeps still looks dead to them and loses its jobs after a minute.

Final failure, cancellation, and stale-attempt exhaustion commit their credit refund in the same transaction as the job state. A failed refund rolls back the transition so a later attempt can recover; retries that remain queued keep their charge.

A job that failed for good can be retried by hand: Retry on the Jobs page, or jobs.retry. It queues a fresh job rather than flipping the old row back to queued. The failed attempt and its error stay in the list, and input.retryOf links the new job to the old one. The kind’s credits are charged again, because the original’s charge was refunded when it gave up. It answers LIMIT_REACHED when the plan is out of credits and CONFLICT when the job is not failed. The page shows both as a toast.

Procedures: jobs.create, jobs.list (filters: workspace, document, status; limit plus an opaque cursor; each row’s result is omitted unless you pass includeResult), jobs.get, jobs.cancel, jobs.retry, and jobs.stream, an oRPC event iterator. The typed client gets an async iterator (for await (const job of await api.jobs.stream({ organizationId }))). REST clients get plain server-sent events at GET /api/v1/organizations/{organizationId}/jobs/stream. The web app subscribes once per page and updates its lists in place. A job’s progress is { done, total } once its handler reports one (Long jobs), kept when it finishes, and null otherwise.

A job in a workspace follows the workspace’s team boundary. A job outside any workspace, created through the API without a workspaceId or left behind when its workspace was deleted, is visible only to the member who created it and to owners and admins. jobs.list and jobs.stream answer NOT_FOUND for a workspaceId the caller cannot see, and leave the jobs a caller cannot see out of the organization-wide list and feed.

Customers see only the kinds the contract lists in JOB_KINDS. A job of an internal kind and a system job, which belongs to no organization, are left out of jobs.list and jobs.stream, and jobs.get, jobs.cancel and jobs.retry answer NOT_FOUND for them, even to an owner (Adding a kind).

An open stream reads its feed when a job_changed notification names its organization, at most once a second, and every 10 seconds without one; with DATABASE_LISTEN=false it reads every second.

A stream re-checks its authorization every 30 seconds. Removing a member, changing their role, revoking their sessions or banning them ends it, or, for a role that may still read jobs, changes what it shows from then on: a demoted admin’s feed loses other teams’ jobs. A stream of one workspace also ends once the member may no longer see it: the workspace moved to a team they are not in, or was deleted. No connection lives longer than 15 minutes; after that the client reconnects. One account may hold five at a time. A sixth is refused with TOO_MANY_REQUESTS, which the app shows as a “too many open job streams” status. Events carry input and result only once a job is finished.

A stream of your own

A live feed of yours takes the shape of jobs.stream in apps/api/src/rpc/jobs.ts: an async generator on authed behind orgScoped, a slot from slotCounter (./limits) released in a finally, and refreshMember(context, member, permissions, input.workspaceId) from ./tenancy on an interval, ending the stream when it answers null, with a ceiling on the connection. Pass the workspaceId of a stream that watches one workspace: the guard checked it once, when the stream opened, and a feed filtered by viewer would only go quiet.

Wait with watchJobChanges(organizationId) from apps/api/src/jobs/changes.ts, or with listenTo from apps/api/src/pg-listen.ts for a channel of your own, and read again after a notification or a quiet interval: a notification is a hint, and the read is what the stream sends. Open the watcher before the first read, so a change committed during it is not missed, and close it in the finally beside the slot. Announce changes of your own after they commit, never inside a busy write transaction: PostgreSQL serializes the commits of transactions that notify.

Refusals before the first event. A stream that refuses at the start, as jobs.stream does past the account’s open streams, has already answered 200: the refusal arrives as the iterator’s first error, an ORPCError whose status is 429, and over REST as an SSE event: error with the error’s JSON. Read code and status off the error, never the HTTP response. The procedure limiter refuses before the stream starts, with a real 429 and data.retryAfter: retryAfterOf(error) from @repo/api-client tells the two apart, so wait out the one and give up on the other, as subscribeToJobs does.

Testing a stream. configureJobStream({ pollMs, idleMs, recheckMs, maxMs }) shortens jobs.stream’s intervals for a test and returns the settings it replaced, for afterEach to put back. configureJobChanges(listen) from apps/api/src/jobs/changes.ts replaces the LISTEN for a test that plays Postgres, or turns it off as DATABASE_LISTEN=false would. Give a stream of yours a seam like it. apps/api/src/rpc/jobs-stream.db.test.ts is the pattern: open the stream with call(), take access away while it is open, and expect next() to end within a re-check, long before the ceiling.

Close every stream a test opens with return() on its iterator. A generator suspended at a yield runs its finally, which releases the slot or unsubscribes, only when it is returned, as oRPC does when the client goes away. Aborting its signal does not end it, so a test that only aborts keeps the slot for every test after it. Abort first and then return: return() on a stream that is still polling waits for its next yield, which may never come, and the abort ends the polling at once.

The worker

The worker runs inside the API process by default:

Variable	Default	Meaning
WORKER_ENABLED	true	Run the worker loop in the API process. false when a dedicated worker runs.
WORKER_CONCURRENCY	2	Jobs processed at the same time by one process, besides the kinds that run on lanes of their own (webhook deliveries).
JOB_TIMEOUT_MS	120000	Per-attempt time budget of a kind without a timeoutMs of its own. The handler’s abort signal fires when it runs out.

With WORKER_CONCURRENCY above 1, slot 0 claims only the kinds customers see. Internal kinds of yours run on the other slots, so a burst of them never holds up the documents and jobs customers are waiting for. A single slot runs everything. Size it for both: a product whose own work is internal and steady needs WORKER_CONCURRENCY of at least 3 for two slots of it, or moves that work into a loop (Loops), which takes no slot.

A kind whose jobs mostly wait on a server someone else chose runs on lanes of its own instead of the slots: lanes: { size, perOrganization } in its definition, size at once in each worker process and at most perOrganization for one organization. The lanes take organizations in turns: a lane passes over an organization that has its share running, or has had its turn, and takes the next one’s job until every organization with a job due has had one. So one organization’s slow servers hold up its own jobs, and another organization’s by one job at most rather than by their backlog. The kit’s webhook deliveries (webhook.deliver) run on four lanes, one per organization.

A job queued anywhere wakes idle workers in every process at once: the commit that queues it sends job_queued, and each worker listens for it on one Postgres connection beyond DATABASE_POOL_MAX. Without notifications (DATABASE_LISTEN=false) a worker asks every second, as it also does for a retry that comes due. A stopping worker stops your loops, lets running jobs finish for 5 seconds, then aborts their signal; a job whose handler stops goes back to the queue at once, its attempt not spent, for the next process.

To scale it separately, run bun run --cwd apps/api worker (compose: COMPOSE_PROFILES=worker in .env) and set WORKER_ENABLED=false on the API replicas. Housekeeping is Bun.cron in whichever process runs the worker: requeue stale jobs and retry storage deletions every minute, delete finished jobs and webhook deliveries older than 30 days, abandoned uploads and unclaimed or expired stored files daily.

In development, bun run setup writes WORKER_ENABLED=true into the root .env, so the API that bun run dev:app starts runs the worker as well. To run it as its own process, set WORKER_ENABLED=false there, restart the API, and start bun run --cwd apps/api worker in another terminal.

Adding a kind

Your job kinds go in apps/api/src/jobs/product.ts, which kit releases never change, built with defineJob from ./define:

export const productJobs = [
  defineJob({
    kind: 'report.export',
    input: z.object({ reportId: z.string() }),
    maxAttempts: 5,
    priority: 5,
    timeoutMs: 10 * 60_000,
    async run({ job, input, signal, progress }) {
      const file = await exportReport(input.reportId, job, signal, progress);
      return { fileId: file.id };
    },
  }),
  defineJob({ kind: 'report.cleanup', internal: true, run: cleanUpReports }),
];

• Input is checked against input before run. A mismatch fails the job for good instead of retrying it, since another attempt would read the same stored input. The retryOf key that jobs.retry adds to a copied job is removed first, so a .strict() schema still accepts a retry. Throw JobError(message, false) for any other failure a retry cannot fix; everything else retries with backoff.
• Visible or internal. Customers see only the kinds the contract lists in JOB_KINDS: the kit’s AI kinds and yours in PRODUCT_JOB_KINDS (packages/api-contract/src/product-jobs.ts). A kind they should see in the Jobs list goes there, with a label in the web app. An internal: true kind must not; it is never listed, streamed or found by id, stays off the worker’s slot 0 when there is more than one slot (The worker), and bun run test fails for a kind that is both or neither. A job of a kind customers see also tells the organization’s webhooks when it succeeds or fails for good (job.succeeded, job.failed).
• Credits. A kind costs credits only when it is priced in jobCredits (packages/billing/src/catalog.ts); everything else is free and needs no plan.
• Priority. priority, -100 to 100, lower runs first among the jobs that are due, 0 by default; enqueueJob({ priority }) overrides it for one job. Lower numbers hold back higher ones for as long as they keep coming, so keep the spread small: a kind customers wait for at -10, a background sweep at 10.
• Time. timeoutMs is this kind’s budget per attempt (JOB_TIMEOUT_MS otherwise): the handler’s signal aborts and deadline passes when it runs out (Long jobs). The worker refuses at start a kind whose priority, timeoutMs (a whole number, at least 1,000) or maxAttempts (at least 1) it cannot use.
• When it fails for good. onFailure({ job, error }) runs once the job has failed for good (Long jobs).
• System jobs. A free kind may run for no organization at all (organizationId: null): work for the deployment itself, which no customer ever sees. A priced kind always needs one.

Queue a job with enqueueJob from apps/api/src/jobs/enqueue.ts, from a procedure, a schedule or another job:

await enqueueJob({ kind: 'report.export', organizationId, workspaceId, createdBy, input });

A priced kind is checked against the plan and charged in one transaction under the organization’s lock, and refuses a caller’s executor: withLimit takes that lock in a transaction of its own, so queueing it inside yours could wait on yourself. Queue it after your transaction commits. A free kind joins your transaction when you pass { executor: tx }, so it commits or rolls back with your writes, and the workers hear of it when you commit. jobs.create and jobs.retry stay the customer’s way to queue the AI kinds.

Two more options. dedupeKey keeps one queued job per kind and key: while one is queued, queueing another returns it with deduplicated: true and changes nothing about it. A running job does not count, so a change made while one runs still gets a job, which may start while the first still runs; name what the job works on (document:<id>). Keys are shared by every organization: a key another organization’s queued job holds throws rather than hand that job over, so never use a word every organization would (rebuild). A key is 1 to 200 characters without control characters, and a priced kind takes none. enqueueJobs([...]) queues many free jobs in one transaction and returns them in order; a priced kind is refused there, because each is charged under the organization’s lock. The jobs of one call are queued at the same instant, so their order among themselves is not kept.

Queue every job through enqueueJob or enqueueJobs, never by inserting into job or with insertJob: they apply the kind’s definition (its attempts and priority), check the input and charge the kind. A query in packages/db that decides whether to queue one runs in your transaction instead:

await db.transaction(async (tx) =>
  (await hasWork(tx))
    ? enqueueJob(
        { kind: 'imports.drain', organizationId: null, dedupeKey: 'drain' },
        { executor: tx },
      )
    : null,
);

The dedupe key keeps one queued, so no lock is needed. The worker logs a warning for a job whose attempts differ from its kind’s definition, which is what a row inserted by hand looks like.

A worker only claims kinds its own build has a handler for. That makes a rolling deploy safe: while old replicas are still running, a job of a new kind waits queued for a replica that knows it. A kind nothing handles is never claimed. It stays queued.

Long jobs

A job gets its kind’s timeoutMs, or JOB_TIMEOUT_MS, per attempt. Its handler’s signal aborts when that runs out, when the process stops, or when another worker has taken the job over, and deadline says when the time is up. Work longer than a reasonable budget goes in pieces: check deadline before each unit, and when the next one would not fit, queue a continuation (the same kind, with what is left in its input or in a row of yours, and a dedupeKey naming the run) and return. progress(done, total) shows how far it has got, in the job and in jobs.stream; it is written at most once a second, the latest value winning. A test that calls a handler itself builds its context with jobContext(job, { signal }) from apps/api/src/jobs/define.ts.

signal.reason says why the signal aborted, with a type from apps/api/src/jobs/define.ts: a JobTimeoutError once deadline has passed (a DOMException named TimeoutError, as AbortSignal.timeout() gives), a JobStoppedError when the process is stopping, and a JobLeaseLostError once the sweep has taken the job back. A handler that queues its continuation when its signal aborts skips it for a lost lease: the sweep has already queued the job again, or failed it, and nothing the attempt records is kept, so the continuation would be a second job for the run.

async run({ job, input, signal, deadline }) {
  // Exports file after file until the run is done, `deadline` is near or `signal` aborts.
  const finished = await exportUntil(input.runId, deadline, signal);
  // Taken back, this job is queued again and goes on with the run itself.
  if (finished || signal.reason instanceof JobLeaseLostError) return { finished };
  await enqueueJob({
    kind: job.kind,
    organizationId: job.organizationId,
    input,
    dedupeKey: `report.export:${input.runId}`,
  });
  return { finished };
}

jobContext(job, { signal: AbortSignal.abort(new JobLeaseLostError()) }) hands a handler under test a lease already lost.

Give the run’s first job the same dedupeKey as its continuations. The sweep sends a job back to the queue with the key it has, so a continuation queued after that finds it and adds nothing; a first job queued without a key goes back without one, and a continuation its aborted handler queues sits beside it. A key only keeps out a second queued job, though. Once a worker has claimed the job the sweep sent back, a continuation is a job of its own, and so is one queued before the sweep took its job back, which then goes back without its key: either way the run has two jobs, which may run at once. Skipping the continuation for a JobLeaseLostError avoids most of that, and the next paragraph covers the rest.

The key is also how a run is canceled while its continuation waits in the queue: cancelQueuedJobByKey({ kind, dedupeKey, organizationId }) from @repo/db cancels the queued job that holds the key, if it is that organization’s, and refunds it as jobs.cancel does. A job of the run that is running holds no key and is left alone, so mark the run canceled in a row of yours as well, and have the handler read it before each unit.

A handler must also cope with a copy of itself. A worker keeps its job leased while the attempt runs, up to twice its timeout; a handler that ignores its signal for longer is taken back by the sweep and runs again while the first copy is still going, and a worker that loses its database connection for a minute loses its jobs the same way, as does one whose machine sleeps while the database or another worker’s sweep runs elsewhere. A handler still running 5 seconds into a shutdown is aborted and its job runs again from the start in the next process. Write so that two runs of one job do no harm: key what you write by the unit of work, or check a row of yours before writing.

onFailure({ job, error }) runs once a job has failed for good: its last attempt threw, its input was refused, or its worker was lost during the last attempt. End what the job was part of there, such as a run of yours that would otherwise stay “running”. It runs at most once, after the failure is recorded; a process that dies in between loses it. A throw is logged and reported. It is not called for a cancellation.

To test what the sweep does to a job of yours, play its worker with claimTestJob from @repo/db, which claims the job you name, due or not, and no other; a negative leaseMs gives it a lease that has already run out. Then sweep it alone: requeueStaleJobs takes { ids } or { kinds }. On a shared database claimJob may claim another suite’s job, or a development worker’s, and a sweep without ids takes back every lost job there.

await claimTestJob(job.id, { leaseMs: -1_000 });
const { requeued } = await requeueStaleJobs(new Date(0), { ids: [job.id] });
// The handler that held it wakes to its signal aborted with the lease lost.
await handler(jobContext(job, { signal: AbortSignal.abort(new JobLeaseLostError()) }));

Schedules

Recurring tasks go next to your job kinds, in productSchedules in the same file:

export const productSchedules = [
  defineSchedule({ name: 'digest.weekly', cron: '0 7 * * 1', run: sendWeeklyDigests }),
];

cron is a Bun.cron expression in UTC, and name must be unique and stable. Every process that runs the worker runs the schedules, so each tick is claimed in the schedule_run table first and runs in exactly one of them: three workers do not send three digests. A failure is logged and does not stop the next tick. A schedule runs at most once per tick: a process that dies mid-task loses that tick. Work that must not be lost belongs in a job, so a task that does real work queues one with enqueueJob and returns. The public demo runs no worker, so no schedule runs there.

Loops

Work that belongs to a process rather than to a job, such as a watcher, a claim loop over a table of your own or work only some machines can do, goes in productLoops (apps/api/src/jobs/product-loops.ts), built with defineLoop:

export const productLoops = [
  defineLoop({
    name: 'imports.watch',
    async run({ signal, sleep }) {
      while (!signal.aborted) {
        await importWaitingFiles(signal);
        await sleep(5_000);
      }
    },
  }),
];

Every process that runs the worker runs every loop, from the worker’s start to its stop. A loop that returns or throws starts again after a pause, 1 s doubling to a minute, back to 1 s once a run has lasted a minute; a failure is logged and reported. sleep(ms) returns early, with false, once the worker stops. At shutdown signal aborts, and the loop has the drain’s 10 seconds to return. A loop runs in each process, so one that must run once across them takes a lock of its own, or is a schedule. Since a loop decides for itself what it can do where it runs, it is also how a product routes work to the machines that can do it: each process claims only what it can reach. To wake one when your rows change, listenTo a channel of yours (apps/api/src/pg-listen.ts). Two loops with one name stop the worker at start.

Search

Semantic search over a workspace’s documents runs on pgvector inside the same Postgres. The compose images are pgvector/pgvector:0.8.7-pg17, and migration 0006_document_chunks.sql runs CREATE EXTENSION IF NOT EXISTS vector before creating document_chunk. On a Postgres without the extension bun run db:migrate fails at that line. Install pgvector (or use the image) first.

Indexing

A document becomes searchable when a document.index job has run for it: Index for search on the Documents page, or jobs.create({ kind: 'document.index', documentId }). The job:

1. reuses the text an extract job stored on the document (extracts first if there is none);
2. splits it into chunks of about 1,200 characters with 150 characters of overlap, cutting on paragraph and sentence boundaries (apps/api/src/jobs/chunk.ts). More than 2,000 chunks is a non-retryable failure;
3. embeds them in batches of 32 through /v1/embed;
4. replaces the document’s previous chunks in one transaction and sets indexedAt / chunkCount (the “Indexed · N chunks” badge; the job result is { chunks, characters, model, dimensions }).

Each chunk stores the embedding and the model that produced it. Re-index after re-extracting a document. After switching the embedding provider (OPENAI_API_KEY, OPENAI_EMBEDDING_MODEL), every chunk indexed by the old model is invisible to search until its document is re-indexed.

Searching

documents.search (POST /api/v1/workspaces/{workspaceId}/documents/search, permission document: read) takes { query, limit } (limit ≤ 50, default 10), embeds the query with one /v1/embed call and returns:

{ "model": "local/hashed-bow-256",
  "results": [{ "chunkId": "…", "documentId": "…", "documentName": "…", "index": 3, "text": "…", "score": 0.83 }] }

score is cosine similarity (1 = identical, 0 = unrelated). Results are the nearest chunks of the workspace whose model matches the query’s, best first. The search box lives on /app/[workspace]/documents.

It is the most expensive procedure in the API (one embed call plus an exact scan of the workspace’s chunks), so it carries its own guards on top of the shared limiter (per address, or per user or key):

• One AI credit per search, unless embedding is local. The service’s /health says which provider is configured, cached for a minute. A failed probe stops the request before embedding. A paid search first reserves its credit: a short transaction under the organization’s lock checks the quota and writes the charge, and the provider is called only once that has committed, so a slow provider holds neither the lock nor a database connection. A failed embedding gets a refund keyed on its reservation, which a retry cannot write twice. If the API stops between the two, the credit stays spent.
• 30 searches per minute per user (TOO_MANY_REQUESTS), keyed on the account rather than the IP address.
• A 10-second timeout on the call to the AI service. Any failure answers a generic SERVICE_UNAVAILABLE (503). The service’s own message goes to the log and never to a tenant.

The query is an exact scan ordered by cosine distance (<=>), restricted to one workspace and one model. There is no HNSW/IVFFlat index: pgvector can only index a column with a fixed dimension, and embedding has none because the local provider emits 256 dimensions and text-embedding-3-small 1,536. Once a deployment settles on one provider, fix the model and add CREATE INDEX … ON document_chunk USING hnsw (embedding vector_cosine_ops) in a migration.

AI credits

Every plan has limits.aiCredits per month (Free 50, Pro 2,000, Team 20,000 by default). Charges are an append-only usage_ledger: +cost when a job is queued or a paid search reserves its credit, −cost when the job fails for good or is canceled, or the search’s embedding fails. Usage is the sum over the current calendar month (UTC). jobs.create checks used + cost ≤ limit with the same assertWithinLimit as workspaces and seats. While billing is off nothing is capped. Over the limit the API answers LIMIT_REACHED with { kind: 'aiCredits', limit, plan }. organizations.usage reports usage.aiCredits and the period.

The AI service

services/ai (FastAPI, uv) is internal. Only the API calls it, with X-Service-Token. The service refuses to start without a token unless SERVICE_AUTH_DISABLED=true. /docs, /redoc and /openapi.json need the header too. Three job-shaped endpoints:

Endpoint	Body	Answer
POST /v1/extract	The document’s raw bytes with its media type as Content-Type (what the API sends), or { file: { name, content_type, data_base64 } }	{ text, characters, pages, truncated, content_type }
POST /v1/summarize	{ text, max_sentences }	{ summary, sentences, model, characters }
POST /v1/embed	{ texts }	{ model, dimensions, vectors }

Every body is bounded before it is parsed (MAX_UPLOAD_BYTES for a document, 413 above it, and a JSON nesting cap). One PDF may have at most MAX_PDF_PAGES pages (413), and one extraction at most EXTRACT_TIMEOUT_SECONDS of wall clock, OCR included (408). All three are 4xx on purpose: the job fails as final and refunds its credit instead of parsing the same bytes three more times. At most MAX_EXTRACT_THREADS extractions (2) run at once, one still running past its answer included, which bounds the service’s memory. One that waits for a thread and runs out of time, before it starts or after, is a 503 the job retries, since it did not have its whole budget. When the API gives up on the call (the job’s attempt timed out or lost its lease, or its worker stops), the service stops the parse: a queued one never starts, a running one stops at its next checkpoint. A summary or a batch of embeddings has OPENAI_TIMEOUT_SECONDS (60), every provider attempt and retry included, and running out is a final 408 too: the call may already be billed upstream. One whose time ran out before its first request went out, while it waited for a thread, is a 502 the job retries, since nothing was billed. Every call carries X-Timeout-Ms, how long the API waits (what is left of the job’s attempt, or a search’s 10 s) less a margin, and the service stops by then when that is sooner. The API waits the margin longer, so the answer it acts on is always the service’s, and no retry pays for a call that is still running. Each response echoes the API’s X-Request-Id, and the service logs one JSON line per request with it, so a failed job maps to exactly one line.

Behind them sits a provider: the deterministic local one (extractive summaries, 256-dimension hashed bag-of-words embeddings) until OPENAI_API_KEY is set, then any OpenAI-compatible endpoint (OPENAI_BASE_URL, OPENAI_CHAT_MODEL, OPENAI_EMBEDDING_MODEL) through plain httpx, retrying 429/5xx and connection errors inside OPENAI_RETRY_BUDGET_SECONDS. Images go through a separate OCR backend (below). /health reports both (provider, ocr) plus auth. When the configuration cannot serve (no token check, an OCR backend that cannot run here) it answers HTTP 200 with ok: false and one sentence per problem in problems. It is not a 503: the container healthcheck asserts liveness only, and a 503 would mark the container unhealthy and fail every deploy over its environment. /admin/system shows the problems next to the queue depth and the storage driver.

The API’s client is apps/api/src/services/ai.ts: three functions, one header, Zod on the way back.

OCR (images)

Images (PNG, JPEG, WebP, TIFF) upload like any other document. document.extract on one sends it to /v1/extract, where an OCR backend chosen once at start-up from OCR_PROVIDER reads it. With no backend configured the service answers 415, the job fails without retrying and its credit is refunded:

OCR_PROVIDER	What reads the image	Types
tesseract	The tesseract CLI as a subprocess (tesseract stdin stdout -l $OCR_LANGUAGES). No Python bindings, no Pillow	PNG, JPEG, TIFF, BMP, WebP
openai	One chat completion against OPENAI_BASE_URL with the image inlined as a data URI (OPENAI_VISION_MODEL). Needs OPENAI_API_KEY. Transcribes rather than recognises, so expect corrected typos	PNG, JPEG, WebP, GIF
auto (default)	tesseract if the binary is on PATH, else openai if a key is set, else none
none	Nothing: images answer 415 (OCR is not configured …)

The Docker image installs tesseract-ocr with the English pack, so the compose stack reads images out of the box. On a development machine install Tesseract yourself (apt install tesseract-ocr, brew install tesseract, the UB Mannheim installer on Windows) or set a key. With neither, images answer 415. An image type the active backend cannot read is 415 too. OCR failures and timeouts are 502, so the job retries with backoff. Scanned PDFs without a text layer are not OCR’d: pypdf returns empty text for them.

Variable	Default	Meaning
OCR_PROVIDER	auto	auto, tesseract, openai or none.
OCR_TESSERACT_BINARY	tesseract	Path of (or name on PATH for) the Tesseract CLI.
OCR_LANGUAGES	eng	Tesseract language pack(s), +-separated (eng+deu). Each must be installed (Docker: add tesseract-ocr-<lang> to services/ai/Dockerfile).
OCR_TIMEOUT_SECONDS	60	Wall-clock limit for one Tesseract run. Exceeding it is a 502.
OPENAI_VISION_MODEL	gpt-4o-mini	Vision-capable chat model for the openai backend.

The seam is ai.ocr.OcrProvider. A class with name, content_types and recognise(data, content_type) plus a branch in build_ocr adds a backend.

Where things are

packages/api-contract/src/index.ts     documents.*, jobs.*, AI_JOB_KINDS, JOB_KINDS, result schemas
packages/api-contract/src/product-jobs.ts   PRODUCT_JOB_KINDS (yours)
packages/storage/src/{index,local,s3}.ts
packages/db/src/schema/{documents,jobs}.ts   packages/db/src/{documents,jobs}.ts
packages/db/drizzle/0005_documents_jobs.sql   0006_document_chunks.sql (pgvector, document_chunk)
apps/api/src/rpc/{documents,jobs}.ts   apps/api/src/uploads.ts (local driver routes)
apps/api/src/jobs/{worker,handlers,chunk,schedules}.ts   apps/api/src/worker.ts (standalone entry)
apps/api/src/jobs/{define,enqueue}.ts   apps/api/src/jobs/product.ts (yours: jobs and schedules)
apps/api/src/services/ai.ts
services/ai/src/ai/{main,extract,ocr,providers,settings}.py
apps/web/src/routes/(app)/app/[workspace]/{documents,jobs}/+page.svelte