apps/api/src/broker/outboxWorker.ts

@@ -26,7 +26,7 @@
 import { schedule } from 'node-cron'
 import { and, eq, lte } from 'drizzle-orm'
 import { db } from '../db/client.js'
-import { calendarOutbox, memberCredentials } from '../db/schema.js'
+import { calendarEvents, calendarOutbox, memberCredentials } from '../db/schema.js'
 import { createFastmailClient } from './client.js'
 import { decryptPassword } from './crypto.js'
 import { syncCalendar } from './sync.js'
@@ -53,6 +53,24 @@ const HARD_FAIL_STATUSES = new Set([400, 401, 403])
 /** HTTP status code for CalDAV If-Match conflict — D-08 conflict flow. */
 const CONFLICT_STATUS = 412
 
+// ── Drain concurrency guard (CR-05) ──────────────────────────────────────────
+
+/**
+ * Module-level drain guard — prevents overlapping 15s drain cycles from
+ * double-dispatching the same still-pending outbox row.
+ *
+ * SINGLE-PROCESS LIMITATION: This guard is valid ONLY for the single-process
+ * Unraid deployment of this two-user app where all drain cycles share the same
+ * Node.js module instance. A multi-process or multi-replica deployment (e.g.
+ * running multiple API containers behind a load balancer) would require a
+ * durable DB row-claim instead:
+ *   UPDATE calendar_outbox SET status='processing'
+ *   WHERE id=? AND status='pending'
+ * and only the process that wins the affected-rows check would dispatch the row.
+ * Do not remove this comment if deploying to multi-process infrastructure.
+ */
+let isDraining = false
+
 // ── Credential + client loading ──────────────────────────────────────────────
 
 /**
@@ -269,113 +287,169 @@ async function dispatchRow(row: OutboxRow): Promise<DispatchResult> {
  * are sorted before operation='delete'. If the create fails, the linked delete is
  * skipped (duplicate is recoverable; lost event is not — D-04 / T-03-14).
  *
+ * Concurrency guard (CR-05): the module-level `isDraining` flag ensures overlapping
+ * 15s scheduler invocations are no-ops for the single-process deployment.
+ *
+ * Durable create-before-delete gate (CR-04): for delete rows with a groupId, the
+ * worker queries the DB for the sibling create row's status. It does NOT rely on
+ * both rows co-occurring in the same in-memory batch.
+ *
  * Per-row errors are caught and logged so one bad row cannot crash the loop.
  */
 export async function runOutboxDrain(): Promise<void> {
-  // Fetch pending rows: WHERE status='pending' AND next_attempt_at <= NOW()
-  const pending = (await db
-    .select()
-    .from(calendarOutbox)
-    .where(
-      and(
-        eq(calendarOutbox.status, 'pending'),
-        lte(calendarOutbox.nextAttemptAt, new Date()),
-      ),
-    )) as OutboxRow[]
+  // CR-05: single-process concurrency guard (see isDraining declaration for limitations)
+  if (isDraining) return
+  isDraining = true
 
-  if (pending.length === 0) return
+  try {
+    // Fetch pending rows: WHERE status='pending' AND next_attempt_at <= NOW()
+    const pending = (await db
+      .select()
+      .from(calendarOutbox)
+      .where(
+        and(
+          eq(calendarOutbox.status, 'pending'),
+          lte(calendarOutbox.nextAttemptAt, new Date()),
+        ),
+      )) as OutboxRow[]
 
-  // D-04 edit-as-move ordering: sort so create rows come before delete rows within the same groupId.
-  // Rows without a groupId are unaffected (stable relative order preserved).
-  const sorted = [...pending].sort((a, b) => {
-    if (a.groupId && b.groupId && a.groupId === b.groupId) {
-      if (a.operation === 'create' && b.operation === 'delete') return -1
-      if (a.operation === 'delete' && b.operation === 'create') return 1
-    }
-    return 0
-  })
+    if (pending.length === 0) return
 
-  // Track groupIds where the create failed so the linked delete is skipped (D-04 / T-03-14)
-  const failedCreateGroups = new Set<string>()
+    // D-04 edit-as-move ordering: sort so create rows come before delete rows within the same groupId.
+    // Rows without a groupId are unaffected (stable relative order preserved).
+    // This is a fast path; the authoritative gate is the durable DB sibling-status check below.
+    const sorted = [...pending].sort((a, b) => {
+      if (a.groupId && b.groupId && a.groupId === b.groupId) {
+        if (a.operation === 'create' && b.operation === 'delete') return -1
+        if (a.operation === 'delete' && b.operation === 'create') return 1
+      }
+      return 0
+    })
 
-  for (const row of sorted) {
-    // D-04: if the create for this group failed, skip the paired delete
-    if (row.operation === 'delete' && row.groupId && failedCreateGroups.has(row.groupId)) {
-      console.warn(
-        `[outboxWorker] Skipping delete row.id=${row.id} — create for groupId=${row.groupId} failed (D-04)`,
-      )
-      continue
-    }
+    // Track groupIds where the create failed within this batch (fast path for same-batch pairs).
+    // Cross-batch ordering is enforced durably by the DB sibling-status check inside the loop.
+    const failedCreateGroups = new Set<string>()
 
-    try {
-      const result = await dispatchRow(row)
+    for (const row of sorted) {
+      // D-04 fast path: if the create for this group already failed in this batch, skip the delete
+      if (row.operation === 'delete' && row.groupId && failedCreateGroups.has(row.groupId)) {
+        console.warn(
+          `[outboxWorker] Skipping delete row.id=${row.id} — create for groupId=${row.groupId} failed this batch (D-04)`,
+        )
+        continue
+      }
 
-      if (result.conflict) {
-        // 412 — mark failed (no retry), re-sync calendar so UI sees authoritative state (D-08)
-        await db
-          .update(calendarOutbox)
-          .set({ status: 'failed', lastError: result.error ?? '412 conflict' })
-          .where(eq(calendarOutbox.id, row.id))
-        await triggerTargetedResync(row.calendarUrl, row.userId)
+      // CR-04: Durable create-before-delete gate — query DB for sibling create status.
+      // This prevents the delete from running when the create pair straddles drain batches.
+      if (row.operation === 'delete' && row.groupId) {
+        const siblingRows = (await db
+          .select({ status: calendarOutbox.status })
+          .from(calendarOutbox)
+          .where(
+            and(
+              eq(calendarOutbox.groupId, row.groupId),
+              eq(calendarOutbox.operation, 'create'),
+            ),
+          )) as Array<{ status: string }>
 
-        if (row.groupId && row.operation === 'create') {
-          failedCreateGroups.add(row.groupId)
+        const siblingStatus = siblingRows[0]?.status
+
+        if (siblingStatus !== 'done') {
+          if (siblingStatus === 'failed' || siblingStatus === 'dead') {
+            // Sibling create failed permanently — skip this delete forever (D-04: original preserved)
+            console.warn(
+              `[outboxWorker] Paired create for groupId=${row.groupId} is ${siblingStatus} — marking delete row.id=${row.id} failed (original event preserved, D-04)`,
+            )
+            await db
+              .update(calendarOutbox)
+              .set({
+                status: 'failed',
+                lastError: 'paired create did not succeed — original preserved',
+              })
+              .where(eq(calendarOutbox.id, row.id))
+          } else {
+            // Sibling create is still pending/processing — defer this delete to a later cycle
+            console.warn(
+              `[outboxWorker] Deferring delete row.id=${row.id} — sibling create (groupId=${row.groupId}) is not yet done (status=${siblingStatus ?? 'not found'})`,
+            )
+            // Leave the delete row pending; do NOT update its status
+          }
+          continue
         }
-      } else if (result.success) {
-        // Success — mark done, trigger targeted re-sync to refresh the cache (D-06)
-        await db
-          .update(calendarOutbox)
-          .set({ status: 'done' })
-          .where(eq(calendarOutbox.id, row.id))
-        await triggerTargetedResync(row.calendarUrl, row.userId)
-      } else if (result.hardFail) {
-        // Hard fail — mark failed immediately, no retry (D-07)
-        await db
-          .update(calendarOutbox)
-          .set({ status: 'failed', lastError: result.error ?? 'Hard fail' })
-          .where(eq(calendarOutbox.id, row.id))
+      }
 
-        if (row.groupId && row.operation === 'create') {
-          failedCreateGroups.add(row.groupId)
-        }
-      } else {
-        // Transient — exponential backoff or dead-letter (D-07 / T-03-12)
-        const nextAttemptCount = row.attemptCount + 1
-        if (nextAttemptCount >= MAX_ATTEMPTS) {
-          // Dead-letter: max attempts reached (T-03-12)
+      try {
+        const result = await dispatchRow(row)
+
+        if (result.conflict) {
+          // 412 — mark failed (no retry), re-sync calendar so UI sees authoritative state (D-08)
           await db
             .update(calendarOutbox)
-            .set({
-              status: 'dead',
-              attemptCount: nextAttemptCount,
-              lastError: result.error ?? 'Max attempts exceeded',
-            })
+            .set({ status: 'failed', lastError: result.error ?? '412 conflict' })
+            .where(eq(calendarOutbox.id, row.id))
+          await triggerTargetedResync(row.calendarUrl, row.userId)
+
+          if (row.groupId && row.operation === 'create') {
+            failedCreateGroups.add(row.groupId)
+          }
+        } else if (result.success) {
+          // Success — mark done, trigger targeted re-sync to refresh the cache (D-06)
+          await db
+            .update(calendarOutbox)
+            .set({ status: 'done' })
+            .where(eq(calendarOutbox.id, row.id))
+          await triggerTargetedResync(row.calendarUrl, row.userId)
+        } else if (result.hardFail) {
+          // Hard fail — mark failed immediately, no retry (D-07)
+          await db
+            .update(calendarOutbox)
+            .set({ status: 'failed', lastError: result.error ?? 'Hard fail' })
             .where(eq(calendarOutbox.id, row.id))
 
           if (row.groupId && row.operation === 'create') {
             failedCreateGroups.add(row.groupId)
           }
         } else {
-          // WR-01: use row.attemptCount (the attempt that just failed, 0-based) as the backoff index.
-          // This makes the first retry wait BACKOFF_SECONDS[0]=15s, not BACKOFF_SECONDS[1]=60s.
-          const backoffMs = (BACKOFF_SECONDS[row.attemptCount] ?? 1800) * 1000
-          await db
-            .update(calendarOutbox)
-            .set({
-              attemptCount: nextAttemptCount,
-              nextAttemptAt: new Date(Date.now() + backoffMs),
-              lastError: result.error,
-            })
-            .where(eq(calendarOutbox.id, row.id))
+          // Transient — exponential backoff or dead-letter (D-07 / T-03-12)
+          const nextAttemptCount = row.attemptCount + 1
+          if (nextAttemptCount >= MAX_ATTEMPTS) {
+            // Dead-letter: max attempts reached (T-03-12)
+            await db
+              .update(calendarOutbox)
+              .set({
+                status: 'dead',
+                attemptCount: nextAttemptCount,
+                lastError: result.error ?? 'Max attempts exceeded',
+              })
+              .where(eq(calendarOutbox.id, row.id))
+
+            if (row.groupId && row.operation === 'create') {
+              failedCreateGroups.add(row.groupId)
+            }
+          } else {
+            // WR-01: use row.attemptCount (the attempt that just failed, 0-based) as the backoff index.
+            // This makes the first retry wait BACKOFF_SECONDS[0]=15s, not BACKOFF_SECONDS[1]=60s.
+            const backoffMs = (BACKOFF_SECONDS[row.attemptCount] ?? 1800) * 1000
+            await db
+              .update(calendarOutbox)
+              .set({
+                attemptCount: nextAttemptCount,
+                nextAttemptAt: new Date(Date.now() + backoffMs),
+                lastError: result.error,
+              })
+              .where(eq(calendarOutbox.id, row.id))
+          }
         }
+      } catch (err) {
+        // Per-row error isolation: log but never crash the loop (T-03-13)
+        console.error(
+          `[outboxWorker] Error dispatching row.id=${row.id} uid=${row.uid}:`,
+          err instanceof Error ? err.message : String(err),
+        )
       }
-    } catch (err) {
-      // Per-row error isolation: log but never crash the loop (T-03-13)
-      console.error(
-        `[outboxWorker] Error dispatching row.id=${row.id} uid=${row.uid}:`,
-        err instanceof Error ? err.message : String(err),
-      )
     }
+  } finally {
+    isDraining = false
   }
 }
 

apps/api/tests/broker/outboxWorker.test.ts

@@ -126,16 +126,12 @@ function wireMockChain() {
   mockUpdate.mockReturnValue({ set: mockUpdateSet })
   // mockFromFn differentiates by table argument:
   // - memberCredentials table → returns FAKE_CRED_ROW (so loadClientForUser succeeds by default)
-  // - anything else → returns mockPendingRows (outbox query)
+  // - anything else (calendarOutbox, calendarEvents) → returns mockWherePending
+  // Use Symbol.for('drizzle:Name') to identify the table — JSON.stringify throws on circular Drizzle
+  // table structures so it cannot be used for table identification.
   mockFromFn.mockImplementation((table: unknown) => {
-    // Drizzle table objects have a Symbol.for('drizzle:Name') property and a [Table.Symbol.Name].
-    // The safest approach: JSON.stringify often includes the table config name.
-    let isCred = false
-    try {
-      isCred = JSON.stringify(table).includes('member_credentials')
-    } catch {
-      // table not serializable — not a credential table
-    }
+    const tableName = (table as Record<symbol, string>)[Symbol.for('drizzle:Name')] ?? ''
+    const isCred = tableName === 'member_credentials'
     return {
       where: isCred
         ? vi.fn().mockResolvedValue([FAKE_CRED_ROW])
@@ -307,6 +303,16 @@ describe('runOutboxDrain — edit-as-move ordering (D-04)', () => {
     // Both rows in the pending list
     mockPendingRows = [deleteRow, createRow]
 
+    // The durable sibling-status check (CR-04) runs for the delete row with groupId.
+    // It queries calendarOutbox for the sibling create's status. By the time the delete
+    // is processed (create was sorted and dispatched first), we simulate the sibling as 'done'.
+    // The base mockWherePending returns mockPendingRows for all calendarOutbox selects; we
+    // override just the sibling-status call with mockImplementationOnce queued after the
+    // pending-rows select call.
+    mockWherePending
+      .mockImplementationOnce(() => Promise.resolve([deleteRow, createRow])) // pending-rows select
+      .mockImplementationOnce(() => Promise.resolve([{ status: 'done' }]))   // sibling-status select
+
     await runOutboxDrain()
 
     // CREATE must be called before DELETE