feat(webhook): dedupe deliveries with logicalEventId

README.md

@@ -14,7 +14,7 @@
   <img src="https://img.shields.io/badge/NestJS-framework-E0234E?logo=nestjs&logoColor=white" alt="NestJS" />
   <img src="https://img.shields.io/badge/PostgreSQL-database-4169E1?logo=postgresql&logoColor=white" alt="PostgreSQL" />
   <img src="https://img.shields.io/badge/License-Apache%202.0-blue" alt="Apache 2.0 License" />
-  <img src="https://img.shields.io/badge/Tests-1591%20passing-brightgreen" alt="1591 Tests Passing" />  <img src="https://img.shields.io/badge/AI%20Agents-12%20built--in-blueviolet" alt="12 AI Agents" />
+  <img src="https://img.shields.io/badge/Tests-1598%20passing-brightgreen" alt="1598 Tests Passing" />  <img src="https://img.shields.io/badge/AI%20Agents-12%20built--in-blueviolet" alt="12 AI Agents" />
 </p>
 
 <p align="center">
@@ -510,7 +510,7 @@ Operator notes for activating existing adapters, metasearch landings on the dire
 | OTA Channels | Booking.com + Expedia (EQC) + SiteMinder + DerbySoft | Direct + aggregated OTA connectivity (ARI + content) |
 | XML Processing | fast-xml-parser | Booking.com OTA XML protocol |
 | Package Manager | pnpm workspaces | Monorepo management |
-| Testing | Vitest (1591 tests across 220 test files) | Unit and integration tests || Build | tsup (packages) + Vite (dashboard) + nest build (API) | Fast builds |
+| Testing | Vitest (1598 tests across 221 test files) | Unit and integration tests || Build | tsup (packages) + Vite (dashboard) + nest build (API) | Fast builds |
 | Containers | Docker + docker-compose | Local dev and production deployment |
 | CI/CD | GitHub Actions | Automated testing, builds, and releases |
 
@@ -642,7 +642,7 @@ Before going live, verify the items in [`docs/deployment.md`](./docs/deployment.
 ### Run tests
 
 ```bash
-# All tests (1591 tests across 220 test files)
+# All tests (1598 tests across 221 test files)
 
 # API tests only
 pnpm --filter @telivityhaip/api test
@@ -1190,7 +1190,7 @@ HAIP is built in public and contributions are welcome.
 pnpm install          # Install dependencies
 pnpm build            # Build all workspace packages
 pnpm dev              # Start API in dev mode (hot reload)
-pnpm test             # Run all tests (1591 tests, 220 files)
+pnpm test             # Run all tests (1598 tests, 221 files)
 pnpm lint             # ESLint
 ```
 

apps/api/src/modules/connect/connect-events.service.spec.ts

@@ -187,6 +187,37 @@ describe('ConnectEventsService', () => {
       );
     });
 
+    it('forwards logicalEventId to WebhookDeliveryService for persisted dedup', async () => {
+      mockDb.select.mockImplementation(() => ({
+        from: vi.fn().mockReturnValue({
+          where: vi.fn().mockResolvedValue([mockSubscription]),
+        }),
+      }));
+
+      const logicalEventId = 'bbbbbbbb-0000-4000-a000-000000000002';
+      const deliveryService = { enqueue: vi.fn().mockResolvedValue({ id: 'del-1' }) };
+      const svc = new ConnectEventsService(mockDb, deliveryService as any);
+
+      await svc.handleEvent({
+        event: 'reservation.created',
+        entityType: 'reservation',
+        entityId: 'res-1',
+        propertyId: 'prop-1',
+        data: { foo: 'bar' },
+        timestamp: new Date().toISOString(),
+        logicalEventId,
+      });
+
+      expect(deliveryService.enqueue).toHaveBeenCalledWith(
+        expect.objectContaining({
+          eventType: 'reservation.created',
+          propertyId: 'prop-1',
+        }),
+        'sub-1',
+        logicalEventId,
+      );
+    });
+
     it('does nothing when no subscriptions match', async () => {
       mockDb.select.mockImplementation(() => ({
         from: vi.fn().mockReturnValue({

apps/api/src/modules/connect/connect-events.service.ts

@@ -5,6 +5,7 @@ import { agentWebhookSubscriptions, auditLogs } from '@telivityhaip/database';
 import { DRIZZLE } from '../../database/database.module';
 import type { CreateSubscriptionDto } from './dto/agent-event-subscription.dto';
 import { WebhookDeliveryService } from '../webhook/webhook-delivery.service';
+import type { WebhookPayload } from '../webhook/webhook.service';
 
 @Injectable()
 export class ConnectEventsService {
@@ -173,7 +174,7 @@ export class ConnectEventsService {
    * Listens to all events via wildcard.
    */
   @OnEvent('**')
-  async handleEvent(payload: any) {
+  async handleEvent(payload: WebhookPayload) {
     if (!payload?.propertyId || !payload?.event) return;
 
     // Find matching subscriptions
@@ -191,18 +192,23 @@ export class ConnectEventsService {
       const events = (sub.events ?? []) as string[];
       if (events.some((pattern: string) => this.matchesEventPattern(payload.event, pattern))) {
         if (this.deliveryService) {
-          // Enqueue a real HTTP delivery (HMAC-signed, retried).
-          await this.deliveryService.enqueue(
-            {
-              eventType: payload.event,
-              propertyId: payload.propertyId,
-              entityType: payload.entityType,
-              entityId: payload.entityId,
-              data: payload.data ?? {},
-              timestamp: payload.timestamp ?? new Date().toISOString(),
-            },
-            sub.id,
-          );
+          const deliveryPayload = {
+            eventType: payload.event,
+            propertyId: payload.propertyId,
+            entityType: payload.entityType,
+            entityId: payload.entityId,
+            data: payload.data ?? {},
+            timestamp: payload.timestamp ?? new Date().toISOString(),
+          };
+          if (payload.logicalEventId) {
+            await this.deliveryService.enqueue(
+              deliveryPayload,
+              sub.id,
+              payload.logicalEventId,
+            );
+          } else {
+            await this.deliveryService.enqueue(deliveryPayload, sub.id);
+          }
         } else {
           // Fallback — just log the match (for tests / environments without delivery service).
           await this.db

apps/api/src/modules/webhook/webhook-delivery.service.spec.ts

@@ -51,14 +51,27 @@ function createStatefulMockDb(subscription: any) {
       };
     }),
     insert: vi.fn((_tbl: any) => ({
-      values: vi.fn((vals: any) => ({
-        returning: vi.fn(() => {
+      values: vi.fn((vals: any) => {
+        const insertOnce = () => {
+          const existing = vals.logicalEventId
+            ? Array.from(deliveries.values()).find((row) =>
+              row.propertyId === vals.propertyId
+              && row.subscriptionId === vals.subscriptionId
+              && row.logicalEventId === vals.logicalEventId)
+            : undefined;
+          if (existing) return Promise.resolve([]);
           const id = `del-${idCounter++}`;
           const row = { id, ...vals };
           deliveries.set(id, row);
           return Promise.resolve([row]);
-        }),
-      })),
+        };
+        return {
+          returning: vi.fn(insertOnce),
+          onConflictDoNothing: vi.fn(() => ({
+            returning: vi.fn(insertOnce),
+          })),
+        };
+      }),
     })),
     update: vi.fn((_tbl: any) => ({
       set: vi.fn((vals: any) => ({
@@ -136,7 +149,7 @@ describe('WebhookDeliveryService', () => {
     expect(stored.attempts).toBe(0);
   });
 
-  it('worker POSTs with HMAC signature + event headers', async () => {
+  it('uses the delivery row ID as the event header for legacy events', async () => {
     fetchMock.mockResolvedValue({ ok: true, status: 200 });
     const db = createStatefulMockDb(subscription);
     const queue = createMockQueue();
@@ -166,6 +179,124 @@ describe('WebhookDeliveryService', () => {
     expect(stored.attempts).toBe(1);
   });
 
+  it('reuses one persisted delivery and stable header across event replay', async () => {
+    fetchMock
+      .mockResolvedValueOnce({ ok: false, status: 500 })
+      .mockResolvedValueOnce({ ok: true, status: 200 });
+    const db = createStatefulMockDb(subscription);
+    const queue = createMockQueue();
+    const service = new WebhookDeliveryService(
+      db as unknown as ConstructorParameters<typeof WebhookDeliveryService>[0],
+      undefined,
+      queue as unknown as ConstructorParameters<typeof WebhookDeliveryService>[2],
+    );
+    const logicalEventId = 'bbbbbbbb-0000-4000-a000-000000000002';
+
+    const first = await service.enqueue(payload, subscription.id, logicalEventId);
+    const replay = await service.enqueue(payload, subscription.id, logicalEventId);
+
+    expect(replay.id).toBe(first.id);
+    expect(db._deliveries.size).toBe(1);
+    expect(db._deliveries.get(first.id).logicalEventId).toBe(logicalEventId);
+    expect(queue.add).toHaveBeenCalledTimes(2);
+    expect(queue.add.mock.calls.map((call) => call[2]?.jobId)).toEqual([
+      first.id,
+      first.id,
+    ]);
+
+    await expect(
+      service.processDeliveryJob({ deliveryId: first.id, propertyId: 'prop-1' }),
+    ).rejects.toThrow('scheduled for retry');
+    await service.processDeliveryJob({ deliveryId: first.id, propertyId: 'prop-1' });
+    await service.processDeliveryJob({ deliveryId: replay.id, propertyId: 'prop-1' });
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(fetchMock.mock.calls.map((call) =>
+      call[1].headers['X-HAIP-Event-Id'])).toEqual([
+      logicalEventId,
+      logicalEventId,
+    ]);
+  });
+
+  it('creates separate deliveries for different persisted logical events', async () => {
+    const db = createStatefulMockDb(subscription);
+    const queue = createMockQueue();
+    const service = new WebhookDeliveryService(
+      db as unknown as ConstructorParameters<typeof WebhookDeliveryService>[0],
+      undefined,
+      queue as unknown as ConstructorParameters<typeof WebhookDeliveryService>[2],
+    );
+
+    const first = await service.enqueue(
+      payload,
+      subscription.id,
+      'bbbbbbbb-0000-4000-a000-000000000002',
+    );
+    const second = await service.enqueue(
+      payload,
+      subscription.id,
+      'bbbbbbbb-0000-4000-a000-000000000003',
+    );
+
+    expect(second.id).not.toBe(first.id);
+    expect(db._deliveries.size).toBe(2);
+    expect(Array.from(db._deliveries.values()).map((row) => row.logicalEventId)).toEqual([
+      'bbbbbbbb-0000-4000-a000-000000000002',
+      'bbbbbbbb-0000-4000-a000-000000000003',
+    ]);
+  });
+
+  it('returns the same delivery from concurrent persisted event creation', async () => {
+    const db = createStatefulMockDb(subscription);
+    const queue = createMockQueue();
+    const service = new WebhookDeliveryService(
+      db as unknown as ConstructorParameters<typeof WebhookDeliveryService>[0],
+      undefined,
+      queue as unknown as ConstructorParameters<typeof WebhookDeliveryService>[2],
+    );
+    const logicalEventId = 'bbbbbbbb-0000-4000-a000-000000000002';
+
+    const [first, replay] = await Promise.all([
+      service.enqueue(payload, subscription.id, logicalEventId),
+      service.enqueue(payload, subscription.id, logicalEventId),
+    ]);
+
+    expect(replay.id).toBe(first.id);
+    expect(db._deliveries.size).toBe(1);
+    expect(queue.add.mock.calls.map((call) => call[2]?.jobId)).toEqual([
+      first.id,
+      first.id,
+    ]);
+  });
+
+  it('requeues the existing delivery when the first queue write is lost', async () => {
+    const db = createStatefulMockDb(subscription);
+    const queue = createMockQueue();
+    queue.add
+      .mockRejectedValueOnce(new Error('Redis unavailable'))
+      .mockResolvedValueOnce({ id: 'job-recovered' });
+    const service = new WebhookDeliveryService(
+      db as unknown as ConstructorParameters<typeof WebhookDeliveryService>[0],
+      undefined,
+      queue as unknown as ConstructorParameters<typeof WebhookDeliveryService>[2],
+    );
+    const logicalEventId = 'bbbbbbbb-0000-4000-a000-000000000002';
+
+    await expect(
+      service.enqueue(payload, subscription.id, logicalEventId),
+    ).rejects.toThrow('Redis unavailable');
+    const [stored] = Array.from(db._deliveries.values());
+
+    const recovered = await service.enqueue(payload, subscription.id, logicalEventId);
+
+    expect(recovered.id).toBe(stored.id);
+    expect(db._deliveries.size).toBe(1);
+    expect(queue.add.mock.calls.map((call) => call[2]?.jobId)).toEqual([
+      stored.id,
+      stored.id,
+    ]);
+  });
+
   it('updates the row and throws so BullMQ retries on non-2xx response', async () => {
     fetchMock.mockResolvedValue({ ok: false, status: 500 });
     const db = createStatefulMockDb(subscription);

apps/api/src/modules/webhook/webhook-delivery.service.ts

@@ -56,6 +56,7 @@ interface WebhookDeliveryJob {
 }
 
 type DeliveryAttemptOutcome = 'delivered' | 'retry' | 'failed' | 'skipped';
+type WebhookDeliveryRow = typeof webhookDeliveries.$inferSelect;
 
 interface WebhookDeliveryQueue {
   add(name: string, data: WebhookDeliveryJob, options?: JobsOptions): Promise<unknown>;
@@ -111,22 +112,49 @@ export class WebhookDeliveryService implements OnModuleInit, OnModuleDestroy {
   }
 
   /**
-   * Enqueue deliveries for an event — one row per matching subscription,
-   * then add a durable BullMQ job for the worker.
+   * Enqueue one delivery per subscription and persisted logical event, then
+   * add its durable BullMQ job. Re-adding an existing row recovers a crash
+   * between the database insert and queue write; BullMQ deduplicates the UUID
+   * delivery job ID while an existing job remains present.
    */
-  async enqueue(payload: DeliveryPayload, subscriptionId: string) {
-    const [delivery] = await this.db
+  async enqueue(
+    payload: DeliveryPayload,
+    subscriptionId: string,
+    logicalEventId?: string,
+  ) {
+    const [inserted] = await this.db
       .insert(webhookDeliveries)
       .values({
         propertyId: payload.propertyId,
         subscriptionId,
+        logicalEventId: logicalEventId ?? null,
         eventType: payload.eventType,
         payload,
         status: 'pending',
         attempts: 0,
       })
+      .onConflictDoNothing()
       .returning();
 
+    let delivery = inserted as WebhookDeliveryRow | undefined;
+    if (!delivery && logicalEventId) {
+      const candidates = await this.db
+        .select()
+        .from(webhookDeliveries)
+        .where(and(
+          eq(webhookDeliveries.propertyId, payload.propertyId),
+          eq(webhookDeliveries.subscriptionId, subscriptionId),
+          eq(webhookDeliveries.logicalEventId, logicalEventId),
+        ));
+      delivery = candidates.find((candidate: WebhookDeliveryRow) =>
+        candidate.propertyId === payload.propertyId
+        && candidate.subscriptionId === subscriptionId
+        && candidate.logicalEventId === logicalEventId);
+    }
+    if (!delivery) {
+      throw new Error('Webhook delivery could not be created or recovered');
+    }
+
     await this.enqueueDeliveryJob(delivery.id, payload.propertyId);
 
     return delivery;
@@ -209,7 +237,7 @@ export class WebhookDeliveryService implements OnModuleInit, OnModuleDestroy {
           headers: {
             'Content-Type': 'application/json',
             'X-HAIP-Signature': signature,
-            'X-HAIP-Event-Id': delivery.id,
+            'X-HAIP-Event-Id': delivery.logicalEventId ?? delivery.id,
             'X-HAIP-Event-Type': delivery.eventType,
           },
           body,