diff --git a/packages/super-sync-server/src/sync/conflict.ts b/packages/super-sync-server/src/sync/conflict.ts index 8ad673fdab..668d382ef8 100644 --- a/packages/super-sync-server/src/sync/conflict.ts +++ b/packages/super-sync-server/src/sync/conflict.ts @@ -68,13 +68,18 @@ export const detectConflictForEntities = async ( start, start + CONFLICT_DETECTION_ENTITY_BATCH_SIZE, ); - // Match each requested id against the op's full entity set: entity_ids when - // present, else the scalar entity_id (pre-migration rows). The `&&`/`= ANY` - // prefilter keeps the GIN(entity_ids) + entity_id indexes usable; the unnest - // + `eid = ANY` keeps only the requested entities, then DISTINCT ON picks the - // latest op per entity. Kept inline (not a shared fragment) so the positional - // params stay stable for the conflict-detection.spec mock; the same shape is - // duplicated in prefetchLatestEntityOpsForBatch — keep them in sync. (#8334) + // Match each requested id against the op's full entity set: the entity_ids + // array UNION the scalar entity_id. The scalar is always folded in (not just + // when entity_ids is empty) so an op whose entity_id is NOT a member of its + // own entity_ids — possible when a multi-entity op dedups to a different + // primary, see getStoredEntityIds — still exposes that scalar entity here. + // DISTINCT ON dedupes the harmless overlap when entity_id is already in the + // array (the common entity_id = entityIds[0] case). The `&&`/`= ANY` + // prefilter keeps the GIN(entity_ids) + entity_id indexes usable; `eid = ANY` + // keeps only the requested entities, then DISTINCT ON picks the latest op per + // entity. Kept inline (not a shared fragment) so the positional params stay + // stable for the conflict-detection.spec mock; the same shape is duplicated in + // prefetchLatestEntityOpsForBatch — keep them in sync. (#8334) const idArray = Prisma.sql`ARRAY[${Prisma.join(batchEntityIds)}]::text[]`; const latestOps = await tx.$queryRaw` SELECT DISTINCT ON (eid) @@ -83,7 +88,7 @@ export const detectConflictForEntities = async ( o.vector_clock AS "vectorClock" FROM operations o CROSS JOIN LATERAL unnest( - CASE WHEN cardinality(o.entity_ids) > 0 THEN o.entity_ids ELSE ARRAY[o.entity_id] END + o.entity_ids || CASE WHEN o.entity_id IS NULL THEN '{}'::text[] ELSE ARRAY[o.entity_id] END ) AS eid WHERE o.user_id = ${userId} AND o.entity_type = ${op.entityType} @@ -386,7 +391,10 @@ export const prefetchLatestEntityOpsForBatch = async ( ); // Prefilter array (all requested ids) so the JOIN below can match a requested // id inside a stored op's entity_ids set, not just its scalar entity_id, while - // keeping the GIN(entity_ids) + entity_id indexes usable. (#8334) + // keeping the GIN(entity_ids) + entity_id indexes usable. The unnest folds the + // scalar entity_id into the entity_ids set (UNION, deduped by DISTINCT ON) so a + // divergent scalar is never missed — see the matching note in + // detectConflictForEntities; keep the two shapes in sync. (#8334) const idArray = Prisma.sql`ARRAY[${Prisma.join( batchPairs.map((pair) => pair.entityId), )}]::text[]`; @@ -399,7 +407,7 @@ export const prefetchLatestEntityOpsForBatch = async ( o.vector_clock AS "vectorClock" FROM operations o CROSS JOIN LATERAL unnest( - CASE WHEN cardinality(o.entity_ids) > 0 THEN o.entity_ids ELSE ARRAY[o.entity_id] END + o.entity_ids || CASE WHEN o.entity_id IS NULL THEN '{}'::text[] ELSE ARRAY[o.entity_id] END ) AS eid JOIN (VALUES ${Prisma.join(touchedRows)}) AS touched(entity_type, entity_id) ON touched.entity_type = o.entity_type diff --git a/packages/super-sync-server/tests/conflict-detection.spec.ts b/packages/super-sync-server/tests/conflict-detection.spec.ts index 475efcef40..bab775e2ed 100644 --- a/packages/super-sync-server/tests/conflict-detection.spec.ts +++ b/packages/super-sync-server/tests/conflict-detection.spec.ts @@ -222,14 +222,16 @@ vi.mock('../src/db', async () => { (entityId): entityId is string => typeof entityId === 'string', ), ); - // An op covers an entity via its full entity_ids set (multi-entity ops), - // falling back to the scalar entity_id for pre-migration rows (#8334). - const coveredEntityIds = (op: any): string[] => - Array.isArray(op.entityIds) && op.entityIds.length > 0 - ? op.entityIds - : op.entityId != null - ? [op.entityId] - : []; + // An op covers every entity in its entity_ids set UNION its scalar + // entity_id — mirrors the `entity_ids || ARRAY[entity_id]` unnest. The + // scalar is always folded in (not just for empty/pre-migration rows) so a + // divergent scalar entity_id is never missed; the Set below dedupes the + // common entity_id = entityIds[0] overlap. (#8334) + const coveredEntityIds = (op: any): string[] => { + const ids = Array.isArray(op.entityIds) ? [...op.entityIds] : []; + if (op.entityId != null) ids.push(op.entityId); + return ids; + }; const latestByEntityId = new Map(); const ops = Array.from(state.operations.values()) .filter((op: any) => op.userId === userId && op.entityType === entityType) diff --git a/packages/super-sync-server/tests/integration/conflict-detection-sql.integration.spec.ts b/packages/super-sync-server/tests/integration/conflict-detection-sql.integration.spec.ts new file mode 100644 index 0000000000..353834facf --- /dev/null +++ b/packages/super-sync-server/tests/integration/conflict-detection-sql.integration.spec.ts @@ -0,0 +1,294 @@ +/** + * Integration test for the multi-entity conflict-detection raw SQL (#8334). + * + * Runs the ACTUAL `detectConflict` / `prefetchLatestEntityOpsForBatch` functions + * (not a copy of the SQL, not a mock) against a REAL PostgreSQL database. Unit + * tests mock `$queryRaw` and can only verify the JS transformation around the + * query — this verifies the literal SQL: the `entity_ids || ARRAY[entity_id]` + * unnest, the `&&` / `= ANY` prefilter, the `DISTINCT ON` latest-per-entity pick, + * and the empty-array → scalar fallback for pre-migration rows. + * + * The decisive case is "divergent scalar": a stored op whose scalar `entity_id` + * is NOT a member of its own `entity_ids`. The previous mutually-exclusive + * `CASE WHEN cardinality(entity_ids) > 0 THEN entity_ids ELSE ARRAY[entity_id]` + * dropped that scalar from the batch lookup, so a later concurrent op touching it + * was wrongly accepted (silent data loss). The union covers it. + * + * Prerequisites (same as snapshot-vector-clock-sql.integration.spec.ts): + * - PostgreSQL running (see docker-compose.yaml), schema applied (prisma db push) + * - DATABASE_URL set, e.g. + * postgresql://supersync:superpassword@localhost:55432/supersync_db + * + * Run with (uses vitest.integration.config.ts — no mocked-Prisma setupFiles): + * DATABASE_URL=postgresql://... npx vitest run --config vitest.integration.config.ts \ + * tests/integration/conflict-detection-sql.integration.spec.ts + * + * NOTE: like the other *.integration.spec.ts files, this is excluded from the + * default `vitest run` (it needs a real DB) and skipped when DATABASE_URL is + * unset. GIN index *usage* is not asserted here (the planner seq-scans tiny test + * tables); it is covered structurally by migration-sql.spec.ts and in production. + */ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { PrismaClient, Prisma } from '@prisma/client'; +import { + detectConflict, + prefetchLatestEntityOpsForBatch, + getEntityConflictKey, +} from '../../src/sync/conflict'; +import { Operation, VectorClock } from '../../src/sync/sync.types'; + +const DATABASE_URL = process.env.DATABASE_URL; +const describeWithDb = DATABASE_URL ? describe : describe.skip; + +describeWithDb('Multi-entity conflict detection SQL (PostgreSQL) (#8334)', () => { + let prisma: PrismaClient; + const TEST_USER_ID = 99998; // Unlikely to collide with real data + const TEST_EMAIL = `test-conflict-sql-${Date.now()}@test.local`; + let opCounter = 0; + + // A PrismaClient satisfies the `$queryRaw` surface the conflict functions use. + const tx = (): Prisma.TransactionClient => + prisma as unknown as Prisma.TransactionClient; + + // Insert a stored operation row with an explicit scalar entity_id + entity_ids + // set, exactly as the upload path persists them. + const insertOp = async (args: { + serverSeq: number; + clientId: string; + entityId: string; + entityIds: string[]; + vectorClock: VectorClock; + }): Promise => { + opCounter++; + await prisma.operation.create({ + data: { + id: `test-conflict-sql-${opCounter}-${Date.now()}`, + userId: TEST_USER_ID, + clientId: args.clientId, + serverSeq: args.serverSeq, + actionType: '[Task] Update', + opType: 'UPD', + entityType: 'TASK', + entityId: args.entityId, + entityIds: args.entityIds, + payload: {}, + // eslint-disable-next-line @typescript-eslint/no-explicit-any -- Prisma JSON input; matches sibling integration specs + vectorClock: args.vectorClock as any, + schemaVersion: 1, + clientTimestamp: BigInt(Date.now()), + receivedAt: BigInt(Date.now()), + }, + }); + }; + + // A multi-entity incoming op forces the raw-SQL path (detectConflictForEntities + // only runs when ≥2 distinct entity ids are checked). The decoy id never + // matches a stored row, so any detected conflict comes from the real target. + const incomingMultiEntityOp = ( + clientId: string, + targetEntityId: string, + vectorClock: VectorClock, + ): Operation => ({ + id: `incoming-${clientId}-${Date.now()}`, + clientId, + actionType: '[Task] Update', + opType: 'UPD', + entityType: 'TASK', + entityIds: [targetEntityId, 'task-decoy-unmatched'], + payload: {}, + vectorClock, + timestamp: Date.now(), + schemaVersion: 1, + }); + + beforeAll(async () => { + prisma = new PrismaClient({ datasources: { db: { url: DATABASE_URL } } }); + await prisma.$connect(); + await prisma.user.create({ + data: { id: TEST_USER_ID, email: TEST_EMAIL, isVerified: 1 }, + }); + await prisma.userSyncState.create({ data: { userId: TEST_USER_ID, lastSeq: 0 } }); + }); + + afterAll(async () => { + await prisma.operation.deleteMany({ where: { userId: TEST_USER_ID } }); + await prisma.userSyncState.deleteMany({ where: { userId: TEST_USER_ID } }); + await prisma.user.deleteMany({ where: { id: TEST_USER_ID } }); + await prisma.$disconnect(); + }); + + beforeEach(async () => { + await prisma.operation.deleteMany({ where: { userId: TEST_USER_ID } }); + opCounter = 0; + }); + + it('detects a concurrent conflict on an entity inside a stored op entity_ids set', async () => { + // Common multi-entity row: entity_id = entityIds[0]. + await insertOp({ + serverSeq: 1, + clientId: 'A', + entityId: 'task-a', + entityIds: ['task-a', 'task-b'], + vectorClock: { A: 1 }, + }); + + // Concurrent op (clocks diverge) touching task-b, which lives in entity_ids. + const result = await detectConflict( + TEST_USER_ID, + incomingMultiEntityOp('B', 'task-b', { B: 1 }), + tx(), + ); + + expect(result.hasConflict).toBe(true); + expect(result.conflictType).toBe('concurrent'); + }); + + it('detects a conflict on a stored op whose scalar entity_id is NOT in its entity_ids (the union fix)', async () => { + // Divergent scalar: entity_id='task-z' is absent from entity_ids=['task-a']. + // The old mutually-exclusive CASE dropped task-z from the batch lookup. + await insertOp({ + serverSeq: 1, + clientId: 'A', + entityId: 'task-z', + entityIds: ['task-a'], + vectorClock: { A: 1 }, + }); + + const result = await detectConflict( + TEST_USER_ID, + incomingMultiEntityOp('B', 'task-z', { B: 1 }), + tx(), + ); + + // Would be { hasConflict: false } under the pre-fix CASE — silent data loss. + expect(result.hasConflict).toBe(true); + expect(result.conflictType).toBe('concurrent'); + }); + + it('falls back to the scalar entity_id for pre-migration rows (empty entity_ids)', async () => { + await insertOp({ + serverSeq: 1, + clientId: 'A', + entityId: 'task-p', + entityIds: [], // pre-migration default '{}' + vectorClock: { A: 1 }, + }); + + const result = await detectConflict( + TEST_USER_ID, + incomingMultiEntityOp('B', 'task-p', { B: 1 }), + tx(), + ); + + expect(result.hasConflict).toBe(true); + expect(result.conflictType).toBe('concurrent'); + }); + + it('does not flag a conflict when the incoming op is a clean successor', async () => { + await insertOp({ + serverSeq: 1, + clientId: 'A', + entityId: 'task-a', + entityIds: ['task-a', 'task-b'], + vectorClock: { A: 1 }, + }); + + // Same client, dominating clock → GREATER_THAN → no conflict. + const result = await detectConflict( + TEST_USER_ID, + incomingMultiEntityOp('A', 'task-b', { A: 2 }), + tx(), + ); + + expect(result.hasConflict).toBe(false); + }); + + it('compares against the LATEST stored op per entity (DISTINCT ON / server_seq DESC)', async () => { + await insertOp({ + serverSeq: 1, + clientId: 'A', + entityId: 'task-a', + entityIds: ['task-a'], + vectorClock: { A: 1 }, + }); + await insertOp({ + serverSeq: 2, + clientId: 'A', + entityId: 'task-a', + entityIds: ['task-a'], + vectorClock: { A: 2 }, + }); + + const result = await detectConflict( + TEST_USER_ID, + incomingMultiEntityOp('B', 'task-a', { B: 1 }), + tx(), + ); + + expect(result.hasConflict).toBe(true); + // existingClock must be the seq-2 clock, not seq-1. + expect(result.existingClock).toEqual({ A: 2 }); + }); + + it('does not consider operations from other users', async () => { + const OTHER_USER_ID = TEST_USER_ID + 1; + const OTHER_EMAIL = `other-conflict-${Date.now()}@test.local`; + await prisma.user.create({ + data: { id: OTHER_USER_ID, email: OTHER_EMAIL, isVerified: 1 }, + }); + try { + await prisma.operation.create({ + data: { + id: `other-user-op-${Date.now()}`, + userId: OTHER_USER_ID, + clientId: 'other', + serverSeq: 1, + actionType: '[Task] Update', + opType: 'UPD', + entityType: 'TASK', + entityId: 'task-shared', + entityIds: ['task-shared'], + payload: {}, + // eslint-disable-next-line @typescript-eslint/no-explicit-any -- Prisma JSON input; matches sibling integration specs + vectorClock: { other: 1 } as any, + schemaVersion: 1, + clientTimestamp: BigInt(Date.now()), + receivedAt: BigInt(Date.now()), + }, + }); + + const result = await detectConflict( + TEST_USER_ID, + incomingMultiEntityOp('B', 'task-shared', { B: 1 }), + tx(), + ); + + // The only writer of task-shared is a different user → no conflict for us. + expect(result.hasConflict).toBe(false); + } finally { + await prisma.operation.deleteMany({ where: { userId: OTHER_USER_ID } }); + await prisma.user.deleteMany({ where: { id: OTHER_USER_ID } }); + } + }); + + it('prefetchLatestEntityOpsForBatch covers a divergent scalar entity_id', async () => { + await insertOp({ + serverSeq: 1, + clientId: 'A', + entityId: 'task-z', + entityIds: ['task-a'], + vectorClock: { A: 1 }, + }); + + const latestByEntity = await prefetchLatestEntityOpsForBatch( + TEST_USER_ID, + [{ entityType: 'TASK', entityId: 'task-z' }], + tx(), + ); + + // Pre-fix this map was empty (task-z invisible); the union exposes it. + const row = latestByEntity.get(getEntityConflictKey('TASK', 'task-z')); + expect(row).toBeDefined(); + expect(row?.clientId).toBe('A'); + }); +}); diff --git a/packages/super-sync-server/vitest.config.ts b/packages/super-sync-server/vitest.config.ts index c1ad8ba7dc..cb43aefaf1 100644 --- a/packages/super-sync-server/vitest.config.ts +++ b/packages/super-sync-server/vitest.config.ts @@ -23,6 +23,7 @@ export default defineConfig({ 'tests/integration/snapshot-skip-optimization.integration.spec.ts', // Integration test that requires real PostgreSQL (run with vitest.integration.config.ts) 'tests/integration/snapshot-vector-clock-sql.integration.spec.ts', + 'tests/integration/conflict-detection-sql.integration.spec.ts', // Tests password reset routes that don't exist - server uses passkey/magic link auth 'tests/password-reset-api.spec.ts', ],