The ingestion queue

One Cloudflare Queue carries small, typed wake-ups for response enrichment and the platform's durable jobs. Request paths do the fast authoritative work first — validate, persist Postgres/R2 state, then enqueue — while the consumer advances one bounded step. A missed wake cannot lose an accepted import, erasure, retention, notification or export: the 15-minute Cron sweep rediscovers its durable ledger. This is the validate-write-enqueue pattern from AGENTS.md.

The message

packages/ports/src/queue.ts defines the envelope and the discriminated payload union. Messages carry only IDs/capabilities (no selectors, recipients or blobs — the 128 KB queue cap means you reference durable state, never embed it):

export interface QueueMessage<T> {
  tenantId: string;
  idempotencyKey: string;   // required for safe at-least-once delivery
  payload: T;
}
export type IngestionPayload =
  | { kind: 'response'; responseId: string; surveyId: string }
  | { kind: 'file-drop-run'; runId: string; notBefore?: string }
  | { kind: 'subject-erasure'; jobId: string; generation: number }
  | { kind: 'retention-purge'; requestId: string; dispatchToken: string }
  | { kind: 'notification'; wakeId: string }
  | { kind: 'export-run'; runId: string; generation: number };

api/src/queue.ts aliases the concrete type as IngestionMessage = QueueMessage<IngestionPayload>.

Producers

Response enrichment remains best-effort because capture is already complete and enrichment is recomputable. enqueueIngestion (api/src/ingest.ts) publishes it inside waitUntil, so it never adds latency to the respondent's response.

Durable producers first commit their immutable request/job state. File drops, erasure, retention, transactional notifications and exports then publish a capability-only wake. Successful workers publish bounded successor wakes where useful; schedule edits, cancellation and erasure advance generations so stale messages cannot mutate current work. If the queue binding is missing or publication fails, the ledger remains due and Cron recovers it.

The makeCfQueues adapter (packages/adapters/src/cf-queues.ts) maps the logical queue name to its wrangler binding (ingestion → QUEUE_INGESTION). When the binding is absent, enqueue is a no-op rather than a throw — so capture stays synchronous and correct until the queue is provisioned at deploy.

Consumer — processQueueMessage

The Worker's queue handler (api/src/index.ts) routes batches to handleQueueBatch:

export async function handleQueueBatch(batch, env): Promise<void> {
  const ports = buildContainer(env);          // one container/DB connection for the whole batch
  try { await runBatch(batch, (body) => processQueueMessage(ports, body)); }
  finally { await ports.dispose?.(); }
}

parseIngestionMessage strictly validates the envelope before any tenant context is derived. The consumer then dispatches by payload.kind:

• response runs sentiment scoring, response roll-up and topic auto-coding;
• file-drop-run, subject-erasure and retention-purge advance one leased, checkpointed job step;
• notification claims and sends one durable outbox delivery;
• export-run builds, manifests, delivers or cleans up one bounded export step.

Response enrichment runs on the consumer's own DB connection (long after the respondent's request connection was disposed):

const ctx = systemContext(msg.tenantId);
const { scored } = await verbatim.analyze(ctx, surveyId);   // score un-scored verbatims
const overall   = await verbatim.rollupResponse(ctx, responseId); // overall response sentiment
const { coded } = await verbatim.autoCodeResponse(ctx, responseId); // topic/theme tagging

systemContext builds a trusted system context: there is no HTTP principal, so it carries no permissions and pins RLS to the envelope's authoritative server-derived tenantId. Durable jobs then re-read and compare their stored capability/generation before doing any work.

Ack / retry contract

runBatch acks on success and retries on failure, per message:

for (const message of batch.messages) {
  try { await process(message.body); message.ack(); }
  catch (err) { console.error(…); message.retry(); }
}

One bad message never fails its siblings. Cloudflare redelivers retried messages, then routes to the DLQ (voc-ingestion-dlq) after max_retries (3) — configured in api/wrangler.toml. process is injected so the ack/retry logic is unit-testable without a DB.

Idempotency

Response enrichment is idempotent: analyze only touches un-scored answers, autoCodeResponse only un-coded ones, and rollupResponse is a pure recompute. Durable handlers add tenant-scoped leases, generations, request fingerprints, provider idempotency keys and positive R2 deletion evidence. Redelivery or overlapping Queue/Cron claims therefore either resume the same work or become stale no-ops. See Sentiment & coding for the enrichment guarantees.

Provisioning

The active [[queues.producers]] / [[queues.consumers]] blocks in api/wrangler.toml require both queues to exist before deployment:

wrangler queues create voc-ingestion
wrangler queues create voc-ingestion-dlq

Without these bindings, capture stays synchronous and response enrichment is skipped; accepted durable jobs remain safe but progress only when their Cron recovery path can run. See Configuration.