fix(notifications): add stable Idempotency-Key header to outbound web…

src/services/notificationDispatchService.ts

@@ -1,3 +1,4 @@
+import { createHash } from 'crypto';
 import prisma from '../utils/prisma.js';
 import config from '../config/default.js';
 import logger from '../utils/logger.js';
@@ -8,10 +9,46 @@ export interface DispatchResult {
   error?: string;
 }
 
+/**
+ * Derives a stable idempotency key for a single logical webhook delivery.
+ *
+ * The key is a SHA-256 hex digest of `${outboxId}:${webhookUrl}`, which
+ * guarantees two properties:
+ *
+ *   1. **Retry-stable**: BullMQ retries of the same job (same outboxId,
+ *      same URL) always produce the identical key, so consumers can detect
+ *      and discard duplicates.
+ *
+ *   2. **Subscriber-unique**: if the same outbox row were ever fanned out
+ *      to two different webhook URLs (not currently the case — each User
+ *      has at most one webhookUrl), each subscriber receives a distinct key.
+ *
+ * NOTE — deliberate re-sends after permanent failure: the current schema
+ * has no "delivery generation" column on NotificationOutbox (no resetAt,
+ * redeliveryCount, etc.). A row that reaches DEAD_LETTER stays there; there
+ * is no operator-facing "reset and re-deliver" path yet. If such a flow is
+ * added in the future, a generation counter MUST be incorporated into this
+ * key (e.g. `${outboxId}:${webhookUrl}:${generation}`) so that a genuine
+ * re-send produces a new key and consumers do not suppress it. Track this
+ * as a follow-up: add a `deliveryGeneration Int @default(0)` column to
+ * NotificationOutbox and thread it through here.
+ *
+ * Consumer expectation: a webhook consumer that receives a POST with an
+ * Idempotency-Key it has already successfully processed SHOULD treat the
+ * repeat request as a no-op and return the original result without
+ * reprocessing. This repo cannot enforce that behaviour server-side — it
+ * is a documented contract for consumer implementations.
+ */
+export function buildWebhookIdempotencyKey(outboxId: string, webhookUrl: string): string {
+  return createHash('sha256').update(`${outboxId}:${webhookUrl}`).digest('hex');
+}
+
 async function sendWebhook(
+  outboxId: string,
   url: string,
   payload: Record<string, unknown>,
 ): Promise<DispatchResult> {
+  const idempotencyKey = buildWebhookIdempotencyKey(outboxId, url);
   const controller = new AbortController();
   const timeout = setTimeout(
     () => controller.abort(),
@@ -20,7 +57,10 @@ async function sendWebhook(
   try {
     const res = await fetch(url, {
       method: 'POST',
-      headers: { 'Content-Type': 'application/json' },
+      headers: {
+        'Content-Type': 'application/json',
+        'Idempotency-Key': idempotencyKey,
+      },
       body: JSON.stringify(payload),
       signal: controller.signal,
     });
@@ -53,6 +93,7 @@ export function sendEmail(to: string, payload: Record<string, unknown>): Dispatc
 
 export async function dispatchNotification(
   notificationId: string,
+  outboxId?: string,
 ): Promise<DispatchResult> {
   const notification = await prisma.notification.findUnique({
     where: { id: notificationId },
@@ -79,7 +120,11 @@ export async function dispatchNotification(
   const results: DispatchResult[] = [];
 
   if (user.webhookUrl) {
-    results.push(await sendWebhook(user.webhookUrl, payload));
+    // Use outboxId for the idempotency key when available (normal dispatch
+    // path). Fall back to notificationId so direct calls (e.g. tests, admin
+    // retrigger) still produce a stable, non-random key.
+    const keyBase = outboxId ?? notificationId;
+    results.push(await sendWebhook(keyBase, user.webhookUrl, payload));
   }
   if (user.email) {
     results.push(sendEmail(user.email, payload));

src/workers/notificationWorker.ts

@@ -70,13 +70,13 @@ export function startNotificationWorker(): void {
   worker = new Worker<NotificationJobData>(
     queueName,
     async (job) => {
-      const { notificationId, requestId } = job.data;
+      const { notificationId, outboxId, requestId } = job.data;
       return runWithRequestContext(requestId, async () => {
         logger.info('Dispatching notification', {
           notificationId,
           ...(requestId ? { requestId } : {}),
         });
-        await dispatchNotification(notificationId);
+        await dispatchNotification(notificationId, outboxId);
       });
     },
     { connection: getConnection() },