fix(postgres): run the installed command, dedupe by instant, refuse out-of-range offsets

packages/sync-postgres/src/cli.ts

@@ -32,6 +32,8 @@ import {
   type SyncEngineConfig,
 } from '@workos/sync-engine'
 import { X509Certificate } from 'node:crypto'
+import { realpathSync } from 'node:fs'
+import { fileURLToPath } from 'node:url'
 import { PostgresDestination } from './destination.js'
 import { buildSchemaSql } from './schema/ddl.js'
 import { fingerprint } from './schema/migrate.js'
@@ -537,9 +539,23 @@ async function runCommand(command: string, flags: Record<string, string | boolea
   }
 }
 
-// Run when invoked directly, not when imported by a test.
-const invokedDirectly = process.argv[1] && (process.argv[1].endsWith('cli.js') || process.argv[1].endsWith('cli.ts'))
-if (invokedDirectly) {
+/**
+ * Whether this module is the script Node was started with, as opposed to an
+ * import by a test. npm installs a `bin` as a symlink named after the
+ * command, so `process.argv[1]` ends in `workos-sync-postgres` rather than
+ * `cli.js`; both sides are resolved to real paths so the installed command
+ * counts.
+ */
+export function invokedDirectly(argv1: string | undefined, moduleUrl: string): boolean {
+  if (!argv1) return false
+  try {
+    return realpathSync(argv1) === realpathSync(fileURLToPath(moduleUrl))
+  } catch {
+    return false
+  }
+}
+
+if (invokedDirectly(process.argv[1], import.meta.url)) {
   main().then(
     (code) => process.exit(code),
     (err) => {

packages/sync-postgres/src/db/upsert.ts

@@ -63,7 +63,7 @@ export function keyValues(stream: TableDef, data: Record<string, unknown>): stri
  * relink and unlink paths, whose rows are whole; upserts use `wavesByKey`.
  */
 export function dedupeByKey(stream: TableDef, rows: Row[]): { rows: Row[]; dropped: number } {
-  const byKey = new Map<string, Row>()
+  const byKey = new Map<string, { row: Row; at: number }>()
   let dropped = 0
   for (const row of rows) {
     const key = keyValues(stream, row.data)
@@ -72,10 +72,35 @@ export function dedupeByKey(stream: TableDef, rows: Row[]): { rows: Row[]; dropp
       continue
     }
     const k = key.join('\0')
+    const at = micros(row.updatedAt)
     const existing = byKey.get(k)
-    if (!existing || row.updatedAt >= existing.updatedAt) byKey.set(k, row)
+    if (!existing || !isBefore(row.updatedAt, at, existing.row.updatedAt, existing.at)) byKey.set(k, { row, at })
   }
-  return { rows: [...byKey.values()], dropped }
+  return { rows: [...byKey.values()].map((e) => e.row), dropped }
+}
+
+/**
+ * The instant `iso` names, in microseconds since the epoch, or NaN.
+ * `Date.parse` reads three fractional digits; a `timestamptz` keeps six, so
+ * the rest of the fraction is read here and two rows a microsecond apart
+ * stay ordered the way the staleness gate will order them.
+ */
+function micros(iso: string): number {
+  const frac = /\.(\d+)/.exec(iso)?.[1] ?? ''
+  const ms = Date.parse(frac.length > 3 ? iso.replace(`.${frac}`, `.${frac.slice(0, 3)}`) : iso)
+  if (Number.isNaN(ms)) return ms
+  return ms * 1000 + Number(frac.slice(3, 6).padEnd(3, '0'))
+}
+
+/**
+ * Whether one timestamp is strictly earlier than another, by the instant
+ * each names: `01:00:00+01:00` sorts after `00:30:00Z` as text and is half an
+ * hour earlier as a time. A value that is not a date is ordered as text; the
+ * database refuses it on the write either way.
+ */
+function isBefore(a: string, atA: number, b: string, atB: number): boolean {
+  if (Number.isNaN(atA) || Number.isNaN(atB)) return a < b
+  return atA < atB
 }
 
 /**

packages/sync-postgres/src/schema/ddl.ts

@@ -105,6 +105,7 @@ as $fn$
     when p[3]::int not between 1 and
       extract(day from (make_date(p[1]::int, p[2]::int, 1) + interval '1 month' - interval '1 day')) then null
     when p[4]::int > 23 or p[5]::int > 59 or p[6]::double precision > 60 then null
+    when p[9]::int > 23 or p[10]::int > 59 then null
     else
     timezone('UTC', make_timestamp(
       p[1]::int, p[2]::int, p[3]::int, p[4]::int, p[5]::int, p[6]::double precision))

packages/sync-postgres/test/cli.test.ts

@@ -1,6 +1,11 @@
 import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'
+import { mkdtempSync, rmSync, symlinkSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { pathToFileURL } from 'node:url'
 import { SyncEngine, WorkOSClient, selectStreams, silentLogger } from '@workos/sync-engine'
 import {
+  invokedDirectly,
   main,
   parseHydrate,
   parseLeaseTtl,
@@ -19,6 +24,42 @@ import { TEST_CA_PEM } from './helpers/ca.js'
 import { FakeWorkOS, resetIds } from '../../sync-engine/test/helpers/fake-workos.js'
 import type { SqlClient } from '../src/db/client.js'
 
+describe('entrypoint', () => {
+  let dir: string
+  beforeEach(() => {
+    dir = mkdtempSync(join(tmpdir(), 'workos-sync-postgres-bin-'))
+  })
+  afterEach(() => {
+    rmSync(dir, { recursive: true, force: true })
+  })
+
+  it('recognises the script itself and the symlink npm installs as the command', () => {
+    const script = join(dir, 'cli.js')
+    writeFileSync(script, '')
+    const url = pathToFileURL(script).href
+    expect(invokedDirectly(script, url)).toBe(true)
+
+    // `npm install` links `node_modules/.bin/workos-sync-postgres` at dist/cli.js,
+    // so argv[1] is the link, not the file.
+    const link = join(dir, 'workos-sync-postgres')
+    symlinkSync(script, link)
+    expect(invokedDirectly(link, url)).toBe(true)
+  })
+
+  it('is false for an import, a missing path, or another script', () => {
+    const script = join(dir, 'cli.js')
+    writeFileSync(script, '')
+    const url = pathToFileURL(script).href
+    expect(invokedDirectly(undefined, url)).toBe(false)
+    expect(invokedDirectly(join(dir, 'missing'), url)).toBe(false)
+    const other = join(dir, 'vitest.mjs')
+    writeFileSync(other, '')
+    expect(invokedDirectly(other, url)).toBe(false)
+    // This test file was started by the test runner, not by the CLI.
+    expect(invokedDirectly(process.argv[1], url)).toBe(false)
+  })
+})
+
 describe('WORKOS_SYNC_HYDRATE', () => {
   it('leaves the engine default in place when unset', () => {
     expect(parseHydrate(undefined)).toBeUndefined()

packages/sync-postgres/test/schema.test.ts

@@ -179,6 +179,10 @@ describe('migrations against real Postgres', () => {
       ['2026-01-01T24:00:00Z', null],
       ['2026-01-01T23:60:00Z', null],
       ['0000-01-01T00:00:00Z', null],
+      // The offset's fields are checked like the clock's.
+      ['2026-01-01T00:00:00+00:99', null],
+      ['2026-01-01T00:00:00-24:00', null],
+      ['2026-01-01T00:00:00+23:59', '2025-12-31T00:01:00.000Z'],
     ]
     for (const [input, expected] of cases) {
       await db.query(`insert into workos.users ("_env","_raw","_updated_at") values ('prod', $1::jsonb, now())`, [

packages/sync-postgres/test/upsert.test.ts

@@ -43,6 +43,27 @@ describe('key extraction', () => {
     expect(dropped).toBe(1)
   })
 
+  it('judges the newest row by its instant, not by the text of its timestamp', () => {
+    // "01:00:00+01:00" sorts after "00:30:00Z" as text and is half an hour earlier as a time.
+    const { rows } = dedupeByKey(users, [
+      { data: userPayload({ email: 'older@example.com' }), updatedAt: '2026-01-01T01:00:00+01:00' },
+      { data: userPayload({ email: 'newer@example.com' }), updatedAt: '2026-01-01T00:30:00Z' },
+    ])
+    expect(rows.map((r) => r.data['email'])).toEqual(['newer@example.com'])
+    // The database keeps six fractional digits, so two rows a microsecond apart stay ordered.
+    const fine = dedupeByKey(users, [
+      { data: userPayload({ email: 'newer@example.com' }), updatedAt: '2026-01-01T00:00:00.000900Z' },
+      { data: userPayload({ email: 'older@example.com' }), updatedAt: '2026-01-01T00:00:00.000100Z' },
+    ])
+    expect(fine.rows.map((r) => r.data['email'])).toEqual(['newer@example.com'])
+    // What is not a date is ordered as text, and the database refuses it on the write.
+    const junk = dedupeByKey(users, [
+      { data: userPayload({ email: 'a@example.com' }), updatedAt: 'later' },
+      { data: userPayload({ email: 'b@example.com' }), updatedAt: 'yesterday' },
+    ])
+    expect(junk.rows.map((r) => r.data['email'])).toEqual(['b@example.com'])
+  })
+
   it('splits repeated keys into ordered waves and leaves distinct keys in one', () => {
     // Postgres refuses to let one INSERT ... ON CONFLICT touch a row twice.
     // The waves are written in turn, so the rows land in the order given.