backspace/packages/server/src/utils/federationWorker.ts

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.federatedId) evt.federatedId = parsed.federatedId;
      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;
      // file_rejected event fields
      if (parsed.attachmentId) evt.attachmentId = parsed.attachmentId;
      if (parsed.sourceFilename) evt.sourceFilename = parsed.sourceFilename;
      if (parsed.rejectionReason) evt.rejectionReason = parsed.rejectionReason;
      if (parsed.rejectionLimit != null) evt.rejectionLimit = parsed.rejectionLimit;
      if (parsed.affectedUserIds) evt.affectedUserIds = parsed.affectedUserIds;
      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
  // Extract the filename from the sourceUrl (e.g., "https://sender/api/uploads/12345.png" → "12345.png")
  const sourceFilename = entry.sourceUrl.split('/').pop() ?? entry.sourceUrl;

  const event: FederationRelayEvent = {
    eventType: 'file_rejected',
    messageId: localMsg.sourceMessageId,
    encryptionVersion: 0,
    timestamp: now,
    attachmentId: att?.id ?? entry.sourceUrl,
    sourceFilename,
    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');
}