mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-26 09:23:50 +00:00
c087ffee8a
Root cause: Operations with `entityId: undefined` (corrupt ops) caused an infinite rejection loop during sync: 1. Upload ops → Rejected with CONFLICT_CONCURRENT 2. Rejection handler creates merged ops (still with invalid entityId) 3. Upload merged ops → Rejected again 4. Loop continues forever This prevented the sync status double checkmark from ever appearing because hasPendingOps never became false. Three-layer defense implemented: 1. Client: Guard time tracking dispatch (task.service.ts) - Changed `if (currentTask &&` to `if (currentTask?.id &&` - Prevents time tracking dispatch with undefined taskId 2. Client: Strengthen entityId validation (operation-log.effects.ts) - Check that entityId is truthy AND a string type - Log error with actual entityId value for debugging 3. Client: Cleanup corrupt ops on startup (operation-log-recovery.service.ts) - New cleanupCorruptOps() method marks corrupt ops as rejected - Runs during hydration to break existing infinite loops - Excludes bulk 'ALL' operations which legitimately lack entityId 4. Server: Validate entityId for all regular operations (validation.service.ts) - Reject ops with missing entityId (except full-state and bulk ops) - Full-state: SYNC_IMPORT, BACKUP_IMPORT, REPAIR - Bulk entity types: ALL, RECOVERY - Returns MISSING_ENTITY_ID error code (permanent rejection)
573 lines
20 KiB
TypeScript
573 lines
20 KiB
TypeScript
/**
|
|
* Tests for sync system fixes (Issues 1-6)
|
|
*
|
|
* These tests verify the fixes implemented for the identified sync issues:
|
|
* - Issue 1: Request deduplication with piggybacking
|
|
* - Issue 3: Encrypted snapshot uploads
|
|
* - Issue 4: Encrypted ops blocking restore endpoint
|
|
* - Issue 5: Dedup before quota check
|
|
* - Issue 6: Vector-clock EQUAL from different client
|
|
*/
|
|
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
|
|
import Fastify, { FastifyInstance } from 'fastify';
|
|
import { uuidv7 } from 'uuidv7';
|
|
|
|
// Store for test data
|
|
let testOperations: Map<string, any>;
|
|
let serverSeqCounter: number;
|
|
let requestCache: Map<string, any>;
|
|
|
|
// Mock the database module with Prisma mocks
|
|
vi.mock('../src/db', () => {
|
|
return {
|
|
prisma: {
|
|
$transaction: vi.fn().mockImplementation(async (callback: any) => {
|
|
const tx = {
|
|
operation: {
|
|
create: vi.fn().mockImplementation(async (args: any) => {
|
|
serverSeqCounter++;
|
|
const op = {
|
|
...args.data,
|
|
serverSeq: serverSeqCounter,
|
|
receivedAt: BigInt(Date.now()),
|
|
};
|
|
testOperations.set(args.data.id, op);
|
|
return op;
|
|
}),
|
|
findFirst: vi.fn().mockImplementation(async (args: any) => {
|
|
// Find by ID
|
|
if (args.where?.id) {
|
|
return testOperations.get(args.where.id) || null;
|
|
}
|
|
// Find full-state operation
|
|
if (args.where?.opType?.in) {
|
|
for (const op of Array.from(testOperations.values()).reverse()) {
|
|
if (
|
|
args.where.opType.in.includes(op.opType) &&
|
|
args.where.userId === op.userId
|
|
) {
|
|
return op;
|
|
}
|
|
}
|
|
}
|
|
return null;
|
|
}),
|
|
findMany: vi.fn().mockImplementation(async (args: any) => {
|
|
const ops = Array.from(testOperations.values());
|
|
return ops
|
|
.filter((op) => {
|
|
if (args.where?.userId !== undefined && args.where.userId !== op.userId)
|
|
return false;
|
|
if (
|
|
args.where?.serverSeq?.gt !== undefined &&
|
|
op.serverSeq <= args.where.serverSeq.gt
|
|
)
|
|
return false;
|
|
if (
|
|
args.where?.clientId?.not &&
|
|
op.clientId === args.where.clientId.not
|
|
)
|
|
return false;
|
|
return true;
|
|
})
|
|
.sort((a, b) => a.serverSeq - b.serverSeq)
|
|
.slice(0, args.take || 500);
|
|
}),
|
|
aggregate: vi.fn().mockResolvedValue({ _min: { serverSeq: 1 } }),
|
|
findUnique: vi.fn().mockImplementation(async (args: any) => {
|
|
if (args.where?.id) {
|
|
return testOperations.get(args.where.id) || null;
|
|
}
|
|
return null;
|
|
}),
|
|
},
|
|
userSyncState: {
|
|
findUnique: vi.fn().mockResolvedValue({ lastSeq: serverSeqCounter }),
|
|
upsert: vi.fn().mockResolvedValue({}),
|
|
update: vi.fn().mockResolvedValue({}),
|
|
},
|
|
syncDevice: {
|
|
upsert: vi.fn().mockResolvedValue({}),
|
|
count: vi.fn().mockResolvedValue(1),
|
|
},
|
|
};
|
|
return callback(tx);
|
|
}),
|
|
operation: {
|
|
findFirst: vi.fn(),
|
|
findMany: vi.fn().mockImplementation(async (args: any) => {
|
|
const ops = Array.from(testOperations.values());
|
|
return ops
|
|
.filter((op) => {
|
|
if (args.where?.userId !== undefined && args.where.userId !== op.userId)
|
|
return false;
|
|
if (
|
|
args.where?.serverSeq?.gt !== undefined &&
|
|
op.serverSeq <= args.where.serverSeq.gt
|
|
)
|
|
return false;
|
|
if (args.where?.clientId?.not && op.clientId === args.where.clientId.not)
|
|
return false;
|
|
return true;
|
|
})
|
|
.sort((a, b) => a.serverSeq - b.serverSeq)
|
|
.slice(0, args.take || 500);
|
|
}),
|
|
aggregate: vi.fn().mockResolvedValue({ _min: { serverSeq: 1 } }),
|
|
findUnique: vi.fn().mockImplementation(async (args: any) => {
|
|
if (args.where?.id) {
|
|
return testOperations.get(args.where.id) || null;
|
|
}
|
|
return null;
|
|
}),
|
|
},
|
|
userSyncState: {
|
|
findUnique: vi.fn().mockImplementation(async () => ({
|
|
lastSeq: serverSeqCounter,
|
|
})),
|
|
upsert: vi.fn().mockResolvedValue({}),
|
|
update: vi.fn().mockResolvedValue({}),
|
|
findMany: vi.fn().mockResolvedValue([]),
|
|
},
|
|
syncDevice: {
|
|
upsert: vi.fn().mockResolvedValue({}),
|
|
count: vi.fn().mockResolvedValue(1),
|
|
deleteMany: vi.fn().mockResolvedValue({ count: 0 }),
|
|
},
|
|
user: {
|
|
findUnique: vi.fn().mockResolvedValue({
|
|
id: 1,
|
|
storageQuotaBytes: BigInt(100 * 1024 * 1024),
|
|
storageUsedBytes: BigInt(0),
|
|
}),
|
|
update: vi.fn().mockResolvedValue({}),
|
|
},
|
|
$queryRaw: vi.fn().mockResolvedValue([{ total: BigInt(0) }]),
|
|
},
|
|
};
|
|
});
|
|
|
|
// Mock auth
|
|
vi.mock('../src/auth', () => ({
|
|
verifyToken: vi.fn().mockResolvedValue({ userId: 1, email: 'test@test.com' }),
|
|
}));
|
|
|
|
// Import after mocking
|
|
import { syncRoutes } from '../src/sync/sync.routes';
|
|
import { initSyncService, getSyncService } from '../src/sync/sync.service';
|
|
|
|
// Helper to create operation
|
|
const createOp = (
|
|
clientId: string,
|
|
overrides: Partial<{
|
|
id: string;
|
|
actionType: string;
|
|
opType: string;
|
|
entityType: string;
|
|
entityId: string;
|
|
payload: unknown;
|
|
vectorClock: Record<string, number>;
|
|
timestamp: number;
|
|
schemaVersion: number;
|
|
}> = {},
|
|
) => ({
|
|
id: uuidv7(),
|
|
clientId,
|
|
actionType: 'ADD_TASK',
|
|
opType: 'CRT',
|
|
entityType: 'TASK',
|
|
entityId: 'task-1',
|
|
payload: { title: 'Test Task' },
|
|
vectorClock: {},
|
|
timestamp: Date.now(),
|
|
schemaVersion: 1,
|
|
...overrides,
|
|
});
|
|
|
|
describe('Sync System Fixes', () => {
|
|
let app: FastifyInstance;
|
|
const authToken = 'mock-token';
|
|
|
|
beforeEach(async () => {
|
|
// Reset test state
|
|
testOperations = new Map();
|
|
serverSeqCounter = 0;
|
|
requestCache = new Map();
|
|
vi.clearAllMocks();
|
|
|
|
// Initialize service
|
|
initSyncService();
|
|
|
|
// Create Fastify app
|
|
app = Fastify();
|
|
await app.register(syncRoutes, { prefix: '/api/sync' });
|
|
await app.ready();
|
|
});
|
|
|
|
afterEach(async () => {
|
|
await app.close();
|
|
});
|
|
|
|
// =============================================================================
|
|
// Issue 1: Request deduplication with piggybacking
|
|
// =============================================================================
|
|
describe('Issue 1: Request deduplication with piggybacking', () => {
|
|
it('should return fresh newOps on deduplicated request', async () => {
|
|
const clientA = 'client-a';
|
|
const clientB = 'client-b';
|
|
const requestId = 'req-123';
|
|
|
|
// Step 1: Client A uploads with requestId
|
|
const firstResponse = await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [createOp(clientA, { entityId: 'task-a' })],
|
|
clientId: clientA,
|
|
lastKnownServerSeq: 0,
|
|
requestId,
|
|
},
|
|
});
|
|
expect(firstResponse.statusCode).toBe(200);
|
|
const firstBody = firstResponse.json();
|
|
expect(firstBody.results[0].accepted).toBe(true);
|
|
const clientASeq = firstBody.latestSeq;
|
|
|
|
// Step 2: Client B uploads ops
|
|
const clientBResponse = await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [createOp(clientB, { entityId: 'task-b' })],
|
|
clientId: clientB,
|
|
},
|
|
});
|
|
expect(clientBResponse.statusCode).toBe(200);
|
|
|
|
// Step 3: Client A retries with same requestId but updated lastKnownServerSeq
|
|
const retryResponse = await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [createOp(clientA, { entityId: 'task-a' })],
|
|
clientId: clientA,
|
|
lastKnownServerSeq: clientASeq,
|
|
requestId,
|
|
},
|
|
});
|
|
|
|
expect(retryResponse.statusCode).toBe(200);
|
|
const retryBody = retryResponse.json();
|
|
|
|
// Should be deduplicated
|
|
expect(retryBody.deduplicated).toBe(true);
|
|
|
|
// CRITICAL: Should have piggybacked ops from client B
|
|
expect(retryBody.newOps).toBeDefined();
|
|
expect(retryBody.newOps.length).toBeGreaterThan(0);
|
|
expect(
|
|
retryBody.newOps.some(
|
|
(op: { op: { clientId: string } }) => op.op.clientId === clientB,
|
|
),
|
|
).toBe(true);
|
|
});
|
|
|
|
it('should use retry request lastKnownServerSeq, not original', async () => {
|
|
const clientA = 'client-a';
|
|
const clientB = 'client-b';
|
|
const requestId = 'req-456';
|
|
|
|
// Client A uploads
|
|
await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [createOp(clientA, { entityId: 'task-1' })],
|
|
clientId: clientA,
|
|
lastKnownServerSeq: 0,
|
|
requestId,
|
|
},
|
|
});
|
|
|
|
// Client B uploads
|
|
await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [createOp(clientB, { entityId: 'task-2' })],
|
|
clientId: clientB,
|
|
},
|
|
});
|
|
|
|
// Client A retries with lastKnownServerSeq=0
|
|
const retryResponse = await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [createOp(clientA, { entityId: 'task-1' })],
|
|
clientId: clientA,
|
|
lastKnownServerSeq: 0,
|
|
requestId,
|
|
},
|
|
});
|
|
|
|
const body = retryResponse.json();
|
|
expect(body.deduplicated).toBe(true);
|
|
// Should get client B's ops since lastKnownServerSeq=0
|
|
expect(body.newOps?.length).toBeGreaterThan(0);
|
|
});
|
|
});
|
|
|
|
// =============================================================================
|
|
// Issue 3: Encrypted snapshot uploads
|
|
// =============================================================================
|
|
describe('Issue 3: Encrypted snapshot uploads', () => {
|
|
it('should store isPayloadEncrypted flag from snapshot upload', async () => {
|
|
const snapshotResponse = await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/snapshot',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
state: 'encrypted-string-here',
|
|
clientId: 'test-client',
|
|
reason: 'recovery',
|
|
vectorClock: { 'test-client': 1 },
|
|
schemaVersion: 1,
|
|
isPayloadEncrypted: true,
|
|
},
|
|
});
|
|
|
|
expect(snapshotResponse.statusCode).toBe(200);
|
|
const snapshotBody = snapshotResponse.json();
|
|
expect(snapshotBody.accepted).toBe(true);
|
|
|
|
// Download ops and verify the SYNC_IMPORT has isPayloadEncrypted
|
|
const downloadResponse = await app.inject({
|
|
method: 'GET',
|
|
url: '/api/sync/ops?sinceSeq=0',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
});
|
|
|
|
const downloadBody = downloadResponse.json();
|
|
const syncImportOp = downloadBody.ops.find(
|
|
(op: { op: { opType: string } }) => op.op.opType === 'SYNC_IMPORT',
|
|
);
|
|
expect(syncImportOp).toBeDefined();
|
|
expect(syncImportOp.op.isPayloadEncrypted).toBe(true);
|
|
});
|
|
|
|
it('should default isPayloadEncrypted to false when not provided', async () => {
|
|
const snapshotResponse = await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/snapshot',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
state: { tasks: [] },
|
|
clientId: 'test-client',
|
|
reason: 'initial',
|
|
vectorClock: { 'test-client': 1 },
|
|
schemaVersion: 1,
|
|
},
|
|
});
|
|
|
|
expect(snapshotResponse.statusCode).toBe(200);
|
|
|
|
// Download and verify default
|
|
const downloadResponse = await app.inject({
|
|
method: 'GET',
|
|
url: '/api/sync/ops?sinceSeq=0',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
});
|
|
|
|
const downloadBody = downloadResponse.json();
|
|
const syncImportOp = downloadBody.ops.find(
|
|
(op: { op: { opType: string } }) => op.op.opType === 'SYNC_IMPORT',
|
|
);
|
|
expect(syncImportOp.op.isPayloadEncrypted).toBe(false);
|
|
});
|
|
});
|
|
|
|
// =============================================================================
|
|
// Issue 4: Encrypted ops blocking restore
|
|
// NOTE: These tests require integration tests against real database
|
|
// The mock layer doesn't fully support the restore endpoint behavior.
|
|
// See tests/integration/ for full restore tests.
|
|
// =============================================================================
|
|
describe('Issue 4: Encrypted ops blocking restore', () => {
|
|
it.skip('should reject restore when encrypted ops exist in range (requires integration test)', async () => {
|
|
// This test requires the real database to test the encrypted ops check
|
|
// in generateSnapshotAtSeq. The mock doesn't implement operation.count.
|
|
// See integration tests for full coverage.
|
|
});
|
|
|
|
it.skip('should allow restore when no encrypted ops exist (requires integration test)', async () => {
|
|
// This test requires the real database to test restore functionality.
|
|
// See integration tests for full coverage.
|
|
});
|
|
});
|
|
|
|
// =============================================================================
|
|
// Issue 5: Dedup before quota check
|
|
// Tests that deduplication happens before quota check to allow retries
|
|
// =============================================================================
|
|
describe('Issue 5: Dedup before quota check', () => {
|
|
it('should return cached results even when quota would be exceeded on retry', async () => {
|
|
const clientId = uuidv7();
|
|
const requestId = uuidv7();
|
|
|
|
// First request with requestId
|
|
const firstResponse = await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [createOp(clientId, { entityId: 'task-1' })],
|
|
clientId,
|
|
lastKnownServerSeq: 0,
|
|
requestId,
|
|
},
|
|
});
|
|
|
|
expect(firstResponse.statusCode).toBe(200);
|
|
const firstBody = firstResponse.json();
|
|
expect(firstBody.results[0].accepted).toBe(true);
|
|
|
|
// Second request with same requestId (retry) should return cached results
|
|
// even if the real quota check would fail (because dedup runs first)
|
|
const retryResponse = await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [createOp(clientId, { entityId: 'task-1' })],
|
|
clientId,
|
|
lastKnownServerSeq: 0,
|
|
requestId,
|
|
},
|
|
});
|
|
|
|
expect(retryResponse.statusCode).toBe(200);
|
|
const retryBody = retryResponse.json();
|
|
expect(retryBody.deduplicated).toBe(true);
|
|
expect(retryBody.results[0].accepted).toBe(true);
|
|
});
|
|
});
|
|
|
|
// =============================================================================
|
|
// Issue 7: serverTime in download response
|
|
// =============================================================================
|
|
describe('Issue 7: serverTime in download response', () => {
|
|
it('should include serverTime in download response for clock drift detection', async () => {
|
|
// First upload an operation
|
|
await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [createOp('test-client', { entityId: 'task-1' })],
|
|
clientId: 'test-client',
|
|
},
|
|
});
|
|
|
|
const beforeTime = Date.now();
|
|
const downloadResponse = await app.inject({
|
|
method: 'GET',
|
|
url: '/api/sync/ops?sinceSeq=0',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
});
|
|
const afterTime = Date.now();
|
|
|
|
expect(downloadResponse.statusCode).toBe(200);
|
|
const body = downloadResponse.json();
|
|
|
|
// serverTime should be present and within expected range
|
|
expect(body.serverTime).toBeDefined();
|
|
expect(typeof body.serverTime).toBe('number');
|
|
expect(body.serverTime).toBeGreaterThanOrEqual(beforeTime);
|
|
expect(body.serverTime).toBeLessThanOrEqual(afterTime);
|
|
});
|
|
});
|
|
|
|
// =============================================================================
|
|
// Issue 8: Cleanup should return affected userIds for storage update
|
|
// =============================================================================
|
|
describe('Issue 8: deleteOldSyncedOpsForAllUsers returns affectedUserIds', () => {
|
|
it('should return totalDeleted and affectedUserIds structure', async () => {
|
|
const syncService = getSyncService();
|
|
const cutoffTime = Date.now() - 50 * 24 * 60 * 60 * 1000; // 50 days ago
|
|
|
|
// Call the method - with default mock it returns empty since no users have snapshots
|
|
const result = await syncService.deleteOldSyncedOpsForAllUsers(cutoffTime);
|
|
|
|
// Verify return structure
|
|
expect(result).toHaveProperty('totalDeleted');
|
|
expect(result).toHaveProperty('affectedUserIds');
|
|
expect(typeof result.totalDeleted).toBe('number');
|
|
expect(Array.isArray(result.affectedUserIds)).toBe(true);
|
|
});
|
|
|
|
it('should return empty affectedUserIds when no operations deleted', async () => {
|
|
const syncService = getSyncService();
|
|
const cutoffTime = Date.now() - 50 * 24 * 60 * 60 * 1000;
|
|
|
|
const result = await syncService.deleteOldSyncedOpsForAllUsers(cutoffTime);
|
|
|
|
expect(result.totalDeleted).toBe(0);
|
|
expect(result.affectedUserIds).toHaveLength(0);
|
|
});
|
|
});
|
|
|
|
// =============================================================================
|
|
// Issue 6: Vector-clock EQUAL from different client
|
|
// NOTE: Conflict detection requires the mock to properly implement
|
|
// operation.findFirst to return existing operations. The current mock
|
|
// doesn't support this, so these tests are skipped in favor of
|
|
// integration tests.
|
|
// =============================================================================
|
|
describe('Issue 6: Vector-clock EQUAL from different client', () => {
|
|
it.skip('should reject EQUAL clocks from different clients as conflict (requires integration test)', async () => {
|
|
// This test requires the mock's operation.findFirst to return
|
|
// the existing operation for conflict detection.
|
|
// See integration tests for full coverage.
|
|
});
|
|
|
|
it('should allow operations when no prior operation exists', async () => {
|
|
const clientA = uuidv7();
|
|
const entityId = 'task-' + uuidv7();
|
|
|
|
// Client A creates entity - should succeed since no prior op exists
|
|
const response1 = await app.inject({
|
|
method: 'POST',
|
|
url: '/api/sync/ops',
|
|
headers: { authorization: `Bearer ${authToken}` },
|
|
payload: {
|
|
ops: [
|
|
{
|
|
id: uuidv7(),
|
|
clientId: clientA,
|
|
actionType: 'AddTask',
|
|
opType: 'CRT',
|
|
entityType: 'TASK',
|
|
entityId,
|
|
payload: { title: 'Task' },
|
|
vectorClock: { [clientA]: 1 },
|
|
timestamp: Date.now(),
|
|
schemaVersion: 1,
|
|
},
|
|
],
|
|
clientId: clientA,
|
|
lastKnownServerSeq: 0,
|
|
},
|
|
});
|
|
|
|
expect(response1.statusCode).toBe(200);
|
|
expect(response1.json().results[0].accepted).toBe(true);
|
|
});
|
|
});
|
|
});
|