feat(write-queue): outbox drain worker

Completes the outbox pattern. drainOutboxTick scans for unsent
envelopes older than 1s and XADDs each to its stream, marking the
row sent on success. Any envelope the inline submitWrite XADD
couldn't deliver eventually lands via this loop.

Wired into apps/api alongside the main processor and the demo
processor. Runs only when PROCESSOR_ENABLED is true; moves to
apps/worker once that deploys separately.

Duplicate XADDs under contention are absorbed by the
skip-if-completed guard in processNext.

Author: Devansh-365

apps/api/src/index.ts

@@ -3,6 +3,7 @@ import { createDb } from '@sheetforge/shared-db';
 import { createLogger } from '@sheetforge/shared-logger';
 import { createIoredisQueueClient, createUpstashQueueClient } from '@sheetforge/shared-redis';
 import { createRouter } from '@sheetforge/slice-rest-api';
+import { drainOutboxTick } from '@sheetforge/slice-write-queue';
 import { demoProcessorTick } from './demo-processor.js';
 import { loadEnv } from './env.js';
 import { processorTick } from './processor.js';
@@ -89,6 +90,30 @@ if (env.PROCESSOR_ENABLED) {
       // its blockMs (or the Upstash adapter's setTimeout fallback).
     }
   })();
+
+  // Outbox drain — catches envelopes whose inline XADD from submitWrite
+  // didn't land (process death between the ledger commit and Redis, or a
+  // transient Redis outage). Without this loop those rows would sit in
+  // Postgres forever and the corresponding writes would never reach the
+  // handler.
+  (async () => {
+    log.info({}, 'outbox-drain-starting');
+    while (true) {
+      try {
+        const outcome = await drainOutboxTick({ db, redis });
+        if (outcome.failed > 0) {
+          await new Promise((res) => setTimeout(res, 1000));
+        }
+      } catch (err) {
+        log.error(
+          { err: err instanceof Error ? err.message : String(err) },
+          'outbox-drain-tick-failed',
+        );
+        await new Promise((res) => setTimeout(res, 5000));
+      }
+      await new Promise((res) => setTimeout(res, 500));
+    }
+  })();
 }
 
 process.on('SIGTERM', () => {

slices/write-queue/src/drain.ts

@@ -0,0 +1,86 @@
+import { type QueueRedisClient, enqueue } from '@sheetforge/queue';
+import { type Db, schema } from '@sheetforge/shared-db';
+import { type Logger, createLogger } from '@sheetforge/shared-logger';
+import { and, asc, isNull, lt } from 'drizzle-orm';
+import { markOutboxSent } from './repo.js';
+
+export interface DrainOutcome {
+  claimed: number;
+  sent: number;
+  failed: number;
+}
+
+/**
+ * One tick of the outbox drain loop. Scans for unsent envelopes older than
+ * `staleMs` and XADDs each to its stream. Marks the row sent on success;
+ * leaves it for the next tick on failure.
+ *
+ * The staleness threshold deliberately skips rows the inline submitWrite
+ * XADD is still trying to deliver. Without it the drain would race the
+ * inline path and produce a duplicate stream message on every write.
+ *
+ * Duplicates remain possible if two drain workers run against the same
+ * database (or if mark-sent lost a write after a successful inline XADD).
+ * processNext's skip-if-completed guard absorbs them harmlessly — the row
+ * lands once, later stream deliveries short-circuit to ack.
+ */
+export async function drainOutboxTick({
+  db,
+  redis,
+  batchSize = 50,
+  staleMs = 1000,
+  logger,
+}: {
+  db: Db;
+  redis: QueueRedisClient;
+  batchSize?: number;
+  staleMs?: number;
+  logger?: Logger;
+}): Promise<DrainOutcome> {
+  const log = logger ?? createLogger({ service: 'write-queue-drain' });
+  const cutoff = new Date(Date.now() - staleMs);
+
+  const candidates = await db
+    .select({
+      id: schema.writeOutbox.id,
+      writeId: schema.writeOutbox.writeId,
+      streamKey: schema.writeOutbox.streamKey,
+      envelope: schema.writeOutbox.envelope,
+    })
+    .from(schema.writeOutbox)
+    .where(and(isNull(schema.writeOutbox.sentAt), lt(schema.writeOutbox.createdAt, cutoff)))
+    .orderBy(asc(schema.writeOutbox.createdAt))
+    .limit(batchSize);
+
+  let sent = 0;
+  let failed = 0;
+  for (const row of candidates) {
+    try {
+      // envelope is jsonb on the column. submitWrite wrote it as an
+      // EnqueueMessage<WritePayload>; trust that shape round-trips cleanly.
+      await enqueue({
+        redis,
+        streamKey: row.streamKey,
+        message: row.envelope as { writeId: string; idempotencyKey?: string; payload: unknown },
+      });
+      await markOutboxSent({ db, id: row.id });
+      sent++;
+    } catch (err) {
+      log.warn(
+        {
+          err: err instanceof Error ? err.message : String(err),
+          outboxId: row.id,
+          writeId: row.writeId,
+        },
+        'drain-xadd-failed',
+      );
+      failed++;
+    }
+  }
+
+  if (candidates.length > 0) {
+    log.debug({ claimed: candidates.length, sent, failed }, 'drain-tick');
+  }
+
+  return { claimed: candidates.length, sent, failed };
+}

slices/write-queue/src/index.ts

@@ -5,6 +5,8 @@ export {
   getLedgerStats,
 } from './service.js';
 export type { ProcessOutcome, LedgerStats } from './service.js';
+export { drainOutboxTick } from './drain.js';
+export type { DrainOutcome } from './drain.js';
 export { WritePayloadSchema } from './types.js';
 export type {
   SubmitResult,