feat(pipeline): a durable job queue

apps/server/src/index.ts

@@ -24,7 +24,13 @@ import { ClaudeModel } from '@outreachgraph/ai';
 import { ResendMailer } from '@outreachgraph/email';
 import { createApp } from '../../api/src/app';
 import { pruneSessions } from '../../api/src/auth';
-import { expireSignals, processDeletion } from '@outreachgraph/pipeline';
+import {
+  drainQueue,
+  expireSignals,
+  processDeletion,
+  rescoreProspect,
+  type QueuedJob,
+} from '@outreachgraph/pipeline';
 
 const PORT = Number(process.env.PORT ?? 8080);
 const WEB_PORT = Number(process.env.WEB_PORT ?? 3001);
@@ -211,6 +217,34 @@ console.log(`outreachgraph listening on :${PORT} (${ENVIRONMENT})`);
 let running = true;
 let inFlight: Promise<void> = Promise.resolve();
 
+/**
+ * Runs one queued job.
+ *
+ * Dispatch is explicit rather than a lookup table so an unhandled kind is a
+ * loud failure with the kind in the message, not a job that quietly reports
+ * success. A throw here is how the queue is told to retry or, once attempts
+ * run out, to keep the row as `failed` with the reason attached.
+ */
+async function runJob(job: QueuedJob): Promise<void> {
+  switch (job.kind) {
+    case 'rescore_prospect': {
+      const { campaignId, personId } = job.payload as { campaignId?: string; personId?: string };
+      if (!campaignId || !personId)
+        throw new Error('rescore_prospect needs campaignId and personId');
+      await rescoreProspect(db, campaignId, personId);
+      return;
+    }
+    case 'process_deletion': {
+      const { deletionJobId } = job.payload as { deletionJobId?: string };
+      if (!deletionJobId) throw new Error('process_deletion needs deletionJobId');
+      await processDeletion(db, deletionJobId);
+      return;
+    }
+    default:
+      throw new Error(`no handler for job kind ${job.kind}`);
+  }
+}
+
 async function tick(): Promise<void> {
   const workspaces = await queryAll<{ id: string }>(
     db,
@@ -235,6 +269,17 @@ async function tick(): Promise<void> {
 
   const pruned = await pruneSessions(db);
   if (pruned > 0) console.log(`pruned ${pruned} expired sessions`);
+
+  // The queue drains last. The sweeps above are bounded and quick; this is the
+  // part that can take the whole tick, and `limit` is what stops it running
+  // away with the loop.
+  const drained = await drainQueue(db, runJob);
+  if (drained.processed > 0 || drained.reclaimed > 0) {
+    console.log(
+      `jobs: ${drained.succeeded} done, ${drained.retried} retrying, ` +
+        `${drained.dead} failed, ${drained.reclaimed} reclaimed`,
+    );
+  }
 }
 
 async function loop(): Promise<void> {

migrations/0007_jobs.sql

@@ -0,0 +1,53 @@
+-- A durable job queue (PRD §1.1 principle 7, "background jobs never run inside
+-- a request").
+--
+-- `JOB_KINDS` has described this table since the first commit; nothing ever
+-- created it. Work was either run inline in the request — `POST /prospects`
+-- still runs the whole enrich/resolve/research/score/recommend/draft chain
+-- before responding — or swept for by the worker loop querying its own domain
+-- tables directly. Neither survives a batch: one URL fans out to a fetch, an
+-- extraction, an identity search per candidate person and a draft per
+-- recommendation, and a hundred URLs is not a slow request but an impossible
+-- one.
+--
+-- Deliberately a table rather than Redis. `REDIS_URL` sits empty in the vault
+-- and unreferenced in the code, and Turso is already here, already backed up
+-- and already the thing every other durable fact lives in. A queue that can
+-- lose its contents on a restart is not a queue.
+
+CREATE TABLE jobs (
+  id            TEXT PRIMARY KEY,
+  workspace_id  TEXT NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
+  kind          TEXT NOT NULL,
+  payload_json  TEXT NOT NULL DEFAULT '{}',
+  -- pending | running | done | failed
+  status        TEXT NOT NULL DEFAULT 'pending',
+  attempts      INTEGER NOT NULL DEFAULT 0,
+  max_attempts  INTEGER NOT NULL DEFAULT 5,
+  -- Not before this instant. Carries both the initial delay and the retry
+  -- backoff, so there is one thing to order by rather than two.
+  run_after     TEXT NOT NULL,
+  last_error    TEXT,
+  -- When the current attempt claimed the row. A container that dies mid-job
+  -- leaves `running` behind forever without this; the reclaim sweep reads it.
+  started_at    TEXT,
+  finished_at   TEXT,
+  -- Caller-supplied identity for work that must not be queued twice — the same
+  -- URL arriving in two of the same bulk paste, say. See the index below.
+  dedupe_key    TEXT,
+  created_at    TEXT NOT NULL,
+  updated_at    TEXT NOT NULL
+);
+
+-- The claim query's access path: the oldest runnable job.
+CREATE INDEX idx_jobs_runnable ON jobs(status, run_after);
+
+CREATE INDEX idx_jobs_workspace ON jobs(workspace_id, created_at DESC);
+
+-- Uniqueness that expires on its own. The predicate keeps only outstanding
+-- work in the index, so re-enqueuing a key whose job already finished is
+-- allowed — which is what makes this a "don't queue it twice right now" guard
+-- rather than a "never do this again" one.
+CREATE UNIQUE INDEX idx_jobs_dedupe
+  ON jobs(workspace_id, dedupe_key)
+  WHERE dedupe_key IS NOT NULL AND status IN ('pending', 'running');

packages/domain/src/ids.ts

@@ -31,6 +31,7 @@ export const ID_PREFIXES = {
   suppression: 'sup',
   privacyRequest: 'pri',
   deletionJob: 'del',
+  job: 'job',
   policyRule: 'pol',
   usageEvent: 'usg',
   auditEvent: 'aud',

packages/pipeline/src/index.ts

@@ -17,3 +17,19 @@ export {
   type JobKind,
   type JobResult,
 } from './jobs';
+export {
+  claimNext,
+  completeJob,
+  drainQueue,
+  enqueue,
+  failJob,
+  queueDepth,
+  reclaimStalled,
+  JOB_STATUSES,
+  type DrainSummary,
+  type EnqueueInput,
+  type EnqueueResult,
+  type JobHandler,
+  type JobStatus,
+  type QueuedJob,
+} from './queue';