From a0b48e7947d665cf0950535bdfc752679a7c440d Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 14:09:15 +0000 Subject: [PATCH] fix(objectql,client): subscribeData delivers a real DataEvent (#4626) The producer (ObjectQL engine) published a raw RealtimeEventPayload envelope with `{ recordId, after, changes }` nested under `payload` and never generated `id`/`userId`, while `@objectstack/client`'s `subscribeData` force-cast that envelope into the callback (`callback(event as any as DataEvent)`). Subscribers reading the declared top-level `event.recordId` / `event.changes` compiled green and got `undefined` at runtime. Data-side twin of #4602. Producer fulfils the contract: insert/update/delete build a true DataEvent (uuid `id`, flattened top-level fields, `userId` from the execution context) and `DataEventSchema.parse` it before publish. A multi-row updateMany/deleteMany names no single record, so it publishes nothing (warn) instead of the previous `recordId: ''` fabrication; bulk contract tracked in #4639. Consumers read the fulfilled shape: the client validates at the boundary and rejects off-contract payloads loudly; the webhook auto-enqueuer drops its `recordId ?? id ?? after?.id ?? 'unknown'` tolerance chain; service-knowledge reads the record from `after` and the delete id from `recordId` instead of indexing the envelope as if it were the row. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012C2cd7tL8QDoZ2QKN3djJ5 --- .changeset/data-event-contract.md | 60 +++++ content/docs/automation/webhooks.mdx | 30 ++- packages/client/src/realtime-api-data.test.ts | 176 ++++++++++++ packages/client/src/realtime-api.ts | 33 ++- .../objectql/src/engine-data-events.test.ts | 253 ++++++++++++++++++ packages/objectql/src/engine.ts | 225 +++++++++++----- .../plugin-webhooks/src/auto-enqueuer.test.ts | 54 +++- .../plugin-webhooks/src/auto-enqueuer.ts | 37 ++- .../__tests__/event-sync-data-events.test.ts | 117 ++++++++ .../src/knowledge-service-plugin.ts | 33 ++- 10 files changed, 919 insertions(+), 99 deletions(-) create mode 100644 .changeset/data-event-contract.md create mode 100644 packages/client/src/realtime-api-data.test.ts create mode 100644 packages/objectql/src/engine-data-events.test.ts create mode 100644 packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts diff --git a/.changeset/data-event-contract.md b/.changeset/data-event-contract.md new file mode 100644 index 0000000000..366b4bc3f5 --- /dev/null +++ b/.changeset/data-event-contract.md @@ -0,0 +1,60 @@ +--- +"@objectstack/objectql": minor +"@objectstack/client": patch +"@objectstack/plugin-webhooks": patch +"@objectstack/service-knowledge": patch +--- + +fix(objectql,client): `subscribeData` callbacks receive real `DataEvent`s — the producer now fulfils the declared contract (#4626) + +`@objectstack/spec/api`'s `DataEvent` declares top-level `id` (uuid, +required), `type`, `object`, `recordId` (required), `changes?`, `before?`, +`after?`, `userId?`, `timestamp`. But the producer (the ObjectQL engine) +published a raw `RealtimeEventPayload` envelope with `{ recordId, after, +changes }` nested under `payload` and never generated `id`/`userId`, while the +client SDK force-cast that envelope into the callback (`callback(event as any +as DataEvent)`). Subscribers who wrote `event.recordId` / `event.changes` — +exactly what the types promised — compiled green and read `undefined` at +runtime. The data-side twin of #4602. + +Producer now fulfils the contract: + +- `ObjectQL.insert()` / `update()` / `delete()` build a true `DataEvent` + (generated uuid `id`, flattened top-level fields, `userId` from the + execution context when the write names an actor) and validate it with + `DataEventSchema.parse` before publishing. The transport envelope is + unchanged (`RealtimeEventPayload`, with `payload` carrying the complete + `DataEvent`), so subscribers keep receiving `{ type, object, payload, + timestamp }` on the wire. +- A batch insert publishes one event **per record** (as before), each with its + own event id. +- **A multi-row write (`multi: true` → `updateMany` / `deleteMany`) now + publishes nothing.** Those driver methods return only an affected count, so + there is no record for a required `recordId` to name; the engine logs a + warning naming the gap instead of publishing the previous fabrication + (`recordId: ''`, `after: `), which every schema-compliant + consumer had to reject. **Consequence: webhooks and knowledge sync no longer + fire for bulk writes** — they previously fired once with an unusable body. A + real bulk event contract is tracked in #4639. + +Consumers validate or read the fulfilled shape instead of guessing: + +- `@objectstack/client`'s `subscribeData` (and therefore + `@objectstack/client-react`'s `useDataSubscription` / + `useDataSubscriptionCallback` / `useAutoRefresh`, which delegate to it) + unwraps the envelope and runs `DataEventSchema.safeParse` at the boundary. + An off-contract payload is rejected loudly (handler error, callback never + invoked) — never coerced or passed through. The `as any as DataEvent` + double-cast is gone, and the `recordId` option now filters on the fulfilled + event. +- `@objectstack/plugin-webhooks`' auto-enqueuer reads the required + `recordId` directly; its `recordId ?? id ?? after?.id ?? before?.id ?? + 'unknown'` fallback chain is gone, and an off-contract event is dropped with + a warning rather than delivered under the literal id `'unknown'`. Delivered + webhook bodies now also carry the event's `id`/`type`/`userId`; the record + itself stays nested under `after` and the envelope keys (`object`, + `recordId`, `action`, `timestamp`) still win. +- `@objectstack/service-knowledge`'s event sync reads the record from `after` + (create/update) and the id from `recordId` (delete) for `data.record.*`. + It previously indexed the envelope itself as if it were the row, and never + resolved an id for deletes. diff --git a/content/docs/automation/webhooks.mdx b/content/docs/automation/webhooks.mdx index 3d578aa663..22a497ff4c 100644 --- a/content/docs/automation/webhooks.mdx +++ b/content/docs/automation/webhooks.mdx @@ -192,9 +192,33 @@ Five stages, each implemented as a thin layer over an existing primitive. Producers (the ObjectQL engine's insert/update/delete handlers) call `IRealtimeService.publish(event)` after the write commits, where `event` is a -plain `{ type, object, payload, timestamp }` record — e.g. -`{ type: 'data.record.updated', object: 'account', payload: { recordId, -changes, after }, timestamp: }`. +plain `{ type, object, payload, timestamp }` transport envelope whose `payload` +is the spec's `DataEvent` (`@objectstack/spec/api`) — validated against +`DataEventSchema` before it is published (#4626): + +```ts +{ + type: 'data.record.updated', + object: 'account', + timestamp: '', + payload: { + id: '', // unique event id + type: 'data.record.updated', + object: 'account', + recordId: 'acc_1', // REQUIRED — the record the event is about + changes: { status: 'active' },// update only: the submitted payload + after: { id: 'acc_1', … }, // create/update only: the written row + userId: 'usr_1', // when the write names an actor + timestamp: '', + }, +} +``` + +> **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). > **Not yet cluster-aware.** The only shipped `IRealtimeService` implementation > is `InMemoryRealtimeAdapter`, an in-process, single-node pub/sub with no diff --git a/packages/client/src/realtime-api-data.test.ts b/packages/client/src/realtime-api-data.test.ts new file mode 100644 index 0000000000..4c73ae088e --- /dev/null +++ b/packages/client/src/realtime-api-data.test.ts @@ -0,0 +1,176 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #4626 — subscribeData delivers TRUE `DataEvent`s, validated at the boundary + * (the data-side twin of #4602's metadata pins). + * + * The callback is typed `(event: DataEvent) => void` — top-level `id` (uuid), + * `type`, `object`, `recordId` (required), `changes?`, `before?`, `after?`, + * `userId?`, `timestamp`. Before this fix the handler delivered the raw + * `RealtimeEventPayload` envelope via `callback(event as any as DataEvent)`, + * so `event.recordId` / `event.changes` / `event.id` were `undefined` at + * runtime while the types said otherwise. + * + * Pins: + * - the subscriber receives the top-level fields (fails on the pre-fix + * envelope passthrough); + * - an off-contract payload — including the pre-fix producer's + * `{ recordId, after }` shape — is rejected LOUDLY: callback never invoked, + * error surfaced, nothing coerced; + * - the `recordId` filter narrows on the FULFILLED event. + */ + +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import type { RealtimeEventPayload } from '@objectstack/spec/contracts'; +import { RealtimeAPI } from './realtime-api'; + +const VALID_EVENT = { + id: 'f47ac10b-58cc-4372-a567-0e02b2c3d479', + type: 'data.record.updated', + object: 'project_task', + recordId: 'task_1', + changes: { status: 'done' }, + after: { id: 'task_1', title: 'Ship it', status: 'done' }, + userId: 'usr_123', + timestamp: '2026-08-02T12:00:00.000Z', +} as const; + +function envelopeOf( + payload: Record, + type = 'data.record.updated', + object = 'project_task', +): RealtimeEventPayload { + return { type, object, payload, timestamp: '2026-08-02T12:00:00.000Z' }; +} + +describe('#4626 — RealtimeAPI.subscribeData 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); // poll interval drains the buffer + } + + it('delivers the DataEvent with top-level fields to the callback', () => { + const seen: unknown[] = []; + api.subscribeData('project_task', (event) => seen.push(event)); + + deliver(envelopeOf({ ...VALID_EVENT })); + + expect(seen).toHaveLength(1); + const event = seen[0] as typeof VALID_EVENT; + // Top-level, as the type declares — NOT nested under `payload`. + expect(event.id).toBe(VALID_EVENT.id); + expect(event.type).toBe('data.record.updated'); + expect(event.object).toBe('project_task'); + expect(event.recordId).toBe('task_1'); + expect(event.changes).toEqual({ status: 'done' }); + expect(event.after).toEqual({ id: 'task_1', title: 'Ship it', status: 'done' }); + expect(event.userId).toBe('usr_123'); + expect(event.timestamp).toBe(VALID_EVENT.timestamp); + }); + + it('rejects the pre-fix producer shape LOUDLY instead of passing it through', () => { + // The old engine payload: `{ recordId, after }` with no id/type/object/ + // timestamp. The envelope carried those — which is precisely why the + // double-cast compiled and lied. + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined); + const callback = vi.fn(); + api.subscribeData('project_task', callback); + + deliver(envelopeOf({ recordId: 'task_1', after: { id: 'task_1', title: 'Ship it' } })); + + expect(callback).not.toHaveBeenCalled(); + expect(errorSpy).toHaveBeenCalled(); + const logged = String(errorSpy.mock.calls.map((c) => c.join(' ')).join('\n')); + expect(logged).toContain('realtime event handler'); + }); + + it('rejects a payload with a wrong field type instead of coercing it', () => { + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined); + const callback = vi.fn(); + api.subscribeData('project_task', callback); + + deliver(envelopeOf({ ...VALID_EVENT, id: 'not-a-uuid' })); + expect(callback).not.toHaveBeenCalled(); + + // A numeric recordId is a producer bug, not something to String() here. + deliver(envelopeOf({ ...VALID_EVENT, recordId: 42 })); + expect(callback).not.toHaveBeenCalled(); + expect(errorSpy).toHaveBeenCalled(); + }); + + it('filters by recordId on the fulfilled event', () => { + const callback = vi.fn(); + api.subscribeData('project_task', callback, { recordId: 'task_2' }); + + deliver(envelopeOf({ ...VALID_EVENT })); + expect(callback).not.toHaveBeenCalled(); + + deliver(envelopeOf({ ...VALID_EVENT, recordId: 'task_2', after: { id: 'task_2' } })); + expect(callback).toHaveBeenCalledTimes(1); + expect(callback.mock.calls[0][0].recordId).toBe('task_2'); + }); + + it('ignores events for another object', () => { + const callback = vi.fn(); + api.subscribeData('project_task', callback); + + deliver(envelopeOf( + { ...VALID_EVENT, object: 'account', recordId: 'acc_1' }, + 'data.record.updated', + 'account', + )); + + expect(callback).not.toHaveBeenCalled(); + }); + + it('delivers created and deleted events too', () => { + const seen: any[] = []; + api.subscribeData('project_task', (event) => seen.push(event)); + + deliver(envelopeOf({ + id: '9f8b0f6e-1c2d-4a3b-8c9d-0e1f2a3b4c5d', + type: 'data.record.created', + object: 'project_task', + recordId: 'task_9', + after: { id: 'task_9', title: 'New' }, + timestamp: '2026-08-02T12:00:01.000Z', + }, 'data.record.created')); + + deliver(envelopeOf({ + id: '1b2c3d4e-5f60-4718-9a2b-3c4d5e6f7081', + type: 'data.record.deleted', + object: 'project_task', + recordId: 'task_9', + timestamp: '2026-08-02T12:00:02.000Z', + }, 'data.record.deleted')); + + expect(seen.map((e) => e.type)).toEqual(['data.record.created', 'data.record.deleted']); + expect(seen.map((e) => e.recordId)).toEqual(['task_9', 'task_9']); + expect(seen[1].after).toBeUndefined(); + }); + + it('unsubscribe stops delivery', () => { + const callback = vi.fn(); + const off = api.subscribeData('project_task', callback); + + deliver(envelopeOf({ ...VALID_EVENT })); + expect(callback).toHaveBeenCalledTimes(1); + + off(); + deliver(envelopeOf({ ...VALID_EVENT })); + expect(callback).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/client/src/realtime-api.ts b/packages/client/src/realtime-api.ts index 7d3446d89e..d39fe072dd 100644 --- a/packages/client/src/realtime-api.ts +++ b/packages/client/src/realtime-api.ts @@ -8,7 +8,12 @@ */ import type { RealtimeEventPayload } from '@objectstack/spec/contracts'; -import { MetadataEventSchema, type MetadataEvent, type DataEvent } from '@objectstack/spec/api'; +import { + MetadataEventSchema, + DataEventSchema, + type MetadataEvent, + type DataEvent, +} from '@objectstack/spec/api'; export interface RealtimeSubscriptionFilter { /** Metadata/object type filter */ @@ -121,12 +126,28 @@ export class RealtimeAPI { ] }, handler: (event) => { - // Type guard and filter - if (event.type.startsWith('data.') && event.object === object) { - if (!options?.recordId || (event.payload as any)?.recordId === options.recordId) { - callback(event as any as DataEvent); - } + if (!event.type.startsWith('data.') || event.object !== object) 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 + // is typed `(event: DataEvent) => void`, so delivering the envelope + // itself (what `event as any as DataEvent` used to do) left every + // subscriber reading `undefined` for the top-level `recordId` / + // `changes` / `id` the type promised. An off-contract payload is + // rejected LOUDLY (throw → surfaced by emitEvent's handler-error log), + // never coerced or passed through: a malformed event means the + // producer is broken and must be fixed there, not tolerated here. + const parsed = DataEventSchema.safeParse(event.payload); + if (!parsed.success) { + throw new Error( + `subscribeData('${object}'): event '${event.type}' payload does not satisfy ` + + `DataEventSchema — rejecting off-contract event (fix the producer): ${parsed.error.message}` + ); } + // Narrow on the FULFILLED event, not the envelope — `recordId` is a + // declared top-level field of what the subscriber receives. + if (options?.recordId && parsed.data.recordId !== options.recordId) return; + callback(parsed.data); } }); diff --git a/packages/objectql/src/engine-data-events.test.ts b/packages/objectql/src/engine-data-events.test.ts new file mode 100644 index 0000000000..2721aca2b9 --- /dev/null +++ b/packages/objectql/src/engine-data-events.test.ts @@ -0,0 +1,253 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #4626 — the ObjectQL engine publishes TRUE `DataEvent`s (contract-first). + * + * `@objectstack/spec/api`'s `DataEvent` is the declared contract for realtime + * record changes: top-level `id` (uuid), `type`, `object`, `recordId` + * (REQUIRED), `changes?`, `before?`, `after?`, `userId?`, `timestamp`. Before + * this fix the engine published a bare `RealtimeEventPayload` envelope with + * `{ recordId, after, changes }` nested under `payload` and never generated + * `id`/`userId` — so every `subscribeData` subscriber that wrote + * `event.recordId` / `event.changes` (exactly what the types promised) read + * `undefined` at runtime. + * + * These tests pin the producer half of the contract: + * - 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 publish failure never fails the write. + */ + +import { describe, it, expect, beforeEach, vi } from 'vitest'; +import { DataEventSchema } from '@objectstack/spec/api'; +import type { IRealtimeService, RealtimeEventPayload } from '@objectstack/spec/contracts'; +import { ObjectQL } from './engine.js'; + +const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + +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))); }, + // The driver contract returns an AFFECTED COUNT for the bulk verbs — the + // reason a multi-row write can name no record (see the pins below). + 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, stores }; +} + +describe('#4626 — engine writes publish true DataEvents', () => { + 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('insert publishes an envelope whose payload IS a schema-valid DataEvent', async () => { + const record = await engine.insert('task', { title: 'Write the pin' }); + + expect(published).toHaveLength(1); + const envelope = published[0]; + + // The transport envelope keeps its shape — nothing else on the wire moves. + expect(envelope.type).toBe('data.record.created'); + expect(envelope.object).toBe('task'); + expect(typeof envelope.timestamp).toBe('string'); + + // The payload is the full DataEvent — parsed with the SPEC schema, not a + // hand-rolled shape, so this pin fails the moment either side drifts. + const event = DataEventSchema.parse(envelope.payload); + expect(event.id).toMatch(UUID_RE); + expect(event.type).toBe('data.record.created'); + expect(event.object).toBe('task'); + expect(event.recordId).toBe(record.id); + expect(event.after).toMatchObject({ id: record.id, title: 'Write the pin' }); + expect(event.changes).toBeUndefined(); + expect(event.timestamp).toBe(envelope.timestamp); + }); + + it('a batch insert publishes ONE event per record, each with its own uuid', async () => { + await engine.insert('task', [{ title: 'a' }, { title: 'b' }, { title: 'c' }]); + + expect(published).toHaveLength(3); + const events = published.map((e) => DataEventSchema.parse(e.payload)); + expect(events.map((e) => (e.after as any).title)).toEqual(['a', 'b', 'c']); + expect(events.every((e) => e.type === 'data.record.created')).toBe(true); + expect(events.every((e) => UUID_RE.test(e.id))).toBe(true); + expect(new Set(events.map((e) => e.id)).size).toBe(3); + expect(new Set(events.map((e) => e.recordId)).size).toBe(3); + }); + + it('update publishes changes AND recordId at the TOP LEVEL (the #4626 defect)', async () => { + const record = await engine.insert('task', { title: 'v1', status: 'open' }); + published.length = 0; + + await engine.update('task', { id: record.id, status: 'done' }); + + expect(published).toHaveLength(1); + const event = DataEventSchema.parse(published[0].payload); + expect(event.type).toBe('data.record.updated'); + // Pre-fix these were `undefined` on what the subscriber received. + expect(event.recordId).toBe(record.id); + expect(event.changes).toMatchObject({ status: 'done' }); + expect(event.after).toMatchObject({ id: record.id, status: 'done', title: 'v1' }); + expect(event.id).toMatch(UUID_RE); + }); + + it('delete publishes a schema-valid event with recordId and no after', async () => { + const record = await engine.insert('task', { title: 'gone' }); + published.length = 0; + + await engine.delete('task', { where: { id: record.id } } as any); + + expect(published).toHaveLength(1); + const event = DataEventSchema.parse(published[0].payload); + expect(event.type).toBe('data.record.deleted'); + expect(event.recordId).toBe(record.id); + expect(event.after).toBeUndefined(); + expect(event.changes).toBeUndefined(); + }); + + it('carries userId when the execution context names an actor', async () => { + const record = await engine.insert('task', { title: 'mine' }, { context: { userId: 'usr_123' } } as any); + const created = DataEventSchema.parse(published[0].payload); + expect(created.userId).toBe('usr_123'); + + published.length = 0; + await engine.update('task', { id: record.id, title: 'edited' }, { context: { userId: 'usr_456' } } as any); + expect(DataEventSchema.parse(published[0].payload).userId).toBe('usr_456'); + }); + + it('omits userId for system-initiated writes (no actor known)', async () => { + await engine.insert('task', { title: 'boot' }); + expect(DataEventSchema.parse(published[0].payload).userId).toBeUndefined(); + }); + + it('publishes NOTHING for a multi-row update — and says so', 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'); + }); + + it('publishes NOTHING for a multi-row delete — and says so', async () => { + await engine.insert('task', [{ title: 'a', status: 'stale' }, { title: 'b', status: 'stale' }]); + 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'); + }); + + 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); + }); +}); diff --git a/packages/objectql/src/engine.ts b/packages/objectql/src/engine.ts index 60cb801716..a196877a18 100644 --- a/packages/objectql/src/engine.ts +++ b/packages/objectql/src/engine.ts @@ -59,6 +59,7 @@ 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 type { ICryptoProvider, CryptoHandle } from '@objectstack/spec/contracts'; import { collectSecretFields, @@ -604,6 +605,55 @@ function isEmptyReferenceValue(v: unknown): boolean { return false; } +/** + * RFC-4122 v4 uuid for the realtime `DataEvent.id` (#4626). + * + * The twin of the generator `MetadataManager` uses for `MetadataEvent.id` + * (#4602/#4628) — same shape, same fallback, kept local rather than shared so + * the engine's `core` import closure gains nothing (ADR-0076 D2 ratchet). + * Prefers `crypto.randomUUID`; the fallback keeps browser-compatible (Pure) + * environments without WebCrypto working while still satisfying + * `DataEventSchema`'s `z.string().uuid()`. + */ +function generateEventUuid(): string { + const c = globalThis.crypto; + if (c && typeof c.randomUUID === 'function') { + return c.randomUUID(); + } + return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, (ch) => { + const r = (Math.random() * 16) | 0; + const v = ch === 'x' ? r : (r & 0x3) | 0x8; + return v.toString(16); + }); +} + +/** + * Coerce a driver-returned primary key into `DataEvent.recordId` (a required + * `string`). Returns `undefined` when the write has no single record identity + * — a bulk `updateMany`/`deleteMany` returns only a count — so the caller can + * decline to publish rather than fabricate one (#4626). + */ +function eventRecordId(value: unknown): string | undefined { + if (typeof value === 'string') return value === '' ? undefined : value; + if (typeof value === 'number' || typeof value === 'bigint') return String(value); + return undefined; +} + +/** `DataEvent.changes`/`before`/`after` are `z.record(...)` — only a plain object qualifies. */ +function eventRecordBody(value: unknown): Record | undefined { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +/** `DataEvent.userId` — the acting user, when the execution context names one. */ +function eventUserId(execCtx?: ExecutionContextInput): string | undefined { + const userId = execCtx?.userId; + if (userId == null) return undefined; + const asString = String(userId); + return asString === '' ? undefined : asString; +} + export class ObjectQL implements IObjectQLEngine { /** * Ambient transaction store (ADR-0034). While a `transaction()` callback @@ -1938,6 +1988,85 @@ export class ObjectQL implements IObjectQLEngine { this.logger.info('RealtimeService configured for data events'); } + /** + * Publish a realtime {@link DataEvent} for a record write (#4626 — + * contract-first, the data-side twin of #4602/#4628). + * + * What reaches a `subscribeData` callback must BE the spec's `DataEvent` + * (`@objectstack/spec/api`): top-level `id` (uuid), `type`, `object`, + * `recordId` (REQUIRED), plus `changes`/`after`/`userId` when they apply. + * The transport keeps its `RealtimeEventPayload` envelope — `payload` + * carries the complete `DataEvent`, and the client SDK unwraps + validates + * it at the boundary instead of double-casting the envelope. + * + * 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. + * - 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. + * + * Never throws and never fails the write: a realtime transport problem must + * not roll back a committed record. + */ + private async publishDataEvent( + action: 'created' | 'updated' | 'deleted', + object: string, + input: { + recordId: unknown; + changes?: unknown; + after?: unknown; + context?: ExecutionContextInput; + }, + ): Promise { + if (!this.realtimeService) return; + + 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)`, + { object }, + ); + return; + } + + try { + const timestamp = new Date().toISOString(); + const changes = eventRecordBody(input.changes); + const after = eventRecordBody(input.after); + const userId = eventUserId(input.context); + const event: DataEvent = DataEventSchema.parse({ + id: generateEventUuid(), + type: `data.record.${action}`, + object, + recordId, + ...(changes !== undefined ? { changes } : {}), + ...(after !== undefined ? { after } : {}), + ...(userId !== undefined ? { userId } : {}), + timestamp, + }); + + const envelope: RealtimeEventPayload = { + type: event.type, + object, + payload: { ...event }, + timestamp, + }; + + await this.realtimeService.publish(envelope); + this.logger.debug(`Published data.record.${action} event`, { object, recordId }); + } catch (error) { + this.logger.warn('Failed to publish data event', { object, recordId, error }); + } + } + /** * Set the i18n service used to localize write-path validation messages and * the field labels inside them (#3957). Bridged by `ObjectQLPlugin` on start, @@ -4055,39 +4184,17 @@ export class ObjectQL implements IObjectQLEngine { // Roll-up: recompute parent summary fields that aggregate this object. const summaryFailures = await this.recomputeSummaries(object, result, null, opCtx.context); - // Publish data.record.created event to realtime service + // Publish one data.record.created DataEvent per written record (#4626). + // A batch insert is N record events, not one event about N records — + // `DataEvent.recordId` is per record. if (this.realtimeService) { - try { - if (Array.isArray(result)) { - // Bulk insert - publish event for each record - for (const record of result) { - const event: RealtimeEventPayload = { - type: 'data.record.created', - object, - payload: { - recordId: record.id, - after: record, - }, - timestamp: new Date().toISOString(), - }; - await this.realtimeService.publish(event); - } - this.logger.debug(`Published ${result.length} data.record.created events`, { object }); - } else { - const event: RealtimeEventPayload = { - type: 'data.record.created', - object, - payload: { - recordId: result.id, - after: result, - }, - timestamp: new Date().toISOString(), - }; - await this.realtimeService.publish(event); - this.logger.debug('Published data.record.created event', { object, recordId: result.id }); - } - } catch (error) { - this.logger.warn('Failed to publish data event', { object, error }); + const createdRows: any[] = Array.isArray(result) ? result : [result]; + for (const record of createdRows) { + await this.publishDataEvent('created', object, { + recordId: record?.id, + after: record, + context: opCtx.context, + }); } } @@ -4402,26 +4509,17 @@ 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 data.record.updated event to realtime service + // 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. if (this.realtimeService) { - try { - const resultId = (typeof result === 'object' && result && 'id' in result) ? (result as any).id : undefined; - const recordId = String(hookContext.input.id || resultId || ''); - const event: RealtimeEventPayload = { - type: 'data.record.updated', - object, - payload: { - recordId, - changes: hookContext.input.data, - after: result, - }, - timestamp: new Date().toISOString(), - }; - await this.realtimeService.publish(event); - this.logger.debug('Published data.record.updated event', { object, recordId }); - } catch (error) { - this.logger.warn('Failed to publish data event', { object, error }); - } + 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 @@ -4653,24 +4751,15 @@ export class ObjectQL implements IObjectQLEngine { ? await this.recomputeSummaries(object, null, summaryPrev, opCtx.context) : []; - // Publish data.record.deleted event to realtime service + // 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. if (this.realtimeService) { - try { - const resultId = (typeof result === 'object' && result && 'id' in result) ? (result as any).id : undefined; - const recordId = String(hookContext.input.id || resultId || ''); - const event: RealtimeEventPayload = { - type: 'data.record.deleted', - object, - payload: { - recordId, - }, - timestamp: new Date().toISOString(), - }; - await this.realtimeService.publish(event); - this.logger.debug('Published data.record.deleted event', { object, recordId }); - } catch (error) { - this.logger.warn('Failed to publish data event', { object, error }); - } + 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 a9a0815ed1..b0dac6cd5c 100644 --- a/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts +++ b/packages/plugins/plugin-webhooks/src/auto-enqueuer.test.ts @@ -16,7 +16,9 @@ * - The deterministic dedupKey (`:`) collapses replays. */ +import { randomUUID } from 'node:crypto'; import { describe, expect, it, vi } from 'vitest'; +import { DataEventSchema } from '@objectstack/spec/api'; import type { IDataEngine, IRealtimeService, @@ -132,18 +134,27 @@ function webhook(over: Partial = {}): any { }; } +/** + * A `data.record.*` envelope whose `payload` is a full `DataEvent` + * (`@objectstack/spec/api`) — what the engine publishes since #4626. Built + * through the spec schema so the fixture cannot drift from the contract the + * enqueuer now reads (`recordId` is a required top-level string). + */ function event( type: 'created' | 'updated' | 'deleted', object: string, record: any, timestamp = '2026-05-24T00:00:00.000Z', ): RealtimeEventPayload { - return { + const payload = DataEventSchema.parse({ + id: randomUUID(), type: `data.record.${type}`, object, - payload: { recordId: record.id, after: record }, + recordId: String(record.id), + ...(type === 'deleted' ? {} : { after: record }), timestamp, - }; + }); + return { type: payload.type, object, payload: { ...payload }, timestamp }; } async function flush() { @@ -171,6 +182,43 @@ describe('AutoEnqueuer', () => { expect(calls[0].url).toBe('https://hooks.example/wh'); expect(calls[0].label).toBe('data.record.created'); expect((calls[0].payload as any).recordId).toBe('c-1'); + // [#4626] The delivered body carries the fulfilled DataEvent — the + // record itself stays nested under `after`, and the envelope keys + // (object/recordId/action/timestamp) still win. + expect((calls[0].payload as any).after).toEqual({ id: 'c-1', name: 'Alice' }); + expect((calls[0].payload as any).object).toBe('contact'); + expect((calls[0].payload as any).action).toBe('created'); + await ae.stop(); + }); + + it('[#4626] drops an off-contract data event instead of enqueuing it as "unknown"', async () => { + // Pre-#4626 the enqueuer read `recordId ?? id ?? after?.id ?? 'unknown'`, + // so a payload that named no record still produced a delivery whose + // recordId was the literal string 'unknown'. The payload IS a DataEvent + // now: no top-level string `recordId` means the producer is broken, and + // the event is dropped loudly rather than tolerated here. + const engine = new FakeEngine({ sys_webhook: [webhook()] }); + const realtime = new FakeRealtime(); + const { enqueue, calls } = makeRecorder(); + const warn = vi.fn(); + const ae = new AutoEnqueuer(engine, realtime, enqueue, { + refreshIntervalMs: 0, + logger: { warn } as any, + }); + await ae.start(); + + await realtime.publish({ + type: 'data.record.created', + object: 'contact', + // The pre-fix engine shape for a bulk write: no usable record id. + payload: { recordId: '', after: 2 }, + timestamp: '2026-05-24T00:00:00.000Z', + }); + await flush(); + + expect(calls).toHaveLength(0); + expect(warn).toHaveBeenCalled(); + expect(String(warn.mock.calls[0][0])).toContain('off-contract'); await ae.stop(); }); diff --git a/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts b/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts index 2161a4b8db..13f2c62c1e 100644 --- a/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts +++ b/packages/plugins/plugin-webhooks/src/auto-enqueuer.ts @@ -318,13 +318,25 @@ export class AutoEnqueuer { ]; if (subs.length === 0) return; + // [#4626] The envelope's `payload` IS the spec's `DataEvent` + // (`@objectstack/spec/api`): `recordId` is a REQUIRED top-level string + // the ObjectQL engine validates before publishing. Read it directly. + // The old `recordId ?? id ?? after?.id ?? before?.id ?? 'unknown'` + // chain was consumer-side tolerance for a producer that never filled + // the contract (AGENTS.md PD #12) — and its `'unknown'` fallback + // silently turned an unnameable record into a delivered webhook. An + // off-contract event is now DROPPED loudly: the producer is broken and + // gets fixed there. const payload = event.payload ?? {}; - const recordId = - (payload as any).recordId ?? - (payload as any).id ?? - (payload as any).after?.id ?? - (payload as any).before?.id ?? - 'unknown'; + const recordId = (payload as { recordId?: unknown }).recordId; + if (typeof recordId !== 'string' || recordId === '') { + this.logger.warn?.( + '[webhook-auto-enqueuer] dropping off-contract data event: payload is not a DataEvent ' + + '(no top-level string `recordId`) — fix the producer', + { type: event.type, object: event.object }, + ); + return; + } // Deterministic eventId — same input on any node → same id. // Includes timestamp so two distinct updates to the same record @@ -352,12 +364,13 @@ export class AutoEnqueuer { timeoutMs: sub.timeoutMs, // [#3946] Envelope keys are written LAST so the event payload // cannot rewrite them. Behaviour-neutral for the engine's own - // publishers — `data.record.*` payloads are - // `{ recordId, after, changes }`, with record fields nested - // under `after`, so none of these four keys collide today. It - // is the shape that was wrong: a publisher that flattened - // record fields into the payload (the `payload.id` fallback - // above suggests some do) would have silently rewritten the + // publishers — since #4626 a `data.record.*` payload is a + // `DataEvent` (`id`, `type`, `object`, `recordId`, `changes?`, + // `after?`, `userId?`, `timestamp`), whose `object` / + // `recordId` / `timestamp` carry the SAME values written here + // and whose record fields stay nested under `after`. It is the + // shape that was wrong: a publisher that flattened record + // fields into the payload would have silently rewritten the // `object` / `action` / `timestamp` a subscriber receives. payload: { ...payload, 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 new file mode 100644 index 0000000000..61957bfcbe --- /dev/null +++ b/packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts @@ -0,0 +1,117 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #4626 — the knowledge event sync reads the FULFILLED `DataEvent`. + * + * `data.record.*` envelopes carry the spec's `DataEvent` in `payload`: the + * written row is `after`, the id is the required top-level `recordId`. The + * shared branch this sync used to take read the payload itself as if it were + * the record, so an object source indexed `{ recordId, after }` — a document + * with none of the record's fields — and a delete never resolved an id at all. + */ + +import { describe, it, expect, vi } from 'vitest'; +import type { RealtimeEventHandler, RealtimeEventPayload } from '@objectstack/spec/contracts'; +import { KnowledgeServicePlugin } from '../knowledge-service-plugin'; +import type { KnowledgeService } from '../knowledge-service'; + +function makeCtx() { + let readyHook: (() => Promise) | undefined; + let handler: RealtimeEventHandler | undefined; + let service: KnowledgeService | undefined; + + const ctx: any = { + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + hook: (name: string, fn: () => Promise) => { + if (name === 'kernel:ready') readyHook = fn; + }, + registerService: (_name: string, svc: KnowledgeService) => { service = svc; }, + getService: (name: string) => { + if (name === 'realtime') { + return { + publish: async () => undefined, + subscribe: async (_channel: string, h: RealtimeEventHandler) => { handler = h; return 'sub-1'; }, + unsubscribe: async () => undefined, + }; + } + throw new Error(`no service ${name}`); + }, + }; + + return { + ctx, + async boot(plugin: KnowledgeServicePlugin) { + await plugin.init(ctx); + await plugin.start(ctx); + await readyHook?.(); + return { + service: service!, + deliver: (event: RealtimeEventPayload) => handler!(event), + }; + }, + }; +} + +const CREATED: RealtimeEventPayload = { + type: 'data.record.created', + object: 'task', + payload: { + id: 'f47ac10b-58cc-4372-a567-0e02b2c3d479', + type: 'data.record.created', + object: 'task', + recordId: 'task_1', + after: { id: 'task_1', title: 'Ship it', notes: 'the record body' }, + timestamp: '2026-08-02T12:00:00.000Z', + }, + timestamp: '2026-08-02T12:00:00.000Z', +}; + +const DELETED: RealtimeEventPayload = { + type: 'data.record.deleted', + object: 'task', + payload: { + id: '1b2c3d4e-5f60-4718-9a2b-3c4d5e6f7081', + type: 'data.record.deleted', + object: 'task', + recordId: 'task_1', + timestamp: '2026-08-02T12:00:01.000Z', + }, + timestamp: '2026-08-02T12:00:01.000Z', +}; + +describe('#4626 — KnowledgeServicePlugin event sync on data.record.*', () => { + it('upserts the RECORD (payload.after), not the event envelope', async () => { + const harness = makeCtx(); + const { service, deliver } = await harness.boot(new KnowledgeServicePlugin()); + const upsert = vi.spyOn(service, 'handleRecordUpsert').mockResolvedValue(undefined); + + await deliver(CREATED); + + expect(upsert).toHaveBeenCalledTimes(1); + expect(upsert.mock.calls[0][0]).toBe('task'); + expect(upsert.mock.calls[0][1]).toEqual({ id: 'task_1', title: 'Ship it', notes: 'the record body' }); + }); + + it('resolves the delete id from the required top-level recordId', async () => { + const harness = makeCtx(); + const { service, deliver } = await harness.boot(new KnowledgeServicePlugin()); + const del = vi.spyOn(service, 'handleRecordDelete').mockResolvedValue(undefined); + + await deliver(DELETED); + + expect(del).toHaveBeenCalledWith('task', 'task_1'); + }); + + it('ignores a data event that carries no record body', async () => { + const harness = makeCtx(); + const { service, deliver } = await harness.boot(new KnowledgeServicePlugin()); + const upsert = vi.spyOn(service, 'handleRecordUpsert').mockResolvedValue(undefined); + + await deliver({ + ...CREATED, + payload: { ...(CREATED.payload as Record), after: undefined }, + }); + + expect(upsert).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/services/service-knowledge/src/knowledge-service-plugin.ts b/packages/services/service-knowledge/src/knowledge-service-plugin.ts index b3fd297d0f..cb20c5463c 100644 --- a/packages/services/service-knowledge/src/knowledge-service-plugin.ts +++ b/packages/services/service-knowledge/src/knowledge-service-plugin.ts @@ -132,12 +132,31 @@ export class KnowledgeServicePlugin implements Plugin { if (!object) return; const type = event.type; const payload = (event.payload ?? {}) as Record; - if ( - type === 'record.created' || - type === 'record.updated' || - type === 'data.record.created' || - type === 'data.record.updated' - ) { + + // [#4626] `data.record.*` payloads ARE the spec's `DataEvent` + // (`@objectstack/spec/api`): the record body lives in `after`, the id + // in the required top-level `recordId`. Reading the payload itself as + // a record (what the shared branch below did) indexed the ENVELOPE — + // `{ recordId, after }` — as if it were the row, so object sources + // were syncing documents with none of the record's fields, and a + // delete never resolved an id at all. Kept separate from the legacy + // `record.*` shape rather than merged behind fallbacks. + if (type === 'data.record.created' || type === 'data.record.updated') { + const record = payload.after as Record | undefined; + if (record && typeof record === 'object') { + await service.handleRecordUpsert(object, record); + } + return; + } + if (type === 'data.record.deleted') { + const recordId = payload.recordId; + if (typeof recordId === 'string' && recordId !== '') { + await service.handleRecordDelete(object, recordId); + } + return; + } + + if (type === 'record.created' || type === 'record.updated') { const record = (payload.record as Record | undefined) ?? payload; if (record && typeof record === 'object') { @@ -145,7 +164,7 @@ export class KnowledgeServicePlugin implements Plugin { } return; } - if (type === 'record.deleted' || type === 'data.record.deleted') { + if (type === 'record.deleted') { const recordObj = payload.record as Record | undefined; const id = (payload.id as string | undefined) ?? (recordObj?.id as string | undefined);