feat(federation): implement relay receive endpoint with dedup, channel mapping, SSRF validation

packages/server/src/routes/federation.ts

@@ -1,11 +1,17 @@
 import type { FastifyInstance } from 'fastify';
 import { randomBytes } from 'node:crypto';
-import { eq } from 'drizzle-orm';
+import { eq, and, or, isNull } from 'drizzle-orm';
 import { authenticate, requireAdmin } from '../utils/auth.js';
-import { generateHmacSecret } from '../utils/federationAuth.js';
+import { generateHmacSecret, parseFederationHeaders, verifySignature } from '../utils/federationAuth.js';
 import { generateSnowflake } from '../utils/snowflake.js';
 import { getDb, 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';
 
 /** Fields safe to expose to admin callers (everything except hmacSecret). */
 interface SanitizedPeer {
@@ -368,4 +374,762 @@ export async function federationRoutes(app: FastifyInstance): Promise<void> {
       return reply.code(200).send({ success: true });
     },
   );
+
+  // ─── POST /api/federation/relay ────────────────────────────────────────────
+  // Server-to-server: receive relayed DM events from a peer instance.
+  // Authenticated via HMAC-SHA256 signature, NOT JWT.
+  app.post<{ Body: FederationRelayRequest }>(
+    '/api/federation/relay',
+    { bodyLimit: 10 * 1024 * 1024 },
+    async (request, reply) => {
+      const db = getDb();
+
+      // 1. Verify HMAC signature
+      const fedHeaders = parseFederationHeaders(request.headers as Record<string, string | string[] | undefined>);
+      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 });
+      }
+
+      // Serialize body back to JSON for HMAC verification (we control both sides)
+      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 request body shape
+      const body = request.body;
+      if (!body || body.version !== 1 || !Array.isArray(body.events)) {
+        return reply.code(400).send({ error: 'Invalid relay request format', statusCode: 400 });
+      }
+
+      if (body.events.length > 50) {
+        return reply.code(400).send({ error: 'Maximum 50 events per batch', statusCode: 400 });
+      }
+
+      const sourceInstance = body.sourceInstance;
+      if (!sourceInstance || typeof sourceInstance !== 'string') {
+        return reply.code(400).send({ error: 'sourceInstance is required', statusCode: 400 });
+      }
+
+      // 3. Process each event
+      const accepted: string[] = [];
+      const rejected: Array<{ messageId: string; reason: string }> = [];
+
+      for (const event of body.events) {
+        try {
+          switch (event.eventType) {
+            case 'create':
+              processCreateEvent(event, sourceInstance, peer.origin, db, accepted, rejected);
+              break;
+            case 'update':
+              processUpdateEvent(event, sourceInstance, db, accepted, rejected);
+              break;
+            case 'delete':
+              processDeleteEvent(event, sourceInstance, db, accepted, rejected);
+              break;
+            case 'reaction_add':
+              processReactionAddEvent(event, sourceInstance, db, accepted, rejected);
+              break;
+            case 'reaction_remove':
+              processReactionRemoveEvent(event, sourceInstance, db, accepted, rejected);
+              break;
+            default:
+              rejected.push({ messageId: event.messageId, reason: 'unknown_event_type' });
+              break;
+          }
+        } catch (err) {
+          const errMsg = err instanceof Error ? err.message : 'unknown_error';
+          console.error(`[federation-relay] Error processing event ${event.messageId}:`, errMsg);
+          rejected.push({ messageId: event.messageId, reason: 'processing_error' });
+        }
+      }
+
+      // 4. Update peer status
+      db.update(schema.federationPeers)
+        .set({
+          lastSeenAt: Date.now(),
+          consecutiveFailures: 0,
+        })
+        .where(eq(schema.federationPeers.id, peer.id))
+        .run();
+
+      // 5. Return response with max upload size info
+      const settings = db
+        .select({ maxUploadSizeBytes: schema.instanceSettings.maxUploadSizeBytes })
+        .from(schema.instanceSettings)
+        .where(eq(schema.instanceSettings.id, 1))
+        .get();
+
+      const response: FederationRelayResponse = {
+        accepted,
+        rejected,
+        maxUploadSize: settings?.maxUploadSizeBytes ?? config.maxUploadSize,
+      };
+
+      return reply.code(200).send(response);
+    },
+  );
+}
+
+// ─── Relay Event Processors ──────────────────────────────────────────────────
+
+/**
+ * Resolve a home user ID to a local user.
+ * Matches users where home_user_id = homeUserId, or where
+ * the user's own id equals homeUserId and they have no home_instance set (local user).
+ */
+function resolveLocalUser(
+  homeUserId: string,
+  db: ReturnType<typeof getDb>,
+): typeof schema.users.$inferSelect | undefined {
+  return db
+    .select()
+    .from(schema.users)
+    .where(
+      or(
+        eq(schema.users.homeUserId, homeUserId),
+        and(eq(schema.users.id, homeUserId), isNull(schema.users.homeInstance)),
+      ),
+    )
+    .get();
+}
+
+/**
+ * Find or create a local DM channel for a federated 1-on-1 pair.
+ * Uses canonical_pair_id for deterministic cross-instance lookup.
+ */
+function findOrCreateDmChannel(
+  canonicalPairId: string,
+  localUserIdA: string,
+  localUserIdB: string,
+  db: ReturnType<typeof getDb>,
+): string {
+  // Try to find existing channel by canonical pair ID
+  const existing = db
+    .select()
+    .from(schema.dmChannels)
+    .where(eq(schema.dmChannels.canonicalPairId, canonicalPairId))
+    .get();
+
+  if (existing) {
+    // Ensure both users are members (they might have been removed)
+    for (const userId of [localUserIdA, localUserIdB]) {
+      const member = db
+        .select()
+        .from(schema.dmMembers)
+        .where(
+          and(
+            eq(schema.dmMembers.dmChannelId, existing.id),
+            eq(schema.dmMembers.userId, userId),
+          ),
+        )
+        .get();
+
+      if (!member) {
+        db.insert(schema.dmMembers)
+          .values({
+            dmChannelId: existing.id,
+            userId,
+            closed: 0,
+          })
+          .run();
+      }
+    }
+    return existing.id;
+  }
+
+  // Create new DM channel with canonical pair ID
+  const channelId = generateSnowflake();
+  const now = Date.now();
+
+  db.insert(schema.dmChannels)
+    .values({
+      id: channelId,
+      canonicalPairId,
+      createdAt: now,
+    })
+    .run();
+
+  for (const userId of [localUserIdA, localUserIdB]) {
+    db.insert(schema.dmMembers)
+      .values({
+        dmChannelId: channelId,
+        userId,
+        closed: 0,
+      })
+      .run();
+  }
+
+  return channelId;
+}
+
+/**
+ * Build a DmMessageWithUser payload for WebSocket broadcasting.
+ */
+function buildDmMessagePayload(
+  messageRow: {
+    id: string;
+    dmChannelId: string;
+    userId: string;
+    content: string | null;
+    replyToId: string | null;
+    editedAt: number | null;
+    createdAt: number;
+  },
+  userRow: typeof schema.users.$inferSelect,
+): DmMessageWithUser {
+  return {
+    id: messageRow.id,
+    dmChannelId: messageRow.dmChannelId,
+    channelId: messageRow.dmChannelId,
+    userId: messageRow.userId,
+    content: messageRow.content,
+    replyToId: messageRow.replyToId,
+    editedAt: messageRow.editedAt,
+    createdAt: messageRow.createdAt,
+    user: sanitizeUser(userRow),
+    attachments: [],
+    embeds: [],
+    reactions: [],
+  };
+}
+
+/**
+ * Validate that a URL's hostname matches the peer origin's hostname (SSRF protection).
+ */
+function isUrlFromPeer(sourceUrl: string, peerOrigin: string): boolean {
+  try {
+    const sourceHost = new URL(sourceUrl).hostname;
+    const peerHost = new URL(peerOrigin).hostname;
+    return sourceHost === peerHost;
+  } catch {
+    return false;
+  }
+}
+
+function processCreateEvent(
+  event: FederationRelayEvent,
+  sourceInstance: string,
+  peerOrigin: string,
+  db: ReturnType<typeof getDb>,
+  accepted: string[],
+  rejected: Array<{ messageId: string; reason: string }>,
+): void {
+  if (!event.message) {
+    rejected.push({ messageId: event.messageId, reason: 'missing_message_payload' });
+    return;
+  }
+
+  // Resolve the message author to a local user
+  const authorUser = resolveLocalUser(event.message.homeUserId, db);
+  if (!authorUser) {
+    rejected.push({ messageId: event.messageId, reason: 'user_not_found' });
+    return;
+  }
+
+  // Dedup: check for existing message with same source
+  const existingMsg = db
+    .select()
+    .from(schema.dmMessages)
+    .where(
+      and(
+        eq(schema.dmMessages.sourceInstance, sourceInstance),
+        eq(schema.dmMessages.sourceMessageId, event.messageId),
+      ),
+    )
+    .get();
+
+  if (existingMsg) {
+    rejected.push({ messageId: event.messageId, reason: 'duplicate' });
+    return;
+  }
+
+  // Resolve the DM recipient. Federated DMs are 1-on-1: one side is the author,
+  // the other is a local user on this instance. We match via canonical_pair_id.
+  const authorHomeUserId = event.message.homeUserId;
+
+  // First, search existing DM channels where the author is already a member
+  const authorMemberships = db
+    .select({ dmChannelId: schema.dmMembers.dmChannelId })
+    .from(schema.dmMembers)
+    .where(eq(schema.dmMembers.userId, authorUser.id))
+    .all();
+
+  let localDmChannelId: string | null = null;
+
+  // Check each of the author's DM channels to find the matching one
+  for (const membership of authorMemberships) {
+    const channelMembers = db
+      .select()
+      .from(schema.dmMembers)
+      .where(eq(schema.dmMembers.dmChannelId, membership.dmChannelId))
+      .all();
+
+    // For 1-on-1 DMs, there should be exactly 2 members
+    if (channelMembers.length === 2) {
+      const otherMember = channelMembers.find(m => m.userId !== authorUser.id);
+      if (otherMember) {
+        const otherUser = db
+          .select()
+          .from(schema.users)
+          .where(eq(schema.users.id, otherMember.userId))
+          .get();
+
+        if (otherUser) {
+          const otherHomeUserId = otherUser.homeUserId || otherUser.id;
+          const pairId = canonicalDmPairId(authorHomeUserId, otherHomeUserId);
+          const channel = db
+            .select()
+            .from(schema.dmChannels)
+            .where(eq(schema.dmChannels.id, membership.dmChannelId))
+            .get();
+
+          if (channel?.canonicalPairId === pairId) {
+            localDmChannelId = membership.dmChannelId;
+            break;
+          }
+        }
+      }
+    }
+  }
+
+  // If no existing channel found, search the author's friends for the recipient.
+  // On cold start (first federated DM), we use the friends list as a hint to
+  // find the local user and create the DM channel.
+  if (!localDmChannelId) {
+    const friendRows = db
+      .select()
+      .from(schema.friends)
+      .where(
+        or(
+          eq(schema.friends.userId, authorUser.id),
+          eq(schema.friends.friendId, authorUser.id),
+        ),
+      )
+      .all();
+
+    const friendIds = friendRows.map(f =>
+      f.userId === authorUser.id ? f.friendId : f.userId,
+    );
+
+    for (const friendId of friendIds) {
+      const friendUser = db
+        .select()
+        .from(schema.users)
+        .where(eq(schema.users.id, friendId))
+        .get();
+
+      if (friendUser) {
+        const friendHomeUserId = friendUser.homeUserId || friendUser.id;
+        const pairId = canonicalDmPairId(authorHomeUserId, friendHomeUserId);
+
+        // Check if a channel already exists with this pair ID
+        const existingChannel = db
+          .select()
+          .from(schema.dmChannels)
+          .where(eq(schema.dmChannels.canonicalPairId, pairId))
+          .get();
+
+        if (existingChannel) {
+          localDmChannelId = existingChannel.id;
+          break;
+        }
+      }
+    }
+
+    // If still no channel with a matching canonical pair ID, create one.
+    // The recipient must be a local (non-federated) user who is friends with the author.
+    if (!localDmChannelId && friendIds.length > 0) {
+      for (const friendId of friendIds) {
+        const friendUser = db
+          .select()
+          .from(schema.users)
+          .where(eq(schema.users.id, friendId))
+          .get();
+
+        if (friendUser && !friendUser.homeInstance) {
+          // This is a local user — they're a candidate recipient
+          const friendHomeUserId = friendUser.homeUserId || friendUser.id;
+          const pairId = canonicalDmPairId(authorHomeUserId, friendHomeUserId);
+
+          localDmChannelId = findOrCreateDmChannel(pairId, authorUser.id, friendId, db);
+          break;
+        }
+      }
+    }
+  }
+
+  if (!localDmChannelId) {
+    rejected.push({ messageId: event.messageId, reason: 'recipient_not_found' });
+    return;
+  }
+
+  // Insert the message
+  const localMessageId = generateSnowflake();
+  db.insert(schema.dmMessages)
+    .values({
+      id: localMessageId,
+      dmChannelId: localDmChannelId,
+      userId: authorUser.id,
+      content: event.message.content,
+      replyToId: null,
+      createdAt: event.message.createdAt,
+      editedAt: null,
+      sourceInstance,
+      sourceMessageId: event.messageId,
+      encryptionVersion: 0,
+    })
+    .run();
+
+  // Queue attachment downloads (SSRF-validated)
+  if (event.message.attachments && event.message.attachments.length > 0) {
+    const now = Date.now();
+    for (const attachment of event.message.attachments) {
+      if (!isUrlFromPeer(attachment.sourceUrl, peerOrigin)) {
+        console.warn(
+          `[federation-relay] Rejecting attachment URL ${attachment.sourceUrl} — hostname does not match peer ${peerOrigin}`,
+        );
+        continue;
+      }
+
+      db.insert(schema.federationFileQueue)
+        .values({
+          id: generateSnowflake(),
+          peerOrigin,
+          dmMessageId: localMessageId,
+          sourceUrl: attachment.sourceUrl,
+          originalName: attachment.originalName,
+          mimetype: attachment.mimetype,
+          size: attachment.size,
+          status: 'pending',
+          nextRetryAt: now,
+          expiresAt: now + 30 * 86_400_000,
+          createdAt: now,
+        })
+        .run();
+    }
+  }
+
+  // Broadcast to local WebSocket clients
+  const messagePayload = buildDmMessagePayload(
+    {
+      id: localMessageId,
+      dmChannelId: localDmChannelId,
+      userId: authorUser.id,
+      content: event.message.content,
+      replyToId: null,
+      editedAt: null,
+      createdAt: event.message.createdAt,
+    },
+    authorUser,
+  );
+  broadcastDmMessage(localDmChannelId, messagePayload);
+
+  accepted.push(event.messageId);
+}
+
+function processUpdateEvent(
+  event: FederationRelayEvent,
+  sourceInstance: string,
+  db: ReturnType<typeof getDb>,
+  accepted: string[],
+  rejected: Array<{ messageId: string; reason: string }>,
+): void {
+  const localMsg = db
+    .select()
+    .from(schema.dmMessages)
+    .where(
+      and(
+        eq(schema.dmMessages.sourceInstance, sourceInstance),
+        eq(schema.dmMessages.sourceMessageId, event.messageId),
+      ),
+    )
+    .get();
+
+  if (!localMsg) {
+    rejected.push({ messageId: event.messageId, reason: 'unknown_message' });
+    return;
+  }
+
+  const content = event.message?.content ?? null;
+  const editedAt = event.message?.editedAt ?? Date.now();
+
+  db.update(schema.dmMessages)
+    .set({ content, editedAt })
+    .where(eq(schema.dmMessages.id, localMsg.id))
+    .run();
+
+  // Broadcast update to local clients
+  const authorUser = db
+    .select()
+    .from(schema.users)
+    .where(eq(schema.users.id, localMsg.userId))
+    .get();
+
+  if (authorUser) {
+    const updatedPayload = buildDmMessagePayload(
+      {
+        id: localMsg.id,
+        dmChannelId: localMsg.dmChannelId,
+        userId: localMsg.userId,
+        content,
+        replyToId: localMsg.replyToId,
+        editedAt,
+        createdAt: localMsg.createdAt,
+      },
+      authorUser,
+    );
+
+    // Re-fetch reactions and attachments for the complete payload
+    const reactions = db
+      .select()
+      .from(schema.dmReactions)
+      .where(eq(schema.dmReactions.dmMessageId, localMsg.id))
+      .all();
+
+    const attachments = db
+      .select()
+      .from(schema.attachments)
+      .where(eq(schema.attachments.dmMessageId, localMsg.id))
+      .all();
+
+    updatedPayload.reactions = reactions.map(r => ({
+      id: r.id,
+      messageId: r.dmMessageId,
+      userId: r.userId,
+      emoji: r.emoji,
+      createdAt: r.createdAt,
+    }));
+
+    updatedPayload.attachments = attachments.map(a => ({
+      id: a.id,
+      messageId: a.dmMessageId ?? a.messageId ?? '',
+      filename: a.filename,
+      originalName: a.originalName,
+      mimetype: a.mimetype,
+      size: a.size,
+      thumbnailFilename: a.thumbnailFilename,
+      width: a.width,
+      height: a.height,
+      duration: a.duration,
+      createdAt: a.createdAt,
+    }));
+
+    connectionManager.sendToDmMembers(localMsg.dmChannelId, {
+      type: 'dm_message_updated',
+      message: updatedPayload,
+    });
+  }
+
+  accepted.push(event.messageId);
+}
+
+function processDeleteEvent(
+  event: FederationRelayEvent,
+  sourceInstance: string,
+  db: ReturnType<typeof getDb>,
+  accepted: string[],
+  rejected: Array<{ messageId: string; reason: string }>,
+): void {
+  const localMsg = db
+    .select()
+    .from(schema.dmMessages)
+    .where(
+      and(
+        eq(schema.dmMessages.sourceInstance, sourceInstance),
+        eq(schema.dmMessages.sourceMessageId, event.messageId),
+      ),
+    )
+    .get();
+
+  if (!localMsg) {
+    rejected.push({ messageId: event.messageId, reason: 'unknown_message' });
+    return;
+  }
+
+  // Collect attachment filenames before deletion for disk cleanup
+  const attachmentRows = db
+    .select({ filename: schema.attachments.filename })
+    .from(schema.attachments)
+    .where(eq(schema.attachments.dmMessageId, localMsg.id))
+    .all();
+
+  // Delete attachments, reactions, and message atomically
+  db.transaction((tx) => {
+    tx.delete(schema.attachments)
+      .where(eq(schema.attachments.dmMessageId, localMsg.id))
+      .run();
+    tx.delete(schema.dmReactions)
+      .where(eq(schema.dmReactions.dmMessageId, localMsg.id))
+      .run();
+    tx.delete(schema.dmMessages)
+      .where(eq(schema.dmMessages.id, localMsg.id))
+      .run();
+  });
+
+  // Clean up files from disk
+  deleteAttachmentFiles(attachmentRows);
+
+  // Broadcast deletion to local clients
+  connectionManager.sendToDmMembers(localMsg.dmChannelId, {
+    type: 'dm_message_deleted',
+    messageId: localMsg.id,
+    dmChannelId: localMsg.dmChannelId,
+  });
+
+  accepted.push(event.messageId);
+}
+
+function processReactionAddEvent(
+  event: FederationRelayEvent,
+  sourceInstance: string,
+  db: ReturnType<typeof getDb>,
+  accepted: string[],
+  rejected: Array<{ messageId: string; reason: string }>,
+): void {
+  if (!event.reaction) {
+    rejected.push({ messageId: event.messageId, reason: 'missing_reaction_payload' });
+    return;
+  }
+
+  // Find the local message corresponding to the source message
+  const localMsg = db
+    .select()
+    .from(schema.dmMessages)
+    .where(
+      and(
+        eq(schema.dmMessages.sourceInstance, sourceInstance),
+        eq(schema.dmMessages.sourceMessageId, event.messageId),
+      ),
+    )
+    .get();
+
+  if (!localMsg) {
+    rejected.push({ messageId: event.messageId, reason: 'unknown_message' });
+    return;
+  }
+
+  // Resolve the reacting user
+  const reactingUser = resolveLocalUser(event.reaction.homeUserId, db);
+  if (!reactingUser) {
+    rejected.push({ messageId: event.messageId, reason: 'user_not_found' });
+    return;
+  }
+
+  // Dedup: check if this user already reacted with this emoji
+  const existingReaction = db
+    .select()
+    .from(schema.dmReactions)
+    .where(
+      and(
+        eq(schema.dmReactions.dmMessageId, localMsg.id),
+        eq(schema.dmReactions.userId, reactingUser.id),
+        eq(schema.dmReactions.emoji, event.reaction.emoji),
+      ),
+    )
+    .get();
+
+  if (existingReaction) {
+    // Already exists — treat as accepted (idempotent)
+    accepted.push(event.messageId);
+    return;
+  }
+
+  const reactionId = generateSnowflake();
+  const now = event.reaction.createdAt || Date.now();
+
+  db.insert(schema.dmReactions)
+    .values({
+      id: reactionId,
+      dmMessageId: localMsg.id,
+      userId: reactingUser.id,
+      emoji: event.reaction.emoji,
+      createdAt: now,
+    })
+    .run();
+
+  // Broadcast to local clients
+  connectionManager.sendToDmMembers(localMsg.dmChannelId, {
+    type: 'reaction_added',
+    messageId: localMsg.id,
+    reaction: {
+      id: reactionId,
+      messageId: localMsg.id,
+      userId: reactingUser.id,
+      emoji: event.reaction.emoji,
+      createdAt: now,
+      user: sanitizeUser(reactingUser),
+    },
+  });
+
+  accepted.push(event.messageId);
+}
+
+function processReactionRemoveEvent(
+  event: FederationRelayEvent,
+  sourceInstance: string,
+  db: ReturnType<typeof getDb>,
+  accepted: string[],
+  rejected: Array<{ messageId: string; reason: string }>,
+): void {
+  if (!event.reaction) {
+    rejected.push({ messageId: event.messageId, reason: 'missing_reaction_payload' });
+    return;
+  }
+
+  // Find the local message
+  const localMsg = db
+    .select()
+    .from(schema.dmMessages)
+    .where(
+      and(
+        eq(schema.dmMessages.sourceInstance, sourceInstance),
+        eq(schema.dmMessages.sourceMessageId, event.messageId),
+      ),
+    )
+    .get();
+
+  if (!localMsg) {
+    rejected.push({ messageId: event.messageId, reason: 'unknown_message' });
+    return;
+  }
+
+  // Resolve the reacting user
+  const reactingUser = resolveLocalUser(event.reaction.homeUserId, db);
+  if (!reactingUser) {
+    rejected.push({ messageId: event.messageId, reason: 'user_not_found' });
+    return;
+  }
+
+  const result = db
+    .delete(schema.dmReactions)
+    .where(
+      and(
+        eq(schema.dmReactions.dmMessageId, localMsg.id),
+        eq(schema.dmReactions.userId, reactingUser.id),
+        eq(schema.dmReactions.emoji, event.reaction.emoji),
+      ),
+    )
+    .run();
+
+  if (result.changes > 0) {
+    connectionManager.sendToDmMembers(localMsg.dmChannelId, {
+      type: 'reaction_removed',
+      messageId: localMsg.id,
+      userId: reactingUser.id,
+      emoji: event.reaction.emoji,
+    });
+  }
+
+  accepted.push(event.messageId);
 }