diff --git a/src/lib/illness/red-flag-notify.ts b/src/lib/illness/red-flag-notify.ts index 6e9a2be5c..2ac3c23a6 100644 --- a/src/lib/illness/red-flag-notify.ts +++ b/src/lib/illness/red-flag-notify.ts @@ -22,6 +22,7 @@ import { prisma } from "@/lib/db"; import { getEvent } from "@/lib/logging/context"; import { dispatchLocalisedNotification } from "@/lib/notifications/dispatch-localised"; +import { claimNotificationEvent } from "@/lib/notifications/reminder-dedup"; import type { IllnessRedFlag } from "@/lib/illness/correlation"; /** One escalation per episode per this window. */ @@ -52,20 +53,18 @@ export async function notifyIllnessRedFlag(input: { }); if (user?.managedProfileAt) return; - // Dedupe: skip when an escalation for THIS episode already landed in the - // record event stream within the window. + // Claim this episode's record event before provider egress. The claim + // serializes concurrent readers without holding a transaction open for + // delivery itself. const reason = `${LEDGER_REASON_PREFIX}${episodeId}`; const since = new Date(Date.now() - DEDUPE_WINDOW_MS); - const prior = await prisma.notificationEvent.findFirst({ - where: { - recordUserId: userId, - eventType: "SYSTEM_ALERT", - dedupKey: reason, - createdAt: { gte: since }, - }, - select: { id: true }, + const claimed = await claimNotificationEvent(prisma, { + recordUserId: userId, + eventType: "SYSTEM_ALERT", + dedupKey: reason, + since, }); - if (prior) return; + if (!claimed) return; // Prefer the most clinically actionable reason for the body. Both copy // keys already exist in every locale bundle (i18n integrity guarantee). @@ -74,16 +73,6 @@ export async function notifyIllnessRedFlag(input: { ? "illness.correlation.redFlagFever" : "illness.correlation.redFlagSpo2"; - // Stamp the record event before dispatching so a slow/duplicate concurrent - // read cannot double-fire. Delivery attempts remain channel diagnostics. - await prisma.notificationEvent.create({ - data: { - recordUserId: userId, - eventType: "SYSTEM_ALERT", - dedupKey: reason, - }, - }); - await dispatchLocalisedNotification({ userId, titleKey: "illness.correlation.redFlagTitle", diff --git a/src/lib/illness/safety-floor-notify.ts b/src/lib/illness/safety-floor-notify.ts index d11cc70d7..9065b8fa9 100644 --- a/src/lib/illness/safety-floor-notify.ts +++ b/src/lib/illness/safety-floor-notify.ts @@ -20,6 +20,7 @@ import { prisma } from "@/lib/db"; import { getEvent } from "@/lib/logging/context"; import { dispatchLocalisedNotification } from "@/lib/notifications/dispatch-localised"; +import { claimNotificationEvent } from "@/lib/notifications/reminder-dedup"; import { convertGlucose, type GlucoseUnit } from "@/lib/glucose"; import { GLUCOSE_HYPO, @@ -123,26 +124,13 @@ export async function notifySafetyFloor(input: { const reason = `${LEDGER_REASON_PREFIX}${decision.reason}`; const since = new Date(Date.now() - DEDUPE_WINDOW_MS); - const prior = await prisma.notificationEvent.findFirst({ - where: { - recordUserId: userId, - eventType: "SYSTEM_ALERT", - dedupKey: reason, - createdAt: { gte: since }, - }, - select: { id: true }, - }); - if (prior) return; - - // Stamp the record event before dispatching so a concurrent confirm - // cannot double-fire. Delivery attempts remain channel diagnostics. - await prisma.notificationEvent.create({ - data: { - recordUserId: userId, - eventType: "SYSTEM_ALERT", - dedupKey: reason, - }, + const claimed = await claimNotificationEvent(prisma, { + recordUserId: userId, + eventType: "SYSTEM_ALERT", + dedupKey: reason, + since, }); + if (!claimed) return; const { titleKey, messageKey } = copyKeysFor(decision); await dispatchLocalisedNotification({ diff --git a/src/lib/jobs/__tests__/medication-reminder-check.test.ts b/src/lib/jobs/__tests__/medication-reminder-check.test.ts index 2db333d96..55ab1bfcd 100644 --- a/src/lib/jobs/__tests__/medication-reminder-check.test.ts +++ b/src/lib/jobs/__tests__/medication-reminder-check.test.ts @@ -103,9 +103,9 @@ vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({ recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined), })); vi.mock("@/lib/notifications/reminder-dedup", () => ({ - hasReminderDedupAnchor: vi.fn().mockResolvedValue(false), + claimNotificationEvent: vi.fn().mockResolvedValue(true), medicationReminderDedupKey: vi.fn(() => "dedup-key"), - writeReminderDedupAnchor: vi.fn().mockResolvedValue(undefined), + REMINDER_DEDUP_LOOKBACK_MS: 48 * 60 * 60 * 1000, })); import { handleReminderCheck } from "@/lib/jobs/reminder/medication-reminder-check"; diff --git a/src/lib/jobs/__tests__/medication-reminder-dedup.test.ts b/src/lib/jobs/__tests__/medication-reminder-dedup.test.ts index 1a0607e77..dfa47aadd 100644 --- a/src/lib/jobs/__tests__/medication-reminder-dedup.test.ts +++ b/src/lib/jobs/__tests__/medication-reminder-dedup.test.ts @@ -34,6 +34,8 @@ interface FakeNotificationEvent { const notificationEvents: FakeNotificationEvent[] = []; const prismaMock = { + $queryRaw: vi.fn().mockResolvedValue([{ locked: 1 }]), + $transaction: vi.fn(), telegramScheduledDeletion: { findMany: vi.fn().mockResolvedValue([]), deleteMany: vi.fn().mockResolvedValue({ count: 0 }), @@ -188,6 +190,7 @@ async function tickAt(iso: string): Promise { beforeEach(() => { vi.clearAllMocks(); notificationEvents.length = 0; + prismaMock.$transaction.mockImplementation(async (run) => run(prismaMock)); vi.useFakeTimers(); prismaMock.telegramScheduledDeletion.findMany.mockResolvedValue([]); prismaMock.telegramPromptContext.deleteMany.mockResolvedValue({ count: 0 }); diff --git a/src/lib/jobs/reminder/medication-reminder-check.ts b/src/lib/jobs/reminder/medication-reminder-check.ts index 7cd887cac..1643f69ce 100644 --- a/src/lib/jobs/reminder/medication-reminder-check.ts +++ b/src/lib/jobs/reminder/medication-reminder-check.ts @@ -31,9 +31,9 @@ import { getPhaseKeyboard, } from "@/lib/jobs/reminder-phases"; import { - hasReminderDedupAnchor, + claimNotificationEvent, medicationReminderDedupKey, - writeReminderDedupAnchor, + REMINDER_DEDUP_LOOKBACK_MS, } from "@/lib/notifications/reminder-dedup"; import { getUserTodayBounds as getUserTodayBoundsUtil } from "@/lib/tz/local-day"; import { getWorkerPrisma, parseTimeToMinutes } from "./shared"; @@ -544,13 +544,9 @@ export async function handleReminderCheck( continue; } - // Already notified for this slot, phase and local day? The - // anchor lives in `push_attempts` and is stamped for every - // dispatched slot regardless of which channels delivered. It - // used to be the Telegram message ledger, which is written only - // when a Telegram send succeeds — so a user on email / APNs / - // ntfy had no row and the tick re-sent the same overdue notice - // every 15 minutes for the rest of the day. + // Claim this slot/phase/local-day record event before mutable RED + // handling or provider egress. The short locked transaction keeps + // concurrent reminder ticks from both minting and delivering it. const dedupReason = medicationReminderDedupKey({ medicationId: med.id, scheduleId: schedule.id, @@ -561,15 +557,14 @@ export async function handleReminderCheck( }), }); - if ( - await hasReminderDedupAnchor(prisma, { - userId: med.user.id, + if (med.notificationsEnabled) { + const claimed = await claimNotificationEvent(prisma, { + recordUserId: med.user.id, eventType: "MEDICATION_REMINDER", - reason: dedupReason, - now, - }) - ) { - continue; + dedupKey: dedupReason, + since: new Date(now.getTime() - REMINDER_DEDUP_LOOKBACK_MS), + }); + if (!claimed) continue; } const doseInfo = schedule.dose ?? med.dose; @@ -651,15 +646,6 @@ export async function handleReminderCheck( `${currentPhase}:${med.name}:${slotTime}`, ); - // Stamp the dedup anchor BEFORE dispatching: a crash between - // the two costs this phase's notice, a repeat costs the user - // a notification every 15 minutes until the day rolls. - await writeReminderDedupAnchor(prisma, { - userId: med.user.id, - eventType: "MEDICATION_REMINDER", - reason: dedupReason, - }); - try { await dispatchNotification({ eventType: "MEDICATION_REMINDER", diff --git a/src/lib/notifications/reminder-dedup.ts b/src/lib/notifications/reminder-dedup.ts index 6f3fd2a11..16f5ee741 100644 --- a/src/lib/notifications/reminder-dedup.ts +++ b/src/lib/notifications/reminder-dedup.ts @@ -9,7 +9,7 @@ * pass for the rest of the local day. * * The anchor now lives in `notification_events`, a record-scoped event ledger - * rather than a channel delivery record. Two properties matter and both are + * rather than a channel delivery record. Three properties matter and all are * deliberate: * * - **Written per dispatched slot, not per successful send.** Anchoring @@ -18,6 +18,9 @@ * - **Written before the dispatch.** A crash mid-dispatch burns the slot * for that phase rather than risking a repeat; the next phase still * escalates, and the next local day starts clean. + * - **Claimed atomically.** A transaction-scoped advisory lock serializes + * same-key workers through the window check and append, while provider + * egress remains outside that short transaction. * */ import type { PrismaClient } from "@/generated/prisma/client"; @@ -28,7 +31,7 @@ import type { PrismaClient } from "@/generated/prisma/client"; * the read on the record/event/key index. Two days covers every timezone * offset plus a DST shift. */ -const ANCHOR_LOOKBACK_MS = 48 * 60 * 60 * 1000; +export const REMINDER_DEDUP_LOOKBACK_MS = 48 * 60 * 60 * 1000; /** * Dedup key for one medication dose slot in one phase on one local day. @@ -59,49 +62,59 @@ export function medicationReminderDedupKey(input: { } /** - * Has this exact slot+phase+day already been dispatched? Best-effort: a DB - * error biases toward "not yet", so a transient blip delays nothing — the - * cost of a rare duplicate is far below the cost of a silently dropped - * dose reminder. + * Atomically claim a record-scoped notification event within one rolling + * window. A `true` result is the sole permission to continue to provider + * egress; `false` means another worker already claimed it or the database + * could not establish the claim. The advisory lock, lookup, and append stay + * inside the short transaction. Callers always perform delivery afterward. */ -export async function hasReminderDedupAnchor( +export async function claimNotificationEvent( prisma: PrismaClient, - input: { userId: string; eventType: string; reason: string; now: Date }, + input: { + recordUserId: string; + eventType: string; + dedupKey: string; + since: Date; + }, ): Promise { try { - const row = await prisma.notificationEvent.findFirst({ - where: { - recordUserId: input.userId, - eventType: input.eventType, - dedupKey: input.reason, - createdAt: { gte: new Date(input.now.getTime() - ANCHOR_LOOKBACK_MS) }, - }, - select: { id: true }, + return await prisma.$transaction(async (tx) => { + const lockKey = [ + "notification-event", + input.recordUserId, + input.eventType, + input.dedupKey, + ].join(":"); + await tx.$queryRaw` + SELECT 1 AS locked + FROM pg_advisory_xact_lock(hashtextextended(${lockKey}, 0)) + `; + + const existing = await tx.notificationEvent.findFirst({ + where: { + recordUserId: input.recordUserId, + eventType: input.eventType, + dedupKey: input.dedupKey, + createdAt: { gte: input.since }, + }, + orderBy: { createdAt: "desc" }, + select: { id: true }, + }); + if (existing) return false; + + await tx.notificationEvent.create({ + data: { + recordUserId: input.recordUserId, + eventType: input.eventType, + dedupKey: input.dedupKey, + }, + }); + return true; }); - return row !== null; } catch { + // Fail closed: a worker that cannot durably claim the event must not + // contact a provider. A later tick may retry after the transaction rolls + // back, so an outage delays rather than duplicates a notification. return false; } } - -/** - * Stamp the anchor. Never throws — a failed stamp costs a duplicate - * notification at the next tick, while a thrown error would cost the - * dispatch itself. - */ -export async function writeReminderDedupAnchor( - prisma: PrismaClient, - input: { userId: string; eventType: string; reason: string }, -): Promise { - try { - await prisma.notificationEvent.create({ - data: { - recordUserId: input.userId, - eventType: input.eventType, - dedupKey: input.reason, - }, - }); - } catch { - // Best-effort by contract; the next tick re-derives the same slot. - } -} diff --git a/tests/integration/notification-attribution.test.ts b/tests/integration/notification-attribution.test.ts index a36131ce6..7f9e411b5 100644 --- a/tests/integration/notification-attribution.test.ts +++ b/tests/integration/notification-attribution.test.ts @@ -4,8 +4,6 @@ import { getPrismaClient, truncateAllTables } from "./setup"; import { resolveNotificationDeliveryIdentity } from "@/lib/notifications/delivery-identity"; import { claimNotificationEvent, - hasReminderDedupAnchor, - writeReminderDedupAnchor, } from "@/lib/notifications/reminder-dedup"; import { recordPushAttempt } from "@/lib/notifications/senders/push-attempt-record"; @@ -115,26 +113,28 @@ describe("notification attribution (real Postgres)", () => { const client = getPrismaClient(); const now = new Date(); - await writeReminderDedupAnchor(client, { - userId: recordUser.id, - eventType: "MEDICATION_REMINDER", - reason: "med:slot", - }); - await expect( - hasReminderDedupAnchor(client, { - userId: recordUser.id, + claimNotificationEvent(client, { + recordUserId: recordUser.id, eventType: "MEDICATION_REMINDER", - reason: "med:slot", - now, + dedupKey: "med:slot", + since: new Date(now.getTime() - 48 * 60 * 60 * 1000), }), ).resolves.toBe(true); - expect(await client.pushAttempt.count()).toBe(0); + await expect( + claimNotificationEvent(client, { + recordUserId: recordUser.id, + eventType: "MEDICATION_REMINDER", + dedupKey: "med:slot", + since: new Date(now.getTime() - 48 * 60 * 60 * 1000), + }), + ).resolves.toBe(false); const event = await client.notificationEvent.findFirstOrThrow({ where: { recordUserId: recordUser.id, dedupKey: "med:slot" }, select: { id: true }, }); + expect(await client.pushAttempt.count()).toBe(0); await client.user.delete({ where: { id: recordUser.id } }); expect( await client.notificationEvent.findUnique({ where: { id: event.id } }),