From bfd742e94ee590ac4dddb846e66aec56d2773499 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Elberte=20Pl=C3=ADnio?= Date: Sun, 19 Jul 2026 16:17:46 -0300 Subject: [PATCH] refactor(sync): make LWW one durable Postgres operation Replace the settings-sync last-writer-wins choreography (conditional `.update()` guarded by `updated_at <`, falling back to an `ignoreDuplicates` `.upsert()`, falling back to a third read to learn the outcome) with one durable `settings_sync_lww_write` SQL function that locks the (user_id, field_group) row, compares/writes/tombstones, and always returns the authoritative row in the same call. pushGroup/deleteGroup keep all validation/sanitization and now just call the durable RPC; `written`/`stale` result semantics are unchanged. The migration is additive only (#35): it adds the function, nothing else. Adds a local-Postgres contract lane (packages/sync/test/lww.contract.test.ts) following PR #47's pattern (skips cleanly without Postgres, wired into `test:supabase`), covering real concurrent push/push, first-insert/ first-insert, and push/delete races, equal-timestamp ties, and older data racing in after a newer tombstone. pgTAP coverage (supabase/tests/database/settings_sync_lww.test.sql) adds microsecond canonicalization, the existing future-timestamp check constraint, and per-user isolation. Issue #37 CAND-3. --- package.json | 2 +- packages/sync/src/index.ts | 129 +++--- packages/sync/test/lww-fixtures.ts | 99 +++++ packages/sync/test/lww.contract.test.ts | 370 ++++++++++++++++++ packages/sync/test/postgres-sync-client.ts | 202 ++++++++++ packages/sync/test/sync.test.ts | 168 ++++---- .../20260720000000_settings_sync_lww.sql | 100 +++++ .../tests/database/settings_sync_lww.test.sql | 240 ++++++++++++ 8 files changed, 1148 insertions(+), 162 deletions(-) create mode 100644 packages/sync/test/lww-fixtures.ts create mode 100644 packages/sync/test/lww.contract.test.ts create mode 100644 packages/sync/test/postgres-sync-client.ts create mode 100644 supabase/migrations/20260720000000_settings_sync_lww.sql create mode 100644 supabase/tests/database/settings_sync_lww.test.sql diff --git a/package.json b/package.json index 019041c..1168234 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,7 @@ "scripts": { "build": "bun run --cwd packages/tauri-release build && bun run --cwd packages/auth build && bun run --cwd packages/brand build && bun run --cwd packages/flags build && bun run --cwd packages/billing build && bun run --cwd packages/edge-shared build && bun run --cwd packages/sync build", "test": "vitest run", - "test:supabase": "supabase test db supabase/tests/database --local && bun run supabase/tests/welcome-credits-concurrency.ts && bun test packages/billing/test/checkout-lifecycle.contract.test.ts", + "test:supabase": "supabase test db supabase/tests/database --local && bun run supabase/tests/welcome-credits-concurrency.ts && bun test packages/billing/test/checkout-lifecycle.contract.test.ts && bun test packages/sync/test/lww.contract.test.ts", "test:coverage": "vitest run --coverage", "typecheck": "tsc -p tsconfig.json --noEmit", "check": "bun run typecheck && bun run test && bun run test:coverage && bun run build" diff --git a/packages/sync/src/index.ts b/packages/sync/src/index.ts index 57b5698..a7039e8 100644 --- a/packages/sync/src/index.ts +++ b/packages/sync/src/index.ts @@ -33,17 +33,8 @@ export interface SupabaseQueryResult { export interface SupabaseQueryBuilderLike extends PromiseLike> { select(columns?: string): SupabaseQueryBuilderLike; - update(values: unknown): SupabaseQueryBuilderLike; - upsert( - values: unknown, - options?: { - onConflict?: string; - ignoreDuplicates?: boolean; - }, - ): SupabaseQueryBuilderLike; eq(column: string, value: unknown): SupabaseQueryBuilderLike; is(column: string, value: unknown): SupabaseQueryBuilderLike; - lt(column: string, value: unknown): SupabaseQueryBuilderLike; order( column: string, options?: { @@ -55,6 +46,7 @@ export interface SupabaseQueryBuilderLike extends PromiseLike(table: string): SupabaseQueryBuilderLike; + rpc(fn: string, args?: Record): unknown; } export interface PushGroupOptions { @@ -171,25 +163,20 @@ export async function pushGroup({ const validGroup = validateFieldGroup(group); const validUpdatedAt = validateUpdatedAt(updatedAt); const sanitizedPayload = sanitizeSyncPayload(validGroup, payload); - const row = { + + const { written, row } = await applyLwwWrite(supabase, { user_id: validUserId, field_group: validGroup, payload: sanitizedPayload, updated_at: validUpdatedAt, deleted_at: null, - }; - - const written = await writeRowWithLww(supabase, row); - if (written !== null) { - return { status: "written", record: toSyncRecord(written) }; - } + }); - const server = await selectStoredGroup(supabase, validUserId, validGroup); - if (server === null) { - throw new SyncError("database_error", "Failed to read stale sync group"); + if (written) { + return { status: "written", record: toSyncRecord(row) }; } - return { status: "stale", record: server.deleted_at === null ? toSyncRecord(server) : null }; + return { status: "stale", record: row.deleted_at === null ? toSyncRecord(row) : null }; } export async function pullAll({ supabase, userId }: PullAllOptions): Promise { @@ -225,23 +212,19 @@ export async function deleteGroup({ const validGroup = validateFieldGroup(group); const validUpdatedAt = validateUpdatedAt(updatedAt); - const written = await writeRowWithLww(supabase, { + const { written, row } = await applyLwwWrite(supabase, { user_id: validUserId, field_group: validGroup, payload: {}, updated_at: validUpdatedAt, deleted_at: validUpdatedAt, }); - if (written !== null) { - return { status: "deleted" }; - } - const server = await selectStoredGroup(supabase, validUserId, validGroup); - if (server === null) { - throw new SyncError("database_error", "Failed to read stale sync group"); + if (written) { + return { status: "deleted" }; } - return { status: "stale", record: server.deleted_at === null ? toSyncRecord(server) : null }; + return { status: "stale", record: row.deleted_at === null ? toSyncRecord(row) : null }; } async function selectGroup( @@ -263,58 +246,60 @@ async function selectGroup( return data === null ? null : toSyncRecord(data); } -async function selectStoredGroup( +interface LwwWriteResult { + written: boolean; + row: SettingsSyncRow; +} + +/** + * Calls the durable `settings_sync_lww_write` Postgres function (see + * supabase/migrations/20260720000000_settings_sync_lww.sql), which locks the + * (user_id, field_group) row, compares `row.updated_at` against it, and + * performs whichever of insert/update/no-op wins, all inside one durable + * operation. It always returns the row that is now authoritative, so callers + * never need a follow-up read to learn the outcome. + */ +async function applyLwwWrite( supabase: SupabaseClientLike, - userId: string, - group: SyncFieldGroup, -): Promise { - const { data, error } = await supabase - .from("settings_sync") - .select(SYNC_COLUMNS) - .eq("user_id", userId) - .eq("field_group", group) - .maybeSingle(); + row: SettingsSyncRow, +): Promise { + const { data, error } = (await supabase.rpc("settings_sync_lww_write", { + target_user: row.user_id, + target_group: row.field_group, + new_payload: row.payload, + new_updated_at: row.updated_at, + new_deleted_at: row.deleted_at, + })) as SupabaseQueryResult; if (error !== null) { - throw databaseError("Failed to pull stored sync group", error); + throw databaseError("Failed to durably write sync group", error); } - return data; + return toLwwWriteResult(data); } -async function writeRowWithLww( - supabase: SupabaseClientLike, - row: SettingsSyncRow, -): Promise { - for (let attempt = 0; attempt < 2; attempt += 1) { - const updated = await supabase - .from("settings_sync") - .update(row) - .eq("user_id", row.user_id) - .eq("field_group", row.field_group) - .lt("updated_at", row.updated_at) - .select(SYNC_COLUMNS) - .maybeSingle(); - if (updated.error !== null) { - throw databaseError("Failed to update sync group", updated.error); - } - if (updated.data !== null) { - return updated.data; - } - - const inserted = await supabase - .from("settings_sync") - .upsert(row, { onConflict: "user_id,field_group", ignoreDuplicates: true }) - .select(SYNC_COLUMNS) - .maybeSingle(); - if (inserted.error !== null) { - throw databaseError("Failed to insert sync group", inserted.error); - } - if (inserted.data !== null) { - return inserted.data; - } +function toLwwWriteResult(data: unknown): LwwWriteResult { + if ( + !isRecord(data) || + typeof data.written !== "boolean" || + typeof data.user_id !== "string" || + typeof data.field_group !== "string" || + typeof data.updated_at !== "string" || + !isJson(data.payload) || + (data.deleted_at !== null && typeof data.deleted_at !== "string") + ) { + throw new SyncError("database_error", "settings_sync_lww_write returned an invalid result"); } - return null; + return { + written: data.written, + row: { + user_id: data.user_id, + field_group: data.field_group, + payload: data.payload, + updated_at: data.updated_at, + deleted_at: data.deleted_at, + }, + }; } function walkPayload(group: SyncFieldGroup, value: Json, path: string[]): void { diff --git a/packages/sync/test/lww-fixtures.ts b/packages/sync/test/lww-fixtures.ts new file mode 100644 index 0000000..77316af --- /dev/null +++ b/packages/sync/test/lww-fixtures.ts @@ -0,0 +1,99 @@ +// Local-Postgres contract-lane fixtures. Mirrors the `SUPABASE_DB_URL` / +// localhost-only guard used by `packages/billing/test/lifecycle-fixtures.ts` +// and `supabase/tests/welcome-credits-concurrency.ts` so this lane can never +// accidentally target a non-disposable database, and gracefully reports why +// it is skipped when no local Supabase is running. +import { SQL } from "bun"; + +export const DEFAULT_DATABASE_URL = "postgresql://postgres:postgres@127.0.0.1:54322/postgres"; + +export function resolveLocalDatabaseUrl(): string | null { + const raw = process.env.SUPABASE_DB_URL ?? DEFAULT_DATABASE_URL; + let parsed: URL; + try { + parsed = new URL(raw); + } catch { + return null; + } + const isDisposableLocalDatabase = + ["127.0.0.1", "localhost", "[::1]"].includes(parsed.hostname) && + parsed.port === "54322" && + parsed.pathname === "/postgres" && + parsed.username === "postgres"; + return isDisposableLocalDatabase ? raw : null; +} + +/** + * Connects to the local Supabase Postgres started by `supabase start`. Returns + * `null` (never throws) when the database is unreachable so the contract lane + * can skip cleanly instead of failing CI runs that have no local Postgres. + */ +export async function connectToLocalDatabase(): Promise { + const databaseUrl = resolveLocalDatabaseUrl(); + if (databaseUrl === null) { + return null; + } + + const sql = new SQL(databaseUrl, { max: 10 }); + try { + await sql.unsafe("select 1"); + } catch { + await sql.close().catch(() => {}); + return null; + } + return sql; +} + +export function uniqueId(prefix: string): string { + return `${prefix}_${crypto.randomUUID().replaceAll("-", "")}`; +} + +export async function insertAuthUser( + sql: Pick, + user: { id: string; email: string }, +): Promise { + await sql.unsafe( + `insert into auth.users ( + id, aud, role, email, raw_app_meta_data, raw_user_meta_data, is_anonymous, created_at, updated_at + ) values ($1, 'authenticated', 'authenticated', $2, '{}'::jsonb, '{}'::jsonb, false, now(), now())`, + [user.id, user.email], + ); +} + +/** + * Cleans up rows the contract lane cannot roll back transactionally: used + * only by the true-concurrency tests, which need two independent + * connections/transactions racing for real, so they commit instead of + * rolling back. Deleting the auth user cascades its settings_sync rows. + */ +export async function cleanupSyncFixtures( + sql: Pick, + fixtures: { userIds: string[] }, +): Promise { + if (fixtures.userIds.length === 0) { + return; + } + await sql.unsafe("delete from auth.users where id = any($1::uuid[])", [ + sql.array(fixtures.userIds, "uuid"), + ]); +} + +/** + * Runs `run` inside a transaction that always rolls back, so each contract + * test is fully isolated without needing bespoke per-table cleanup. Only the + * true-concurrency tests opt out of this helper, since they need two + * independently committing connections. + */ +export async function withRollback(sql: SQL, run: (tx: SQL) => Promise): Promise { + const rollbackSentinel = Symbol("contract-lane-rollback"); + try { + await sql.begin(async (tx) => { + await run(tx); + throw rollbackSentinel; + }); + } catch (error) { + if (error !== rollbackSentinel) { + throw error; + } + } +} diff --git a/packages/sync/test/lww.contract.test.ts b/packages/sync/test/lww.contract.test.ts new file mode 100644 index 0000000..e06085e --- /dev/null +++ b/packages/sync/test/lww.contract.test.ts @@ -0,0 +1,370 @@ +// Settings sync LWW contract lane (issue #37, CAND-3). +// +// Exercises the production `pushGroup` / `deleteGroup` / `pullGroup` policy +// against the REAL durable `settings_sync_lww_write` SQL function in +// supabase/migrations/20260720000000_settings_sync_lww.sql, running on a +// local Supabase Postgres. The compare/write/winner decision lives entirely +// behind the `SupabaseClientLike` boundary — this file never reimplements +// LWW policy, it only asserts on outcomes, including genuine concurrent +// races between independently committing Postgres transactions. +// +// Prerequisite: a local Supabase Postgres via `supabase start` (requires +// Docker). When it is not reachable, every test below is skipped with a +// console message instead of failing the run. CI's `database` job starts +// Supabase and runs this lane for real via `bun run test:supabase`. +import { afterAll, beforeAll, describe, expect, it } from "bun:test"; +import type { SQL } from "bun"; +import { deleteGroup, pullGroup, pushGroup, type PushGroupResult } from "../src/index.js"; +import { + cleanupSyncFixtures, + connectToLocalDatabase, + DEFAULT_DATABASE_URL, + insertAuthUser, + resolveLocalDatabaseUrl, + uniqueId, + withRollback, +} from "./lww-fixtures.js"; +import { createPostgresSyncClient } from "./postgres-sync-client.js"; + +const sql = await connectToLocalDatabase(); + +if (sql === null) { + console.log( + "[sync contract lane] Skipping settings-sync LWW Postgres contract tests: " + + `no local Supabase Postgres reachable at ${resolveLocalDatabaseUrl() ?? process.env.SUPABASE_DB_URL ?? DEFAULT_DATABASE_URL}. ` + + "Run `supabase start` (requires Docker) to exercise this lane locally; " + + "CI's `database` job runs it automatically via `bun run test:supabase`.", + ); +} + +describe.skipIf(sql === null)("@pickforge/sync settings sync LWW contract (local Postgres)", () => { + const db = sql as SQL; + + afterAll(async () => { + await db.close(); + }); + + async function createUser(tx: SQL, label: string): Promise { + const userId = crypto.randomUUID(); + await insertAuthUser(tx, { id: userId, email: `${uniqueId(label)}@example.invalid` }); + return userId; + } + + it("inserts an absent group and lets a strictly newer push win", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "push-newer-wins"); + const client = createPostgresSyncClient(tx, userId); + + await expect( + pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "dark" }, + updatedAt: "2026-07-09T04:00:00.000000Z", + }), + ).resolves.toMatchObject({ status: "written" }); + + await expect( + pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "light" }, + updatedAt: "2026-07-09T04:05:00.000000Z", + }), + ).resolves.toEqual({ + status: "written", + record: { fieldGroup: "appSettings", payload: { theme: "light" }, updatedAt: "2026-07-09T04:05:00.000000Z" }, + }); + + await expect(pullGroup({ supabase: client, userId, group: "appSettings" })).resolves.toEqual({ + fieldGroup: "appSettings", + payload: { theme: "light" }, + updatedAt: "2026-07-09T04:05:00.000000Z", + }); + }); + }); + + it("returns the winning row for a stale push without a follow-up read", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "push-stale"); + const client = createPostgresSyncClient(tx, userId); + + await pushGroup({ + supabase: client, + userId, + group: "operatorConfig", + payload: { threads: 8 }, + updatedAt: "2026-07-09T04:05:00.000000Z", + }); + + await expect( + pushGroup({ + supabase: client, + userId, + group: "operatorConfig", + payload: { threads: 2 }, + updatedAt: "2026-07-09T04:00:00.000000Z", + }), + ).resolves.toEqual({ + status: "stale", + record: { fieldGroup: "operatorConfig", payload: { threads: 8 }, updatedAt: "2026-07-09T04:05:00.000000Z" }, + }); + }); + }); + + it("treats an equal updated_at push as stale (first writer wins the tie)", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "push-equal-tie"); + const client = createPostgresSyncClient(tx, userId); + const tie = "2026-07-09T04:05:00.000000Z"; + + await pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "first" }, + updatedAt: tie, + }); + + await expect( + pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "second" }, + updatedAt: tie, + }), + ).resolves.toEqual({ + status: "stale", + record: { fieldGroup: "appSettings", payload: { theme: "first" }, updatedAt: tie }, + }); + }); + }); + + it("keeps older data from resurrecting a group after a newer tombstone", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "tombstone-resurrect"); + const client = createPostgresSyncClient(tx, userId); + + await pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "dark" }, + updatedAt: "2026-07-09T04:00:00.000000Z", + }); + await expect( + deleteGroup({ + supabase: client, + userId, + group: "appSettings", + updatedAt: "2026-07-09T04:10:00.000000Z", + }), + ).resolves.toEqual({ status: "deleted" }); + + await expect( + pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "late-arrival" }, + updatedAt: "2026-07-09T04:05:00.000000Z", + }), + ).resolves.toEqual({ status: "stale", record: null }); + await expect(pullGroup({ supabase: client, userId, group: "appSettings" })).resolves.toBeNull(); + + await expect( + pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "restored" }, + updatedAt: "2026-07-09T04:15:00.000000Z", + }), + ).resolves.toEqual({ + status: "written", + record: { fieldGroup: "appSettings", payload: { theme: "restored" }, updatedAt: "2026-07-09T04:15:00.000000Z" }, + }); + }); + }); + + it("resolves two genuinely concurrent pushes to exactly one durable, order-independent winner", async () => { + const userId = crypto.randomUUID(); + await insertAuthUser(db, { id: userId, email: `${uniqueId("concurrent-push-push")}@example.invalid` }); + + try { + const [olderOutcome, newerOutcome] = await Promise.all([ + db.begin(async (tx) => { + const client = createPostgresSyncClient(tx, userId); + return pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "older" }, + updatedAt: "2026-07-09T04:00:00.000000Z", + }); + }), + db.begin(async (tx) => { + const client = createPostgresSyncClient(tx, userId); + return pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "newer" }, + updatedAt: "2026-07-09T04:05:00.000000Z", + }); + }), + ]); + + // Whichever transaction committed first durably decided the outcome, + // but the newer updated_at always wins the durable comparison: no + // interleaving lets the older write survive. + expect(newerOutcome).toEqual({ + status: "written", + record: { fieldGroup: "appSettings", payload: { theme: "newer" }, updatedAt: "2026-07-09T04:05:00.000000Z" }, + }); + expect(olderOutcome.status === "stale" || olderOutcome.status === "written").toBe(true); + const olderResult = olderOutcome as PushGroupResult; + if (olderResult.status === "stale") { + expect(olderResult.record).toEqual({ + fieldGroup: "appSettings", + payload: { theme: "newer" }, + updatedAt: "2026-07-09T04:05:00.000000Z", + }); + } + + const rows = await db.unsafe( + "select payload, updated_at from public.settings_sync where user_id = $1 and field_group = 'appSettings'", + [userId], + ); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ payload: { theme: "newer" } }); + } finally { + await cleanupSyncFixtures(db, { userIds: [userId] }); + } + }); + + it("resolves a push racing a delete to exactly one durable, order-independent winner", async () => { + const userId = crypto.randomUUID(); + await insertAuthUser(db, { id: userId, email: `${uniqueId("concurrent-push-delete")}@example.invalid` }); + await db.unsafe( + `insert into public.settings_sync (user_id, field_group, payload, updated_at, deleted_at) + values ($1, 'appSettings', $2, $3, null)`, + [userId, { theme: "seed" }, "2026-07-09T03:00:00.000000Z"], + ); + + try { + const [pushOutcome, deleteOutcome] = await Promise.all([ + db.begin(async (tx) => { + const client = createPostgresSyncClient(tx, userId); + return pushGroup({ + supabase: client, + userId, + group: "appSettings", + payload: { theme: "pushed" }, + updatedAt: "2026-07-09T04:00:00.000000Z", + }); + }), + db.begin(async (tx) => { + const client = createPostgresSyncClient(tx, userId); + return deleteGroup({ + supabase: client, + userId, + group: "appSettings", + updatedAt: "2026-07-09T04:05:00.000000Z", + }); + }), + ]); + + // The delete has the strictly newer updated_at than both the seed row + // and the push, so it always durably wins in the end no matter which + // transaction's compare/write runs first. The push's own outcome + // depends on whether it ran before or after the delete (it can + // legitimately win against the older seed row first and then get + // superseded), so only the delete's outcome and the final row state + // are order-independent. + expect(deleteOutcome.status).toBe("deleted"); + expect(pushOutcome.status === "stale" || pushOutcome.status === "written").toBe(true); + + const rows = await db.unsafe( + "select deleted_at from public.settings_sync where user_id = $1 and field_group = 'appSettings'", + [userId], + ); + expect(rows).toHaveLength(1); + expect(rows[0]?.deleted_at).not.toBeNull(); + } finally { + await cleanupSyncFixtures(db, { userIds: [userId] }); + } + }); + + it("resolves two genuinely concurrent first inserts to exactly one durable, order-independent winner", async () => { + const userId = crypto.randomUUID(); + await insertAuthUser(db, { id: userId, email: `${uniqueId("concurrent-first-insert")}@example.invalid` }); + + try { + const [firstOutcome, secondOutcome] = await Promise.all([ + db.begin(async (tx) => { + const client = createPostgresSyncClient(tx, userId); + return pushGroup({ + supabase: client, + userId, + group: "keybindings", + payload: { bindings: [{ command: "save", shortcut: "Ctrl+S" }] }, + updatedAt: "2026-07-09T04:00:00.000000Z", + }); + }), + db.begin(async (tx) => { + const client = createPostgresSyncClient(tx, userId); + return pushGroup({ + supabase: client, + userId, + group: "keybindings", + payload: { bindings: [{ command: "save", shortcut: "Ctrl+Shift+S" }] }, + updatedAt: "2026-07-09T04:05:00.000000Z", + }); + }), + ]); + + // The strictly newer insert always durably wins, even though both + // transactions raced to insert the first row for this group. + expect(secondOutcome).toEqual({ + status: "written", + record: { + fieldGroup: "keybindings", + payload: { bindings: [{ command: "save", shortcut: "Ctrl+Shift+S" }] }, + updatedAt: "2026-07-09T04:05:00.000000Z", + }, + }); + expect(firstOutcome.status === "stale" || firstOutcome.status === "written").toBe(true); + + const rows = await db.unsafe( + "select payload from public.settings_sync where user_id = $1 and field_group = 'keybindings'", + [userId], + ); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ payload: { bindings: [{ command: "save", shortcut: "Ctrl+Shift+S" }] } }); + } finally { + await cleanupSyncFixtures(db, { userIds: [userId] }); + } + }); + + it("rejects a durable write to another user's row (per-user isolation)", async () => { + await withRollback(db, async (tx) => { + const ownerId = await createUser(tx, "isolation-owner"); + const attackerId = await createUser(tx, "isolation-attacker"); + const attackerClient = createPostgresSyncClient(tx, attackerId); + + await expect( + pushGroup({ + supabase: attackerClient, + userId: ownerId, + group: "appSettings", + payload: { theme: "stolen" }, + updatedAt: "2026-07-09T04:00:00.000000Z", + }), + ).rejects.toMatchObject({ name: "SyncError", code: "database_error" }); + }); + }); +}); diff --git a/packages/sync/test/postgres-sync-client.ts b/packages/sync/test/postgres-sync-client.ts new file mode 100644 index 0000000..2dadc49 --- /dev/null +++ b/packages/sync/test/postgres-sync-client.ts @@ -0,0 +1,202 @@ +// Local-Postgres contract-lane adapter. Implements the same `SupabaseClientLike` +// boundary the production `pushGroup` / `pullAll` / `pullGroup` / `deleteGroup` +// code is written against, but backed by a real `bun:sql` connection instead +// of the in-memory fake. No LWW decision vocabulary is reimplemented here: +// the compare/write/winner decision is made entirely by the durable +// `settings_sync_lww_write` Postgres function, exactly as production does. +// +// Unlike the billing contract lane's Postgres client, `settings_sync` is +// guarded by row-level security scoped to `auth.uid()`, not a service-role- +// only private schema, so every call here runs as the `authenticated` role +// with `request.jwt.claim.sub` set to the acting user — the same boundary a +// real client app session presents to Postgres. +import type { SQL } from "bun"; +import type { + SupabaseClientLike, + SupabaseErrorLike, + SupabaseQueryBuilderLike, + SupabaseQueryResult, +} from "../src/index.js"; + +const JSONB_RPCS = new Set(["settings_sync_lww_write"]); +const KNOWN_TABLES = ["settings_sync"] as const; +type KnownTable = (typeof KNOWN_TABLES)[number]; + +export function createPostgresSyncClient(sql: SQL, userId: string): SupabaseClientLike { + return { + from(table: string) { + if (!isKnownTable(table)) { + throw new Error(`Unknown table: ${table}`); + } + return new PostgresQuery(sql, table, userId); + }, + rpc(fn: string, args: Record = {}) { + return callRpc(sql, userId, fn, args); + }, + }; +} + +/** + * Every call authenticates as `userId` (mirroring a real PostgREST request) + * before running its query, and — when nested inside an outer per-test + * transaction (`withRollback`) — runs inside its own SAVEPOINT so an + * expected error (e.g. a cross-user isolation check) rolls back only that + * call, keeping the outer transaction usable, exactly like independent + * requests would behave in production. + */ +async function withUserSession( + sql: SQL, + userId: string, + run: (handle: SQL) => Promise, +): Promise { + const authenticate = async (handle: SQL) => { + await handle.unsafe("set local role authenticated"); + await handle.unsafe("select set_config('request.jwt.claim.sub', $1, true)", [userId]); + return run(handle); + }; + + const transactional = sql as SQL & { + savepoint?: (fn: (sp: SQL) => Promise) => Promise; + }; + if (typeof transactional.savepoint === "function") { + return transactional.savepoint((sp) => authenticate(sp)); + } + return authenticate(sql); +} + +async function callRpc( + sql: SQL, + userId: string, + fn: string, + args: Record, +): Promise> { + const keys = Object.keys(args); + const placeholders = keys.map((key, index) => `${key} => $${index + 1}`).join(", "); + // Bun's `postgres` driver double-encodes JSON.stringify'd object params + // bound to a jsonb-typed function argument (it JSON-serializes them again + // client-side), so pass objects (like new_payload) through as-is and let + // the driver encode them. + const values = keys.map((key) => args[key]); + const text = `select public.${fn}(${placeholders}) as result`; + + try { + const rows = await withUserSession(sql, userId, (handle) => handle.unsafe(text, values)); + const raw = rows[0]?.result; + if (JSONB_RPCS.has(fn)) { + return { data: typeof raw === "string" ? JSON.parse(raw) : (raw ?? null), error: null }; + } + return { data: raw === "" ? null : raw, error: null }; + } catch (error) { + return { data: null, error: toSupabaseError(error) }; + } +} + +function toSupabaseError(error: unknown): SupabaseErrorLike { + if (error !== null && typeof error === "object" && "message" in error) { + const pgError = error as { message: unknown; code?: unknown; errno?: unknown }; + const code = + typeof pgError.errno === "string" + ? pgError.errno + : typeof pgError.code === "string" + ? pgError.code + : undefined; + return { + code, + message: typeof pgError.message === "string" ? pgError.message : String(pgError.message), + }; + } + return { message: String(error) }; +} + +class PostgresQuery implements SupabaseQueryBuilderLike { + private columns = "*"; + private readonly filters: Array<{ column: string; op: "eq" | "is"; value: unknown }> = []; + private orderBy: { column: string; ascending: boolean } | null = null; + + constructor( + private readonly sql: SQL, + private readonly table: KnownTable, + private readonly userId: string, + ) {} + + select(columns = "*"): SupabaseQueryBuilderLike { + this.columns = columns; + return this; + } + + eq(column: string, value: unknown): SupabaseQueryBuilderLike { + this.filters.push({ column, op: "eq", value }); + return this; + } + + is(column: string, value: unknown): SupabaseQueryBuilderLike { + this.filters.push({ column, op: "is", value }); + return this; + } + + order(column: string, options?: { ascending?: boolean }): SupabaseQueryBuilderLike { + this.orderBy = { column, ascending: options?.ascending !== false }; + return this; + } + + async maybeSingle(): Promise> { + const result = await this.execute(); + const rows = Array.isArray(result.data) ? result.data : result.data === null ? [] : [result.data]; + return { data: (rows[0] ?? null) as T | null, error: result.error }; + } + + then, TResult2 = never>( + onfulfilled?: ((value: SupabaseQueryResult) => TResult1 | PromiseLike) | null, + onrejected?: ((reason: unknown) => TResult2 | PromiseLike) | null, + ): PromiseLike { + return this.execute().then(onfulfilled, onrejected); + } + + private async execute(): Promise> { + try { + return await withUserSession(this.sql, this.userId, (handle) => this.runSelect(handle)); + } catch (error) { + return { data: null, error: toSupabaseError(error) }; + } + } + + private async runSelect(handle: SQL): Promise> { + const params: unknown[] = []; + const where = this.filters + .map((filter) => { + if (filter.op === "is" && filter.value === null) { + return `${filter.column} is null`; + } + params.push(filter.value); + return filter.op === "is" + ? `${filter.column} is $${params.length}` + : `${filter.column} = $${params.length}`; + }) + .join(" and "); + let text = `select ${this.columns} from public.${this.table}`; + if (where.length > 0) { + text += ` where ${where}`; + } + if (this.orderBy !== null) { + text += ` order by ${this.orderBy.column} ${this.orderBy.ascending ? "asc" : "desc"}`; + } + const rows = await handle.unsafe(text, params); + return { data: rows.map(toTextRow) as T, error: null }; + } +} + +// Bun's `postgres` driver decodes `timestamptz` columns fetched via a plain +// SELECT into JS `Date` objects (unlike the RPC path, whose jsonb payload +// already comes back as ISO text), but `SupabaseClientLike` consumers expect +// ISO strings, matching what a real supabase-js/PostgREST client returns. +function toTextRow(row: Record): Record { + const converted: Record = {}; + for (const [key, value] of Object.entries(row)) { + converted[key] = value instanceof Date ? value.toISOString() : value; + } + return converted; +} + +function isKnownTable(table: string): table is KnownTable { + return (KNOWN_TABLES as readonly string[]).includes(table); +} diff --git a/packages/sync/test/sync.test.ts b/packages/sync/test/sync.test.ts index ee730bf..0979041 100644 --- a/packages/sync/test/sync.test.ts +++ b/packages/sync/test/sync.test.ts @@ -318,16 +318,14 @@ describe("@pickforge/sync", () => { }); }); - it("retries the conditional update after an ignored duplicate insert race", async () => { + it("treats an equal updated_at as stale, so the first writer wins a tie", async () => { const supabase = new MemorySupabase(); - supabase.beforeNextUpsert(() => { - supabase.tables.settings_sync.push({ - user_id: USER_ID, - field_group: "appSettings", - payload: { theme: "older" }, - updated_at: OLDER, - deleted_at: null, - }); + await pushGroup({ + supabase, + userId: USER_ID, + group: "appSettings", + payload: { theme: "first" }, + updatedAt: NEWER, }); await expect( @@ -335,14 +333,14 @@ describe("@pickforge/sync", () => { supabase, userId: USER_ID, group: "appSettings", - payload: { theme: "newer" }, + payload: { theme: "second" }, updatedAt: NEWER, }), ).resolves.toEqual({ - status: "written", + status: "stale", record: { fieldGroup: "appSettings", - payload: { theme: "newer" }, + payload: { theme: "first" }, updatedAt: NEWER, }, }); @@ -350,12 +348,35 @@ describe("@pickforge/sync", () => { { user_id: USER_ID, field_group: "appSettings", - payload: { theme: "newer" }, + payload: { theme: "first" }, updated_at: NEWER, deleted_at: null, }, ]); }); + + it("treats an equal updated_at delete race the same way: first writer wins", async () => { + const supabase = new MemorySupabase(); + await pushGroup({ + supabase, + userId: USER_ID, + group: "appSettings", + payload: { theme: "dark" }, + updatedAt: OLDER, + }); + await deleteGroup({ supabase, userId: USER_ID, group: "appSettings", updatedAt: DELETED }); + + await expect( + pushGroup({ + supabase, + userId: USER_ID, + group: "appSettings", + payload: { theme: "racing-push" }, + updatedAt: DELETED, + }), + ).resolves.toEqual({ status: "stale", record: null }); + await expect(pullGroup({ supabase, userId: USER_ID, group: "appSettings" })).resolves.toBeNull(); + }); }); interface SettingsSyncRow { @@ -375,8 +396,6 @@ class MemorySupabase implements SupabaseClientLike { settings_sync: [], }; - private readonly upsertHooks: Array<() => void> = []; - from(table: string): SupabaseQueryBuilderLike { if (table !== "settings_sync") { throw new Error(`Unknown table: ${table}`); @@ -385,22 +404,58 @@ class MemorySupabase implements SupabaseClientLike { return new MemoryQuery(this, table); } - beforeNextUpsert(hook: () => void): void { - this.upsertHooks.push(hook); + // Mirrors the durable `settings_sync_lww_write` Postgres function (see + // supabase/migrations/20260720000000_settings_sync_lww.sql): compares the + // incoming `updated_at` against any existing row for (user_id, field_group) + // and performs whichever of insert/update/no-op wins, returning whichever + // row is now authoritative. This fake is single-threaded, so it is + // inherently atomic; real concurrent-race coverage lives in the local- + // Postgres contract lane (packages/sync/test/lww.contract.test.ts). + rpc(fn: string, args: Record = {}): PromiseLike> { + if (fn !== "settings_sync_lww_write") { + throw new Error(`Unknown rpc: ${fn}`); + } + + return Promise.resolve(this.applyLwwWrite(args)); } - runBeforeUpsert(): void { - this.upsertHooks.shift()?.(); + private applyLwwWrite(args: Record): SupabaseQueryResult { + const targetUser = args.target_user as string; + const targetGroup = args.target_group as string; + const newPayload = args.new_payload as Json; + const newUpdatedAt = args.new_updated_at as string; + const newDeletedAt = (args.new_deleted_at as string | null) ?? null; + + const existing = this.tables.settings_sync.find( + (row) => row.user_id === targetUser && row.field_group === targetGroup, + ); + + if (existing === undefined) { + const row: SettingsSyncRow = { + user_id: targetUser, + field_group: targetGroup, + payload: newPayload, + updated_at: newUpdatedAt, + deleted_at: newDeletedAt, + }; + this.tables.settings_sync.push(row); + return { data: { written: true, ...row }, error: null }; + } + + if (newUpdatedAt > existing.updated_at) { + existing.payload = newPayload; + existing.updated_at = newUpdatedAt; + existing.deleted_at = newDeletedAt; + return { data: { written: true, ...existing }, error: null }; + } + + return { data: { written: false, ...existing }, error: null }; } } class MemoryQuery implements SupabaseQueryBuilderLike { - private action: "select" | "update" | "upsert" = "select"; - private values: Record | null = null; - private filters: Array<{ column: string; op: "eq" | "is" | "lt"; value: unknown }> = []; + private filters: Array<{ column: string; op: "eq" | "is"; value: unknown }> = []; private orderBy: { column: string; ascending: boolean } | null = null; - private onConflict: string | undefined; - private ignoreDuplicates = false; constructor( private readonly supabase: MemorySupabase, @@ -411,23 +466,6 @@ class MemoryQuery implements SupabaseQueryBuilderLike { return this; } - update(values: unknown): SupabaseQueryBuilderLike { - this.action = "update"; - this.values = values as Record; - return this; - } - - upsert( - values: unknown, - options?: { onConflict?: string; ignoreDuplicates?: boolean }, - ): SupabaseQueryBuilderLike { - this.action = "upsert"; - this.values = values as Record; - this.onConflict = options?.onConflict; - this.ignoreDuplicates = options?.ignoreDuplicates === true; - return this; - } - eq(column: string, value: unknown): SupabaseQueryBuilderLike { this.filters.push({ column, op: "eq", value }); return this; @@ -438,11 +476,6 @@ class MemoryQuery implements SupabaseQueryBuilderLike { return this; } - lt(column: string, value: unknown): SupabaseQueryBuilderLike { - this.filters.push({ column, op: "lt", value }); - return this; - } - order(column: string, options?: { ascending?: boolean }): SupabaseQueryBuilderLike { this.orderBy = { column, ascending: options?.ascending !== false }; return this; @@ -466,19 +499,6 @@ class MemoryQuery implements SupabaseQueryBuilderLike { } private async execute(): Promise> { - if (this.action === "update") { - const rows = this.matchingRows(); - for (const row of rows) { - Object.assign(row, this.values); - } - - return { data: rows as T, error: null }; - } - - if (this.action === "upsert") { - return this.upsertRow() as SupabaseQueryResult; - } - let rows = this.matchingRows(); if (this.orderBy !== null) { rows = [...rows].sort((left, right) => { @@ -493,40 +513,10 @@ class MemoryQuery implements SupabaseQueryBuilderLike { return { data: rows as T, error: null }; } - private upsertRow(): SupabaseQueryResult { - const value = this.values as unknown as SettingsSyncRow; - const conflictColumns = (this.onConflict ?? "").split(",").map((column) => column.trim()); - this.supabase.runBeforeUpsert(); - const existing = this.supabase.tables[this.table].find((row) => - conflictColumns.every( - (column) => row[column as keyof SettingsSyncRow] === value[column as keyof SettingsSyncRow], - ), - ); - - if (existing !== undefined) { - if (this.ignoreDuplicates) { - return { data: null, error: null }; - } - - Object.assign(existing, value); - return { data: [existing], error: null }; - } - - this.supabase.tables[this.table].push(value); - return { data: [value], error: null }; - } - private matchingRows(): SettingsSyncRow[] { return this.supabase.tables[this.table].filter((row) => this.filters.every((filter) => { const value = row[filter.column as keyof SettingsSyncRow]; - if (filter.op === "lt") { - return String(value) < String(filter.value); - } - if (filter.op === "is") { - return value === filter.value; - } - return value === filter.value; }), ); diff --git a/supabase/migrations/20260720000000_settings_sync_lww.sql b/supabase/migrations/20260720000000_settings_sync_lww.sql new file mode 100644 index 0000000..3d63ab7 --- /dev/null +++ b/supabase/migrations/20260720000000_settings_sync_lww.sql @@ -0,0 +1,100 @@ +-- Issue #37 (CAND-3): the settings sync last-writer-wins decision used to +-- cross multiple client/server round trips (a `.update()` guarded by +-- `updated_at < new updated_at`, then, only on a miss, an `.upsert()` with +-- `ignoreDuplicates`, then, only on a further miss, a third read to hand the +-- caller the winning row). That choreography reconstructed a durable +-- concurrency decision from client-driven retries instead of making it one +-- durable operation. +-- +-- `settings_sync_lww_write` is that one operation: it locks the +-- (user_id, field_group) row (if any), compares the caller's `updated_at` +-- against it, and performs whichever of insert/update/no-op wins, all inside +-- one function call/transaction. It always returns the row that is now +-- authoritative (the caller's write when it won, the existing row when it +-- didn't), so callers never need a follow-up read to learn the outcome. +-- +-- `security invoker` (the default) intentionally keeps this function inside +-- the existing `settings_sync` row-level-security boundary: it does not +-- re-implement per-user isolation, it relies on the same +-- `auth.uid() = user_id` select/insert/update policies the table already +-- enforces for direct client access, exactly like `public.credit_balance_cents` +-- (supabase/migrations/20260709010000_billing_customers_credit_ledger.sql). +create or replace function public.settings_sync_lww_write( + target_user uuid, + target_group text, + new_payload jsonb, + new_updated_at timestamptz, + new_deleted_at timestamptz +) +returns jsonb +language plpgsql +security invoker +set search_path = '' +as $$ +declare + current_row public.settings_sync%rowtype; +begin + select * + into current_row + from public.settings_sync + where user_id = target_user + and field_group = target_group + for update; + + if not found then + insert into public.settings_sync (user_id, field_group, payload, updated_at, deleted_at) + values (target_user, target_group, new_payload, new_updated_at, new_deleted_at) + on conflict (user_id, field_group) do nothing + returning * into current_row; + + if found then + return pg_catalog.jsonb_build_object( + 'written', true, + 'user_id', current_row.user_id, + 'field_group', current_row.field_group, + 'payload', current_row.payload, + 'updated_at', current_row.updated_at, + 'deleted_at', current_row.deleted_at + ); + end if; + + -- Lost the insert race: the winner's row now exists (and is visible, + -- since the conflicting insert waited for it to commit). Lock it and + -- fall through to the normal compare-and-decide path below. + select * + into current_row + from public.settings_sync + where user_id = target_user + and field_group = target_group + for update; + end if; + + if new_updated_at > current_row.updated_at then + update public.settings_sync + set payload = new_payload, + updated_at = new_updated_at, + deleted_at = new_deleted_at + where user_id = target_user + and field_group = target_group + returning * into current_row; + + return pg_catalog.jsonb_build_object( + 'written', true, + 'user_id', current_row.user_id, + 'field_group', current_row.field_group, + 'payload', current_row.payload, + 'updated_at', current_row.updated_at, + 'deleted_at', current_row.deleted_at + ); + end if; + + return pg_catalog.jsonb_build_object( + 'written', false, + 'user_id', current_row.user_id, + 'field_group', current_row.field_group, + 'payload', current_row.payload, + 'updated_at', current_row.updated_at, + 'deleted_at', current_row.deleted_at + ); +end; +$$; diff --git a/supabase/tests/database/settings_sync_lww.test.sql b/supabase/tests/database/settings_sync_lww.test.sql new file mode 100644 index 0000000..713c39e --- /dev/null +++ b/supabase/tests/database/settings_sync_lww.test.sql @@ -0,0 +1,240 @@ +begin; + +select plan(15); + +select ok( + has_function_privilege( + 'authenticated', + 'public.settings_sync_lww_write(uuid, text, jsonb, timestamptz, timestamptz)', + 'execute' + ), + 'authenticated can execute the durable LWW write function' +); + +insert into auth.users ( + id, aud, role, email, raw_app_meta_data, raw_user_meta_data, is_anonymous, created_at, updated_at +) +values + ( + '00000000-0000-0000-0000-000000000301', 'authenticated', 'authenticated', + 'settings-sync-lww-a@example.invalid', '{}'::jsonb, '{}'::jsonb, false, now(), now() + ), + ( + '00000000-0000-0000-0000-000000000302', 'authenticated', 'authenticated', + 'settings-sync-lww-b@example.invalid', '{}'::jsonb, '{}'::jsonb, false, now(), now() + ); + +set local role authenticated; +select set_config('request.jwt.claim.sub', '00000000-0000-0000-0000-000000000301', true); + +-- One durable operation both inserts an absent row and reports it as written. +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'appSettings', '{"theme":"dark"}'::jsonb, + '2026-07-09T04:00:00.000000Z'::timestamptz, null + ), + jsonb_build_object( + 'written', true, + 'user_id', '00000000-0000-0000-0000-000000000301', + 'field_group', 'appSettings', + 'payload', '{"theme":"dark"}'::jsonb, + 'updated_at', '2026-07-09T04:00:00.000000Z'::timestamptz, + 'deleted_at', null + ), + 'the first write for an absent row is durably inserted and reported written' +); + +-- A strictly newer write wins and replaces the row in the same call. +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'appSettings', '{"theme":"light"}'::jsonb, + '2026-07-09T04:05:00.000000Z'::timestamptz, null + ), + jsonb_build_object( + 'written', true, + 'user_id', '00000000-0000-0000-0000-000000000301', + 'field_group', 'appSettings', + 'payload', '{"theme":"light"}'::jsonb, + 'updated_at', '2026-07-09T04:05:00.000000Z'::timestamptz, + 'deleted_at', null + ), + 'a strictly newer updated_at overwrites the stored row' +); + +-- A strictly older write is stale and the winning row comes back without a +-- second read. +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'appSettings', '{"theme":"stale"}'::jsonb, + '2026-07-09T04:01:00.000000Z'::timestamptz, null + ), + jsonb_build_object( + 'written', false, + 'user_id', '00000000-0000-0000-0000-000000000301', + 'field_group', 'appSettings', + 'payload', '{"theme":"light"}'::jsonb, + 'updated_at', '2026-07-09T04:05:00.000000Z'::timestamptz, + 'deleted_at', null + ), + 'an older updated_at is stale and returns the still-current winning row' +); + +-- An equal updated_at is also stale: the first writer wins ties. +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'appSettings', '{"theme":"tie"}'::jsonb, + '2026-07-09T04:05:00.000000Z'::timestamptz, null + ), + jsonb_build_object( + 'written', false, + 'user_id', '00000000-0000-0000-0000-000000000301', + 'field_group', 'appSettings', + 'payload', '{"theme":"light"}'::jsonb, + 'updated_at', '2026-07-09T04:05:00.000000Z'::timestamptz, + 'deleted_at', null + ), + 'an equal updated_at is stale, so the first writer wins the tie' +); + +-- Microsecond precision is preserved through the compare/write decision. +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'operatorConfig', '{"threads":2}'::jsonb, + '2026-07-09T04:00:00.000001Z'::timestamptz, null + ), + jsonb_build_object( + 'written', true, + 'user_id', '00000000-0000-0000-0000-000000000301', + 'field_group', 'operatorConfig', + 'payload', '{"threads":2}'::jsonb, + 'updated_at', '2026-07-09T04:00:00.000001Z'::timestamptz, + 'deleted_at', null + ), + 'a microsecond-precision first write is durably inserted' +); +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'operatorConfig', '{"threads":8}'::jsonb, + '2026-07-09T04:00:00.000002Z'::timestamptz, null + ), + jsonb_build_object( + 'written', true, + 'user_id', '00000000-0000-0000-0000-000000000301', + 'field_group', 'operatorConfig', + 'payload', '{"threads":8}'::jsonb, + 'updated_at', '2026-07-09T04:00:00.000002Z'::timestamptz, + 'deleted_at', null + ), + 'a one-microsecond-newer write wins over the prior microsecond' +); + +-- A tombstone (deleted_at = updated_at) is just another LWW-compared write: +-- a newer tombstone wins over the live row it replaces. +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'appSettings', '{}'::jsonb, + '2026-07-09T04:10:00.000000Z'::timestamptz, '2026-07-09T04:10:00.000000Z'::timestamptz + ), + jsonb_build_object( + 'written', true, + 'user_id', '00000000-0000-0000-0000-000000000301', + 'field_group', 'appSettings', + 'payload', '{}'::jsonb, + 'updated_at', '2026-07-09T04:10:00.000000Z'::timestamptz, + 'deleted_at', '2026-07-09T04:10:00.000000Z'::timestamptz + ), + 'a tombstone write durably replaces the live row' +); + +-- Older data racing in after a newer tombstone stays stale and does not +-- resurrect the tombstoned group. +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'appSettings', '{"theme":"late-arrival"}'::jsonb, + '2026-07-09T04:05:00.000000Z'::timestamptz, null + ), + jsonb_build_object( + 'written', false, + 'user_id', '00000000-0000-0000-0000-000000000301', + 'field_group', 'appSettings', + 'payload', '{}'::jsonb, + 'updated_at', '2026-07-09T04:10:00.000000Z'::timestamptz, + 'deleted_at', '2026-07-09T04:10:00.000000Z'::timestamptz + ), + 'a push older than a tombstone stays stale and the tombstone is not resurrected' +); + +-- A newer push after the tombstone restores the group. +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'appSettings', '{"theme":"restored"}'::jsonb, + '2026-07-09T04:15:00.000000Z'::timestamptz, null + ), + jsonb_build_object( + 'written', true, + 'user_id', '00000000-0000-0000-0000-000000000301', + 'field_group', 'appSettings', + 'payload', '{"theme":"restored"}'::jsonb, + 'updated_at', '2026-07-09T04:15:00.000000Z'::timestamptz, + 'deleted_at', null + ), + 'a push newer than the tombstone clears it and restores the group' +); + +-- Far-future timestamps still hit the existing check constraint: the durable +-- operation does not loosen that invariant. +select throws_ok( + $$select public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000301', 'keybindings', '{}'::jsonb, + now() + interval '10 minutes', null + )$$, + '23514', + null, + 'a far-future updated_at still violates the existing check constraint' +); + +-- Per-user isolation: one authenticated user cannot durably write another +-- user's row through this function, exactly like direct table access today. +select throws_ok( + $$select public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000302', 'appSettings', '{}'::jsonb, now(), null + )$$, + '42501', + null, + 'the function cannot be used to write another user''s row' +); + +-- Nothing was written to user 302's rows by the isolation attempt above. +select is( + (select count(*)::integer from public.settings_sync where user_id = '00000000-0000-0000-0000-000000000302'), + 0, + 'the isolation attempt left the other user with no sync rows' +); + +-- A second user's groups are independent of the first user's LWW state. +select set_config('request.jwt.claim.sub', '00000000-0000-0000-0000-000000000302', true); +select is( + public.settings_sync_lww_write( + '00000000-0000-0000-0000-000000000302', 'appSettings', '{"theme":"own"}'::jsonb, + '2026-07-09T04:00:00.000000Z'::timestamptz, null + ), + jsonb_build_object( + 'written', true, + 'user_id', '00000000-0000-0000-0000-000000000302', + 'field_group', 'appSettings', + 'payload', '{"theme":"own"}'::jsonb, + 'updated_at', '2026-07-09T04:00:00.000000Z'::timestamptz, + 'deleted_at', null + ), + 'a second user has their own independent LWW state for the same group' +); +reset role; +select is( + (select payload from public.settings_sync where user_id = '00000000-0000-0000-0000-000000000301' and field_group = 'appSettings'), + '{"theme":"restored"}'::jsonb, + 'the first user''s row is unaffected by the second user''s write' +); + +select * from finish(); + +rollback;