refactor(server): thread undeliverable collector through processRelayEvents (#18)

Additive plumbing. No behavior change — every existing event-type path
continues to push to accepted/rejected only. Response serializes the new
bucket only when non-empty (byte-identical responses in the normal case).
Tasks 3-4 add actual undeliverable pushes for dm_call_start paths.
This commit is contained in:
Jannis Braun
2026-04-24 19:17:27 +02:00
parent 3988c5823a
commit 0057cb4d42
+11 -4
View File
@@ -1532,7 +1532,7 @@ export async function federationRoutes(app: FastifyInstance): Promise<void> {
} }
// 3. Process each event // 3. Process each event
const { accepted, rejected } = await processRelayEvents(body.events, sourceInstance, peer.origin, db); const { accepted, rejected, undeliverable } = await processRelayEvents(body.events, sourceInstance, peer.origin, db);
// 4. Update peer status // 4. Update peer status
db.update(schema.federationPeers) db.update(schema.federationPeers)
@@ -1555,6 +1555,7 @@ export async function federationRoutes(app: FastifyInstance): Promise<void> {
accepted, accepted,
rejected, rejected,
maxUploadSize: settings?.maxUploadSizeBytes ?? config.maxUploadSize, maxUploadSize: settings?.maxUploadSizeBytes ?? config.maxUploadSize,
...(undeliverable.length > 0 ? { undeliverable } : {}),
}; };
return reply.code(200).send(response); return reply.code(200).send(response);
@@ -2066,9 +2067,14 @@ export async function processRelayEvents(
sourceInstance: string, sourceInstance: string,
peerOrigin: string, peerOrigin: string,
db: ReturnType<typeof getDb>, db: ReturnType<typeof getDb>,
): Promise<{ accepted: string[]; rejected: Array<{ messageId: string; reason: string }> }> { ): Promise<{
accepted: string[];
rejected: Array<{ messageId: string; reason: string }>;
undeliverable: Array<{ messageId: string; reason: string }>;
}> {
const accepted: string[] = []; const accepted: string[] = [];
const rejected: Array<{ messageId: string; reason: string }> = []; const rejected: Array<{ messageId: string; reason: string }> = [];
const undeliverable: Array<{ messageId: string; reason: string }> = [];
for (const event of events) { for (const event of events) {
try { try {
@@ -2116,7 +2122,7 @@ export async function processRelayEvents(
processFileRejectedEvent(event, sourceInstance, db, accepted, rejected); processFileRejectedEvent(event, sourceInstance, db, accepted, rejected);
break; break;
case 'dm_call_start': case 'dm_call_start':
processDmCallStartEvent(event, sourceInstance, db, accepted, rejected); processDmCallStartEvent(event, sourceInstance, db, accepted, rejected, undeliverable);
break; break;
case 'dm_call_accept': case 'dm_call_accept':
processDmCallAcceptEvent(event, sourceInstance, db, accepted, rejected); processDmCallAcceptEvent(event, sourceInstance, db, accepted, rejected);
@@ -2156,7 +2162,7 @@ export async function processRelayEvents(
} }
} }
return { accepted, rejected }; return { accepted, rejected, undeliverable };
} }
// ─── Helpers ──────────────────────────────────────────────────────────────── // ─── Helpers ────────────────────────────────────────────────────────────────
@@ -4253,6 +4259,7 @@ function processDmCallStartEvent(
db: ReturnType<typeof getDb>, db: ReturnType<typeof getDb>,
accepted: string[], accepted: string[],
rejected: Array<{ messageId: string; reason: string }>, rejected: Array<{ messageId: string; reason: string }>,
undeliverable: Array<{ messageId: string; reason: string }>,
): void { ): void {
if (!event.call?.caller || !event.call.livekitUrl || !event.call.tokens || !event.federatedId) { if (!event.call?.caller || !event.call.livekitUrl || !event.call.tokens || !event.federatedId) {
rejected.push({ messageId: event.messageId, reason: 'missing_call_payload' }); rejected.push({ messageId: event.messageId, reason: 'missing_call_payload' });