Lucas Berger
4f42b7535b
- Add reminderLeadMinutes to eventFieldsSchema (ingress validation, T-11-06) - Add reminderLeadMinutes to outboxPayloadSchema (drain re-validation, IN-03 defense-in-depth) - Import extractValarms + computeAlertInstantUtc from vevent.ts - UPDATE branch: hasExplicitReminder gate mirrors hasExplicitRecurrence (WR-01 pattern) - absent field → extractValarms(rawVevent) preserved verbatim (CAL-14, D-08) - explicit null → clear VALARM (no valarmsToPreserve, null passed to buildVeventString) - explicit value + allDay → computeAlertInstantUtc at 9 AM local (D-04) - explicit value + timed → passed through to buildTimedValarm via buildVeventString - CREATE branch: always explicit picker value; compute allDayAlertInstantUtc when allDay
956 lines
42 KiB
TypeScript
956 lines
42 KiB
TypeScript
/**
|
|
* 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 { 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).nullable().optional(),
|
|
})
|
|
.passthrough();
|
|
|
|
type OutboxPayloadFields = z.infer<typeof outboxPayloadSchema>;
|
|
|
|
// ── 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<FastmailClient> {
|
|
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<number, FastmailClient>,
|
|
): Promise<void> {
|
|
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;
|
|
}
|
|
|
|
async function dispatchRow(row: OutboxRow): Promise<DispatchResult> {
|
|
// 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<string, unknown>;
|
|
try {
|
|
rawFields = JSON.parse(row.payload) as Record<string, unknown>;
|
|
} 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
|
|
) {
|
|
const tz = process.env.TZ ?? Intl.DateTimeFormat().resolvedOptions().timeZone;
|
|
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<string, unknown>;
|
|
try {
|
|
rawFields = JSON.parse(row.payload) as Record<string, unknown>;
|
|
} 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) {
|
|
const tz = process.env.TZ ?? Intl.DateTimeFormat().resolvedOptions().timeZone;
|
|
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<typeof createCalendarEvent>[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<void> {
|
|
// 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<string>();
|
|
|
|
// 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<number, FastmailClient>();
|
|
|
|
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.
|
|
//
|
|
// 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<void>((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);
|
|
}
|