From 45a23e3d2635d7ada91b3dea54c2bda4b44d9158 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 26 Aug 2026 18:33:05 +0000 Subject: [PATCH 1/5] feat(webhook): dedupe deliveries with logicalEventId Add logical_event_id column and unique index on webhook_deliveries. Skip duplicate enqueue when the same logical event is already recorded. Co-authored-by: Agus --- .../webhook/webhook-delivery.service.spec.ts | 141 +++++++++++++++++- .../webhook/webhook-delivery.service.ts | 38 ++++- .../modules/webhook/webhook.service.spec.ts | 58 +++++++ .../src/modules/webhook/webhook.service.ts | 23 ++- .../0022_webhook_logical_event_id.sql | 6 + packages/database/src/push-schema.ts | 3 + packages/database/src/schema/connect.ts | 20 ++- 7 files changed, 275 insertions(+), 14 deletions(-) create mode 100644 apps/api/src/modules/webhook/webhook.service.spec.ts create mode 100644 packages/database/src/migrations/0022_webhook_logical_event_id.sql diff --git a/apps/api/src/modules/webhook/webhook-delivery.service.spec.ts b/apps/api/src/modules/webhook/webhook-delivery.service.spec.ts index f7016dfe..83213d21 100644 --- a/apps/api/src/modules/webhook/webhook-delivery.service.spec.ts +++ b/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[0], + undefined, + queue as unknown as ConstructorParameters[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[0], + undefined, + queue as unknown as ConstructorParameters[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[0], + undefined, + queue as unknown as ConstructorParameters[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[0], + undefined, + queue as unknown as ConstructorParameters[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); diff --git a/apps/api/src/modules/webhook/webhook-delivery.service.ts b/apps/api/src/modules/webhook/webhook-delivery.service.ts index 6c93eeb9..b0be8612 100644 --- a/apps/api/src/modules/webhook/webhook-delivery.service.ts +++ b/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; @@ -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, diff --git a/apps/api/src/modules/webhook/webhook.service.spec.ts b/apps/api/src/modules/webhook/webhook.service.spec.ts new file mode 100644 index 00000000..c6a50025 --- /dev/null +++ b/apps/api/src/modules/webhook/webhook.service.spec.ts @@ -0,0 +1,58 @@ +import { describe, expect, it, vi } from 'vitest'; +import { WebhookService, type WebhookPayload } from './webhook.service'; + +describe('WebhookService persisted dispatch', () => { + it('adds the stable logical event ID only to the internal dispatch envelope', async () => { + const db = { insert: vi.fn() }; + const eventEmitter = { emitAsync: vi.fn().mockResolvedValue([]) }; + const service = new WebhookService( + db as unknown as ConstructorParameters[0], + eventEmitter as unknown as ConstructorParameters[1], + ); + const payload: WebhookPayload = { + event: 'booking_request.created', + entityType: 'booking_request', + entityId: 'bbbbbbbb-0000-4000-a000-000000000001', + propertyId: 'aaaaaaaa-0000-4000-a000-000000000001', + data: { + requestId: 'bbbbbbbb-0000-4000-a000-000000000001', + status: 'pending', + }, + timestamp: '2026-08-24T17:15:00.000Z', + }; + + await service.dispatchPersisted( + payload, + 'bbbbbbbb-0000-4000-a000-000000000002', + ); + + expect(eventEmitter.emitAsync).toHaveBeenCalledWith( + 'booking_request.created', + { + ...payload, + logicalEventId: 'bbbbbbbb-0000-4000-a000-000000000002', + }, + ); + expect(payload).not.toHaveProperty('logicalEventId'); + expect(db.insert).not.toHaveBeenCalled(); + }); + + it('rejects a persisted dispatch name outside the shared WebhookEvent catalog', async () => { + const eventEmitter = { emitAsync: vi.fn().mockResolvedValue([]) }; + const service = new WebhookService( + { insert: vi.fn() } as unknown as ConstructorParameters[0], + eventEmitter as unknown as ConstructorParameters[1], + ); + const payload = { + event: 'payment.retained', + entityType: 'booking_request_payment_resolution', + entityId: 'bbbbbbbb-0000-4000-a000-000000000001', + data: {}, + timestamp: '2026-08-25T00:00:00.000Z', + } as unknown as WebhookPayload; + + await expect(service.dispatchPersisted(payload, 'logical-event-1')) + .rejects.toThrow(/unknown persisted webhook event/i); + expect(eventEmitter.emitAsync).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/api/src/modules/webhook/webhook.service.ts b/apps/api/src/modules/webhook/webhook.service.ts index 07c18d35..cfda8f32 100644 --- a/apps/api/src/modules/webhook/webhook.service.ts +++ b/apps/api/src/modules/webhook/webhook.service.ts @@ -1,16 +1,17 @@ import { Injectable, Inject } from '@nestjs/common'; import { EventEmitter2 } from '@nestjs/event-emitter'; import { auditLogs } from '@telivityhaip/database'; -import { type WebhookEvent } from '@telivityhaip/shared'; +import { WEBHOOK_EVENTS, type WebhookEvent } from '@telivityhaip/shared'; import { DRIZZLE } from '../../database/database.module'; export interface WebhookPayload { - event: string; + event: WebhookEvent; entityType: string; entityId: string; propertyId?: string; data: Record; timestamp: string; + logicalEventId?: string; } @Injectable() @@ -20,6 +21,24 @@ export class WebhookService { private readonly eventEmitter: EventEmitter2, ) {} + /** + * Dispatch a payload whose audit/outbox records were already committed by + * the domain transaction. Async listeners are awaited so a durable caller + * can retain and retry its pending consequence on delivery failure. + */ + async dispatchPersisted( + payload: WebhookPayload, + logicalEventId: string, + ): Promise { + if (!Object.hasOwn(WEBHOOK_EVENTS, payload.event)) { + throw new Error(`Unknown persisted webhook event: ${String(payload.event)}`); + } + await this.eventEmitter.emitAsync(payload.event, { + ...payload, + logicalEventId, + } satisfies WebhookPayload); + } + /** * Emit a webhook event and log it to the audit trail. */ diff --git a/packages/database/src/migrations/0022_webhook_logical_event_id.sql b/packages/database/src/migrations/0022_webhook_logical_event_id.sql new file mode 100644 index 00000000..3a9900fe --- /dev/null +++ b/packages/database/src/migrations/0022_webhook_logical_event_id.sql @@ -0,0 +1,6 @@ +-- Webhook delivery deduplication by logical event id (crash-safe re-enqueue). +ALTER TABLE webhook_deliveries + ADD COLUMN IF NOT EXISTS logical_event_id uuid; + +CREATE UNIQUE INDEX IF NOT EXISTS webhook_deliveries_property_subscription_logical_event_unique + ON webhook_deliveries (property_id, subscription_id, logical_event_id); diff --git a/packages/database/src/push-schema.ts b/packages/database/src/push-schema.ts index 3fcf7950..75222eda 100644 --- a/packages/database/src/push-schema.ts +++ b/packages/database/src/push-schema.ts @@ -715,6 +715,7 @@ async function main() { id uuid PRIMARY KEY DEFAULT gen_random_uuid(), property_id uuid NOT NULL REFERENCES properties(id), subscription_id uuid NOT NULL REFERENCES agent_webhook_subscriptions(id), + logical_event_id uuid, event_type varchar(100) NOT NULL, payload jsonb NOT NULL, status webhook_delivery_status NOT NULL DEFAULT 'pending', @@ -1472,6 +1473,8 @@ async function main() { `ALTER TABLE guest_reviews ADD COLUMN IF NOT EXISTS provider_channel_id varchar(255)`, `ALTER TABLE guest_reviews ADD COLUMN IF NOT EXISTS last_synced_at timestamptz`, `CREATE UNIQUE INDEX IF NOT EXISTS guest_reviews_property_source_external_unique ON guest_reviews (property_id, source, external_id)`, + `ALTER TABLE webhook_deliveries ADD COLUMN IF NOT EXISTS logical_event_id uuid`, + `CREATE UNIQUE INDEX IF NOT EXISTS webhook_deliveries_property_subscription_logical_event_unique ON webhook_deliveries (property_id, subscription_id, logical_event_id)`, ]; for (const a of alters) { await db.execute(sql.raw(a)); diff --git a/packages/database/src/schema/connect.ts b/packages/database/src/schema/connect.ts index 180cfc72..5b8f5520 100644 --- a/packages/database/src/schema/connect.ts +++ b/packages/database/src/schema/connect.ts @@ -1,4 +1,15 @@ -import { pgTable, uuid, varchar, boolean, timestamp, jsonb, integer, text, pgEnum } from 'drizzle-orm/pg-core'; +import { + pgTable, + uuid, + varchar, + boolean, + timestamp, + jsonb, + integer, + text, + pgEnum, + uniqueIndex, +} from 'drizzle-orm/pg-core'; import { properties } from './property.js'; /** @@ -46,6 +57,7 @@ export const webhookDeliveries = pgTable('webhook_deliveries', { id: uuid('id').primaryKey().defaultRandom(), propertyId: uuid('property_id').notNull().references(() => properties.id), subscriptionId: uuid('subscription_id').notNull().references(() => agentWebhookSubscriptions.id), + logicalEventId: uuid('logical_event_id'), eventType: varchar('event_type', { length: 100 }).notNull(), payload: jsonb('payload').notNull(), @@ -59,7 +71,11 @@ export const webhookDeliveries = pgTable('webhook_deliveries', { createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(), deliveredAt: timestamp('delivered_at', { withTimezone: true }), -}); +}, (table) => ({ + propertySubscriptionLogicalEventUnique: + uniqueIndex('webhook_deliveries_property_subscription_logical_event_unique') + .on(table.propertyId, table.subscriptionId, table.logicalEventId), +})); /** * Connect API credentials — tenant-bound API keys for the /api/v1/connect/* surface. From 9417a34ded86ec64b3822c047aae14580cafb24c Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 26 Aug 2026 19:57:03 +0000 Subject: [PATCH 2/5] chore: sync README test counts for CI Co-authored-by: telivity-otaip --- README.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 0459cd55..17d677da 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,7 @@ NestJS PostgreSQL Apache 2.0 License -1568 Tests Passing 12 AI Agents +1574 Tests Passing 12 AI Agents

@@ -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 (1568 tests across 218 test files) | Unit and integration tests || Build | tsup (packages) + Vite (dashboard) + nest build (API) | Fast builds | +| Testing | Vitest (1574 tests across 219 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 (1568 tests across 218 test files) +# All tests (1574 tests across 219 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 (1568 tests, 218 files) +pnpm test # Run all tests (1574 tests, 219 files) pnpm lint # ESLint ``` From 5cec6220f47d31ff59af0d7a2f0fef9289e09f08 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 26 Aug 2026 20:08:41 +0000 Subject: [PATCH 3/5] fix(webhook): use reservation.created in logicalEventId dispatch spec booking_request.created is not in the core WEBHOOK_EVENTS catalog; the dedup envelope test should use a shipped event name. Co-authored-by: telivity-otaip --- apps/api/src/modules/webhook/webhook.service.spec.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/apps/api/src/modules/webhook/webhook.service.spec.ts b/apps/api/src/modules/webhook/webhook.service.spec.ts index c6a50025..3874c5ef 100644 --- a/apps/api/src/modules/webhook/webhook.service.spec.ts +++ b/apps/api/src/modules/webhook/webhook.service.spec.ts @@ -10,12 +10,12 @@ describe('WebhookService persisted dispatch', () => { eventEmitter as unknown as ConstructorParameters[1], ); const payload: WebhookPayload = { - event: 'booking_request.created', - entityType: 'booking_request', + event: 'reservation.created', + entityType: 'reservation', entityId: 'bbbbbbbb-0000-4000-a000-000000000001', propertyId: 'aaaaaaaa-0000-4000-a000-000000000001', data: { - requestId: 'bbbbbbbb-0000-4000-a000-000000000001', + reservationId: 'bbbbbbbb-0000-4000-a000-000000000001', status: 'pending', }, timestamp: '2026-08-24T17:15:00.000Z', @@ -27,7 +27,7 @@ describe('WebhookService persisted dispatch', () => { ); expect(eventEmitter.emitAsync).toHaveBeenCalledWith( - 'booking_request.created', + 'reservation.created', { ...payload, logicalEventId: 'bbbbbbbb-0000-4000-a000-000000000002', From 73b42b19cc859bbedbc3489d707ed22a97040799 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 27 Aug 2026 00:51:28 +0000 Subject: [PATCH 4/5] =?UTF-8?q?fix(webhook):=20address=20#350=20review=20?= =?UTF-8?q?=E2=80=94=20connect=20dedup=20wiring=20and=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Forward logicalEventId from Connect event listener to delivery enqueue. Keep logical_event_id DDL in migration/CREATE only (not push-schema alters). Document X-HAIP-Event-Id logical-vs-legacy contract. Co-authored-by: telivity-otaip --- .../connect/connect-events.service.spec.ts | 31 ++++++++++++++++++ .../modules/connect/connect-events.service.ts | 32 +++++++++++-------- docs/webhooks.md | 2 +- packages/database/src/push-schema.ts | 2 -- 4 files changed, 51 insertions(+), 16 deletions(-) diff --git a/apps/api/src/modules/connect/connect-events.service.spec.ts b/apps/api/src/modules/connect/connect-events.service.spec.ts index db70fc46..bb7d6760 100644 --- a/apps/api/src/modules/connect/connect-events.service.spec.ts +++ b/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({ diff --git a/apps/api/src/modules/connect/connect-events.service.ts b/apps/api/src/modules/connect/connect-events.service.ts index c840f19d..9643ef6e 100644 --- a/apps/api/src/modules/connect/connect-events.service.ts +++ b/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 diff --git a/docs/webhooks.md b/docs/webhooks.md index dd1f85f3..011c231f 100644 --- a/docs/webhooks.md +++ b/docs/webhooks.md @@ -52,7 +52,7 @@ Each matching event is POSTed to your `callbackUrl`: POST Content-Type: application/json X-HAIP-Signature: sha256= -X-HAIP-Event-Id: +X-HAIP-Event-Id: X-HAIP-Event-Type: reservation.checked_in ``` diff --git a/packages/database/src/push-schema.ts b/packages/database/src/push-schema.ts index 75222eda..c59fd597 100644 --- a/packages/database/src/push-schema.ts +++ b/packages/database/src/push-schema.ts @@ -1473,8 +1473,6 @@ async function main() { `ALTER TABLE guest_reviews ADD COLUMN IF NOT EXISTS provider_channel_id varchar(255)`, `ALTER TABLE guest_reviews ADD COLUMN IF NOT EXISTS last_synced_at timestamptz`, `CREATE UNIQUE INDEX IF NOT EXISTS guest_reviews_property_source_external_unique ON guest_reviews (property_id, source, external_id)`, - `ALTER TABLE webhook_deliveries ADD COLUMN IF NOT EXISTS logical_event_id uuid`, - `CREATE UNIQUE INDEX IF NOT EXISTS webhook_deliveries_property_subscription_logical_event_unique ON webhook_deliveries (property_id, subscription_id, logical_event_id)`, ]; for (const a of alters) { await db.execute(sql.raw(a)); From 89cf718def77f79f0300837b4926db466d86fb7a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 27 Aug 2026 08:28:49 +0000 Subject: [PATCH 5/5] fix(database): run numbered SQL migrations from 0022 via ledger Implement Agus review feedback for PR #350: - Add migration-runner with schema_migrations ledger; execute 0022+ SQL files once inside transactions after push-schema baseline (0001-0021) - Point db:migrate, Docker, Render, and release smoke at run-migrations.js - Copy src/migrations into dist/ on build so deployment ships SQL files - Remove logical_event_id DDL from push-schema CREATE TABLE (0022 only) - Guard push-schema CLI so importing it does not double-run baseline DDL - Add PostgreSQL integration tests: fresh DB, pre-0022 upgrade, ledger idempotency, and unique-index duplicate rejection Co-authored-by: telivity-otaip --- .../api/src/release-smoke.integration.spec.ts | 2 +- docker-compose.yml | 2 +- packages/database/package.json | 4 +- .../database/src/migration-runner.spec.ts | 236 ++++++++++++++++++ packages/database/src/migration-runner.ts | 139 +++++++++++ packages/database/src/push-schema.ts | 20 +- packages/database/src/run-migrations.ts | 12 + packages/database/tsup.config.ts | 16 +- render.yaml | 2 +- 9 files changed, 418 insertions(+), 15 deletions(-) create mode 100644 packages/database/src/migration-runner.spec.ts create mode 100644 packages/database/src/migration-runner.ts create mode 100644 packages/database/src/run-migrations.ts diff --git a/apps/api/src/release-smoke.integration.spec.ts b/apps/api/src/release-smoke.integration.spec.ts index 9fe9ed59..aa6d4014 100644 --- a/apps/api/src/release-smoke.integration.spec.ts +++ b/apps/api/src/release-smoke.integration.spec.ts @@ -43,7 +43,7 @@ describe.runIf(runSmoke)('release smoke (migrate → reservation lifecycle)', () process.env.REDIS_URL = process.env.REDIS_URL ?? 'redis://localhost:6379'; const root = join(__dirname, '..', '..', '..'); - execSync('node packages/database/dist/push-schema.js', { + execSync('node packages/database/dist/run-migrations.js', { cwd: root, env: { ...process.env, DATABASE_URL: dbUrl }, stdio: 'inherit', diff --git a/docker-compose.yml b/docker-compose.yml index d770f6f9..7a0022f7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -73,7 +73,7 @@ services: dockerfile: apps/api/Dockerfile image: haip-api container_name: haip-init - command: sh -c "node packages/database/dist/push-schema.js && node packages/database/dist/seed.js" + command: sh -c "node packages/database/dist/run-migrations.js && node packages/database/dist/seed.js" environment: DATABASE_URL: postgresql://haip:haip@postgres:5432/haip depends_on: diff --git a/packages/database/package.json b/packages/database/package.json index 3399c590..be130b82 100644 --- a/packages/database/package.json +++ b/packages/database/package.json @@ -42,9 +42,9 @@ "test:coverage": "vitest run --coverage", "seed": "tsx src/seed.ts", "link-integration": "tsx src/link-integration-principal.ts", - "migrate": "tsx src/push-schema.ts", + "migrate": "tsx src/run-migrations.ts", "db:generate": "drizzle-kit generate", - "db:migrate": "tsx src/push-schema.ts", + "db:migrate": "tsx src/run-migrations.ts", "db:studio": "drizzle-kit studio" }, "dependencies": { diff --git a/packages/database/src/migration-runner.spec.ts b/packages/database/src/migration-runner.spec.ts new file mode 100644 index 00000000..4c3196bc --- /dev/null +++ b/packages/database/src/migration-runner.spec.ts @@ -0,0 +1,236 @@ +/** + * PostgreSQL integration tests for the numbered SQL migration runner (0022+). + * + * Uses ephemeral databases when Postgres is reachable (haip_test pattern). + */ +import { randomBytes } from 'node:crypto'; +import { dirname, join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import postgres from 'postgres'; +import { postgresOptionsFromEnv } from './postgres-options.js'; +import { pushSchema } from './push-schema.js'; +import { + FIRST_TRACKED_MIGRATION_VERSION, + getAppliedMigrationVersions, + listPendingMigrations, + resolveMigrationsDirectory, + runAllMigrations, + runPendingSqlMigrations, +} from './migration-runner.js'; + +const DEFAULT_DATABASE_URL = + process.env['DATABASE_URL'] ?? 'postgresql://haip:haip@localhost:5432/haip_test'; + +const ADMIN_DATABASE_URL = DEFAULT_DATABASE_URL.replace(/\/[^/]+$/, '/postgres'); + +const MIGRATIONS_DIR = resolveMigrationsDirectory( + dirname(fileURLToPath(import.meta.url)), +); + +async function postgresReachable(url: string): Promise { + const client = postgres(url, { ...postgresOptionsFromEnv(), max: 1 }); + try { + await client`SELECT 1`; + return true; + } catch { + return false; + } finally { + await client.end(); + } +} + +async function withEphemeralDatabase(run: (databaseUrl: string) => Promise): Promise { + const dbName = `haip_mig_${randomBytes(6).toString('hex')}`; + const admin = postgres(ADMIN_DATABASE_URL, { ...postgresOptionsFromEnv(), max: 1 }); + const databaseUrl = DEFAULT_DATABASE_URL.replace(/\/[^/]+$/, `/${dbName}`); + + await admin.unsafe(`CREATE DATABASE "${dbName}"`); + try { + await run(databaseUrl); + } finally { + await admin.unsafe(`DROP DATABASE IF EXISTS "${dbName}" WITH (FORCE)`); + await admin.end(); + } +} + +async function columnExists(sql: postgres.Sql, column: string): Promise { + const rows = await sql<{ exists: boolean }[]>` + SELECT EXISTS ( + SELECT 1 + FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'webhook_deliveries' + AND column_name = ${column} + ) AS exists + `; + return rows[0]?.exists ?? false; +} + +async function uniqueIndexExists(sql: postgres.Sql, indexName: string): Promise { + const rows = await sql<{ exists: boolean }[]>` + SELECT EXISTS ( + SELECT 1 + FROM pg_indexes + WHERE schemaname = 'public' + AND indexname = ${indexName} + ) AS exists + `; + return rows[0]?.exists ?? false; +} + +async function seedWebhookDedupFixtures(sql: postgres.Sql) { + const propertyId = 'f3500001-0000-4000-a000-000000000001'; + const subscriptionId = 'f3500002-0000-4000-a000-000000000002'; + const logicalEventId = 'f3500003-0000-4000-a000-000000000003'; + + await sql` + INSERT INTO properties ( + id, name, code, country_code, timezone, currency_code, total_rooms + ) VALUES ( + ${propertyId}, + 'Migration test property', + ${`M350${randomBytes(2).toString('hex').toUpperCase()}`}, + 'US', + 'America/New_York', + 'USD', + 10 + ) + `; + + await sql` + INSERT INTO agent_webhook_subscriptions ( + id, property_id, subscriber_id, callback_url, events + ) VALUES ( + ${subscriptionId}, + ${propertyId}, + 'migration-test-subscriber', + 'https://example.com/webhooks', + ${JSON.stringify(['reservation.created'])} + ) + `; + + return { propertyId, subscriptionId, logicalEventId }; +} + +const postgresReady = await postgresReachable(DEFAULT_DATABASE_URL); + +describe.skipIf(!postgresReady)('migration runner (PostgreSQL)', () => { + beforeAll(async () => { + if (!postgresReady) return; + const canCreateDb = await postgresReachable(ADMIN_DATABASE_URL); + if (!canCreateDb) { + throw new Error('Postgres admin connection required for ephemeral migration tests'); + } + }); + + it('applies 0022 on a fresh database', async () => { + await withEphemeralDatabase(async (databaseUrl) => { + await runAllMigrations(databaseUrl, { migrationsDir: MIGRATIONS_DIR }); + + const sql = postgres(databaseUrl, postgresOptionsFromEnv()); + try { + expect(await columnExists(sql, 'logical_event_id')).toBe(true); + expect(await uniqueIndexExists( + sql, + 'webhook_deliveries_property_subscription_logical_event_unique', + )).toBe(true); + + const applied = await getAppliedMigrationVersions(sql); + expect(applied.has(FIRST_TRACKED_MIGRATION_VERSION)).toBe(true); + } finally { + await sql.end(); + } + }); + }); + + it('upgrades an existing pre-0022 database (push-schema baseline only)', async () => { + await withEphemeralDatabase(async (databaseUrl) => { + await pushSchema(databaseUrl); + + const sql = postgres(databaseUrl, postgresOptionsFromEnv()); + try { + expect(await columnExists(sql, 'logical_event_id')).toBe(false); + expect(await getAppliedMigrationVersions(sql).then((v) => v.size)).toBe(0); + + const pendingBefore = await listPendingMigrations(sql, MIGRATIONS_DIR); + expect(pendingBefore.some((m) => m.version === FIRST_TRACKED_MIGRATION_VERSION)).toBe(true); + + await runPendingSqlMigrations(sql, MIGRATIONS_DIR); + + expect(await columnExists(sql, 'logical_event_id')).toBe(true); + expect(await uniqueIndexExists( + sql, + 'webhook_deliveries_property_subscription_logical_event_unique', + )).toBe(true); + } finally { + await sql.end(); + } + }); + }); + + it('skips already-applied migrations when run twice (ledger idempotency)', async () => { + await withEphemeralDatabase(async (databaseUrl) => { + await runAllMigrations(databaseUrl, { migrationsDir: MIGRATIONS_DIR }); + + const sql = postgres(databaseUrl, postgresOptionsFromEnv()); + try { + const firstApplied = await getAppliedMigrationVersions(sql); + expect(firstApplied.has(FIRST_TRACKED_MIGRATION_VERSION)).toBe(true); + + const secondPass = await runPendingSqlMigrations(sql, MIGRATIONS_DIR); + expect(secondPass).toEqual([]); + + const appliedAfter = await getAppliedMigrationVersions(sql); + expect(appliedAfter.size).toBe(firstApplied.size); + } finally { + await sql.end(); + } + }); + }); + + it('rejects duplicate (property_id, subscription_id, logical_event_id) rows', async () => { + await withEphemeralDatabase(async (databaseUrl) => { + await runAllMigrations(databaseUrl, { migrationsDir: MIGRATIONS_DIR }); + + const sql = postgres(databaseUrl, postgresOptionsFromEnv()); + try { + const { propertyId, subscriptionId, logicalEventId } = await seedWebhookDedupFixtures(sql); + const payload = JSON.stringify({ eventType: 'reservation.created' }); + + await sql` + INSERT INTO webhook_deliveries ( + property_id, subscription_id, logical_event_id, event_type, payload + ) VALUES ( + ${propertyId}, + ${subscriptionId}, + ${logicalEventId}, + 'reservation.created', + ${payload}::jsonb + ) + `; + + await expect(sql` + INSERT INTO webhook_deliveries ( + property_id, subscription_id, logical_event_id, event_type, payload + ) VALUES ( + ${propertyId}, + ${subscriptionId}, + ${logicalEventId}, + 'reservation.created', + ${payload}::jsonb + ) + `).rejects.toMatchObject({ code: '23505' }); + } finally { + await sql.end(); + } + }); + }); +}); + +describe('migration runner (unit)', () => { + it('resolves migrations adjacent to the compiled entry directory', () => { + const dir = resolveMigrationsDirectory(join(dirname(fileURLToPath(import.meta.url)))); + expect(dir.endsWith(`${join('src', 'migrations')}`) || dir.endsWith(`${join('dist', 'migrations')}`)).toBe(true); + }); +}); diff --git a/packages/database/src/migration-runner.ts b/packages/database/src/migration-runner.ts new file mode 100644 index 00000000..3e5f9d41 --- /dev/null +++ b/packages/database/src/migration-runner.ts @@ -0,0 +1,139 @@ +/** + * Numbered SQL migration runner for schema changes from 0022 onward. + * + * Migrations 0001–0021 are incorporated idempotently by push-schema.ts (legacy + * baseline). This runner records applied versions in schema_migrations and + * executes each new .sql file once, inside a transaction. + */ +import { readdir, readFile } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import type postgres from 'postgres'; +import { pushSchema } from './push-schema.js'; + +/** Last migration version baked into push-schema.ts (baseline, not re-run). */ +export const BASELINE_MIGRATION_VERSION = 21; + +/** First version executed by this runner. */ +export const FIRST_TRACKED_MIGRATION_VERSION = BASELINE_MIGRATION_VERSION + 1; + +const MIGRATION_FILENAME_RE = /^(\d{4})_.+\.sql$/; + +export type MigrationFile = { + version: number; + filename: string; + path: string; +}; + +export function parseMigrationFilename(filename: string): number | null { + const match = MIGRATION_FILENAME_RE.exec(filename); + if (!match) return null; + return Number.parseInt(match[1]!, 10); +} + +export function resolveMigrationsDirectory(entryDir = dirname(fileURLToPath(import.meta.url))): string { + return join(entryDir, 'migrations'); +} + +export async function listTrackedMigrationFiles(migrationsDir: string): Promise { + const entries = await readdir(migrationsDir); + const files: MigrationFile[] = []; + + for (const filename of entries) { + const version = parseMigrationFilename(filename); + if (version === null || version < FIRST_TRACKED_MIGRATION_VERSION) continue; + files.push({ + version, + filename, + path: join(migrationsDir, filename), + }); + } + + files.sort((a, b) => a.version - b.version); + return files; +} + +export async function ensureMigrationLedger(sql: postgres.Sql): Promise { + await sql` + CREATE TABLE IF NOT EXISTS schema_migrations ( + version integer PRIMARY KEY, + filename varchar(255) NOT NULL, + applied_at timestamptz NOT NULL DEFAULT now() + ) + `; +} + +export async function getAppliedMigrationVersions(sql: postgres.Sql): Promise> { + await ensureMigrationLedger(sql); + const rows = await sql<{ version: number }[]>` + SELECT version FROM schema_migrations ORDER BY version + `; + return new Set(rows.map((row) => row.version)); +} + +export async function runPendingSqlMigrations( + sql: postgres.Sql, + migrationsDir: string, +): Promise { + const pending = await listPendingMigrations(sql, migrationsDir); + const applied: string[] = []; + + for (const migration of pending) { + const body = await readFile(migration.path, 'utf8'); + await sql.begin(async (tx) => { + await tx.unsafe(body); + await tx` + INSERT INTO schema_migrations (version, filename) + VALUES (${migration.version}, ${migration.filename}) + `; + }); + applied.push(migration.filename); + console.log(`Applied migration ${migration.filename}`); + } + + return applied; +} + +export async function listPendingMigrations( + sql: postgres.Sql, + migrationsDir: string, +): Promise { + const applied = await getAppliedMigrationVersions(sql); + const files = await listTrackedMigrationFiles(migrationsDir); + return files.filter((file) => !applied.has(file.version)); +} + +/** + * Push legacy baseline schema, then apply tracked SQL migrations (0022+). + */ +export async function runAllMigrations( + databaseUrl: string, + options: { + migrationsDir?: string; + skipPushSchema?: boolean; + } = {}, +): Promise { + const postgresModule = await import('postgres'); + const defaultPostgres = postgresModule.default; + const { postgresOptionsFromEnv } = await import('./postgres-options.js'); + + const migrationsDir = options.migrationsDir ?? resolveMigrationsDirectory(); + + if (!options.skipPushSchema) { + console.log('Pushing baseline schema (migrations 0001–0021 via push-schema)...'); + await pushSchema(databaseUrl); + } + + const sql = defaultPostgres(databaseUrl, postgresOptionsFromEnv()); + + try { + const applied = await runPendingSqlMigrations(sql, migrationsDir); + if (applied.length === 0) { + console.log('No pending SQL migrations.'); + } else { + console.log(`SQL migrations applied: ${applied.join(', ')}`); + } + } finally { + await sql.end(); + } +} diff --git a/packages/database/src/push-schema.ts b/packages/database/src/push-schema.ts index c59fd597..5be679c2 100644 --- a/packages/database/src/push-schema.ts +++ b/packages/database/src/push-schema.ts @@ -2,6 +2,7 @@ * Push schema to database using drizzle-orm's migrate API. * Workaround for drizzle-kit CJS/.js extension issue. */ +import { fileURLToPath } from 'node:url'; import postgres from 'postgres'; import { drizzle } from 'drizzle-orm/postgres-js'; import { sql } from 'drizzle-orm'; @@ -12,8 +13,8 @@ import { postgresOptionsFromEnv } from './postgres-options.js'; const DATABASE_URL = process.env['DATABASE_URL'] ?? 'postgresql://haip:haip@localhost:5432/haip'; -async function main() { - const client = postgres(DATABASE_URL, postgresOptionsFromEnv()); +export async function pushSchema(databaseUrl: string = DATABASE_URL) { + const client = postgres(databaseUrl, postgresOptionsFromEnv()); const db = drizzle(client, { schema }); // Create enums @@ -715,7 +716,6 @@ async function main() { id uuid PRIMARY KEY DEFAULT gen_random_uuid(), property_id uuid NOT NULL REFERENCES properties(id), subscription_id uuid NOT NULL REFERENCES agent_webhook_subscriptions(id), - logical_event_id uuid, event_type varchar(100) NOT NULL, payload jsonb NOT NULL, status webhook_delivery_status NOT NULL DEFAULT 'pending', @@ -1482,7 +1482,13 @@ async function main() { await client.end(); } -main().catch((err) => { - console.error('Push failed:', err); - process.exit(1); -}); +async function main() { + await pushSchema(); +} + +if (process.argv[1] === fileURLToPath(import.meta.url)) { + main().catch((err) => { + console.error('Push failed:', err); + process.exit(1); + }); +} diff --git a/packages/database/src/run-migrations.ts b/packages/database/src/run-migrations.ts new file mode 100644 index 00000000..71884144 --- /dev/null +++ b/packages/database/src/run-migrations.ts @@ -0,0 +1,12 @@ +/** + * HAIP database migrate entrypoint — baseline push-schema + tracked SQL (0022+). + */ +import { runAllMigrations } from './migration-runner.js'; + +const DATABASE_URL = + process.env['DATABASE_URL'] ?? 'postgresql://haip:haip@localhost:5432/haip'; + +runAllMigrations(DATABASE_URL).catch((err) => { + console.error('Migration failed:', err); + process.exit(1); +}); diff --git a/packages/database/tsup.config.ts b/packages/database/tsup.config.ts index 58332f86..d4adcb64 100644 --- a/packages/database/tsup.config.ts +++ b/packages/database/tsup.config.ts @@ -1,11 +1,21 @@ import { defineConfig } from 'tsup'; +import { cpSync } from 'node:fs'; export default defineConfig({ - // push-schema and seed are emitted as runnable scripts so the production - // Docker image (which ships only dist/, no tsx) can migrate+seed via `node`. - entry: ['src/index.ts', 'src/schema/index.ts', 'src/push-schema.ts', 'src/seed.ts'], + // push-schema, run-migrations, and seed are emitted as runnable scripts so the + // production Docker image (which ships only dist/, no tsx) can migrate+seed via `node`. + entry: [ + 'src/index.ts', + 'src/schema/index.ts', + 'src/push-schema.ts', + 'src/run-migrations.ts', + 'src/seed.ts', + ], format: ['esm', 'cjs'], dts: true, clean: true, sourcemap: true, + onSuccess: async () => { + cpSync('src/migrations', 'dist/migrations', { recursive: true }); + }, }); diff --git a/render.yaml b/render.yaml index cbeb6f63..7f454972 100644 --- a/render.yaml +++ b/render.yaml @@ -31,7 +31,7 @@ services: # Push schema + seed the demo hotel before each deploy goes live. # Idempotent (schema IF NOT EXISTS; seed skips if property 'TGH' exists), # so redeploys are safe. Runs in the built image (compiled dist, no tsx). - preDeployCommand: 'node packages/database/dist/push-schema.js && node packages/database/dist/seed.js' + preDeployCommand: 'node packages/database/dist/run-migrations.js && node packages/database/dist/seed.js' envVars: - key: NODE_ENV value: production