From 07589dd67fa3dba0e1c7115c16a0f56f708d4260 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Sat, 27 Dec 2025 20:43:25 +0100 Subject: [PATCH] refactor(sync): extract DeviceService and OperationDownloadService Continue SyncService extraction with two more services: - DeviceService: device ownership and online status queries (isDeviceOwner, getAllUserIds, getOnlineDeviceCount) - OperationDownloadService: operation retrieval with gap detection and snapshot optimization (getOpsSince, getOpsSinceWithSeq, getLatestSeq) SyncService now delegates to 6 extracted services: - ValidationService - RateLimitService - RequestDeduplicationService - DeviceService - OperationDownloadService All 236 tests passing. --- .../src/sync/services/device.service.ts | 47 ++++ .../src/sync/services/index.ts | 2 + .../services/operation-download.service.ts | 238 ++++++++++++++++++ .../src/sync/sync.service.ts | 236 ++--------------- .../tests/device.service.spec.ts | 139 ++++++++++ 5 files changed, 449 insertions(+), 213 deletions(-) create mode 100644 packages/super-sync-server/src/sync/services/device.service.ts create mode 100644 packages/super-sync-server/src/sync/services/operation-download.service.ts create mode 100644 packages/super-sync-server/tests/device.service.spec.ts diff --git a/packages/super-sync-server/src/sync/services/device.service.ts b/packages/super-sync-server/src/sync/services/device.service.ts new file mode 100644 index 0000000000..a48deaf8f3 --- /dev/null +++ b/packages/super-sync-server/src/sync/services/device.service.ts @@ -0,0 +1,47 @@ +/** + * DeviceService - Handles device-related queries + * + * Extracted from SyncService for better separation of concerns. + * This service handles device ownership and online status queries. + */ +import { prisma } from '../../db'; +import { ONLINE_DEVICE_THRESHOLD_MS } from '../sync.types'; + +export class DeviceService { + /** + * Check if a device (identified by clientId) belongs to a user. + */ + async isDeviceOwner(userId: number, clientId: string): Promise { + const count = await prisma.syncDevice.count({ + where: { userId, clientId }, + }); + return count > 0; + } + + /** + * Get all user IDs that have sync state. + * Used for batch operations like cleanup. + */ + async getAllUserIds(): Promise { + const users = await prisma.userSyncState.findMany({ + select: { userId: true }, + distinct: ['userId'], + }); + return users.map((u) => u.userId); + } + + /** + * Get count of devices that have been seen recently for a user. + * A device is considered "online" if it was seen within the threshold. + */ + async getOnlineDeviceCount(userId: number): Promise { + const threshold = Date.now() - ONLINE_DEVICE_THRESHOLD_MS; + const count = await prisma.syncDevice.count({ + where: { + userId, + lastSeenAt: { gt: BigInt(threshold) }, + }, + }); + return count; + } +} diff --git a/packages/super-sync-server/src/sync/services/index.ts b/packages/super-sync-server/src/sync/services/index.ts index ee1029f3af..f3a9c93696 100644 --- a/packages/super-sync-server/src/sync/services/index.ts +++ b/packages/super-sync-server/src/sync/services/index.ts @@ -8,3 +8,5 @@ export { ValidationService, ALLOWED_ENTITY_TYPES } from './validation.service'; export type { ValidationResult } from './validation.service'; export { RateLimitService } from './rate-limit.service'; export { RequestDeduplicationService } from './request-deduplication.service'; +export { DeviceService } from './device.service'; +export { OperationDownloadService } from './operation-download.service'; diff --git a/packages/super-sync-server/src/sync/services/operation-download.service.ts b/packages/super-sync-server/src/sync/services/operation-download.service.ts new file mode 100644 index 0000000000..c64fb9339a --- /dev/null +++ b/packages/super-sync-server/src/sync/services/operation-download.service.ts @@ -0,0 +1,238 @@ +/** + * OperationDownloadService - Handles downloading operations for clients + * + * Extracted from SyncService for better separation of concerns. + * This service handles operation retrieval with gap detection and snapshot optimization. + */ +import { prisma } from '../../db'; +import { Operation, ServerOperation, VectorClock } from '../sync.types'; +import { Logger } from '../../logger'; + +export class OperationDownloadService { + /** + * Get operations since a given sequence number. + * Simple version without gap detection. + */ + async getOpsSince( + userId: number, + sinceSeq: number, + excludeClient?: string, + limit: number = 500, + ): Promise { + const ops = await prisma.operation.findMany({ + where: { + userId, + serverSeq: { gt: sinceSeq }, + ...(excludeClient ? { clientId: { not: excludeClient } } : {}), + }, + orderBy: { + serverSeq: 'asc', + }, + take: limit, + }); + + return ops.map((row) => ({ + serverSeq: row.serverSeq, + op: { + id: row.id, + clientId: row.clientId, + actionType: row.actionType, + opType: row.opType as Operation['opType'], + entityType: row.entityType, + entityId: row.entityId ?? undefined, + payload: row.payload, + vectorClock: row.vectorClock as unknown as VectorClock, + schemaVersion: row.schemaVersion, + timestamp: Number(row.clientTimestamp), + parentOpId: row.parentOpId ?? undefined, + isPayloadEncrypted: row.isPayloadEncrypted, + }, + receivedAt: Number(row.receivedAt), + })); + } + + /** + * Get operations and latest sequence atomically with gap detection. + * + * OPTIMIZATION: When sinceSeq is before the latest full-state operation (SYNC_IMPORT, + * BACKUP_IMPORT, REPAIR), we skip to that operation's sequence instead. This prevents + * sending operations that will be filtered out by the client anyway, saving bandwidth + * and processing time. + */ + async getOpsSinceWithSeq( + userId: number, + sinceSeq: number, + excludeClient?: string, + limit: number = 500, + ): Promise<{ + ops: ServerOperation[]; + latestSeq: number; + gapDetected: boolean; + latestSnapshotSeq?: number; + snapshotVectorClock?: VectorClock; + }> { + return prisma.$transaction(async (tx) => { + // Find the latest full-state operation (SYNC_IMPORT, BACKUP_IMPORT, REPAIR) + // These operations supersede all previous operations + const latestFullStateOp = await tx.operation.findFirst({ + where: { + userId, + opType: { in: ['SYNC_IMPORT', 'BACKUP_IMPORT', 'REPAIR'] }, + }, + orderBy: { serverSeq: 'desc' }, + select: { serverSeq: true }, + }); + + const latestSnapshotSeq = latestFullStateOp?.serverSeq ?? undefined; + + // OPTIMIZATION: If client is requesting ops from before the latest full-state op, + // start from the full-state op instead. Pre-import ops are superseded and will + // be filtered out by the client anyway. + let effectiveSinceSeq = sinceSeq; + let snapshotVectorClock: VectorClock | undefined; + + if (latestSnapshotSeq !== undefined && sinceSeq < latestSnapshotSeq) { + // Start from one before the snapshot so it's included in results + effectiveSinceSeq = latestSnapshotSeq - 1; + Logger.info( + `[user:${userId}] Optimized download: skipping from sinceSeq=${sinceSeq} to ${effectiveSinceSeq} ` + + `(latest snapshot at seq ${latestSnapshotSeq})`, + ); + + // Compute aggregated vector clock from all ops up to and including the snapshot. + // This ensures clients know about all clock entries from skipped ops. + const skippedOps = await tx.operation.findMany({ + where: { + userId, + serverSeq: { lte: latestSnapshotSeq }, + }, + select: { vectorClock: true }, + }); + + snapshotVectorClock = {}; + for (const op of skippedOps) { + const clock = op.vectorClock as unknown as VectorClock; + if (clock && typeof clock === 'object') { + for (const [clientId, value] of Object.entries(clock)) { + if (typeof value === 'number') { + snapshotVectorClock[clientId] = Math.max( + snapshotVectorClock[clientId] ?? 0, + value, + ); + } + } + } + } + + Logger.info( + `[user:${userId}] Computed snapshotVectorClock from ${skippedOps.length} ops: ${JSON.stringify(snapshotVectorClock)}`, + ); + } + + const ops = await tx.operation.findMany({ + where: { + userId, + serverSeq: { gt: effectiveSinceSeq }, + ...(excludeClient ? { clientId: { not: excludeClient } } : {}), + }, + orderBy: { + serverSeq: 'asc', + }, + take: limit, + }); + + const seqRow = await tx.userSyncState.findUnique({ + where: { userId }, + select: { lastSeq: true }, + }); + + // Get min sequence efficiently + const minSeqAgg = await tx.operation.aggregate({ + where: { userId }, + _min: { serverSeq: true }, + }); + + const latestSeq = seqRow?.lastSeq ?? 0; + const minSeq = minSeqAgg._min.serverSeq ?? null; + + // Gap detection logic + let gapDetected = false; + + // Case 1: Client has history but server is empty + if (sinceSeq > 0 && latestSeq === 0) { + gapDetected = true; + Logger.warn( + `[user:${userId}] Gap detected: client at sinceSeq=${sinceSeq} but server is empty (latestSeq=0)`, + ); + } + + // Case 2: Client is ahead of server + if (sinceSeq > latestSeq && latestSeq > 0) { + gapDetected = true; + Logger.warn( + `[user:${userId}] Gap detected: client ahead sinceSeq=${sinceSeq} > latestSeq=${latestSeq}`, + ); + } + + if (sinceSeq > 0 && latestSeq > 0) { + // Case 3: Requested seq is purged + if (minSeq !== null && sinceSeq < minSeq - 1) { + gapDetected = true; + Logger.warn( + `[user:${userId}] Gap detected: sinceSeq=${sinceSeq} but minSeq=${minSeq}`, + ); + } + + // Case 4: Gap in returned operations (use original sinceSeq for gap detection) + if ( + !excludeClient && + ops.length > 0 && + ops[0].serverSeq > effectiveSinceSeq + 1 + ) { + gapDetected = true; + Logger.warn( + `[user:${userId}] Gap detected: expected seq ${effectiveSinceSeq + 1} but got ${ops[0].serverSeq}`, + ); + } + } + + const mappedOps = ops.map((row) => ({ + serverSeq: row.serverSeq, + op: { + id: row.id, + clientId: row.clientId, + actionType: row.actionType, + opType: row.opType as Operation['opType'], + entityType: row.entityType, + entityId: row.entityId ?? undefined, + payload: row.payload, + vectorClock: row.vectorClock as unknown as VectorClock, + schemaVersion: row.schemaVersion, + timestamp: Number(row.clientTimestamp), + parentOpId: row.parentOpId ?? undefined, + isPayloadEncrypted: row.isPayloadEncrypted, + }, + receivedAt: Number(row.receivedAt), + })); + + return { + ops: mappedOps, + latestSeq, + gapDetected, + latestSnapshotSeq, + snapshotVectorClock, + }; + }); + } + + /** + * Get the latest sequence number for a user. + */ + async getLatestSeq(userId: number): Promise { + const row = await prisma.userSyncState.findUnique({ + where: { userId }, + select: { lastSeq: true }, + }); + return row?.lastSeq ?? 0; + } +} diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index 032630977f..dce60934a7 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -6,7 +6,6 @@ import { UploadResult, SyncConfig, DEFAULT_SYNC_CONFIG, - ONLINE_DEVICE_THRESHOLD_MS, compareVectorClocks, VectorClock, SYNC_ERROR_CODES, @@ -25,6 +24,8 @@ import { ALLOWED_ENTITY_TYPES, RateLimitService, RequestDeduplicationService, + DeviceService, + OperationDownloadService, } from './services'; /** @@ -66,6 +67,8 @@ export class SyncService { private validationService: ValidationService; private rateLimitService: RateLimitService; private requestDeduplicationService: RequestDeduplicationService; + private deviceService: DeviceService; + private operationDownloadService: OperationDownloadService; /** * FIX 1.7: In-memory lock to prevent concurrent snapshot generation for the same user. @@ -87,6 +90,8 @@ export class SyncService { this.validationService = new ValidationService(this.config); this.rateLimitService = new RateLimitService(this.config); this.requestDeduplicationService = new RequestDeduplicationService(); + this.deviceService = new DeviceService(); + this.operationDownloadService = new OperationDownloadService(); } // === Conflict Detection === @@ -510,6 +515,7 @@ export class SyncService { } // === Download Operations === + // Delegated to OperationDownloadService async getOpsSince( userId: number, @@ -517,46 +523,14 @@ export class SyncService { excludeClient?: string, limit: number = 500, ): Promise { - const ops = await prisma.operation.findMany({ - where: { - userId, - serverSeq: { gt: sinceSeq }, - ...(excludeClient ? { clientId: { not: excludeClient } } : {}), - }, - orderBy: { - serverSeq: 'asc', - }, - take: limit, - }); - - return ops.map((row) => ({ - serverSeq: row.serverSeq, - op: { - id: row.id, - clientId: row.clientId, - actionType: row.actionType, - opType: row.opType as Operation['opType'], - entityType: row.entityType, - entityId: row.entityId ?? undefined, - payload: row.payload, - vectorClock: row.vectorClock as unknown as VectorClock, - schemaVersion: row.schemaVersion, - timestamp: Number(row.clientTimestamp), - parentOpId: row.parentOpId ?? undefined, - isPayloadEncrypted: row.isPayloadEncrypted, - }, - receivedAt: Number(row.receivedAt), - })); + return this.operationDownloadService.getOpsSince( + userId, + sinceSeq, + excludeClient, + limit, + ); } - /** - * Get operations and latest sequence atomically with gap detection. - * - * OPTIMIZATION: When sinceSeq is before the latest full-state operation (SYNC_IMPORT, - * BACKUP_IMPORT, REPAIR), we skip to that operation's sequence instead. This prevents - * sending operations that will be filtered out by the client anyway, saving bandwidth - * and processing time. - */ async getOpsSinceWithSeq( userId: number, sinceSeq: number, @@ -569,166 +543,16 @@ export class SyncService { latestSnapshotSeq?: number; snapshotVectorClock?: VectorClock; }> { - return prisma.$transaction(async (tx) => { - // Find the latest full-state operation (SYNC_IMPORT, BACKUP_IMPORT, REPAIR) - // These operations supersede all previous operations - const latestFullStateOp = await tx.operation.findFirst({ - where: { - userId, - opType: { in: ['SYNC_IMPORT', 'BACKUP_IMPORT', 'REPAIR'] }, - }, - orderBy: { serverSeq: 'desc' }, - select: { serverSeq: true }, - }); - - const latestSnapshotSeq = latestFullStateOp?.serverSeq ?? undefined; - - // OPTIMIZATION: If client is requesting ops from before the latest full-state op, - // start from the full-state op instead. Pre-import ops are superseded and will - // be filtered out by the client anyway. - let effectiveSinceSeq = sinceSeq; - let snapshotVectorClock: VectorClock | undefined; - - if (latestSnapshotSeq !== undefined && sinceSeq < latestSnapshotSeq) { - // Start from one before the snapshot so it's included in results - effectiveSinceSeq = latestSnapshotSeq - 1; - Logger.info( - `[user:${userId}] Optimized download: skipping from sinceSeq=${sinceSeq} to ${effectiveSinceSeq} ` + - `(latest snapshot at seq ${latestSnapshotSeq})`, - ); - - // Compute aggregated vector clock from all ops up to and including the snapshot. - // This ensures clients know about all clock entries from skipped ops. - const skippedOps = await tx.operation.findMany({ - where: { - userId, - serverSeq: { lte: latestSnapshotSeq }, - }, - select: { vectorClock: true }, - }); - - snapshotVectorClock = {}; - for (const op of skippedOps) { - const clock = op.vectorClock as unknown as VectorClock; - if (clock && typeof clock === 'object') { - for (const [clientId, value] of Object.entries(clock)) { - if (typeof value === 'number') { - snapshotVectorClock[clientId] = Math.max( - snapshotVectorClock[clientId] ?? 0, - value, - ); - } - } - } - } - - Logger.info( - `[user:${userId}] Computed snapshotVectorClock from ${skippedOps.length} ops: ${JSON.stringify(snapshotVectorClock)}`, - ); - } - - const ops = await tx.operation.findMany({ - where: { - userId, - serverSeq: { gt: effectiveSinceSeq }, - ...(excludeClient ? { clientId: { not: excludeClient } } : {}), - }, - orderBy: { - serverSeq: 'asc', - }, - take: limit, - }); - - const seqRow = await tx.userSyncState.findUnique({ - where: { userId }, - select: { lastSeq: true }, - }); - - // Get min sequence efficiently - const minSeqAgg = await tx.operation.aggregate({ - where: { userId }, - _min: { serverSeq: true }, - }); - - const latestSeq = seqRow?.lastSeq ?? 0; - const minSeq = minSeqAgg._min.serverSeq ?? null; - - // Gap detection logic - let gapDetected = false; - - // Case 1: Client has history but server is empty - if (sinceSeq > 0 && latestSeq === 0) { - gapDetected = true; - Logger.warn( - `[user:${userId}] Gap detected: client at sinceSeq=${sinceSeq} but server is empty (latestSeq=0)`, - ); - } - - // Case 2: Client is ahead of server - if (sinceSeq > latestSeq && latestSeq > 0) { - gapDetected = true; - Logger.warn( - `[user:${userId}] Gap detected: client ahead sinceSeq=${sinceSeq} > latestSeq=${latestSeq}`, - ); - } - - if (sinceSeq > 0 && latestSeq > 0) { - // Case 3: Requested seq is purged - if (minSeq !== null && sinceSeq < minSeq - 1) { - gapDetected = true; - Logger.warn( - `[user:${userId}] Gap detected: sinceSeq=${sinceSeq} but minSeq=${minSeq}`, - ); - } - - // Case 4: Gap in returned operations (use original sinceSeq for gap detection) - if ( - !excludeClient && - ops.length > 0 && - ops[0].serverSeq > effectiveSinceSeq + 1 - ) { - gapDetected = true; - Logger.warn( - `[user:${userId}] Gap detected: expected seq ${effectiveSinceSeq + 1} but got ${ops[0].serverSeq}`, - ); - } - } - - const mappedOps = ops.map((row) => ({ - serverSeq: row.serverSeq, - op: { - id: row.id, - clientId: row.clientId, - actionType: row.actionType, - opType: row.opType as Operation['opType'], - entityType: row.entityType, - entityId: row.entityId ?? undefined, - payload: row.payload, - vectorClock: row.vectorClock as unknown as VectorClock, - schemaVersion: row.schemaVersion, - timestamp: Number(row.clientTimestamp), - parentOpId: row.parentOpId ?? undefined, - isPayloadEncrypted: row.isPayloadEncrypted, - }, - receivedAt: Number(row.receivedAt), - })); - - return { - ops: mappedOps, - latestSeq, - gapDetected, - latestSnapshotSeq, - snapshotVectorClock, - }; - }); + return this.operationDownloadService.getOpsSinceWithSeq( + userId, + sinceSeq, + excludeClient, + limit, + ); } async getLatestSeq(userId: number): Promise { - const row = await prisma.userSyncState.findUnique({ - where: { userId }, - select: { lastSeq: true }, - }); - return row?.lastSeq ?? 0; + return this.operationDownloadService.getLatestSeq(userId); } // === Snapshot Management === @@ -1684,29 +1508,15 @@ export class SyncService { } async isDeviceOwner(userId: number, clientId: string): Promise { - const count = await prisma.syncDevice.count({ - where: { userId, clientId }, - }); - return count > 0; + return this.deviceService.isDeviceOwner(userId, clientId); } async getAllUserIds(): Promise { - const users = await prisma.userSyncState.findMany({ - select: { userId: true }, - distinct: ['userId'], - }); - return users.map((u) => u.userId); + return this.deviceService.getAllUserIds(); } async getOnlineDeviceCount(userId: number): Promise { - const threshold = Date.now() - ONLINE_DEVICE_THRESHOLD_MS; - const count = await prisma.syncDevice.count({ - where: { - userId, - lastSeenAt: { gt: BigInt(threshold) }, - }, - }); - return count; + return this.deviceService.getOnlineDeviceCount(userId); } } diff --git a/packages/super-sync-server/tests/device.service.spec.ts b/packages/super-sync-server/tests/device.service.spec.ts new file mode 100644 index 0000000000..d174c498bd --- /dev/null +++ b/packages/super-sync-server/tests/device.service.spec.ts @@ -0,0 +1,139 @@ +import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest'; +import { DeviceService } from '../src/sync/services/device.service'; +import { ONLINE_DEVICE_THRESHOLD_MS } from '../src/sync/sync.types'; + +// Mock prisma +vi.mock('../src/db', () => ({ + prisma: { + syncDevice: { + count: vi.fn(), + }, + userSyncState: { + findMany: vi.fn(), + }, + }, +})); + +import { prisma } from '../src/db'; + +describe('DeviceService', () => { + let service: DeviceService; + + beforeEach(() => { + vi.clearAllMocks(); + service = new DeviceService(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + describe('isDeviceOwner', () => { + it('should return true when device exists for user', async () => { + vi.mocked(prisma.syncDevice.count).mockResolvedValue(1); + + const result = await service.isDeviceOwner(1, 'client-123'); + + expect(result).toBe(true); + expect(prisma.syncDevice.count).toHaveBeenCalledWith({ + where: { userId: 1, clientId: 'client-123' }, + }); + }); + + it('should return false when device does not exist', async () => { + vi.mocked(prisma.syncDevice.count).mockResolvedValue(0); + + const result = await service.isDeviceOwner(1, 'unknown-client'); + + expect(result).toBe(false); + }); + + it('should return true when multiple devices match (edge case)', async () => { + vi.mocked(prisma.syncDevice.count).mockResolvedValue(2); + + const result = await service.isDeviceOwner(1, 'client-123'); + + expect(result).toBe(true); + }); + }); + + describe('getAllUserIds', () => { + it('should return empty array when no users exist', async () => { + vi.mocked(prisma.userSyncState.findMany).mockResolvedValue([]); + + const result = await service.getAllUserIds(); + + expect(result).toEqual([]); + expect(prisma.userSyncState.findMany).toHaveBeenCalledWith({ + select: { userId: true }, + distinct: ['userId'], + }); + }); + + it('should return user IDs from sync state', async () => { + vi.mocked(prisma.userSyncState.findMany).mockResolvedValue([ + { userId: 1 }, + { userId: 2 }, + { userId: 3 }, + ] as any); + + const result = await service.getAllUserIds(); + + expect(result).toEqual([1, 2, 3]); + }); + + it('should handle single user', async () => { + vi.mocked(prisma.userSyncState.findMany).mockResolvedValue([{ userId: 42 }] as any); + + const result = await service.getAllUserIds(); + + expect(result).toEqual([42]); + }); + }); + + describe('getOnlineDeviceCount', () => { + it('should return count of online devices', async () => { + vi.useFakeTimers(); + const now = Date.now(); + vi.setSystemTime(now); + + vi.mocked(prisma.syncDevice.count).mockResolvedValue(3); + + const result = await service.getOnlineDeviceCount(1); + + expect(result).toBe(3); + expect(prisma.syncDevice.count).toHaveBeenCalledWith({ + where: { + userId: 1, + lastSeenAt: { gt: BigInt(now - ONLINE_DEVICE_THRESHOLD_MS) }, + }, + }); + }); + + it('should return zero when no online devices', async () => { + vi.mocked(prisma.syncDevice.count).mockResolvedValue(0); + + const result = await service.getOnlineDeviceCount(1); + + expect(result).toBe(0); + }); + + it('should calculate threshold correctly', async () => { + vi.useFakeTimers(); + const now = 1700000000000; // Fixed timestamp + vi.setSystemTime(now); + + vi.mocked(prisma.syncDevice.count).mockResolvedValue(1); + + await service.getOnlineDeviceCount(1); + + const expectedThreshold = BigInt(now - ONLINE_DEVICE_THRESHOLD_MS); + expect(prisma.syncDevice.count).toHaveBeenCalledWith({ + where: { + userId: 1, + lastSeenAt: { gt: expectedThreshold }, + }, + }); + }); + }); +});