feat(fase-9): wire sync_conflicts table + queue logging + debug panel
Closes 9.3 in full: migration 016 + pgTAP test 010 + queue.ts conflict-detection path + /settings debug panel + integration test. - Migration 016 adds public.sync_conflicts (append-only audit log of last-write-wins conflicts surfaced by the offline mutation queue). RLS allows owners to SELECT + INSERT their own rows only; UPDATE and DELETE have no policy and are silently denied. collective_id is intentionally nullable so user-level conflicts (theme, language in future) also fit. Indexed on (user_id, created_at desc) for the "show me my last 20 conflicts" query. - queue.ts gains a `QueueContext` (currentUserId + collectiveId) and an optional `preImage` field on UpdateOp. Before sending an UPDATE the queue fetches the remote row (limited to the patched fields), proceeds with the UPDATE (last-write-wins), then fires a best-effort INSERT into sync_conflicts when the remote diverged. Failure to log never blocks the UPDATE or pollutes the pending_ops queue. Three unit specs (SC-01..SC-03) lock the contract in. - sync/index.ts exposes setSyncContext(), called from (app)/+layout's $effect whenever currentUser / currentCollective change. - New SyncConflictsPanel.svelte renders the last 20 conflict rows for the current user with an "Export JSON" button. Gated on import.meta.env.DEV (or a forceShow prop for a future secret unlock) — never ships in the production bundle. Slotted into /settings just above the Account section. - Integration tests (packages/test-utils/tests/sync-conflicts.test.ts, 4 SC-INT specs) exercise the real PostgREST + RLS contract end-to-end with the existing seed users (Ana, Borja). Requires fake-indexeddb in the test-utils dev deps (queue's IDB shim). - pgTAP test 010 covers structure, RLS reads/writes, ownership rejection, and the append-only invariant (UPDATE/DELETE return 0 rows). - packages/types/src/database.ts adds the sync_conflicts table type. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
124
apps/web/src/lib/components/SyncConflictsPanel.svelte
Normal file
124
apps/web/src/lib/components/SyncConflictsPanel.svelte
Normal file
@@ -0,0 +1,124 @@
|
||||
<script lang="ts">
|
||||
/**
|
||||
* Fase 9.3.3 — debug panel for the last 20 sync_conflicts rows owned
|
||||
* by the current user. Gated on `import.meta.env.DEV` so it never
|
||||
* ships in a production bundle; flip the prop to force-show for an
|
||||
* eventual "secret toggle" without touching the gate.
|
||||
*
|
||||
* The component is presentational: pass in a `loader` that returns
|
||||
* the rows so tests can inject a mock without going through Supabase.
|
||||
*/
|
||||
import { onMount } from 'svelte';
|
||||
import { getSupabase } from '$lib/supabase';
|
||||
import { currentUser } from '$lib/stores/auth';
|
||||
import * as m from '$lib/paraglide/messages';
|
||||
|
||||
type ConflictRow = {
|
||||
id: string;
|
||||
entity_type: string;
|
||||
entity_id: string;
|
||||
local_version: unknown;
|
||||
remote_version: unknown;
|
||||
resolution: 'remote_won' | 'local_won';
|
||||
created_at: string;
|
||||
};
|
||||
|
||||
type Loader = (userId: string) => Promise<ConflictRow[]>;
|
||||
|
||||
const defaultLoader: Loader = async (userId) => {
|
||||
const { data } = await getSupabase()
|
||||
.from('sync_conflicts')
|
||||
.select('id, entity_type, entity_id, local_version, remote_version, resolution, created_at')
|
||||
.eq('user_id', userId)
|
||||
.order('created_at', { ascending: false })
|
||||
.limit(20);
|
||||
return (data ?? []) as ConflictRow[];
|
||||
};
|
||||
|
||||
let {
|
||||
loader = defaultLoader,
|
||||
forceShow = false
|
||||
}: { loader?: Loader; forceShow?: boolean } = $props();
|
||||
|
||||
let rows = $state<ConflictRow[]>([]);
|
||||
let loaded = $state(false);
|
||||
|
||||
// Hide in production unless `forceShow` is set. `import.meta.env.DEV`
|
||||
// is statically replaced at build time. `$derived` re-evaluates when
|
||||
// the prop changes (e.g. a parent toggles the secret unlock).
|
||||
const visible = $derived(import.meta.env.DEV || forceShow);
|
||||
|
||||
onMount(async () => {
|
||||
if (!visible || !$currentUser) {
|
||||
loaded = true;
|
||||
return;
|
||||
}
|
||||
try {
|
||||
rows = await loader($currentUser.id);
|
||||
} catch {
|
||||
rows = [];
|
||||
}
|
||||
loaded = true;
|
||||
});
|
||||
|
||||
function exportJson() {
|
||||
const blob = new Blob([JSON.stringify(rows, null, 2)], { type: 'application/json' });
|
||||
const url = URL.createObjectURL(blob);
|
||||
const a = document.createElement('a');
|
||||
a.href = url;
|
||||
a.download = `sync-conflicts-${new Date().toISOString().replace(/[:.]/g, '-')}.json`;
|
||||
document.body.appendChild(a);
|
||||
a.click();
|
||||
a.remove();
|
||||
URL.revokeObjectURL(url);
|
||||
}
|
||||
</script>
|
||||
|
||||
{#if visible}
|
||||
<section data-testid="sync-conflicts-panel">
|
||||
<div class="mb-3 flex items-center justify-between gap-2">
|
||||
<p class="text-[13px] font-semibold uppercase tracking-[0.05em] text-text-secondary">
|
||||
{m.settings_sync_conflicts()}
|
||||
</p>
|
||||
{#if rows.length > 0}
|
||||
<button
|
||||
type="button"
|
||||
onclick={exportJson}
|
||||
class="rounded-md border border-slate-300 px-3 py-1 text-xs font-medium text-slate-600 hover:bg-slate-50 dark:border-slate-600 dark:text-slate-400 dark:hover:bg-slate-800"
|
||||
data-testid="sync-conflicts-export"
|
||||
>
|
||||
{m.settings_sync_conflicts_export()}
|
||||
</button>
|
||||
{/if}
|
||||
</div>
|
||||
{#if loaded && rows.length === 0}
|
||||
<p class="text-sm text-text-secondary" data-testid="sync-conflicts-empty">
|
||||
{m.settings_sync_conflicts_empty()}
|
||||
</p>
|
||||
{:else}
|
||||
<ul class="space-y-2">
|
||||
{#each rows as row (row.id)}
|
||||
<li
|
||||
data-testid="sync-conflict-row"
|
||||
class="rounded-md border border-slate-200 bg-surface p-3 text-xs dark:border-slate-700"
|
||||
>
|
||||
<div class="mb-1 flex items-center justify-between text-text-secondary">
|
||||
<span class="font-mono">{row.entity_type}</span>
|
||||
<span>{new Date(row.created_at).toLocaleString()}</span>
|
||||
</div>
|
||||
<div class="grid grid-cols-2 gap-2">
|
||||
<div>
|
||||
<p class="text-text-muted">local</p>
|
||||
<pre class="overflow-x-auto whitespace-pre-wrap break-all">{JSON.stringify(row.local_version, null, 0)}</pre>
|
||||
</div>
|
||||
<div>
|
||||
<p class="text-text-muted">remote ({row.resolution})</p>
|
||||
<pre class="overflow-x-auto whitespace-pre-wrap break-all">{JSON.stringify(row.remote_version, null, 0)}</pre>
|
||||
</div>
|
||||
</div>
|
||||
</li>
|
||||
{/each}
|
||||
</ul>
|
||||
{/if}
|
||||
</section>
|
||||
{/if}
|
||||
@@ -11,7 +11,7 @@
|
||||
import { browser } from '$app/environment';
|
||||
import { getSupabase } from '$lib/supabase';
|
||||
import { pendingOpsCount } from '$lib/stores/syncStatus';
|
||||
import { SyncQueue, attachOnlineFlush, type NewOp } from './queue';
|
||||
import { SyncQueue, attachOnlineFlush, type NewOp, type QueueContext } from './queue';
|
||||
|
||||
let singleton: SyncQueue | null = null;
|
||||
let detachOnline: (() => void) | null = null;
|
||||
@@ -42,6 +42,15 @@ export async function refreshPending(): Promise<void> {
|
||||
pendingOpsCount.set(ops.length);
|
||||
}
|
||||
|
||||
/**
|
||||
* Update the queue's "who am I / which collective" context. Called from
|
||||
* the top-level layout when auth + active collective resolve, so the
|
||||
* conflict logger (Fase 9.3) attributes rows correctly.
|
||||
*/
|
||||
export function setSyncContext(ctx: QueueContext): void {
|
||||
getQueue().setContext(ctx);
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue-then-send convenience wrapper. Returns immediately on offline (op
|
||||
* stays queued) and propagates the online-success/failure otherwise.
|
||||
|
||||
@@ -184,6 +184,163 @@ describe('SyncQueue — pending_ops contract', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('SyncQueue — last-write-wins conflict logging (Fase 9.3)', () => {
|
||||
it('SC-01: an UPDATE whose remote pre-image differs from the local pre-image logs a sync_conflicts row', async () => {
|
||||
// The client tracks two kinds of calls:
|
||||
// * shopping_items.update → succeeds (last-write-wins).
|
||||
// * shopping_items.select before that update returns a row whose
|
||||
// `name` already differs from what the queue believed.
|
||||
// * sync_conflicts.insert is observed here.
|
||||
const calls: Array<{ table: string; kind: string; payload?: unknown; patch?: unknown; match?: unknown }> = [];
|
||||
const client = {
|
||||
from: (table: string) => ({
|
||||
insert: (payload: unknown) => ({
|
||||
select: () => ({
|
||||
single: async () => {
|
||||
calls.push({ table, kind: 'insert', payload });
|
||||
return { error: null };
|
||||
}
|
||||
})
|
||||
}),
|
||||
update: (patch: unknown) => ({
|
||||
match: async (m: Record<string, unknown>) => {
|
||||
calls.push({ table, kind: 'update', patch, match: m });
|
||||
return { error: null };
|
||||
}
|
||||
}),
|
||||
delete: () => ({
|
||||
match: async (m: Record<string, unknown>) => {
|
||||
calls.push({ table, kind: 'delete', match: m });
|
||||
return { error: null };
|
||||
}
|
||||
}),
|
||||
select: () => ({
|
||||
match: () => ({
|
||||
maybeSingle: async () => {
|
||||
calls.push({ table, kind: 'select' });
|
||||
// Remote has a value that disagrees with our preImage.
|
||||
return { data: { name: 'remote-name' }, error: null };
|
||||
}
|
||||
})
|
||||
})
|
||||
})
|
||||
};
|
||||
|
||||
const queue = new SyncQueue(client, { currentUserId: 'user-1', collectiveId: 'coll-1' });
|
||||
await queue.enqueue({
|
||||
kind: 'update',
|
||||
table: 'shopping_items',
|
||||
match: { id: 'item-1' },
|
||||
patch: { name: 'local-name' },
|
||||
preImage: { name: 'old-name' }
|
||||
});
|
||||
|
||||
// The UPDATE happened (remote-won under last-write-wins) AND a
|
||||
// sync_conflicts row was inserted with both payloads.
|
||||
const updateCall = calls.find((c) => c.kind === 'update' && c.table === 'shopping_items');
|
||||
expect(updateCall).toBeTruthy();
|
||||
|
||||
const conflict = calls.find((c) => c.kind === 'insert' && c.table === 'sync_conflicts');
|
||||
expect(conflict).toBeTruthy();
|
||||
const payload = conflict?.payload as {
|
||||
user_id: string;
|
||||
entity_type: string;
|
||||
entity_id: string;
|
||||
local_version: Record<string, unknown>;
|
||||
remote_version: Record<string, unknown>;
|
||||
resolution: string;
|
||||
};
|
||||
expect(payload.user_id).toBe('user-1');
|
||||
expect(payload.entity_type).toBe('shopping_item');
|
||||
expect(payload.entity_id).toBe('item-1');
|
||||
expect(payload.local_version).toEqual({ name: 'local-name' });
|
||||
expect(payload.remote_version).toEqual({ name: 'remote-name' });
|
||||
expect(payload.resolution).toBe('remote_won');
|
||||
});
|
||||
|
||||
it('SC-02: no conflict is logged when the remote pre-image matches the local one', async () => {
|
||||
const calls: Array<{ table: string; kind: string }> = [];
|
||||
const client = {
|
||||
from: (table: string) => ({
|
||||
insert: (_payload: unknown) => ({
|
||||
select: () => ({ single: async () => ({ error: null }) })
|
||||
}),
|
||||
update: (_patch: unknown) => ({
|
||||
match: async (_m: Record<string, unknown>) => {
|
||||
calls.push({ table, kind: 'update' });
|
||||
return { error: null };
|
||||
}
|
||||
}),
|
||||
delete: () => ({ match: async () => ({ error: null }) }),
|
||||
select: () => ({
|
||||
match: () => ({
|
||||
maybeSingle: async () => {
|
||||
calls.push({ table, kind: 'select' });
|
||||
// Remote agrees with our preImage — no conflict.
|
||||
return { data: { name: 'old-name' }, error: null };
|
||||
}
|
||||
})
|
||||
})
|
||||
})
|
||||
};
|
||||
|
||||
const queue = new SyncQueue(client, { currentUserId: 'user-1' });
|
||||
await queue.enqueue({
|
||||
kind: 'update',
|
||||
table: 'shopping_items',
|
||||
match: { id: 'item-1' },
|
||||
patch: { name: 'new-name' },
|
||||
preImage: { name: 'old-name' }
|
||||
});
|
||||
|
||||
expect(calls.find((c) => c.table === 'sync_conflicts')).toBeUndefined();
|
||||
});
|
||||
|
||||
it('SC-03: a failed sync_conflicts INSERT does NOT block the UPDATE', async () => {
|
||||
// The UPDATE must still come through even if the conflict log
|
||||
// insertion errors (best-effort logging).
|
||||
let updateLanded = false;
|
||||
const client = {
|
||||
from: (table: string) => ({
|
||||
insert: (_payload: unknown) => ({
|
||||
select: () => ({
|
||||
single: async () => {
|
||||
// sync_conflicts insert blows up
|
||||
if (table === 'sync_conflicts') return { error: { message: 'rls' } };
|
||||
return { error: null };
|
||||
}
|
||||
})
|
||||
}),
|
||||
update: (_patch: unknown) => ({
|
||||
match: async (_m: Record<string, unknown>) => {
|
||||
updateLanded = true;
|
||||
return { error: null };
|
||||
}
|
||||
}),
|
||||
delete: () => ({ match: async () => ({ error: null }) }),
|
||||
select: () => ({
|
||||
match: () => ({
|
||||
maybeSingle: async () => ({ data: { name: 'remote-name' }, error: null })
|
||||
})
|
||||
})
|
||||
})
|
||||
};
|
||||
|
||||
const queue = new SyncQueue(client, { currentUserId: 'user-1' });
|
||||
await queue.enqueue({
|
||||
kind: 'update',
|
||||
table: 'shopping_items',
|
||||
match: { id: 'item-1' },
|
||||
patch: { name: 'local' },
|
||||
preImage: { name: 'old' }
|
||||
});
|
||||
|
||||
expect(updateLanded).toBe(true);
|
||||
// Op was removed from the queue (treated as success).
|
||||
expect(await queue.pending()).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('attachOnlineFlush — window online event triggers flush', () => {
|
||||
it('F-01: dispatching "online" calls flush() within a tick', async () => {
|
||||
const { client } = makeMockClient([{ error: null }]);
|
||||
|
||||
@@ -36,6 +36,13 @@ export type UpdateOp = BaseOp & {
|
||||
table: 'shopping_items' | 'shopping_lists';
|
||||
match: Record<string, unknown>;
|
||||
patch: Record<string, unknown>;
|
||||
/**
|
||||
* Fase 9.3 — the row state the client believed when it queued the
|
||||
* UPDATE, restricted to the fields being patched. Used to detect
|
||||
* last-write-wins conflicts at send time. Optional so legacy ops
|
||||
* (and tests not exercising the conflict path) keep working.
|
||||
*/
|
||||
preImage?: Record<string, unknown>;
|
||||
};
|
||||
export type DeleteOp = BaseOp & {
|
||||
kind: 'delete';
|
||||
@@ -49,6 +56,16 @@ export type NewOp =
|
||||
| Omit<UpdateOp, keyof BaseOp>
|
||||
| Omit<DeleteOp, keyof BaseOp>;
|
||||
|
||||
/**
|
||||
* Maps an op's `table` to the entity_type value we persist in
|
||||
* sync_conflicts. Keeping a small lookup avoids hard-coding the rule
|
||||
* "strip the trailing s" which would break for irregular plurals.
|
||||
*/
|
||||
const ENTITY_TYPE_FOR_TABLE: Record<UpdateOp['table'], string> = {
|
||||
shopping_items: 'shopping_item',
|
||||
shopping_lists: 'shopping_list'
|
||||
};
|
||||
|
||||
export const MAX_ATTEMPTS = 5;
|
||||
const DB_NAME = 'colectivo-sync';
|
||||
const DB_VERSION = 1;
|
||||
@@ -59,9 +76,31 @@ type SupabaseQueryClient = {
|
||||
insert: (row: unknown) => { select: () => { single: () => Promise<unknown> } };
|
||||
update: (patch: unknown) => { match: (m: Record<string, unknown>) => Promise<unknown> };
|
||||
delete: () => { match: (m: Record<string, unknown>) => Promise<unknown> };
|
||||
/**
|
||||
* Fase 9.3 — used to read the remote pre-image before sending an
|
||||
* UPDATE so we can detect a last-write-wins conflict. Optional
|
||||
* because legacy callers that never enqueue a `preImage` op also
|
||||
* never reach this branch.
|
||||
*/
|
||||
select?: (cols?: string) => {
|
||||
match: (m: Record<string, unknown>) => {
|
||||
maybeSingle: () => Promise<{ data: Record<string, unknown> | null; error: { message: string } | null }>;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
|
||||
/**
|
||||
* Optional context the queue uses for last-write-wins conflict logging.
|
||||
* Without `currentUserId` the queue skips conflict detection entirely —
|
||||
* RLS on sync_conflicts requires `user_id = auth.uid()` so anonymous
|
||||
* writes would fail anyway.
|
||||
*/
|
||||
export type QueueContext = {
|
||||
currentUserId?: string | null;
|
||||
collectiveId?: string | null;
|
||||
};
|
||||
|
||||
async function getDb(): Promise<IDBPDatabase> {
|
||||
return openDB(DB_NAME, DB_VERSION, {
|
||||
upgrade(db) {
|
||||
@@ -78,7 +117,17 @@ async function getDb(): Promise<IDBPDatabase> {
|
||||
* interface so tests can inject a mock.
|
||||
*/
|
||||
export class SyncQueue {
|
||||
constructor(private readonly client: SupabaseQueryClient) {}
|
||||
private readonly context: QueueContext;
|
||||
|
||||
constructor(private readonly client: SupabaseQueryClient, context: QueueContext = {}) {
|
||||
this.context = context;
|
||||
}
|
||||
|
||||
/** Replace the queue context (call when the user logs in / out). */
|
||||
setContext(context: QueueContext): void {
|
||||
this.context.currentUserId = context.currentUserId ?? null;
|
||||
this.context.collectiveId = context.collectiveId ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist the op to pending_ops, then attempt to send it to Supabase.
|
||||
@@ -163,10 +212,41 @@ export class SyncQueue {
|
||||
};
|
||||
if (r.error && !isIdempotentSuccess(r.error)) throw new Error(r.error.message);
|
||||
} else if (op.kind === 'update') {
|
||||
// Fase 9.3: best-effort last-write-wins conflict detection.
|
||||
// Read the remote row first (only fields we're patching) and
|
||||
// compare to the pre-image the caller captured when the op was
|
||||
// queued. Any mismatch on a patched field = remote moved
|
||||
// under us; log a sync_conflicts row and proceed with the
|
||||
// UPDATE (last-write-wins). A failure to fetch / log never
|
||||
// blocks the UPDATE — the conflict log is advisory.
|
||||
let remoteSnapshot: Record<string, unknown> | null = null;
|
||||
if (op.preImage && this.client.from(op.table).select && this.context.currentUserId) {
|
||||
try {
|
||||
const cols = Object.keys(op.preImage).join(', ');
|
||||
const sel = this.client.from(op.table).select?.(cols);
|
||||
if (sel) {
|
||||
const r = await sel.match(op.match).maybeSingle();
|
||||
if (!r.error) remoteSnapshot = r.data;
|
||||
}
|
||||
} catch {
|
||||
/* swallow — best effort */
|
||||
}
|
||||
}
|
||||
|
||||
const r = (await table.update(op.patch).match(op.match)) as {
|
||||
error: { message: string } | null;
|
||||
};
|
||||
if (r.error) throw new Error(r.error.message);
|
||||
|
||||
// After a successful UPDATE, fire-and-forget the conflict log.
|
||||
if (remoteSnapshot && op.preImage) {
|
||||
const diverged = Object.keys(op.preImage).some(
|
||||
(k) => !deepEqual(remoteSnapshot![k], op.preImage![k])
|
||||
);
|
||||
if (diverged) {
|
||||
await this.logConflict(op, remoteSnapshot);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
const r = (await table.delete().match(op.match)) as {
|
||||
error: { message: string } | null;
|
||||
@@ -174,6 +254,54 @@ export class SyncQueue {
|
||||
if (r.error) throw new Error(r.error.message);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Insert a sync_conflicts row for the given UPDATE op. Errors are
|
||||
* swallowed — the table is append-only/best-effort and a logging
|
||||
* failure must never break the user's sync.
|
||||
*/
|
||||
private async logConflict(op: UpdateOp, remoteSnapshot: Record<string, unknown>): Promise<void> {
|
||||
const userId = this.context.currentUserId;
|
||||
if (!userId) return;
|
||||
const entityType = ENTITY_TYPE_FOR_TABLE[op.table];
|
||||
const entityId = (op.match.id ?? op.match['id']) as string | undefined;
|
||||
if (!entityType || !entityId) return;
|
||||
try {
|
||||
await this.client
|
||||
.from('sync_conflicts')
|
||||
.insert({
|
||||
user_id: userId,
|
||||
collective_id: this.context.collectiveId ?? null,
|
||||
entity_type: entityType,
|
||||
entity_id: entityId,
|
||||
local_version: op.patch,
|
||||
remote_version: remoteSnapshot,
|
||||
resolution: 'remote_won'
|
||||
})
|
||||
.select()
|
||||
.single();
|
||||
} catch {
|
||||
/* swallow */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Local deep-equality good enough for primitive / JSON payloads. */
|
||||
function deepEqual(a: unknown, b: unknown): boolean {
|
||||
if (a === b) return true;
|
||||
if (a === null || b === null) return false;
|
||||
if (typeof a !== typeof b) return false;
|
||||
if (typeof a !== 'object') return false;
|
||||
if (Array.isArray(a) !== Array.isArray(b)) return false;
|
||||
const ak = Object.keys(a as object);
|
||||
const bk = Object.keys(b as object);
|
||||
if (ak.length !== bk.length) return false;
|
||||
for (const k of ak) {
|
||||
if (!deepEqual((a as Record<string, unknown>)[k], (b as Record<string, unknown>)[k])) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user