fix(messaging): store outbox audit timestamps as datetime for Postgres (#2228)

xuyushun441-sys · os-zhuang · claude · web-flow · commit 6d3bf5466d03 · 2026-06-23T10:32:37.000+08:00
created_at/updated_at are builtin audit columns the SQL driver always creates
as native TIMESTAMP columns, regardless of the declared field type. The
notification + HTTP outboxes declared them as Field.number and wrote epoch-ms
via Date.now(), so on Postgres both enqueue inserts and the retention sweep
threw "date/time field value out of range" (bigint vs timestamp). SQLite's
lenient affinity hid it until the multi-node Postgres E2E.

- objects: created_at/updated_at -&gt; Field.datetime (matches the real column and
  registers them for the driver's per-dialect filter coercion)
- outboxes: enqueue writes Date; toRecord normalises read-back via toEpochMs so
  the record contract stays epoch ms
- retention: one ISO-8601 cutoff for every target (drop format:'epoch')
- tests: retention asserts an ISO-string cutoff; new driver-sql prune regression
  against a builtin created_at timestamp column

service-job job-run-retention was already ISO (no change).

Co-authored-by: Jack Zhuang &lt;277994282+os-zhuang@users.noreply.github.com&gt;
Co-authored-by: Claude Opus 4.8 &lt;noreply@anthropic.com&gt;
diff --git a/.changeset/messaging-pg-audit-timestamp.md b/.changeset/messaging-pg-audit-timestamp.md
@@ -0,0 +1,18 @@
+---
+"@objectstack/service-messaging": patch
+---
+
+fix(messaging): store outbox audit timestamps as datetime so Postgres retention works
+
+`created_at`/`updated_at` are builtin audit columns that the SQL driver always
+provisions as native `TIMESTAMP` columns, regardless of the declared field type.
+The notification and HTTP outboxes declared them as `Field.number` and wrote
+epoch-ms via `Date.now()`, so on Postgres both the `enqueue` insert and the
+retention sweep failed with `date/time field value out of range` (a bigint
+compared to a timestamp column). SQLite's lenient column affinity hid the bug
+until the multi-node Postgres E2E.
+
+The outbox objects now declare these as `Field.datetime` and write `Date`s; the
+retention sweep uses one ISO-8601 cutoff for every target (dropping the
+`format: 'epoch'` special case); `toRecord` normalises read-back to epoch ms so
+the record contract is unchanged. `sys_job_run` retention was already ISO.
diff --git a/packages/plugins/driver-sql/src/sql-driver-retention-prune.test.ts b/packages/plugins/driver-sql/src/sql-driver-retention-prune.test.ts
@@ -0,0 +1,81 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * Regression for the messaging / job retention sweeps on Postgres
+ * (service-messaging `NotificationRetention`, service-job `JobRunRetention`):
+ *
+ * The builtin `created_at` audit column is provisioned as a native `TIMESTAMP`
+ * (`table.timestamp`), so a retention prune MUST filter it with an ISO-8601
+ * cutoff. An earlier sweep passed a bare epoch-ms number, which compares a
+ * bigint to a timestamp column — Postgres rejects it ("date/time field value
+ * out of range"). SQLite's lenient column affinity hid the bug.
+ *
+ * This proves the path the sweep now relies on: declaring the builtin
+ * `created_at` column as `datetime` registers it for per-dialect filter
+ * coercion, so a `created_at < <ISO>` delete prunes exactly the aged rows. Rows
+ * are written as JS `Date`s, exactly as the outboxes now do.
+ */
+
+import { describe, it, expect, beforeEach, afterEach } from 'vitest';
+import { SqlDriver } from '../src/index.js';
+
+describe('SqlDriver retention prune on a builtin created_at timestamp column', () => {
+  let driver: SqlDriver;
+
+  beforeEach(async () => {
+    driver = new SqlDriver({
+      client: 'better-sqlite3',
+      connection: { filename: ':memory:' },
+      useNullAsDefault: true,
+    });
+
+    // `created_at` is a builtin audit column; declaring it `datetime` (as the
+    // outbox objects now do) registers it for filter coercion without creating a
+    // duplicate column.
+    await driver.initObjects([
+      {
+        name: 'retention_probe',
+        fields: { created_at: { type: 'datetime' }, label: { type: 'string' } },
+      },
+    ]);
+
+    // Write `created_at` as `Date` objects — the outbox enqueue convention. On
+    // SQLite better-sqlite3 stores these as INTEGER milliseconds.
+    await driver.create('retention_probe',
+      { id: 'old1', label: 'old', created_at: new Date('2025-01-01T00:00:00Z') }, { bypassTenantAudit: true });
+    await driver.create('retention_probe',
+      { id: 'old2', label: 'old', created_at: new Date('2025-02-01T00:00:00Z') }, { bypassTenantAudit: true });
+    await driver.create('retention_probe',
+      { id: 'new1', label: 'new', created_at: new Date('2026-06-01T00:00:00Z') }, { bypassTenantAudit: true });
+  });
+
+  afterEach(async () => {
+    await driver.disconnect();
+  });
+
+  it('prunes rows older than an ISO-8601 cutoff and keeps recent ones', async () => {
+    const cutoffIso = new Date('2026-01-01T00:00:00.000Z').toISOString();
+
+    const deleted = await driver.deleteMany(
+      'retention_probe',
+      { where: { created_at: { $lt: cutoffIso } } },
+      { bypassTenantAudit: true },
+    );
+    expect(deleted).toBe(2);
+
+    const remaining = await driver.find('retention_probe', {});
+    expect(remaining.map((r: any) => r.id)).toEqual(['new1']);
+  });
+
+  it('an ISO-8601 cutoff before every row deletes nothing', async () => {
+    const deleted = await driver.deleteMany(
+      'retention_probe',
+      { where: { created_at: { $lt: '2020-01-01T00:00:00.000Z' } } },
+      { bypassTenantAudit: true },
+    );
+    expect(deleted).toBe(0);
+
+    const remaining = await driver.find('retention_probe', {});
+    expect(remaining.map((r: any) => r.id).sort()).toEqual(['new1', 'old1', 'old2']);
+  });
+});
diff --git a/packages/services/service-messaging/src/audit-timestamp.ts b/packages/services/service-messaging/src/audit-timestamp.ts
@@ -0,0 +1,24 @@
+// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * Normalise a builtin `created_at` / `updated_at` value to epoch milliseconds.
+ *
+ * Those two columns are provisioned as native `TIMESTAMP` columns by the SQL
+ * driver (Postgres/MySQL), so the outboxes WRITE them as `Date`s — never a bare
+ * epoch-ms number, which a real timestamp column rejects (the bug that broke the
+ * `sys_notification_delivery` retention sweep on Postgres). The read-back form is
+ * therefore dialect-dependent: epoch-ms on SQLite, a `Date` (or ISO string) on
+ * Postgres. This collapses all of those back to epoch-ms so the outbox record
+ * contract (`createdAt` / `updatedAt: number`) stays driver-independent.
+ */
+export function toEpochMs(value: unknown): number {
+    if (typeof value === 'number') return value;
+    if (value instanceof Date) return value.getTime();
+    if (typeof value === 'string') {
+        const parsed = Date.parse(value);
+        if (Number.isFinite(parsed)) return parsed;
+        const numeric = Number(value);
+        if (Number.isFinite(numeric)) return numeric;
+    }
+    return 0;
+}
diff --git a/packages/services/service-messaging/src/objects/http-delivery.object.ts b/packages/services/service-messaging/src/objects/http-delivery.object.ts
@@ -149,8 +149,11 @@ export const HttpDelivery = ObjectSchema.create({
         response_body: Field.textarea({ label: 'Response Body (capped)', required: false }),
         error: Field.textarea({ label: 'Error', required: false }),
 
-        created_at: Field.number({ label: 'Created At (ms)', required: true }),
-        updated_at: Field.number({ label: 'Updated At (ms)', required: true }),
+        // Builtin audit columns are native TIMESTAMP columns (Postgres/MySQL),
+        // so declare them `datetime` and write `Date`s (not epoch-ms numbers,
+        // which a real timestamp column rejects). See SqlHttpOutbox.
+        created_at: Field.datetime({ label: 'Created At', required: true }),
+        updated_at: Field.datetime({ label: 'Updated At', required: true }),
     },
 
     indexes: [
diff --git a/packages/services/service-messaging/src/objects/notification-delivery.object.ts b/packages/services/service-messaging/src/objects/notification-delivery.object.ts
@@ -13,7 +13,9 @@ import { ObjectSchema, Field } from '@objectstack/spec/data';
  *
  * `payload` snapshots the rendered notification content at enqueue time so a
  * later edit of the L2 event can't rewrite an in-flight delivery (and so the
- * dispatcher needs no second read to send). Timestamps are epoch ms.
+ * dispatcher needs no second read to send). Scheduling fields (`claimed_at`,
+ * `next_attempt_at`, `last_attempted_at`) are epoch ms; the builtin
+ * `created_at` / `updated_at` audit columns are native timestamps.
  */
 export const NotificationDelivery = ObjectSchema.create({
     name: 'sys_notification_delivery',
@@ -67,8 +69,12 @@ export const NotificationDelivery = ObjectSchema.create({
         last_attempted_at: Field.number({ label: 'Last Attempted At (ms)' }),
         error: Field.textarea({ label: 'Error' }),
 
-        created_at: Field.number({ label: 'Created At (ms)', readonly: true }),
-        updated_at: Field.number({ label: 'Updated At (ms)' }),
+        // Builtin audit columns: the SQL driver provisions `created_at` /
+        // `updated_at` as native TIMESTAMP columns (Postgres/MySQL), so they are
+        // declared `datetime` and written as `Date`s — a bare epoch-ms number is
+        // rejected by a real timestamp column. See SqlNotificationOutbox.
+        created_at: Field.datetime({ label: 'Created At', readonly: true }),
+        updated_at: Field.datetime({ label: 'Updated At' }),
     },
 
     indexes: [
diff --git a/packages/services/service-messaging/src/retention.test.ts b/packages/services/service-messaging/src/retention.test.ts
@@ -30,7 +30,7 @@ function captureEngine(deleteImpl?: (object: string, opts: any) => any) {
 const FIXED_NOW = 1_700_000_000_000; // fixed clock for deterministic cutoffs
 
 describe('NotificationRetention', () => {
-    it('prunes every target older than the cutoff, formatting the cutoff per target', async () => {
+    it('prunes every target older than an ISO-8601 cutoff (never a bare epoch-ms number)', async () => {
         const { engine, deletes } = captureEngine();
         const retention = new NotificationRetention({
             getData: () => engine,
@@ -50,12 +50,17 @@ describe('NotificationRetention', () => {
             // Cross-tenant system context — retention is an operator policy.
             expect(d.context).toEqual({ isSystem: true });
         }
-        // The delivery row stores epoch-ms; everything else stores ISO strings.
+        // Every target's `created_at` is a native timestamp column, so the cutoff
+        // is an ISO-8601 string — NEVER a bare epoch-ms number (a bigint compared
+        // to a timestamp column makes Postgres throw "date/time field value out of
+        // range"; SQLite's lenient affinity used to hide it for the delivery outbox).
         const byObject = Object.fromEntries(deletes.map((d) => [d.object, d.where]));
-        expect(byObject['sys_notification_delivery']).toEqual({ created_at: { $lt: cutoffMs } });
-        expect(byObject['sys_notification']).toEqual({ created_at: { $lt: cutoffIso } });
-        expect(byObject['sys_inbox_message']).toEqual({ created_at: { $lt: cutoffIso } });
-        expect(byObject['sys_notification_receipt']).toEqual({ created_at: { $lt: cutoffIso } });
+        for (const obj of DEFAULT_RETENTION_TARGETS.map((t) => t.object)) {
+            expect(byObject[obj]).toEqual({ created_at: { $lt: cutoffIso } });
+            const lt = (byObject[obj].created_at as { $lt: unknown }).$lt;
+            expect(typeof lt).toBe('string');
+            expect(lt).toMatch(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/);
+        }
 
         expect(outcomes.every((o) => o.deleted === 1 && !o.error)).toBe(true);
     });
@@ -100,7 +105,7 @@ describe('NotificationRetention', () => {
             getData: () => engine,
             logger: silentLogger(),
             now: () => FIXED_NOW,
-            targets: [{ object: 'sys_notification', tsField: 'created_at', format: 'iso' }],
+            targets: [{ object: 'sys_notification', tsField: 'created_at' }],
         });
         const outcomes = await retention.prune(1);
         expect(outcomes).toEqual([{ object: 'sys_notification', deleted: undefined }]);
diff --git a/packages/services/service-messaging/src/retention.ts b/packages/services/service-messaging/src/retention.ts
@@ -21,16 +21,17 @@ interface RetentionLogger {
 }
 
 /**
- * One object the sweeper prunes, plus how to read its age. The pipeline stores
- * timestamps two ways: the event / inbox / receipt rows carry ISO-8601 strings
- * (`created_at`), while the delivery outbox rows carry epoch-ms numbers — so the
- * cutoff value is formatted per target. `$lt` works for both (ISO-8601 sorts
- * lexicographically; epoch-ms numerically).
+ * One object the sweeper prunes, plus the field that carries its age. Every
+ * target's `created_at` is a builtin audit column — a native `TIMESTAMP` on
+ * Postgres/MySQL — so the cutoff is always an ISO-8601 string. (An earlier
+ * version passed an epoch-ms number for the delivery outbox; that compared a
+ * bigint to a timestamp column and Postgres rejected it with "date/time field
+ * value out of range". SQLite's lenient column affinity hid the bug.) The
+ * driver coerces the ISO comparand to the column's storage form per dialect.
  */
 export interface RetentionTarget {
     readonly object: string;
     readonly tsField: string;
-    readonly format: 'iso' | 'epoch';
 }
 
 /**
@@ -42,10 +43,10 @@ export interface RetentionTarget {
  * unaffected.
  */
 export const DEFAULT_RETENTION_TARGETS: readonly RetentionTarget[] = [
-    { object: RECEIPT_OBJECT, tsField: 'created_at', format: 'iso' },
-    { object: INBOX_OBJECT, tsField: 'created_at', format: 'iso' },
-    { object: DELIVERY_OBJECT, tsField: 'created_at', format: 'epoch' },
-    { object: NOTIFICATION_EVENT_OBJECT, tsField: 'created_at', format: 'iso' },
+    { object: RECEIPT_OBJECT, tsField: 'created_at' },
+    { object: INBOX_OBJECT, tsField: 'created_at' },
+    { object: DELIVERY_OBJECT, tsField: 'created_at' },
+    { object: NOTIFICATION_EVENT_OBJECT, tsField: 'created_at' },
 ];
 
 export interface NotificationRetentionOptions {
@@ -103,15 +104,16 @@ export class NotificationRetention {
             return [];
         }
 
-        const cutoffMs = this.now() - retentionDays * 86_400_000;
-        const cutoffIso = new Date(cutoffMs).toISOString();
+        const cutoffIso = new Date(this.now() - retentionDays * 86_400_000).toISOString();
         const outcomes: PruneOutcome[] = [];
 
         for (const t of this.targets) {
-            const cutoff = t.format === 'epoch' ? cutoffMs : cutoffIso;
             try {
                 const res = await data.delete(t.object, {
-                    where: { [t.tsField]: { $lt: cutoff } },
+                    // ISO-8601 cutoff for every target: `created_at` is a native
+                    // timestamp column, which rejects a bare epoch-ms number on
+                    // Postgres. The driver coerces this per dialect on the way down.
+                    where: { [t.tsField]: { $lt: cutoffIso } },
                     multi: true,
                     // System context: retention is an operator policy that spans
                     // tenants, so it must not be scoped by the caller's RLS.
diff --git a/packages/services/service-messaging/src/sql-http-outbox.ts b/packages/services/service-messaging/src/sql-http-outbox.ts
@@ -3,6 +3,7 @@
 import { randomUUID } from 'node:crypto';
 import type { IDataEngine } from '@objectstack/spec/contracts';
 import { hashPartition } from './backoff.js';
+import { toEpochMs } from './audit-timestamp.js';
 import {
     HttpRedeliverError,
     type EnqueueHttpInput,
@@ -46,8 +47,11 @@ interface DeliveryRow {
     response_code?: number | null;
     response_body?: string | null;
     error?: string | null;
-    created_at: number;
-    updated_at: number;
+    // Builtin audit columns (native TIMESTAMP on Postgres/MySQL): WRITTEN as
+    // `Date`s. Read-back form is dialect-dependent; `toRecord` normalises via
+    // `toEpochMs`.
+    created_at: number | string | Date;
+    updated_at: number | string | Date;
 }
 
 /**
@@ -83,7 +87,9 @@ export class SqlHttpOutbox implements IHttpOutbox {
         if (existing?.id) return existing.id as string;
 
         const id = randomUUID();
-        const now = Date.now();
+        // `Date`, not epoch-ms: `created_at` / `updated_at` are native TIMESTAMP
+        // columns and a real timestamp column rejects a bare number on Postgres.
+        const now = new Date();
         const row: DeliveryRow = {
             id,
             source: input.source,
@@ -270,8 +276,8 @@ export class SqlHttpOutbox implements IHttpOutbox {
             responseCode: r.response_code ?? undefined,
             responseBody: r.response_body ?? undefined,
             error: r.error ?? undefined,
-            createdAt: r.created_at,
-            updatedAt: r.updated_at,
+            createdAt: toEpochMs(r.created_at),
+            updatedAt: toEpochMs(r.updated_at),
         };
     }
 }
diff --git a/packages/services/service-messaging/src/sql-outbox.ts b/packages/services/service-messaging/src/sql-outbox.ts
@@ -11,6 +11,7 @@ import type {
     NotificationDeliveryRecord,
 } from './outbox.js';
 import { hashPartition } from './backoff.js';
+import { toEpochMs } from './audit-timestamp.js';
 
 export const DELIVERY_OBJECT = 'sys_notification_delivery';
 
@@ -38,8 +39,11 @@ interface DeliveryRow {
     last_attempted_at?: number | null;
     error?: string | null;
     digest_key?: string | null;
-    created_at: number;
-    updated_at: number;
+    // Builtin audit columns (native TIMESTAMP on Postgres/MySQL): WRITTEN as
+    // `Date`s. Read-back form is dialect-dependent (epoch-ms on SQLite, a
+    // `Date`/ISO string on Postgres); `toRecord` normalises via `toEpochMs`.
+    created_at: number | string | Date;
+    updated_at: number | string | Date;
 }
 
 /**
@@ -70,7 +74,9 @@ export class SqlNotificationOutbox implements INotificationOutbox {
         if (existing?.id) return String(existing.id);
 
         const id = randomUUID();
-        const now = Date.now();
+        // `Date`, not epoch-ms: `created_at` / `updated_at` are native TIMESTAMP
+        // columns and a real timestamp column rejects a bare number on Postgres.
+        const now = new Date();
         const row: DeliveryRow = {
             id,
             notification_id: input.notificationId,
@@ -254,8 +260,8 @@ export class SqlNotificationOutbox implements INotificationOutbox {
             lastAttemptedAt: r.last_attempted_at ?? undefined,
             error: r.error ?? undefined,
             digestKey: r.digest_key ?? undefined,
-            createdAt: r.created_at,
-            updatedAt: r.updated_at,
+            createdAt: toEpochMs(r.created_at),
+            updatedAt: toEpochMs(r.updated_at),
         };
     }
 }
