From 7f20a4517ce7aad2a66dc7fbebd27eeae2c5a4e2 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Wed, 25 Mar 2026 21:33:47 +0100 Subject: [PATCH] feat(federation): implement checkpoint sync endpoint using mutation log Add POST /api/federation/sync for peer instances to catch up on missed DM mutations after downtime. Uses HMAC-SHA256 authentication (same as relay), determines shared channels between peers via raw SQL join on dm_members/users, queries the federation_mutation_log with timestamp pagination, and returns FederationRelayEvent payloads for each mutation type (create, update, delete, reaction_add, reaction_remove). --- packages/server/src/routes/federation.ts | 329 ++++++++++++++++++++++- 1 file changed, 327 insertions(+), 2 deletions(-) diff --git a/packages/server/src/routes/federation.ts b/packages/server/src/routes/federation.ts index af7116b3..bd184251 100644 --- a/packages/server/src/routes/federation.ts +++ b/packages/server/src/routes/federation.ts @@ -4,14 +4,14 @@ import { eq, and, or, isNull } from 'drizzle-orm'; import { authenticate, requireAdmin } from '../utils/auth.js'; import { generateHmacSecret, parseFederationHeaders, verifySignature } from '../utils/federationAuth.js'; import { generateSnowflake } from '../utils/snowflake.js'; -import { getDb, schema } from '../db/index.js'; +import { getDb, getRawDb, schema } from '../db/index.js'; import { config } from '../config.js'; import { connectionManager } from '../ws/handler.js'; import { sanitizeUser } from '../utils/sanitize.js'; import { deleteAttachmentFiles } from '../utils/fileCleanup.js'; import { canonicalDmPairId } from '../utils/federationOutbox.js'; import { broadcastDmMessage } from './dm.js'; -import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent, DmMessageWithUser } from '@backspace/shared'; +import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent, FederationRelayAttachment, FederationSyncRequest, FederationSyncResponse, DmMessageWithUser } from '@backspace/shared'; /** Fields safe to expose to admin callers (everything except hmacSecret). */ interface SanitizedPeer { @@ -479,6 +479,331 @@ export async function federationRoutes(app: FastifyInstance): Promise { return reply.code(200).send(response); }, ); + + // ─── POST /api/federation/sync ────────────────────────────────────────────── + // Server-to-server: checkpoint catch-up sync. A peer calls this after downtime + // to retrieve missed DM mutations from the mutation log. + // Authenticated via HMAC-SHA256 signature, same as /relay. + app.post<{ Body: FederationSyncRequest }>( + '/api/federation/sync', + { bodyLimit: 1024 * 64 }, + async (request, reply) => { + const db = getDb(); + const rawDb = getRawDb(); + + // 1. Verify HMAC signature + const fedHeaders = parseFederationHeaders(request.headers as Record); + if (!fedHeaders) { + return reply.code(401).send({ error: 'Missing or malformed federation headers', statusCode: 401 }); + } + + const peer = db + .select() + .from(schema.federationPeers) + .where(eq(schema.federationPeers.origin, fedHeaders.origin)) + .get(); + + if (!peer || peer.status !== 'active') { + return reply.code(403).send({ error: 'Unknown or inactive peer', statusCode: 403 }); + } + + const bodyString = JSON.stringify(request.body); + if (!verifySignature(bodyString, fedHeaders.signature, peer.hmacSecret, fedHeaders.timestamp)) { + return reply.code(401).send({ error: 'Invalid signature', statusCode: 401 }); + } + + // 2. Validate & normalize request body + const body = request.body; + if (!body || typeof body.sinceTimestamp !== 'number' || body.sinceTimestamp < 0) { + return reply.code(400).send({ error: 'sinceTimestamp must be a non-negative number', statusCode: 400 }); + } + + const sinceTimestamp = body.sinceTimestamp; + const dmChannelIdFilter = body.dmChannelId && typeof body.dmChannelId === 'string' ? body.dmChannelId : null; + + // Clamp limit: min 1, max 500, default 100 + let limit = typeof body.limit === 'number' ? body.limit : 100; + limit = Math.max(1, Math.min(500, Math.floor(limit))); + + // 3. Determine shared DM channels between this instance and the requesting peer + // Shared channels have one local user (home_instance IS NULL) and + // one peer user (home_instance = peer hostname), with exactly 2 members. + let peerHost: string; + try { + peerHost = new URL(peer.origin).host; + } catch { + return reply.code(500).send({ error: 'Invalid peer origin in database', statusCode: 500 }); + } + + // Find shared 1-on-1 DM channel IDs using raw SQL for the complex JOIN. + // A shared channel is one where: + // - There are exactly 2 members + // - One member is a local user (home_instance IS NULL) + // - One member is a user from the peer (home_instance = peerHost) + const sharedChannelRows = rawDb.prepare(` + SELECT DISTINCT dm1.dm_channel_id + FROM dm_members dm1 + JOIN dm_members dm2 ON dm1.dm_channel_id = dm2.dm_channel_id AND dm1.user_id != dm2.user_id + JOIN users u1 ON dm1.user_id = u1.id + JOIN users u2 ON dm2.user_id = u2.id + WHERE u1.home_instance IS NULL + AND u2.home_instance = ? + AND ( + SELECT COUNT(*) FROM dm_members dm3 + WHERE dm3.dm_channel_id = dm1.dm_channel_id + ) = 2 + `).all(peerHost) as Array<{ dm_channel_id: string }>; + + const sharedChannelIds = sharedChannelRows.map(r => r.dm_channel_id); + + if (sharedChannelIds.length === 0) { + const syncResponse: FederationSyncResponse = { + events: [], + hasMore: false, + checkpoint: sinceTimestamp, + }; + return reply.code(200).send(syncResponse); + } + + // 4. Query mutation log for the relevant channels + // Only return mutations for messages authored by LOCAL users + // (source_instance IS NULL) — each instance is authoritative for its own users' messages. + let mutationRows: Array<{ + id: string; + dm_message_id: string; + dm_channel_id: string; + mutation_type: string; + mutated_at: number; + payload: string | null; + }>; + + if (dmChannelIdFilter) { + // Validate that the requested channel is actually shared with this peer + if (!sharedChannelIds.includes(dmChannelIdFilter)) { + const syncResponse: FederationSyncResponse = { + events: [], + hasMore: false, + checkpoint: sinceTimestamp, + }; + return reply.code(200).send(syncResponse); + } + + mutationRows = rawDb.prepare(` + SELECT ml.id, ml.dm_message_id, ml.dm_channel_id, ml.mutation_type, ml.mutated_at, ml.payload + FROM federation_mutation_log ml + JOIN dm_messages dm ON ml.dm_message_id = dm.id + WHERE ml.dm_channel_id = ? + AND ml.mutated_at > ? + AND dm.source_instance IS NULL + ORDER BY ml.mutated_at ASC + LIMIT ? + `).all(dmChannelIdFilter, sinceTimestamp, limit) as typeof mutationRows; + + // For delete mutations, the dm_messages row won't exist — handle separately + const deleteMutations = rawDb.prepare(` + SELECT ml.id, ml.dm_message_id, ml.dm_channel_id, ml.mutation_type, ml.mutated_at, ml.payload + FROM federation_mutation_log ml + WHERE ml.dm_channel_id = ? + AND ml.mutated_at > ? + AND ml.mutation_type = 'delete' + AND ml.dm_message_id NOT IN (SELECT dm.id FROM dm_messages dm WHERE dm.id = ml.dm_message_id) + ORDER BY ml.mutated_at ASC + LIMIT ? + `).all(dmChannelIdFilter, sinceTimestamp, limit) as typeof mutationRows; + + // Merge, deduplicate, sort, and re-limit + const seen = new Set(mutationRows.map(r => r.id)); + for (const row of deleteMutations) { + if (!seen.has(row.id)) { + mutationRows.push(row); + seen.add(row.id); + } + } + mutationRows.sort((a, b) => a.mutated_at - b.mutated_at); + if (mutationRows.length > limit) { + mutationRows = mutationRows.slice(0, limit); + } + } else { + // All shared channels — build IN clause with placeholders + const placeholders = sharedChannelIds.map(() => '?').join(','); + + mutationRows = rawDb.prepare(` + SELECT ml.id, ml.dm_message_id, ml.dm_channel_id, ml.mutation_type, ml.mutated_at, ml.payload + FROM federation_mutation_log ml + JOIN dm_messages dm ON ml.dm_message_id = dm.id + WHERE ml.dm_channel_id IN (${placeholders}) + AND ml.mutated_at > ? + AND dm.source_instance IS NULL + ORDER BY ml.mutated_at ASC + LIMIT ? + `).all(...sharedChannelIds, sinceTimestamp, limit) as typeof mutationRows; + + // For delete mutations, the dm_messages row won't exist — handle separately + const deleteMutations = rawDb.prepare(` + SELECT ml.id, ml.dm_message_id, ml.dm_channel_id, ml.mutation_type, ml.mutated_at, ml.payload + FROM federation_mutation_log ml + WHERE ml.dm_channel_id IN (${placeholders}) + AND ml.mutated_at > ? + AND ml.mutation_type = 'delete' + AND ml.dm_message_id NOT IN (SELECT dm.id FROM dm_messages dm WHERE dm.id = ml.dm_message_id) + ORDER BY ml.mutated_at ASC + LIMIT ? + `).all(...sharedChannelIds, sinceTimestamp, limit) as typeof mutationRows; + + // Merge, deduplicate, sort, and re-limit + const seen = new Set(mutationRows.map(r => r.id)); + for (const row of deleteMutations) { + if (!seen.has(row.id)) { + mutationRows.push(row); + seen.add(row.id); + } + } + mutationRows.sort((a, b) => a.mutated_at - b.mutated_at); + if (mutationRows.length > limit) { + mutationRows = mutationRows.slice(0, limit); + } + } + + // 5. Build response events from mutation log entries + const events: FederationRelayEvent[] = []; + + for (const mutation of mutationRows) { + const mutationType = mutation.mutation_type as 'create' | 'update' | 'delete' | 'reaction_add' | 'reaction_remove'; + + if (mutationType === 'delete') { + // For deletes, we don't need the message content — just the ID and channel + events.push({ + eventType: 'delete', + dmChannelId: mutation.dm_channel_id, + messageId: mutation.dm_message_id, + encryptionVersion: 0, + timestamp: mutation.mutated_at, + }); + continue; + } + + if (mutationType === 'reaction_add' || mutationType === 'reaction_remove') { + // Use the stored payload from the mutation log + if (mutation.payload) { + let reactionData: { userId: string; homeUserId: string; emoji: string; createdAt?: number } | null = null; + try { + reactionData = JSON.parse(mutation.payload) as { userId: string; homeUserId: string; emoji: string; createdAt?: number }; + } catch { + // Skip malformed payload + continue; + } + + events.push({ + eventType: mutationType, + dmChannelId: mutation.dm_channel_id, + messageId: mutation.dm_message_id, + encryptionVersion: 0, + timestamp: mutation.mutated_at, + reaction: { + userId: reactionData.userId, + homeUserId: reactionData.homeUserId, + emoji: reactionData.emoji, + createdAt: reactionData.createdAt ?? mutation.mutated_at, + }, + }); + } + continue; + } + + // For create and update: fetch the current message state + const message = db + .select() + .from(schema.dmMessages) + .where(eq(schema.dmMessages.id, mutation.dm_message_id)) + .get(); + + if (!message) { + // Message was deleted after this create/update mutation was logged — skip it. + // The delete mutation will handle the cleanup on the peer side. + continue; + } + + // Resolve the author user to get homeUserId and homeInstance + const authorUser = db + .select() + .from(schema.users) + .where(eq(schema.users.id, message.userId)) + .get(); + + if (!authorUser) { + continue; + } + + const homeUserId = authorUser.homeUserId || authorUser.id; + const homeInstance = authorUser.homeInstance || (config.domain ? `https://${config.domain}` : ''); + + // Fetch attachments for the message + const attachmentRows = db + .select() + .from(schema.attachments) + .where(eq(schema.attachments.dmMessageId, message.id)) + .all(); + + let localOrigin: string; + try { + localOrigin = resolveLocalOrigin(request); + } catch { + localOrigin = config.domain ? `https://${config.domain}` : ''; + } + + const attachments: FederationRelayAttachment[] = attachmentRows.map(a => ({ + id: a.id, + filename: a.filename, + originalName: a.originalName, + mimetype: a.mimetype, + size: a.size, + width: a.width ?? undefined, + height: a.height ?? undefined, + duration: a.duration ?? undefined, + thumbnailFilename: a.thumbnailFilename ?? undefined, + sourceUrl: `${localOrigin}/api/uploads/${a.filename}`, + })); + + events.push({ + eventType: mutationType, + dmChannelId: mutation.dm_channel_id, + messageId: message.id, + encryptionVersion: 0, + timestamp: mutation.mutated_at, + message: { + userId: message.userId, + homeUserId, + homeInstance, + content: message.content, + replyToId: message.replyToId ?? null, + editedAt: message.editedAt ?? null, + createdAt: message.createdAt, + attachments: attachments.length > 0 ? attachments : undefined, + }, + }); + } + + // 6. Compute pagination metadata + const hasMore = mutationRows.length >= limit; + const checkpoint = mutationRows.length > 0 + ? mutationRows[mutationRows.length - 1]!.mutated_at + : sinceTimestamp; + + // 7. Update peer last-seen timestamp + db.update(schema.federationPeers) + .set({ lastSeenAt: Date.now() }) + .where(eq(schema.federationPeers.id, peer.id)) + .run(); + + const syncResponse: FederationSyncResponse = { + events, + hasMore, + checkpoint, + }; + + return reply.code(200).send(syncResponse); + }, + ); } // ─── Relay Event Processors ──────────────────────────────────────────────────