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 working 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 needed — the signature is the authorization).
3. documents.complete measures the stored object with the driver, rewrites document.size to what it really is, re-checks the workspace quota against that size, and marks the document ready. An object larger than was declared (capped by MAX_UPLOAD_BYTES) is deleted and answered 413 PAYLOAD_TOO_LARGE; one that never arrived is 412 PRECONDITION_FAILED; a smaller one is accepted and logged, because the row — which is what the quota counts — must say what is really in the bucket.

The PUT is bounded rather than trusted. 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) and counts the bytes as they stream, aborting a chunked body that lied about its length; it returns no ETag. A bucket enforces none of that, which is why step 3 measures the object instead of believing the client.

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; what happens to images afterwards is OCR); 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 rather than the page’s. The Documents page pages it with a Load more button, and uploads through XMLHttpRequest rather than fetch — the only one of the two that reports progress, so every file shows its own percentage.

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/:org/:doc
set	s3	Cloudflare R2, AWS S3, MinIO… via Bun’s built-in S3Client	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. Same idea as the console email provider: the local driver is right for development and single-box self-hosting; production points at 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, but documents live in workspaces and a workspace is what a team sees fill up. Usage is the sum of document.size over the workspace’s pending and ready rows, so an upload in flight already counts; 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 frees its bytes.

Jobs

Postgres is the queue. A job row is claimed with UPDATE … WHERE id = (SELECT … 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 job still running after twice JOB_TIMEOUT_MS is presumed lost and requeued by the minute sweep.

A job that failed for good can be retried by hand — Retry on the Jobs page, or jobs.retry. It charges the kind’s credits again, so it answers LIMIT_REACHED when the plan is out of them and CONFLICT when the job is not failed; the page surfaces both as a toast. It queues a fresh job rather than flipping the old row back to queued: the failed attempt and its error stay in the list, input.retryOf links the new job to the old one, and the credits are charged again — the original’s charge was refunded when it gave up, and usage_ledger is unique per (job_id, reason), so re-charging the same id would be dropped silently and the retry would run for free.

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 stream re-earns the right to exist every 30 seconds — removing a member, changing their role, revoking their sessions or banning them ends it — and no connection lives longer than 15 minutes, after which 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 rather than failing silently. Events carry input and result only once a job is finished.

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.
JOB_TIMEOUT_MS	120000	Per-attempt time budget; the handler’s abort signal fires when it runs out.

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 every minute, delete finished jobs older than 30 days and abandoned uploads daily.

Adding a kind

1. Add it to JOB_KINDS and the JobCreateInputSchema union in packages/api-contract (plus a result schema if the UI should render it).
2. Price it in jobCredits (packages/billing/src/catalog.ts) — the API refuses unpriced kinds at compile time.
3. Implement the handler in apps/api/src/jobs/handlers.ts; throw JobError(message, false) for failures a retry cannot fix.
4. Label it in the web app ($lib/jobs.ts, messages app_jobs_kind_*).

A worker only claims kinds its own build has a handler for, which is what makes a rolling deploy safe: while the old replicas are still running, a job of a new kind waits queued for a replica that knows it instead of being claimed and failed for good. The other side of that rule is that a kind nothing handles is never claimed at all — it stays queued rather than being destroyed.

Search

Semantic search over a workspace’s documents runs on pgvector inside the same Postgres — the compose images are pgvector/pgvector: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; and 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 rather than only the shared per-IP limiter:

• 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 counts as free: charging on a guess would take credits for work the buyer’s own hardware did.
• 30 searches per minute per user (TOO_MANY_REQUESTS), keyed on the account rather than the IP address a whole office shares.
• A 10-second timeout on the call to the AI service. Any failure answers a generic SERVICE_UNAVAILABLE (503) — the service’s own message quotes provider URLs, model names and its own exceptions, so it 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 deliberately 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, −cost when it fails for good or is canceled; 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. 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 and — when the configuration cannot serve (no token check, an OCR backend that cannot run here) — HTTP 200 with ok: false and one sentence per problem in problems. Not a 503: a restart cannot fix an environment, so the container healthcheck asserts liveness only and /admin/system shows the problems in red 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 — 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 still answer 415 with the same message as before. 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; rasterising pages is a follow-up.

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.*, JOB_KINDS, result schemas
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/services/ai.ts
services/ai/src/ai/{main,extract,ocr,providers,settings}.py
apps/web/src/routes/(app)/app/[workspace]/{documents,jobs}/+page.svelte