fix(federation): add participants array to relay events and fix recipient resolution

The relay was failing because processCreateEvent relied on the friends
table to discover the DM recipient, but friendships aren't federated
across instances. Also, resolveLocalUser matched deleted replicated
users before active ones.

- Add participants[] to FederationRelayEvent with homeUserId/homeInstance
  for all DM channel members
- Add getDmParticipants() helper to look up member identities
- Include participants in outbox payloads (create/update) and sync events
- Rewrite processCreateEvent to resolve participants directly, compute
  canonicalDmPairId, and findOrCreateDmChannel — removing the entire
  friends-list fallback (60+ lines)
- Fix resolveLocalUser to filter out deleted users (is_deleted = 0)
  and prefer the replicated user match when multiple candidates exist
This commit is contained in:
Jannis Braun
2026-03-26 05:36:41 +01:00
parent cb9a70d7a6
commit cf9fcb78ed
6 changed files with 94 additions and 126 deletions
+4 -1
View File
@@ -21,7 +21,7 @@ import {
import { sanitizeUser } from '../utils/sanitize.js'; import { sanitizeUser } from '../utils/sanitize.js';
import { deleteUploadFile, deleteAttachmentFiles } from '../utils/fileCleanup.js'; import { deleteUploadFile, deleteAttachmentFiles } from '../utils/fileCleanup.js';
import { fetchDmEmbedsForMessages, resolveEmbeds, reResolveEmbeds, embedRowToEmbed } from '../utils/embedResolver.js'; import { fetchDmEmbedsForMessages, resolveEmbeds, reResolveEmbeds, embedRowToEmbed } from '../utils/embedResolver.js';
import { appendMutationLog, queueOutboxEvent, buildRelayPayload } from '../utils/federationOutbox.js'; import { appendMutationLog, queueOutboxEvent, buildRelayPayload, getDmParticipants } from '../utils/federationOutbox.js';
/** /**
* Batch-fetch reactions for a set of DM message IDs. * Batch-fetch reactions for a set of DM message IDs.
@@ -973,8 +973,10 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
// Federation: log mutation and queue for relay // Federation: log mutation and queue for relay
appendMutationLog(messageId, id, 'create'); appendMutationLog(messageId, id, 'create');
const participants = getDmParticipants(id);
queueOutboxEvent(messageId, id, 'create', JSON.stringify({ queueOutboxEvent(messageId, id, 'create', JSON.stringify({
message: { ...buildRelayPayload(message, message.user), attachments: [] }, message: { ...buildRelayPayload(message, message.user), attachments: [] },
participants,
})); }));
// Resolve embeds asynchronously after responding // Resolve embeds asynchronously after responding
@@ -1042,6 +1044,7 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
appendMutationLog(id, msg.dmChannelId, 'update'); appendMutationLog(id, msg.dmChannelId, 'update');
queueOutboxEvent(id, msg.dmChannelId, 'update', JSON.stringify({ queueOutboxEvent(id, msg.dmChannelId, 'update', JSON.stringify({
message: buildRelayPayload(updated, updated.user), message: buildRelayPayload(updated, updated.user),
participants: getDmParticipants(msg.dmChannelId),
})); }));
// Resolve new embeds asynchronously (old ones already deleted above) // Resolve new embeds asynchronously (old ones already deleted above)
+48 -120
View File
@@ -9,7 +9,7 @@ import { config } from '../config.js';
import { connectionManager } from '../ws/handler.js'; import { connectionManager } from '../ws/handler.js';
import { sanitizeUser } from '../utils/sanitize.js'; import { sanitizeUser } from '../utils/sanitize.js';
import { deleteAttachmentFiles } from '../utils/fileCleanup.js'; import { deleteAttachmentFiles } from '../utils/fileCleanup.js';
import { canonicalDmPairId } from '../utils/federationOutbox.js'; import { canonicalDmPairId, getDmParticipants } from '../utils/federationOutbox.js';
import { broadcastDmMessage } from './dm.js'; import { broadcastDmMessage } from './dm.js';
import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent, FederationRelayAttachment, FederationSyncRequest, FederationSyncResponse, DmMessageWithUser } from '@backspace/shared'; import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent, FederationRelayAttachment, FederationSyncRequest, FederationSyncResponse, DmMessageWithUser } from '@backspace/shared';
@@ -748,6 +748,7 @@ export async function federationRoutes(app: FastifyInstance): Promise<void> {
messageId: message.id, messageId: message.id,
encryptionVersion: 0, encryptionVersion: 0,
timestamp: mutation.mutated_at, timestamp: mutation.mutated_at,
participants: getDmParticipants(mutation.dm_channel_id),
message: { message: {
userId: message.userId, userId: message.userId,
homeUserId, homeUserId,
@@ -795,16 +796,25 @@ function resolveLocalUser(
homeUserId: string, homeUserId: string,
db: ReturnType<typeof getDb>, db: ReturnType<typeof getDb>,
): typeof schema.users.$inferSelect | undefined { ): typeof schema.users.$inferSelect | undefined {
return db const candidates = db
.select() .select()
.from(schema.users) .from(schema.users)
.where( .where(
and(
or( or(
eq(schema.users.homeUserId, homeUserId), eq(schema.users.homeUserId, homeUserId),
and(eq(schema.users.id, homeUserId), isNull(schema.users.homeInstance)), and(eq(schema.users.id, homeUserId), isNull(schema.users.homeInstance)),
), ),
eq(schema.users.isDeleted, 0),
),
) )
.get(); .all();
// Prefer non-deleted active users; if multiple, prefer the one with homeUserId set
// (replicated user) over a local user match
if (candidates.length === 0) return undefined;
if (candidates.length === 1) return candidates[0];
return candidates.find(u => u.homeUserId === homeUserId) ?? candidates[0];
} }
/** /**
@@ -933,10 +943,8 @@ function processCreateEvent(
return; return;
} }
// Resolve the message author to a local user if (!event.participants || event.participants.length < 2) {
const authorUser = resolveLocalUser(event.message.homeUserId, db); rejected.push({ messageId: event.messageId, reason: 'missing_participants' });
if (!authorUser) {
rejected.push({ messageId: event.messageId, reason: 'user_not_found' });
return; return;
} }
@@ -957,126 +965,46 @@ function processCreateEvent(
return; return;
} }
// Resolve the DM recipient. Federated DMs are 1-on-1: one side is the author, // Resolve ALL participants to local users
// the other is a local user on this instance. We match via canonical_pair_id. const resolvedParticipants: Array<{
const authorHomeUserId = event.message.homeUserId; localUser: typeof schema.users.$inferSelect;
homeUserId: string;
}> = [];
// First, search existing DM channels where the author is already a member for (const p of event.participants) {
const authorMemberships = db const localUser = resolveLocalUser(p.homeUserId, db);
.select({ dmChannelId: schema.dmMembers.dmChannelId }) if (localUser) {
.from(schema.dmMembers) resolvedParticipants.push({ localUser, homeUserId: p.homeUserId });
.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) { if (resolvedParticipants.length < 2) {
rejected.push({ messageId: event.messageId, reason: 'recipient_not_found' }); rejected.push({ messageId: event.messageId, reason: 'participant_not_found' });
return; return;
} }
// Find the author among the resolved participants
const authorEntry = resolvedParticipants.find(
p => p.homeUserId === event.message!.homeUserId,
);
if (!authorEntry) {
rejected.push({ messageId: event.messageId, reason: 'author_not_found' });
return;
}
const authorUser = authorEntry.localUser;
// Compute canonical pair ID from participants' home user IDs and find/create channel
const pairId = canonicalDmPairId(
resolvedParticipants[0]!.homeUserId,
resolvedParticipants[1]!.homeUserId,
);
const localDmChannelId = findOrCreateDmChannel(
pairId,
resolvedParticipants[0]!.localUser.id,
resolvedParticipants[1]!.localUser.id,
db,
);
// Insert the message // Insert the message
const localMessageId = generateSnowflake(); const localMessageId = generateSnowflake();
db.insert(schema.dmMessages) db.insert(schema.dmMessages)
+28 -1
View File
@@ -3,7 +3,7 @@ import * as schema from '../db/schema.js';
import { eq, and } from 'drizzle-orm'; import { eq, and } from 'drizzle-orm';
import { generateSnowflake } from './snowflake.js'; import { generateSnowflake } from './snowflake.js';
import crypto from 'node:crypto'; import crypto from 'node:crypto';
import type { FederationRelayEvent } from '@backspace/shared'; import type { FederationRelayEvent, FederationRelayParticipant } from '@backspace/shared';
import { config } from '../config.js'; import { config } from '../config.js';
// ─── Settings Cache ────────────────────────────────────────────────────────── // ─── Settings Cache ──────────────────────────────────────────────────────────
@@ -210,6 +210,33 @@ export function canonicalDmPairId(homeUserIdA: string, homeUserIdB: string): str
return crypto.createHash('sha256').update(sorted.join(':')).digest('hex').slice(0, 32); return crypto.createHash('sha256').update(sorted.join(':')).digest('hex').slice(0, 32);
} }
/**
* Look up all members of a DM channel and return their federated identities.
* Used to include participants in relay events so the receiving instance can
* resolve both parties without relying on the friends list.
*/
export function getDmParticipants(dmChannelId: string): FederationRelayParticipant[] {
const db = getDb();
const members = db
.select({
userId: schema.dmMembers.userId,
homeUserId: schema.users.homeUserId,
homeInstance: schema.users.homeInstance,
id: schema.users.id,
})
.from(schema.dmMembers)
.innerJoin(schema.users, eq(schema.dmMembers.userId, schema.users.id))
.where(eq(schema.dmMembers.dmChannelId, dmChannelId))
.all();
const domainOrigin = config.domain ? `https://${config.domain}` : '';
return members.map(m => ({
homeUserId: m.homeUserId || m.id,
homeInstance: m.homeInstance || domainOrigin,
}));
}
/** /**
* Build the relay payload object for a DM message. * Build the relay payload object for a DM message.
* The caller may augment the returned object with attachments before serialization. * The caller may augment the returned object with attachments before serialization.
@@ -172,6 +172,7 @@ async function processOutboxTick(): Promise<void> {
messageId: entry.messageId, messageId: entry.messageId,
encryptionVersion: (entry.encryptionVersion ?? 0) as 0, encryptionVersion: (entry.encryptionVersion ?? 0) as 0,
timestamp: entry.createdAt, timestamp: entry.createdAt,
...(parsed.participants ? { participants: parsed.participants } : {}),
...(parsed.message ? { message: parsed.message } : {}), ...(parsed.message ? { message: parsed.message } : {}),
...(parsed.reactions ? { reactions: parsed.reactions } : {}), ...(parsed.reactions ? { reactions: parsed.reactions } : {}),
...(parsed.reaction ? { reaction: parsed.reaction } : {}), ...(parsed.reaction ? { reaction: parsed.reaction } : {}),
+4 -1
View File
@@ -11,7 +11,7 @@ import { ACTIVITY_LIMITS } from '@backspace/shared/src/activities.js';
import { sanitizeUser } from '../utils/sanitize.js'; import { sanitizeUser } from '../utils/sanitize.js';
import { deleteAttachmentFiles } from '../utils/fileCleanup.js'; import { deleteAttachmentFiles } from '../utils/fileCleanup.js';
import { resolveEmbeds, reResolveEmbeds, embedRowToEmbed } from '../utils/embedResolver.js'; import { resolveEmbeds, reResolveEmbeds, embedRowToEmbed } from '../utils/embedResolver.js';
import { appendMutationLog, queueOutboxEvent, buildRelayPayload } from '../utils/federationOutbox.js'; import { appendMutationLog, queueOutboxEvent, buildRelayPayload, getDmParticipants } from '../utils/federationOutbox.js';
/** /**
* Re-evaluate SPEAK permission for all participants in voice channels * Re-evaluate SPEAK permission for all participants in voice channels
@@ -888,8 +888,10 @@ function handleDmMessageCreate(event: Record<string, unknown>, userId: string):
// Federation: log mutation and queue for relay // Federation: log mutation and queue for relay
appendMutationLog(messageId, dmChannelId, 'create'); appendMutationLog(messageId, dmChannelId, 'create');
const participants = getDmParticipants(dmChannelId);
queueOutboxEvent(messageId, dmChannelId, 'create', JSON.stringify({ queueOutboxEvent(messageId, dmChannelId, 'create', JSON.stringify({
message: { ...buildRelayPayload(dmMessage, dmMessage.user), attachments: [] }, message: { ...buildRelayPayload(dmMessage, dmMessage.user), attachments: [] },
participants,
})); }));
// Resolve embeds asynchronously // Resolve embeds asynchronously
@@ -994,6 +996,7 @@ function handleDmMessageEdit(event: Record<string, unknown>, userId: string): vo
appendMutationLog(messageId, msg.dmChannelId, 'update'); appendMutationLog(messageId, msg.dmChannelId, 'update');
queueOutboxEvent(messageId, msg.dmChannelId, 'update', JSON.stringify({ queueOutboxEvent(messageId, msg.dmChannelId, 'update', JSON.stringify({
message: buildRelayPayload(updated, updated.user), message: buildRelayPayload(updated, updated.user),
participants: getDmParticipants(msg.dmChannelId),
})); }));
// Resolve new embeds asynchronously (old ones already deleted above) // Resolve new embeds asynchronously (old ones already deleted above)
+6
View File
@@ -717,12 +717,18 @@ export interface AdminResetPasswordResponse {
// ─── Federation Relay Types ────────────────────────────────────────────────── // ─── Federation Relay Types ──────────────────────────────────────────────────
export interface FederationRelayParticipant {
homeUserId: string;
homeInstance: string;
}
export interface FederationRelayEvent { export interface FederationRelayEvent {
eventType: 'create' | 'update' | 'delete' | 'reaction_add' | 'reaction_remove'; eventType: 'create' | 'update' | 'delete' | 'reaction_add' | 'reaction_remove';
dmChannelId: string; dmChannelId: string;
messageId: string; messageId: string;
encryptionVersion: 0; encryptionVersion: 0;
timestamp: number; timestamp: number;
participants?: FederationRelayParticipant[];
message?: { message?: {
userId: string; userId: string;
homeUserId: string; homeUserId: string;