fix(260610-hbu-01): catch-up window + per-uid dedup in reminderScheduler
- 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:
@@ -3,39 +3,48 @@
|
|||||||
*
|
*
|
||||||
* Fires every minute via node-cron. Each tick calls runReminderCheck() which:
|
* Fires every minute via node-cron. Each tick calls runReminderCheck() which:
|
||||||
* 1. Queries shared (isShared=true) timed (allDay=false) events whose dtstartUtc
|
* 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).
|
* 2. Cross-joins with ALL push_subscriptions (shared reminder → every member).
|
||||||
* 3. Deduplicates on (uid, minuteBucket) so a window-boundary double-tick never
|
* 3. Deduplicates on uid ALONE so a recovered reminder fires EXACTLY ONCE
|
||||||
* fires twice (D-06, RESEARCH Pitfall 5 / Open Question 3).
|
* no matter how many consecutive ticks the event spends in the window.
|
||||||
* 4. Calls dispatchPush once per subscription per due event.
|
* 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:
|
* Decisions enforced here:
|
||||||
* D-05 — isShared=true in the SQL WHERE (not in copy); personal events excluded.
|
* 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-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-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
|
* 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).
|
* sends and no crash (cross-join on empty table returns no rows).
|
||||||
*
|
*
|
||||||
* Threat mitigations:
|
* Threat mitigations:
|
||||||
* T-05-17 — D-05 enforced in QUERY (WHERE isShared=true), not in copy.
|
* 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.
|
* T-05-19 — per-subscription try/catch; dispatchPush already swallows 410/404.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
import { schedule } from 'node-cron'
|
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 { db } from '../db/client.js'
|
||||||
import { calendars, calendarEvents, pushSubscriptions } from '../db/schema.js'
|
import { calendars, calendarEvents, pushSubscriptions } from '../db/schema.js'
|
||||||
import { dispatchPush } from '../lib/pushDispatcher.js'
|
import { dispatchPush } from '../lib/pushDispatcher.js'
|
||||||
import type { NotificationPayload } from '../lib/pushDispatcher.js'
|
import type { NotificationPayload } from '../lib/pushDispatcher.js'
|
||||||
|
|
||||||
// ── In-memory dedup (D-12: single-process, no Redis) ─────────────────────────
|
// ── In-memory dedup (D-12: single-process, no Redis) ─────────────────────────
|
||||||
// Key: `${eventUid}:${minuteBucket}` where minuteBucket = floor(now ms / 60000).
|
// Key: event uid (bare string — no minuteBucket suffix).
|
||||||
// Prevents double-fire when the same event sits in the window across two consecutive ticks.
|
// 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.
|
// Acceptable data loss on process restart for a two-person household.
|
||||||
const sentReminders = new Set<string>()
|
const sentReminders = new Map<string, number>()
|
||||||
|
|
||||||
// ── Helpers ───────────────────────────────────────────────────────────────────
|
// ── Helpers ───────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
@@ -59,14 +68,16 @@ function yyyyMmDd(d: Date): string {
|
|||||||
* default = new Date() (current wall-clock time).
|
* default = new Date() (current wall-clock time).
|
||||||
*/
|
*/
|
||||||
export async function runReminderCheck(now = new Date()): Promise<void> {
|
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)
|
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
|
// 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).
|
// 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
|
// 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).
|
// on sql`1=1` returns no rows — zero sends, no crash (D-16 empty-calendar analogue).
|
||||||
const rows = await db
|
const rows = await db
|
||||||
@@ -89,7 +100,7 @@ export async function runReminderCheck(now = new Date()): Promise<void> {
|
|||||||
and(
|
and(
|
||||||
eq(calendars.isShared, true), // D-05: shared calendar only (in QUERY, not copy)
|
eq(calendars.isShared, true), // D-05: shared calendar only (in QUERY, not copy)
|
||||||
eq(calendarEvents.allDay, false), // D-07: timed events only
|
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),
|
lte(calendarEvents.dtstartUtc, windowEnd),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
@@ -108,7 +119,7 @@ export async function runReminderCheck(now = new Date()): Promise<void> {
|
|||||||
byUid.set(row.uid, {
|
byUid.set(row.uid, {
|
||||||
uid: row.uid,
|
uid: row.uid,
|
||||||
title: row.title ?? null,
|
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,
|
dtstartUtc: row.dtstartUtc as Date,
|
||||||
subs: [],
|
subs: [],
|
||||||
})
|
})
|
||||||
@@ -128,15 +139,21 @@ export async function runReminderCheck(now = new Date()): Promise<void> {
|
|||||||
// Dedup and dispatch
|
// Dedup and dispatch
|
||||||
for (const [uid, event] of byUid) {
|
for (const [uid, event] of byUid) {
|
||||||
try {
|
try {
|
||||||
const key = `${uid}:${minuteBucket}`
|
// Per-uid exactly-once dedup (D-12): keyed on bare uid, no minuteBucket.
|
||||||
if (sentReminders.has(key)) continue
|
// 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)
|
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 = {
|
const notification: NotificationPayload = {
|
||||||
// Null-safe title fallback (D-02 / NOTIF-01): title column may be NULL on rows
|
// 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".
|
// created before the Plan 05-07 sync update. Use the uid rather than "undefined".
|
||||||
title: event.title ?? uid,
|
title: event.title ?? uid,
|
||||||
body: 'Starts in 15 min',
|
body: `Starts in ${minutes} min`,
|
||||||
tag: `reminder-${uid}`,
|
tag: `reminder-${uid}`,
|
||||||
navigate: `/calendar?date=${dateStr}&event=${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
|
// 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
|
// dispatch prevents retry when dispatchPush throws — at-least-once delivery
|
||||||
// delivery requires not pre-marking the key.
|
// requires not pre-marking the uid.
|
||||||
sentReminders.add(key)
|
sentReminders.set(uid, event.dtstartUtc.getTime())
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
// Per-event error isolation (T-05-18): one bad event never aborts remaining events.
|
// Per-event error isolation (T-05-18): one bad event never aborts remaining events.
|
||||||
console.error(
|
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.
|
// CR-01: Prune started events from sentReminders to prevent unbounded growth.
|
||||||
// Entries for the previous minute bucket and older are no longer needed — the
|
// Any entry whose stored dtstart is <= now means the event has already started;
|
||||||
// dedup window is (uid, minuteBucket), and the current minute has now been
|
// it can be removed safely (it will never re-enter the (now, now+16min] window).
|
||||||
// processed. Keep only the current bucket; discard everything older.
|
for (const [uid, dtstartMs] of sentReminders) {
|
||||||
const staleBucket = minuteBucket - 1
|
if (dtstartMs <= now.getTime()) {
|
||||||
for (const key of sentReminders) {
|
sentReminders.delete(uid)
|
||||||
const colonIdx = key.lastIndexOf(':')
|
|
||||||
if (colonIdx !== -1 && Number(key.slice(colonIdx + 1)) < staleBucket) {
|
|
||||||
sentReminders.delete(key)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user