fix(federation): branching message lookup in reaction relay processors
Uses messageHomeInstance to determine whether the target message originated locally (lookup by ID) or was replicated from another instance (lookup by sourceInstance + sourceMessageId). Supports N-instance federation and falls back to relay-sender lookup for backward compatibility with in-flight outbox entries.
This commit is contained in:
@@ -2,7 +2,7 @@ import type { FastifyInstance } from 'fastify';
|
|||||||
import { randomBytes } from 'node:crypto';
|
import { randomBytes } from 'node:crypto';
|
||||||
import { eq, and, or, isNull } from 'drizzle-orm';
|
import { eq, and, or, isNull } from 'drizzle-orm';
|
||||||
import { authenticate, requireAdmin } from '../utils/auth.js';
|
import { authenticate, requireAdmin } from '../utils/auth.js';
|
||||||
import { generateHmacSecret, parseFederationHeaders, verifySignature } from '../utils/federationAuth.js';
|
import { generateHmacSecret, getOurOrigin, parseFederationHeaders, verifySignature } from '../utils/federationAuth.js';
|
||||||
import { generateSnowflake } from '../utils/snowflake.js';
|
import { generateSnowflake } from '../utils/snowflake.js';
|
||||||
import { getDb, getRawDb, schema } from '../db/index.js';
|
import { getDb, getRawDb, schema } from '../db/index.js';
|
||||||
import { config } from '../config.js';
|
import { config } from '../config.js';
|
||||||
@@ -1259,6 +1259,44 @@ function processDeleteEvent(
|
|||||||
accepted.push(event.messageId);
|
accepted.push(event.messageId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Resolve a local DM message from a federation relay event's canonical identity.
|
||||||
|
* Uses messageHomeInstance to branch the lookup:
|
||||||
|
* - If the message originated on THIS instance → find by local ID
|
||||||
|
* - Otherwise → find by sourceInstance + sourceMessageId tracking
|
||||||
|
* Falls back to relay sender origin when messageHomeInstance is absent (backward compat).
|
||||||
|
*/
|
||||||
|
function resolveLocalDmMessage(
|
||||||
|
canonicalMessageId: string,
|
||||||
|
messageHomeInstance: string | undefined,
|
||||||
|
sourceInstance: string,
|
||||||
|
db: ReturnType<typeof getDb>,
|
||||||
|
): typeof schema.dmMessages.$inferSelect | undefined {
|
||||||
|
if (messageHomeInstance && messageHomeInstance === getOurOrigin()) {
|
||||||
|
return db
|
||||||
|
.select()
|
||||||
|
.from(schema.dmMessages)
|
||||||
|
.where(
|
||||||
|
and(
|
||||||
|
eq(schema.dmMessages.id, canonicalMessageId),
|
||||||
|
isNull(schema.dmMessages.sourceInstance),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
.get();
|
||||||
|
}
|
||||||
|
const originInstance = messageHomeInstance || sourceInstance;
|
||||||
|
return db
|
||||||
|
.select()
|
||||||
|
.from(schema.dmMessages)
|
||||||
|
.where(
|
||||||
|
and(
|
||||||
|
eq(schema.dmMessages.sourceInstance, originInstance),
|
||||||
|
eq(schema.dmMessages.sourceMessageId, canonicalMessageId),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
.get();
|
||||||
|
}
|
||||||
|
|
||||||
function processReactionAddEvent(
|
function processReactionAddEvent(
|
||||||
event: FederationRelayEvent,
|
event: FederationRelayEvent,
|
||||||
sourceInstance: string,
|
sourceInstance: string,
|
||||||
@@ -1271,19 +1309,13 @@ function processReactionAddEvent(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Find the local message — use the actual message ID from the reaction payload,
|
const canonicalMessageId = event.reaction.messageId ?? event.messageId;
|
||||||
// not event.messageId which is the outbox dedup key (reactionId)
|
const localMsg = resolveLocalDmMessage(
|
||||||
const sourceMessageId = event.reaction.messageId ?? event.messageId;
|
canonicalMessageId,
|
||||||
const localMsg = db
|
event.reaction.messageHomeInstance,
|
||||||
.select()
|
sourceInstance,
|
||||||
.from(schema.dmMessages)
|
db,
|
||||||
.where(
|
);
|
||||||
and(
|
|
||||||
eq(schema.dmMessages.sourceInstance, sourceInstance),
|
|
||||||
eq(schema.dmMessages.sourceMessageId, sourceMessageId),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
.get();
|
|
||||||
|
|
||||||
if (!localMsg) {
|
if (!localMsg) {
|
||||||
rejected.push({ messageId: event.messageId, reason: 'unknown_message' });
|
rejected.push({ messageId: event.messageId, reason: 'unknown_message' });
|
||||||
@@ -1358,19 +1390,13 @@ function processReactionRemoveEvent(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Find the local message — use the actual message ID from the reaction payload,
|
const canonicalMessageId = event.reaction.messageId ?? event.messageId;
|
||||||
// not event.messageId which is the outbox dedup key (composite string)
|
const localMsg = resolveLocalDmMessage(
|
||||||
const sourceMessageId = event.reaction.messageId ?? event.messageId;
|
canonicalMessageId,
|
||||||
const localMsg = db
|
event.reaction.messageHomeInstance,
|
||||||
.select()
|
sourceInstance,
|
||||||
.from(schema.dmMessages)
|
db,
|
||||||
.where(
|
);
|
||||||
and(
|
|
||||||
eq(schema.dmMessages.sourceInstance, sourceInstance),
|
|
||||||
eq(schema.dmMessages.sourceMessageId, sourceMessageId),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
.get();
|
|
||||||
|
|
||||||
if (!localMsg) {
|
if (!localMsg) {
|
||||||
rejected.push({ messageId: event.messageId, reason: 'unknown_message' });
|
rejected.push({ messageId: event.messageId, reason: 'unknown_message' });
|
||||||
|
|||||||
Reference in New Issue
Block a user