fix: claim notification events atomically

- serialize rolling event claims before delivery\n- keep provider egress outside the transaction

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",

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({

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";

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<void> {
 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 });

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",

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<boolean> {
   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<void> {
-  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.
-  }
-}

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