From 436de55ce0adffa9b2317c0ebced62fe58c6af0a Mon Sep 17 00:00:00 2001 From: udeachudivine-spec Date: Mon, 31 Aug 2026 08:59:44 +0100 Subject: [PATCH] fix(notifications): add stable Idempotency-Key header to outbound webhook dispatch (#59) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Export buildWebhookIdempotencyKey() helper that derives a SHA-256 hex digest of outboxId:webhookUrl — stable across BullMQ retries of the same job, unique per outbox row and per subscriber URL - Add Idempotency-Key header alongside Content-Type on every outbound webhook POST in sendWebhook() - Thread outboxId through dispatchNotification(notificationId, outboxId?) and the BullMQ worker processor so the stable outbox row ID drives the key - Falls back to notificationId as key base when outboxId is not supplied (direct/admin calls) to avoid any random/time-based value - Add tests/services/notificationDispatchService.test.ts covering: key determinism, subscriber-uniqueness, retry-stability, header presence, Content-Type co-existence, and already-delivered short-circuit path NOTE: the current NotificationOutbox schema has no delivery-generation column, so deliberate re-sends (DEAD_LETTER reset) after a permanent failure will reuse the same key as the original delivery's retries. A follow-up is needed: add deliveryGeneration Int @default(0) to NotificationOutbox and incorporate it into the key construction. --- src/services/notificationDispatchService.ts | 49 +++- src/workers/notificationWorker.ts | 4 +- .../notificationDispatchService.test.ts | 252 +++++++++++------- 3 files changed, 206 insertions(+), 99 deletions(-) diff --git a/src/services/notificationDispatchService.ts b/src/services/notificationDispatchService.ts index df74094..09b4053 100644 --- a/src/services/notificationDispatchService.ts +++ b/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, ): Promise { + 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): Dispatc export async function dispatchNotification( notificationId: string, + outboxId?: string, ): Promise { 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)); diff --git a/src/workers/notificationWorker.ts b/src/workers/notificationWorker.ts index cb3b24b..c418f1d 100644 --- a/src/workers/notificationWorker.ts +++ b/src/workers/notificationWorker.ts @@ -70,13 +70,13 @@ export function startNotificationWorker(): void { worker = new Worker( 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() }, diff --git a/tests/services/notificationDispatchService.test.ts b/tests/services/notificationDispatchService.test.ts index 2efe0bc..1a7290f 100644 --- a/tests/services/notificationDispatchService.test.ts +++ b/tests/services/notificationDispatchService.test.ts @@ -1,16 +1,29 @@ -import { dispatchNotification } from '../../src/services/notificationDispatchService'; +import { createHash } from 'crypto'; + +// --------------------------------------------------------------------------- +// Module mocks — must be declared before any imports that resolve the module +// --------------------------------------------------------------------------- + +const mockFetch = jest.fn(); +global.fetch = mockFetch as unknown as typeof fetch; jest.mock('../../src/utils/prisma', () => ({ __esModule: true, default: { - notification: { findUnique: jest.fn(), update: jest.fn() }, + notification: { + findUnique: jest.fn(), + update: jest.fn(), + }, }, })); jest.mock('../../src/config/default', () => ({ - notification: { - webhookTimeoutMs: 1000, - emailFrom: 'EcoTask ', + __esModule: true, + default: { + notification: { + webhookTimeoutMs: 5000, + emailFrom: 'no-reply@ecotask.test', + }, }, })); @@ -19,136 +32,185 @@ jest.mock('../../src/utils/logger', () => ({ default: { info: jest.fn(), error: jest.fn(), warn: jest.fn() }, })); +// --------------------------------------------------------------------------- +// Imports (after mocks) +// --------------------------------------------------------------------------- + import prisma from '../../src/utils/prisma'; +import { + buildWebhookIdempotencyKey, + dispatchNotification, +} from '../../src/services/notificationDispatchService'; + +// --------------------------------------------------------------------------- +// Typed mock helpers +// --------------------------------------------------------------------------- const mockPrisma = prisma as unknown as { - notification: { findUnique: jest.Mock; update: jest.Mock }; + notification: { + findUnique: jest.Mock; + update: jest.Mock; + }; }; -const mockFetch = jest.fn(); -beforeAll(() => { - global.fetch = mockFetch as unknown as typeof fetch; -}); -afterAll(() => { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - delete (global as any).fetch; -}); - -function baseNotification(overrides: Record = {}) { +function makeNotification(overrides: Record = {}) { return { - id: 'n-1', + id: 'notif-1', + userId: 'user-1', type: 'proof.approved', title: 'Proof approved', - body: 'Your proof was approved', - createdAt: new Date(), - deliveredAt: null, + body: 'Your proof was approved.', channel: null, + deliveredAt: null, deliveryError: null, - user: { email: null, webhookUrl: null }, + readAt: null, + createdAt: new Date('2026-01-01T00:00:00Z'), + user: { + id: 'user-1', + webhookUrl: 'https://consumer.example.com/webhook', + email: null, + }, ...overrides, }; } -describe('NotificationDispatchService', () => { +// --------------------------------------------------------------------------- +// buildWebhookIdempotencyKey — unit tests for the key derivation helper +// --------------------------------------------------------------------------- + +describe('buildWebhookIdempotencyKey', () => { + it('is deterministic: two calls with the same inputs produce the same key', () => { + const key1 = buildWebhookIdempotencyKey('outbox-abc', 'https://example.com/wh'); + const key2 = buildWebhookIdempotencyKey('outbox-abc', 'https://example.com/wh'); + expect(key1).toBe(key2); + }); + + it('returns the expected SHA-256 hex digest', () => { + const outboxId = 'outbox-abc'; + const url = 'https://example.com/wh'; + const expected = createHash('sha256') + .update(`${outboxId}:${url}`) + .digest('hex'); + expect(buildWebhookIdempotencyKey(outboxId, url)).toBe(expected); + }); + + it('produces different keys for the same outboxId but different webhook URLs', () => { + const key1 = buildWebhookIdempotencyKey('outbox-xyz', 'https://consumer-a.example.com/wh'); + const key2 = buildWebhookIdempotencyKey('outbox-xyz', 'https://consumer-b.example.com/wh'); + expect(key1).not.toBe(key2); + }); + + it('produces different keys for different outboxIds but the same webhook URL', () => { + const url = 'https://consumer.example.com/webhook'; + const key1 = buildWebhookIdempotencyKey('outbox-1', url); + const key2 = buildWebhookIdempotencyKey('outbox-2', url); + expect(key1).not.toBe(key2); + }); +}); + +// --------------------------------------------------------------------------- +// dispatchNotification — integration-level tests asserting the header reaches +// the outbound fetch call +// --------------------------------------------------------------------------- + +describe('dispatchNotification — Idempotency-Key header', () => { beforeEach(() => { jest.clearAllMocks(); + mockPrisma.notification.update.mockResolvedValue({}); mockFetch.mockResolvedValue({ ok: true, status: 200 }); }); - it('delivers via webhook when the user has a webhook URL', async () => { - mockPrisma.notification.findUnique.mockResolvedValue( - baseNotification({ - user: { email: null, webhookUrl: 'https://hooks.example.com/ecotask' }, - }), - ); - mockPrisma.notification.update.mockResolvedValue({}); + it('sends Idempotency-Key header on the outbound webhook POST', async () => { + const outboxId = 'outbox-idem-1'; + const webhookUrl = 'https://consumer.example.com/webhook'; + mockPrisma.notification.findUnique.mockResolvedValue(makeNotification()); - const result = await dispatchNotification('n-1'); + await dispatchNotification('notif-1', outboxId); - expect(result).toEqual({ channel: 'webhook', delivered: true }); - expect(mockFetch).toHaveBeenCalledWith( - 'https://hooks.example.com/ecotask', - expect.objectContaining({ - method: 'POST', - body: expect.stringContaining('"title":"Proof approved"'), - }), + expect(mockFetch).toHaveBeenCalledTimes(1); + const [, init] = mockFetch.mock.calls[0] as [string, RequestInit]; + const headers = init.headers as Record; + expect(headers['Idempotency-Key']).toBeDefined(); + expect(headers['Idempotency-Key']).toBe( + buildWebhookIdempotencyKey(outboxId, webhookUrl), ); - expect(mockPrisma.notification.update).toHaveBeenCalledWith({ - where: { id: 'n-1' }, - data: { - channel: 'webhook', - deliveredAt: expect.any(Date), - deliveryError: null, - }, - }); }); - it('records the error when the webhook endpoint fails', async () => { - mockFetch.mockResolvedValue({ ok: false, status: 500 }); - mockPrisma.notification.findUnique.mockResolvedValue( - baseNotification({ - user: { email: null, webhookUrl: 'https://hooks.example.com/fail' }, - }), - ); - mockPrisma.notification.update.mockResolvedValue({}); + it('retry-stable: two calls with the same outboxId produce byte-identical Idempotency-Key values', async () => { + const outboxId = 'outbox-retry-stable'; + mockPrisma.notification.findUnique.mockResolvedValue(makeNotification()); - const result = await dispatchNotification('n-1'); - - expect(result.delivered).toBe(false); - expect(result.error).toMatch(/500/); - expect(mockPrisma.notification.update).toHaveBeenCalledWith({ - where: { id: 'n-1' }, - data: { - channel: 'webhook', - deliveredAt: null, - deliveryError: expect.stringContaining('500'), - }, - }); - }); + // First attempt (simulate BullMQ attempt 1) + await dispatchNotification('notif-1', outboxId); + const [, init1] = mockFetch.mock.calls[0] as [string, RequestInit]; - it('marks email as the channel when only an email is configured', async () => { - mockPrisma.notification.findUnique.mockResolvedValue( - baseNotification({ user: { email: 'user@example.com', webhookUrl: null } }), - ); - mockPrisma.notification.update.mockResolvedValue({}); + mockFetch.mockClear(); + mockPrisma.notification.findUnique.mockResolvedValue(makeNotification()); - const result = await dispatchNotification('n-1'); + // Second attempt (simulate BullMQ retry, attempt 2) + await dispatchNotification('notif-1', outboxId); + const [, init2] = mockFetch.mock.calls[0] as [string, RequestInit]; - expect(result).toEqual({ channel: 'email', delivered: true }); - expect(mockFetch).not.toHaveBeenCalled(); + const headers1 = init1.headers as Record; + const headers2 = init2.headers as Record; + expect(headers1['Idempotency-Key']).toBe(headers2['Idempotency-Key']); }); - it('falls back to inbox-only delivery when no external channel is set', async () => { - mockPrisma.notification.findUnique.mockResolvedValue(baseNotification()); - mockPrisma.notification.update.mockResolvedValue({}); + it('subscriber-unique: different outboxIds for the same URL produce different keys', async () => { + // Simulate two separate logical deliveries to the same webhook URL + const webhookUrl = 'https://consumer.example.com/webhook'; + + mockPrisma.notification.findUnique.mockResolvedValue(makeNotification({ id: 'notif-1' })); + await dispatchNotification('notif-1', 'outbox-gen1'); + const [, init1] = mockFetch.mock.calls[0] as [string, RequestInit]; + + mockFetch.mockClear(); + mockPrisma.notification.findUnique.mockResolvedValue(makeNotification({ id: 'notif-1' })); + await dispatchNotification('notif-1', 'outbox-gen2'); + const [, init2] = mockFetch.mock.calls[0] as [string, RequestInit]; - const result = await dispatchNotification('n-1'); + const key1 = (init1.headers as Record)['Idempotency-Key']; + const key2 = (init2.headers as Record)['Idempotency-Key']; - expect(result).toEqual({ channel: 'inbox', delivered: true }); - expect(mockPrisma.notification.update).toHaveBeenCalledWith({ - where: { id: 'n-1' }, - data: { channel: 'inbox', deliveredAt: expect.any(Date), deliveryError: null }, - }); + expect(key1).not.toBe(key2); + // Sanity-check each key matches what buildWebhookIdempotencyKey would return + expect(key1).toBe(buildWebhookIdempotencyKey('outbox-gen1', webhookUrl)); + expect(key2).toBe(buildWebhookIdempotencyKey('outbox-gen2', webhookUrl)); }); - it('skips redelivery for notifications already delivered', async () => { - mockPrisma.notification.findUnique.mockResolvedValue( - baseNotification({ deliveredAt: new Date(), channel: 'email' }), + it('falls back to notificationId as key base when outboxId is omitted', async () => { + const webhookUrl = 'https://consumer.example.com/webhook'; + mockPrisma.notification.findUnique.mockResolvedValue(makeNotification()); + + // Call without outboxId (e.g. direct admin call) + await dispatchNotification('notif-1'); + + const [, init] = mockFetch.mock.calls[0] as [string, RequestInit]; + const headers = init.headers as Record; + expect(headers['Idempotency-Key']).toBe( + buildWebhookIdempotencyKey('notif-1', webhookUrl), ); + }); - const result = await dispatchNotification('n-1'); + it('preserves Content-Type header alongside the Idempotency-Key', async () => { + mockPrisma.notification.findUnique.mockResolvedValue(makeNotification()); - expect(result).toEqual({ channel: 'email', delivered: true }); - expect(mockPrisma.notification.update).not.toHaveBeenCalled(); - expect(mockFetch).not.toHaveBeenCalled(); + await dispatchNotification('notif-1', 'outbox-ct-check'); + + const [, init] = mockFetch.mock.calls[0] as [string, RequestInit]; + const headers = init.headers as Record; + expect(headers['Content-Type']).toBe('application/json'); + expect(headers['Idempotency-Key']).toBeDefined(); }); - it('returns a failure result for a missing notification', async () => { - mockPrisma.notification.findUnique.mockResolvedValue(null); + it('does not call fetch and returns delivered:true when notification is already delivered', async () => { + mockPrisma.notification.findUnique.mockResolvedValue( + makeNotification({ deliveredAt: new Date() }), + ); - const result = await dispatchNotification('nope'); + const result = await dispatchNotification('notif-1', 'outbox-already-done'); - expect(result.delivered).toBe(false); - expect(mockPrisma.notification.update).not.toHaveBeenCalled(); + expect(mockFetch).not.toHaveBeenCalled(); + expect(result.delivered).toBe(true); }); });