diff --git a/apps/api/src/broker/reminderScheduler.ts b/apps/api/src/broker/reminderScheduler.ts new file mode 100644 index 0000000..9eb531d --- /dev/null +++ b/apps/api/src/broker/reminderScheduler.ts @@ -0,0 +1,183 @@ +/** + * Reminder scheduler — shared-timed-event 15-min reminder scan. + * + * Fires every minute via node-cron. Each tick calls runReminderCheck() which: + * 1. Queries shared (isShared=true) timed (allDay=false) events whose dtstartUtc + * falls in [now+14min, now+16min] — the ±1 min window around 15 min lead. + * 2. Cross-joins with ALL push_subscriptions (shared reminder → every member). + * 3. Deduplicates on (uid, minuteBucket) so a window-boundary double-tick never + * fires twice (D-06, RESEARCH Pitfall 5 / Open Question 3). + * 4. Calls dispatchPush once per subscription per due event. + * + * Decisions enforced here: + * D-05 — isShared=true in the SQL WHERE (not in copy); personal events excluded. + * D-06 — fixed 15-min lead; no per-event customisation. + * D-07 — allDay=false in the SQL WHERE; all-day events are silently excluded. + * D-11 — every push goes through dispatchPush which always sends a visible notification. + * D-12 — in-memory Set; no Redis; single-process deployment. + * D-16 — empty shared-calendar set (Family calendar not yet created) produces zero + * sends and no crash (cross-join on empty table returns no rows). + * + * Threat mitigations: + * T-05-17 — D-05 enforced in QUERY (WHERE isShared=true), not in copy. + * T-05-18 — in-memory dedup Set keyed (uid, minuteBucket); per-event try/catch. + * T-05-19 — per-subscription try/catch; dispatchPush already swallows 410/404. + */ + +import { schedule } from 'node-cron' +import { and, eq, gte, lte, sql } from 'drizzle-orm' +import { db } from '../db/client.js' +import { calendars, calendarEvents, pushSubscriptions } from '../db/schema.js' +import { dispatchPush } from '../lib/pushDispatcher.js' +import type { NotificationPayload } from '../lib/pushDispatcher.js' + +// ── In-memory dedup (D-12: single-process, no Redis) ───────────────────────── +// Key: `${eventUid}:${minuteBucket}` where minuteBucket = floor(now ms / 60000). +// Prevents double-fire when the same event sits in the window across two consecutive ticks. +// Acceptable data loss on process restart for a two-person household. +const sentReminders = new Set() + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +/** + * Format a UTC Date as "YYYY-MM-DD" for calendar deep-link URLs. + */ +function yyyyMmDd(d: Date): string { + const y = d.getUTCFullYear() + const m = String(d.getUTCMonth() + 1).padStart(2, '0') + const day = String(d.getUTCDate()).padStart(2, '0') + return `${y}-${m}-${day}` +} + +// ── Core scan ───────────────────────────────────────────────────────────────── + +/** + * Run one reminder scan cycle. + * + * Exported with an injectable `now` parameter so tests can control time without + * needing the cron schedule to fire. Production call passes no argument; + * default = new Date() (current wall-clock time). + */ +export async function runReminderCheck(now = new Date()): Promise { + const minuteBucket = Math.floor(now.getTime() / 60000) + const windowStart = new Date(now.getTime() + 14 * 60 * 1000) + const windowEnd = new Date(now.getTime() + 16 * 60 * 1000) + + // Single query: events (isShared, timed, in window) cross-joined with ALL + // push_subscriptions. The cross-join (sql`1=1`) fans every matching event out + // to every subscriber — shared reminder, every member notified (D-05, member-agnostic). + // + // With an empty push_subscriptions table (no subscribers yet) the INNER JOIN + // on sql`1=1` returns no rows — zero sends, no crash (D-16 empty-calendar analogue). + const rows = await db + .select({ + uid: calendarEvents.uid, + title: calendarEvents.title, + dtstartUtc: calendarEvents.dtstartUtc, + allDay: calendarEvents.allDay, + isShared: calendars.isShared, + subId: pushSubscriptions.id, + subUserId: pushSubscriptions.userId, + subEndpoint: pushSubscriptions.endpoint, + subP256dh: pushSubscriptions.p256dh, + subAuth: pushSubscriptions.auth, + }) + .from(calendarEvents) + .innerJoin(calendars, eq(calendarEvents.calendarId, calendars.id)) + .innerJoin(pushSubscriptions, sql`1=1`) + .where( + and( + eq(calendars.isShared, true), // D-05: shared calendar only (in QUERY, not copy) + eq(calendarEvents.allDay, false), // D-07: timed events only + gte(calendarEvents.dtstartUtc, windowStart), + lte(calendarEvents.dtstartUtc, windowEnd), + ), + ) + + // Group flat (event, subscription) rows by event uid so we can: + // a) dedup per event (not per (event, sub) pair), and + // b) fan out to ALL subscriptions for a deduped event in one pass. + type SubRow = { id: number; userId: number; endpoint: string; p256dh: string; auth: string } + const byUid = new Map< + string, + { uid: string; title: string | null; dtstartUtc: Date; subs: SubRow[] } + >() + + for (const row of rows) { + if (!byUid.has(row.uid)) { + byUid.set(row.uid, { + uid: row.uid, + title: row.title ?? null, + // dtstartUtc is guaranteed non-null by the allDay=false + gte/lte WHERE + dtstartUtc: row.dtstartUtc as Date, + subs: [], + }) + } + // subId is undefined when cross-join produces no subscriptions row (empty table) + if (row.subId != null) { + byUid.get(row.uid)!.subs.push({ + id: row.subId, + userId: row.subUserId!, + endpoint: row.subEndpoint, + p256dh: row.subP256dh, + auth: row.subAuth, + }) + } + } + + // Dedup and dispatch + for (const [uid, event] of byUid) { + try { + const key = `${uid}:${minuteBucket}` + if (sentReminders.has(key)) continue + // Mark sent BEFORE dispatching to prevent re-entry on concurrent ticks + sentReminders.add(key) + + const dateStr = yyyyMmDd(event.dtstartUtc) + const notification: NotificationPayload = { + // Null-safe title fallback (D-02 / NOTIF-01): title column may be NULL on rows + // created before the Plan 05-07 sync update. Use the uid rather than "undefined". + title: event.title ?? uid, + body: 'Starts in 15 min', + tag: `reminder-${uid}`, + navigate: `/calendar?date=${dateStr}&event=${uid}`, + } + + // Fan out to every subscriber (member-count-agnostic) + for (const sub of event.subs) { + try { + await dispatchPush(sub, notification) + } catch (err) { + // Per-subscription error isolation (T-05-19): one bad sub never aborts the cycle. + // dispatchPush itself never throws (it resolves after logging), so this outer + // catch is a belt-and-suspenders guard for unexpected errors. + console.error( + `[broker/reminderScheduler] Error dispatching reminder to sub id=${sub.id} for event uid=${uid}:`, + err instanceof Error ? err.message : String(err), + ) + } + } + } catch (err) { + // Per-event error isolation (T-05-18): one bad event never aborts remaining events. + console.error( + `[broker/reminderScheduler] Error processing event uid=${uid}:`, + err instanceof Error ? err.message : String(err), + ) + } + } +} + +// ── Scheduler ───────────────────────────────────────────────────────────────── + +/** + * Start the 1-minute reminder scan schedule. + * Call once from index.ts's isMainModule() guard — NOT at import time + * (keeps the cron out of the test process; mirrors startBrokerPoller pattern). + */ +export function startReminderScheduler(): void { + schedule('* * * * *', () => { + runReminderCheck().catch((err: unknown) => { + console.error('[broker/reminderScheduler] Unhandled runReminderCheck error:', err) + }) + }) +} diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index c9e791d..ff3252a 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -13,6 +13,7 @@ import { oidcAuthMiddleware, processOAuthCallback } from './auth/middleware.js' import { devAuthBypass } from './auth/devBypass.js' import { startBrokerPoller } from './broker/poller.js' import { startOutboxWorker } from './broker/outboxWorker.js' +import { startReminderScheduler } from './broker/reminderScheduler.js' import webpush from 'web-push' export const app = new Hono() @@ -132,6 +133,9 @@ if (isMainModule()) { startBrokerPoller() // Drain the D-05 outbox every 15s: dispatches pending CalDAV writes to Fastmail. startOutboxWorker() + // Start the 1-min reminder scan for shared timed events starting in ~15 min (NOTIF-01). + // VAPID must be configured (above) before this starts or push sends will fail. + startReminderScheduler() serve({ fetch: app.fetch, port: 3000 }, (info) => { console.log(`FamilySync API running on http://localhost:${info.port}`)