/** * 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 { calendarEvents, calendarOutbox, calendars, 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 { buildVeventString, extractRruleString, RRULE_PRESETS } from './vevent.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 // ── 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 ────────────────────────────────────────────── /** * 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. * * IN-01: accepts an optional per-drain-cycle client cache. Without it, every settled or * conflicted row independently reloaded + AES-GCM-decrypted the member credential, * widening the window the decrypted app password lives in memory (T-03-13). When a cache * is supplied, the decrypted client is built at most once per userId per drain cycle. */ async function triggerTargetedResync( calendarUrl: string, userId: number, clientCache?: Map, ): Promise { try { // loadClientForUser may throw in test environments — caught below let client = clientCache?.get(userId) if (!client) { client = await loadClientForUser(userId) clientCache?.set(userId, client) } 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 { // CR-03: fail closed on credential errors — let loadClientForUser throw. // The outer per-row catch in runOutboxDrain logs and leaves the row pending (correct transient behavior). // Do NOT add an empty-credential fallback — that would silently PUT with no authentication. const client = await loadClientForUser(row.userId) 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', } } // CR-02: parse the stored form JSON and build a real VCALENDAR string let fields: Record try { fields = JSON.parse(row.payload) as Record } catch { return { success: false, conflict: false, hardFail: true, transient: false, error: 'payload parse failed' } } // WR-01: recurrence preservation. The PWA omits `recurrence` from an edit payload // (it cannot read the existing RRULE — not in the occurrence contract, D-03), so on // update we must NOT rebuild the VEVENT with no RRULE — that would silently convert a // recurring series into a single event. When the payload carries no explicit // recurrence, fall back to the RRULE already stored in calendarEvents.rawVevent. // An explicit recurrence value (including 'none') still overrides — that is a // deliberate user change. Read rawVevent in the same scoped query as the fresh etag. let preservedRrule: string | undefined const hasExplicitRecurrence = Object.prototype.hasOwnProperty.call(fields, 'recurrence') const rruleFromPayload = fields.recurrence && fields.recurrence !== 'none' ? RRULE_PRESETS[fields.recurrence as string] : undefined // WR-02: re-read the freshest etag from calendarEvents just before PUT. // Rapid successive edits to the same uid enqueue multiple update rows, each // carrying the etag at enqueue time. If a prior edit succeeded and triggered // a re-sync, calendarEvents.etag was updated but the next update row still // carries the old enqueue-time etag — guaranteed 412 on the second edit. // Using the freshest cached etag here prevents the spurious conflict toast // while still preserving genuine conflict detection (D-08): a real external // change updates calendarEvents.etag differently from any pending row's etag. // CR-02: calendar_events is keyed (calendarId, uid), and a shared Fastmail // account (D-16) caches the same uid once per member's calendar. A uid-only // re-read returns multiple rows and an arbitrary [0] — potentially the OTHER // member's etag, which would spuriously 412 (false conflict → edit dropped, // D-08) or coincidentally match and overwrite. Scope the re-read to THIS row's // own calendar by joining through calendars on the outbox row's userId + // calendarUrl so the freshest etag belongs to the writing member. let etagForPut: string | null = row.etag ?? null const freshEtagRows = (await db .select({ etag: calendarEvents.etag, rawVevent: calendarEvents.rawVevent }) .from(calendarEvents) .innerJoin(calendars, eq(calendarEvents.calendarId, calendars.id)) .where( and( eq(calendarEvents.uid, row.uid), eq(calendars.userId, row.userId), eq(calendars.url, row.calendarUrl), ), ) .limit(1)) as Array<{ etag: string | null; rawVevent: string | null }> if (freshEtagRows.length > 0 && freshEtagRows[0].etag != null) { etagForPut = freshEtagRows[0].etag } // WR-01: when the edit payload carries no explicit recurrence, preserve the RRULE // already on the stored event so an edit does not strip a recurring series. if (!hasExplicitRecurrence && freshEtagRows.length > 0 && freshEtagRows[0].rawVevent) { preservedRrule = extractRruleString(freshEtagRows[0].rawVevent) } const { icsString } = buildVeventString({ uid: row.uid, summary: fields.title as string, allDay: fields.allDay as boolean, dtstart: fields.allDay ? (fields.start as string) : new Date(fields.start as string), dtend: fields.allDay ? (fields.end as string) : new Date(fields.end as string), location: fields.location as string | undefined, description: fields.description as string | undefined, rruleString: hasExplicitRecurrence ? rruleFromPayload : (preservedRrule ?? rruleFromPayload), }) response = await updateCalendarEvent( client, row.calendarObjectUrl, icsString, etagForPut, ) } else { // create if (!row.payload) { return { success: false, conflict: false, hardFail: true, transient: false, error: 'create operation missing payload', } } // CR-02: parse the stored form JSON and build a real VCALENDAR string let fields: Record try { fields = JSON.parse(row.payload) as Record } catch { return { success: false, conflict: false, hardFail: true, transient: false, error: 'payload parse failed' } } // CR-01: edit-as-move RRULE preservation. The same-calendar `update` branch // preserves a recurring series' RRULE by reading rawVevent; the `create` branch // (used for the create half of an edit-as-move, D-04) has no source for the // original RRULE because it writes under a brand-new uid. The edit route extracts // the source event's RRULE and stashes it on the payload as `_preservedRrule` so // the worker can re-apply it here. An explicit `recurrence` on the payload still // wins (deliberate user change); the preserved RRULE only fills the gap when the // edit omitted recurrence — matching the update-branch semantics and the WR-01 fix. const hasExplicitRecurrence = Object.prototype.hasOwnProperty.call(fields, 'recurrence') const rruleFromPayload = fields.recurrence && fields.recurrence !== 'none' ? RRULE_PRESETS[fields.recurrence as string] : undefined const preservedRrule = typeof fields._preservedRrule === 'string' && fields._preservedRrule.length > 0 ? fields._preservedRrule : undefined const { icsString } = buildVeventString({ uid: row.uid, summary: fields.title as string, allDay: fields.allDay as boolean, dtstart: fields.allDay ? (fields.start as string) : new Date(fields.start as string), dtend: fields.allDay ? (fields.end as string) : new Date(fields.end as string), location: fields.location as string | undefined, description: fields.description as string | undefined, rruleString: hasExplicitRecurrence ? rruleFromPayload : (preservedRrule ?? rruleFromPayload), }) // 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, icsString) } 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 } } // IN-02: an unmapped 4xx (e.g. 405, 409, 422) is a permanent client error — retrying it // for the full backoff window just delays settling and burns the attempt budget before // dead-lettering. The transient-eligible 4xx codes (408 request timeout, 429 too many // requests) are already in TRANSIENT_STATUSES and handled above, so any remaining 4xx // here is a hard fail. 5xx, network failures, and truly unknown statuses still fall // through to transient so genuinely recoverable conditions keep their retries. if (status >= 400 && status < 500) { return { success: false, conflict: false, hardFail: true, transient: false, error: `Hard fail: HTTP ${status} for uid=${row.uid}`, } } // Unknown / 5xx 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). * * 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 { // CR-05: single-process concurrency guard (see isDraining declaration for limitations) if (isDraining) return isDraining = true 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[] 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). // 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 }) // 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() // IN-01: per-drain-cycle client cache so triggerTargetedResync decrypts each member's // credential at most once per cycle. Discarded when the drain returns — never persisted. const clientCache = new Map() 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 } // 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 }> 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 } } try { const result = await dispatchRow(row) if (result.conflict) { // WR-06: distinguish an edit-as-move create-412 from a same-calendar conflict. // For a move (D-04) the create runs first; on 412 the paired delete is later // marked failed and the original event survives — so this is NOT a "the event // changed elsewhere" conflict, it is "the move could not be applied". The PWA // set lastSyncedUid to the NEW (move) uid, whose only outbox row is this failed // create, so without a distinct message the user sees the wrong conflict copy // and has no cue to retry. Emit a move-specific lastError that does NOT contain // '412' so the toast routes it to the dedicated move-failed copy instead of the // generic etag-conflict copy. const isMoveCreate = !!row.groupId && row.operation === 'create' const conflictError = isMoveCreate ? 'move-failed: the event could not be moved — re-open it and save again' : (result.error ?? '412 conflict') // 412 — mark failed (no retry), re-sync calendar so UI sees authoritative state (D-08) await db .update(calendarOutbox) .set({ status: 'failed', lastError: conflictError }) .where(eq(calendarOutbox.id, row.id)) await triggerTargetedResync(row.calendarUrl, row.userId, clientCache) if (row.groupId && row.operation === 'create') { failedCreateGroups.add(row.groupId) } } else if (result.success) { // Success — refresh the local cache BEFORE marking done. The PWA's // SyncStateToast polls sync-status and invalidates ['events'] the // instant it sees status='done'; if we marked done first, that refetch // raced the re-sync and returned stale cache (deleted event still // present, edit not yet applied) — forcing a manual refresh. Re-syncing // first means 'done' guarantees the cache already reflects the write. await triggerTargetedResync(row.calendarUrl, row.userId, clientCache) await db .update(calendarOutbox) .set({ status: 'done' }) .where(eq(calendarOutbox.id, row.id)) } 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 { // 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), ) } } } finally { isDraining = false } } // ── 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) }) }) }