896 lines
29 KiB
TypeScript
896 lines
29 KiB
TypeScript
import { getDb } from '../db/index.js';
|
|
import * as schema from '../db/schema.js';
|
|
import { eq, and, lte, asc, inArray } from 'drizzle-orm';
|
|
import { config } from '../config.js';
|
|
import { isFederationRelayEnabled, queueOutboxEvent } from './federationOutbox.js';
|
|
import { runFederationJanitor } from './storageJanitor.js';
|
|
import { buildFederationHeaders, getOurOrigin } from './federationAuth.js';
|
|
import { generateSnowflake } from './snowflake.js';
|
|
import { getDmMessageWithUser } from '../routes/dm.js';
|
|
import { connectionManager } from '../ws/handler.js';
|
|
import { generateThumbnail } from './thumbnail.js';
|
|
import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent } from '@backspace/shared';
|
|
import fs from 'node:fs';
|
|
import path from 'node:path';
|
|
import crypto from 'node:crypto';
|
|
import { pipeline } from 'node:stream/promises';
|
|
import { Readable } from 'node:stream';
|
|
|
|
// ─── Constants ──────────────────────────────────────────────────────────────
|
|
|
|
const OUTBOX_INTERVAL_MS = 10_000; // 10 seconds
|
|
const FILE_QUEUE_INTERVAL_MS = 30_000; // 30 seconds
|
|
const HEALTH_CHECK_INTERVAL_MS = 3_600_000; // 1 hour
|
|
const JANITOR_INTERVAL_MS = 3_600_000; // 1 hour
|
|
|
|
const OUTBOX_BATCH_LIMIT = 50;
|
|
const FILE_QUEUE_BATCH_LIMIT = 5;
|
|
|
|
const OUTBOX_FETCH_TIMEOUT_MS = 30_000;
|
|
const FILE_DOWNLOAD_TIMEOUT_MS = 60_000;
|
|
const HEALTH_CHECK_TIMEOUT_MS = 10_000;
|
|
|
|
/** Exponential backoff schedule by attempt number (1-indexed). Values in milliseconds. */
|
|
const BACKOFF_SCHEDULE_MS: readonly number[] = [
|
|
30_000, // attempt 1: 30s
|
|
60_000, // attempt 2: 1min
|
|
300_000, // attempt 3: 5min
|
|
900_000, // attempt 4: 15min
|
|
3_600_000, // attempt 5: 1hr
|
|
21_600_000, // attempt 6: 6hr
|
|
86_400_000, // attempt 7+: 24hr cap
|
|
];
|
|
|
|
const MAX_FILE_ATTEMPTS = 10;
|
|
const PEER_UNREACHABLE_THRESHOLD = 10;
|
|
|
|
// ─── Worker State ───────────────────────────────────────────────────────────
|
|
|
|
let outboxTimer: ReturnType<typeof setTimeout> | null = null;
|
|
let fileQueueTimer: ReturnType<typeof setTimeout> | null = null;
|
|
let healthCheckTimer: ReturnType<typeof setTimeout> | null = null;
|
|
let janitorTimer: ReturnType<typeof setTimeout> | null = null;
|
|
|
|
let outboxAbortController: AbortController | null = null;
|
|
let fileQueueAbortController: AbortController | null = null;
|
|
let healthCheckAbortController: AbortController | null = null;
|
|
|
|
// ─── Helpers ────────────────────────────────────────────────────────────────
|
|
|
|
/**
|
|
* Compute the backoff delay for a given attempt number (1-indexed).
|
|
* Caps at the last entry in BACKOFF_SCHEDULE_MS.
|
|
*/
|
|
function getBackoffMs(attempt: number): number {
|
|
const index = Math.min(attempt - 1, BACKOFF_SCHEDULE_MS.length - 1);
|
|
return BACKOFF_SCHEDULE_MS[Math.max(0, index)] ?? 86_400_000;
|
|
}
|
|
|
|
/**
|
|
* Get the effective max upload size from instance settings or config fallback.
|
|
*/
|
|
function getMaxUploadSize(): number {
|
|
try {
|
|
const db = getDb();
|
|
const row = db
|
|
.select({ maxUploadSizeBytes: schema.instanceSettings.maxUploadSizeBytes })
|
|
.from(schema.instanceSettings)
|
|
.where(eq(schema.instanceSettings.id, 1))
|
|
.get();
|
|
return row?.maxUploadSizeBytes ?? config.maxUploadSize;
|
|
} catch {
|
|
return config.maxUploadSize;
|
|
}
|
|
}
|
|
|
|
// ─── Outbox Delivery Worker ─────────────────────────────────────────────────
|
|
|
|
function scheduleOutboxTick(): void {
|
|
outboxTimer = setTimeout(() => {
|
|
processOutboxTick().catch((err) => {
|
|
console.error('[federation-worker] Outbox tick error:', err);
|
|
}).finally(() => {
|
|
scheduleOutboxTick();
|
|
});
|
|
}, OUTBOX_INTERVAL_MS);
|
|
}
|
|
|
|
async function processOutboxTick(): Promise<void> {
|
|
if (!isFederationRelayEnabled()) {
|
|
return;
|
|
}
|
|
|
|
const db = getDb();
|
|
const now = Date.now();
|
|
|
|
// Fetch outbox entries ready for delivery, joined with active peers
|
|
const entries = db
|
|
.select({
|
|
outboxId: schema.federationOutbox.id,
|
|
peerId: schema.federationOutbox.peerId,
|
|
contextId: schema.federationOutbox.contextId,
|
|
entityId: schema.federationOutbox.entityId,
|
|
contextType: schema.federationOutbox.contextType,
|
|
eventType: schema.federationOutbox.eventType,
|
|
payload: schema.federationOutbox.payload,
|
|
encryptionVersion: schema.federationOutbox.encryptionVersion,
|
|
attempts: schema.federationOutbox.attempts,
|
|
createdAt: schema.federationOutbox.createdAt,
|
|
peerOrigin: schema.federationPeers.origin,
|
|
peerHmacSecret: schema.federationPeers.hmacSecret,
|
|
peerStatus: schema.federationPeers.status,
|
|
})
|
|
.from(schema.federationOutbox)
|
|
.innerJoin(
|
|
schema.federationPeers,
|
|
eq(schema.federationOutbox.peerId, schema.federationPeers.id),
|
|
)
|
|
.where(
|
|
and(
|
|
lte(schema.federationOutbox.nextRetryAt, now),
|
|
eq(schema.federationPeers.status, 'active'),
|
|
),
|
|
)
|
|
.orderBy(asc(schema.federationOutbox.createdAt))
|
|
.limit(OUTBOX_BATCH_LIMIT)
|
|
.all();
|
|
|
|
if (entries.length === 0) {
|
|
return;
|
|
}
|
|
|
|
// Group by peer
|
|
const byPeer = new Map<string, typeof entries>();
|
|
for (const entry of entries) {
|
|
const group = byPeer.get(entry.peerId);
|
|
if (group) {
|
|
group.push(entry);
|
|
} else {
|
|
byPeer.set(entry.peerId, [entry]);
|
|
}
|
|
}
|
|
|
|
const ourOrigin = getOurOrigin();
|
|
|
|
for (const [peerId, peerEntries] of byPeer) {
|
|
const firstEntry = peerEntries[0];
|
|
if (!firstEntry) continue; // Should never happen given grouping logic above
|
|
|
|
const peerOrigin = firstEntry.peerOrigin;
|
|
const peerHmacSecret = firstEntry.peerHmacSecret;
|
|
|
|
// Build relay events from outbox entries
|
|
const events: FederationRelayEvent[] = peerEntries.map((entry) => {
|
|
const parsed = JSON.parse(entry.payload) as Partial<FederationRelayEvent>;
|
|
const isDm = entry.contextType === 'dm' || !entry.contextType;
|
|
const evt: FederationRelayEvent = {
|
|
eventType: entry.eventType as FederationRelayEvent['eventType'],
|
|
contextType: (entry.contextType ?? 'dm') as 'dm' | 'friend',
|
|
messageId: entry.entityId ?? '',
|
|
encryptionVersion: (entry.encryptionVersion ?? 0) as 0,
|
|
timestamp: entry.createdAt,
|
|
};
|
|
if (isDm && entry.contextId) evt.dmChannelId = entry.contextId;
|
|
if (parsed.participants) evt.participants = parsed.participants;
|
|
if (parsed.message) evt.message = parsed.message;
|
|
if (parsed.reactions) evt.reactions = parsed.reactions;
|
|
if (parsed.reaction) evt.reaction = parsed.reaction;
|
|
if (parsed.membership) evt.membership = parsed.membership;
|
|
if (parsed.ownership) evt.ownership = parsed.ownership;
|
|
if (parsed.group) evt.group = parsed.group;
|
|
if (parsed.friendship) evt.friendship = parsed.friendship;
|
|
return evt;
|
|
});
|
|
|
|
const request: FederationRelayRequest = {
|
|
version: 1,
|
|
sourceInstance: ourOrigin,
|
|
events,
|
|
};
|
|
|
|
const bodyString = JSON.stringify(request);
|
|
const headers = buildFederationHeaders(bodyString, peerHmacSecret, ourOrigin);
|
|
|
|
// Create an abort controller for this specific request
|
|
outboxAbortController = new AbortController();
|
|
|
|
try {
|
|
const response = await fetch(`${peerOrigin}/api/federation/relay`, {
|
|
method: 'POST',
|
|
headers,
|
|
body: bodyString,
|
|
signal: AbortSignal.any([
|
|
outboxAbortController.signal,
|
|
AbortSignal.timeout(OUTBOX_FETCH_TIMEOUT_MS),
|
|
]),
|
|
});
|
|
|
|
if (response.ok) {
|
|
const result = await response.json() as FederationRelayResponse;
|
|
|
|
// Delete accepted entries
|
|
if (result.accepted.length > 0) {
|
|
// Map accepted messageIds to outbox IDs
|
|
const acceptedSet = new Set(result.accepted);
|
|
const acceptedOutboxIds = peerEntries
|
|
.filter((e) => acceptedSet.has(e.entityId))
|
|
.map((e) => e.outboxId);
|
|
|
|
if (acceptedOutboxIds.length > 0) {
|
|
db.delete(schema.federationOutbox)
|
|
.where(inArray(schema.federationOutbox.id, acceptedOutboxIds))
|
|
.run();
|
|
}
|
|
}
|
|
|
|
// Log rejected entries (they remain in outbox for retry)
|
|
for (const rejection of result.rejected) {
|
|
console.warn(
|
|
`[federation-worker] Peer ${peerOrigin} rejected message ${rejection.messageId}: ${rejection.reason}`,
|
|
);
|
|
}
|
|
|
|
// Store the peer's max upload size for informational display
|
|
if (typeof result.maxUploadSize === 'number') {
|
|
db.update(schema.federationPeers)
|
|
.set({ remoteMaxUploadSize: result.maxUploadSize })
|
|
.where(eq(schema.federationPeers.origin, peerOrigin))
|
|
.run();
|
|
}
|
|
|
|
// Update peer health
|
|
db.update(schema.federationPeers)
|
|
.set({
|
|
lastSeenAt: now,
|
|
consecutiveFailures: 0,
|
|
})
|
|
.where(eq(schema.federationPeers.id, peerId))
|
|
.run();
|
|
} else {
|
|
console.warn(
|
|
`[federation-worker] Peer ${peerOrigin} returned HTTP ${response.status}`,
|
|
);
|
|
handleOutboxDeliveryFailure(db, peerId, peerEntries, now);
|
|
}
|
|
} catch (err) {
|
|
if (err instanceof DOMException && err.name === 'AbortError') {
|
|
// Worker is stopping — don't update anything
|
|
return;
|
|
}
|
|
console.error(
|
|
`[federation-worker] Failed to deliver to peer ${peerOrigin}:`,
|
|
err instanceof Error ? err.message : err,
|
|
);
|
|
handleOutboxDeliveryFailure(db, peerId, peerEntries, now);
|
|
}
|
|
}
|
|
}
|
|
|
|
function handleOutboxDeliveryFailure(
|
|
db: ReturnType<typeof getDb>,
|
|
peerId: string,
|
|
entries: Array<{ outboxId: string; attempts: number | null }>,
|
|
now: number,
|
|
): void {
|
|
// Increment attempts and compute next retry for each entry
|
|
for (const entry of entries) {
|
|
const newAttempts = (entry.attempts ?? 0) + 1;
|
|
const backoffMs = getBackoffMs(newAttempts);
|
|
|
|
db.update(schema.federationOutbox)
|
|
.set({
|
|
attempts: newAttempts,
|
|
nextRetryAt: now + backoffMs,
|
|
})
|
|
.where(eq(schema.federationOutbox.id, entry.outboxId))
|
|
.run();
|
|
}
|
|
|
|
// Update peer failure tracking
|
|
const peer = db
|
|
.select({ consecutiveFailures: schema.federationPeers.consecutiveFailures })
|
|
.from(schema.federationPeers)
|
|
.where(eq(schema.federationPeers.id, peerId))
|
|
.get();
|
|
|
|
const newFailures = (peer?.consecutiveFailures ?? 0) + 1;
|
|
const updates: Record<string, number | string> = {
|
|
lastFailureAt: now,
|
|
consecutiveFailures: newFailures,
|
|
};
|
|
|
|
if (newFailures >= PEER_UNREACHABLE_THRESHOLD) {
|
|
(updates as Record<string, number | string>)['status'] = 'unreachable';
|
|
console.warn(
|
|
`[federation-worker] Peer ${peerId} marked unreachable after ${newFailures} consecutive failures`,
|
|
);
|
|
}
|
|
|
|
db.update(schema.federationPeers)
|
|
.set(updates)
|
|
.where(eq(schema.federationPeers.id, peerId))
|
|
.run();
|
|
}
|
|
|
|
// ─── File Queue Download Worker ─────────────────────────────────────────────
|
|
|
|
function scheduleFileQueueTick(): void {
|
|
fileQueueTimer = setTimeout(() => {
|
|
processFileQueueTick().catch((err) => {
|
|
console.error('[federation-worker] File queue tick error:', err);
|
|
}).finally(() => {
|
|
scheduleFileQueueTick();
|
|
});
|
|
}, FILE_QUEUE_INTERVAL_MS);
|
|
}
|
|
|
|
async function processFileQueueTick(): Promise<void> {
|
|
if (!isFederationRelayEnabled()) {
|
|
return;
|
|
}
|
|
|
|
const db = getDb();
|
|
const now = Date.now();
|
|
|
|
const pending = db
|
|
.select()
|
|
.from(schema.federationFileQueue)
|
|
.where(
|
|
and(
|
|
eq(schema.federationFileQueue.status, 'pending'),
|
|
lte(schema.federationFileQueue.nextRetryAt, now),
|
|
),
|
|
)
|
|
.limit(FILE_QUEUE_BATCH_LIMIT)
|
|
.all();
|
|
|
|
if (pending.length === 0) {
|
|
return;
|
|
}
|
|
|
|
const maxUploadSize = getMaxUploadSize();
|
|
|
|
for (const entry of pending) {
|
|
await processFileQueueEntry(db, entry, maxUploadSize, now);
|
|
}
|
|
}
|
|
|
|
function handleSizeRejection(
|
|
db: ReturnType<typeof getDb>,
|
|
entry: typeof schema.federationFileQueue.$inferSelect,
|
|
maxUploadSize: number,
|
|
now: number,
|
|
): void {
|
|
// Look up the local DM message to find the sender and channel info
|
|
const localMsg = db.select()
|
|
.from(schema.dmMessages)
|
|
.where(eq(schema.dmMessages.id, entry.dmMessageId))
|
|
.get();
|
|
|
|
if (!localMsg || !localMsg.sourceInstance || !localMsg.sourceMessageId) return;
|
|
|
|
// Find the attachment row to get its ID
|
|
const att = db.select()
|
|
.from(schema.attachments)
|
|
.where(
|
|
and(
|
|
eq(schema.attachments.dmMessageId, entry.dmMessageId),
|
|
eq(schema.attachments.sourceUrl, entry.sourceUrl),
|
|
),
|
|
)
|
|
.get();
|
|
|
|
// Resolve the sender's username from their local replicated user stub
|
|
const senderUser = db.select()
|
|
.from(schema.users)
|
|
.where(eq(schema.users.id, localMsg.userId))
|
|
.get();
|
|
const sourceUsername = senderUser?.displayName || senderUser?.username || 'unknown';
|
|
|
|
// Update attachment federation status
|
|
if (att) {
|
|
db.update(schema.attachments)
|
|
.set({
|
|
federationStatus: 'remote',
|
|
federationMeta: JSON.stringify({
|
|
sourceInstance: localMsg.sourceInstance,
|
|
sourceUserId: localMsg.userId,
|
|
sourceUsername,
|
|
}),
|
|
})
|
|
.where(eq(schema.attachments.id, att.id))
|
|
.run();
|
|
}
|
|
|
|
// Find local DM members native to THIS instance
|
|
const ourOrigin = getOurOrigin();
|
|
const dmMembers = db.select()
|
|
.from(schema.dmMembers)
|
|
.where(eq(schema.dmMembers.dmChannelId, localMsg.dmChannelId))
|
|
.all();
|
|
const affectedUserIds: string[] = [];
|
|
for (const member of dmMembers) {
|
|
const user = db.select()
|
|
.from(schema.users)
|
|
.where(eq(schema.users.id, member.userId))
|
|
.get();
|
|
if (user && (!user.homeInstance || user.homeInstance === ourOrigin)) {
|
|
affectedUserIds.push(user.homeUserId || user.id);
|
|
}
|
|
}
|
|
|
|
// Queue a file_rejected reverse relay event to the sender's instance
|
|
const event: FederationRelayEvent = {
|
|
eventType: 'file_rejected',
|
|
messageId: localMsg.sourceMessageId,
|
|
encryptionVersion: 0,
|
|
timestamp: now,
|
|
attachmentId: att?.id ?? entry.sourceUrl,
|
|
rejectionReason: 'size_limit_exceeded',
|
|
rejectionLimit: maxUploadSize,
|
|
affectedUserIds,
|
|
};
|
|
|
|
queueOutboxEvent(
|
|
localMsg.sourceMessageId,
|
|
localMsg.dmChannelId,
|
|
'file_rejected',
|
|
JSON.stringify(event),
|
|
[localMsg.sourceInstance],
|
|
);
|
|
|
|
// Broadcast updated message to local clients so they see the 'remote' badge
|
|
const updatedMsg = getDmMessageWithUser(entry.dmMessageId);
|
|
if (updatedMsg) {
|
|
connectionManager.sendToDmMembers(updatedMsg.dmChannelId, {
|
|
type: 'dm_message_updated',
|
|
message: updatedMsg,
|
|
});
|
|
}
|
|
}
|
|
|
|
async function processFileQueueEntry(
|
|
db: ReturnType<typeof getDb>,
|
|
entry: typeof schema.federationFileQueue.$inferSelect,
|
|
maxUploadSize: number,
|
|
now: number,
|
|
): Promise<void> {
|
|
// SSRF protection: validate sourceUrl hostname matches peerOrigin hostname
|
|
try {
|
|
const sourceHostname = new URL(entry.sourceUrl).hostname;
|
|
const peerHostname = new URL(entry.peerOrigin).hostname;
|
|
if (sourceHostname !== peerHostname) {
|
|
console.warn(
|
|
`[federation-worker] SSRF blocked: sourceUrl hostname "${sourceHostname}" does not match peer origin "${peerHostname}" for file queue entry ${entry.id}`,
|
|
);
|
|
db.update(schema.federationFileQueue)
|
|
.set({
|
|
status: 'rejected',
|
|
rejectionReason: 'ssrf_hostname_mismatch',
|
|
})
|
|
.where(eq(schema.federationFileQueue.id, entry.id))
|
|
.run();
|
|
return;
|
|
}
|
|
} catch {
|
|
db.update(schema.federationFileQueue)
|
|
.set({
|
|
status: 'rejected',
|
|
rejectionReason: 'invalid_url',
|
|
})
|
|
.where(eq(schema.federationFileQueue.id, entry.id))
|
|
.run();
|
|
return;
|
|
}
|
|
|
|
// Check file size against limit
|
|
if (entry.size > maxUploadSize) {
|
|
db.update(schema.federationFileQueue)
|
|
.set({
|
|
status: 'rejected',
|
|
rejectionReason: 'size_limit_exceeded',
|
|
})
|
|
.where(eq(schema.federationFileQueue.id, entry.id))
|
|
.run();
|
|
handleSizeRejection(db, entry, maxUploadSize, now);
|
|
return;
|
|
}
|
|
|
|
// Create abort controller for this download
|
|
fileQueueAbortController = new AbortController();
|
|
|
|
try {
|
|
const response = await fetch(entry.sourceUrl, {
|
|
signal: AbortSignal.any([
|
|
fileQueueAbortController.signal,
|
|
AbortSignal.timeout(FILE_DOWNLOAD_TIMEOUT_MS),
|
|
]),
|
|
});
|
|
|
|
if (!response.ok || !response.body) {
|
|
throw new Error(`HTTP ${response.status} downloading file from ${entry.sourceUrl}`);
|
|
}
|
|
|
|
// Generate a unique local filename using the same pattern as the upload route
|
|
const ext = path.extname(entry.originalName) || '';
|
|
const localId = generateSnowflake();
|
|
const localFilename = `${localId}${ext}`;
|
|
const localPath = path.join(config.uploadDir, localFilename);
|
|
|
|
// Ensure upload directory exists
|
|
fs.mkdirSync(config.uploadDir, { recursive: true });
|
|
|
|
// Stream the response body to disk
|
|
const nodeStream = Readable.fromWeb(response.body as ReadableStream);
|
|
const writeStream = fs.createWriteStream(localPath);
|
|
|
|
try {
|
|
await pipeline(nodeStream, writeStream);
|
|
} catch (pipeErr) {
|
|
// Clean up partial file on failure
|
|
try { fs.unlinkSync(localPath); } catch { /* ignore cleanup errors */ }
|
|
throw pipeErr;
|
|
}
|
|
|
|
// Verify file was written and get actual size
|
|
const stat = fs.statSync(localPath);
|
|
|
|
// Check actual downloaded size against limit (defense in depth)
|
|
if (stat.size > maxUploadSize) {
|
|
try { fs.unlinkSync(localPath); } catch { /* ignore */ }
|
|
db.update(schema.federationFileQueue)
|
|
.set({
|
|
status: 'rejected',
|
|
rejectionReason: 'size_limit_exceeded',
|
|
})
|
|
.where(eq(schema.federationFileQueue.id, entry.id))
|
|
.run();
|
|
handleSizeRejection(db, entry, maxUploadSize, now);
|
|
return;
|
|
}
|
|
|
|
// Generate thumbnail for images (same as local upload flow)
|
|
let thumbnailFilename: string | null = null;
|
|
try {
|
|
thumbnailFilename = await generateThumbnail(localPath, entry.mimetype, config.uploadDir);
|
|
} catch {
|
|
// Non-fatal — full image will be used instead
|
|
}
|
|
|
|
// Update the existing attachment row (created by processCreateEvent with
|
|
// sourceUrl as interim filename) to point to the local file
|
|
const updated = db.update(schema.attachments)
|
|
.set({
|
|
filename: localFilename,
|
|
size: stat.size,
|
|
thumbnailFilename,
|
|
})
|
|
.where(
|
|
and(
|
|
eq(schema.attachments.dmMessageId, entry.dmMessageId),
|
|
eq(schema.attachments.sourceUrl, entry.sourceUrl),
|
|
),
|
|
)
|
|
.run();
|
|
|
|
// Fallback: if no existing row was found (e.g., legacy queue entry from
|
|
// before processCreateEvent created rows), insert a new one
|
|
if (updated.changes === 0) {
|
|
db.insert(schema.attachments)
|
|
.values({
|
|
id: generateSnowflake(),
|
|
dmMessageId: entry.dmMessageId,
|
|
uploaderId: null,
|
|
filename: localFilename,
|
|
originalName: entry.originalName,
|
|
mimetype: entry.mimetype,
|
|
size: stat.size,
|
|
thumbnailFilename,
|
|
sourceUrl: entry.sourceUrl,
|
|
createdAt: now,
|
|
})
|
|
.run();
|
|
}
|
|
|
|
// Mark file queue entry as completed
|
|
db.update(schema.federationFileQueue)
|
|
.set({
|
|
status: 'completed',
|
|
targetFilename: localFilename,
|
|
})
|
|
.where(eq(schema.federationFileQueue.id, entry.id))
|
|
.run();
|
|
|
|
console.log(
|
|
`[federation-worker] Downloaded federated file: ${entry.originalName} -> ${localFilename}`,
|
|
);
|
|
|
|
// Notify connected clients that the attachment is now available locally
|
|
const updatedMsg = getDmMessageWithUser(entry.dmMessageId);
|
|
if (updatedMsg) {
|
|
connectionManager.sendToDmMembers(updatedMsg.dmChannelId, {
|
|
type: 'dm_message_updated',
|
|
message: updatedMsg,
|
|
});
|
|
}
|
|
} catch (err) {
|
|
if (err instanceof DOMException && err.name === 'AbortError') {
|
|
// Worker is stopping — leave entry as pending for next tick
|
|
return;
|
|
}
|
|
|
|
console.error(
|
|
`[federation-worker] Failed to download file ${entry.originalName} from ${entry.peerOrigin}:`,
|
|
err instanceof Error ? err.message : err,
|
|
);
|
|
|
|
const newAttempts = (entry.attempts ?? 0) + 1;
|
|
|
|
if (newAttempts > MAX_FILE_ATTEMPTS) {
|
|
db.update(schema.federationFileQueue)
|
|
.set({
|
|
status: 'failed',
|
|
attempts: newAttempts,
|
|
rejectionReason: 'max_attempts_exceeded',
|
|
})
|
|
.where(eq(schema.federationFileQueue.id, entry.id))
|
|
.run();
|
|
} else {
|
|
const backoffMs = getBackoffMs(newAttempts);
|
|
db.update(schema.federationFileQueue)
|
|
.set({
|
|
attempts: newAttempts,
|
|
nextRetryAt: now + backoffMs,
|
|
})
|
|
.where(eq(schema.federationFileQueue.id, entry.id))
|
|
.run();
|
|
}
|
|
}
|
|
}
|
|
|
|
// ─── Health Check Worker ────────────────────────────────────────────────────
|
|
|
|
function scheduleHealthCheckTick(): void {
|
|
healthCheckTimer = setTimeout(() => {
|
|
processHealthCheckTick().catch((err) => {
|
|
console.error('[federation-worker] Health check tick error:', err);
|
|
}).finally(() => {
|
|
scheduleHealthCheckTick();
|
|
});
|
|
}, HEALTH_CHECK_INTERVAL_MS);
|
|
}
|
|
|
|
async function processHealthCheckTick(): Promise<void> {
|
|
const db = getDb();
|
|
|
|
const unreachablePeers = db
|
|
.select()
|
|
.from(schema.federationPeers)
|
|
.where(eq(schema.federationPeers.status, 'unreachable'))
|
|
.all();
|
|
|
|
if (unreachablePeers.length === 0) {
|
|
return;
|
|
}
|
|
|
|
const now = Date.now();
|
|
|
|
for (const peer of unreachablePeers) {
|
|
healthCheckAbortController = new AbortController();
|
|
|
|
try {
|
|
const response = await fetch(`${peer.origin}/api/instance/info`, {
|
|
signal: AbortSignal.any([
|
|
healthCheckAbortController.signal,
|
|
AbortSignal.timeout(HEALTH_CHECK_TIMEOUT_MS),
|
|
]),
|
|
});
|
|
|
|
if (response.ok) {
|
|
db.update(schema.federationPeers)
|
|
.set({
|
|
status: 'active',
|
|
consecutiveFailures: 0,
|
|
lastSeenAt: now,
|
|
})
|
|
.where(eq(schema.federationPeers.id, peer.id))
|
|
.run();
|
|
|
|
console.log(
|
|
`[federation-worker] Peer ${peer.origin} recovered — marked active`,
|
|
);
|
|
}
|
|
// If not ok, leave as unreachable — will check again next cycle
|
|
} catch (err) {
|
|
if (err instanceof DOMException && err.name === 'AbortError') {
|
|
return;
|
|
}
|
|
// Leave as unreachable — will check again next cycle
|
|
}
|
|
}
|
|
}
|
|
|
|
// ─── Janitor Worker ──────────────────────────────────────────────────────────
|
|
|
|
function scheduleJanitorTick(): void {
|
|
janitorTimer = setTimeout(() => {
|
|
runFederationJanitor();
|
|
scheduleJanitorTick();
|
|
}, JANITOR_INTERVAL_MS);
|
|
}
|
|
|
|
// ─── Lifecycle ──────────────────────────────────────────────────────────────
|
|
|
|
/**
|
|
* Trigger checkpoint sync for peers that have never been synced (lastSyncedAt === 0).
|
|
* This catches historical messages that existed before the relay was enabled.
|
|
*/
|
|
async function runInitialSyncForNewPeers(): Promise<void> {
|
|
if (!isFederationRelayEnabled()) return;
|
|
|
|
const db = getDb();
|
|
const unsyncedPeers = db
|
|
.select()
|
|
.from(schema.federationPeers)
|
|
.where(and(
|
|
eq(schema.federationPeers.status, 'active'),
|
|
eq(schema.federationPeers.lastSyncedAt, 0),
|
|
))
|
|
.all();
|
|
|
|
if (unsyncedPeers.length === 0) return;
|
|
|
|
const ourOrigin = getOurOrigin();
|
|
|
|
for (const peer of unsyncedPeers) {
|
|
try {
|
|
console.log(`[federation-worker] Running initial sync with ${peer.origin}...`);
|
|
let sinceTimestamp = 0;
|
|
let totalEvents = 0;
|
|
|
|
// Paginate through all events from the peer
|
|
while (true) {
|
|
const body = JSON.stringify({ sinceTimestamp, limit: 100 });
|
|
const headers = buildFederationHeaders(body, peer.hmacSecret, ourOrigin);
|
|
|
|
const response = await fetch(`${peer.origin}/api/federation/sync`, {
|
|
method: 'POST',
|
|
headers,
|
|
body,
|
|
signal: AbortSignal.timeout(30_000),
|
|
});
|
|
|
|
if (!response.ok) {
|
|
console.error(`[federation-worker] Sync with ${peer.origin} failed: ${response.status}`);
|
|
break;
|
|
}
|
|
|
|
const data = await response.json() as { events: FederationRelayEvent[]; hasMore: boolean; checkpoint: number };
|
|
|
|
if (data.events.length === 0) break;
|
|
|
|
// Relay the events through our own relay endpoint logic
|
|
// For simplicity, POST them to ourselves
|
|
const relayBody = JSON.stringify({
|
|
version: 1,
|
|
sourceInstance: peer.origin,
|
|
events: data.events,
|
|
});
|
|
const relayHeaders = buildFederationHeaders(relayBody, peer.hmacSecret, peer.origin);
|
|
|
|
await fetch(`${ourOrigin}/api/federation/relay`, {
|
|
method: 'POST',
|
|
headers: relayHeaders,
|
|
body: relayBody,
|
|
signal: AbortSignal.timeout(30_000),
|
|
});
|
|
|
|
totalEvents += data.events.length;
|
|
sinceTimestamp = data.checkpoint;
|
|
|
|
if (!data.hasMore) break;
|
|
}
|
|
|
|
// Second pass: sync friend events
|
|
let friendSinceTimestamp = 0;
|
|
while (true) {
|
|
const friendBody = JSON.stringify({ sinceTimestamp: friendSinceTimestamp, contextType: 'friend', limit: 100 });
|
|
const friendHeaders = buildFederationHeaders(friendBody, peer.hmacSecret, ourOrigin);
|
|
|
|
const friendResponse = await fetch(`${peer.origin}/api/federation/sync`, {
|
|
method: 'POST',
|
|
headers: friendHeaders,
|
|
body: friendBody,
|
|
signal: AbortSignal.timeout(30_000),
|
|
});
|
|
|
|
if (!friendResponse.ok) {
|
|
console.error(`[federation-worker] Friend sync with ${peer.origin} failed: ${friendResponse.status}`);
|
|
break;
|
|
}
|
|
|
|
const friendData = await friendResponse.json() as { events: FederationRelayEvent[]; hasMore: boolean; checkpoint: number };
|
|
|
|
if (friendData.events.length === 0) break;
|
|
|
|
const friendRelayBody = JSON.stringify({
|
|
version: 1,
|
|
sourceInstance: peer.origin,
|
|
events: friendData.events,
|
|
});
|
|
const friendRelayHeaders = buildFederationHeaders(friendRelayBody, peer.hmacSecret, peer.origin);
|
|
|
|
await fetch(`${ourOrigin}/api/federation/relay`, {
|
|
method: 'POST',
|
|
headers: friendRelayHeaders,
|
|
body: friendRelayBody,
|
|
signal: AbortSignal.timeout(30_000),
|
|
});
|
|
|
|
totalEvents += friendData.events.length;
|
|
friendSinceTimestamp = friendData.checkpoint;
|
|
|
|
if (!friendData.hasMore) break;
|
|
}
|
|
|
|
// Update lastSyncedAt so this doesn't run again
|
|
db.update(schema.federationPeers)
|
|
.set({ lastSyncedAt: Date.now() })
|
|
.where(eq(schema.federationPeers.id, peer.id))
|
|
.run();
|
|
|
|
if (totalEvents > 0) {
|
|
console.log(`[federation-worker] Initial sync with ${peer.origin}: ${totalEvents} events synced`);
|
|
} else {
|
|
console.log(`[federation-worker] Initial sync with ${peer.origin}: no events to sync`);
|
|
}
|
|
} catch (err) {
|
|
console.error(`[federation-worker] Initial sync with ${peer.origin} failed:`, err);
|
|
// Don't update lastSyncedAt — will retry next startup
|
|
}
|
|
}
|
|
}
|
|
|
|
export function startFederationWorkers(): void {
|
|
console.log('[federation-worker] Federation workers started');
|
|
scheduleOutboxTick();
|
|
scheduleFileQueueTick();
|
|
scheduleHealthCheckTick();
|
|
scheduleJanitorTick();
|
|
// Run initial sync for newly peered instances (async, non-blocking)
|
|
runInitialSyncForNewPeers().catch((err) => {
|
|
console.error('[federation-worker] Initial sync error:', err);
|
|
});
|
|
}
|
|
|
|
export function stopFederationWorkers(): void {
|
|
if (outboxTimer) {
|
|
clearTimeout(outboxTimer);
|
|
outboxTimer = null;
|
|
}
|
|
if (fileQueueTimer) {
|
|
clearTimeout(fileQueueTimer);
|
|
fileQueueTimer = null;
|
|
}
|
|
if (healthCheckTimer) {
|
|
clearTimeout(healthCheckTimer);
|
|
healthCheckTimer = null;
|
|
}
|
|
|
|
if (janitorTimer) {
|
|
clearTimeout(janitorTimer);
|
|
janitorTimer = null;
|
|
}
|
|
|
|
outboxAbortController?.abort();
|
|
outboxAbortController = null;
|
|
|
|
fileQueueAbortController?.abort();
|
|
fileQueueAbortController = null;
|
|
|
|
healthCheckAbortController?.abort();
|
|
healthCheckAbortController = null;
|
|
|
|
console.log('[federation-worker] Federation workers stopped');
|
|
}
|