feat(federation): hook outbox and mutation log into DM message and reaction handlers

Wire appendMutationLog + queueOutboxEvent + buildRelayPayload into all
DM mutation paths so federation peers receive relay events:

- REST: POST /api/dm/:id/messages, PATCH /api/dm/messages/:id,
  DELETE /api/dm/messages/:id
- WebSocket: dm_message_create, dm_message_edit, dm_message_delete,
  reaction_add (DM path), reaction_remove (DM path)
- Fix buildRelayPayload parameter types to accept optional replyToId
  and editedAt (matching DmMessageWithUser's optional fields)
This commit is contained in:
Jannis Braun
2026-03-25 21:14:34 +01:00
parent 88ad1c676d
commit 32a0c2e618
3 changed files with 74 additions and 4 deletions
+17
View File
@@ -21,6 +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';
/** /**
* Batch-fetch reactions for a set of DM message IDs. * Batch-fetch reactions for a set of DM message IDs.
@@ -970,6 +971,12 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
// Broadcast to all DM members (including those who closed the channel) // Broadcast to all DM members (including those who closed the channel)
broadcastDmMessage(id, message); broadcastDmMessage(id, message);
// Federation: log mutation and queue for relay
appendMutationLog(messageId, id, 'create');
queueOutboxEvent(messageId, id, 'create', JSON.stringify({
message: { ...buildRelayPayload(message, message.user), attachments: [] },
}));
// Resolve embeds asynchronously after responding // Resolve embeds asynchronously after responding
setImmediate(() => { setImmediate(() => {
resolveEmbeds(messageId, content?.trim() || null, id, true, null).catch(() => {}); resolveEmbeds(messageId, content?.trim() || null, id, true, null).catch(() => {});
@@ -1031,6 +1038,12 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
}); });
} }
// Federation: log mutation and queue for relay
appendMutationLog(id, msg.dmChannelId, 'update');
queueOutboxEvent(id, msg.dmChannelId, 'update', JSON.stringify({
message: buildRelayPayload(updated, updated.user),
}));
// Resolve new embeds asynchronously (old ones already deleted above) // Resolve new embeds asynchronously (old ones already deleted above)
setImmediate(() => { setImmediate(() => {
resolveEmbeds(id, content.trim(), msg.dmChannelId, true, null).catch(() => {}); resolveEmbeds(id, content.trim(), msg.dmChannelId, true, null).catch(() => {});
@@ -1093,6 +1106,10 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
}); });
} }
// Federation: log mutation and queue for relay
appendMutationLog(id, msg.dmChannelId, 'delete');
queueOutboxEvent(id, msg.dmChannelId, 'delete', JSON.stringify({ deleted: true }));
return reply.code(200).send({ success: true }); return reply.code(200).send({ success: true });
}); });
} }
@@ -217,8 +217,8 @@ export function buildRelayPayload(
message: { message: {
id: string; id: string;
content: string | null; content: string | null;
replyToId: string | null; replyToId?: string | null;
editedAt: number | null; editedAt?: number | null;
createdAt: number; createdAt: number;
}, },
user: { user: {
@@ -232,8 +232,8 @@ export function buildRelayPayload(
homeUserId: user.homeUserId || user.id, homeUserId: user.homeUserId || user.id,
homeInstance: user.homeInstance || '', homeInstance: user.homeInstance || '',
content: message.content, content: message.content,
replyToId: message.replyToId, replyToId: message.replyToId ?? null,
editedAt: message.editedAt, editedAt: message.editedAt ?? null,
createdAt: message.createdAt, createdAt: message.createdAt,
}; };
} }
+53
View File
@@ -11,6 +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';
/** /**
* Re-evaluate SPEAK permission for all participants in voice channels * Re-evaluate SPEAK permission for all participants in voice channels
@@ -885,6 +886,12 @@ function handleDmMessageCreate(event: Record<string, unknown>, userId: string):
// Broadcast to all DM members (including those who closed the channel) // Broadcast to all DM members (including those who closed the channel)
broadcastDmMessage(dmChannelId, dmMessage); broadcastDmMessage(dmChannelId, dmMessage);
// Federation: log mutation and queue for relay
appendMutationLog(messageId, dmChannelId, 'create');
queueOutboxEvent(messageId, dmChannelId, 'create', JSON.stringify({
message: { ...buildRelayPayload(dmMessage, dmMessage.user), attachments: [] },
}));
// Resolve embeds asynchronously // Resolve embeds asynchronously
setImmediate(() => { setImmediate(() => {
resolveEmbeds(messageId, hasContent ? content!.trim() : null, dmChannelId, true, null).catch(() => {}); resolveEmbeds(messageId, hasContent ? content!.trim() : null, dmChannelId, true, null).catch(() => {});
@@ -983,6 +990,12 @@ function handleDmMessageEdit(event: Record<string, unknown>, userId: string): vo
}); });
} }
// Federation: log mutation and queue for relay
appendMutationLog(messageId, msg.dmChannelId, 'update');
queueOutboxEvent(messageId, msg.dmChannelId, 'update', JSON.stringify({
message: buildRelayPayload(updated, updated.user),
}));
// Resolve new embeds asynchronously (old ones already deleted above) // Resolve new embeds asynchronously (old ones already deleted above)
setImmediate(() => { setImmediate(() => {
resolveEmbeds(messageId, content.trim(), msg.dmChannelId, true, null).catch(() => {}); resolveEmbeds(messageId, content.trim(), msg.dmChannelId, true, null).catch(() => {});
@@ -1041,6 +1054,10 @@ function handleDmMessageDelete(event: Record<string, unknown>, userId: string):
dmChannelId: msg.dmChannelId, dmChannelId: msg.dmChannelId,
}); });
} }
// Federation: log mutation and queue for relay
appendMutationLog(messageId, msg.dmChannelId, 'delete');
queueOutboxEvent(messageId, msg.dmChannelId, 'delete', JSON.stringify({ deleted: true }));
} }
// ─── Reaction Handlers ───────────────────────────────────────────────────── // ─── Reaction Handlers ─────────────────────────────────────────────────────
@@ -1114,6 +1131,22 @@ function handleReactionAdd(event: Record<string, unknown>, userId: string): void
messageId, messageId,
reaction: { id: reactionId, messageId, userId, emoji, createdAt: now, user: userObj }, reaction: { id: reactionId, messageId, userId, emoji, createdAt: now, user: userObj },
}); });
// Federation: log reaction mutation and queue for relay
appendMutationLog(messageId, dmMsg.dmChannelId, 'reaction_add', JSON.stringify({
userId,
homeUserId: reactionUser?.homeUserId || userId,
emoji,
createdAt: now,
}));
queueOutboxEvent(reactionId, dmMsg.dmChannelId, 'reaction_add', JSON.stringify({
reaction: {
userId,
homeUserId: reactionUser?.homeUserId || userId,
emoji,
createdAt: now,
},
}));
} catch (err) { } catch (err) {
// Unique constraint violation (already reacted) // Unique constraint violation (already reacted)
} }
@@ -1171,6 +1204,26 @@ function handleReactionRemove(event: Record<string, unknown>, userId: string): v
userId, userId,
emoji, emoji,
}); });
// Federation: log reaction removal and queue for relay
const removingUser = db.select().from(schema.users).where(eq(schema.users.id, userId)).get();
appendMutationLog(messageId, dmMsg.dmChannelId, 'reaction_remove', JSON.stringify({
userId,
homeUserId: removingUser?.homeUserId || userId,
emoji,
}));
queueOutboxEvent(
`${messageId}:${userId}:${emoji}`,
dmMsg.dmChannelId,
'reaction_remove',
JSON.stringify({
reaction: {
userId,
homeUserId: removingUser?.homeUserId || userId,
emoji,
},
}),
);
} }
} }