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)
This commit is contained in:
Lucas Berger
2026-06-09 21:38:32 -04:00
parent 9b04528fd6
commit b95f671485
2 changed files with 187 additions and 0 deletions
+183
View File
@@ -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)
})
})
}
+4
View File
@@ -13,6 +13,7 @@ import { oidcAuthMiddleware, processOAuthCallback } from './auth/middleware.js'
import { devAuthBypass } from './auth/devBypass.js' import { devAuthBypass } from './auth/devBypass.js'
import { startBrokerPoller } from './broker/poller.js' import { startBrokerPoller } from './broker/poller.js'
import { startOutboxWorker } from './broker/outboxWorker.js' import { startOutboxWorker } from './broker/outboxWorker.js'
import { startReminderScheduler } from './broker/reminderScheduler.js'
import webpush from 'web-push' import webpush from 'web-push'
export const app = new Hono() export const app = new Hono()
@@ -132,6 +133,9 @@ if (isMainModule()) {
startBrokerPoller() startBrokerPoller()
// Drain the D-05 outbox every 15s: dispatches pending CalDAV writes to Fastmail. // Drain the D-05 outbox every 15s: dispatches pending CalDAV writes to Fastmail.
startOutboxWorker() 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) => { serve({ fetch: app.fetch, port: 3000 }, (info) => {
console.log(`FamilySync API running on http://localhost:${info.port}`) console.log(`FamilySync API running on http://localhost:${info.port}`)