diff --git a/apps/sync-server/src/routes/sync.test.ts b/apps/sync-server/src/routes/sync.test.ts index 937772464..cf32fb215 100644 --- a/apps/sync-server/src/routes/sync.test.ts +++ b/apps/sync-server/src/routes/sync.test.ts @@ -24,7 +24,20 @@ vi.mock('../services/sync', () => ({ hasMore: false, nextCursor: 0 }), - processPushItem: vi.fn().mockResolvedValue({ accepted: true, serverCursor: 1 }), + processRecordPushBatch: vi.fn().mockResolvedValue({ + accepted: ['550e8400-e29b-41d4-a716-446655440000'], + rejected: [], + serverTime: 1000, + maxCursor: 1, + outcomes: [ + { + id: '550e8400-e29b-41d4-a716-446655440000', + type: 'note', + accepted: true, + serverCursor: 1 + } + ] + }), pullItems: vi.fn().mockResolvedValue([]), getItem: vi.fn().mockResolvedValue({ itemId: '550e8400-e29b-41d4-a716-446655440000', @@ -36,6 +49,15 @@ vi.mock('../services/sync', () => ({ updateDeviceCursor: vi.fn().mockResolvedValue(undefined) })) +vi.mock('../services/crdt', () => ({ + storeUpdates: vi.fn().mockResolvedValue([1]), + getUpdates: vi.fn().mockResolvedValue({ updates: [], hasMore: false }), + getBatchUpdates: vi.fn().mockResolvedValue({}), + storeSnapshot: vi.fn().mockResolvedValue({ sequenceNum: 0 }), + getSnapshot: vi.fn().mockResolvedValue(null), + pruneUpdatesBeforeSnapshot: vi.fn().mockResolvedValue(0) +})) + vi.mock('../services/device', () => ({ updateDevice: vi.fn().mockResolvedValue(undefined) })) @@ -65,11 +87,12 @@ import { getSyncStatus, getManifest, getChanges, - processPushItem, + processRecordPushBatch, pullItems, getItem, updateDeviceCursor } from '../services/sync' +import { storeUpdates } from '../services/crdt' import { authMiddleware } from '../middleware/auth' import { updateDevice } from '../services/device' @@ -133,6 +156,24 @@ const makePushItem = (overrides: Record = {}) => ({ ...overrides }) +const makePushBatchResult = ( + overrides: Partial>> = {} +) => ({ + accepted: [VALID_UUID], + rejected: [], + serverTime: 1000, + maxCursor: 1, + outcomes: [ + { + id: VALID_UUID, + type: 'note' as const, + accepted: true, + serverCursor: 1 + } + ], + ...overrides +}) + // ============================================================================ // Tests // ============================================================================ @@ -354,10 +395,21 @@ describe('sync routes', () => { it('should collect rejected items with reasons', async () => { // #given - vi.mocked(processPushItem).mockResolvedValueOnce({ - accepted: false, - reason: 'VERSION_CONFLICT' - }) + vi.mocked(processRecordPushBatch).mockResolvedValueOnce( + makePushBatchResult({ + accepted: [], + rejected: [{ id: VALID_UUID, reason: 'VERSION_CONFLICT' }], + maxCursor: 0, + outcomes: [ + { + id: VALID_UUID, + type: 'note', + accepted: false, + reason: 'VERSION_CONFLICT' + } + ] + }) + ) const body = { items: [makePushItem()] } // #when @@ -448,10 +500,21 @@ describe('sync routes', () => { it('should not update device last_sync_at when all items rejected', async () => { // #given - vi.mocked(processPushItem).mockResolvedValueOnce({ - accepted: false, - reason: 'VERSION_CONFLICT' - }) + vi.mocked(processRecordPushBatch).mockResolvedValueOnce( + makePushBatchResult({ + accepted: [], + rejected: [{ id: VALID_UUID, reason: 'VERSION_CONFLICT' }], + maxCursor: 0, + outcomes: [ + { + id: VALID_UUID, + type: 'note', + accepted: false, + reason: 'VERSION_CONFLICT' + } + ] + }) + ) const body = { items: [makePushItem()] } // #when @@ -468,6 +531,19 @@ describe('sync routes', () => { it('should accept non-UUID item IDs when payload otherwise validates', async () => { // #given + vi.mocked(processRecordPushBatch).mockResolvedValueOnce( + makePushBatchResult({ + accepted: ['not-a-uuid'], + outcomes: [ + { + id: 'not-a-uuid', + type: 'note', + accepted: true, + serverCursor: 1 + } + ] + }) + ) const body = { items: [makePushItem({ id: 'not-a-uuid' })] } // #when @@ -488,6 +564,20 @@ describe('sync routes', () => { }) ) }) + + it('should pass the parsed record batch to processRecordPushBatch', async () => { + const body = { items: [makePushItem()] } + + await app.request('http://localhost/sync/push', jsonPost('/sync/push', body), env, executionCtx) + + expect(processRecordPushBatch).toHaveBeenCalledWith( + env.DB, + env.STORAGE, + 'user-1', + 'device-1', + [makePushItem()] + ) + }) }) // ========================================================================== @@ -608,4 +698,70 @@ describe('sync routes', () => { expect(json.error.code).toBe(ErrorCodes.VALIDATION_ERROR) }) }) + + describe('record transport aliases', () => { + it('supports /sync/records/push while keeping legacy /sync/push intact', async () => { + const body = { items: [makePushItem()] } + + const res = await app.request( + 'http://localhost/sync/records/push', + jsonPost('/sync/records/push', body), + env, + executionCtx + ) + + expect(res.status).toBe(200) + expect(processRecordPushBatch).toHaveBeenCalledWith( + env.DB, + env.STORAGE, + 'user-1', + 'device-1', + [makePushItem()] + ) + }) + + it('supports /sync/records/pull', async () => { + const body = { itemIds: [VALID_UUID] } + + const res = await app.request( + 'http://localhost/sync/records/pull', + jsonPost('/sync/records/pull', body), + env, + executionCtx + ) + + expect(res.status).toBe(200) + expect(pullItems).toHaveBeenCalledWith(env.DB, env.STORAGE, 'user-1', [VALID_UUID]) + }) + }) + + describe('CRDT transport separation', () => { + it('routes /sync/crdt/updates through CRDT services instead of record push', async () => { + const payload = { noteId: 'note_1', updates: [btoa('hello')] } + + const res = await app.request( + 'http://localhost/sync/crdt/updates', + jsonPost('/sync/crdt/updates', payload), + env, + executionCtx + ) + + expect(res.status).toBe(200) + expect(storeUpdates).toHaveBeenCalledTimes(1) + expect(processRecordPushBatch).not.toHaveBeenCalled() + }) + + it('returns 400 for invalid CRDT update payloads', async () => { + const res = await app.request( + 'http://localhost/sync/crdt/updates', + jsonPost('/sync/crdt/updates', { noteId: 'bad note id!', updates: [] }), + env, + executionCtx + ) + + expect(res.status).toBe(400) + const json = (await res.json()) as { error: { code: string } } + expect(json.error.code).toBe(ErrorCodes.VALIDATION_ERROR) + }) + }) }) diff --git a/apps/sync-server/src/routes/sync.ts b/apps/sync-server/src/routes/sync.ts index 2434ffaef..b59a98cce 100644 --- a/apps/sync-server/src/routes/sync.ts +++ b/apps/sync-server/src/routes/sync.ts @@ -1,7 +1,9 @@ +import type { Context } from 'hono' import { Hono } from 'hono' import { z } from 'zod' import { PullRequestSchema, PushRequestSchema } from '@memry/contracts/sync-api' +import { safeBase64Decode } from '../lib/encoding' import { AppError, ErrorCodes } from '../lib/errors' import { authMiddleware } from '../middleware/auth' import { createRateLimiter } from '../middleware/rate-limit' @@ -10,10 +12,16 @@ import { getItem, getManifest, getSyncStatus, - processPushItem, + processRecordPushBatch, pullItems, updateDeviceCursor } from '../services/sync' +import { + logCrdtTraffic, + logRecordPushBatch, + logRecordQueryBatch, + logSyncValidationFailure +} from '../services/sync-telemetry' import { updateDevice } from '../services/device' import { checkQuota } from '../services/quota' import { getStorageBreakdown } from '../services/storage' @@ -31,6 +39,73 @@ export const sync = new Hono() sync.use('*', authMiddleware) +const MAX_UPDATE_BYTES = 5 * 1024 * 1024 // 5MB per individual update +const BASE64_CHUNK_SIZE = 8192 + +const getRequestPath = (c: Context): string => new URL(c.req.url).pathname + +function parseTransportRequest( + schema: z.ZodType, + body: unknown, + params: { + transport: 'record' | 'crdt' + endpoint: string + label: string + } +): T { + const parsed = schema.safeParse(body) + if (!parsed.success) { + const issue = parsed.error.issues[0]?.message ?? 'validation failed' + logSyncValidationFailure({ + transport: params.transport, + endpoint: params.endpoint, + issue + }) + throw new AppError(ErrorCodes.VALIDATION_ERROR, `Invalid ${params.label}: ${issue}`, 400) + } + + return parsed.data +} + +function logQueryValidationFailure( + transport: 'record' | 'crdt', + endpoint: string, + issue: string, + code: keyof typeof ErrorCodes = 'VALIDATION_ERROR' +): never { + logSyncValidationFailure({ transport, endpoint, issue }) + throw new AppError(ErrorCodes[code], issue, 400) +} + +function safeBase64Encode(buffer: ArrayBuffer): string { + const bytes = new Uint8Array(buffer) + let result = '' + for (let i = 0; i < bytes.length; i += BASE64_CHUNK_SIZE) { + const chunk = bytes.subarray(i, i + BASE64_CHUNK_SIZE) + result += String.fromCharCode(...chunk) + } + return btoa(result) +} + +function decodeCrdtPayload(base64: string, endpoint: string, tooLargeMessage: string): ArrayBuffer { + try { + const bytes = safeBase64Decode(base64) + if (bytes.byteLength > MAX_UPDATE_BYTES) { + throw new AppError(ErrorCodes.VALIDATION_ERROR, tooLargeMessage, 413) + } + return bytes.slice().buffer as ArrayBuffer + } catch (error) { + if (error instanceof AppError) { + logSyncValidationFailure({ + transport: 'crdt', + endpoint, + issue: error.message + }) + } + throw error + } +} + const pushRateLimit = createRateLimiter({ keyPrefix: 'sync_push', maxRequests: 60, @@ -93,34 +168,36 @@ sync.get('/storage', storageRateLimit, async (c) => { return c.json(breakdown) }) -sync.get('/status', statusRateLimit, async (c) => { +const handleRecordStatus = async (c: Context): Promise => { const userId = c.get('userId')! const deviceId = c.get('deviceId')! const status = await getSyncStatus(c.env.DB, userId, deviceId) return c.json(status) -}) +} -sync.get('/manifest', manifestRateLimit, async (c) => { +const handleRecordManifest = async (c: Context): Promise => { const userId = c.get('userId')! const manifest = await getManifest(c.env.DB, userId) return c.json(manifest) -}) +} -sync.get('/changes', changesRateLimit, async (c) => { +const handleRecordChanges = async (c: Context): Promise => { const userId = c.get('userId')! const deviceId = c.get('deviceId')! + const endpoint = getRequestPath(c) + const startedAt = Date.now() const cursorParam = c.req.query('cursor') const limitParam = c.req.query('limit') const cursor = cursorParam ? parseInt(cursorParam, 10) : 0 if (isNaN(cursor) || cursor < 0) { - throw new AppError(ErrorCodes.SYNC_INVALID_CURSOR, 'Invalid cursor value', 400) + logQueryValidationFailure('record', endpoint, 'Invalid cursor value', 'SYNC_INVALID_CURSOR') } const limit = limitParam ? parseInt(limitParam, 10) : undefined if (limit !== undefined && (isNaN(limit) || limit < 1)) { - throw new AppError(ErrorCodes.VALIDATION_ERROR, 'Invalid limit value', 400) + logQueryValidationFailure('record', endpoint, 'Invalid limit value') } const changes = await getChanges(c.env.DB, userId, cursor, limit) @@ -132,47 +209,54 @@ sync.get('/changes', changesRateLimit, async (c) => { }) } + logRecordQueryBatch({ + endpoint, + operation: 'changes', + latencyMs: Date.now() - startedAt, + itemTypes: changes.items.map((item) => item.type), + deletedCount: changes.deleted.length + }) + return c.json(changes) -}) +} -sync.post('/push', pushRateLimit, async (c) => { +const handleRecordPush = async (c: Context): Promise => { const userId = c.get('userId')! const deviceId = c.get('deviceId')! + const endpoint = getRequestPath(c) + const startedAt = Date.now() const body: unknown = await c.req.json() - const parsed = PushRequestSchema.safeParse(body) - if (!parsed.success) { - throw new AppError( - ErrorCodes.VALIDATION_ERROR, - `Invalid push request: ${parsed.error.issues[0]?.message ?? 'validation failed'}`, - 400 - ) - } + const parsed = parseTransportRequest(PushRequestSchema, body, { + transport: 'record', + endpoint, + label: 'push request' + }) - const estimatedBytes = JSON.stringify(parsed.data.items).length - await checkQuota(c.env.DB, userId, estimatedBytes) - - const accepted: string[] = [] - const rejected: Array<{ id: string; reason: string }> = [] - let maxCursor = 0 - - for (const item of parsed.data.items) { - const result = await processPushItem(c.env.DB, c.env.STORAGE, userId, deviceId, item) - if (result.accepted) { - accepted.push(item.id) - if (result.serverCursor && result.serverCursor > maxCursor) { - maxCursor = result.serverCursor - } - } else { - rejected.push({ id: item.id, reason: result.reason ?? 'UNKNOWN' }) + let result + try { + result = await processRecordPushBatch(c.env.DB, c.env.STORAGE, userId, deviceId, parsed.items) + } catch (error) { + if (error instanceof AppError && error.code === ErrorCodes.STORAGE_QUOTA_EXCEEDED) { + logRecordPushBatch({ + endpoint, + latencyMs: Date.now() - startedAt, + outcomes: parsed.items.map((item) => ({ + id: item.id, + type: item.type, + accepted: false, + reason: error.code + })) + }) } + throw error } - if (maxCursor > 0) { - await updateDeviceCursor(c.env.DB, deviceId, userId, maxCursor) + if (result.maxCursor > 0) { + await updateDeviceCursor(c.env.DB, deviceId, userId, result.maxCursor) } - if (accepted.length > 0) { + if (result.accepted.length > 0) { await updateDevice(c.env.DB, deviceId, userId, { last_sync_at: Math.floor(Date.now() / 1000) }) @@ -183,38 +267,50 @@ sync.post('/push', pushRateLimit, async (c) => { new Request(new URL('/broadcast', c.req.url), { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ excludeDeviceId: deviceId, cursor: maxCursor }) + body: JSON.stringify({ excludeDeviceId: deviceId, cursor: result.maxCursor }) }) ) ) } + logRecordPushBatch({ + endpoint, + latencyMs: Date.now() - startedAt, + outcomes: result.outcomes + }) + return c.json({ - accepted, - rejected, - serverTime: Math.floor(Date.now() / 1000), - maxCursor + accepted: result.accepted, + rejected: result.rejected, + serverTime: result.serverTime, + maxCursor: result.maxCursor }) -}) +} -sync.post('/pull', pullRateLimit, async (c) => { +const handleRecordPull = async (c: Context): Promise => { const userId = c.get('userId')! + const endpoint = getRequestPath(c) + const startedAt = Date.now() const body: unknown = await c.req.json() - const parsed = PullRequestSchema.safeParse(body) - if (!parsed.success) { - throw new AppError( - ErrorCodes.VALIDATION_ERROR, - `Invalid pull request: ${parsed.error.issues[0]?.message ?? 'validation failed'}`, - 400 - ) - } + const parsed = parseTransportRequest(PullRequestSchema, body, { + transport: 'record', + endpoint, + label: 'pull request' + }) + + const items = await pullItems(c.env.DB, c.env.STORAGE, userId, parsed.itemIds) + logRecordQueryBatch({ + endpoint, + operation: 'pull', + latencyMs: Date.now() - startedAt, + itemTypes: items.map((item) => item.type) + }) - const items = await pullItems(c.env.DB, c.env.STORAGE, userId, parsed.data.itemIds) return c.json({ items }) -}) +} -sync.get('/items/:id', async (c) => { +const handleRecordItem = async (c: Context): Promise => { const userId = c.get('userId')! const itemId = c.req.param('id') @@ -223,35 +319,31 @@ sync.get('/items/:id', async (c) => { throw new AppError(ErrorCodes.VALIDATION_ERROR, 'Invalid item ID format', 400) } - const item = await getItem(c.env.DB, c.env.STORAGE, userId, itemId) + const item = await getItem(c.env.DB, c.env.STORAGE, userId, parseResult.data) return c.json(item) -}) +} -// ============================================================================ -// CRDT Endpoints -// ============================================================================ +const recordSync = new Hono() -const MAX_UPDATE_BYTES = 5 * 1024 * 1024 // 5MB per individual update -const BASE64_CHUNK_SIZE = 8192 +recordSync.get('/status', statusRateLimit, handleRecordStatus) +recordSync.get('/manifest', manifestRateLimit, handleRecordManifest) +recordSync.get('/changes', changesRateLimit, handleRecordChanges) +recordSync.post('/push', pushRateLimit, handleRecordPush) +recordSync.post('/pull', pullRateLimit, handleRecordPull) +recordSync.get('/items/:id', handleRecordItem) -function safeBase64Encode(buffer: ArrayBuffer): string { - const bytes = new Uint8Array(buffer) - let result = '' - for (let i = 0; i < bytes.length; i += BASE64_CHUNK_SIZE) { - const chunk = bytes.subarray(i, i + BASE64_CHUNK_SIZE) - result += String.fromCharCode(...chunk) - } - return btoa(result) -} +sync.route('/records', recordSync) -function safeBase64Decode(b64: string): Uint8Array { - const raw = atob(b64) - const bytes = new Uint8Array(raw.length) - for (let i = 0; i < raw.length; i++) { - bytes[i] = raw.charCodeAt(i) - } - return bytes -} +sync.get('/status', statusRateLimit, handleRecordStatus) +sync.get('/manifest', manifestRateLimit, handleRecordManifest) +sync.get('/changes', changesRateLimit, handleRecordChanges) +sync.post('/push', pushRateLimit, handleRecordPush) +sync.post('/pull', pullRateLimit, handleRecordPull) +sync.get('/items/:id', handleRecordItem) + +// ============================================================================ +// CRDT Endpoints +// ============================================================================ const NoteIdSchema = z .string() @@ -298,22 +390,44 @@ const CrdtPushSchema = z.object({ updates: z.array(z.string().max(MAX_UPDATE_BYTES * 2)).max(100) }) -sync.post('/crdt/updates', crdtPushRateLimit, async (c) => { +const CrdtSnapshotPushSchema = z.object({ + noteId: NoteIdSchema, + snapshot: z.string() +}) + +const handleCrdtUpdatePush = async (c: Context): Promise => { const userId = c.get('userId')! const deviceId = c.get('deviceId')! + const endpoint = getRequestPath(c) + const startedAt = Date.now() const body = await c.req.json() - const parsed = CrdtPushSchema.parse(body) - - const buffers = parsed.updates.map((b64) => { - const bytes = safeBase64Decode(b64) - if (bytes.byteLength > MAX_UPDATE_BYTES) { - throw new AppError(ErrorCodes.VALIDATION_ERROR, 'Individual update exceeds 5MB limit', 413) - } - return bytes.buffer as ArrayBuffer + const parsed = parseTransportRequest(CrdtPushSchema, body, { + transport: 'crdt', + endpoint, + label: 'CRDT updates request' }) + const buffers = parsed.updates.map((payload) => + decodeCrdtPayload(payload, endpoint, 'Individual update exceeds 5MB limit') + ) + const totalBytes = buffers.reduce((sum, buf) => sum + buf.byteLength, 0) - await checkQuota(c.env.DB, userId, totalBytes) + try { + await checkQuota(c.env.DB, userId, totalBytes) + } catch (error) { + if (error instanceof AppError && error.code === ErrorCodes.STORAGE_QUOTA_EXCEEDED) { + logCrdtTraffic({ + endpoint, + event: 'updates_rejected', + noteId: parsed.noteId, + updateCount: parsed.updates.length, + totalBytes, + latencyMs: Date.now() - startedAt, + reason: error.code + }) + } + throw error + } const sequences = await storeUpdates(c.env.DB, userId, parsed.noteId, deviceId, buffers) @@ -333,25 +447,42 @@ sync.post('/crdt/updates', crdtPushRateLimit, async (c) => { ) ) + logCrdtTraffic({ + endpoint, + event: 'updates_stored', + noteId: parsed.noteId, + updateCount: sequences.length, + totalBytes, + latencyMs: Date.now() - startedAt + }) + return c.json({ sequences }) -}) +} -sync.get('/crdt/updates', crdtPullRateLimit, async (c) => { +const handleCrdtUpdatePull = async (c: Context): Promise => { const userId = c.get('userId')! + const endpoint = getRequestPath(c) + const startedAt = Date.now() const noteIdRaw = c.req.query('note_id') const since = parseInt(c.req.query('since') ?? '0', 10) const limit = parseInt(c.req.query('limit') ?? '100', 10) if (!noteIdRaw) { - throw new AppError(ErrorCodes.VALIDATION_ERROR, 'note_id is required', 400) + logQueryValidationFailure('crdt', endpoint, 'note_id is required') } - const noteIdResult = NoteIdSchema.safeParse(noteIdRaw) + const noteId = noteIdRaw + const noteIdResult = NoteIdSchema.safeParse(noteId) if (!noteIdResult.success) { - throw new AppError(ErrorCodes.VALIDATION_ERROR, 'Invalid note_id format', 400) + logQueryValidationFailure('crdt', endpoint, 'Invalid note_id format') + } + if (isNaN(since) || since < 0) { + logQueryValidationFailure('crdt', endpoint, 'Invalid since value') + } + if (isNaN(limit) || limit < 1) { + logQueryValidationFailure('crdt', endpoint, 'Invalid limit value') } - const noteId = noteIdResult.data - const result = await getUpdates(c.env.DB, userId, noteId, since, Math.min(limit, 500)) + const result = await getUpdates(c.env.DB, userId, noteIdResult.data, since, Math.min(limit, 500)) const encoded = result.updates.map((u) => ({ sequenceNum: u.sequence_num, @@ -360,89 +491,159 @@ sync.get('/crdt/updates', crdtPullRateLimit, async (c) => { createdAt: u.created_at })) + logCrdtTraffic({ + endpoint, + event: 'updates_fetched', + noteId: noteIdResult.data, + updateCount: encoded.length, + totalBytes: result.updates.reduce((sum, update) => sum + update.update_data.byteLength, 0), + latencyMs: Date.now() - startedAt + }) + return c.json({ updates: encoded, hasMore: result.hasMore }) -}) +} -sync.post('/crdt/updates/batch', crdtBatchPullRateLimit, async (c) => { +const handleCrdtBatchPull = async (c: Context): Promise => { const userId = c.get('userId')! + const endpoint = getRequestPath(c) + const startedAt = Date.now() const body: unknown = await c.req.json() - const parsed = CrdtBatchPullSchema.safeParse(body) - if (!parsed.success) { - throw new AppError( - ErrorCodes.VALIDATION_ERROR, - `Invalid batch request: ${parsed.error.issues[0]?.message ?? 'validation failed'}`, - 400 - ) - } + const parsed = parseTransportRequest(CrdtBatchPullSchema, body, { + transport: 'crdt', + endpoint, + label: 'CRDT batch request' + }) - const { notes, limit } = parsed.data - const batchResult = await getBatchUpdates(c.env.DB, userId, notes, limit) + const batchResult = await getBatchUpdates(c.env.DB, userId, parsed.notes, parsed.limit) const response: Record = {} - for (const [noteId, r] of Object.entries(batchResult)) { + for (const [noteId, result] of Object.entries(batchResult)) { response[noteId] = { - updates: r.updates.map((u) => ({ - sequenceNum: u.sequence_num, - data: safeBase64Encode(u.update_data as ArrayBuffer), - signerDeviceId: u.signer_device_id, - createdAt: u.created_at + updates: result.updates.map((update) => ({ + sequenceNum: update.sequence_num, + data: safeBase64Encode(update.update_data as ArrayBuffer), + signerDeviceId: update.signer_device_id, + createdAt: update.created_at })), - hasMore: r.hasMore + hasMore: result.hasMore } } - return c.json({ notes: response }) -}) -const CrdtSnapshotPushSchema = z.object({ - noteId: NoteIdSchema, - snapshot: z.string() -}) + logCrdtTraffic({ + endpoint, + event: 'batch_fetched', + noteCount: parsed.notes.length, + updateCount: Object.values(batchResult).reduce((sum, result) => sum + result.updates.length, 0), + totalBytes: Object.values(batchResult).reduce( + (sum, result) => + sum + result.updates.reduce((noteSum, update) => noteSum + update.update_data.byteLength, 0), + 0 + ), + latencyMs: Date.now() - startedAt + }) + + return c.json({ notes: response }) +} -sync.post('/crdt/snapshot', crdtPushRateLimit, async (c) => { +const handleCrdtSnapshotPush = async (c: Context): Promise => { const userId = c.get('userId')! const deviceId = c.get('deviceId')! + const endpoint = getRequestPath(c) + const startedAt = Date.now() const body = await c.req.json() - const parsed = CrdtSnapshotPushSchema.parse(body) + const parsed = parseTransportRequest(CrdtSnapshotPushSchema, body, { + transport: 'crdt', + endpoint, + label: 'CRDT snapshot request' + }) - const bytes = safeBase64Decode(parsed.snapshot) - if (bytes.byteLength > MAX_UPDATE_BYTES) { - throw new AppError(ErrorCodes.VALIDATION_ERROR, 'Snapshot exceeds 5MB limit', 413) + const snapshotBytes = decodeCrdtPayload(parsed.snapshot, endpoint, 'Snapshot exceeds 5MB limit') + + try { + await checkQuota(c.env.DB, userId, snapshotBytes.byteLength) + } catch (error) { + if (error instanceof AppError && error.code === ErrorCodes.STORAGE_QUOTA_EXCEEDED) { + logCrdtTraffic({ + endpoint, + event: 'snapshot_rejected', + noteId: parsed.noteId, + totalBytes: snapshotBytes.byteLength, + latencyMs: Date.now() - startedAt, + reason: error.code + }) + } + throw error } - await checkQuota(c.env.DB, userId, bytes.byteLength) - const result = await storeSnapshot( c.env.DB, c.env.STORAGE, userId, parsed.noteId, deviceId, - bytes.buffer as ArrayBuffer + snapshotBytes ) await pruneUpdatesBeforeSnapshot(c.env.DB, userId, parsed.noteId) + logCrdtTraffic({ + endpoint, + event: 'snapshot_stored', + noteId: parsed.noteId, + totalBytes: snapshotBytes.byteLength, + sequenceNum: result.sequenceNum, + latencyMs: Date.now() - startedAt + }) + return c.json({ sequenceNum: result.sequenceNum }) -}) +} -sync.get('/crdt/snapshot/:noteId', crdtPullRateLimit, async (c) => { +const handleCrdtSnapshotPull = async (c: Context): Promise => { const userId = c.get('userId')! + const endpoint = getRequestPath(c) + const startedAt = Date.now() const noteIdRaw = c.req.param('noteId') const noteIdResult = NoteIdSchema.safeParse(noteIdRaw) if (!noteIdResult.success) { - throw new AppError(ErrorCodes.VALIDATION_ERROR, 'Invalid noteId format', 400) + logQueryValidationFailure('crdt', endpoint, 'Invalid noteId format') } const result = await getSnapshot(c.env.DB, c.env.STORAGE, userId, noteIdResult.data) if (!result) { + logCrdtTraffic({ + endpoint, + event: 'snapshot_fetched', + noteId: noteIdResult.data, + totalBytes: 0, + sequenceNum: 0, + latencyMs: Date.now() - startedAt + }) return c.json({ snapshot: null, sequenceNum: 0, signerDeviceId: null }) } - const encoded = safeBase64Encode(result.snapshotData) + logCrdtTraffic({ + endpoint, + event: 'snapshot_fetched', + noteId: noteIdResult.data, + totalBytes: result.snapshotData.byteLength, + sequenceNum: result.sequenceNum, + latencyMs: Date.now() - startedAt + }) + return c.json({ - snapshot: encoded, + snapshot: safeBase64Encode(result.snapshotData), sequenceNum: result.sequenceNum, signerDeviceId: result.signerDeviceId }) -}) +} + +const crdtSync = new Hono() + +crdtSync.post('/updates', crdtPushRateLimit, handleCrdtUpdatePush) +crdtSync.get('/updates', crdtPullRateLimit, handleCrdtUpdatePull) +crdtSync.post('/updates/batch', crdtBatchPullRateLimit, handleCrdtBatchPull) +crdtSync.post('/snapshot', crdtPushRateLimit, handleCrdtSnapshotPush) +crdtSync.get('/snapshot/:noteId', crdtPullRateLimit, handleCrdtSnapshotPull) + +sync.route('/crdt', crdtSync) diff --git a/apps/sync-server/src/services/sync-telemetry.ts b/apps/sync-server/src/services/sync-telemetry.ts new file mode 100644 index 000000000..16b93d673 --- /dev/null +++ b/apps/sync-server/src/services/sync-telemetry.ts @@ -0,0 +1,189 @@ +import type { SyncItemType } from '@memry/contracts/sync-api' + +import { createLogger } from '../lib/logger' +import type { RecordPushBatchOutcome } from './sync' + +type SyncTransport = 'record' | 'crdt' +type RecordQueryOperation = 'changes' | 'pull' +type SyncDomain = + | 'notes' + | 'tasks' + | 'projects' + | 'settings' + | 'inbox' + | 'filters' + | 'attachments' + | 'tags' + +const logger = createLogger('SyncTelemetry') + +const LATENCY_BUCKETS = [ + { max: 25, bucket: 'under_25ms' }, + { max: 100, bucket: '25_to_99ms' }, + { max: 250, bucket: '100_to_249ms' }, + { max: 1000, bucket: '250_to_999ms' } +] as const + +const toSyncDomain = (itemType: SyncItemType): SyncDomain => { + switch (itemType) { + case 'note': + case 'journal': + return 'notes' + case 'task': + return 'tasks' + case 'project': + return 'projects' + case 'settings': + return 'settings' + case 'inbox': + return 'inbox' + case 'filter': + return 'filters' + case 'attachment': + return 'attachments' + case 'tag_definition': + return 'tags' + } +} + +const toLatencyBucket = (latencyMs: number): string => { + for (const entry of LATENCY_BUCKETS) { + if (latencyMs < entry.max) { + return entry.bucket + } + } + + return '1s_plus' +} + +const summarizeItemTypes = (itemTypes: SyncItemType[]): Partial> => { + const summary: Partial> = {} + for (const itemType of itemTypes) { + const domain = toSyncDomain(itemType) + summary[domain] = (summary[domain] ?? 0) + 1 + } + return summary +} + +export const logSyncValidationFailure = (params: { + transport: SyncTransport + endpoint: string + issue: string +}): void => { + logger.warn('Sync request validation failed', params) +} + +export const logRecordPushBatch = (params: { + endpoint: string + latencyMs: number + outcomes: RecordPushBatchOutcome[] +}): void => { + const domains: Partial< + Record< + SyncDomain, + { + accepted: number + rejected: number + replayRejected: number + conflictRejected: number + quotaRejected: number + otherRejected: number + } + > + > = {} + + let accepted = 0 + let rejected = 0 + + for (const outcome of params.outcomes) { + const domain = toSyncDomain(outcome.type) + const entry = domains[domain] ?? { + accepted: 0, + rejected: 0, + replayRejected: 0, + conflictRejected: 0, + quotaRejected: 0, + otherRejected: 0 + } + + if (outcome.accepted) { + entry.accepted += 1 + accepted += 1 + } else { + entry.rejected += 1 + rejected += 1 + switch (outcome.reason) { + case 'SYNC_REPLAY_DETECTED': + entry.replayRejected += 1 + break + case 'SYNC_VERSION_CONFLICT': + entry.conflictRejected += 1 + break + case 'STORAGE_QUOTA_EXCEEDED': + entry.quotaRejected += 1 + break + default: + entry.otherRejected += 1 + } + } + + domains[domain] = entry + } + + logger.info('Record sync push processed', { + transport: 'record', + operation: 'push', + endpoint: params.endpoint, + accepted, + rejected, + totalMutations: params.outcomes.length, + domains, + latencyMs: params.latencyMs, + latencyBucket: toLatencyBucket(params.latencyMs) + }) +} + +export const logRecordQueryBatch = (params: { + endpoint: string + operation: RecordQueryOperation + latencyMs: number + itemTypes: SyncItemType[] + deletedCount?: number +}): void => { + logger.info('Record sync query processed', { + transport: 'record', + operation: params.operation, + endpoint: params.endpoint, + itemCount: params.itemTypes.length, + deletedCount: params.deletedCount ?? 0, + domains: summarizeItemTypes(params.itemTypes), + latencyMs: params.latencyMs, + latencyBucket: toLatencyBucket(params.latencyMs) + }) +} + +export const logCrdtTraffic = (params: { + endpoint: string + event: + | 'updates_stored' + | 'updates_rejected' + | 'updates_fetched' + | 'snapshot_stored' + | 'snapshot_rejected' + | 'snapshot_fetched' + | 'batch_fetched' + noteId?: string + updateCount?: number + noteCount?: number + totalBytes?: number + sequenceNum?: number + latencyMs: number + reason?: string +}): void => { + logger.info('CRDT sync activity', { + transport: 'crdt', + domain: 'notes', + ...params, + latencyBucket: toLatencyBucket(params.latencyMs) + }) +} diff --git a/apps/sync-server/src/services/sync.ts b/apps/sync-server/src/services/sync.ts index de5796e78..0ed8606de 100644 --- a/apps/sync-server/src/services/sync.ts +++ b/apps/sync-server/src/services/sync.ts @@ -2,7 +2,9 @@ import { CRYPTO_VERSION, ED25519_PARAMS, XCHACHA20_PARAMS } from '@memry/contrac import type { ChangesResponse, EncryptedItemPayload, + PullItemResponse, PushItemInput, + PushResponse, SyncItemRef, SyncManifest, SyncStatus, @@ -14,6 +16,7 @@ import { AppError, ErrorCodes } from '../lib/errors' import { generateBlobKey, getBlob, putBlob } from './blob' import { getNextCursor } from './cursor' import { getDevice } from './device' +import { checkQuota } from './quota' import { getUserById } from './user' const MAX_ENCRYPTED_DATA_BYTES = 5 * 1024 * 1024 @@ -28,6 +31,32 @@ interface ExistingSyncItemRow { createdAt?: number | null } +interface StoredSyncItemPullRow { + item_id: string + item_type: string + blob_key: string + crypto_version: number + operation: string + signer_device_id: string | null + signature: string | null + state_vector: string | null + clock: string | null + deleted_at: number | null + server_cursor: number +} + +export interface RecordPushBatchOutcome { + id: string + type: PushItemInput['type'] + accepted: boolean + reason?: string + serverCursor?: number +} + +export interface RecordPushBatchResult extends PushResponse { + outcomes: RecordPushBatchOutcome[] +} + export const validateEncryptedFields = (item: PushItemInput): void => { const dataNonce = safeBase64Decode(item.dataNonce) if (dataNonce.length !== XCHACHA20_PARAMS.NONCE_LENGTH) { @@ -156,6 +185,118 @@ export const serializePayload = (item: PushItemInput): string => { return JSON.stringify(payload, Object.keys(payload).sort()) } +const estimatePushBatchBytes = (items: PushItemInput[]): number => JSON.stringify(items).length + +const parseStoredClock = (itemId: string, clock: string | null): VectorClock | undefined => { + if (!clock) { + return undefined + } + + try { + return JSON.parse(clock) as VectorClock + } catch { + throw new AppError(ErrorCodes.INTERNAL_ERROR, `Corrupt clock payload for item ${itemId}`, 500) + } +} + +const readEncryptedPayload = async ( + storage: R2Bucket, + blobKey: string, + userId: string, + itemId: string +): Promise => { + const blob = await getBlob(storage, blobKey, userId) + if (!blob) { + throw new AppError( + ErrorCodes.STORAGE_BLOB_NOT_FOUND, + `Blob missing for item ${itemId}`, + 404 + ) + } + + try { + const text = await new Response(blob.body).text() + return JSON.parse(text) as EncryptedItemPayload + } catch { + throw new AppError(ErrorCodes.INTERNAL_ERROR, `Corrupt blob payload for item ${itemId}`, 500) + } +} + +const toPullItemResponse = async ( + storage: R2Bucket, + userId: string, + row: StoredSyncItemPullRow +): Promise => { + const payload = await readEncryptedPayload(storage, row.blob_key, userId, row.item_id) + + if (!row.signer_device_id || !row.signature) { + throw new AppError( + ErrorCodes.INTERNAL_ERROR, + `Sync item ${row.item_id} missing signer metadata`, + 500 + ) + } + + const parsedClock = parseStoredClock(row.item_id, row.clock) + + return { + id: row.item_id, + type: row.item_type as PullItemResponse['type'], + operation: row.operation as PullItemResponse['operation'], + cryptoVersion: row.crypto_version, + signature: row.signature, + signerDeviceId: row.signer_device_id, + ...(row.deleted_at ? { deletedAt: row.deleted_at } : {}), + ...(parsedClock ? { clock: parsedClock } : {}), + ...(row.state_vector ? { stateVector: row.state_vector } : {}), + blob: payload + } +} + +export const processRecordPushBatch = async ( + db: D1Database, + storage: R2Bucket, + userId: string, + deviceId: string, + items: PushItemInput[] +): Promise => { + await checkQuota(db, userId, estimatePushBatchBytes(items)) + + const accepted: string[] = [] + const rejected: Array<{ id: string; reason: string }> = [] + const outcomes: RecordPushBatchOutcome[] = [] + let maxCursor = 0 + + for (const item of items) { + const result = await processPushItem(db, storage, userId, deviceId, item) + outcomes.push({ + id: item.id, + type: item.type, + accepted: result.accepted, + reason: result.reason, + serverCursor: result.serverCursor + }) + + if (result.accepted) { + accepted.push(item.id) + if (result.serverCursor && result.serverCursor > maxCursor) { + maxCursor = result.serverCursor + } + continue + } + + rejected.push({ id: item.id, reason: result.reason ?? 'UNKNOWN' }) + } + + return { + accepted, + rejected, + serverTime: Math.floor(Date.now() / 1000), + maxCursor, + outcomes + } +} + export const processPushItem = async ( db: D1Database, storage: R2Bucket, @@ -416,39 +557,14 @@ export const pullItems = async ( storage: R2Bucket, userId: string, itemIds: string[] -): Promise< - Array<{ - id: string - type: string - operation: string - cryptoVersion: number - signature: string - signerDeviceId: string - deletedAt?: number - clock?: VectorClock - stateVector?: string - blob: EncryptedItemPayload - }> -> => { +): Promise => { if (itemIds.length === 0) { return [] } const BATCH_SIZE = D1_MAX_BIND_PARAMS - 1 - const allDbRows: Array<{ - item_id: string - item_type: string - blob_key: string - crypto_version: number - operation: string - signer_device_id: string | null - signature: string | null - state_vector: string | null - clock: string | null - deleted_at: number | null - server_cursor: number - }> = [] + const allDbRows: StoredSyncItemPullRow[] = [] for (let i = 0; i < itemIds.length; i += BATCH_SIZE) { const batch = itemIds.slice(i, i + BATCH_SIZE) @@ -462,92 +578,16 @@ export const pullItems = async ( ORDER BY server_cursor ASC` ) .bind(userId, ...batch) - .all<{ - item_id: string - item_type: string - blob_key: string - crypto_version: number - operation: string - signer_device_id: string | null - signature: string | null - state_vector: string | null - clock: string | null - deleted_at: number | null - server_cursor: number - }>() + .all() allDbRows.push(...(rows.results ?? [])) } allDbRows.sort((a, b) => a.server_cursor - b.server_cursor) - const results: Array<{ - id: string - type: string - operation: string - cryptoVersion: number - signature: string - signerDeviceId: string - deletedAt?: number - clock?: VectorClock - stateVector?: string - blob: EncryptedItemPayload - }> = [] + const results: PullItemResponse[] = [] for (const row of allDbRows) { - const blob = await getBlob(storage, row.blob_key, userId) - if (!blob) { - throw new AppError( - ErrorCodes.STORAGE_BLOB_NOT_FOUND, - `Blob missing for item ${row.item_id}`, - 404 - ) - } - - let payload: EncryptedItemPayload - try { - const text = await new Response(blob.body).text() - payload = JSON.parse(text) as EncryptedItemPayload - } catch { - throw new AppError( - ErrorCodes.INTERNAL_ERROR, - `Corrupt blob payload for item ${row.item_id}`, - 500 - ) - } - - if (!row.signer_device_id || !row.signature) { - throw new AppError( - ErrorCodes.INTERNAL_ERROR, - `Sync item ${row.item_id} missing signer metadata`, - 500 - ) - } - - let parsedClock: VectorClock | undefined - if (row.clock) { - try { - parsedClock = JSON.parse(row.clock) as VectorClock - } catch { - throw new AppError( - ErrorCodes.INTERNAL_ERROR, - `Corrupt clock payload for item ${row.item_id}`, - 500 - ) - } - } - - results.push({ - id: row.item_id, - type: row.item_type, - operation: row.operation, - cryptoVersion: row.crypto_version, - signature: row.signature, - signerDeviceId: row.signer_device_id, - ...(row.deleted_at ? { deletedAt: row.deleted_at } : {}), - ...(parsedClock ? { clock: parsedClock } : {}), - ...(row.state_vector ? { stateVector: row.state_vector } : {}), - blob: payload - }) + results.push(await toPullItemResponse(storage, userId, row)) } return results @@ -584,18 +624,7 @@ export const getItem = async ( throw new AppError(ErrorCodes.SYNC_ITEM_NOT_FOUND, 'Sync item not found', 404) } - const blob = await getBlob(storage, row.blob_key, userId) - if (!blob) { - throw new AppError(ErrorCodes.STORAGE_BLOB_NOT_FOUND, 'Item blob not found in storage', 404) - } - - let payload: EncryptedItemPayload - try { - const text = await new Response(blob.body).text() - payload = JSON.parse(text) as EncryptedItemPayload - } catch { - throw new AppError(ErrorCodes.INTERNAL_ERROR, `Corrupt blob payload for item ${itemId}`, 500) - } + const payload = await readEncryptedPayload(storage, row.blob_key, userId, itemId) return { itemId: row.item_id,