diff --git a/.changeset/bulk-data-event-contract.md b/.changeset/bulk-data-event-contract.md new file mode 100644 index 0000000000..3f3c6232d2 --- /dev/null +++ b/.changeset/bulk-data-event-contract.md @@ -0,0 +1,70 @@ +--- +"@objectstack/spec": minor +"@objectstack/objectql": minor +"@objectstack/client": minor +"@objectstack/plugin-webhooks": minor +"@objectstack/service-knowledge": patch +--- + +feat(spec,objectql,client,plugin-webhooks): predicate writes get an honest bulk event contract (#4639) + +A `multi: true` update/delete reaches `IDataDriver.updateMany` / `deleteMany`, +which are contracted to resolve an affected row COUNT and nothing else. That +satisfies neither `DataEvent.recordId` (required) nor `before` / `after` / +`changes`, so before #4626 the engine fabricated a per-record event with +`recordId: ''` and `after: ` — an event every schema-compliant consumer +must reject, and one the webhook enqueuer's `?? 'unknown'` fallback turned into +a real delivery naming an unidentifiable record. #4626 removed the fabrication +and published nothing instead: honest, but it left webhooks, knowledge sync and +`subscribeData` silent for every predicate write. + +Bulk writes now get their **own** contract rather than impersonating a +per-record one or going dark: + +- **New `BulkDataEvent`** (`@objectstack/spec/api`): `data.records.updated` / + `data.records.deleted` — note the plural — carrying `id`, `type`, `object`, + `matched`, `userId?`, `timestamp`. Deliberately a separate schema from + `DataEvent`, not a widened one: a consumer that receives + `data.records.updated` knows from the type alone that no `recordId` is + coming, instead of discovering an empty string at runtime. +- **Engine** publishes it from the `multi: true` branches of `update()` / + `delete()`, validated with `BulkDataEventSchema.parse` before publish. A + predicate that matched **zero** rows publishes nothing (no data changed — this + is what keeps an idle background sweep from becoming an hourly "0 records" + delivery), and a driver that resolves a non-count publishes nothing and warns + rather than asserting a number it cannot verify. Per-record writes are + untouched, including a scalar `where.id` with `multi: true`, which is still a + single-record target and still emits `data.record.deleted`. +- **Webhooks**: two new opt-in triggers, `bulk_update` and `bulk_delete` + (`WebhookTriggerType`, and the `sys_webhook.triggers` multi-select). They are + **not** extra sources for `create` / `update` / `delete`: the delivered body + has no `recordId` and no record, so routing it to existing per-record + subscribers would hand them a payload missing every field they read — the + same class of breakage as the old `recordId: ''`, from the other direction. A + webhook that wants both subscribes to both. Bulk deliveries dedup on the + producer's event uuid, since two sweeps in the same millisecond are genuinely + different events that a timestamp-based key would collapse. +- **Client SDK**: new `client.events.subscribeBulkData(object, cb)`, with the + same loud boundary validation as `subscribeData`. Kept a separate method for + the same reason — delivering a `BulkDataEvent` to a `(event: DataEvent) => + void` callback would recreate exactly the "typed field, `undefined` at + runtime" defect #4626 removed. `subscribeData`'s own guard was also tightened + from `data.` to `data.record.`, so an aggregate event is ignored rather than + rejected as off-contract. +- **Knowledge sync** now says out loud that a predicate write leaves its index + stale. A knowledge index is a per-record projection and `matched: 40` names no + record, so no event shape could drive it — the durable fix is reconciliation, + tracked in #4672. + +The event carries no `where` predicate. The only one available at publish time +is the middleware-composed AST, whose filter embeds the security layer's +injected row scoping (RLS, sharing) — publishing it would ship tenant scoping +internals to whatever external URL a webhook points at. + +Also pays off a measurement debt from #4655, which claimed the write-path cost +of event publishing had been measured but never published the numbers: +`packages/objectql/src/engine-data-events.bench.ts` measures it. Against an +in-memory driver, publishing costs ~7–9µs per event (insert 0.021ms vs 0.012ms, +single-id update 0.013ms vs 0.007ms). A bulk write pays that **once** regardless +of how many rows matched (0.040ms vs 0.034ms over a 100-row match set), so its +relative cost shrinks as the match set grows. diff --git a/content/docs/automation/webhooks.mdx b/content/docs/automation/webhooks.mdx index 22a497ff4c..dc7e73379d 100644 --- a/content/docs/automation/webhooks.mdx +++ b/content/docs/automation/webhooks.mdx @@ -94,7 +94,7 @@ in `definition_json`, a serialised `Webhook` JSON (canonical schema: | `name` | text | Unique snake_case name — referenced in logs and audit. | | `label` | text | Optional display label. | | `object_name` | text | Short object name whose record events fire this webhook. | -| `triggers` | select | Multi-select of `create` / `update` / `delete`, stored as an array (the enqueuer also accepts a legacy comma-separated string). | +| `triggers` | select | Multi-select of `create` / `update` / `delete` plus the opt-in bulk pair `bulk_update` / `bulk_delete` ([see below](#bulk-writes-bulk_update-and-bulk_delete)), stored as an array (the enqueuer also accepts a legacy comma-separated string). | | `url` | text | External endpoint that receives the POST. | | `method` | select | HTTP method — one of `GET` / `POST` / `PUT` / `PATCH` / `DELETE`. Default `POST`. | | `description` | textarea | Free-text description. | @@ -214,11 +214,59 @@ is the spec's `DataEvent` (`@objectstack/spec/api`) — validated against } ``` -> **A multi-row write emits no record event.** `updateMany` / `deleteMany` -> (`multi: true`) return only an affected count, so there is no record for a -> `DataEvent` to name and the engine publishes nothing rather than an event -> with an empty `recordId` — meaning webhooks do **not** fire for bulk writes -> today. Tracked in [#4639](https://github.com/objectstack-ai/objectstack/issues/4639). +### Bulk writes: `bulk_update` and `bulk_delete` + +A predicate write — `updateMany` / `deleteMany` (`multi: true`) — reports only +an affected count, so there is no record for a `DataEvent` to name. Rather than +publish an event with an empty `recordId` (which every schema-compliant +consumer must reject), the engine publishes a **separate** aggregate event +(#4639): + +```ts +{ + type: 'data.records.updated', // note: recordS — plural + object: 'account', + timestamp: '', + payload: { + id: '', // unique event id + type: 'data.records.updated', + object: 'account', + matched: 40, // how many records the predicate affected + userId: 'usr_1', // when the write names an actor + timestamp: '', + }, +} +``` + +These dispatch under their own triggers, `bulk_update` and `bulk_delete`, and +they are **opt-in**: a webhook declaring `update` does not receive them. That +is deliberate — the body has no `recordId` and no record, so delivering it to a +subscriber written against the per-record shape would hand it a payload missing +everything it reads. + +```ts +webhooks: [{ + name: 'account_bulk_audit', + object: 'account', + triggers: ['update', 'bulk_update'], // subscribe to both if you want both + url: 'https://example.com/hooks/accounts', +}] +``` + +Two properties worth knowing: + +- **No event when nothing matched.** A predicate that affected zero rows + changed no data, so it publishes nothing — an idle background sweep does not + become an hourly "0 records" delivery. +- **The predicate is not included.** The only filter available at publish time + is the query after the security layer composed row scoping into it (RLS, + sharing), so sending it would leak tenant scoping internals to the + destination URL. The event states the count and nothing more. + +Because a count names no rows, a bulk delivery cannot drive an incremental +per-record projection (a cache, a search index, a mirror). Use it to invalidate, +alert, or schedule a refetch; anything that must know *which* records changed +has to reconcile against the source. > **Not yet cluster-aware.** The only shipped `IRealtimeService` implementation > is `InMemoryRealtimeAdapter`, an in-process, single-node pub/sub with no diff --git a/content/docs/protocol/knowledge.mdx b/content/docs/protocol/knowledge.mdx index 105ec9519b..a915ec1958 100644 --- a/content/docs/protocol/knowledge.mdx +++ b/content/docs/protocol/knowledge.mdx @@ -296,9 +296,11 @@ with a `permissions` mapping function at index time. ## 6. Sync model -For `object` sources, `KnowledgeService` subscribes to ObjectQL -`record.created`, `record.updated`, and `record.deleted` events. Each -event triggers a single `adapter.upsert` / `adapter.delete` call. +For `object` sources, `KnowledgeService` subscribes to the ObjectQL engine's +`data.record.created`, `data.record.updated` and `data.record.deleted` events +(the legacy unprefixed `record.*` shape is still accepted). Each event triggers +a single `adapter.upsert` / `adapter.delete` call — the record body is read +from the event's `after`, and a delete's id from its required `recordId`. - **MVP (Phase 1):** synchronous, inline with the originating mutation. Fast for low-volume dev / demo. Indexing failures are logged but do @@ -306,6 +308,19 @@ event triggers a single `adapter.upsert` / `adapter.delete` call. - **Phase 2:** async via `service-queue` for batching, retries, and back-pressure. + + **A predicate write leaves the index stale.** A `multi: true` update/delete + reaches `updateMany` / `deleteMany`, which report only an affected row count, + so it publishes the aggregate `data.records.updated` / `data.records.deleted` + ([#4639](https://github.com/objectstack-ai/objectstack/issues/4639)) rather + than per-record events. A knowledge index is a per-record projection and + `matched: 40` names no record, so there is no upsert or delete to derive — + the service logs a warning naming the object and count instead of failing + silently. Reconciliation against the source object is the durable fix and is + tracked in [#4672](https://github.com/objectstack-ai/objectstack/issues/4672): + events keep the index *fresh*, reconciliation keeps it *correct*. + + `file` / `http` sources rely on explicit `reindexSource` calls (typically triggered by a cron job, a Console button, or a webhook). diff --git a/content/docs/references/api/events.mdx b/content/docs/references/api/events.mdx index 6b27f61acc..93d57b6167 100644 --- a/content/docs/references/api/events.mdx +++ b/content/docs/references/api/events.mdx @@ -26,13 +26,39 @@ Examples: ## TypeScript Usage ```typescript -import { DataEventSchema, DataEventType, MetadataEventSchema, MetadataEventType } from '@objectstack/spec/api'; -import type { DataEvent, DataEventType, MetadataEvent, MetadataEventType } from '@objectstack/spec/api'; +import { BulkDataEventSchema, BulkDataEventType, DataEventSchema, DataEventType, MetadataEventSchema, MetadataEventType } from '@objectstack/spec/api'; +import type { BulkDataEvent, BulkDataEventType, DataEvent, DataEventType, MetadataEvent, MetadataEventType } from '@objectstack/spec/api'; // Validate data -const result = DataEventSchema.parse(data); +const result = BulkDataEventSchema.parse(data); ``` +--- + +## BulkDataEvent + +### Properties + +| Property | Type | Required | Description | +| :--- | :--- | :--- | :--- | +| **id** | `string` | ✅ | Unique event identifier | +| **type** | `Enum<'data.records.updated' \| 'data.records.deleted'>` | ✅ | Event type | +| **object** | `string` | ✅ | Object name | +| **matched** | `integer` | ✅ | Number of records affected | +| **userId** | `string` | optional | User who triggered the event | +| **timestamp** | `string` | ✅ | Event timestamp | + + +--- + +## BulkDataEventType + +### Allowed Values + +* `data.records.updated` +* `data.records.deleted` + + --- ## DataEvent diff --git a/content/docs/references/automation/webhook.mdx b/content/docs/references/automation/webhook.mdx index aa20fa52d0..bd6e6514a5 100644 --- a/content/docs/references/automation/webhook.mdx +++ b/content/docs/references/automation/webhook.mdx @@ -19,6 +19,26 @@ producer are declared here — an author can't subscribe to something that never fires. +**Bulk triggers (#4639).** `bulk_update` / `bulk_delete` map to the engine's + +aggregate `data.records.updated` / `data.records.deleted`, emitted when a + +predicate write (`multi: true` → `IDataDriver.updateMany`/`deleteMany`) + +affects a set of rows the driver reports only as a count. They are separate + +trigger values, not extra sources for `update` / `delete`, because their + +delivery has a different SHAPE: no `recordId`, no record body, just + +`object` + `matched`. Folding them into the per-record triggers would send + +every existing subscriber a body missing the fields it reads — the same + +class of breakage as the pre-#4626 `recordId: ''` fabrication, arriving from + +the other direction. A webhook that wants both subscribes to both. + Deliberately NOT triggers (#3196): - `undelete` — there is no soft-delete / restore capability in the engine @@ -47,7 +67,7 @@ value that silently never fires. ```typescript import { WebhookSchema, WebhookTriggerType } from '@objectstack/spec/automation'; -import type { Webhook } from '@objectstack/spec/automation'; +import type { Webhook, WebhookTriggerType } from '@objectstack/spec/automation'; // Validate data const result = WebhookSchema.parse(data); @@ -63,8 +83,8 @@ const result = WebhookSchema.parse(data); | :--- | :--- | :--- | :--- | | **name** | `string` | ✅ | Webhook unique name (lowercase snake_case) | | **label** | `string` | optional | Human-readable webhook label | -| **object** | `string` | optional | Object whose record events (create/update/delete) trigger this webhook | -| **triggers** | `Enum<'create' \| 'update' \| 'delete'>[]` | optional | Events that trigger execution | +| **object** | `string` | optional | Object whose record events (create/update/delete, bulk_update/bulk_delete) trigger this webhook | +| **triggers** | `Enum<'create' \| 'update' \| 'delete' \| 'bulk_update' \| 'bulk_delete'>[]` | optional | Events that trigger execution | | **url** | `string` | ✅ | External webhook endpoint URL | | **method** | `Enum<'GET' \| 'POST' \| 'PUT' \| 'PATCH' \| 'DELETE'>` | ✅ | HTTP method | | **headers** | `Record` | optional | Custom HTTP headers | @@ -83,6 +103,8 @@ const result = WebhookSchema.parse(data); * `create` * `update` * `delete` +* `bulk_update` +* `bulk_delete` --- diff --git a/content/docs/references/integration/connector.mdx b/content/docs/references/integration/connector.mdx index c3570ce5d4..7fbde4e520 100644 --- a/content/docs/references/integration/connector.mdx +++ b/content/docs/references/integration/connector.mdx @@ -178,7 +178,7 @@ Circuit breaker configuration | **triggers** | `{ key: string; label: string; description?: string; type: Enum<'polling' \| 'webhook'>; … }[]` | optional | Trigger definitions (not yet enforced — never read at registration; see #3197) | | **syncConfig** | `{ strategy?: Enum<'full' \| 'incremental' \| 'upsert' \| 'append_only'>; direction?: Enum<'import' \| 'export' \| 'bidirectional'>; schedule?: string \| { dialect: Enum<'cel' \| 'cron' \| 'template'>; source?: string; ast?: any; meta?: object }; realtimeSync?: boolean; … }` | optional | Data sync configuration | | **fieldMappings** | `{ source: string; target: string; transform?: { type: 'constant'; value: any } \| { type: 'cast'; targetType: Enum<'string' \| 'number' \| 'boolean' \| 'date'> } \| { type: 'lookup'; table: string; keyField: string; valueField: string } \| { type: 'javascript'; expression: string \| { dialect: Enum<'cel' \| 'cron' \| 'template'>; source?: string; ast?: any; meta?: object } } \| { type: 'map'; mappings: Record }; defaultValue?: any; … }[]` | optional | Field mapping rules | -| **webhooks** | `{ name: string; label?: string; object?: string; triggers?: Enum<'create' \| 'update' \| 'delete'>[]; … }[]` | optional | Webhook configurations (not yet enforced — never read at registration; see #3197) | +| **webhooks** | `{ name: string; label?: string; object?: string; triggers?: Enum<'create' \| 'update' \| 'delete' \| 'bulk_update' \| 'bulk_delete'>[]; … }[]` | optional | Webhook configurations (not yet enforced — never read at registration; see #3197) | | **rateLimitConfig** | `{ strategy?: Enum<'fixed_window' \| 'sliding_window' \| 'token_bucket' \| 'leaky_bucket'>; maxRequests: number; windowSeconds: number; burstCapacity?: number; … }` | optional | Rate limiting configuration | | **retryConfig** | `{ strategy?: Enum<'exponential_backoff' \| 'linear_backoff' \| 'fixed_delay' \| 'no_retry'>; maxAttempts?: number; initialDelayMs?: number; maxDelayMs?: number; … }` | optional | Retry configuration | | **connectionTimeoutMs** | `number` | optional | Connection timeout in ms | @@ -336,7 +336,7 @@ Connector type | **triggers** | `{ key: string; label: string; description?: string; type: Enum<'polling' \| 'webhook'>; … }[]` | optional | Trigger definitions (not yet enforced — never read at registration; see #3197) | | **syncConfig** | `{ strategy?: Enum<'full' \| 'incremental' \| 'upsert' \| 'append_only'>; direction?: Enum<'import' \| 'export' \| 'bidirectional'>; schedule?: string \| { dialect: Enum<'cel' \| 'cron' \| 'template'>; source?: string; ast?: any; meta?: object }; realtimeSync?: boolean; … }` | optional | Data sync configuration | | **fieldMappings** | `{ source: string; target: string; transform?: { type: 'constant'; value: any } \| { type: 'cast'; targetType: Enum<'string' \| 'number' \| 'boolean' \| 'date'> } \| { type: 'lookup'; table: string; keyField: string; valueField: string } \| { type: 'javascript'; expression: string \| { dialect: Enum<'cel' \| 'cron' \| 'template'>; source?: string; ast?: any; meta?: object } } \| { type: 'map'; mappings: Record }; defaultValue?: any; … }[]` | optional | Field mapping rules | -| **webhooks** | `{ name: string; label?: string; object?: string; triggers?: Enum<'create' \| 'update' \| 'delete'>[]; … }[]` | optional | Webhook configurations (not yet enforced — never read at registration; see #3197) | +| **webhooks** | `{ name: string; label?: string; object?: string; triggers?: Enum<'create' \| 'update' \| 'delete' \| 'bulk_update' \| 'bulk_delete'>[]; … }[]` | optional | Webhook configurations (not yet enforced — never read at registration; see #3197) | | **rateLimitConfig** | `{ strategy?: Enum<'fixed_window' \| 'sliding_window' \| 'token_bucket' \| 'leaky_bucket'>; maxRequests: number; windowSeconds: number; burstCapacity?: number; … }` | optional | Rate limiting configuration | | **retryConfig** | `{ strategy?: Enum<'exponential_backoff' \| 'linear_backoff' \| 'fixed_delay' \| 'no_retry'>; maxAttempts?: number; initialDelayMs?: number; maxDelayMs?: number; … }` | optional | Retry configuration | | **connectionTimeoutMs** | `number` | optional | Connection timeout in ms | @@ -459,8 +459,8 @@ Synchronization strategy | :--- | :--- | :--- | :--- | | **name** | `string` | ✅ | Webhook unique name (lowercase snake_case) | | **label** | `string` | optional | Human-readable webhook label | -| **object** | `string` | optional | Object whose record events (create/update/delete) trigger this webhook | -| **triggers** | `Enum<'create' \| 'update' \| 'delete'>[]` | optional | Events that trigger execution | +| **object** | `string` | optional | Object whose record events (create/update/delete, bulk_update/bulk_delete) trigger this webhook | +| **triggers** | `Enum<'create' \| 'update' \| 'delete' \| 'bulk_update' \| 'bulk_delete'>[]` | optional | Events that trigger execution | | **url** | `string` | ✅ | External webhook endpoint URL | | **method** | `Enum<'GET' \| 'POST' \| 'PUT' \| 'PATCH' \| 'DELETE'>` | ✅ | HTTP method | | **headers** | `Record` | optional | Custom HTTP headers | diff --git a/packages/client/src/realtime-api-data.test.ts b/packages/client/src/realtime-api-data.test.ts index 4c73ae088e..6dbe8fb1f9 100644 --- a/packages/client/src/realtime-api-data.test.ts +++ b/packages/client/src/realtime-api-data.test.ts @@ -35,6 +35,14 @@ const VALID_EVENT = { timestamp: '2026-08-02T12:00:00.000Z', } as const; +const VALID_BULK_EVENT = { + id: '3f2504e0-4f89-41d3-9a0c-0305e82c3301', + type: 'data.records.updated', + object: 'project_task', + matched: 40, + timestamp: '2026-08-02T12:00:00.000Z', +} as const; + function envelopeOf( payload: Record, type = 'data.record.updated', @@ -173,4 +181,108 @@ describe('#4626 — RealtimeAPI.subscribeData contract boundary', () => { deliver(envelopeOf({ ...VALID_EVENT })); expect(callback).toHaveBeenCalledTimes(1); }); + + it('ignores an aggregate bulk event rather than rejecting it as off-contract', () => { + // [#4639] `data.records.updated` satisfies a DIFFERENT contract, so it must + // not reach `DataEventSchema` here. The pre-#4639 guard tested + // `startsWith('data.')`, which would have thrown on every predicate write. + const callback = vi.fn(); + api.subscribeData('project_task', callback); + + expect(() => + deliver(envelopeOf({ ...VALID_BULK_EVENT }, 'data.records.updated')), + ).not.toThrow(); + expect(callback).not.toHaveBeenCalled(); + }); +}); + +/** + * #4639 — `subscribeBulkData` carries the aggregate contract for predicate + * writes (`multi: true` → `updateMany`/`deleteMany`, which report a count). + * + * Separate from `subscribeData` on purpose: a `BulkDataEvent` has no + * `recordId` and no record body, so delivering it through a + * `(event: DataEvent) => void` callback would recreate the exact defect #4626 + * removed — a typed field that is `undefined` at runtime. + */ +describe('#4639 — RealtimeAPI.subscribeBulkData contract boundary', () => { + let api: RealtimeAPI; + + beforeEach(() => { + vi.useFakeTimers(); + api = new RealtimeAPI('http://localhost:3000'); + }); + + afterEach(() => { + api.disconnect(); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + function deliver(envelope: RealtimeEventPayload): void { + api._bufferEvent(envelope); + vi.advanceTimersByTime(2000); + } + + it('delivers the BulkDataEvent with its top-level count', () => { + const seen: any[] = []; + api.subscribeBulkData('project_task', (event) => seen.push(event)); + + deliver(envelopeOf({ ...VALID_BULK_EVENT }, 'data.records.updated')); + + expect(seen).toHaveLength(1); + expect(seen[0].type).toBe('data.records.updated'); + expect(seen[0].matched).toBe(40); + expect(seen[0].object).toBe('project_task'); + expect(seen[0].recordId).toBeUndefined(); + }); + + it('rejects an off-contract bulk payload LOUDLY instead of coercing it', () => { + const callback = vi.fn(); + const consoleError = vi.spyOn(console, 'error').mockImplementation(() => undefined); + api.subscribeBulkData('project_task', callback); + + // `matched` missing — the one field a bulk event exists to carry. + const { matched: _dropped, ...withoutMatched } = VALID_BULK_EVENT; + deliver(envelopeOf({ ...withoutMatched }, 'data.records.updated')); + + expect(callback).not.toHaveBeenCalled(); + expect(consoleError).toHaveBeenCalled(); + }); + + it('does not receive per-record events', () => { + const callback = vi.fn(); + api.subscribeBulkData('project_task', callback); + + deliver(envelopeOf({ ...VALID_EVENT })); + + expect(callback).not.toHaveBeenCalled(); + }); + + it('ignores bulk events for another object', () => { + const callback = vi.fn(); + api.subscribeBulkData('project_task', callback); + + deliver( + envelopeOf( + { ...VALID_BULK_EVENT, object: 'other_object' }, + 'data.records.deleted', + 'other_object', + ), + ); + + expect(callback).not.toHaveBeenCalled(); + }); + + it('unsubscribe stops delivery', () => { + const callback = vi.fn(); + const off = api.subscribeBulkData('project_task', callback); + + deliver(envelopeOf({ ...VALID_BULK_EVENT }, 'data.records.updated')); + expect(callback).toHaveBeenCalledTimes(1); + + off(); + deliver(envelopeOf({ ...VALID_BULK_EVENT }, 'data.records.updated')); + expect(callback).toHaveBeenCalledTimes(1); + }); }); diff --git a/packages/client/src/realtime-api.ts b/packages/client/src/realtime-api.ts index d39fe072dd..3261c3b55b 100644 --- a/packages/client/src/realtime-api.ts +++ b/packages/client/src/realtime-api.ts @@ -11,8 +11,10 @@ import type { RealtimeEventPayload } from '@objectstack/spec/contracts'; import { MetadataEventSchema, DataEventSchema, + BulkDataEventSchema, type MetadataEvent, type DataEvent, + type BulkDataEvent, } from '@objectstack/spec/api'; export interface RealtimeSubscriptionFilter { @@ -127,6 +129,14 @@ export class RealtimeAPI { }, handler: (event) => { if (!event.type.startsWith('data.') || event.object !== object) return; + // [#4639] An aggregate `data.records.*` from a predicate write is not + // off-contract — it satisfies a DIFFERENT one — so it must be skipped + // before the `DataEventSchema` check below rejects it as malformed. + // Excluded by its own namespace rather than by narrowing the guard to + // `data.record.`, which would also silently drop `data.field.changed` + // (declared in `DataEventType`, no producer today — see #4673) if it + // ever gains one. Bulk events are delivered by `subscribeBulkData`. + if (event.type.startsWith('data.records.')) return; // Contract boundary (#4626): the wire carries a RealtimeEventPayload // envelope whose `payload` is the producer's DataEvent (the ObjectQL // engine builds and validates it). Validate it here too — the callback @@ -163,6 +173,65 @@ export class RealtimeAPI { }; } + /** + * Subscribe to aggregate bulk-write events for an object (#4639). + * + * A predicate write (`multi: true` update/delete) reaches + * `IDataDriver.updateMany`/`deleteMany`, which report an affected COUNT and + * name no rows — so it publishes `data.records.updated` / + * `data.records.deleted` rather than the per-record events + * {@link subscribeData} delivers. The callback receives a + * {@link BulkDataEvent}: `object` and `matched`, no `recordId`, no record + * body. + * + * Kept as a separate method on purpose. Delivering these through + * `subscribeData` would hand a `(event: DataEvent) => void` callback an + * object whose `recordId` is missing — reintroducing, from the producer + * side, exactly the "typed field that is `undefined` at runtime" defect + * #4626 removed. A caller that wants both subscribes twice, and the types + * make the difference unmissable. + * + * Use it to invalidate a whole view, show "40 records changed", or schedule + * a refetch — not to patch a per-record cache, which a count cannot drive. + */ + subscribeBulkData( + object: string, + callback: (event: BulkDataEvent) => void + ): () => void { + const subscriptionId = `bulk-data-${object}-${Date.now()}`; + + this.subscriptions.set(subscriptionId, { + filter: { + type: object, + eventTypes: ['data.records.updated', 'data.records.deleted'] + }, + handler: (event) => { + if (!event.type.startsWith('data.records.') || event.object !== object) return; + // Same boundary discipline as subscribeData: the envelope's `payload` + // is the producer's BulkDataEvent, validated here too. Off-contract is + // rejected LOUDLY (throw → surfaced by emitEvent's handler-error log), + // never coerced — a malformed event means the producer is broken. + const parsed = BulkDataEventSchema.safeParse(event.payload); + if (!parsed.success) { + throw new Error( + `subscribeBulkData('${object}'): event '${event.type}' payload does not satisfy ` + + `BulkDataEventSchema — rejecting off-contract event (fix the producer): ${parsed.error.message}` + ); + } + callback(parsed.data); + } + }); + + this.startPolling(); + + return () => { + this.subscriptions.delete(subscriptionId); + if (this.subscriptions.size === 0) { + this.stopPolling(); + } + }; + } + /** * Emit an event to all matching subscriptions (client-side only) * This is used for in-process event delivery diff --git a/packages/objectql/src/engine-data-events.bench.ts b/packages/objectql/src/engine-data-events.bench.ts new file mode 100644 index 0000000000..e9de563a89 --- /dev/null +++ b/packages/objectql/src/engine-data-events.bench.ts @@ -0,0 +1,202 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * Write-path cost of publishing realtime data events. + * + * #4655 (the #4626 fix) claimed this had been measured and never published the + * numbers — so the per-event `crypto.randomUUID()` + `DataEventSchema.parse()` + * added to every insert/update/delete has had no evidence behind it since. + * #4639 adds a second publisher (`BulkDataEventSchema.parse` on predicate + * writes), which is the moment to actually pay that debt rather than repeat the + * claim. + * + * Run: `pnpm --filter @objectstack/objectql exec vitest bench src/engine-data-events.bench.ts` + * + * Each pair below is the SAME write with and without a realtime service + * attached, so the delta is exactly the event-publishing work: uuid generation, + * schema validation, envelope construction, and the (no-op) publish. The + * in-memory driver keeps driver cost near zero, which flatters the event cost — + * i.e. the relative overhead measured here is an upper bound on what a real + * SQL/HTTP driver would show. + */ + +import { bench, describe } from 'vitest'; +import type { IRealtimeService, RealtimeEventPayload } from '@objectstack/spec/contracts'; +import { ObjectQL } from './engine.js'; + +const task = { + name: 'task', + label: 'Task', + fields: { + id: { name: 'id', type: 'text' as const, primaryKey: true }, + title: { name: 'title', type: 'text' as const }, + status: { name: 'status', type: 'text' as const }, + }, +}; + +function makeMemoryDriver() { + const stores = new Map>>(); + const storeFor = (o: string) => { + let s = stores.get(o); + if (!s) { s = new Map(); stores.set(o, s); } + return s; + }; + let nextId = 0; + const matches = (row: Record, where: any): boolean => { + if (!where || typeof where !== 'object') return true; + for (const [k, v] of Object.entries(where)) { + if (k.startsWith('$')) continue; + const exp = (v && typeof v === 'object' && '$eq' in (v as any)) ? (v as any).$eq : v; + if ((row[k] ?? null) !== (exp ?? null)) return false; + } + return true; + }; + const driver: any = { + name: 'memory', version: '0.0.0', supports: {}, + async connect() {}, async disconnect() {}, async checkHealth() { return true; }, async execute() { return null; }, + async find(o: string, ast: any) { return Array.from(storeFor(o).values()).filter((r) => matches(r, ast?.where)); }, + findStream() { throw new Error('ns'); }, + async findOne(o: string, ast: any) { for (const r of storeFor(o).values()) if (matches(r, ast?.where)) return r; return null; }, + async create(o: string, data: Record) { + nextId += 1; + const id = (data.id as string) ?? `r_${nextId}`; + const row = { ...data, id }; + storeFor(o).set(id, row); + return row; + }, + async update(o: string, id: string, data: Record) { + const s = storeFor(o); const cur = s.get(id); + if (!cur) throw new Error(`nf ${o}/${id}`); + const up = { ...cur, ...data, id }; s.set(id, up); return up; + }, + async upsert(o: string, data: Record) { + const id = data.id as string | undefined; + return id && storeFor(o).has(id) ? this.update(o, id, data) : this.create(o, data); + }, + async delete(o: string, id: string) { return storeFor(o).delete(id); }, + async count(o: string, ast: any) { return (await this.find(o, ast)).length; }, + async bulkCreate(o: string, rows: Record[]) { return Promise.all(rows.map((r) => this.create(o, r))); }, + async updateMany(o: string, ast: any, data: Record) { + const rows = await this.find(o, ast); + for (const r of rows) storeFor(o).set(r.id as string, { ...r, ...data }); + return rows.length; + }, + async deleteMany(o: string, ast: any) { + const rows = await this.find(o, ast); + for (const r of rows) storeFor(o).delete(r.id as string); + return rows.length; + }, + async bulkUpdate() { return []; }, async bulkDelete() {}, + async beginTransaction() { return { commit: async () => {}, rollback: async () => {} }; }, + async commit() {}, async rollback() {}, + }; + return driver; +} + +/** A publish sink that does nothing, so we time the PRODUCER, not a transport. */ +const nullRealtime: IRealtimeService = { + publish: async (_event: RealtimeEventPayload) => undefined, + subscribe: async () => 'sub', + unsubscribe: async () => undefined, +}; + +async function makeEngine(withRealtime: boolean): Promise { + const engine = new ObjectQL(); + engine.registerDriver(makeMemoryDriver(), true); + await engine.init(); + engine.registry.registerObject(task as any, 'bench'); + if (withRealtime) engine.setRealtimeService(nullRealtime); + // Silence the per-write logger so log formatting is not in the measurement. + const logger = (engine as any).logger; + for (const level of ['debug', 'info', 'warn']) logger[level] = () => undefined; + return engine; +} + +/** + * Each case gets its OWN engine pair. Sharing one pair across cases invalidates + * the comparison: vitest runs the two arms of a case for a fixed WALL-CLOCK + * budget, not a fixed iteration count, so the faster arm executes far more + * iterations — and any state those iterations accumulate (rows inserted, rows + * flipped) then diverges between the two engines. A later case measured on + * those engines is comparing different table sizes, not different event + * settings. The first draft of this file shared one pair and duly reported the + * predicate write as *faster* with events enabled, because the events-off + * engine had ~84k rows to scan against the events-on engine's ~47k. + */ +const SEED_ROWS = 100; + +async function makePair(): Promise<[ObjectQL, ObjectQL]> { + const pair = [await makeEngine(true), await makeEngine(false)] as [ObjectQL, ObjectQL]; + for (const engine of pair) { + await engine.insert( + 'task', + Array.from({ length: SEED_ROWS }, (_, i) => ({ title: `seed ${i}`, status: 'open' })), + ); + } + return pair; +} + +/** + * Memoized lazy init. Two constraints rule out the obvious alternatives: + * this package compiles to CommonJS, so a top-level `await` is a TS1309 error; + * and vitest's benchmark mode is experimental and does NOT run `beforeAll`, so + * a hook-based setup leaves every engine `undefined`, every iteration throwing, + * and the summary reporting `NaNx faster` off zero samples. + * + * Awaiting an already-settled promise costs one microtask, paid identically by + * both arms, so the delta each case reports is unaffected — and the actual + * construction happens during vitest's warmup iterations, outside the measured + * samples. + */ +function lazyPair(): () => Promise<[ObjectQL, ObjectQL]> { + let pending: Promise<[ObjectQL, ObjectQL]> | undefined; + return () => (pending ??= makePair()); +} + +const insertPair = lazyPair(); +const updatePair = lazyPair(); +const bulkPair = lazyPair(); + +describe('insert — per-record DataEvent (#4626)', () => { + bench('with realtime service', async () => { + const [on] = await insertPair(); + await on.insert('task', { title: 'bench', status: 'open' }); + }); + + bench('without realtime service', async () => { + const [, off] = await insertPair(); + await off.insert('task', { title: 'bench', status: 'open' }); + }); +}); + +describe('single-id update — per-record DataEvent (#4626)', () => { + bench('with realtime service', async () => { + const [on] = await updatePair(); + await on.update('task', { id: 'r_1', title: 'bench' }); + }); + + bench('without realtime service', async () => { + const [, off] = await updatePair(); + await off.update('task', { id: 'r_1', title: 'bench' }); + }); +}); + +describe('predicate update — one aggregate BulkDataEvent (#4639)', () => { + // The event cost here is per WRITE, not per row: one uuid + one + // `BulkDataEventSchema.parse` regardless of how many rows matched. The gap + // between these two arms is the whole of what #4639 adds to a bulk write. + // + // The predicate writes `title` and filters on `status`, so it matches the + // same SEED_ROWS rows on every iteration — a filter that mutated the column + // it selects on would shrink its own match set as the bench ran, making the + // two arms diverge exactly like the shared-engine bug above. + bench('with realtime service', async () => { + const [on] = await bulkPair(); + await on.update('task', { title: 'bench' }, { multi: true, where: { status: 'open' } } as any); + }); + + bench('without realtime service', async () => { + const [, off] = await bulkPair(); + await off.update('task', { title: 'bench' }, { multi: true, where: { status: 'open' } } as any); + }); +}); diff --git a/packages/objectql/src/engine-data-events.test.ts b/packages/objectql/src/engine-data-events.test.ts index 2721aca2b9..554f86f1bb 100644 --- a/packages/objectql/src/engine-data-events.test.ts +++ b/packages/objectql/src/engine-data-events.test.ts @@ -16,15 +16,18 @@ * - the transport envelope's `payload` IS a schema-valid `DataEvent`; * - a batch insert publishes one event PER RECORD, with unique ids; * - `userId` is carried when the execution context names an actor; - * - a multi-row write (`updateMany`/`deleteMany` → affected count) publishes - * NOTHING, loudly — it has no per-record identity, and `recordId` is - * required, so the pre-fix fabrication (`recordId: ''`, `after: `) - * is not replaced by another one; + * - a multi-row write publishes no PER-RECORD event — it has no per-record + * identity, and `recordId` is required, so the pre-fix fabrication + * (`recordId: ''`, `after: `) is not replaced by another one; * - a publish failure never fails the write. + * + * #4639 then gave the multi-row case its own honest contract rather than + * leaving it silent — see the second describe block: `data.records.updated` / + * `data.records.deleted`, carrying `matched` and NO `recordId`. */ import { describe, it, expect, beforeEach, vi } from 'vitest'; -import { DataEventSchema } from '@objectstack/spec/api'; +import { BulkDataEventSchema, DataEventSchema } from '@objectstack/spec/api'; import type { IRealtimeService, RealtimeEventPayload } from '@objectstack/spec/contracts'; import { ObjectQL } from './engine.js'; @@ -203,51 +206,195 @@ describe('#4626 — engine writes publish true DataEvents', () => { expect(DataEventSchema.parse(published[0].payload).userId).toBeUndefined(); }); - it('publishes NOTHING for a multi-row update — and says so', async () => { + it('a publish failure never fails the write itself', async () => { + (realtime.publish as ReturnType).mockRejectedValueOnce(new Error('transport down')); + + const record = await engine.insert('task', { title: 'still written' }); + expect(record.id).toBeTruthy(); + expect(await engine.findOne('task', { where: { id: record.id } })).toMatchObject({ title: 'still written' }); + }); + + it('publishes nothing at all when no realtime service is configured', async () => { + const bare = new ObjectQL(); + const { driver } = makeMemoryDriver(); + bare.registerDriver(driver, true); + await bare.init(); + bare.registry.registerObject(task as any); + + await expect(bare.insert('task', { title: 'no realtime' })).resolves.toBeTruthy(); + expect(published).toHaveLength(0); + }); +}); + +/** + * #4639 — a predicate write gets its OWN contract instead of impersonating a + * per-record one. + * + * `IDataDriver.updateMany`/`deleteMany` resolve an affected COUNT, so a + * `multi: true` write can satisfy neither `DataEvent.recordId` (required) nor + * any of `before`/`after`/`changes`. #4626 correctly refused to fabricate one + * and published nothing — honest, but it meant webhooks, knowledge sync and + * `subscribeData` all went silent when a predicate write emptied half a table. + * + * These pin the third option: `data.records.updated` / `data.records.deleted`, + * an aggregate event that states the count and claims nothing else. + */ +describe('#4639 — predicate writes publish aggregate BulkDataEvents', () => { + let engine: ObjectQL; + let published: RealtimeEventPayload[]; + let realtime: IRealtimeService; + let warn: ReturnType; + + beforeEach(async () => { + published = []; + realtime = { + publish: vi.fn(async (event: RealtimeEventPayload) => { published.push(event); }), + subscribe: vi.fn(async () => 'sub-1'), + unsubscribe: vi.fn(async () => undefined), + }; + engine = new ObjectQL(); + const { driver } = makeMemoryDriver(); + engine.registerDriver(driver, true); + await engine.init(); + engine.registry.registerObject(task as any); + engine.setRealtimeService(realtime); + warn = vi.spyOn((engine as any).logger, 'warn').mockImplementation(() => undefined); + }); + + it('a multi-row update publishes ONE schema-valid data.records.updated', async () => { await engine.insert('task', [{ title: 'a', status: 'open' }, { title: 'b', status: 'open' }]); published.length = 0; - warn.mockClear(); await engine.update('task', { status: 'done' }, { multi: true, where: { status: 'open' } } as any); - // `updateMany` returns an affected COUNT: there is no record to name, and - // `DataEvent.recordId` is required. Pre-fix this published one event with - // `recordId: ''` and `after: 2` — an event every compliant consumer must - // reject. Absence is loud, not silent. - expect(published).toHaveLength(0); - const logged = warn.mock.calls.map((c) => String(c[0])).join('\n'); - expect(logged).toContain('data.record.updated'); - expect(logged).toContain('#4626'); + expect(published).toHaveLength(1); + const envelope = published[0]; + expect(envelope.type).toBe('data.records.updated'); + expect(envelope.object).toBe('task'); + + const event = BulkDataEventSchema.parse(envelope.payload); + expect(event.id).toMatch(UUID_RE); + expect(event.type).toBe('data.records.updated'); + expect(event.object).toBe('task'); + expect(event.matched).toBe(2); + expect(event.timestamp).toBe(envelope.timestamp); }); - it('publishes NOTHING for a multi-row delete — and says so', async () => { - await engine.insert('task', [{ title: 'a', status: 'stale' }, { title: 'b', status: 'stale' }]); + it('a multi-row delete publishes ONE schema-valid data.records.deleted', async () => { + await engine.insert('task', [{ title: 'a', status: 'stale' }, { title: 'b', status: 'stale' }, { title: 'c', status: 'live' }]); published.length = 0; - warn.mockClear(); await engine.delete('task', { multi: true, where: { status: 'stale' } } as any); - expect(published).toHaveLength(0); - const logged = warn.mock.calls.map((c) => String(c[0])).join('\n'); - expect(logged).toContain('data.record.deleted'); + expect(published).toHaveLength(1); + const event = BulkDataEventSchema.parse(published[0].payload); + expect(event.type).toBe('data.records.deleted'); + expect(event.matched).toBe(2); }); - it('a publish failure never fails the write itself', async () => { - (realtime.publish as ReturnType).mockRejectedValueOnce(new Error('transport down')); + it('a bulk event is NOT a DataEvent — it cannot be mistaken for a per-record one', async () => { + await engine.insert('task', [{ title: 'a', status: 'open' }]); + published.length = 0; - const record = await engine.insert('task', { title: 'still written' }); - expect(record.id).toBeTruthy(); - expect(await engine.findOne('task', { where: { id: record.id } })).toMatchObject({ title: 'still written' }); + await engine.update('task', { status: 'done' }, { multi: true, where: { status: 'open' } } as any); + + // The whole point of a separate type: a consumer validating against the + // per-record contract REJECTS this rather than reading `recordId` as an + // empty string (the pre-#4626 fabrication) or as `undefined` (what a + // widened `DataEvent` would have given it). + const payload = published[0].payload as Record; + expect(DataEventSchema.safeParse(payload).success).toBe(false); + expect(payload.recordId).toBeUndefined(); + expect(payload.after).toBeUndefined(); + expect(payload.changes).toBeUndefined(); }); - it('publishes nothing at all when no realtime service is configured', async () => { - const bare = new ObjectQL(); + it('does NOT carry the query predicate (it embeds composed security scoping)', async () => { + await engine.insert('task', [{ title: 'a', status: 'open' }]); + published.length = 0; + + await engine.update('task', { status: 'done' }, { multi: true, where: { status: 'open' } } as any); + + // The only predicate available at publish time is the middleware-COMPOSED + // AST, whose `where` carries the security layer's injected row scoping + // (RLS, sharing). Shipping that to an external webhook URL would disclose + // tenant scoping internals, so the event states the count and stops. + const payload = published[0].payload as Record; + expect(payload.where).toBeUndefined(); + expect(JSON.stringify(payload)).not.toContain('status'); + }); + + it('carries userId when the execution context names an actor', async () => { + await engine.insert('task', [{ title: 'a', status: 'open' }]); + published.length = 0; + + await engine.update( + 'task', + { status: 'done' }, + { multi: true, where: { status: 'open' }, context: { userId: 'usr_789' } } as any, + ); + + expect(BulkDataEventSchema.parse(published[0].payload).userId).toBe('usr_789'); + }); + + it('publishes NOTHING when the predicate matched no rows', async () => { + await engine.insert('task', [{ title: 'a', status: 'open' }]); + published.length = 0; + + await engine.update('task', { status: 'done' }, { multi: true, where: { status: 'nonexistent' } } as any); + + // No rows matched → no data changed → not a data event. This is what keeps + // an idle hourly LifecycleService sweep from becoming one webhook delivery + // per object per hour saying "0 records". + expect(published).toHaveLength(0); + }); + + it('publishes NOTHING when the driver breaks its count contract — and says so', async () => { + const offContract = new ObjectQL(); const { driver } = makeMemoryDriver(); - bare.registerDriver(driver, true); - await bare.init(); - bare.registry.registerObject(task as any); + // `updateMany` is contracted to resolve the affected count. A driver that + // resolves something else leaves `matched` unknowable, and `matched` is the + // entire substance of a bulk event — so none is published. + driver.updateMany = async () => ({ acknowledged: true } as any); + offContract.registerDriver(driver, true); + await offContract.init(); + offContract.registry.registerObject(task as any); + offContract.setRealtimeService(realtime); + const offWarn = vi.spyOn((offContract as any).logger, 'warn').mockImplementation(() => undefined); + await offContract.insert('task', [{ title: 'a', status: 'open' }]); + published.length = 0; + + await offContract.update('task', { status: 'done' }, { multi: true, where: { status: 'open' } } as any); - await expect(bare.insert('task', { title: 'no realtime' })).resolves.toBeTruthy(); expect(published).toHaveLength(0); + const logged = offWarn.mock.calls.map((c) => String(c[0])).join('\n'); + expect(logged).toContain('data.records.updated'); + expect(logged).toContain('#4639'); + }); + + it('a by-id delete still takes the PER-RECORD path even with multi: true', async () => { + const [record] = await engine.insert('task', [{ title: 'reaped' }]); + published.length = 0; + + // The shape LifecycleService's guarded reap uses: a scalar `where.id` is a + // single-record target regardless of `multi`, so it must keep producing a + // per-record event. Only an operator predicate routes to deleteMany. + await engine.delete('task', { where: { id: record.id }, multi: true } as any); + + expect(published).toHaveLength(1); + expect(published[0].type).toBe('data.record.deleted'); + expect(DataEventSchema.parse(published[0].payload).recordId).toBe(record.id); + }); + + it('a publish failure never fails the predicate write itself', async () => { + await engine.insert('task', [{ title: 'a', status: 'open' }]); + published.length = 0; + (realtime.publish as ReturnType).mockRejectedValueOnce(new Error('transport down')); + + await expect( + engine.update('task', { status: 'done' }, { multi: true, where: { status: 'open' } } as any), + ).resolves.not.toThrow(); + expect(await engine.findOne('task', { where: { status: 'done' } })).toBeTruthy(); + expect(warn).toHaveBeenCalled(); }); }); diff --git a/packages/objectql/src/engine.ts b/packages/objectql/src/engine.ts index 04fe9ddbfe..4f93b7d517 100644 --- a/packages/objectql/src/engine.ts +++ b/packages/objectql/src/engine.ts @@ -59,7 +59,12 @@ export type InsertManyRowOutcome = | { ok: false; error: unknown }; import { CoreServiceName, StorageNameMapping } from '@objectstack/spec/system'; import { IRealtimeService, RealtimeEventPayload } from '@objectstack/spec/contracts'; -import { DataEventSchema, type DataEvent } from '@objectstack/spec/api'; +import { + BulkDataEventSchema, + DataEventSchema, + type BulkDataEvent, + type DataEvent, +} from '@objectstack/spec/api'; import type { ICryptoProvider, CryptoHandle } from '@objectstack/spec/contracts'; import { collectSecretFields, @@ -654,6 +659,21 @@ function eventUserId(execCtx?: ExecutionContextInput): string | undefined { return asString === '' ? undefined : asString; } +/** + * Coerce a multi-row driver result into `BulkDataEvent.matched` (#4639). + * + * `IDataDriver.updateMany`/`deleteMany` are contracted to resolve the affected + * row count (`Promise`). A driver that resolves something else has not + * met that contract, and the count is the ONLY substantive thing a bulk event + * says — so this returns `undefined` and the caller declines to publish rather + * than inventing a `matched: 0` that reads as "nothing was affected" when rows + * very likely were. + */ +function eventMatchedCount(value: unknown): number | undefined { + if (typeof value !== 'number' || !Number.isInteger(value) || value < 0) return undefined; + return value; +} + export class ObjectQL implements IObjectQLEngine { /** * Ambient transaction store (ADR-0034). While a `transaction()` callback @@ -2010,13 +2030,19 @@ export class ObjectQL implements IObjectQLEngine { * carries the complete `DataEvent`, and the client SDK unwraps + validates * it at the boundary instead of double-casting the envelope. * + * Only per-record writes reach here. A predicate (`multi: true`) write has + * its own contract — see {@link publishBulkDataEvent} (#4639) — and the + * multi branches of `update()`/`delete()` route to it directly, so they + * never arrive at the identity gate below. + * * Two loud-by-design gates: - * - **No record identity → no event.** `DataEvent.recordId` is required and - * a bulk `updateMany`/`deleteMany` returns only a count, so a multi-row - * write has no truthful per-record event to publish. It publishes NONE - * (warn log naming the gap) instead of the pre-#4626 fabrication - * (`recordId: ''`, `after: `) that every schema-compliant - * consumer must reject. Tracked for a real bulk contract in #4639. + * - **No record identity → no event.** `DataEvent.recordId` is required, so + * a write that names no record has no truthful per-record event to + * publish. It publishes NONE (warn log) instead of the pre-#4626 + * fabrication (`recordId: ''`, `after: `) that every + * schema-compliant consumer must reject. Reaching this gate now means a + * single-id write whose driver returned no usable primary key — a driver + * bug — because the bulk callers no longer come through here. * - The event body is `DataEventSchema.parse`d before publish, so a * malformed producer fails here (warn log, event not published) rather * than delivering a lie downstream. @@ -2039,10 +2065,10 @@ export class ObjectQL implements IObjectQLEngine { const recordId = eventRecordId(input.recordId); if (!recordId) { this.logger.warn( - `No data.record.${action} event published for '${object}': the write names no single record ` + - `(a multi-row updateMany/deleteMany returns only an affected count), and DataEvent.recordId ` + - `is required — refusing to publish an off-contract event ` + - `(#4626; bulk event contract tracked in #4639)`, + `No data.record.${action} event published for '${object}': the write names no single record, ` + + `and DataEvent.recordId is required — refusing to publish an off-contract event. ` + + `A predicate write publishes data.records.${action} instead (#4639), so reaching this ` + + `means a single-id write whose driver returned no usable primary key (#4626)`, { object }, ); return; @@ -2078,6 +2104,87 @@ export class ObjectQL implements IObjectQLEngine { } } + /** + * Publish a realtime {@link BulkDataEvent} for a predicate write (#4639). + * + * The `multi: true` branches of `update()`/`delete()` reach + * `IDataDriver.updateMany`/`deleteMany`, which resolve an affected COUNT and + * nothing else. That is too little for the per-record `DataEvent` contract + * (`recordId` is required), so those writes were silent from #4626 until now + * — honest, but it meant webhooks and every other event consumer saw + * nothing at all when a predicate write emptied half a table. + * + * They now get their own event instead of impersonating a per-record one: + * `data.records.updated` / `data.records.deleted`, carrying the object and + * the count. A consumer reading the type knows immediately that no + * `recordId` is coming — the failure mode of the pre-#4626 fabrication, + * where `recordId: ''` looked like a record until you tried to use it. + * + * Deliberately NOT carried: the query predicate. The only one in hand here + * is the middleware-composed AST, whose `where` embeds the security layer's + * injected row scoping (RLS, sharing) — publishing it would ship tenant + * internals to whatever external URL a webhook points at. See + * `BulkDataEventSchema`'s TSDoc for the full reasoning. + * + * Same two disciplines as the per-record twin: validate before publish, and + * never throw — a realtime transport problem must not roll back a committed + * write. + */ + private async publishBulkDataEvent( + action: 'updated' | 'deleted', + object: string, + input: { matched: unknown; context?: ExecutionContextInput }, + ): Promise { + if (!this.realtimeService) return; + + const matched = eventMatchedCount(input.matched); + if (matched === undefined) { + this.logger.warn( + `No data.records.${action} event published for '${object}': the driver's multi-row result is ` + + `not an affected-row count (IDataDriver.updateMany/deleteMany are contracted to resolve ` + + `a number). The count is the only thing a bulk event states, so publishing one here would ` + + `assert something unverified (#4639)`, + { object }, + ); + return; + } + + // A predicate that matched nothing changed no data, so it is not a data + // event — the per-record path is silent for the same reason (no rows + // written, no events). This is what keeps an idle hourly LifecycleService + // sweep from becoming a webhook delivery per object per hour saying + // "0 records". Debug, not warn: matching nothing is normal, not a fault. + if (matched === 0) { + this.logger.debug(`No data.records.${action} event for '${object}': predicate matched no rows`, { object }); + return; + } + + try { + const timestamp = new Date().toISOString(); + const userId = eventUserId(input.context); + const event: BulkDataEvent = BulkDataEventSchema.parse({ + id: generateEventUuid(), + type: `data.records.${action}`, + object, + matched, + ...(userId !== undefined ? { userId } : {}), + timestamp, + }); + + const envelope: RealtimeEventPayload = { + type: event.type, + object, + payload: { ...event }, + timestamp, + }; + + await this.realtimeService.publish(envelope); + this.logger.debug(`Published data.records.${action} event`, { object, matched }); + } catch (error) { + this.logger.warn('Failed to publish bulk data event', { object, matched, error }); + } + } + /** * Set the i18n service used to localize write-path validation messages and * the field labels inside them (#3957). Bridged by `ObjectQLPlugin` on start, @@ -4371,6 +4478,14 @@ export class ObjectQL implements IObjectQLEngine { try { let result; + // [#4639] Which event contract this write reports under. A predicate + // write publishes the aggregate `data.records.updated`; anything else + // publishes the per-record `data.record.updated`. Recorded at the + // branch that made the choice rather than re-derived at the publish + // site from an absent id — the two are the same today, and a future + // driver returning a row from `updateMany` would silently reroute a + // bulk write onto the per-record contract if we inferred it. + let isPredicateWrite = false; // Pre-update snapshot. Exposed to after-hooks via `hookContext.previous` // (the HookContext contract documents `previous` for update/delete) and // reused for object-level validation rules. Fetched once, only for @@ -4501,6 +4616,7 @@ export class ObjectQL implements IObjectQLEngine { opCtx.data as Record, opCtx.context, updateMsgCtx, ); result = await driver.updateMany(object, ast, hookContext.input.data as Record, hookContext.input.options as any); + isPredicateWrite = true; } else { throw new Error('Update requires an ID or options.multi=true'); } @@ -4520,17 +4636,25 @@ export class ObjectQL implements IObjectQLEngine { // that moved to a different parent updates BOTH old and new parent. const summaryFailures = await this.recomputeSummaries(object, result, priorRecord, opCtx.context); - // Publish the data.record.updated DataEvent (#4626). A multi-row - // update names no single record (`updateMany` returns a count), so - // `publishDataEvent` declines rather than fabricating a recordId. + // Publish the update event under whichever contract this write can + // honour: per-record `data.record.updated` (#4626), or the aggregate + // `data.records.updated` for a predicate write, whose driver call + // returns an affected count and names no row (#4639). if (this.realtimeService) { - const resultId = (typeof result === 'object' && result && 'id' in result) ? (result as any).id : undefined; - await this.publishDataEvent('updated', object, { - recordId: hookContext.input.id ?? resultId, - changes: hookContext.input.data, - after: result, - context: opCtx.context, - }); + if (isPredicateWrite) { + await this.publishBulkDataEvent('updated', object, { + matched: result, + context: opCtx.context, + }); + } else { + const resultId = (typeof result === 'object' && result && 'id' in result) ? (result as any).id : undefined; + await this.publishDataEvent('updated', object, { + recordId: hookContext.input.id ?? resultId, + changes: hookContext.input.data, + after: result, + context: opCtx.context, + }); + } } // The record IS updated; a summary that could not recompute after @@ -4722,6 +4846,9 @@ export class ObjectQL implements IObjectQLEngine { try { let result; + // [#4639] See update()'s twin: recorded at the branch that chose the + // driver call, not inferred later from a missing id. + let isPredicateWrite = false; // Capture the row's FK values BEFORE deletion so roll-up summaries can // recompute the (now-orphaned) parent. Only when a summary aggregates // this object — avoids an extra read on every delete. @@ -4749,6 +4876,7 @@ export class ObjectQL implements IObjectQLEngine { ); } result = await driver.deleteMany(object, ast, hookContext.input.options as any); + isPredicateWrite = true; } else { throw new Error('Delete requires an ID or options.multi=true'); } @@ -4762,15 +4890,22 @@ export class ObjectQL implements IObjectQLEngine { ? await this.recomputeSummaries(object, null, summaryPrev, opCtx.context) : []; - // Publish the data.record.deleted DataEvent (#4626). Same rule as - // update: a multi-row delete (`deleteMany` → count) names no record, - // so no per-record event is fabricated for it. + // Same split as update(): per-record `data.record.deleted` (#4626), + // or the aggregate `data.records.deleted` when the delete was by + // predicate (#4639). if (this.realtimeService) { - const resultId = (typeof result === 'object' && result && 'id' in result) ? (result as any).id : undefined; - await this.publishDataEvent('deleted', object, { - recordId: hookContext.input.id ?? resultId, - context: opCtx.context, - }); + if (isPredicateWrite) { + await this.publishBulkDataEvent('deleted', object, { + matched: result, + context: opCtx.context, + }); + } else { + const resultId = (typeof result === 'object' && result && 'id' in result) ? (result as any).id : undefined; + await this.publishDataEvent('deleted', object, { + recordId: hookContext.input.id ?? resultId, + context: opCtx.context, + }); + } } // The record IS deleted; a summary that could not recompute after diff --git a/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts b/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts index b0dac6cd5c..fdfb10df21 100644 --- a/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts +++ b/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts @@ -18,7 +18,7 @@ import { randomUUID } from 'node:crypto'; import { describe, expect, it, vi } from 'vitest'; -import { DataEventSchema } from '@objectstack/spec/api'; +import { BulkDataEventSchema, DataEventSchema } from '@objectstack/spec/api'; import type { IDataEngine, IRealtimeService, @@ -157,6 +157,28 @@ function event( return { type: payload.type, object, payload: { ...payload }, timestamp }; } +/** + * A `data.records.*` envelope whose `payload` is a full `BulkDataEvent` + * (#4639) — what the engine publishes for a predicate (`multi: true`) write. + * Built through the spec schema for the same reason as {@link event}: the + * fixture cannot drift from the contract the enqueuer reads. + */ +function bulkEvent( + type: 'updated' | 'deleted', + object: string, + matched: number, + timestamp = '2026-05-24T00:00:00.000Z', +): RealtimeEventPayload { + const payload = BulkDataEventSchema.parse({ + id: randomUUID(), + type: `data.records.${type}`, + object, + matched, + timestamp, + }); + return { type: payload.type, object, payload: { ...payload }, timestamp }; +} + async function flush() { await new Promise((r) => setTimeout(r, 0)); } @@ -504,3 +526,155 @@ describe('AutoEnqueuer', () => { await ae.stop(); }); }); + +/** + * #4639 — predicate writes dispatch under their OWN opt-in triggers. + * + * A `multi: true` update/delete publishes `data.records.*` carrying a count and + * no record. Routing that to the existing `update`/`delete` subscribers would + * hand them a body missing every field they read, so it gets its own trigger + * pair: `bulk_update` / `bulk_delete`. + */ +describe('AutoEnqueuer — bulk data events (#4639)', () => { + it('dispatches data.records.updated to a bulk_update subscriber', async () => { + const engine = new FakeEngine({ sys_webhook: [webhook({ triggers: 'bulk_update' })] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0 }); + await ae.start(); + + await realtime.publish(bulkEvent('updated', 'contact', 40)); + await flush(); + + expect(calls).toHaveLength(1); + expect(calls[0].label).toBe('data.records.updated'); + const payload = calls[0].payload as any; + expect(payload.matched).toBe(40); + expect(payload.object).toBe('contact'); + expect(payload.action).toBe('updated'); + // No record to name — that is the contract, not an omission. + expect(payload.recordId).toBeUndefined(); + expect(payload.after).toBeUndefined(); + await ae.stop(); + }); + + it('dispatches data.records.deleted to a bulk_delete subscriber', async () => { + const engine = new FakeEngine({ sys_webhook: [webhook({ triggers: ['bulk_delete'] })] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0 }); + await ae.start(); + + await realtime.publish(bulkEvent('deleted', 'contact', 7)); + await flush(); + + expect(calls).toHaveLength(1); + expect(calls[0].label).toBe('data.records.deleted'); + expect((calls[0].payload as any).matched).toBe(7); + await ae.stop(); + }); + + it('does NOT deliver a bulk event to a per-record update subscriber', async () => { + // The opt-in half of the decision: an existing `update` webhook keeps + // receiving only bodies shaped the way it already reads them. + const engine = new FakeEngine({ sys_webhook: [webhook({ triggers: 'create,update,delete' })] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0 }); + await ae.start(); + + await realtime.publish(bulkEvent('updated', 'contact', 40)); + await realtime.publish(bulkEvent('deleted', 'contact', 40)); + await flush(); + + expect(calls).toHaveLength(0); + await ae.stop(); + }); + + it('does NOT deliver a per-record event to a bulk-only subscriber', async () => { + const engine = new FakeEngine({ sys_webhook: [webhook({ triggers: 'bulk_update,bulk_delete' })] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0 }); + await ae.start(); + + await realtime.publish(event('updated', 'contact', { id: 'c-1' })); + await flush(); + + expect(calls).toHaveLength(0); + await ae.stop(); + }); + + it('drops an off-contract bulk event instead of guessing a count', 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(); + + // `matched` is the entire substance of a bulk delivery: a wrong one is + // worse than none, so the event is dropped loudly (same discipline as + // the #4626 per-record `recordId` check). + await realtime.publish({ + type: 'data.records.updated', + object: 'contact', + payload: { id: randomUUID(), type: 'data.records.updated', object: 'contact' }, + timestamp: '2026-05-24T00:00:00.000Z', + }); + await flush(); + + expect(calls).toHaveLength(0); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining('BulkDataEvent'), + expect.anything(), + ); + await ae.stop(); + }); + + it('dedups on the event uuid, so same-millisecond sweeps do not collapse', async () => { + const engine = new FakeEngine({ sys_webhook: [webhook({ triggers: 'bulk_delete' })] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0 }); + await ae.start(); + + // Two DISTINCT sweeps sharing a timestamp. A `${object}:${action}:${ts}` + // key would silently drop the second; the producer's per-event uuid + // keeps them apart. + const ts = '2026-05-24T00:00:00.000Z'; + const first = bulkEvent('deleted', 'contact', 3, ts); + const second = bulkEvent('deleted', 'contact', 5, ts); + await realtime.publish(first); + await realtime.publish(second); + await flush(); + expect(calls).toHaveLength(2); + + // …while a genuine redelivery of the SAME event still collapses. + await realtime.publish(first); + await flush(); + expect(calls).toHaveLength(2); + await ae.stop(); + }); + + it('self-heals the cache when sys_webhook is changed by a predicate write', async () => { + // Deactivating every webhook on an object is a bulk update. If only + // `data.record.*` refreshed the cache, the enqueuer would keep + // dispatching from rows the admin just turned off. + const engine = new FakeEngine({ sys_webhook: [webhook()] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { refreshIntervalMs: 0 }); + await ae.start(); + + engine.rows.sys_webhook[0].active = false; + await realtime.publish(bulkEvent('updated', 'sys_webhook', 1)); + await flush(); + + await realtime.publish(event('created', 'contact', { id: 'c-1' })); + await flush(); + + expect(calls).toHaveLength(0); + await ae.stop(); + }); +}); diff --git a/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts b/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts index 13f2c62c1e..f959dd56a2 100644 --- a/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts +++ b/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts @@ -1,8 +1,17 @@ // Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. import type { IDataEngine, IRealtimeService, RealtimeEventPayload } from '@objectstack/spec/contracts'; +import type { WebhookTriggerType } from '@objectstack/spec/automation'; import type { EnqueueHttpInput } from '@objectstack/service-messaging'; +/** + * The authored trigger vocabulary, taken from the spec rather than restated + * here — this file both validates authored triggers and maps events onto them, + * so a locally-spelled union would be a second contract free to drift from the + * one authors are validated against. + */ +type WebhookTrigger = WebhookTriggerType; + /** * Enqueue callback into the shared `service-messaging` HTTP outbox (ADR-0018 M3). * The plugin supplies one bound to `messaging.enqueueHttp(...)`; webhooks no @@ -29,7 +38,7 @@ interface CachedSubscription { id: string; name: string; objectName: string | undefined; // empty = matches all objects - triggers: Set<'create' | 'update' | 'delete'>; + triggers: Set; url: string; method?: string; headers?: Record; @@ -90,6 +99,11 @@ export interface AutoEnqueuerOptions { * `eventId` is computed from `${object}:${recordId}:${type}:${timestamp}` * so the outbox dedup index catches duplicates that could arise from * upstream replay or buggy producers — and is stable across nodes. + * + * An aggregate `data.records.*` event (#4639) has no record to key on, so it + * dedups on the producer's event uuid instead: two predicate sweeps in the + * same millisecond are genuinely different events and must not collapse into + * one delivery, which a timestamp-based key would do. */ export class AutoEnqueuer { private readonly subscriptions = new Map(); @@ -236,12 +250,13 @@ export class AutoEnqueuer { if (unknown.length > 0) { this.logger.warn?.( `[webhook-auto-enqueuer] webhook '${(row.name as string) ?? row.id}' declares trigger(s) the engine never emits: ` + - `${unknown.join(', ')} — ignored. Dispatchable triggers: create, update, delete.`, + `${unknown.join(', ')} — ignored. Dispatchable triggers: ` + + `${[...DISPATCHABLE_WEBHOOK_TRIGGERS].join(', ')}.`, { id: row.id, unknown }, ); } const triggers = new Set( - normalized.filter((t) => DISPATCHABLE_WEBHOOK_TRIGGERS.has(t)) as Array<'create' | 'update' | 'delete'>, + normalized.filter((t) => DISPATCHABLE_WEBHOOK_TRIGGERS.has(t)) as WebhookTrigger[], ); if (triggers.size === 0) { // [ADR-0078 Phase 4] No dispatchable triggers — the webhook can @@ -260,7 +275,8 @@ export class AutoEnqueuer { `[webhook-auto-enqueuer] webhook '${(row.name as string) ?? row.id}' has no dispatchable ` + `triggers — it will NEVER fire (rule webhook/without-triggers): there is no manual fire ` + `path (#3196), so this row is dead while looking armed in Setup. Declare ` + - `triggers: ['create'|'update'|'delete'], or set it inactive if it should be off.`, + `one of: ${[...DISPATCHABLE_WEBHOOK_TRIGGERS].join(', ')}, or set it inactive if it ` + + `should be off.`, { id: row.id }, ); return null; @@ -303,10 +319,18 @@ export class AutoEnqueuer { * webhook persistence. */ private handleEvent(event: RealtimeEventPayload): void { - if (!event.type?.startsWith('data.record.')) return; if (!event.object) return; if (event.object === this.subscriptionsObject) return; // self-heal handles its own + // [#4639] A predicate write publishes the aggregate `data.records.*` + // instead, which has no record to describe — separate path, separate + // trigger, separate delivery shape. + if (event.type?.startsWith('data.records.')) { + this.handleBulkEvent(event); + return; + } + if (!event.type?.startsWith('data.record.')) return; + const action = event.type.slice('data.record.'.length) as | 'created' | 'updated' | 'deleted' | string; const trigger = mapActionToTrigger(action); @@ -389,9 +413,101 @@ export class AutoEnqueuer { } } + /** + * Handler for aggregate `data.records.*` events — a predicate write + * (`multi: true`) that the driver reports only as an affected-row count + * (#4639). + * + * Deliberately NOT folded into {@link handleEvent}'s per-record path. The + * delivered body has no `recordId` and no record fields, so a subscriber + * to `update` that started receiving these would get a payload missing + * everything it reads — which is how the pre-#4626 `recordId: ''` + * fabrication broke consumers, just arriving from the other side. A + * webhook opts in with `bulk_update` / `bulk_delete`. + */ + private handleBulkEvent(event: RealtimeEventPayload): void { + const action = event.type.slice('data.records.'.length); + const trigger = mapBulkActionToTrigger(action); + if (!trigger) return; + + const subs = [ + ...(this.subscriptions.get(event.object!) ?? []), + ...(this.subscriptions.get('*') ?? []), + ]; + if (subs.length === 0) return; + + // Same contract discipline as the per-record path: the payload IS the + // spec's `BulkDataEvent`, whose `matched` the engine validates before + // publishing. An off-contract event is dropped loudly rather than + // delivered with a guessed count — `matched` is the entire substance + // of a bulk delivery, so a wrong one is worse than none. + const payload = event.payload ?? {}; + const matched = (payload as { matched?: unknown }).matched; + if (typeof matched !== 'number' || !Number.isInteger(matched) || matched < 0) { + this.logger.warn?.( + '[webhook-auto-enqueuer] dropping off-contract bulk data event: payload is not a ' + + 'BulkDataEvent (no top-level non-negative integer `matched`) — fix the producer', + { type: event.type, object: event.object }, + ); + return; + } + + // A predicate write has no natural key to build a deterministic id + // from — `${object}:${action}:${timestamp}` would collide between two + // sweeps landing in the same millisecond, and silently drop the + // second. The producer's own event uuid is generated once and travels + // with the event, so it dedups redelivery of the SAME event without + // ever conflating two distinct ones. + const eventUuid = (payload as { id?: unknown }).id; + if (typeof eventUuid !== 'string' || eventUuid === '') { + this.logger.warn?.( + '[webhook-auto-enqueuer] dropping off-contract bulk data event: payload has no ' + + 'top-level string `id` to dedup on — fix the producer', + { type: event.type, object: event.object }, + ); + return; + } + const eventId = `${event.object}:${event.type}:${eventUuid}`; + + for (const sub of subs) { + if (!sub.triggers.has(trigger)) continue; + + void this.enqueue({ + source: 'webhook', + refId: sub.id, + dedupKey: `${sub.id}:${eventId}`, + label: event.type, + url: sub.url, + method: sub.method, + headers: sub.headers, + signingSecret: sub.secret, + timeoutMs: sub.timeoutMs, + // [#3946] Envelope keys last so the payload cannot rewrite them. + payload: { + ...payload, + object: event.object, + matched, + action, + timestamp: event.timestamp, + }, + }).catch((err) => + this.logger.warn?.('[webhook-auto-enqueuer] bulk enqueue failed', { + webhook: sub.name, + eventId, + err: (err as Error)?.message ?? err, + }), + ); + } + } + private handleSelfHealEvent(event: RealtimeEventPayload): void { if (event.object !== this.subscriptionsObject) return; - if (!event.type?.startsWith('data.record.')) return; + // [#4639] A predicate write over `sys_webhook` (deactivate every + // webhook on an object, say) changes the subscription set exactly like + // a per-record edit does, so it must refresh the cache too — matching + // only `data.record.` would leave the enqueuer dispatching from rows + // the admin just turned off. + if (!event.type?.startsWith('data.record.') && !event.type?.startsWith('data.records.')) return; this.refresh().catch((err) => this.logger.warn?.('[webhook-auto-enqueuer] self-heal refresh failed', err), ); @@ -418,5 +534,23 @@ function mapActionToTrigger( } } +/** [#4639] `data.records.{action}` → its opt-in bulk trigger. */ +function mapBulkActionToTrigger(action: string): 'bulk_update' | 'bulk_delete' | null { + switch (action) { + case 'updated': + return 'bulk_update'; + case 'deleted': + return 'bulk_delete'; + default: + return null; + } +} + /** The trigger values the enqueuer can actually map from an emitted record event. */ -const DISPATCHABLE_WEBHOOK_TRIGGERS: ReadonlySet = new Set(['create', 'update', 'delete']); +const DISPATCHABLE_WEBHOOK_TRIGGERS: ReadonlySet = new Set([ + 'create', + 'update', + 'delete', + 'bulk_update', + 'bulk_delete', +]); diff --git a/packages/plugins/plugin-webhooks/src/sys-webhook.object.ts b/packages/plugins/plugin-webhooks/src/sys-webhook.object.ts index 1c3868d9df..56179dd1a2 100644 --- a/packages/plugins/plugin-webhooks/src/sys-webhook.object.ts +++ b/packages/plugins/plugin-webhooks/src/sys-webhook.object.ts @@ -127,7 +127,7 @@ export const SysWebhook = ObjectSchema.create({ }), triggers: Field.select( - ['create', 'update', 'delete'], + ['create', 'update', 'delete', 'bulk_update', 'bulk_delete'], { label: 'Triggers', required: false, @@ -135,7 +135,12 @@ export const SysWebhook = ObjectSchema.create({ // an array; the auto-enqueuer parser also tolerates the legacy // comma-separated / JSON-string forms so existing rows keep working. multiple: true, - description: 'Record events that fire this webhook', + // [#4639] `bulk_*` fire on predicate writes (`multi: true`), whose + // delivery carries `matched` instead of a record — opt-in precisely + // because that body is a different shape. Kept in step with + // `WebhookTriggerType` (`@objectstack/spec/automation`), which is the + // contract the auto-enqueuer validates against. + description: 'Record events that fire this webhook (bulk_* deliver a count, not a record)', group: 'Definition', }, ), diff --git a/packages/plugins/plugin-webhooks/src/translations/en.objects.generated.ts b/packages/plugins/plugin-webhooks/src/translations/en.objects.generated.ts index 4afb76f019..1f51af055b 100644 --- a/packages/plugins/plugin-webhooks/src/translations/en.objects.generated.ts +++ b/packages/plugins/plugin-webhooks/src/translations/en.objects.generated.ts @@ -30,11 +30,13 @@ export const enObjects: NonNullable = { }, triggers: { label: "Triggers", - help: "Comma-separated event list: create,update,delete", + help: "Record events that fire this webhook. bulk_update / bulk_delete fire on predicate writes and deliver a count, not a record.", options: { create: "create", update: "update", - delete: "delete" + delete: "delete", + bulk_update: "bulk_update", + bulk_delete: "bulk_delete" } }, url: { diff --git a/packages/plugins/plugin-webhooks/src/translations/es-ES.objects.generated.ts b/packages/plugins/plugin-webhooks/src/translations/es-ES.objects.generated.ts index f4f3b4b132..b1092f7318 100644 --- a/packages/plugins/plugin-webhooks/src/translations/es-ES.objects.generated.ts +++ b/packages/plugins/plugin-webhooks/src/translations/es-ES.objects.generated.ts @@ -30,11 +30,13 @@ export const esESObjects: NonNullable = { }, triggers: { label: "Desencadenantes", - help: "Lista de eventos separada por comas: create,update,delete.", + help: "Eventos de registro que activan este Webhook. bulk_update / bulk_delete se activan en escrituras por predicado y entregan un recuento, no un registro.", options: { create: "Crear", update: "Actualizar", - delete: "Eliminar" + delete: "Eliminar", + bulk_update: "Actualización masiva", + bulk_delete: "Eliminación masiva" } }, url: { diff --git a/packages/plugins/plugin-webhooks/src/translations/ja-JP.objects.generated.ts b/packages/plugins/plugin-webhooks/src/translations/ja-JP.objects.generated.ts index ad7d56af47..72365f80f2 100644 --- a/packages/plugins/plugin-webhooks/src/translations/ja-JP.objects.generated.ts +++ b/packages/plugins/plugin-webhooks/src/translations/ja-JP.objects.generated.ts @@ -30,11 +30,13 @@ export const jaJPObjects: NonNullable = { }, triggers: { label: "トリガー", - help: "カンマ区切りのイベントリスト: create,update,delete", + help: "このウェブフックを発火するレコードイベント。bulk_update / bulk_delete は述語による一括書き込みで発火し、レコードではなく件数を配信します。", options: { create: "作成", update: "更新", - delete: "削除" + delete: "削除", + bulk_update: "一括更新", + bulk_delete: "一括削除" } }, url: { diff --git a/packages/plugins/plugin-webhooks/src/translations/zh-CN.objects.generated.ts b/packages/plugins/plugin-webhooks/src/translations/zh-CN.objects.generated.ts index d4a1f50811..639fc6aae8 100644 --- a/packages/plugins/plugin-webhooks/src/translations/zh-CN.objects.generated.ts +++ b/packages/plugins/plugin-webhooks/src/translations/zh-CN.objects.generated.ts @@ -30,11 +30,13 @@ export const zhCNObjects: NonNullable = { }, triggers: { label: "触发器", - help: "以逗号分隔的事件列表：create,update,delete", + help: "触发该 Webhook 的记录事件。bulk_update / bulk_delete 由谓词写（批量）触发，投递的是受影响条数而非记录。", options: { create: "创建", update: "更新", - delete: "删除" + delete: "删除", + bulk_update: "批量更新", + bulk_delete: "批量删除" } }, url: { diff --git a/packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts b/packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts index 61957bfcbe..1a728d4ac6 100644 --- a/packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts +++ b/packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts @@ -79,6 +79,20 @@ const DELETED: RealtimeEventPayload = { timestamp: '2026-08-02T12:00:01.000Z', }; +/** [#4639] What a predicate (`multi: true`) write publishes: a count, no rows. */ +const BULK_DELETED: RealtimeEventPayload = { + type: 'data.records.deleted', + object: 'task', + payload: { + id: '9c8b7a65-4321-4fed-8ba9-876543210fed', + type: 'data.records.deleted', + object: 'task', + matched: 12, + timestamp: '2026-08-02T12:00:02.000Z', + }, + timestamp: '2026-08-02T12:00:02.000Z', +}; + describe('#4626 — KnowledgeServicePlugin event sync on data.record.*', () => { it('upserts the RECORD (payload.after), not the event envelope', async () => { const harness = makeCtx(); @@ -114,4 +128,25 @@ describe('#4626 — KnowledgeServicePlugin event sync on data.record.*', () => { expect(upsert).not.toHaveBeenCalled(); }); + + it('[#4639] says out loud that a predicate write leaves the index stale', async () => { + const harness = makeCtx(); + const { service, deliver } = await harness.boot(new KnowledgeServicePlugin()); + const upsert = vi.spyOn(service, 'handleRecordUpsert').mockResolvedValue(undefined); + const del = vi.spyOn(service, 'handleRecordDelete').mockResolvedValue(undefined); + + await deliver(BULK_DELETED); + + // A knowledge index is a per-record projection and `matched: 12` names no + // record, so there is nothing to upsert or delete — the adapters take + // neither a count nor a predicate. The index is now stale in a way this + // subscription cannot repair, and a silent no-op here would read exactly + // like "nothing happened". Reconciliation is tracked in #4672. + expect(upsert).not.toHaveBeenCalled(); + expect(del).not.toHaveBeenCalled(); + expect(harness.ctx.logger.warn).toHaveBeenCalledWith( + expect.stringContaining('may now be stale'), + expect.objectContaining({ object: 'task', matched: 12 }), + ); + }); }); diff --git a/packages/services/service-knowledge/src/knowledge-service-plugin.ts b/packages/services/service-knowledge/src/knowledge-service-plugin.ts index 51dad6d423..79b66fd2c4 100644 --- a/packages/services/service-knowledge/src/knowledge-service-plugin.ts +++ b/packages/services/service-knowledge/src/knowledge-service-plugin.ts @@ -163,6 +163,31 @@ export class KnowledgeServicePlugin implements Plugin { return; } + // [#4639] Aggregate events from a predicate write (`multi: true`). + // A knowledge index is a PER-RECORD projection, and `matched: 40` + // names no record — there is no upsert or delete this handler could + // derive from it, and none of the adapters take a predicate. So the + // index is now stale in a way this subscription cannot repair. + // + // Say so rather than falling through to the `return` below: a silent + // no-op here reads identically to "nothing happened", which is how the + // gap stayed invisible before #4639 gave bulk writes an event at all. + // The durable fix is a reconciliation pass over the object source + // (events keep the index FRESH; reconciliation keeps it CORRECT) — + // tracked separately in #4672. + if (type === 'data.records.updated' || type === 'data.records.deleted') { + const matched = payload.matched; + ctx.logger.warn?.( + `KnowledgeServicePlugin: '${object}' had a predicate write (${type}) affecting ` + + `${typeof matched === 'number' ? matched : 'an unreported number of'} record(s). ` + + 'A bulk event carries a count, not records, so the knowledge index for this object ' + + 'may now be stale and cannot be repaired from the event stream (#4639; ' + + 'reconciliation tracked in #4672).', + { object, type, matched }, + ); + return; + } + if (type === 'record.created' || type === 'record.updated') { const record = (payload.record as Record | undefined) ?? payload; diff --git a/packages/spec/api-surface.json b/packages/spec/api-surface.json index c92efda61b..810f89430d 100644 --- a/packages/spec/api-surface.json +++ b/packages/spec/api-surface.json @@ -2309,7 +2309,7 @@ "Webhook (type)", "WebhookInput (type)", "WebhookSchema (const)", - "WebhookTriggerType (const)", + "WebhookTriggerType (type)", "analyzeRegion (function)", "approverTypeIsOrgScoped (function)", "canonicalApproverType (function)", @@ -2463,6 +2463,9 @@ "BatchUpdateRequestSchema (const)", "BatchUpdateResponse (type)", "BatchUpdateResponseSchema (const)", + "BulkDataEvent (type)", + "BulkDataEventSchema (const)", + "BulkDataEventType (type)", "BulkRequest (type)", "BulkRequestSchema (const)", "BulkResponse (type)", diff --git a/packages/spec/authorable-surface.json b/packages/spec/authorable-surface.json index 189f5c110f..daed1d360b 100644 --- a/packages/spec/authorable-surface.json +++ b/packages/spec/authorable-surface.json @@ -704,6 +704,12 @@ "api/BatchUpdateResponse:succeeded", "api/BatchUpdateResponse:success", "api/BatchUpdateResponse:total", + "api/BulkDataEvent:id", + "api/BulkDataEvent:matched", + "api/BulkDataEvent:object", + "api/BulkDataEvent:timestamp", + "api/BulkDataEvent:type", + "api/BulkDataEvent:userId", "api/BulkRequest:allOrNone", "api/BulkRequest:records", "api/BulkResponse:data", diff --git a/packages/spec/docs-import-surface.baseline.json b/packages/spec/docs-import-surface.baseline.json index d49939a787..da5c463dcb 100644 --- a/packages/spec/docs-import-surface.baseline.json +++ b/packages/spec/docs-import-surface.baseline.json @@ -25,7 +25,6 @@ "automation/GuardRef — no type export", "automation/StateMachine — no type export", "automation/StateNode — no type export", - "automation/WebhookTriggerType — no type export", "data/AddressValue — no type export", "data/AggregationFunction — no type export", "data/AggregationMetricType — no type export", diff --git a/packages/spec/json-schema.manifest.json b/packages/spec/json-schema.manifest.json index 9163533b16..399f0af1a2 100644 --- a/packages/spec/json-schema.manifest.json +++ b/packages/spec/json-schema.manifest.json @@ -130,6 +130,8 @@ "api/BatchRecord", "api/BatchUpdateRequest", "api/BatchUpdateResponse", + "api/BulkDataEvent", + "api/BulkDataEventType", "api/BulkRequest", "api/BulkResponse", "api/CacheControl", diff --git a/packages/spec/liveness/webhook.json b/packages/spec/liveness/webhook.json index 9188b0eb73..3c8f110055 100644 --- a/packages/spec/liveness/webhook.json +++ b/packages/spec/liveness/webhook.json @@ -1,6 +1,6 @@ { "type": "webhook", - "_note": "WebhookSchema (outbound webhook — packages/spec/src/automation/webhook.zod.ts). Governed via a spec-only schema override in the gate (SPEC_ONLY_SCHEMAS): webhook is still NOT a registered metadata type (absent from kernel/metadata-type-schemas.ts) — registering it would turn on Studio webhook CRUD + saveMetaItem overlay + create-seeds; that reassessment is tracked in #3490 and deliberately deferred. TWO THINGS CLOSED THE OLD 'entire surface is dead' classification: (1) #3494 PRUNED the aspirational dead props outright — body / payloadFields / includeSession / retryPolicy / tags / authentication are gone from the schema; (2) the #3461 materializer bridge (PR #3489) makes every REMAINING prop live. `bootstrapDeclaredWebhooks` (plugins/plugin-webhooks/src/bootstrap-declared-webhooks.ts) materializes each stack/connector-authored webhook into a `sys_webhook` DATA row on boot — WebhookSchema.parse (:114) → mapWebhookToRow (:191): `object`→`object_name` (:195), `isActive`→`active` (:202), same-named `name`/`label`/`triggers`/`url`/`method`/`description`, and the FULL envelope → `definition_json` (:203). The dispatcher (AutoEnqueuer) reads those rows (auto-enqueuer.ts:175 `where:{active:true}`) and fans out on data.record.* events, reading `object_name`/`name`/`url`/`method`/`triggers` off the row and `headers`/`secret`/`timeoutMs` back out of `definition_json` (auto-enqueuer.ts:266-277). So all 11 remaining props are LIVE; there is no dead/experimental prop left, so nothing carries an authorWarn (the old per-webhook `url` heads-up is gone — authoring is no longer a no-op). Seed-not-clobber: an admin-edited row (`customized`) is never re-seeded (bootstrap-declared-webhooks.ts:143). The `object` prop carries the ADR-0054 runtime proof for the whole materialization pipeline (bound high-risk class `webhook-materialization`). Field-level line refs: materializer bootstrap-declared-webhooks.ts, runtime object plugins/plugin-webhooks/src/sys-webhook.object.ts, dispatcher auto-enqueuer.ts.", + "_note": "WebhookSchema (outbound webhook — packages/spec/src/automation/webhook.zod.ts). Governed via a spec-only schema override in the gate (SPEC_ONLY_SCHEMAS): webhook is still NOT a registered metadata type (absent from kernel/metadata-type-schemas.ts) — registering it would turn on Studio webhook CRUD + saveMetaItem overlay + create-seeds; that reassessment is tracked in #3490 and deliberately deferred. TWO THINGS CLOSED THE OLD 'entire surface is dead' classification: (1) #3494 PRUNED the aspirational dead props outright — body / payloadFields / includeSession / retryPolicy / tags / authentication are gone from the schema; (2) the #3461 materializer bridge (PR #3489) makes every REMAINING prop live. `bootstrapDeclaredWebhooks` (plugins/plugin-webhooks/src/bootstrap-declared-webhooks.ts) materializes each stack/connector-authored webhook into a `sys_webhook` DATA row on boot — WebhookSchema.parse (:114) → mapWebhookToRow (:191): `object`→`object_name` (:195), `isActive`→`active` (:202), same-named `name`/`label`/`triggers`/`url`/`method`/`description`, and the FULL envelope → `definition_json` (:203). The dispatcher (AutoEnqueuer) reads those rows (auto-enqueuer.ts:175 `where:{active:true}`) and fans out on data.record.* events (plus the aggregate data.records.* a predicate write publishes, dispatched under the opt-in bulk_update/bulk_delete triggers — #4639), reading `object_name`/`name`/`url`/`method`/`triggers` off the row and `headers`/`secret`/`timeoutMs` back out of `definition_json` (auto-enqueuer.ts:266-277). So all 11 remaining props are LIVE; there is no dead/experimental prop left, so nothing carries an authorWarn (the old per-webhook `url` heads-up is gone — authoring is no longer a no-op). Seed-not-clobber: an admin-edited row (`customized`) is never re-seeded (bootstrap-declared-webhooks.ts:143). The `object` prop carries the ADR-0054 runtime proof for the whole materialization pipeline (bound high-risk class `webhook-materialization`). Field-level line refs: materializer bootstrap-declared-webhooks.ts, runtime object plugins/plugin-webhooks/src/sys-webhook.object.ts, dispatcher auto-enqueuer.ts.", "props": { "name": { "status": "live", @@ -20,7 +20,7 @@ }, "triggers": { "status": "live", - "evidence": "Materialized to sys_webhook.triggers (bootstrap-declared-webhooks.ts:196, #3489); the dispatcher parses + maps them to create/update/delete (auto-enqueuer.ts:212, unknown values dropped with a warning #3196).", + "evidence": "Materialized to sys_webhook.triggers (bootstrap-declared-webhooks.ts:196, #3489); the dispatcher parses + maps them to create/update/delete from data.record.* and to bulk_update/bulk_delete from the aggregate data.records.* a predicate write publishes (auto-enqueuer.ts, unknown values dropped with a warning #3196, bulk pair #4639).", "note": "Was dead pre-bridge. Salvaged 1:1." }, "url": { diff --git a/packages/spec/src/api/events.zod.ts b/packages/spec/src/api/events.zod.ts index c01b9df4ae..56dd07c0e0 100644 --- a/packages/spec/src/api/events.zod.ts +++ b/packages/spec/src/api/events.zod.ts @@ -73,6 +73,39 @@ export const DataEventType = z.enum([ export type DataEventType = z.infer; +/** + * Bulk Data Event Types + * + * Triggered when a predicate write touches an unbounded set of records: + * `ObjectQL.update()` / `delete()` with `options.multi = true`, which reach + * `IDataDriver.updateMany` / `deleteMany`. Follows the pattern + * `data.records.{action}` — plural, to read differently at a glance from the + * per-record `data.record.{action}` above. + * + * These exist because the driver contract for a multi-row write returns an + * affected COUNT and nothing else (`Promise` — see + * `contracts/data-driver.ts`), so there is no record identity to put in + * {@link DataEventSchema}'s required `recordId`. Before #4639 the engine had + * only two options, and both were wrong: fabricate a per-record event with + * `recordId: ''` (which every schema-compliant consumer must reject — the + * pre-#4626 behaviour), or publish nothing at all (#4626's honest-but-silent + * stopgap). A bulk write now gets its OWN contract instead of impersonating a + * per-record one: a consumer that receives `data.records.updated` can see from + * the type alone that no `recordId` is coming, rather than discovering it as an + * empty string at runtime. + * + * There is deliberately no `data.records.created`: a batch insert knows every + * row it wrote, so the engine publishes N per-record `data.record.created` + * events for it (each with its own event id) — a count would be strictly less + * information than the contract already delivers. + */ +export const BulkDataEventType = z.enum([ + 'data.records.updated', + 'data.records.deleted', +]); + +export type BulkDataEventType = z.infer; + /** * Metadata Event Payload * @@ -143,3 +176,55 @@ export const DataEventSchema = lazySchema(() => z.object({ })); export type DataEvent = z.infer; + +/** + * Bulk Data Event Payload + * + * Represents a predicate write (`multi: true` update/delete) that affected + * `matched` records without naming any of them. Travels in the same + * `RealtimeEventPayload` envelope as {@link DataEventSchema}, and is validated + * by the producer before publish — but it is a SEPARATE contract, not a + * degraded `DataEvent`: there is no `recordId`, no `before`/`after`, and no + * `changes`, because a multi-row driver call returns none of them. + * + * **What a consumer can and cannot do with this.** It can count, alert, audit + * at aggregate level, or mark a derived index as needing reconciliation. It + * CANNOT incrementally maintain a per-record projection (search index, cache, + * mirror) from it — `matched: 40` names no rows. A consumer whose correctness + * depends on knowing *which* records changed must reconcile against the source + * of truth; that is a property of the driver contract, not a gap in this + * schema, and no field added here could fix it. + * + * **Why there is no `where`.** The only predicate in hand at publish time is + * the middleware-COMPOSED query AST — the caller's filter after the security + * layer injected its row scoping (RLS write filter, sharing's editable-rows + * filter). Publishing it would ship tenant scoping internals to whatever + * external URL a webhook points at. The caller's own pre-composition filter is + * no better: it is a second, divergent answer to "what did this write touch". + * So the event reports the count it can state truthfully and stops there. + */ +export const BulkDataEventSchema = lazySchema(() => z.object({ + /** Unique event identifier */ + id: z.string().uuid().describe('Unique event identifier'), + + /** Event type (data.records.{action}) */ + type: BulkDataEventType.describe('Event type'), + + /** Object name */ + object: z.string().describe('Object name'), + + /** + * Number of records the predicate write affected. The ObjectQL engine does + * not publish an event at all when this would be `0` — a predicate that + * matched nothing changed no data — so a consumer will not see idle sweeps. + */ + matched: z.number().int().nonnegative().describe('Number of records affected'), + + /** User who triggered the event */ + userId: z.string().optional().describe('User who triggered the event'), + + /** Event timestamp (ISO 8601) */ + timestamp: z.string().datetime().describe('Event timestamp'), +})); + +export type BulkDataEvent = z.infer; diff --git a/packages/spec/src/automation/webhook.zod.ts b/packages/spec/src/automation/webhook.zod.ts index f343f06b68..8f113355c1 100644 --- a/packages/spec/src/automation/webhook.zod.ts +++ b/packages/spec/src/automation/webhook.zod.ts @@ -13,6 +13,17 @@ import { SnakeCaseIdentifierSchema } from '../shared/identifiers.zod'; * producer are declared here — an author can't subscribe to something that * never fires. * + * **Bulk triggers (#4639).** `bulk_update` / `bulk_delete` map to the engine's + * aggregate `data.records.updated` / `data.records.deleted`, emitted when a + * predicate write (`multi: true` → `IDataDriver.updateMany`/`deleteMany`) + * affects a set of rows the driver reports only as a count. They are separate + * trigger values, not extra sources for `update` / `delete`, because their + * delivery has a different SHAPE: no `recordId`, no record body, just + * `object` + `matched`. Folding them into the per-record triggers would send + * every existing subscriber a body missing the fields it reads — the same + * class of breakage as the pre-#4626 `recordId: ''` fabrication, arriving from + * the other direction. A webhook that wants both subscribes to both. + * * Deliberately NOT triggers (#3196): * - `undelete` — there is no soft-delete / restore capability in the engine * (`delete` is a hard delete; no `deleted_at` convention, no restore @@ -29,8 +40,12 @@ export const WebhookTriggerType = z.enum([ 'create', 'update', 'delete', + 'bulk_update', + 'bulk_delete', ]); +export type WebhookTriggerType = z.infer; + /** * CANONICAL WEBHOOK DEFINITION * @@ -90,7 +105,7 @@ export const WebhookSchema = lazySchema(() => z.object({ label: z.string().optional().describe('Human-readable webhook label'), /** Scope */ - object: z.string().optional().describe('Object whose record events (create/update/delete) trigger this webhook'), + object: z.string().optional().describe('Object whose record events (create/update/delete, bulk_update/bulk_delete) trigger this webhook'), triggers: z.array(WebhookTriggerType).optional().describe('Events that trigger execution'), /** Target */