Mirrors/super-productivity
1
0
Fork 0
mirror of https://github.com/johannesjo/super-productivity.git synced 2026-08-04 05:22:39 +00:00
Code Issues Projects Releases Wiki Activity

refactor(sync): extract DeviceService and OperationDownloadService

Browse source 
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.
This commit is contained in:
Johannes Millan 2025-12-27 20:43:25 +01:00
parent 2a9e3154bc
commit 07589dd67f
5 changed files with 449 additions and 213 deletions
Download patch file Download diff file Expand all files Collapse all files

47
packages/super-sync-server/src/sync/services/device.service.ts Normal file
View file

@ -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<boolean> {
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<number[]> {
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<number> {
const threshold = Date.now() - ONLINE_DEVICE_THRESHOLD_MS;
const count = await prisma.syncDevice.count({
where: {
userId,
lastSeenAt: { gt: BigInt(threshold) },
},
});
return count;
}
}

2
packages/super-sync-server/src/sync/services/index.ts
View file

@ -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';

238
packages/super-sync-server/src/sync/services/operation-download.service.ts Normal file
View file

@ -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<ServerOperation[]> {
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<number> {
const row = await prisma.userSyncState.findUnique({
where: { userId },
select: { lastSeq: true },
});
return row?.lastSeq ?? 0;
}
}

236
packages/super-sync-server/src/sync/sync.service.ts
View file

@ -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<ServerOperation[]> {
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<number> {
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<boolean> {
const count = await prisma.syncDevice.count({
where: { userId, clientId },
});
return count > 0;
return this.deviceService.isDeviceOwner(userId, clientId);
}
async getAllUserIds(): Promise<number[]> {
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<number> {
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);
}
}

139
packages/super-sync-server/tests/device.service.spec.ts Normal file
View file

@ -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 },
},
});
});
});
});