fix(objectql,client): subscribeData delivers a real DataEvent (#4626)

.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: <affected count>`), 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.

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: <ISO 8601 string> }`.
+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: '<ISO 8601>',
+  payload: {
+    id: '<uuid>',                 // 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: '<ISO 8601>',
+  },
+}
+```
+
+> **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

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<string, unknown>,
+  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);
+  });
+});

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);
       }
     });