/** * 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. * scheduleOutboxDrain is exported for unit testing; it wraps runOutboxDrain with the isDraining * guard + drainRequested trailing-re-drain loop (D-05). * startOutboxWorker wraps scheduleOutboxDrain in a 15-second setInterval. * (node-cron 4.2.1 silently skipped scheduled executions in the long-running server process; * setInterval fires reliably in the same process — replaced to fix the silent skip.) * * Source: poller.ts pattern (runPoll/startBrokerPoller) */ import { z } from 'zod'; import ICAL from 'ical.js'; 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, extractValarms, computeAlertInstantUtc, RRULE_PRESETS, } from './vevent.js'; import type { FastmailClient } from './client.js'; import { dispatchEventChange } from '../lib/eventChangeDispatcher.js'; import { getHouseholdTimezone } from '../lib/householdTimezone.js'; import { onOutboxDrain } from '../lib/outboxTrigger.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; // ── Outbox payload re-validation (IN-03) ───────────────────────────────────── /** * IN-03: re-validate the JSON payload read back out of calendar_outbox before building * a VEVENT from it. The payload was zod-validated at enqueue (routes/events.ts * eventFieldsSchema), but a manually-inserted row or enqueue→drain schema drift could * feed undefined/wrong-typed fields into buildVeventString, producing SUMMARY:undefined * or an Invalid Date. Such a row can NEVER succeed, so on validation failure the caller * hard-fails the row (no retry) instead of burning the backoff budget. * * Mirrors eventFieldsSchema in routes/events.ts. `_preservedRrule` (added by the edit-as- * move route, CR-01) is allowed via .passthrough() so the move payload still validates. */ const outboxPayloadSchema = z .object({ title: z.string().min(1).max(255), allDay: z.boolean(), start: z.string().min(1).max(64), end: z.string().min(1).max(64), location: z.string().max(2000).optional(), description: z.string().max(2000).optional(), recurrence: z.enum(['none', 'daily', 'weekly', 'monthly', 'yearly']).optional(), calendarUrl: z.string().url().max(1024).optional(), _preservedRrule: z.string().max(1024).optional(), // D-06: recurrence bounding (RRULE UNTIL / COUNT) // CR-01: defense-in-depth — validate the exact 'YYYY-MM-DD' shape here too (the route // schema validates on ingress, but the outbox payload is re-parsed from stored JSON). // Guarantees .replace(/-/g,'') in assembleRruleString emits digits-only, closing the // RRULE-part injection vector. int().min(1) prevents zero/negative counts. recurrenceUntil: z .string() .regex(/^\d{4}-\d{2}-\d{2}$/) .optional(), // 'YYYY-MM-DD' → RRULE UNTIL recurrenceCount: z.number().int().min(1).optional(), // integer ≥ 1 → RRULE COUNT // Phase 11: per-event reminder lead in minutes (CAL-13/CAL-14, D-08). // absent — field not present; preserve existing VALARM verbatim (no-change path, D-08) // null — explicit "None" → clear the VALARM on write-back // 0 — same-day all-day reminder (fire 9 AM on event date); timed 0 = None (D-06) // positive int — N minutes before event start (timed) or N/1440 days before (all-day) reminderLeadMinutes: z.number().int().min(0).max(10080).nullable().optional(), }) .passthrough(); type OutboxPayloadFields = z.infer; // ── D-06: RRULE bound assembly ─────────────────────────────────────────────── /** * Assembles a complete RRULE string by appending a COUNT or UNTIL bound to a * base preset string (e.g. 'FREQ=WEEKLY'). * * Rules (RFC 5545 §3.3.10): * - COUNT takes precedence over UNTIL when both are supplied (mutual exclusion). * - For all-day events (allDay=true), UNTIL serializes as DATE form: `YYYYMMDD`. * - For timed events (allDay=false), UNTIL serializes as DATETIME UTC: `YYYYMMDDTHHMMSSZ`. * Using T235959Z (end of UTC day) is a safe universal choice per RESEARCH.md Pitfall 2. * - When neither bound is supplied the base preset is returned unchanged. * * T-06-02 security note: `until.replace(/-/g,'')` emits only digits stripped from a * length-bounded string (max 10 via Zod). The fixed `;UNTIL=` / `;COUNT=` templates * prevent injection of extra `;`-delimited RRULE parts. The assembled string is later * passed through `ICAL.Recur.fromString` which rejects malformed RRULE values. * * @param basePreset - Base RRULE string, e.g. 'FREQ=DAILY' (from RRULE_PRESETS or extracted) * @param until - Optional 'YYYY-MM-DD' end date * @param count - Optional occurrence count (integer ≥ 1) * @param allDay - Whether the event is all-day (controls UNTIL value-type) * @returns Assembled RRULE string */ export function assembleRruleString( basePreset: string, until?: string, count?: number, allDay?: boolean, ): string { let s = basePreset; if (count !== undefined) { // COUNT wins over UNTIL (mutual exclusion) s += `;COUNT=${count}`; } else if (until) { const dateDigits = until.replace(/-/g, ''); if (allDay) { // DATE form for all-day events: YYYYMMDD (RFC 5545 §3.3.10) s += `;UNTIL=${dateDigits}`; } else { // DATETIME UTC form for timed events: YYYYMMDDTHHMMSSZ (end of UTC day) s += `;UNTIL=${dateDigits}T235959Z`; } } return s; } /** * IN-02: shared RRULE-resolution decision tree for the update and create branches. * * The two branches differ only in the SOURCE of `preservedRrule` (the update branch * re-reads it from calendarEvents.rawVevent; the create branch reads the * `_preservedRrule` payload field threaded by the edit-as-move route). The precedence * logic is identical and was previously copy-pasted, risking drift between the two * copies of the RFC-5545 bound-strip (`;(UNTIL|COUNT)=` removal) — see Pitfall 3. * * Precedence: * 1. Explicit recurrence on the payload wins (recurrence:'none' clears the RRULE). * 2. Else, if a preserved RRULE exists and the payload changes only the bound * (UNTIL/COUNT), strip the preserved RRULE's existing bound and re-apply the new * one — never naive-concatenate (would produce a double-UNTIL/COUNT). * 3. Else fall back to the payload's preset (rruleFromPayload), or the preserved * RRULE unchanged when no bound change was requested. */ function resolveFinalRrule( fields: OutboxPayloadFields, hasExplicitRecurrence: boolean, rruleFromPayload: string | undefined, preservedRrule: string | undefined, ): string | undefined { if (hasExplicitRecurrence) { // Explicit recurrence wins — recurrence:'none' yields undefined (no RRULE emitted) return rruleFromPayload ? assembleRruleString( rruleFromPayload, fields.recurrenceUntil, fields.recurrenceCount, fields.allDay, ) : undefined; } if (preservedRrule) { if (fields.recurrenceUntil || fields.recurrenceCount !== undefined) { // Bound change only: strip existing UNTIL/COUNT, then re-apply the new bound (Pitfall 3) const strippedPreset = preservedRrule.replace(/;(UNTIL|COUNT)=[^;]*/g, ''); return assembleRruleString( strippedPreset, fields.recurrenceUntil, fields.recurrenceCount, fields.allDay, ); } return preservedRrule; } return rruleFromPayload; } // ── 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; /** * D-05 / D-06: trailing-re-drain flag. * Set to true when signalOutboxDrain() fires while a drain is already in flight. * Collapses any number of mid-drain signals into exactly one trailing drain — never * rate-limited, never dropped, never more than one extra pass (D-06). * Reset to false BEFORE the recursive scheduleOutboxDrain() call (Pitfall 3 — resetting * after would allow an unbounded re-drain chain against Fastmail rate limits / T-09-01). */ let drainRequested = false; /** * Schedule an outbox drain pass. * * If no drain is currently running, kicks off runOutboxDrain() immediately. * If a drain IS running (isDraining=true), records drainRequested=true so the * currently-running drain triggers exactly one trailing re-drain on completion (D-05). * * Errors from runOutboxDrain are caught and logged (D-02 / T-09-03). * * Called by: * - initOutboxTrigger's onOutboxDrain listener (event-driven path, CAL-15) * - startOutboxWorker's 15-second setInterval (polling fallback, D-08) */ export function scheduleOutboxDrain(): void { if (isDraining) { drainRequested = true; return; } runOutboxDrain() .catch((err: unknown) => { console.error('[outboxWorker] Unhandled runOutboxDrain error:', err); }) .finally(() => { if (drainRequested) { // Reset BEFORE recursive call (Pitfall 3) — prevents unbounded re-drain chain drainRequested = false; scheduleOutboxDrain(); } }); } /** * WR-06: max time to wait on the post-write targeted re-sync before marking the * outbox row 'done'. A stalled Fastmail connection cannot wedge the single-process * drain loop beyond this cap; the PWA's next sync-status poll reconciles any cache * that the timed-out re-sync did not refresh. */ const RESYNC_TIMEOUT_MS = 10_000; // ── 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. */ export 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. */ export 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; } // NOTIF-03: pass onChanges so this-member writes push to the other member. // actor = userId (the member who wrote via the outbox — D-03). await syncCalendar(client, davCal, userId, (changes) => { for (const change of changes) { dispatchEventChange(change, userId).catch((err: unknown) => { console.error( '[outboxWorker] dispatchEventChange error:', err instanceof Error ? err.message : String(err), ); }); } }); } 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; } /** * Lazily resolves the household timezone at most once, caching the promise. * Threaded through a drain cycle (mirroring clientCache, IN-01) so an all-day * create row and an all-day update row in the same cycle share a single * app_config read instead of issuing two identical SELECTs (IN-03). The read is * still lazy: cycles with no all-day work never touch the DB. */ type TimezoneResolver = () => Promise; function makeTimezoneResolver(): TimezoneResolver { let cached: Promise | undefined; return () => { if (cached === undefined) { // D-05: route through the single stored-TZ accessor (no inline fallback duplicated here). // D-06: getHouseholdTimezone falls back to process.env.TZ → Intl when no row is stored. cached = getHouseholdTimezone(db); } return cached; }; } async function dispatchRow( row: OutboxRow, resolveTimezone: TimezoneResolver, ): 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 rawFields: Record; try { rawFields = JSON.parse(row.payload) as Record; } catch { return { success: false, conflict: false, hardFail: true, transient: false, error: 'payload parse failed', }; } // IN-03: re-validate the parsed payload. A schema-invalid row can never succeed — // hard-fail it (no retry) rather than feeding undefined/Invalid Date into the VEVENT. const parsedFields = outboxPayloadSchema.safeParse(rawFields); if (!parsedFields.success) { return { success: false, conflict: false, hardFail: true, transient: false, error: `payload validation failed: ${parsedFields.error.message}`, }; } const fields: OutboxPayloadFields = parsedFields.data; // 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'); // CAL-14: mirrors the WR-01 hasExplicitRecurrence pattern for VALARM preservation. // When the payload omits `reminderLeadMinutes` entirely (no-change, D-08), we preserve // the existing VALARM verbatim from rawVevent via extractValarms. An explicit null clears // the VALARM; an explicit value (timed or all-day) replaces it. const hasExplicitReminder = Object.prototype.hasOwnProperty.call(fields, 'reminderLeadMinutes'); 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); } // CAL-14: resolve VALARM for the UPDATE branch. // absent (no hasExplicitReminder) + rawVevent has VALARMs → preserve verbatim (D-08) // explicit null → clear (no VALARM emitted by buildVeventString) // explicit value + allDay + valid start → compute 9 AM absolute DATE-TIME trigger (D-04) // explicit value + timed → pass through to buildTimedValarm let valarmsToPreserve: ICAL.Component[] | undefined; let allDayAlertInstantUtcUpdate: Date | undefined; if (!hasExplicitReminder && freshEtagRows.length > 0 && freshEtagRows[0].rawVevent) { valarmsToPreserve = extractValarms(freshEtagRows[0].rawVevent); } else if ( hasExplicitReminder && fields.reminderLeadMinutes != null && fields.allDay && fields.start ) { // IN-03: shared per-cycle resolver — one app_config read across all-day rows. const tz = await resolveTimezone(); const leadDays = fields.reminderLeadMinutes / 1440; allDayAlertInstantUtcUpdate = computeAlertInstantUtc(fields.start, leadDays, tz); } // D-06: assemble the final RRULE string, combining the preset or preserved RRULE // with an optional UNTIL/COUNT bound from the payload. // Pitfall 3: on series edit with bound change only (no new preset), parse the preserved // RRULE, STRIP any existing UNTIL/COUNT, then re-apply the new bound — never naive- // concatenate onto `FREQ=WEEKLY;BYDAY=...` which would produce double-UNTIL. // WR-01 note: preservedRrule is only set when !hasExplicitRecurrence (see above), // so the hasExplicitRecurrence branch always takes precedence over preserved RRULE. // IN-02: shared decision tree extracted to resolveFinalRrule (mirrored in create branch). const finalRruleString = resolveFinalRrule( fields, hasExplicitRecurrence, rruleFromPayload, preservedRrule, ); const { icsString } = buildVeventString({ uid: row.uid, summary: fields.title, allDay: fields.allDay, dtstart: fields.allDay ? fields.start : new Date(fields.start), dtend: fields.allDay ? fields.end : new Date(fields.end), location: fields.location, description: fields.description, rruleString: finalRruleString, // CAL-13/CAL-14: VALARM wiring — absent preserves, null clears, value replaces reminderLeadMinutes: hasExplicitReminder ? fields.reminderLeadMinutes : undefined, valarms: valarmsToPreserve, allDayAlertInstantUtc: allDayAlertInstantUtcUpdate, }); 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 rawFields: Record; try { rawFields = JSON.parse(row.payload) as Record; } catch { return { success: false, conflict: false, hardFail: true, transient: false, error: 'payload parse failed', }; } // IN-03: re-validate the parsed payload — hard-fail a schema-invalid create row. const parsedFields = outboxPayloadSchema.safeParse(rawFields); if (!parsedFields.success) { return { success: false, conflict: false, hardFail: true, transient: false, error: `payload validation failed: ${parsedFields.error.message}`, }; } const fields: OutboxPayloadFields = parsedFields.data; // 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; // D-06: assemble the final RRULE string with optional UNTIL/COUNT bound. // CR-01: an explicit recurrence preset wins over _preservedRrule (deliberate user choice). // recurrence:'none' explicitly clears any RRULE — including when _preservedRrule is present. // If no explicit recurrence, fall back to _preservedRrule (edit-as-move RRULE carry-through). // IN-02: shared decision tree extracted to resolveFinalRrule (mirrored in update branch). const finalRruleString = resolveFinalRrule( fields, hasExplicitRecurrence, rruleFromPayload, preservedRrule, ); // CAL-13: VALARM wiring for CREATE branch (no rawVevent source — new event always // carries an explicit picker value or no reminder at all; no preserve path needed). let allDayAlertInstantUtcCreate: Date | undefined; if (fields.reminderLeadMinutes != null && fields.allDay && fields.start) { // IN-03: shared per-cycle resolver — one app_config read across all-day rows. const tz = await resolveTimezone(); const leadDays = fields.reminderLeadMinutes / 1440; allDayAlertInstantUtcCreate = computeAlertInstantUtc(fields.start, leadDays, tz); } const { icsString } = buildVeventString({ uid: row.uid, summary: fields.title, allDay: fields.allDay, dtstart: fields.allDay ? fields.start : new Date(fields.start), dtend: fields.allDay ? fields.end : new Date(fields.end), location: fields.location, description: fields.description, rruleString: finalRruleString, // CAL-13: per-event reminder — pass through; null=clear, value=set, absent=no VALARM reminderLeadMinutes: fields.reminderLeadMinutes, allDayAlertInstantUtc: allDayAlertInstantUtcCreate, }); // 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(); // IN-03: per-drain-cycle timezone resolver so multiple all-day rows in the same cycle // share one app_config read. Lazy: cycles with no all-day work never hit the DB. const resolveTimezone = makeTimezoneResolver(); 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, resolveTimezone); 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. // // WR-06: bound the re-sync with a timeout. triggerTargetedResync does // unbounded network I/O against Fastmail (fetchCalendars + syncCalendar); // a hang would leave this row 'pending' from the DB's view for the full // duration, the 15s isDraining guard would no-op the next cycle, and the // PWA would poll 'pending' indefinitely — wedging the single-process // drain loop. On timeout we proceed to mark 'done' and let the PWA's next // poll/refetch reconcile (the same documented refetch path the eager // re-sync was optimizing). triggerTargetedResync already swallows its own // errors, so the race only needs to cap the wait. await Promise.race([ triggerTargetedResync(row.calendarUrl, row.userId, clientCache), new Promise((resolve) => setTimeout(resolve, RESYNC_TIMEOUT_MS)), ]); 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 ──────────────────────────────────────────────────────────────── /** * Register the in-process EventEmitter drain signal listener (CAL-15 / D-01). * Call once at API startup (in index.ts, after startOutboxWorker). * After registration, any signalOutboxDrain() call (fired post-enqueue) immediately * routes through scheduleOutboxDrain — eliminating the up-to-15s polling delay. * * T-09-02: initOutboxTrigger is called only under isMainModule() in index.ts; * tests import runOutboxDrain/scheduleOutboxDrain directly and never register * a listener — no open-handle leak preventing process exit. * * WR-02: idempotent. The unsubscribe handle is retained so a duplicate call is a * no-op (never double-registers the 'drain' listener) and stopOutboxTrigger() can * remove it for a clean teardown (tests, graceful shutdown). */ let unsubscribeDrain: (() => void) | null = null; export function initOutboxTrigger(): void { if (unsubscribeDrain) return; // idempotent — never double-register unsubscribeDrain = onOutboxDrain(() => scheduleOutboxDrain()); } /** * Remove the 'drain' listener registered by initOutboxTrigger() and reset the * idempotency guard so a later initOutboxTrigger() can re-register cleanly. * Used by the test suite's afterAll teardown (WR-03) and available for graceful * shutdown. */ export function stopOutboxTrigger(): void { unsubscribeDrain?.(); unsubscribeDrain = null; } /** * IN-03: test-only reset for the module-level concurrency flags. * isDraining/drainRequested are the entire concurrency contract for the * single-process deployment and are NOT touched by vi.resetAllMocks(). Tests call * this in beforeEach so each test starts from a known-quiescent state instead of * relying on the previous test having drained cleanly. Not for production use. */ export function __resetDrainState(): void { if (process.env.NODE_ENV === 'production') return; // test-only; never clear the guard in prod isDraining = false; drainRequested = false; } /** * Starts the 15-second background outbox drain schedule (polling fallback, D-08). * Call once at API startup (wired in index.ts beside startBrokerPoller). * Uses setInterval instead of node-cron: node-cron 4.2.1 silently skipped executions * in the long-running server process; setInterval fires reliably. * The interval body calls scheduleOutboxDrain() so errors are absorbed by its .catch * and mid-drain signals collapse correctly via drainRequested (D-05). */ export function startOutboxWorker(): void { setInterval(() => { scheduleOutboxDrain(); }, 15 * 1000); }