luckberg/familysync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(260610-hbu-01): catch-up window + per-uid dedup in reminderScheduler

Browse Source 
- Replace [now+14min, now+16min] window with (now, now+16min] catch-up
- Replace minuteBucket-keyed Set with uid-keyed Map for exactly-once dedup
- Lead-accurate body: 'Starts in N min' (Math.max(1, round(lead/60000)))
- CR-01 pruning: drop entries whose dtstart <= now (event started)
- WR-01 preserved: mark uid sent after all dispatches complete
- Drop gte import; add gt import from drizzle-orm
This commit is contained in:
Lucas Berger
2026-06-10 12:32:43 -04:00
parent ec38dea1dc
commit 3fdb242f7e
1 changed files with 45 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/broker/reminderScheduler.ts
+45 -31
View File
@@ -3,39 +3,48 @@
*
* 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.
* falls in (now, now+16min] — strictly after now (future only; excludes
* already-started events) and at most 16 min out. Because the lower bound
* is `now` and not `now+14min`, a missed/late cron tick is recovered on the
* next scan: as long as the event hasn't started, it re-appears in the window.
* 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.
* 3. Deduplicates on uid ALONE so a recovered reminder fires EXACTLY ONCE
* no matter how many consecutive ticks the event spends in the window.
* CAVEAT: if an event's dtstart is rescheduled earlier after a reminder
* already fired, it will not re-fire in v1 (acceptable for household use).
* 4. Calls dispatchPush once per subscription per deduped 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-06 — fixed 16-min catch-up window; 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-12 — in-memory Map; no Redis; single-process deployment.
* Key: bare event uid. Value: event's dtstart in ms (for pruning).
* Acceptable data loss on process restart for a two-person household.
* 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-18 — in-memory dedup Map keyed by uid; 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 { and, eq, gt, 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.
// Key: event uid (bare string — no minuteBucket suffix).
// Value: event's dtstart in ms — used by CR-01 pruning to drop started events.
// Prevents double-fire when the same event sits in the catch-up window across
// multiple consecutive ticks (cross-tick exactly-once guarantee).
// Acceptable data loss on process restart for a two-person household.
const sentReminders = new Set<string>()
const sentReminders = new Map<string, number>()
// ── Helpers ───────────────────────────────────────────────────────────────────
@@ -59,14 +68,16 @@ function yyyyMmDd(d: Date): string {
* 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
// Single query: events (isShared, timed, in catch-up 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).
//
// Window: (now, now+16min] — strictly greater-than now excludes already-started events;
// lte now+16min is the upper bound. Catching up: if the ideal 15-min tick was missed,
// the event is still > now on the next tick and will be found.
//
// 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
@@ -89,7 +100,7 @@ export async function runReminderCheck(now = new Date()): Promise<void> {
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),
gt(calendarEvents.dtstartUtc, now), // strictly future (excludes already-started)
lte(calendarEvents.dtstartUtc, windowEnd),
),
)
@@ -108,7 +119,7 @@ export async function runReminderCheck(now = new Date()): Promise<void> {
byUid.set(row.uid, {
uid: row.uid,
title: row.title ?? null,
// dtstartUtc is guaranteed non-null by the allDay=false + gte/lte WHERE
// dtstartUtc is guaranteed non-null by the allDay=false + gt/lte WHERE
dtstartUtc: row.dtstartUtc as Date,
subs: [],
})
@@ -128,15 +139,21 @@ export async function runReminderCheck(now = new Date()): Promise<void> {
// Dedup and dispatch
for (const [uid, event] of byUid) {
try {
const key = `${uid}:${minuteBucket}`
if (sentReminders.has(key)) continue
// Per-uid exactly-once dedup (D-12): keyed on bare uid, no minuteBucket.
// Same event fires at most once regardless of how many ticks it sits in window.
// CAVEAT: rescheduling dtstart earlier after a reminder fired will not re-fire (v1).
if (sentReminders.has(uid)) continue
const dateStr = yyyyMmDd(event.dtstartUtc)
// Lead-accurate body: compute actual minutes to start, guarded to minimum 1.
const minutes = Math.max(1, Math.round((event.dtstartUtc.getTime() - now.getTime()) / 60000))
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',
body: `Starts in ${minutes} min`,
tag: `reminder-${uid}`,
navigate: `/calendar?date=${dateStr}&event=${uid}`,
}
@@ -157,9 +174,9 @@ export async function runReminderCheck(now = new Date()): Promise<void> {
}
// WR-01: mark sent AFTER all dispatches have been attempted. Pre-marking before
// dispatch prevents retry in the same bucket when dispatchPush throws — at-least-once
// delivery requires not pre-marking the key.
sentReminders.add(key)
// dispatch prevents retry when dispatchPush throws — at-least-once delivery
// requires not pre-marking the uid.
sentReminders.set(uid, event.dtstartUtc.getTime())
} catch (err) {
// Per-event error isolation (T-05-18): one bad event never aborts remaining events.
console.error(
@@ -169,15 +186,12 @@ export async function runReminderCheck(now = new Date()): Promise<void> {
}
}
// CR-01: Prune stale entries from sentReminders to prevent unbounded growth.
// Entries for the previous minute bucket and older are no longer needed — the
// dedup window is (uid, minuteBucket), and the current minute has now been
// processed. Keep only the current bucket; discard everything older.
const staleBucket = minuteBucket - 1
for (const key of sentReminders) {
const colonIdx = key.lastIndexOf(':')
if (colonIdx !== -1 && Number(key.slice(colonIdx + 1)) < staleBucket) {
sentReminders.delete(key)
// CR-01: Prune started events from sentReminders to prevent unbounded growth.
// Any entry whose stored dtstart is <= now means the event has already started;
// it can be removed safely (it will never re-enter the (now, now+16min] window).
for (const [uid, dtstartMs] of sentReminders) {
if (dtstartMs <= now.getTime()) {
sentReminders.delete(uid)
}
}
}