feat(05-06): implement reminderScheduler — shared timed 15-min reminder scan

- Create apps/api/src/broker/reminderScheduler.ts:
  - runReminderCheck(now): queries isShared=true + allDay=false events
    with dtstartUtc in [now+14min, now+16min] via calendarEvents→calendars→
    pushSubscriptions cross-join (2 innerJoins; fans out to all subscribers)
  - In-memory sentReminders Set keyed uid:minuteBucket prevents double-fire
    at window boundary (D-06, T-05-18)
  - Per-event and per-subscription try/catch for error isolation (T-05-19)
  - Null title fallback (event.title ?? uid) — handles rows before Plan 05-07
  - Empty shared-calendar set produces zero sends and no crash (D-16)
  - startReminderScheduler(): node-cron 1-min schedule wrapping runReminderCheck
- Wire startReminderScheduler() into index.ts isMainModule() guard after
  startOutboxWorker() and VAPID setVapidDetails (NOTIF-01)
- Tests: 3/3 GREEN (all-day excluded, non-shared excluded, dedup)

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<string>()
+
+// ── 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<void> {
+  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)
+    })
+  })
+}

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}`)