Drain document lifecycle events through a durable processor

Before this commit `createDocumentWithLifecycle` enqueued ledger rows
inside the save transaction but consumed them from an in-memory
`pendingSideEffects` array straight after commit. A crash between the
commit and the drain stranded rows as `pending` forever; a crash mid
handler left them as `running`; and there was no retry budget.

The new `DocumentLifecycleProcessor` is the single drainer. It:

  - claims work via `claimPendingBatch`, an atomic `UPDATE … RETURNING`
    that flips rows from `pending` to `running` while incrementing
    `attempts`, so two drainers cannot race onto the same row;
  - dispatches each row through a handler registered by `hookType`,
    parsing the frozen `payload_json` back into the call arguments;
  - records `succeeded` / `failed` / `retryable` outcomes through the
    ledger, so a transient failure stays on the queue until the attempt
    cap expires, after which it is marked `failed` for operators;
  - reclaims rows stuck in `running` past `staleRunningMs` back to
    `pending` on every sweep tick plus at startup, so a crashed process
    never leaves a row orphaned;
  - runs a periodic sweep so every document — including those whose
    immediate post-commit drain failed — is eventually advanced.

`createDocumentWithLifecycle` now only enqueues and then calls
`lifecycleProcessor.drainForDocument` for the immediate happy-path pass.
Handlers live in `document-lifecycle-handlers.ts` so the processor stays
free of service-graph knowledge. The processor is constructed in
`bootstrap-application`, started there, and stopped on `before-quit`.

src/main/database/document-lifecycle-events.ts

@@ -30,6 +30,15 @@ export interface EnqueueInput {
   payloadJson?: string;
 }
 
+export interface ClaimBatchInput {
+  /** Upper bound on rows claimed in one pass. */
+  limit: number;
+  /** Rows whose `attempts` have already reached this cap stay `pending` and are not claimed here; a dead-letter sweep surfaces them separately. */
+  maxAttempts: number;
+  /** When provided, only claim rows for the given document so the post-commit immediate drain stays scoped. */
+  documentId?: string;
+}
+
 export function buildDocumentLifecycleEventMethods(
   getDb: () => DatabaseSync,
   readAll: ReadAll,
@@ -54,11 +63,51 @@ export function buildDocumentLifecycleEventMethods(
       return id;
     },
 
-    markRunning(id: string): void {
+    /**
+     * Atomically claim up to `limit` pending events, flipping each to
+     * `running` and incrementing `attempts` in a single statement.
+     * Returns the claimed rows so the processor can dispatch them
+     * without racing another worker against the same ids. Rows that
+     * have already reached `maxAttempts` stay `pending` here — the
+     * operator sees them via `listExhausted` and either retries
+     * manually or purges.
+     */
+    claimPendingBatch(input: ClaimBatchInput): DocumentLifecycleEventRow[] {
       const ts = nowIso();
-      getDb().prepare(
-        "UPDATE document_lifecycle_events SET status = 'running', attempts = attempts + 1, updated_at = ? WHERE id = ?",
-      ).run(ts, id);
+      const documentScope = input.documentId ? "AND document_id = ?" : "";
+      const selectSql = `
+        SELECT id FROM document_lifecycle_events
+        WHERE status = 'pending' AND attempts < ? ${documentScope}
+        ORDER BY created_at ASC
+        LIMIT ?
+      `;
+      const updateSql = `
+        UPDATE document_lifecycle_events
+        SET status = 'running', attempts = attempts + 1, updated_at = ?
+        WHERE id IN (${selectSql})
+        RETURNING id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at
+      `;
+      const params: SQLInputValue[] = [ts, input.maxAttempts];
+      if (input.documentId) params.push(input.documentId);
+      params.push(input.limit);
+      const raw = readAll<LedgerRawRow>(updateSql, ...params);
+      return raw.map(mapLedgerRow);
+    },
+
+    /**
+     * Rescue rows that were claimed by a prior process but never
+     * reached `succeeded` or `failed` (typically because the process
+     * crashed mid-hook). Rows whose `updated_at` is older than
+     * `cutoffIso` move back to `pending` so the next `claimPendingBatch`
+     * picks them up; `attempts` is NOT rewound because the earlier
+     * claim already counted against the cap.
+     */
+    reclaimStaleRunning(cutoffIso: string): number {
+      const ts = nowIso();
+      const result = getDb().prepare(
+        "UPDATE document_lifecycle_events SET status = 'pending', updated_at = ? WHERE status = 'running' AND updated_at < ?",
+      ).run(ts, cutoffIso);
+      return Number(result.changes ?? 0);
     },
 
     markSucceeded(id: string): void {
@@ -76,10 +125,17 @@ export function buildDocumentLifecycleEventMethods(
     },
 
     /**
-     * Claim the batch of pending events for a document so the caller can
-     * drain them post-commit. Ordered by `created_at` so the fan-out
-     * reproduces the enqueue order.
+     * Return to `pending` after a handler failure that the caller wants
+     * the processor to retry (e.g. transient network failure). The
+     * attempt counter is preserved so the cap still bounds retries.
      */
+    markRetryable(id: string, error: string): void {
+      const ts = nowIso();
+      getDb().prepare(
+        "UPDATE document_lifecycle_events SET status = 'pending', last_error = ?, updated_at = ? WHERE id = ?",
+      ).run(error.slice(0, 2000), ts, id);
+    },
+
     listPendingForDocument(documentId: string): DocumentLifecycleEventRow[] {
       return readRows(readAll,
         "SELECT id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at FROM document_lifecycle_events WHERE document_id = ? AND status = 'pending' ORDER BY created_at ASC",
@@ -94,28 +150,38 @@ export function buildDocumentLifecycleEventMethods(
         limit,
       );
     },
+
+    /**
+     * Rows that have reached the attempt cap without succeeding. Surfaced
+     * to operators so a failing hook is visible instead of disappearing
+     * into the `pending` queue once the cap locks it out.
+     */
+    listExhausted(maxAttempts: number, limit = 200): DocumentLifecycleEventRow[] {
+      return readRows(readAll,
+        "SELECT id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at FROM document_lifecycle_events WHERE status = 'pending' AND attempts >= ? ORDER BY updated_at DESC LIMIT ?",
+        maxAttempts,
+        limit,
+      );
+    },
   };
 }
 
-function readRows(
-  readAll: ReadAll,
-  sql: string,
-  ...params: SQLInputValue[]
-): DocumentLifecycleEventRow[] {
-  const raw = readAll<{
-    id: string;
-    company_id: string;
-    document_id: string;
-    hook_type: string;
-    status: "pending" | "running" | "succeeded" | "failed";
-    attempts: number;
-    last_error: string | null;
-    payload_json: string;
-    created_at: string;
-    updated_at: string;
-    completed_at: string | null;
-  }>(sql, ...params);
-  return raw.map((row) => ({
+interface LedgerRawRow {
+  id: string;
+  company_id: string;
+  document_id: string;
+  hook_type: string;
+  status: "pending" | "running" | "succeeded" | "failed";
+  attempts: number;
+  last_error: string | null;
+  payload_json: string;
+  created_at: string;
+  updated_at: string;
+  completed_at: string | null;
+}
+
+function mapLedgerRow(row: LedgerRawRow): DocumentLifecycleEventRow {
+  return {
     id: row.id,
     companyId: row.company_id,
     documentId: row.document_id,
@@ -127,5 +193,13 @@ function readRows(
     createdAt: row.created_at,
     updatedAt: row.updated_at,
     completedAt: row.completed_at,
-  }));
+  };
+}
+
+function readRows(
+  readAll: ReadAll,
+  sql: string,
+  ...params: SQLInputValue[]
+): DocumentLifecycleEventRow[] {
+  return readAll<LedgerRawRow>(sql, ...params).map(mapLedgerRow);
 }