fix(sync): cover scalar entity_id in multi-entity conflict lookup (#8334)

The two raw-SQL latest-writer lookups unnested a mutually-exclusive
`CASE WHEN cardinality(entity_ids) > 0 THEN entity_ids ELSE ARRAY[entity_id]`,
so a stored op whose scalar entity_id is NOT a member of its own entity_ids
(producible via getStoredEntityIds dedup) was invisible to the batch lookup -
a later concurrent op touching that entity was wrongly accepted (silent data
loss). This only bites once batch upload is enabled in prod, but is a real
latent divergence between the single-entity (Prisma OR) and batch (raw SQL)
paths.

Fold the scalar into the set with a union:
  entity_ids || CASE WHEN entity_id IS NULL THEN '{}'::text[] ELSE ARRAY[entity_id] END
(entity_id is nullable; the guard prevents array||NULL nulling the whole
array). DISTINCT ON dedupes the common entity_id = entityIds[0] overlap. The
WHERE prefilter is unchanged, so GIN(entity_ids) + entity_id btree usage is
preserved. Update the conflict-detection.spec mock to union too, and add a
DATABASE_URL-gated PG integration test exercising the real detectConflict /
prefetchLatestEntityOpsForBatch (divergent-scalar case fails under the old
CASE); excluded from the default vitest run like the sibling SQL integration test.

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<LatestEntityOperationRow[]>`
       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

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<string, any>();
       const ops = Array.from(state.operations.values())
         .filter((op: any) => op.userId === userId && op.entityType === entityType)

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<void> => {
+    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');
+  });
+});

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',
     ],