From 927dfa9bafdc85ec8b485c2249fb639e75f57b98 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 7 Sep 2026 15:27:54 +0000 Subject: [PATCH 1/3] fix(plugin-webhooks): match webhook subscriptions on the organization dimension (#13566) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `AutoEnqueuer.handleEvent` / `handleBulkEvent` selected subscriptions by object name and trigger only, from one cache holding every organization's `sys_webhook` rows — so on a walled deployment organization A's record events reached organization B's endpoint, signed with B's secret. Both fan-out paths now compare the subscription's own organization (`CachedSubscription.organizationId`, #13546) with the organization the producer stamped on the event (`DataEvent.organizationId`, #14970; `BulkDataEvent.organizationId`, #15225 / #15813): one equality per candidate, no lookup on the hot path. A subscription with no organization ownership does not receive an organization-walled event (loud refusal, said once per subscription); an organization-owned subscription receives only its own organization's events and is fail-closed on an event that names none, on both paths. Nothing stamps either side on a `single` posture, so delivery there is unchanged. A present-but-off-contract `organizationId` drops the event loudly, delivering to nobody. Pins assert on which subscriptions the enqueuer selected (the enqueue seam), never on delivery rows — #13565 stamps a delivery with the SUBSCRIPTION's organization, so a leaked delivery reads as natively owned by the receiver. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012zTkyNHJ7TkuN2oXtP5x37 --- .../plugin-webhooks/src/auto-enqueuer.test.ts | 346 ++++++++++++++++++ .../plugin-webhooks/src/auto-enqueuer.ts | 215 ++++++++++- 2 files changed, 558 insertions(+), 3 deletions(-) diff --git a/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts b/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts index ac327ddd6d..27e3e2e1e3 100644 --- a/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts +++ b/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts @@ -154,6 +154,10 @@ function event( object: string, record: any, timestamp = '2026-05-24T00:00:00.000Z', + // [#13566] The RECORD's organization, the way the engine stamps it + // (#14970). Omitted = the spec's one spelling for "belongs to no + // organization"; the schema refuses the empty string. + extra: { organizationId?: string } = {}, ): RealtimeEventPayload { const payload = DataEventSchema.parse({ id: randomUUID(), @@ -161,6 +165,7 @@ function event( object, recordId: String(record.id), ...(type === 'deleted' ? {} : { after: record }), + ...(extra.organizationId !== undefined ? { organizationId: extra.organizationId } : {}), timestamp, }); return { type: payload.type, object, payload: { ...payload }, timestamp }; @@ -177,11 +182,16 @@ function bulkEvent( object: string, matched: number, timestamp = '2026-05-24T00:00:00.000Z', + // [#13566] The ONE organization the tenant wall named for the batch, the + // way the engine stamps it (#15225 / #15813). Omitted = "the producer did + // not assert one organization for the batch" — a routine value there. + extra: { organizationId?: string } = {}, ): RealtimeEventPayload { const payload = BulkDataEventSchema.parse({ id: randomUUID(), type: `data.records.${type}`, object, + ...(extra.organizationId !== undefined ? { organizationId: extra.organizationId } : {}), matched, timestamp, }); @@ -748,3 +758,339 @@ describe('AutoEnqueuer — bulk data events (#4639)', () => { await ae.stop(); }); }); + +/** + * #13566 — the organization dimension of the match. + * + * On a walled deployment (`OS_TENANCY_POSTURE=isolated|group`) every + * organization's `sys_webhook` rows sit in ONE cache keyed by object name, and + * the producers now stamp the event with the organization the record belongs + * to (#14970) or the one the tenant wall named for the batch (#15225 / + * #15813). Matching on object name alone delivered organization A's records + * to organization B's endpoint, signed with B's secret. + * + * ⭐ Every pin here asserts on WHICH SUBSCRIPTIONS THE ENQUEUER SELECTED — + * the `refId`s handed to the enqueue seam — never on delivery rows. #13565 + * stamps each delivery with the SUBSCRIPTION's organization, so a leaked + * delivery reads as natively owned by the receiver while carrying the + * sender's payload: a test over `sys_http_delivery` rows passes on a live + * leak. + */ +describe('AutoEnqueuer — organization dimension (#13566)', () => { + const selected = (calls: EnqueueHttpInput[]) => calls.map((c) => c.refId).sort(); + + describe('per-record path (data.record.*)', () => { + it("fans out ONLY to the subscription of the organization the record belongs to", async () => { + // The leak pin. Two organizations, each with its own webhook on + // `contact`; before the fix both received both organizations' + // records. + const engine = new FakeEngine({ + sys_webhook: [ + webhook({ id: 'wh-a', name: 'a', organization_id: 'org_a' }), + webhook({ id: 'wh-b', name: 'b', organization_id: 'org_b' }), + ], + }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish(event('created', 'contact', { id: 'c-a' }, undefined, { organizationId: 'org_a' })); + await flush(); + expect(selected(calls)).toEqual(['wh-a']); + + await realtime.publish( + event('updated', 'contact', { id: 'c-b' }, '2026-05-24T00:00:01.000Z', { organizationId: 'org_b' }), + ); + await flush(); + expect(selected(calls)).toEqual(['wh-a', 'wh-b']); + expect(calls.find((c) => c.refId === 'wh-b')!.organizationId).toBe('org_b'); + expect((calls.find((c) => c.refId === 'wh-b')!.payload as any).recordId).toBe('c-b'); + // A foreign organization's subscription is simply not a candidate + // — the match term working, not a refusal worth a warning. + expect(warn).not.toHaveBeenCalled(); + await ae.stop(); + }); + + it('an org-less subscription does NOT receive an organization-walled record event — loud, once', async () => { + // The ruling's case, verbatim: a subscription with no organisation + // ownership does not fan out — loud refusal, never a silent + // cross-organisation delivery. + const engine = new FakeEngine({ sys_webhook: [webhook()] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const debug = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { + refreshIntervalMs: 0, + logger: { warn, debug }, + }); + await ae.start(); + + await realtime.publish(event('created', 'contact', { id: 'c-1' }, undefined, { organizationId: 'org_a' })); + await flush(); + expect(selected(calls)).toEqual([]); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining('belongs to NO organization'), + expect.objectContaining({ id: 'wh-1', type: 'data.record.created', object: 'contact' }), + ); + expect(String(warn.mock.calls[0][0])).toContain('#13566'); + + // Said once per subscription: the next refused event, from another + // organization even, is debug-level. + await realtime.publish( + event('created', 'contact', { id: 'c-2' }, '2026-05-24T00:00:01.000Z', { organizationId: 'org_b' }), + ); + await flush(); + expect(selected(calls)).toEqual([]); + expect(warn).toHaveBeenCalledTimes(1); + expect(debug).toHaveBeenCalledWith(expect.stringContaining('still refused'), expect.objectContaining({ id: 'wh-1' })); + await ae.stop(); + }); + + it('an organization-owned subscription does NOT receive a record event that names no organization (fail-closed)', async () => { + // Absent on a DataEvent is "belongs to no organization" — an + // environment-wide row, an object outside the wall, or the producer + // publishing absent rather than substituting the caller's org when it + // had no row in hand. None of those names organization A. + const engine = new FakeEngine({ sys_webhook: [webhook({ organization_id: 'org_a' })] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish(event('created', 'contact', { id: 'c-1' })); + await flush(); + expect(selected(calls)).toEqual([]); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining('names no organization'), + expect.objectContaining({ id: 'wh-1', subscriptionOrganizationId: 'org_a', eventNamesOrganization: false }), + ); + + // …and its own organization's record still arrives (the positive + // control on the same subscription). + await realtime.publish( + event('created', 'contact', { id: 'c-2' }, '2026-05-24T00:00:01.000Z', { organizationId: 'org_a' }), + ); + await flush(); + expect(selected(calls)).toEqual(['wh-1']); + await ae.stop(); + }); + + it('an org-less subscription still receives an org-less record event (the single-posture control)', async () => { + // A `single`-posture deployment stamps nothing on either side — + // every webhook on every non-walled install lives in this cell. + const engine = new FakeEngine({ sys_webhook: [webhook()] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish(event('created', 'contact', { id: 'c-1' })); + await flush(); + expect(selected(calls)).toEqual(['wh-1']); + expect(calls[0].organizationId).toBeUndefined(); + expect(warn).not.toHaveBeenCalled(); + await ae.stop(); + }); + + it("an any-object ('*') subscription goes through the same organization filter", async () => { + const engine = new FakeEngine({ + sys_webhook: [ + webhook({ id: 'wh-star-orgless', name: 'star-orgless', object_name: '' }), + webhook({ id: 'wh-star-a', name: 'star-a', object_name: '', organization_id: 'org_a' }), + webhook({ id: 'wh-star-b', name: 'star-b', object_name: '', organization_id: 'org_b' }), + ], + }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish(event('created', 'lead', { id: 'l-1' }, undefined, { organizationId: 'org_a' })); + await flush(); + expect(selected(calls)).toEqual(['wh-star-a']); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ id: 'wh-star-orgless' })); + await ae.stop(); + }); + + it('drops a record event whose organizationId is present but off-contract, delivering to nobody', async () => { + const engine = new FakeEngine({ + sys_webhook: [webhook({ id: 'wh-orgless' }), webhook({ id: 'wh-a', organization_id: 'org_a' })], + }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + // The schema refuses '' at the publish site; a producer that did not + // validate is broken, and the event is dropped loudly — never + // coerced, never read as "no organization". + await realtime.publish({ + type: 'data.record.created', + object: 'contact', + payload: { recordId: 'c-1', organizationId: '' }, + timestamp: '2026-05-24T00:00:00.000Z', + }); + await flush(); + expect(selected(calls)).toEqual([]); + expect(warn).toHaveBeenCalledTimes(1); + expect(String(warn.mock.calls[0][0])).toContain('off-contract'); + expect(String(warn.mock.calls[0][0])).toContain('organizationId'); + await ae.stop(); + }); + }); + + describe('bulk path (data.records.*)', () => { + it('fans out ONLY to the subscription of the organization the tenant wall named for the batch', async () => { + const engine = new FakeEngine({ + sys_webhook: [ + webhook({ id: 'wh-a', name: 'a', triggers: 'bulk_update', organization_id: 'org_a' }), + webhook({ id: 'wh-b', name: 'b', triggers: 'bulk_update', organization_id: 'org_b' }), + ], + }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish(bulkEvent('updated', 'contact', 12, undefined, { organizationId: 'org_a' })); + await flush(); + expect(selected(calls)).toEqual(['wh-a']); + expect((calls[0].payload as any).matched).toBe(12); + expect(warn).not.toHaveBeenCalled(); + await ae.stop(); + }); + + it('an organization-owned subscription does NOT receive a bulk event the producer could not attribute (absent = fail-closed)', async () => { + // On the bulk path absent is a ROUTINE value: the producer stamps + // the key only when the Layer 0 wall named exactly one organization + // (#15687). A system sweep or a cross-membership `group` write + // publishes it absent, and the contract says a tenant-scoped + // consumer must not deliver that inside an organization wall. + const engine = new FakeEngine({ + sys_webhook: [webhook({ triggers: 'bulk_update,bulk_delete', organization_id: 'org_a' })], + }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish(bulkEvent('updated', 'contact', 40)); + await realtime.publish(bulkEvent('deleted', 'contact', 3)); + await flush(); + expect(selected(calls)).toEqual([]); + expect(warn).toHaveBeenCalledTimes(1); // said once, not once per event + expect(warn).toHaveBeenCalledWith( + expect.stringContaining('names no organization'), + expect.objectContaining({ id: 'wh-1', type: 'data.records.updated', eventNamesOrganization: false }), + ); + + // Positive control on the same subscription: a batch the wall + // attributed to its organization is delivered. + await realtime.publish(bulkEvent('updated', 'contact', 5, undefined, { organizationId: 'org_a' })); + await flush(); + expect(selected(calls)).toEqual(['wh-1']); + await ae.stop(); + }); + + it('an org-less subscription receives an unattributed bulk event (the deployment-wide consumer the contract names)', async () => { + const engine = new FakeEngine({ sys_webhook: [webhook({ triggers: 'bulk_update' })] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish(bulkEvent('updated', 'contact', 40)); + await flush(); + expect(selected(calls)).toEqual(['wh-1']); + expect(warn).not.toHaveBeenCalled(); + await ae.stop(); + }); + + it('an org-less subscription does NOT receive an organization-walled bulk event — loud, once', async () => { + const engine = new FakeEngine({ sys_webhook: [webhook({ triggers: 'bulk_update' })] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish(bulkEvent('updated', 'contact', 40, undefined, { organizationId: 'org_a' })); + await realtime.publish(bulkEvent('updated', 'contact', 41, undefined, { organizationId: 'org_b' })); + await flush(); + expect(selected(calls)).toEqual([]); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining('belongs to NO organization'), + expect.objectContaining({ id: 'wh-1', type: 'data.records.updated' }), + ); + await ae.stop(); + }); + + it('drops a bulk event whose organizationId is present but off-contract, delivering to nobody', async () => { + const engine = new FakeEngine({ + sys_webhook: [ + webhook({ id: 'wh-orgless', triggers: 'bulk_update' }), + webhook({ id: 'wh-a', triggers: 'bulk_update', organization_id: 'org_a' }), + ], + }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish({ + type: 'data.records.updated', + object: 'contact', + payload: { id: randomUUID(), type: 'data.records.updated', object: 'contact', matched: 4, organizationId: 42 }, + timestamp: '2026-05-24T00:00:00.000Z', + }); + await flush(); + expect(selected(calls)).toEqual([]); + expect(warn).toHaveBeenCalledTimes(1); + expect(String(warn.mock.calls[0][0])).toContain('off-contract bulk data event'); + expect(String(warn.mock.calls[0][0])).toContain('organizationId'); + await ae.stop(); + }); + }); + + it('the say-once ledger forgets a row the refresh no longer sees, so a re-created row reports again', async () => { + const engine = new FakeEngine({ sys_webhook: [webhook()] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0, logger: { warn } }); + await ae.start(); + + await realtime.publish(event('created', 'contact', { id: 'c-1' }, undefined, { organizationId: 'org_a' })); + await flush(); + expect(warn).toHaveBeenCalledTimes(1); + + // Row deleted → refresh prunes the ledger; row re-created under the + // same id → its first refusal is loud again. + engine.rows.sys_webhook = []; + await ae.refresh(); + engine.rows.sys_webhook = [webhook()]; + await ae.refresh(); + await realtime.publish( + event('created', 'contact', { id: 'c-2' }, '2026-05-24T00:00:01.000Z', { organizationId: 'org_a' }), + ); + await flush(); + expect(selected(calls)).toEqual([]); + expect(warn).toHaveBeenCalledTimes(2); + await ae.stop(); + }); +}); diff --git a/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts b/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts index 7f5269275a..2158854eb3 100644 --- a/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts +++ b/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts @@ -128,6 +128,15 @@ interface CachedSubscription { * when the row itself carries no organization (a `single`-posture install): * the delivery then lands NULL, which is honest for a subscription that * belongs to no organization. Threaded, never fabricated (#11303's rule). + * + * [#13566] ALSO the subscription half of the organization MATCH — compared + * against the organization the producer stamps on the event + * (`DataEvent.organizationId`, #14970; `BulkDataEvent.organizationId`, + * #15225 / #15813) before anything is enqueued. See + * {@link AutoEnqueuer.admitsOrganization} for the matrix; the short form + * is that a subscription with no organization ownership does not receive + * an organization-walled event, and a subscription with one receives only + * its own organization's. */ organizationId?: string; /** @@ -176,11 +185,17 @@ export interface AutoEnqueuerOptions { * The handler: * 1. Looks up matching subscriptions in an in-memory `Map` * — O(1) per event, no DB hit on the write path. - * 2. Calls `outbox.enqueue()` fire-and-forget for each match. The + * 2. [#13566] Compares each candidate's own organization with the one the + * producer stamped on the event — one equality per candidate, no + * lookup (see {@link admitsOrganization}). On a walled deployment this + * is what keeps organization A's record events out of organization B's + * webhook endpoints. + * 3. Calls `outbox.enqueue()` fire-and-forget for each match. The * enqueue itself is a single INSERT, which runs *after* the user's * request has already returned. * - * Net cost on the write path: one synchronous Map lookup (~microseconds). + * Net cost on the write path: one synchronous Map lookup (~microseconds) + * plus one string comparison per candidate subscription. * * ## Cache freshness * The cache is rebuilt: @@ -244,6 +259,17 @@ export class AutoEnqueuer { * refresh, forever. */ private readonly droppedForSecret = new Set(); + /** + * [#13566] Webhook ids whose FIRST organization-dimension refusal has been + * said out loud (see {@link admitsOrganization}). Same say-once shape as + * {@link droppedForSecret}, for the same reason: a subscription that + * cannot receive a class of events must be reported once, with the + * remedy, and not once per event forever — an org-less `'*'` subscription + * on a walled deployment would otherwise warn on every write of every + * organization. Pruned on refresh to the rows still live, so a row that + * is deleted and re-created reports again. + */ + private readonly organizationRefusalReported = new Set(); constructor( private readonly engine: IDataEngine, @@ -395,11 +421,15 @@ export class AutoEnqueuer { // deactivated. Otherwise the set grows for the life of the process, and // a webhook turned off while broken and later turned back on still // broken would have its first report suppressed as a repeat. - if (this.droppedForSecret.size > 0) { + if (this.droppedForSecret.size > 0 || this.organizationRefusalReported.size > 0) { const live = new Set(rows.map((r) => String(r?.id))); for (const id of this.droppedForSecret) { if (!live.has(id)) this.droppedForSecret.delete(id); } + // [#13566] Same pruning for the organization-refusal ledger. + for (const id of this.organizationRefusalReported) { + if (!live.has(id)) this.organizationRefusalReported.delete(id); + } } this.logger?.debug?.('[webhook-auto-enqueuer] cache refreshed', { @@ -856,6 +886,21 @@ export class AutoEnqueuer { return; } + // [#13566] The organization term rides beside `recordId` on the same + // payload (`DataEvent.organizationId`, stamped by the engine from the + // RECORD's own tenant column, #14970). Read once per event here, + // compared per candidate subscription below — never resolved. + const organization = readEventOrganizationId(payload as Record); + if (!organization) { + this.logger?.warn?.( + '[webhook-auto-enqueuer] dropping off-contract data event: `organizationId` is present but ' + + 'not a non-empty string (DataEventSchema refuses that at the publish site) — fix the ' + + 'producer; never coerced and never read as "no organization" (#13566)', + { type: event.type, object: event.object }, + ); + return; + } + // Deterministic eventId — same input on any node → same id. // Includes timestamp so two distinct updates to the same record // don't accidentally dedup. @@ -863,6 +908,11 @@ export class AutoEnqueuer { for (const sub of subs) { if (!sub.triggers.has(trigger)) continue; + // [#13566] The organization dimension of the match. Decided BEFORE + // the parked branch below on purpose: a parked subscription records + // the payload on a dead `sys_http_delivery` row (#8069), and another + // organization's record must not land there either. + if (!this.admitsOrganization(sub, organization.organizationId, event)) continue; // Fire-and-forget — never await on the hot path. Map the webhook // delivery onto the generic HTTP-outbox shape (ADR-0018 M3): @@ -969,8 +1019,28 @@ export class AutoEnqueuer { } const eventId = `${event.object}:${event.type}:${eventUuid}`; + // [#13566] `BulkDataEvent.organizationId` — the ONE organization the + // tenant wall named for the whole batch, or absent when the producer + // could not assert one (#15225 / #15813). Same reader as the + // per-record path; the two paths diverge only in what ABSENT means, + // which {@link admitsOrganization} spells out — on this path absent + // is a routine, expected value, and it is handled fail-closed. + const organization = readEventOrganizationId(payload as Record); + if (!organization) { + this.logger?.warn?.( + '[webhook-auto-enqueuer] dropping off-contract bulk data event: `organizationId` is present ' + + 'but not a non-empty string (BulkDataEventSchema refuses that at the publish site) — fix ' + + 'the producer; never coerced and never read as "not asserted" (#13566)', + { type: event.type, object: event.object }, + ); + return; + } + for (const sub of subs) { if (!sub.triggers.has(trigger)) continue; + // [#13566] See the per-record path — decided before the parked + // branch for the same reason. + if (!this.admitsOrganization(sub, organization.organizationId, event)) continue; void this.enqueue({ source: 'webhook', @@ -1000,6 +1070,117 @@ export class AutoEnqueuer { } } + /** + * [#13566] The organization dimension of the match — ONE comparison + * between the subscription's own organization (cached off its + * `sys_webhook` row, #13546) and the organization the producer stamped on + * the event (#14970 per record; #15225 / #15813 per batch). ⛔ No lookup: + * the enqueuer exists to keep this path O(1), and both halves are already + * in hand — the filter is a comparison, never a resolution. + * + * The cells, and why each falls where it does: + * + * | subscription | event | verdict | + * |--------------|--------|------------------------------------------------| + * | org A | org A | deliver | + * | org A | org B | not a match — the routine outcome of the new | + * | | | term, silent like an object-name mismatch | + * | org A | absent | REFUSE (fail-closed), said once per sub | + * | none | org A | REFUSE, said once per sub — the ruling's case | + * | none | absent | deliver — no wall on either side | + * + * **`none` × `org A` — the ruling.** A `sys_webhook` row with no + * organization on a walled deployment (a package-declared row bootstrapped + * under `isSystem`, a row authored before the column was provisioned) + * would otherwise receive EVERY organization's records at its URL, signed + * with its secret — the leak this card is. Maintainer's ruling + * (2026-09-07), verbatim: *a subscription with no organisation ownership + * does not fan out — loud refusal, never a silent cross-organisation + * delivery.* ADR-0131 D1 is why there is no third reading — NULL is not a + * state, so "no organization" is never "every organization". + * + * **`org A` × `absent` — fail-closed, on BOTH paths.** The two event + * families spell absence with the same key and mean different things by + * it (`packages/spec/src/api/events.zod.ts`, both members' docs). On a + * `DataEvent` it is "this record belongs to no organization" — an + * environment-wide row, an object outside the wall, or (the per-record + * producer's own docblock, `eventOrganizationId` in + * `packages/objectql/src/engine.ts`) no row in hand at the publish site, + * published absent rather than substituting the caller's organization. + * On a `BulkDataEvent` it is "the producer did not assert one + * organization for this batch" — a system or cross-membership predicate + * write — and the contract says outright that a tenant-scoped consumer + * must not deliver it inside an organization wall. Neither reading names + * organization A, so a subscription that belongs to A delivers on + * neither. ⛔ Absent is never "no tenancy concern": delivering on it + * would turn each of those cases into a cross-organization delivery. The + * cost is an environment-wide record not reaching an organization's + * webhook — accepted, and said once so the subscription is not dead while + * looking armed. + * + * **`none` × `absent` — deliver.** A `single`-posture deployment stamps + * nothing on either side (`postureStampsOrganization` is false there), so + * this cell is every event on every non-walled install; on a walled one + * it is an environment-wide subscription taking an environment-wide + * event. No organization is named anywhere, so there is no wall to cross. + * + * Both refusals are said ONCE per subscription (the #8022 say-once rule, + * ledger {@link organizationRefusalReported}): the first refused event is + * a `warn` naming the consequence and the remedy, later ones are + * debug-level. Without that a refused subscription reads active:true in + * Setup with nothing ever arriving — the "dead while looking armed" shape + * ADR-0078 refuses. + */ + private admitsOrganization( + sub: CachedSubscription, + eventOrganizationId: string | undefined, + event: RealtimeEventPayload, + ): boolean { + // org A × org A, and none × absent. + if (sub.organizationId === eventOrganizationId) return true; + // org A × org B — another organization's subscription is simply not a + // candidate for this event. Not logged: it is the match term working, + // once per event per foreign subscription. + if (sub.organizationId !== undefined && eventOrganizationId !== undefined) return false; + + // Exactly one side names an organization — refuse, and say so once. + const orgless = sub.organizationId === undefined; + const meta = { + id: sub.id, + webhook: sub.name, + type: event.type, + object: event.object, + subscriptionOrganizationId: sub.organizationId, + eventNamesOrganization: eventOrganizationId !== undefined, + }; + if (this.organizationRefusalReported.has(sub.id)) { + this.logger?.debug?.( + `[webhook-auto-enqueuer] webhook '${sub.name}' still refused on the organization ` + + 'dimension (#13566)', + meta, + ); + return false; + } + this.organizationRefusalReported.add(sub.id); + const message = orgless + ? `[webhook-auto-enqueuer] webhook '${sub.name}' belongs to NO organization, but this ` + + `${event.type} event on '${event.object}' is organization-walled (the producer stamped ` + + 'organizationId) — refusing to fan out (#13566): a subscription with no organization ' + + "ownership does not receive an organization's records. It will receive NO " + + 'organization-walled event while reading active:true in Setup; author the webhook ' + + 'inside the organization that should receive these events. Said once per subscription.' + : `[webhook-auto-enqueuer] webhook '${sub.name}' belongs to organization ` + + `'${sub.organizationId}', but this ${event.type} event on '${event.object}' names no ` + + 'organization — refusing to deliver it inside an organization wall (#13566): on the ' + + 'per-record path an absent organizationId is an environment-wide row or an object outside ' + + 'the wall; on the bulk path it is a batch the producer could not attribute to one ' + + 'organization (a system or cross-membership predicate write). A tenant-scoped ' + + "subscription receives only events attributable to its own organization. Said once per " + + 'subscription.'; + this.logger?.warn?.(message, meta); + return false; + } + private handleSelfHealEvent(event: RealtimeEventPayload): void { if (event.object !== this.subscriptionsObject) return; // [#4639] A predicate write over `sys_webhook` (deactivate every @@ -1035,6 +1216,34 @@ function mapActionToTrigger( } /** [#4639] `data.records.{action}` → its opt-in bulk trigger. */ +/** + * [#13566] The event's organization term, read off the payload the way the + * spec publishes it — `DataEvent.organizationId` / `BulkDataEvent.organizationId` + * (`@objectstack/spec/api`), `z.string().min(1).optional()` on both. + * + * Three answers, kept apart on purpose: + * - `{ organizationId: '' }` — present: the producer named the + * organization (the RECORD's on the per-record path; the ONE organization + * the tenant wall named for the batch on the bulk path); + * - `{ organizationId: undefined }` — absent: the key is not on the payload + * (or carries `undefined`, which is what the schema's `.optional()` admits + * and what any JSON hop turns into no key at all) — the schema's one + * spelling for "no organization named"; + * - `undefined` — OFF-CONTRACT: the key is present but is not a non-empty + * string (`''`, `null`, a number). The schema refuses every one of those + * at the publish site, so its arrival here means a producer that did not + * validate; the caller drops the whole event loudly, exactly like a + * missing `recordId` (#4626) — ⛔ never coerced, never read as absent. + */ +function readEventOrganizationId( + payload: Record, +): { organizationId: string | undefined } | undefined { + const value = payload.organizationId; + if (value === undefined) return { organizationId: undefined }; + if (typeof value !== 'string' || value === '') return undefined; + return { organizationId: value }; +} + function mapBulkActionToTrigger(action: string): 'bulk_update' | 'bulk_delete' | null { switch (action) { case 'updated': From 9e6d20c79d01722976fe6480c843655f7c1a7aa6 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 7 Sep 2026 15:31:12 +0000 Subject: [PATCH 2/3] test(plugin-webhooks): re-feed the #13565 stamp pins with an event naming the subscription's organization; add the changeset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two #13565 pins fed an organization-owned subscription an event that named no organization — the leniency #13566 removes. The pinned assertion (the enqueue input carries the SUBSCRIPTION's organization) is unchanged; the event now names that same organization, so the delivery it is pinned on still happens. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012zTkyNHJ7TkuN2oXtP5x37 --- .../webhook-fanout-organization-dimension.md | 14 ++++++++++++++ .../plugin-webhooks/src/auto-enqueuer.test.ts | 9 +++++++-- 2 files changed, 21 insertions(+), 2 deletions(-) create mode 100644 .changeset/webhook-fanout-organization-dimension.md diff --git a/.changeset/webhook-fanout-organization-dimension.md b/.changeset/webhook-fanout-organization-dimension.md new file mode 100644 index 0000000000..3d0e783331 --- /dev/null +++ b/.changeset/webhook-fanout-organization-dimension.md @@ -0,0 +1,14 @@ +--- +"@objectstack/plugin-webhooks": patch +--- + +Webhook fan-out now matches subscriptions on the organization dimension, closing a cross-organization delivery on walled deployments (`OS_TENANCY_POSTURE=isolated|group`). + +`AutoEnqueuer` selected the subscriptions to deliver to by object name and trigger only, and every organization's `sys_webhook` rows live in one cache — so organization A's record events reached organization B's webhook endpoint, signed with B's secret, on first delivery. Both the per-record (`data.record.*`) and the bulk (`data.records.*`) fan-out paths now compare the subscription's own organization (`sys_webhook.organization_id`) with the organization the engine stamps on the event (`DataEvent.organizationId`, `BulkDataEvent.organizationId`): one equality per candidate, no lookup on the hot path. + +What changes for a subscription: + +- **Owned by organization A** — receives only events stamped A. An event that names no organization (an environment-wide row or an object outside the wall on the per-record path; a batch the tenant wall could not attribute to one organization on the bulk path) is not delivered inside the wall — fail-closed — and the first such refusal is logged once with the reason. +- **With no organization** (`organization_id` NULL — for example a package-declared webhook on a walled deployment) — no longer receives any organization-stamped event; the refusal is logged once per subscription. It still receives events that name no organization. On a `single`-posture deployment nothing stamps either side, so delivery there is unchanged. + +An event whose `organizationId` is present but not a non-empty string is dropped loudly as off-contract, delivering to nobody. diff --git a/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts b/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts index 27e3e2e1e3..f89825a132 100644 --- a/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts +++ b/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts @@ -247,7 +247,10 @@ describe('AutoEnqueuer', () => { const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0 }); await ae.start(); - await realtime.publish(event('created', 'contact', { id: 'c-1' })); + // [#13566] The event names the subscription's own organization: an + // organization-owned subscription receives only its organization's + // events now, so the stamp is pinned on a delivery that still happens. + await realtime.publish(event('created', 'contact', { id: 'c-1' }, undefined, { organizationId: 'org_pin_alpha' })); await flush(); expect(calls).toHaveLength(1); @@ -646,7 +649,9 @@ describe('AutoEnqueuer — bulk data events (#4639)', () => { const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0 }); await ae.start(); - await realtime.publish(bulkEvent('updated', 'contact', 3)); + // [#13566] Same as the per-record pin: the batch is attributed to the + // subscription's own organization, so the delivery still happens. + await realtime.publish(bulkEvent('updated', 'contact', 3, undefined, { organizationId: 'org_pin_alpha' })); await flush(); expect(calls).toHaveLength(1); From d8d521c9336d85c4ca75ec2df0e2085bd97003d9 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 7 Sep 2026 22:28:48 +0000 Subject: [PATCH 3/3] fix(plugin-webhooks): keep issue ids out of the enqueuer's runtime message prose `check:doc-authoring` refuses a tracker id inside customer-facing string prose (maintainer ruling 2026-08-12); the ids stay in the code comments, the warn/debug texts name the rule and the remedy without them. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012zTkyNHJ7TkuN2oXtP5x37 --- .../plugins/plugin-webhooks/src/auto-enqueuer.test.ts | 2 +- packages/plugins/plugin-webhooks/src/auto-enqueuer.ts | 10 +++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts b/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts index f89825a132..f84bbb22ae 100644 --- a/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts +++ b/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts @@ -841,7 +841,7 @@ describe('AutoEnqueuer — organization dimension (#13566)', () => { expect.stringContaining('belongs to NO organization'), expect.objectContaining({ id: 'wh-1', type: 'data.record.created', object: 'contact' }), ); - expect(String(warn.mock.calls[0][0])).toContain('#13566'); + expect(String(warn.mock.calls[0][0])).toContain('refusing to fan out'); // Said once per subscription: the next refused event, from another // organization even, is debug-level. diff --git a/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts b/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts index 2158854eb3..69cc4e1cca 100644 --- a/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts +++ b/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts @@ -895,7 +895,7 @@ export class AutoEnqueuer { this.logger?.warn?.( '[webhook-auto-enqueuer] dropping off-contract data event: `organizationId` is present but ' + 'not a non-empty string (DataEventSchema refuses that at the publish site) — fix the ' + - 'producer; never coerced and never read as "no organization" (#13566)', + 'producer; never coerced and never read as "no organization"', { type: event.type, object: event.object }, ); return; @@ -1030,7 +1030,7 @@ export class AutoEnqueuer { this.logger?.warn?.( '[webhook-auto-enqueuer] dropping off-contract bulk data event: `organizationId` is present ' + 'but not a non-empty string (BulkDataEventSchema refuses that at the publish site) — fix ' + - 'the producer; never coerced and never read as "not asserted" (#13566)', + 'the producer; never coerced and never read as "not asserted"', { type: event.type, object: event.object }, ); return; @@ -1156,7 +1156,7 @@ export class AutoEnqueuer { if (this.organizationRefusalReported.has(sub.id)) { this.logger?.debug?.( `[webhook-auto-enqueuer] webhook '${sub.name}' still refused on the organization ` + - 'dimension (#13566)', + 'dimension', meta, ); return false; @@ -1165,13 +1165,13 @@ export class AutoEnqueuer { const message = orgless ? `[webhook-auto-enqueuer] webhook '${sub.name}' belongs to NO organization, but this ` + `${event.type} event on '${event.object}' is organization-walled (the producer stamped ` + - 'organizationId) — refusing to fan out (#13566): a subscription with no organization ' + + 'organizationId) — refusing to fan out: a subscription with no organization ' + "ownership does not receive an organization's records. It will receive NO " + 'organization-walled event while reading active:true in Setup; author the webhook ' + 'inside the organization that should receive these events. Said once per subscription.' : `[webhook-auto-enqueuer] webhook '${sub.name}' belongs to organization ` + `'${sub.organizationId}', but this ${event.type} event on '${event.object}' names no ` + - 'organization — refusing to deliver it inside an organization wall (#13566): on the ' + + 'organization — refusing to deliver it inside an organization wall: on the ' + 'per-record path an absent organizationId is an environment-wide row or an object outside ' + 'the wall; on the bulk path it is a batch the producer could not attribute to one ' + 'organization (a system or cross-membership predicate write). A tenant-scoped ' +