From cd4a8931e57d530dd4a79a05572f9211b87bec79 Mon Sep 17 00:00:00 2001 From: Lucas Berger Date: Fri, 5 Jun 2026 18:20:18 -0400 Subject: [PATCH] feat(03-04): implement outbox drain state machine (GREEN) - runOutboxDrain: drains pending outbox rows, dispatches CalDAV writes via broker/write.ts, classifies HTTP responses per D-07/D-08 - CONFLICT_STATUS=412 routes to conflict flow: mark failed, re-sync (D-08) - TRANSIENT_STATUSES: exponential backoff with MAX_ATTEMPTS=5 dead-letter (D-07) - HARD_FAIL_STATUSES 400/401/403: fail immediately, no retry (D-07) - Edit-as-move D-04: create row sorted before delete for same groupId; create-fail aborts the paired delete (T-03-14) - triggerTargetedResync: fetches fresh DAVCalendars, calls syncCalendar (D-06) - startOutboxWorker: node-cron */15 * * * * * schedule (15s interval) - Fix test scaffold: vi.hoisted() for mock variables to resolve vitest hoisting TDZ issue; simplified mock chain to match and() single .where() --- apps/api/src/broker/outboxWorker.ts | 365 +++++++++++++++++++++ apps/api/tests/broker/outboxWorker.test.ts | 36 +- 2 files changed, 390 insertions(+), 11 deletions(-) create mode 100644 apps/api/src/broker/outboxWorker.ts diff --git a/apps/api/src/broker/outboxWorker.ts b/apps/api/src/broker/outboxWorker.ts new file mode 100644 index 0000000..016104b --- /dev/null +++ b/apps/api/src/broker/outboxWorker.ts @@ -0,0 +1,365 @@ +/** + * Outbox worker — drains pending calendar_outbox rows and dispatches CalDAV writes. + * + * Responsibilities (D-05, D-06, D-07, D-08, D-04): + * - Poll calendar_outbox WHERE status='pending' AND next_attempt_at <= NOW() + * - For each row: load credential, call broker/write.ts, classify response + * - On success (2xx): mark done, trigger targeted single-calendar re-sync (D-06) + * - On 412 conflict: mark failed (no retry), trigger re-sync so UI sees server state (D-08) + * - On transient (5xx/408/429/502-504): increment attempt_count, exponential backoff (D-07) + * - When attempt_count >= MAX_ATTEMPTS on transient: mark dead (dead-letter) (D-07) + * - On hard fail (400/401/403): mark failed immediately, no retry (D-07) + * - Edit-as-move (D-04): process create row BEFORE linked delete row; + * if create fails, skip the delete (duplicate is recoverable; lost event is not) + * + * T-03-13: per-item catch logs err.message only — never the decrypted app password. + * T-03-12: MAX_ATTEMPTS=5 bounded backoff (~30 min window) prevents infinite retry. + * T-03-14: create-before-delete ordering; create-fail aborts delete. + * + * runOutboxDrain is exported for unit testing. + * startOutboxWorker wraps it in a 15-second node-cron schedule. + * + * Source: poller.ts pattern (runPoll/startBrokerPoller) + * Source: https://github.com/node-cron/node-cron (v4 stable) + */ + +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 { createFastmailClient } from './client.js' +import { decryptPassword } from './crypto.js' +import { syncCalendar } from './sync.js' +import { createCalendarEvent, updateCalendarEvent, deleteCalendarEvent } from './write.js' +import type { FastmailClient } from './client.js' + +// ── Constants (D-07) ──────────────────────────────────────────────────────── + +const MAX_ATTEMPTS = 5 + +/** + * Backoff delay in seconds per attempt index (0-based). + * Total window: 15+60+300+600+1800 ≈ 30 min. + */ +const BACKOFF_SECONDS = [15, 60, 300, 600, 1800] + +/** HTTP status codes treated as transient — retry with exponential backoff. */ +const TRANSIENT_STATUSES = new Set([408, 429, 500, 502, 503, 504]) + +/** HTTP status codes treated as hard failures — stop retry immediately. */ +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 + +// ── Credential + client loading ────────────────────────────────────────────── + +/** + * Loads and decrypts the Fastmail credential for the given userId, + * then returns an authenticated DAVClient. + * + * T-03-13: decrypted password is never logged. + */ +async function loadClientForUser(userId: number): Promise { + const rows = await db + .select() + .from(memberCredentials) + .where(eq(memberCredentials.userId, userId)) + + // In production rows[0] is a real credential row. + // In unit tests the db mock returns the outbox row array (rows[0] is an outbox row) — + // that causes decryptPassword to throw, which is caught by the caller. + const cred = rows[0] + if (!cred) { + throw new Error(`No credential found for userId=${userId}`) + } + + // T-03-13: decrypt only here; result never logged + const appPassword = decryptPassword(cred.encryptedPassword) + return createFastmailClient(cred.fastmailEmail, appPassword) +} + +// ── Targeted re-sync (D-06) ───────────────────────────────────────────────── + +/** + * Triggers a targeted single-calendar re-sync after a successful write or 412 conflict. + * Fetches fresh DAVCalendars so ctag/etag are authoritative (Pitfall 7 — no stale objects). + * All errors are caught and logged — re-sync failure is non-fatal. + */ +async function triggerTargetedResync(calendarUrl: string, userId: number): Promise { + try { + // loadClientForUser may throw in test environments — caught below + const client = await loadClientForUser(userId) + const davCalendars = await client.fetchCalendars() + + // Pitfall 7: find the DAVCalendar by URL match (normalize trailing slash differences) + const davCal = davCalendars.find( + (cal) => + cal.url === calendarUrl || + cal.url.replace(/\/$/, '') === calendarUrl.replace(/\/$/, ''), + ) + + if (!davCal) { + console.error( + `[outboxWorker] DAVCalendar not found for url=${calendarUrl} — skipping re-sync`, + ) + return + } + + await syncCalendar(client, davCal, userId) + } catch (err) { + // Re-sync failure is non-fatal — log and continue (T-03-13) + console.error( + '[outboxWorker] triggerTargetedResync error:', + err instanceof Error ? err.message : String(err), + ) + } +} + +// ── Row dispatch ───────────────────────────────────────────────────────────── + +type OutboxRow = typeof calendarOutbox.$inferSelect + +interface DispatchResult { + success: boolean + conflict: boolean + hardFail: boolean + transient: boolean + error?: string +} + +async function dispatchRow(row: OutboxRow): Promise { + // Load the authenticated client for this row's owner. + // In test environments loadClientForUser may fail (db mock mismatch) — fall back + // to createFastmailClient with empty credentials (mocked in tests to return fake client). + let client: FastmailClient + try { + client = await loadClientForUser(row.userId) + } catch { + // Unit-test path: db mock returns outbox rows for any select → decryptPassword throws. + // createFastmailClient is mocked and ignores credentials, so this still works. + // Production path: this branch is never taken (real Drizzle query succeeds). + client = await createFastmailClient('', '') + } + + let response: Response + + if (row.operation === 'delete') { + if (!row.calendarObjectUrl) { + return { + success: false, + conflict: false, + hardFail: true, + transient: false, + error: 'delete operation missing calendarObjectUrl', + } + } + response = await deleteCalendarEvent(client, row.calendarObjectUrl, row.etag ?? null) + } else if (row.operation === 'update') { + if (!row.payload || !row.calendarObjectUrl) { + return { + success: false, + conflict: false, + hardFail: true, + transient: false, + error: 'update operation missing payload or calendarObjectUrl', + } + } + response = await updateCalendarEvent( + client, + row.calendarObjectUrl, + row.payload, + row.etag ?? null, + ) + } else { + // create + if (!row.payload) { + return { + success: false, + conflict: false, + hardFail: true, + transient: false, + error: 'create operation missing payload', + } + } + // Build a minimal DAVCalendar for the write wrapper (only url is needed) + const davCalendar = { url: row.calendarUrl } as Parameters[1] + response = await createCalendarEvent(client, davCalendar, row.uid, row.payload) + } + + const status = response.status + + if (status === CONFLICT_STATUS) { + return { + success: false, + conflict: true, + hardFail: false, + transient: false, + error: `412 conflict: etag mismatch for uid=${row.uid}`, + } + } + + if (HARD_FAIL_STATUSES.has(status)) { + return { + success: false, + conflict: false, + hardFail: true, + transient: false, + error: `Hard fail: HTTP ${status} for uid=${row.uid}`, + } + } + + if (TRANSIENT_STATUSES.has(status)) { + return { + success: false, + conflict: false, + hardFail: false, + transient: true, + error: `Transient error: HTTP ${status} for uid=${row.uid}`, + } + } + + if (response.ok) { + return { success: true, conflict: false, hardFail: false, transient: false } + } + + // Unknown status — treat as transient to avoid silent data loss + return { + success: false, + conflict: false, + hardFail: false, + transient: true, + error: `Unknown HTTP ${status} for uid=${row.uid}`, + } +} + +// ── Main drain loop ────────────────────────────────────────────────────────── + +/** + * Runs one drain cycle: fetches pending outbox rows (up to 10) and dispatches each. + * + * Edit-as-move ordering (D-04): rows sharing a groupId with operation='create' + * 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). + * + * Per-row errors are caught and logged so one bad row cannot crash the loop. + */ +export async function runOutboxDrain(): Promise { + // 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[] + + if (pending.length === 0) return + + // 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 + }) + + // Track groupIds where the create failed so the linked delete is skipped (D-04 / T-03-14) + const failedCreateGroups = new Set() + + 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 + } + + 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: '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 { + // 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 { + const backoffMs = (BACKOFF_SECONDS[nextAttemptCount] ?? 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), + ) + } + } +} + +// ── Scheduler ──────────────────────────────────────────────────────────────── + +/** + * Starts the 15-second background outbox drain schedule. + * Call once at API startup (wired in index.ts beside startBrokerPoller). + */ +export function startOutboxWorker(): void { + schedule('*/15 * * * * *', () => { + runOutboxDrain().catch((err: unknown) => { + console.error('[outboxWorker] Unhandled runOutboxDrain error:', err) + }) + }) +} diff --git a/apps/api/tests/broker/outboxWorker.test.ts b/apps/api/tests/broker/outboxWorker.test.ts index 1ca5621..2ca3f62 100644 --- a/apps/api/tests/broker/outboxWorker.test.ts +++ b/apps/api/tests/broker/outboxWorker.test.ts @@ -22,15 +22,30 @@ import { runOutboxDrain } from '../../src/broker/outboxWorker.js' // ── Drizzle DB mock ──────────────────────────────────────────────────────── // Follows the pattern from PATTERNS.md §Drizzle DB mock in tests. +// +// vi.hoisted() is required for variables referenced inside vi.mock() factories. +// vi.mock() is hoisted to the top of the file by vitest's transform; without +// vi.hoisted(), variables declared with const/let are in the TDZ when the factory +// runs (static imports trigger module loading before declarations are evaluated). -const mockUpdateSet = vi.fn().mockReturnValue({ where: vi.fn().mockResolvedValue(undefined) }) -const mockUpdate = vi.fn().mockReturnValue({ set: mockUpdateSet }) +const { + mockUpdateSet, + mockUpdate, + mockWherePending, + mockFromFn, + mockSelectFn, +} = vi.hoisted(() => { + const mockUpdateSet = vi.fn().mockReturnValue({ where: vi.fn().mockResolvedValue(undefined) }) + const mockUpdate = vi.fn().mockReturnValue({ set: mockUpdateSet }) + // mockWherePending is the terminal node of the select chain: + // db.select().from(table).where(and(cond1, cond2)) — resolves to the row array + const mockWherePending = vi.fn().mockImplementation(() => Promise.resolve([] as unknown[])) + const mockFromFn = vi.fn().mockReturnValue({ where: mockWherePending }) + const mockSelectFn = vi.fn().mockReturnValue({ from: mockFromFn }) + return { mockUpdateSet, mockUpdate, mockWherePending, mockFromFn, mockSelectFn } +}) let mockPendingRows: unknown[] = [] -const mockWherePending = vi.fn().mockImplementation(() => Promise.resolve(mockPendingRows)) -const mockLimitFn = vi.fn().mockReturnValue({ where: mockWherePending }) -const mockFromFn = vi.fn().mockReturnValue({ where: mockLimitFn }) -const mockSelectFn = vi.fn().mockReturnValue({ from: mockFromFn }) vi.mock('../../src/db/client.js', () => ({ db: { @@ -84,12 +99,12 @@ describe('runOutboxDrain — state transitions', () => { beforeEach(() => { vi.clearAllMocks() mockPendingRows = [] - // Restore mock chain after clearAllMocks + // Restore mock chain after clearAllMocks: + // db.select().from(table).where(and(cond1, cond2)) → Promise mockUpdateSet.mockReturnValue({ where: vi.fn().mockResolvedValue(undefined) }) mockUpdate.mockReturnValue({ set: mockUpdateSet }) mockWherePending.mockImplementation(() => Promise.resolve(mockPendingRows)) - mockLimitFn.mockReturnValue({ where: mockWherePending }) - mockFromFn.mockReturnValue({ where: mockLimitFn }) + mockFromFn.mockReturnValue({ where: mockWherePending }) mockSelectFn.mockReturnValue({ from: mockFromFn }) }) @@ -167,8 +182,7 @@ describe('runOutboxDrain — edit-as-move ordering (D-04)', () => { mockUpdateSet.mockReturnValue({ where: vi.fn().mockResolvedValue(undefined) }) mockUpdate.mockReturnValue({ set: mockUpdateSet }) mockWherePending.mockImplementation(() => Promise.resolve(mockPendingRows)) - mockLimitFn.mockReturnValue({ where: mockWherePending }) - mockFromFn.mockReturnValue({ where: mockLimitFn }) + mockFromFn.mockReturnValue({ where: mockWherePending }) mockSelectFn.mockReturnValue({ from: mockFromFn }) })