diff --git a/.changeset/24835-driver-update-leaves-caller-data.md b/.changeset/24835-driver-update-leaves-caller-data.md new file mode 100644 index 00000000000..5fa54b11bd5 --- /dev/null +++ b/.changeset/24835-driver-update-leaves-caller-data.md @@ -0,0 +1,11 @@ +--- +'@objectstack/driver-sql': patch +--- + +fix(driver-sql): `update()` no longer writes its `updated_at` stamp into the caller's `data`, so webhook deliveries for record updates are stored again on PostgreSQL and MySQL + +Clause-②: no + +- **What was lost.** On PostgreSQL and MySQL, every webhook delivery for a record update was lost. `SqlDriver.update()` assigned its `updated_at` stamp onto the patch object the caller passed in. ObjectQL publishes that same object as the update event's `changes`. On those two dialects the stamp is a knex `Raw` (`CURRENT_TIMESTAMP`) that holds the database client, its pool and a timer. The webhook outbox could not serialise the event (`Converting circular structure to JSON`), so no `sys_http_delivery` row was written. The record write still answered 200, and the webhook still read active. Every other reader of the update event's `changes`, and of the patch after the driver call, saw the same `Raw`. On SQLite the stamp is an ISO string, so the delivery was stored, but the caller's patch and the event's `changes` still carried an `updated_at` the caller never wrote. +- **What changed.** `update()`, including its rotated-table path, stamps `updated_at` into a fresh object, as `updateMany()` already did. The caller's `data` is left exactly as it was passed. +- **Unchanged.** The stored `updated_at` value on every dialect, `preserveAudit` imports, tables without an `updated_at` column, `upsert()` (it already stamps a copy of its input), and every export and option. diff --git a/packages/drivers/driver-sql/src/sql-driver-24835-update-leaves-caller-data.test.ts b/packages/drivers/driver-sql/src/sql-driver-24835-update-leaves-caller-data.test.ts new file mode 100644 index 00000000000..3ebae5a221d --- /dev/null +++ b/packages/drivers/driver-sql/src/sql-driver-24835-update-leaves-caller-data.test.ts @@ -0,0 +1,139 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#24835] `update()` leaves the caller's `data` argument exactly as it was + * handed in — on every path that stamps `updated_at`. + * + * ## What was wrong + * + * `formatInput` and `applyWriteColumnMap` are copy-on-write: a patch that needs + * no coercion comes back as the CALLER'S OWN OBJECT. `update()` and + * `rotatedUpdateById()` then assigned the `updated_at` stamp onto it, so the + * stamp landed in the caller's patch. ObjectQL hands `driver.update` the same + * object it later publishes as `DataEvent.changes`, so every reader of the + * update event saw a value the caller never wrote: + * + * - on PostgreSQL / MySQL the stamp is `knex.fn.now()`, a knex `Raw` that + * holds the client, its pool and a Node `Timeout`. `JSON.stringify` of the + * event payload throws ("Converting circular structure to JSON"), so the + * webhook outbox could not store the delivery and EVERY webhook delivery for + * a record update was lost there; + * - on SQLite it is an ISO-8601 string — serialisable, which is why the loss + * never showed there, and just as foreign to the caller's patch. + * + * `updateMany()` already stamped into a fresh object; these two doors now do + * the same. + * + * ## What is pinned + * + * The caller's patch is deep-equal to a clone taken before the call, on each + * path that stamps: the plain path (a table this driver's DDL created), the + * presumed-timestamp path (`skipSchemaSync`-style registration, both when the + * column exists and when the fallback has to drop the stamp), and the rotated + * path. Each case also proves the stamp really ran on that path, so a pin + * cannot pass because nothing was stamped at all. SQLite only: the mutation is + * dialect-independent (only the stamped VALUE differs), and the live-PG/MySQL + * driver lane does not run on pull requests. + */ + +import { describe, it, expect, afterEach } from 'vitest'; +import { SqlDriver } from './index.js'; + +/** These fixtures are not tenant-scoped. */ +const OPTS = { bypassTenantAudit: true } as const; +const BACKDATED = '2020-01-01T00:00:00.000Z'; + +function sqlite(): SqlDriver { + return new SqlDriver({ client: 'better-sqlite3', connection: { filename: ':memory:' }, useNullAsDefault: true }); +} + +/** Read `updated_at` straight out of storage, past every read-side coercion. */ +async function storedUpdatedAt(driver: SqlDriver, table: string, id: string): Promise { + const row: any = await (driver as any).knex(table).where('id', id).first(); + return row?.updated_at; +} + +/** Update with `patch`, and return what the caller's object looked like before and after. */ +async function updateAndCompare(driver: SqlDriver, object: string, id: string, patch: Record) { + const before = structuredClone(patch); + await driver.update(object, id, patch, OPTS); + return { before, after: patch }; +} + +describe('#24835 — SqlDriver.update() never writes into the caller\'s data', () => { + let driver: SqlDriver | undefined; + afterEach(async () => { + await driver?.disconnect(); + driver = undefined; + }); + + it('plain path: a table the driver created stamps updated_at, and the patch is untouched', async () => { + driver = sqlite(); + await driver.initObjects([{ name: 'n24835_plain', fields: { title: { type: 'text' } } } as any]); + await driver.create('n24835_plain', { id: 'p1', title: 'seed' }, OPTS); + await (driver as any).knex('n24835_plain').where('id', 'p1').update({ updated_at: BACKDATED }); + + const { before, after } = await updateAndCompare(driver, 'n24835_plain', 'p1', { title: 'renamed' }); + + expect(after).toStrictEqual(before); + expect(Object.keys(after)).toEqual(['title']); + // The stamp ran on this path — it landed in storage, not in the patch. + expect(await storedUpdatedAt(driver, 'n24835_plain', 'p1')).not.toBe(BACKDATED); + }); + + it('presumed path, column present: the speculative stamped attempt leaves the patch untouched', async () => { + driver = sqlite(); + const knex = (driver as any).knex; + await knex.raw( + `create table "n24835_presumed" (id text primary key, title text, ` + + `created_at text default (strftime('%Y-%m-%dT%H:%M:%fZ','now')), ` + + `updated_at text default (strftime('%Y-%m-%dT%H:%M:%fZ','now')))`, + ); + driver.registerObjectMetadata([{ name: 'n24835_presumed', fields: { title: { type: 'text' } } } as any]); + await driver.create('n24835_presumed', { id: 'q1', title: 'seed' }, OPTS); + await knex('n24835_presumed').where('id', 'q1').update({ updated_at: BACKDATED }); + expect((driver as any).updatedAtStampIsPresumed('n24835_presumed')).toBe(true); + + const { before, after } = await updateAndCompare(driver, 'n24835_presumed', 'q1', { title: 'renamed' }); + + expect(after).toStrictEqual(before); + expect(await storedUpdatedAt(driver, 'n24835_presumed', 'q1')).not.toBe(BACKDATED); + // The presumption was settled by this very call. + expect((driver as any).updatedAtStampIsPresumed('n24835_presumed')).toBe(false); + }); + + it('presumed path, column absent: the fallback that drops the stamp leaves the patch untouched', async () => { + driver = sqlite(); + const knex = (driver as any).knex; + await knex.raw(`create table "n24835_nocol" (id text primary key, title text)`); + driver.registerObjectMetadata([{ name: 'n24835_nocol', fields: { title: { type: 'text' } } } as any]); + await knex('n24835_nocol').insert({ id: 'r1', title: 'seed' }); + expect((driver as any).updatedAtStampIsPresumed('n24835_nocol')).toBe(true); + + const { before, after } = await updateAndCompare(driver, 'n24835_nocol', 'r1', { title: 'renamed' }); + + expect(after).toStrictEqual(before); + expect((await knex('n24835_nocol').where('id', 'r1').first()).title).toBe('renamed'); + // The fallback path really ran: the driver now knows the column is absent. + expect((driver as any).updatedAtColumnState.get('n24835_nocol')).toBe('absent'); + }); + + it('rotated path: a rotation-managed object stamps updated_at in its shard, and the patch is untouched', async () => { + driver = sqlite(); + const rotated = { + name: 'n24835_rot', + fields: { payload: { type: 'text' }, created_at: { type: 'datetime' } }, + lifecycle: { class: 'telemetry', storage: { strategy: 'rotation', shards: 3, unit: 'day' } }, + }; + await driver.initObjects([rotated as any]); + const shards: string[] | undefined = (driver as any).rotationShardsOf('n24835_rot'); + expect(shards?.length).toBeGreaterThan(0); + await driver.create('n24835_rot', { id: 's1', payload: 'seed' }, OPTS); + await (driver as any).knex(shards![0]).where('id', 's1').update({ updated_at: BACKDATED }); + + const { before, after } = await updateAndCompare(driver, 'n24835_rot', 's1', { payload: 'patched' }); + + expect(after).toStrictEqual(before); + expect(await storedUpdatedAt(driver, shards![0], 's1')).not.toBe(BACKDATED); + }); +}); diff --git a/packages/drivers/driver-sql/src/sql-driver.ts b/packages/drivers/driver-sql/src/sql-driver.ts index 206bca5f0bc..ecc2686b44a 100644 --- a/packages/drivers/driver-sql/src/sql-driver.ts +++ b/packages/drivers/driver-sql/src/sql-driver.ts @@ -8817,11 +8817,19 @@ export class SqlDriver implements IDataDriver { }; if (this.stampsUpdatedAt(object) && !this.keepSuppliedUpdatedAt(formatted, options)) { - formatted.updated_at = this.updatedAtStamp(); + // [#24835] A FRESH object, never `formatted`: `formatInput` and + // `applyWriteColumnMap` are copy-on-write, so for a patch that needs no + // coercion `formatted` IS the caller's `data`. Assigning the stamp onto it + // put `updated_at` into the caller's patch — which ObjectQL publishes as + // `DataEvent.changes` — and on Postgres/MySQL that value is a knex `Raw` + // holding the client, its pool and a timer, so every reader that + // serialises the event (the webhook outbox first) threw. `updateMany`'s + // shape, below. + const stamped = { ...formatted, updated_at: this.updatedAtStamp() }; if (this.updatedAtStampIsPresumed(object)) { - await this.updateWithPresumedTimestamp(object, formatted, options, issue); + await this.updateWithPresumedTimestamp(object, stamped, options, issue); } else { - await issue(formatted, options); + await issue(stamped, options); } } else { await issue(formatted, options); @@ -10245,13 +10253,16 @@ export class SqlDriver implements IDataDriver { // fact on the line immediately before that call — so on this path the // answer can never come from the declared-shape presumption, and there is // nothing speculative to recover from. - if (this.stampsUpdatedAt(object) && !this.keepSuppliedUpdatedAt(formatted, options)) { - formatted.updated_at = this.updatedAtStamp(); - } + // + // [#24835] Stamped into a FRESH object, for the reason {@link update} gives: + // `formatted` may be the caller's own `data`. + const payload = this.stampsUpdatedAt(object) && !this.keepSuppliedUpdatedAt(formatted, options) + ? { ...formatted, updated_at: this.updatedAtStamp() } + : formatted; for (const shard of shards) { const builder = this.getBuilder(shard, options).where('id', id); this.applyTenantScope(builder, object, options); - const count = await builder.update(formatted); + const count = await builder.update(payload); if (count > 0) { const readback = this.getBuilder(shard, options).where('id', id); this.applyTenantScope(readback, object, options); diff --git a/packages/plugins/plugin-webhooks/package.json b/packages/plugins/plugin-webhooks/package.json index 5ca409d90ad..b78d669e2b0 100644 --- a/packages/plugins/plugin-webhooks/package.json +++ b/packages/plugins/plugin-webhooks/package.json @@ -42,6 +42,7 @@ "@objectstack/types": "workspace:*" }, "devDependencies": { + "@objectstack/driver-sql": "workspace:*", "@objectstack/metadata-core": "workspace:*", "@objectstack/objectql": "workspace:*", "@types/node": "^26.6.3", diff --git a/packages/plugins/plugin-webhooks/src/webhook-delivery-real-sql-driver.test.ts b/packages/plugins/plugin-webhooks/src/webhook-delivery-real-sql-driver.test.ts new file mode 100644 index 00000000000..8f1b6b94fd7 --- /dev/null +++ b/packages/plugins/plugin-webhooks/src/webhook-delivery-real-sql-driver.test.ts @@ -0,0 +1,142 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#24835] An insert and a by-id update, on a REAL `SqlDriver` under a REAL + * `ObjectQL` engine, with an active declared webhook, each store one + * `sys_http_delivery` row whose `payload_json` is plain JSON. + * + * ## Why the whole chain on a real driver + * + * The loss this pins was invisible to every fake in this package. The driver's + * `update()` used to stamp `updated_at` into the CALLER'S patch object, and + * ObjectQL publishes that same object as `DataEvent.changes`. On PostgreSQL and + * MySQL the stamp is a knex `Raw` holding the client, its pool and a Node + * `Timeout`, so the outbox's `JSON.stringify(payload)` threw and no delivery + * row was written for ANY record update — the record write still answered 200 + * and the webhook still read active. A fake driver never stamps, and a fake + * outbox never serialises, so only this composition can see it: + * `ObjectQL` → `SqlDriver` → realtime → `AutoEnqueuer` → + * `MessagingService.enqueueHttp` → `SqlHttpOutbox` → `sys_http_delivery`. + * + * SQLite here. On SQLite the stamp was an ISO string, so the row WAS written, + * and the assertion that would have caught the PG/MySQL loss is the plain-JSON + * round trip below. The same body was run against a throwaway local + * PostgreSQL 16 cluster when this landed (before: 1 row, the update lost; + * after: 2 rows); no live-PG lane runs on pull requests, so that run is not + * committed. + */ + +import { describe, expect, it } from 'vitest'; +import { ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import type { IRealtimeService, RealtimeEventHandler, RealtimeEventPayload } from '@objectstack/spec/contracts'; +import { HttpDelivery, MessagingService, SqlHttpOutbox, type EnqueueHttpInput } from '@objectstack/service-messaging'; +import { AutoEnqueuer } from './auto-enqueuer.js'; +import { DeclaredWebhookSource, declaredWebhookRefId } from './declared-webhooks.js'; +import { SysWebhook } from './sys-webhook.object.js'; + +const SYSTEM = { context: { isSystem: true } }; + +function driverUnderTest(): SqlDriver { + return new SqlDriver({ client: 'better-sqlite3', connection: { filename: ':memory:' }, useNullAsDefault: true }); +} + +class FakeRealtime implements IRealtimeService { + private subs = new Map(); + private n = 0; + async publish(event: RealtimeEventPayload): Promise { + for (const sub of this.subs.values()) { + if (sub.opts?.object && event.object !== sub.opts.object) continue; + await sub.handler(event); + } + } + async subscribe(_channel: string, handler: any, opts?: any): Promise { + const id = `s-${++this.n}`; + this.subs.set(id, { handler, opts }); + return id; + } + async unsubscribe(id: string): Promise { this.subs.delete(id); } +} + +/** A declared webhook, as `defineStack({ webhooks })` would author it — active by default. */ +const DECLARED = { + name: 'probe_task_changed', + object: 'probe_task', + triggers: ['create', 'update'], + url: 'https://receiver.invalid/hook', + method: 'POST', +}; + +/** True when `value` is made only of what JSON can carry. */ +function isPlainJson(value: unknown): boolean { + if (value === null || ['string', 'number', 'boolean'].includes(typeof value)) return true; + if (Array.isArray(value)) return value.every(isPlainJson); + if (typeof value !== 'object') return false; + if (Object.getPrototypeOf(value) !== Object.prototype) return false; + return Object.values(value as Record).every(isPlainJson); +} + +async function waitFor(check: () => Promise, ms = 3000): Promise { + const end = Date.now() + ms; + while (Date.now() < end) { + if (await check()) return; + await new Promise((r) => setTimeout(r, 25)); + } +} + +describe('#24835 — a webhook delivery is stored for an update on a real SqlDriver', () => { + it('an insert and a by-id update each store one delivery row whose payload_json is plain JSON', async () => { + const driver = driverUnderTest(); + const engine = new ObjectQL(); + engine.registerDriver(driver, true); + await engine.init(); + engine.registry.registerObject({ name: 'probe_task', label: 'Probe Task', fields: { title: { type: 'text', label: 'Title' } } } as any, 'test'); + engine.registry.registerObject(HttpDelivery as any, 'test'); + engine.registry.registerObject(SysWebhook as any, 'test'); + await engine.syncSchemas(); + + const realtime = new FakeRealtime(); + engine.setRealtimeService(realtime); + const lost: string[] = []; + const logger = { + info() {}, debug() {}, + warn: (m: string) => { lost.push(m); }, + error: (m: string) => { lost.push(m); }, + }; + const messaging = new MessagingService({ logger: { info() {}, warn() {}, error() {}, debug() {} } as any }); + messaging.setHttpOutbox(new SqlHttpOutbox(engine as any, { partitionCount: 1 })); + const enqueuer = new AutoEnqueuer(engine as any, realtime, (input: EnqueueHttpInput) => messaging.enqueueHttp(input), { + refreshIntervalMs: 0, + declaredWebhooks: new DeclaredWebhookSource(engine, { list: (type: string) => (type === 'webhook' ? [DECLARED] : []) }), + logger, + }); + const deliveries = async () => (await engine.find('sys_http_delivery', {}, SYSTEM)) as any[]; + + try { + await enqueuer.start(); + const created: any = await engine.insert('probe_task', { title: 'seed' }); + await waitFor(async () => (await deliveries()).length >= 1); + const patch = { title: 'webhook probe' }; + await engine.update('probe_task', patch, { where: { id: created.id } }); + await waitFor(async () => (await deliveries()).length >= 2); + + const rows = await deliveries(); + expect(rows.map((r) => r.label).sort()).toEqual(['data.record.created', 'data.record.updated']); + for (const row of rows) { + expect(row.ref_id).toBe(declaredWebhookRefId('probe_task_changed')); + expect(typeof row.payload_json).toBe('string'); + const parsed = JSON.parse(row.payload_json); + expect(isPlainJson(parsed)).toBe(true); + expect(JSON.stringify(parsed)).toBe(row.payload_json); + expect(parsed).toMatchObject({ object: 'probe_task', recordId: created.id }); + } + const updated = JSON.parse(rows.find((r) => r.label === 'data.record.updated').payload_json); + expect(updated.action).toBe('updated'); + expect(updated.changes.title).toBe('webhook probe'); + expect(lost).toEqual([]); + } finally { + await enqueuer.stop(); + await driver.disconnect(); + } + }); +}); diff --git a/packages/plugins/plugin-webhooks/vitest.config.ts b/packages/plugins/plugin-webhooks/vitest.config.ts index aa5ded710d7..4b3e2368374 100644 --- a/packages/plugins/plugin-webhooks/vitest.config.ts +++ b/packages/plugins/plugin-webhooks/vitest.config.ts @@ -54,6 +54,12 @@ export default defineConfig({ // session-read rule). Same reason and the same anchoring as the // `core` entry above. { find: /^@objectstack\/types$/, replacement: path.resolve(__dirname, '../../types/src/index.ts') }, + // [#24835] `webhook-delivery-real-sql-driver.test.ts` takes a VALUE + // import on `@objectstack/driver-sql` (a real `SqlDriver` under the + // engine), so the delivery verdict is about the driver source in the + // checkout, not `driver-sql/dist`. Same reason and anchoring as the + // `core` entry above. + { find: /^@objectstack\/driver-sql$/, replacement: path.resolve(__dirname, '../../drivers/driver-sql/src/index.ts') }, ], }, }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index be40adef2f0..f6136050a3b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2044,6 +2044,9 @@ importers: specifier: workspace:* version: link:../../types devDependencies: + '@objectstack/driver-sql': + specifier: workspace:* + version: link:../../drivers/driver-sql '@objectstack/metadata-core': specifier: workspace:* version: link:../../metadata-core