diff --git a/packages/sync-postgres/src/cli.ts b/packages/sync-postgres/src/cli.ts index a9303ef..d04c404 100644 --- a/packages/sync-postgres/src/cli.ts +++ b/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 process.exit(code), (err) => { diff --git a/packages/sync-postgres/src/db/upsert.ts b/packages/sync-postgres/src/db/upsert.ts index 7b45d68..acc6be0 100644 --- a/packages/sync-postgres/src/db/upsert.ts +++ b/packages/sync-postgres/src/db/upsert.ts @@ -63,7 +63,7 @@ export function keyValues(stream: TableDef, data: Record): 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() + const byKey = new Map() 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 } /** diff --git a/packages/sync-postgres/src/schema/ddl.ts b/packages/sync-postgres/src/schema/ddl.ts index bba9e89..a813f0b 100644 --- a/packages/sync-postgres/src/schema/ddl.ts +++ b/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)) diff --git a/packages/sync-postgres/test/cli.test.ts b/packages/sync-postgres/test/cli.test.ts index 2db2034..1d35ecb 100644 --- a/packages/sync-postgres/test/cli.test.ts +++ b/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() diff --git a/packages/sync-postgres/test/schema.test.ts b/packages/sync-postgres/test/schema.test.ts index 678be5d..accd0da 100644 --- a/packages/sync-postgres/test/schema.test.ts +++ b/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())`, [ diff --git a/packages/sync-postgres/test/upsert.test.ts b/packages/sync-postgres/test/upsert.test.ts index 192a6bc..bd1b87a 100644 --- a/packages/sync-postgres/test/upsert.test.ts +++ b/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.