diff --git a/package-lock.json b/package-lock.json index da23932..6b72828 100644 --- a/package-lock.json +++ b/package-lock.json @@ -52404,7 +52404,7 @@ "version": "1.1.7", "license": "MIT", "dependencies": { - "@friggframework/core": "^2.0.0-next.88" + "@friggframework/core": "^2.0.0-next.89" }, "devDependencies": { "@friggframework/devtools": "^1.1.2", @@ -52417,9 +52417,9 @@ } }, "packages/v1-ready/hubspot/node_modules/@friggframework/core": { - "version": "2.0.0-next.88", - "resolved": "https://registry.npmjs.org/@friggframework/core/-/core-2.0.0-next.88.tgz", - "integrity": "sha512-DmvG035UJQloD3vEwJJPrH4RIPTXQA603yFY1jzSWKxKCnkQl1E1E6mUcGTBcC9r8QjPTLtj9ZDJlgOWESn8aA==", + "version": "2.0.0-next.89", + "resolved": "https://registry.npmjs.org/@friggframework/core/-/core-2.0.0-next.89.tgz", + "integrity": "sha512-AhxMCO+6aw9RGti6vzo8VOMTD8bufsFPNn+5ZVkMc61XjcGyLMsOUiCeC01ZH33ny4+davhTOYj8zGRKFaXIkQ==", "license": "MIT", "dependencies": { "@aws-sdk/client-apigatewaymanagementapi": "^3.588.0", diff --git a/packages/v1-ready/hubspot/README.md b/packages/v1-ready/hubspot/README.md index 47a334d..c2ad5b2 100644 --- a/packages/v1-ready/hubspot/README.md +++ b/packages/v1-ready/hubspot/README.md @@ -28,10 +28,18 @@ for the full Tier 3 contract. | Portal → integration routing | `findIntegrationByEntityExternalId` via friggCommands | Inbound `portalId` from the payload | | Per-account business logic | Your bound `HUBSPOT_WEBHOOK` handler | Per-portal OAuth credentials (loaded by the worker) | -At request time HubSpot POSTs the whole app's events to the receiver. The receiver -verifies the v3 signature, then resolves each event's `portalId` to the owning Frigg -integration and enqueues a per-event `HUBSPOT_WEBHOOK` job. The queue worker hydrates -that integration with its own per-portal credentials and runs your handler. +At request time HubSpot POSTs the whole app's events to the receiver. The receiver is +**DB-free** (`useDatabase: false`): it verifies the v3 signature and enqueues one +`HUBSPOT_WEBHOOK_RESOLVE` job per event — no database access. In the queue worker (which +has the database), the resolve step looks up each event's `portalId` → owning integration +and re-enqueues a `HUBSPOT_WEBHOOK` job bound to that integration id. The worker hydrates +that integration with its own per-portal credentials and runs your handler. Keeping the +lookup in the worker is what lets the public receiver endpoint stay DB-free. + +> **Make your `HUBSPOT_WEBHOOK` handler idempotent.** Delivery is at-least-once: the +> two SQS hops (resolve → webhook) and HubSpot's own retry-on-non-2xx mean a given +> event can be delivered more than once. Key your processing on a stable identifier +> (e.g. `objectId` + `subscriptionType` + `occurredAt`) so duplicates are no-ops. ### Enabling it on an integration @@ -71,9 +79,19 @@ class HubSpotIntegration extends IntegrationBase { } ``` -The framework auto-mounts the receiver route at the integration's base path -(`POST //webhooks`) — register that URL as the app's webhook target -in your HubSpot app settings. +The framework mounts the receiver route **under your binding key**, on its own +DB-free Lambda function: + +``` +POST /api/-integration//webhooks +``` + +So the binding `hubspotWebhooks` above yields +`POST /api/hubspot-integration/hubspotWebhooks/webhooks`. Pick a clean binding key +(e.g. `hubspot` → `.../hubspot/webhooks`) and register that URL as the app's webhook +target in your HubSpot app settings. Namespacing by the binding key means a second +module's webhooks extension (e.g. Clockwork) can live on the same integration without +colliding. ### Configuration @@ -83,12 +101,15 @@ in your HubSpot app settings. ### What the bundle contributes -- **Route:** `POST /webhooks` → `HUBSPOT_WEBHOOK_RECEIVED` -- **`HUBSPOT_WEBHOOK_RECEIVED`** — default receiver: verifies the signature, resolves - each event's `portalId`, and queues matched events. Events whose portal does not map - to any integration are skipped (HubSpot broadcasts every portal's events to the app). - Events are resolved and enqueued in parallel; if a `portalId` resolves ambiguously - (one external ID owned by multiple integrations) the whole batch is rejected rather - than risk cross-tenant routing. -- **`HUBSPOT_WEBHOOK`** — default no-op; override it via `binding.handlers.HUBSPOT_WEBHOOK` - to run your per-event logic. +- `useDatabase: false` — the receiver route runs without a DB connection. +- **Route:** `POST /webhooks` (namespaced to `/{bindingKey}/webhooks`) → `HUBSPOT_WEBHOOK_RECEIVED`. +- **`HUBSPOT_WEBHOOK_RECEIVED`** — DB-free receiver: verifies the v3 signature and + enqueues one `HUBSPOT_WEBHOOK_RESOLVE` per event (in parallel). Events missing a + `portalId` are skipped. No database access. +- **`HUBSPOT_WEBHOOK_RESOLVE`** — queue event (worker, DB): reverse-looks up + `portalId` → integration id and re-enqueues a `HUBSPOT_WEBHOOK` bound to it. Events + whose portal maps to no integration are skipped; an ambiguous resolution (one external + id owned by multiple integrations) throws rather than risk cross-tenant routing. +- **`HUBSPOT_WEBHOOK`** — queue event (worker, hydrated integration): default no-op; + override via `binding.handlers.HUBSPOT_WEBHOOK` to run your per-event logic with the + correct per-account context. diff --git a/packages/v1-ready/hubspot/extensions/webhooks/handlers.js b/packages/v1-ready/hubspot/extensions/webhooks/handlers.js index eb711a9..902eec7 100644 --- a/packages/v1-ready/hubspot/extensions/webhooks/handlers.js +++ b/packages/v1-ready/hubspot/extensions/webhooks/handlers.js @@ -2,25 +2,11 @@ const { verifyHubSpotSignature } = require('./signature-verifier'); const { findIntegrationByPortalId } = require('./lookup'); /** - * Receiver handler for `POST /webhooks`. - * - * Verifies HubSpot's v3 signature using `process.env.HUBSPOT_CLIENT_SECRET` - * (the same env var the api-module already reads for OAuth), iterates the - * inbound batch, resolves each event's `portalId` to a Frigg integration via - * the platform-neutral reverse lookup, and enqueues a per-event - * `HUBSPOT_WEBHOOK` job for the matched integration. Events whose portal - * does not map to any integration are silently skipped (HubSpot sends events - * for the whole app, not per-account). - * - * Per the Tier 3 contract, this function is bound as a plain function on the - * integration instance — `this` is the IntegrationBase instance and exposes - * `commands.findIntegrationByEntityExternalId` and `queueWebhook`. - * - * @this {import('@friggframework/core').IntegrationBase} - * @param {Object} args - * @param {import('express').Request} args.req - * @param {import('express').Response} args.res - * @returns {Promise} + * Receiver for `POST /webhooks` — DB-free. Verifies the v3 signature, then + * enqueues one HUBSPOT_WEBHOOK_RESOLVE per event (carrying the raw payload incl. + * portalId). It does NOT resolve the owning integration — that needs the DB and + * happens in the worker, which is what lets the extension declare + * useDatabase: false. Events missing a portalId are skipped. */ async function onHubSpotWebhookReceived({ req, res }) { const verification = verifyHubSpotSignature({ @@ -37,14 +23,8 @@ async function onHubSpotWebhookReceived({ req, res }) { const events = Array.isArray(req.body) ? req.body : []; - // Phase 1 — resolve every portalId in parallel before queueing anything. - // Two reasons for two-phase: (a) HubSpot batches can be large and each - // lookup is an independent DB round-trip; running them serially blows - // the HubSpot response budget; (b) if any lookup is ambiguous, the core - // command throws by design — we want that throw to land BEFORE any - // queueWebhook fires so we never leave the batch in a partial-enqueue - // state that HubSpot's retry would then duplicate. - const resolutions = await Promise.all( + // Parallel: each is an independent SQS send and HubSpot batches can be large. + const outcomes = await Promise.all( events.map(async (evt, i) => { const portalId = evt && evt.portalId; if (portalId === undefined || portalId === null) { @@ -52,54 +32,66 @@ async function onHubSpotWebhookReceived({ req, res }) { `[hubspot-webhooks] event[${i}] missing portalId ` + `(subscriptionType=${evt && evt.subscriptionType}); skipping` ); - return null; + return 'skipped'; } - const integrationId = await findIntegrationByPortalId( - this, - portalId - ); - if (!integrationId) return null; - return { integrationId, evt }; - }) - ); - - // Phase 2 — enqueue matched events in parallel. Every match resolved - // cleanly above, so any failure here is genuinely SQS-side and should - // surface to HubSpot for retry. - const matches = resolutions.filter(Boolean); - await Promise.all( - matches.map(({ integrationId, evt }) => - this.queueWebhook({ - integrationId, + await this.queueWebhook({ + event: 'HUBSPOT_WEBHOOK_RESOLVE', body: evt, - event: 'HUBSPOT_WEBHOOK', - }) - ) + }); + return 'queued'; + }) ); + const queued = outcomes.filter((o) => o === 'queued').length; res.status(200).json({ received: events.length, - queued: matches.length, - skipped: events.length - matches.length, + queued, + skipped: events.length - queued, + }); +} + +/** + * HUBSPOT_WEBHOOK_RESOLVE — runs in the queue worker (DB available) on a dry + * integration instance. Reverse-looks up portalId → owning integration, then + * re-enqueues HUBSPOT_WEBHOOK bound to that id (the worker hydrates the real + * record on that hop). Keeping the lookup here is what lets the receiver stay + * DB-free. Ambiguous resolution propagates rather than risk cross-tenant misrouting. + */ +async function onHubSpotWebhookResolve({ data }) { + const body = data && data.body; + const portalId = body && body.portalId; + if (portalId === undefined || portalId === null) { + console.warn( + '[hubspot-webhooks] resolve: event missing portalId; skipping' + ); + return; + } + + const integrationId = await findIntegrationByPortalId(this, portalId); + if (!integrationId) { + console.warn( + `[hubspot-webhooks] resolve: no integration for portalId=${portalId}; skipping` + ); + return; + } + + await this.queueWebhook({ + event: 'HUBSPOT_WEBHOOK', + integrationId, + body, }); } /** - * Default per-event handler. Integration consumers override this via - * `binding.handlers.HUBSPOT_WEBHOOK = 'methodName'`; the default is a no-op - * so a misconfigured binding fails predictably rather than crashing the - * queue worker. - * - * @this {import('@friggframework/core').IntegrationBase} - * @param {Object} args - * @param {Object} args.data - * @returns {Promise} + * HUBSPOT_WEBHOOK — runs in the worker with the owning integration hydrated. + * Default no-op; consumers override via binding.handlers.HUBSPOT_WEBHOOK. */ async function onHubSpotWebhook({ data }) { - // intentional no-op — override via binding.handlers.HUBSPOT_WEBHOOK + // no-op — override via binding.handlers.HUBSPOT_WEBHOOK } module.exports = { onHubSpotWebhookReceived, + onHubSpotWebhookResolve, onHubSpotWebhook, }; diff --git a/packages/v1-ready/hubspot/extensions/webhooks/index.js b/packages/v1-ready/hubspot/extensions/webhooks/index.js index ed1cb07..9ea0e6e 100644 --- a/packages/v1-ready/hubspot/extensions/webhooks/index.js +++ b/packages/v1-ready/hubspot/extensions/webhooks/index.js @@ -1,24 +1,23 @@ const { onHubSpotWebhookReceived, + onHubSpotWebhookResolve, onHubSpotWebhook, } = require('./handlers'); /** * HubSpot Webhooks — Tier 3 Integration Extension bundle. * - * Contributes a single receiver route (`POST /webhooks`) plus two events: + * DB-free receiver (`useDatabase: false`): HUBSPOT_WEBHOOK_RECEIVED verifies the + * v3 signature and enqueues a resolve job. The portal→integration lookup needs + * the DB, so it runs in the worker (HUBSPOT_WEBHOOK_RESOLVE), which re-enqueues + * HUBSPOT_WEBHOOK bound to the resolved integration. Consumers override + * HUBSPOT_WEBHOOK via binding.handlers. * - * HUBSPOT_WEBHOOK_RECEIVED — bound to the route; verifies signature, - * resolves portalId → integrationId, queues - * one HUBSPOT_WEBHOOK per matched event. - * HUBSPOT_WEBHOOK — default no-op; integrations override via - * `binding.handlers.HUBSPOT_WEBHOOK`. - * - * See the binding contract at: - * https://github.com/friggframework/frigg/blob/next/packages/core/integrations/EXTENSIONS.md + * Contract: https://github.com/friggframework/frigg/blob/next/packages/core/integrations/EXTENSIONS.md */ module.exports = { name: 'hubspot-webhooks', + useDatabase: false, routes: [ { path: '/webhooks', @@ -31,6 +30,10 @@ module.exports = { type: 'LIFE_CYCLE_EVENT', handler: onHubSpotWebhookReceived, }, + HUBSPOT_WEBHOOK_RESOLVE: { + type: 'LIFE_CYCLE_EVENT', + handler: onHubSpotWebhookResolve, + }, HUBSPOT_WEBHOOK: { type: 'LIFE_CYCLE_EVENT', handler: onHubSpotWebhook, diff --git a/packages/v1-ready/hubspot/package.json b/packages/v1-ready/hubspot/package.json index 27c4e8f..19db157 100644 --- a/packages/v1-ready/hubspot/package.json +++ b/packages/v1-ready/hubspot/package.json @@ -20,6 +20,6 @@ "prettier": "^2.7.1" }, "dependencies": { - "@friggframework/core": "^2.0.0-next.88" + "@friggframework/core": "^2.0.0-next.89" } } diff --git a/packages/v1-ready/hubspot/tests/extensions/webhooks.test.js b/packages/v1-ready/hubspot/tests/extensions/webhooks.test.js index 55f44a7..ece2cd6 100644 --- a/packages/v1-ready/hubspot/tests/extensions/webhooks.test.js +++ b/packages/v1-ready/hubspot/tests/extensions/webhooks.test.js @@ -11,6 +11,7 @@ const { } = require('../../extensions/webhooks/lookup'); const { onHubSpotWebhookReceived, + onHubSpotWebhookResolve, onHubSpotWebhook, } = require('../../extensions/webhooks/handlers'); @@ -86,16 +87,23 @@ describe('hubspot-webhooks extension bundle shape', () => { }); }); + it('declares useDatabase: false so the receiver route is DB-free', () => { + expect(webhooksExtension.useDatabase).toBe(false); + }); + it('every route event references a declared event', () => { for (const route of webhooksExtension.routes) { expect(webhooksExtension.events).toHaveProperty(route.event); } }); - it('declares both HUBSPOT_WEBHOOK_RECEIVED and HUBSPOT_WEBHOOK with function handlers', () => { - expect(typeof webhooksExtension.events.HUBSPOT_WEBHOOK_RECEIVED.handler).toBe( - 'function' - ); + it('declares the receiver, resolve, and webhook events with function handlers', () => { + expect( + typeof webhooksExtension.events.HUBSPOT_WEBHOOK_RECEIVED.handler + ).toBe('function'); + expect( + typeof webhooksExtension.events.HUBSPOT_WEBHOOK_RESOLVE.handler + ).toBe('function'); expect(typeof webhooksExtension.events.HUBSPOT_WEBHOOK.handler).toBe( 'function' ); @@ -335,7 +343,7 @@ describe('findIntegrationByPortalId wrapper', () => { }); }); -describe('onHubSpotWebhookReceived (default receiver handler)', () => { +describe('onHubSpotWebhookReceived (DB-free receiver)', () => { let previousClientSecret; beforeAll(() => { @@ -351,24 +359,16 @@ describe('onHubSpotWebhookReceived (default receiver handler)', () => { } }); - const makeIntegration = (overrides = {}) => { - const commandsOverride = overrides.commands; - delete overrides.commands; - const integration = { - commands: { - findIntegrationByEntityExternalId: jest.fn(async (portalId) => { - if (portalId === 999) return null; // simulate unknown portal - return `integration-for-portal-${portalId}`; - }), - ...commandsOverride, - }, - queueWebhook: jest.fn().mockResolvedValue(undefined), - ...overrides, - }; - return integration; - }; + // The receiver must NOT touch the database. We give it a commands spy so we + // can assert it is never called, plus a queueWebhook spy. + const makeIntegration = () => ({ + commands: { + findIntegrationByEntityExternalId: jest.fn(), + }, + queueWebhook: jest.fn().mockResolvedValue(undefined), + }); - it('returns 401 on invalid signature', async () => { + it('returns 401 on invalid signature and enqueues nothing', async () => { const integration = makeIntegration(); const req = buildSignedRequest({ overrideSignature: 'AAAA' }); const res = makeRes(); @@ -377,16 +377,32 @@ describe('onHubSpotWebhookReceived (default receiver handler)', () => { expect(integration.queueWebhook).not.toHaveBeenCalled(); }); - it('returns 401 on missing signature header', async () => { + it('returns 401 when the signature header is missing', async () => { const integration = makeIntegration(); const req = buildSignedRequest(); delete req.headers['x-hubspot-signature-v3']; const res = makeRes(); await onHubSpotWebhookReceived.call(integration, { req, res }); expect(res.statusCode).toBe(401); + expect(integration.queueWebhook).not.toHaveBeenCalled(); }); - it('iterates events, queues each matched event, and returns 200 with counts', async () => { + it('returns 401 when HUBSPOT_CLIENT_SECRET is not set', async () => { + const saved = process.env.HUBSPOT_CLIENT_SECRET; + delete process.env.HUBSPOT_CLIENT_SECRET; + try { + const integration = makeIntegration(); + const req = buildSignedRequest(); + const res = makeRes(); + await onHubSpotWebhookReceived.call(integration, { req, res }); + expect(res.statusCode).toBe(401); + expect(integration.queueWebhook).not.toHaveBeenCalled(); + } finally { + process.env.HUBSPOT_CLIENT_SECRET = saved; + } + }); + + it('enqueues one HUBSPOT_WEBHOOK_RESOLVE per event WITHOUT any DB lookup', async () => { const integration = makeIntegration(); const body = [ { portalId: 111, subscriptionType: 'contact.creation', objectId: 1 }, @@ -397,54 +413,25 @@ describe('onHubSpotWebhookReceived (default receiver handler)', () => { await onHubSpotWebhookReceived.call(integration, { req, res }); - expect(integration.commands.findIntegrationByEntityExternalId).toHaveBeenCalledTimes( - 2 - ); - expect(integration.commands.findIntegrationByEntityExternalId).toHaveBeenCalledWith( - 111, - 'hubspot' - ); - expect(integration.commands.findIntegrationByEntityExternalId).toHaveBeenCalledWith( - 222, - 'hubspot' - ); + // The receiver is DB-free: it must never perform the portal lookup. + expect( + integration.commands.findIntegrationByEntityExternalId + ).not.toHaveBeenCalled(); + expect(integration.queueWebhook).toHaveBeenCalledTimes(2); expect(integration.queueWebhook).toHaveBeenCalledWith({ - integrationId: 'integration-for-portal-111', + event: 'HUBSPOT_WEBHOOK_RESOLVE', body: body[0], - event: 'HUBSPOT_WEBHOOK', }); expect(integration.queueWebhook).toHaveBeenCalledWith({ - integrationId: 'integration-for-portal-222', + event: 'HUBSPOT_WEBHOOK_RESOLVE', body: body[1], - event: 'HUBSPOT_WEBHOOK', }); expect(res.statusCode).toBe(200); expect(res.body).toEqual({ received: 2, queued: 2, skipped: 0 }); }); - it('skips events whose portalId does not resolve to an integration', async () => { - const integration = makeIntegration(); - const body = [ - { portalId: 111, subscriptionType: 'contact.creation', objectId: 1 }, - { portalId: 999, subscriptionType: 'contact.creation', objectId: 2 }, // null lookup - ]; - const req = buildSignedRequest({ body }); - const res = makeRes(); - - await onHubSpotWebhookReceived.call(integration, { req, res }); - - expect(integration.queueWebhook).toHaveBeenCalledTimes(1); - expect(integration.queueWebhook).toHaveBeenCalledWith({ - integrationId: 'integration-for-portal-111', - body: body[0], - event: 'HUBSPOT_WEBHOOK', - }); - expect(res.statusCode).toBe(200); - expect(res.body).toEqual({ received: 2, queued: 1, skipped: 1 }); - }); - - it('skips events missing a portalId entirely', async () => { + it('skips events missing a portalId (nothing to resolve later)', async () => { const integration = makeIntegration(); const body = [ { subscriptionType: 'contact.creation', objectId: 1 }, // no portalId @@ -455,11 +442,11 @@ describe('onHubSpotWebhookReceived (default receiver handler)', () => { await onHubSpotWebhookReceived.call(integration, { req, res }); - expect(integration.commands.findIntegrationByEntityExternalId).toHaveBeenCalledTimes( - 1 - ); expect(integration.queueWebhook).toHaveBeenCalledTimes(1); - expect(res.statusCode).toBe(200); + expect(integration.queueWebhook).toHaveBeenCalledWith({ + event: 'HUBSPOT_WEBHOOK_RESOLVE', + body: body[1], + }); expect(res.body).toEqual({ received: 2, queued: 1, skipped: 1 }); }); @@ -472,112 +459,78 @@ describe('onHubSpotWebhookReceived (default receiver handler)', () => { expect(res.body).toEqual({ received: 0, queued: 0, skipped: 0 }); expect(integration.queueWebhook).not.toHaveBeenCalled(); }); +}); + +describe('onHubSpotWebhookResolve (queue worker, DB)', () => { + const makeIntegration = (lookupImpl) => ({ + commands: { + findIntegrationByEntityExternalId: jest.fn( + lookupImpl || + (async (portalId) => + portalId === 999 + ? null + : `integration-for-portal-${portalId}`) + ), + }, + queueWebhook: jest.fn().mockResolvedValue(undefined), + }); - it('propagates ambiguous-resolution errors instead of catching them', async () => { - const integration = makeIntegration({ - commands: { - findIntegrationByEntityExternalId: jest - .fn() - .mockRejectedValue(new Error('ambiguous resolution')), - }, + it('resolves portalId → integrationId and re-enqueues HUBSPOT_WEBHOOK bound to it', async () => { + const integration = makeIntegration(); + const body = { portalId: 111, subscriptionType: 'contact.creation' }; + + await onHubSpotWebhookResolve.call(integration, { data: { body } }); + + expect( + integration.commands.findIntegrationByEntityExternalId + ).toHaveBeenCalledWith(111, 'hubspot'); + expect(integration.queueWebhook).toHaveBeenCalledTimes(1); + expect(integration.queueWebhook).toHaveBeenCalledWith({ + event: 'HUBSPOT_WEBHOOK', + integrationId: 'integration-for-portal-111', + body, }); - const req = buildSignedRequest({ body: [{ portalId: 111 }] }); - const res = makeRes(); - await expect( - onHubSpotWebhookReceived.call(integration, { req, res }) - ).rejects.toThrow(/ambiguous/); - expect(integration.queueWebhook).not.toHaveBeenCalled(); }); - it('queues zero events when any one resolution is ambiguous (all-or-nothing)', async () => { - // Multi-event batch where event[1] resolves ambiguous. With the - // two-phase implementation, NO queueWebhook calls should fire for - // any event — including event[0] which would have resolved cleanly. - // Prevents partial-enqueue + HubSpot-retry-duplication. - const integration = makeIntegration({ - commands: { - findIntegrationByEntityExternalId: jest.fn(async (portalId) => { - if (portalId === 222) - throw new Error('ambiguous resolution'); - return `integration-for-portal-${portalId}`; - }), - }, - }); - const req = buildSignedRequest({ - body: [ - { portalId: 111, subscriptionType: 'contact.creation' }, - { portalId: 222, subscriptionType: 'contact.creation' }, - { portalId: 333, subscriptionType: 'contact.creation' }, - ], + it('skips (does not re-enqueue) when no integration owns the portal', async () => { + const integration = makeIntegration(); + await onHubSpotWebhookResolve.call(integration, { + data: { body: { portalId: 999 } }, }); - const res = makeRes(); - await expect( - onHubSpotWebhookReceived.call(integration, { req, res }) - ).rejects.toThrow(/ambiguous/); expect(integration.queueWebhook).not.toHaveBeenCalled(); }); - it('resolves and queues events in parallel rather than serially', async () => { - // Each lookup takes a tick; if dispatch were sequential, the second - // lookup would only start after the first lookup AND the first - // queueWebhook had resolved. We assert all three lookups have begun - // before any of them complete. - const inFlightLookups = jest.fn(); - const inFlightQueueWrites = jest.fn(); - let outstandingLookups = 0; - let outstandingQueueWrites = 0; - const integration = makeIntegration({ - commands: { - findIntegrationByEntityExternalId: jest.fn(async (portalId) => { - outstandingLookups += 1; - inFlightLookups(outstandingLookups); - await new Promise((resolve) => setImmediate(resolve)); - outstandingLookups -= 1; - return `integration-for-portal-${portalId}`; - }), - }, - queueWebhook: jest.fn(async () => { - outstandingQueueWrites += 1; - inFlightQueueWrites(outstandingQueueWrites); - await new Promise((resolve) => setImmediate(resolve)); - outstandingQueueWrites -= 1; - }), - }); - const req = buildSignedRequest({ - body: [ - { portalId: 111, subscriptionType: 'contact.creation' }, - { portalId: 222, subscriptionType: 'contact.creation' }, - { portalId: 333, subscriptionType: 'contact.creation' }, - ], + it('skips when the queued body has no portalId', async () => { + const integration = makeIntegration(); + await onHubSpotWebhookResolve.call(integration, { + data: { body: { subscriptionType: 'contact.creation' } }, }); - const res = makeRes(); - await onHubSpotWebhookReceived.call(integration, { req, res }); - - // At some point all three lookups (and all three queue writes) - // should have been in flight simultaneously — that's what - // distinguishes parallel from serial dispatch. - expect(Math.max(...inFlightLookups.mock.calls.map((c) => c[0]))).toBe( - 3 - ); expect( - Math.max(...inFlightQueueWrites.mock.calls.map((c) => c[0])) - ).toBe(3); - expect(res.body).toEqual({ received: 3, queued: 3, skipped: 0 }); + integration.commands.findIntegrationByEntityExternalId + ).not.toHaveBeenCalled(); + expect(integration.queueWebhook).not.toHaveBeenCalled(); }); - it('rejects with 401 when HUBSPOT_CLIENT_SECRET is not set', async () => { - const saved = process.env.HUBSPOT_CLIENT_SECRET; - delete process.env.HUBSPOT_CLIENT_SECRET; - try { - const integration = makeIntegration(); - const req = buildSignedRequest(); - const res = makeRes(); - await onHubSpotWebhookReceived.call(integration, { req, res }); - expect(res.statusCode).toBe(401); - expect(integration.queueWebhook).not.toHaveBeenCalled(); - } finally { - process.env.HUBSPOT_CLIENT_SECRET = saved; - } + it('propagates ambiguous-resolution errors instead of swallowing them', async () => { + const integration = makeIntegration(async () => { + throw new Error('ambiguous resolution'); + }); + await expect( + onHubSpotWebhookResolve.call(integration, { + data: { body: { portalId: 111 } }, + }) + ).rejects.toThrow(/ambiguous/); + expect(integration.queueWebhook).not.toHaveBeenCalled(); + }); + + it('throws a clear error when the integration has no commands wired (createFriggCommands missing)', async () => { + const integration = { queueWebhook: jest.fn() }; // no `commands` + await expect( + onHubSpotWebhookResolve.call(integration, { + data: { body: { portalId: 111 } }, + }) + ).rejects.toThrow(/commands\.findIntegrationByEntityExternalId/); + expect(integration.queueWebhook).not.toHaveBeenCalled(); }); });