Merge branch 'feat/look-for-additional-super-sync-server-dae67c'

* feat/look-for-additional-super-sync-server-dae67c:
  perf(super-sync): optimize status and conflict checks

packages/shared-schema/src/index.ts

@@ -45,6 +45,7 @@ export {
   SUPER_SYNC_CLIENT_ID_REGEX,
   SUPER_SYNC_MAX_CLIENT_ID_LENGTH,
   SUPER_SYNC_MAX_OPS_PER_UPLOAD,
+  SUPER_SYNC_MAX_ENTITY_IDS_PER_OP,
   SUPER_SYNC_OP_TYPES,
   SUPER_SYNC_IMPORT_REASONS,
   SUPER_SYNC_SNAPSHOT_REASONS,

packages/shared-schema/src/supersync-http-contract.ts

@@ -3,6 +3,7 @@ import { z } from 'zod';
 export const SUPER_SYNC_CLIENT_ID_REGEX = /^[a-zA-Z0-9_-]+$/;
 export const SUPER_SYNC_MAX_CLIENT_ID_LENGTH = 255;
 export const SUPER_SYNC_MAX_OPS_PER_UPLOAD = 100;
+export const SUPER_SYNC_MAX_ENTITY_IDS_PER_OP = 1000;
 
 export const SUPER_SYNC_OP_TYPES = [
   'CRT',
@@ -50,7 +51,10 @@ export const SuperSyncOperationSchema = z.object({
   opType: z.enum(SUPER_SYNC_OP_TYPES),
   entityType: z.string().min(1).max(255),
   entityId: z.string().max(255).optional(),
-  entityIds: z.array(z.string().max(255)).optional(),
+  entityIds: z
+    .array(z.string().max(255))
+    .max(SUPER_SYNC_MAX_ENTITY_IDS_PER_OP)
+    .optional(),
   payload: z.unknown(),
   vectorClock: SuperSyncVectorClockSchema,
   timestamp: z.number(),

packages/shared-schema/tests/supersync-http-contract.spec.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it } from 'vitest';
 import {
+  SUPER_SYNC_MAX_ENTITY_IDS_PER_OP,
   SUPER_SYNC_MAX_OPS_PER_UPLOAD,
   SuperSyncDownloadOpsQuerySchema,
   SuperSyncDownloadOpsResponseSchema,
@@ -45,6 +46,23 @@ describe('SuperSync HTTP contract schemas', () => {
     expect('extraOpField' in parsed.ops[0]).toBe(false);
   });
 
+  it('caps entityIds per operation', () => {
+    expect(() =>
+      SuperSyncUploadOpsRequestSchema.parse({
+        ops: [
+          {
+            ...createValidOperation(),
+            entityIds: Array.from(
+              { length: SUPER_SYNC_MAX_ENTITY_IDS_PER_OP + 1 },
+              (_, i) => `task-${i}`,
+            ),
+          },
+        ],
+        clientId: 'client_1',
+      }),
+    ).toThrow();
+  });
+
   it('rejects invalid client IDs in upload requests', () => {
     expect(() =>
       SuperSyncUploadOpsRequestSchema.parse({

packages/super-sync-server/src/sync/services/snapshot.service.ts

@@ -191,6 +191,20 @@ export class SnapshotService {
     return Number(result[0]?.bytes ?? 0);
   }
 
+  /**
+   * Return cached snapshot timestamp without loading the snapshot blob.
+   * Status polling only needs snapshot age; reading, gunzipping, and parsing
+   * snapshotData there can turn a cheap endpoint into a large CPU/memory hit.
+   */
+  async getCachedSnapshotGeneratedAt(userId: number): Promise<number | null> {
+    const row = await prisma.userSyncState.findUnique({
+      where: { userId },
+      select: { snapshotAt: true },
+    });
+
+    return row?.snapshotAt != null ? Number(row.snapshotAt) : null;
+  }
+
   /**
    * Get cached snapshot for a user.
    */

packages/super-sync-server/src/sync/sync.routes.ts

@@ -1154,13 +1154,15 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise<void> => {
         const userId = getAuthUser(req).userId;
         const syncService = getSyncService();
 
-        const latestSeq = await syncService.getLatestSeq(userId);
-        const devicesOnline = await syncService.getOnlineDeviceCount(userId);
-
-        const cached = await syncService.getCachedSnapshot(userId);
-        const snapshotAge = cached ? Date.now() - cached.generatedAt : undefined;
-
-        const storageInfo = await syncService.getStorageInfo(userId);
+        const [latestSeq, devicesOnline, snapshotGeneratedAt, storageInfo] =
+          await Promise.all([
+            syncService.getLatestSeq(userId),
+            syncService.getOnlineDeviceCount(userId),
+            syncService.getCachedSnapshotGeneratedAt(userId),
+            syncService.getStorageInfo(userId),
+          ]);
+        const snapshotAge =
+          snapshotGeneratedAt !== null ? Date.now() - snapshotGeneratedAt : undefined;
 
         Logger.debug(
           `[user:${userId}] Status: seq=${latestSeq}, devices=${devicesOnline}`,

packages/super-sync-server/src/sync/sync.service.ts

@@ -42,6 +42,16 @@ interface DuplicateOperationCandidate {
   syncImportReason: string | null;
 }
 
+interface LatestEntityOperationRow {
+  entityId: string;
+  clientId: string;
+  vectorClock: unknown;
+}
+
+// Conservative enough to avoid planner-heavy BitmapOr + Sort plans on large
+// histories while still replacing up to 100 per-entity round trips with one query.
+const CONFLICT_DETECTION_ENTITY_BATCH_SIZE = 100;
+
 /**
  * Main sync orchestration service.
  *
@@ -98,60 +108,82 @@ export class SyncService {
 
     // Build list of entity IDs to check for conflicts.
     // Operations may have either entityId (singular) or entityIds (batch operations).
-    const entityIdsToCheck: string[] = op.entityIds?.length
+    const rawEntityIdsToCheck = op.entityIds?.length
       ? op.entityIds
       : op.entityId
         ? [op.entityId]
         : [];
 
     // Skip if no entity IDs (can't have entity-level conflicts)
-    if (entityIdsToCheck.length === 0) {
+    if (rawEntityIdsToCheck.length === 0) {
       return { hasConflict: false };
     }
 
-    // Check each entity for conflicts
-    for (const entityId of entityIdsToCheck) {
-      const conflictResult = await this.detectConflictForEntity(userId, op, entityId, tx);
-      if (conflictResult.hasConflict) {
-        return conflictResult;
+    if (rawEntityIdsToCheck.length === 1) {
+      return this.detectConflictForEntity(userId, op, rawEntityIdsToCheck[0], tx);
+    }
+
+    const entityIdsToCheck = Array.from(new Set(rawEntityIdsToCheck));
+    return this.detectConflictForEntities(userId, op, entityIdsToCheck, tx);
+  }
+
+  private async detectConflictForEntities(
+    userId: number,
+    op: Operation,
+    entityIdsToCheck: string[],
+    tx: Prisma.TransactionClient,
+  ): Promise<ConflictResult> {
+    for (
+      let start = 0;
+      start < entityIdsToCheck.length;
+      start += CONFLICT_DETECTION_ENTITY_BATCH_SIZE
+    ) {
+      const batchEntityIds = entityIdsToCheck.slice(
+        start,
+        start + CONFLICT_DETECTION_ENTITY_BATCH_SIZE,
+      );
+      const latestOps = await tx.$queryRaw<LatestEntityOperationRow[]>`
+        SELECT DISTINCT ON (entity_id)
+          entity_id AS "entityId",
+          client_id AS "clientId",
+          vector_clock AS "vectorClock"
+        FROM operations
+        WHERE user_id = ${userId}
+          AND entity_type = ${op.entityType}
+          AND entity_id IN (${Prisma.join(batchEntityIds)})
+        ORDER BY entity_id, server_seq DESC
+      `;
+
+      const latestOpByEntityId = new Map<string, LatestEntityOperationRow>();
+      for (const latestOp of latestOps) {
+        latestOpByEntityId.set(latestOp.entityId, latestOp);
+      }
+
+      for (const entityId of batchEntityIds) {
+        const existingOp = latestOpByEntityId.get(entityId);
+        if (!existingOp) continue;
+
+        const conflictResult = this.resolveConflictForExistingOp(
+          op,
+          entityId,
+          existingOp,
+        );
+        if (conflictResult.hasConflict) {
+          return conflictResult;
+        }
       }
     }
 
     return { hasConflict: false };
   }
 
-  /**
-   * Checks for conflicts on a single entity.
-   * Extracted from detectConflict to support multi-entity operations.
-   */
-  private async detectConflictForEntity(
-    userId: number,
+  private resolveConflictForExistingOp(
     op: Operation,
     entityId: string,
-    tx: Prisma.TransactionClient,
-  ): Promise<ConflictResult> {
-    // Get the latest operation for this entity
-    const existingOp = await tx.operation.findFirst({
-      where: {
-        userId,
-        entityType: op.entityType,
-        entityId,
-      },
-      select: {
-        clientId: true,
-        vectorClock: true,
-      },
-      orderBy: {
-        serverSeq: 'desc',
-      },
-    });
-
-    // No existing operation = no conflict
-    if (!existingOp) {
-      return { hasConflict: false };
-    }
-
-    // Parse the existing operation's vector clock (Prisma returns Json, cast to VectorClock)
+    existingOp: { clientId: string; vectorClock: unknown },
+  ): ConflictResult {
+    // Stored JSON/vector_clock values arrive as unknown from both Prisma model
+    // reads and raw SQL rows; cast only at the vector-clock comparison boundary.
     const existingClock = existingOp.vectorClock as unknown as VectorClock;
 
     // Compare vector clocks
@@ -208,6 +240,41 @@ export class SyncService {
     };
   }
 
+  /**
+   * Checks conflicts for the common single-entity upload path using Prisma's
+   * typed model API. Multi-entity operations use the batched raw-SQL path above
+   * to avoid one round trip per entity.
+   */
+  private async detectConflictForEntity(
+    userId: number,
+    op: Operation,
+    entityId: string,
+    tx: Prisma.TransactionClient,
+  ): Promise<ConflictResult> {
+    // Get the latest operation for this entity
+    const existingOp = await tx.operation.findFirst({
+      where: {
+        userId,
+        entityType: op.entityType,
+        entityId,
+      },
+      select: {
+        clientId: true,
+        vectorClock: true,
+      },
+      orderBy: {
+        serverSeq: 'desc',
+      },
+    });
+
+    // No existing operation = no conflict
+    if (!existingOp) {
+      return { hasConflict: false };
+    }
+
+    return this.resolveConflictForExistingOp(op, entityId, existingOp);
+  }
+
   private isSameDuplicateOperation(
     existingOp: DuplicateOperationCandidate,
     userId: number,
@@ -855,6 +922,10 @@ export class SyncService {
     return this.snapshotService.getCachedSnapshotBytes(userId);
   }
 
+  async getCachedSnapshotGeneratedAt(userId: number): Promise<number | null> {
+    return this.snapshotService.getCachedSnapshotGeneratedAt(userId);
+  }
+
   async cacheSnapshot(
     userId: number,
     state: unknown,

packages/super-sync-server/tests/conflict-detection.spec.ts

@@ -64,6 +64,7 @@ vi.mock('../src/db', async () => {
           );
         }
         if (args.where?.entityId && args.where?.entityType) {
+          state.entityConflictFindFirstCount++;
           const ops = Array.from(state.operations.values())
             .filter(
               (op: any) =>
@@ -162,6 +163,48 @@ vi.mock('../src/db', async () => {
     },
     // Upload transaction writes the storage counter atomically via $executeRaw.
     $executeRaw: vi.fn().mockResolvedValue(0),
+    $queryRaw: vi
+      .fn()
+      .mockImplementation(
+        async (
+          _strings: any,
+          userId: number,
+          entityType: string,
+          entityIdsSql: Prisma.Sql,
+        ) => {
+          state.batchConflictQueryCount++;
+          if (!Array.isArray(entityIdsSql.values)) {
+            throw new Error(
+              'Expected batched conflict query entity IDs to be passed via Prisma.join(...)',
+            );
+          }
+          const batchEntityIds = new Set(
+            entityIdsSql.values.filter(
+              (entityId): entityId is string => typeof entityId === 'string',
+            ),
+          );
+          const latestByEntityId = new Map<string, any>();
+          const ops = Array.from(state.operations.values())
+            .filter(
+              (op: any) =>
+                op.userId === userId &&
+                op.entityType === entityType &&
+                batchEntityIds.has(op.entityId),
+            )
+            .sort((a: any, b: any) => b.serverSeq - a.serverSeq);
+
+          for (const op of ops) {
+            if (!op.entityId || latestByEntityId.has(op.entityId)) continue;
+            latestByEntityId.set(op.entityId, op);
+          }
+
+          return Array.from(latestByEntityId.values()).map((op: any) => ({
+            entityId: op.entityId,
+            clientId: op.clientId,
+            vectorClock: op.vectorClock,
+          }));
+        },
+      ),
   });
 
   return {
@@ -877,5 +920,38 @@ describe('Conflict Detection', () => {
       const result = await service.uploadOps(userId, clientA, [op]);
       expect(result[0].accepted).toBe(true);
     });
+
+    it('should batch latest-op lookups for multi-entity conflict detection', async () => {
+      const service = getSyncService();
+
+      const existingOp = createOp({
+        entityId: 'task-2',
+        clientId: clientB,
+        vectorClock: { [clientB]: 1 },
+      });
+      const existingResult = await service.uploadOps(userId, clientB, [existingOp]);
+      expect(existingResult[0].accepted).toBe(true);
+      const beforeMultiEntityFindFirstCount = testState.entityConflictFindFirstCount;
+
+      const multiEntityOp = createOp({
+        entityId: 'task-1',
+        entityIds: ['task-1', 'task-2', 'task-3'],
+        clientId: clientA,
+        vectorClock: { [clientA]: 1 },
+      });
+
+      const result = await service.uploadOps(userId, clientA, [multiEntityOp]);
+
+      expect(testState.batchConflictQueryCount).toBeGreaterThan(0);
+      expect(testState.entityConflictFindFirstCount).toBe(
+        beforeMultiEntityFindFirstCount,
+      );
+      expect(result[0]).toMatchObject({
+        accepted: false,
+        errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT,
+        existingClock: { [clientB]: 1 },
+      });
+      expect(result[0].error).toContain('TASK:task-2');
+    });
   });
 });

packages/super-sync-server/tests/snapshot.service.spec.ts

@@ -10,6 +10,7 @@ vi.mock('../src/db', () => ({
   prisma: {
     userSyncState: {
       findUnique: vi.fn(),
+      findFirst: vi.fn(),
       update: vi.fn(),
       updateMany: vi.fn(),
       create: vi.fn(),
@@ -119,6 +120,31 @@ describe('SnapshotService', () => {
     });
   });
 
+  describe('getCachedSnapshotGeneratedAt', () => {
+    it('should return null when no cached snapshot metadata exists', async () => {
+      vi.mocked(prisma.userSyncState.findUnique).mockResolvedValue(null);
+
+      const result = await service.getCachedSnapshotGeneratedAt(1);
+
+      expect(result).toBeNull();
+    });
+
+    it('should read only snapshot metadata', async () => {
+      const snapshotAt = BigInt(1700000000000);
+      vi.mocked(prisma.userSyncState.findUnique).mockResolvedValue({
+        snapshotAt,
+      } as any);
+
+      const result = await service.getCachedSnapshotGeneratedAt(1);
+
+      expect(result).toBe(Number(snapshotAt));
+      expect(prisma.userSyncState.findUnique).toHaveBeenCalledWith({
+        where: { userId: 1 },
+        select: { snapshotAt: true },
+      });
+    });
+  });
+
   describe('cacheSnapshot', () => {
     it('should compress and conditionally update an existing row', async () => {
       vi.useFakeTimers();

packages/super-sync-server/tests/sync.service.test-state.ts

@@ -10,6 +10,8 @@ export const testState = {
   userSyncStates: new Map<number, any>(),
   users: new Map<number, any>(),
   serverSeqCounter: 0,
+  batchConflictQueryCount: 0,
+  entityConflictFindFirstCount: 0,
 };
 
 export function resetTestState(): void {
@@ -18,6 +20,8 @@ export function resetTestState(): void {
   testState.userSyncStates = new Map();
   testState.users = new Map();
   testState.serverSeqCounter = 0;
+  testState.batchConflictQueryCount = 0;
+  testState.entityConflictFindFirstCount = 0;
 }
 
 export function applyOperationSelect(op: any, select?: Record<string, boolean>): any {