backspace/packages/server/src/utils/federationPeerActivation.test.ts

import { describe, it, expect, beforeEach, vi } from 'vitest';
import Database from 'better-sqlite3';
import { drizzle } from 'drizzle-orm/better-sqlite3';
import fs from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import * as schema from '../db/schema.js';
import { eq } from 'drizzle-orm';
import * as federationRouteMock from '../routes/federation.js';

const __dirname = path.dirname(fileURLToPath(import.meta.url));

type TestDb = ReturnType<typeof drizzle<typeof schema>>;
let sqlite: Database.Database;
let testDb: TestDb;

vi.mock('../db/index.js', () => ({
  getDb: () => testDb,
  schema,
}));

vi.mock('../utils/federationOutbox.js', () => ({
  isFederationRelayEnabled: () => true,
}));

vi.mock('../utils/federationAuth.js', () => ({
  getOurOrigin: () => 'https://local.example',
  buildFederationHeaders: (_body: string, _secret: string, _origin: string) => ({
    'Content-Type': 'application/json',
    'X-Federation-Origin': _origin,
  }),
}));

vi.mock('../routes/federation.js', () => ({
  processRelayEvents: vi.fn().mockResolvedValue({ accepted: [], rejected: [], undeliverable: [] }),
}));

vi.mock('../ws/handler.js', () => ({
  connectionManager: {
    sendToAdmins: vi.fn(),
    getAllOnlineUserIds: () => [],
    sendToUser: vi.fn(),
    sendToDmMembers: vi.fn(),
    evictFederatedCallsForHost: vi.fn().mockReturnValue(0),
  },
}));

function applyMigrations(db: Database.Database): void {
  const migrationsDir = path.resolve(__dirname, '../../drizzle');
  const files = fs.readdirSync(migrationsDir).filter(f => f.endsWith('.sql')).sort();
  for (const f of files) {
    const sql = fs.readFileSync(path.join(migrationsDir, f), 'utf8');
    const statements = sql.split(/-->\s*statement-breakpoint/);
    for (const stmt of statements) {
      const clean = stmt.trim();
      if (clean) db.exec(clean);
    }
  }
}

function seedPeer(id: string, status: string, lastSyncedAt = 0): void {
  testDb.insert(schema.federationPeers).values({
    id, origin: `https://${id}.example`, hmacSecret: 'secret',
    status, lastSyncedAt, createdAt: Date.now(),
  }).run();
}

function seedOutboxEntry(id: string, peerId: string, nextRetryAt: number, attempts: number): void {
  testDb.insert(schema.federationOutbox).values({
    id, peerId, contextId: 'ch-1', entityId: `msg-${id}`,
    contextType: 'dm', eventType: 'create', payload: '{}',
    encryptionVersion: 0, attempts, nextRetryAt,
    expiresAt: Date.now() + 30 * 86_400_000,
    createdAt: Date.now(),
  }).run();
}

describe('resetOutboxBackoff', () => {
  beforeEach(() => {
    sqlite = new Database(':memory:');
    testDb = drizzle(sqlite, { schema });
    applyMigrations(sqlite);
  });

  it('resets nextRetryAt=now and attempts=0 for all peer entries — including past-due ones', async () => {
    const { resetOutboxBackoff } = await import('./federationPeerActivation.js');
    seedPeer('peer-a', 'active');
    seedPeer('peer-b', 'active');
    const now = Date.now();

    // Three entries for peer-a: past-due (already eligible), near-future, far-future
    seedOutboxEntry('entry-1', 'peer-a', now - 1000, 5);
    seedOutboxEntry('entry-2', 'peer-a', now + 60_000, 3);
    seedOutboxEntry('entry-3', 'peer-a', now + 86_400_000, 7);
    // Entry for unrelated peer-b (must NOT be touched)
    seedOutboxEntry('entry-4', 'peer-b', now + 86_400_000, 9);

    resetOutboxBackoff('peer-a');

    const a1 = testDb.select().from(schema.federationOutbox).where(eq(schema.federationOutbox.id, 'entry-1')).get();
    const a2 = testDb.select().from(schema.federationOutbox).where(eq(schema.federationOutbox.id, 'entry-2')).get();
    const a3 = testDb.select().from(schema.federationOutbox).where(eq(schema.federationOutbox.id, 'entry-3')).get();
    const b4 = testDb.select().from(schema.federationOutbox).where(eq(schema.federationOutbox.id, 'entry-4')).get();

    // All peer-a entries reset — including the past-due one (correctness: attempts=0 on those too)
    expect(a1?.attempts).toBe(0);
    expect(a2?.attempts).toBe(0);
    expect(a3?.attempts).toBe(0);
    expect(a1?.nextRetryAt).toBeGreaterThanOrEqual(now);
    expect(a2?.nextRetryAt).toBeLessThanOrEqual(Date.now());
    expect(a3?.nextRetryAt).toBeLessThanOrEqual(Date.now());
    // peer-b untouched
    expect(b4?.attempts).toBe(9);
    expect(b4?.nextRetryAt).toBe(now + 86_400_000);
  });

  it('is a no-op when the peer has no outbox entries', async () => {
    const { resetOutboxBackoff } = await import('./federationPeerActivation.js');
    seedPeer('peer-empty', 'active');
    expect(() => resetOutboxBackoff('peer-empty')).not.toThrow();
  });
});

describe('syncPeerMutationLog', () => {
  beforeEach(() => {
    sqlite = new Database(':memory:');
    testDb = drizzle(sqlite, { schema });
    applyMigrations(sqlite);
    vi.restoreAllMocks();
  });

  it('seeds sinceTimestamp from peer.lastSyncedAt for each pass', async () => {
    const { syncPeerMutationLog } = await import('./federationPeerActivation.js');
    testDb.insert(schema.federationPeers).values({
      id: 'peer-1', origin: 'https://peer-1.example', hmacSecret: 'secret',
      status: 'active', lastSyncedAt: 5000, createdAt: Date.now(),
    }).run();

    const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
      new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: 5000 }), { status: 200 })
    );

    await syncPeerMutationLog('peer-1', 'health_check_recovery');

    // Three passes: dm (no contextType), friend, profile
    expect(fetchSpy).toHaveBeenCalledTimes(3);
    for (const call of fetchSpy.mock.calls) {
      const body = JSON.parse(call[1]?.body as string) as { sinceTimestamp: number };
      expect(body.sinceTimestamp).toBe(5000);
    }
    const calls = fetchSpy.mock.calls.map(c => JSON.parse(c[1]?.body as string) as { contextType?: string });
    expect(calls[0]!.contextType).toBeUndefined();       // DM pass (no contextType filter)
    expect(calls[1]!.contextType).toBe('friend');
    expect(calls[2]!.contextType).toBe('profile');
  });

  it('advances lastSyncedAt on success', async () => {
    const { syncPeerMutationLog } = await import('./federationPeerActivation.js');
    testDb.insert(schema.federationPeers).values({
      id: 'peer-2', origin: 'https://peer-2.example', hmacSecret: 'secret',
      status: 'active', lastSyncedAt: 0, createdAt: Date.now(),
    }).run();

    vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
      new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: 1000 }), { status: 200 })
    );

    const before = Date.now();
    await syncPeerMutationLog('peer-2', 'startup_bootstrap');
    const after = Date.now();

    const row = testDb.select().from(schema.federationPeers)
      .where(eq(schema.federationPeers.id, 'peer-2')).get();
    expect(row?.lastSyncedAt).toBeGreaterThanOrEqual(before);
    expect(row?.lastSyncedAt).toBeLessThanOrEqual(after);
  });

  it('does NOT update lastSyncedAt on transient failure', async () => {
    const { syncPeerMutationLog } = await import('./federationPeerActivation.js');
    testDb.insert(schema.federationPeers).values({
      id: 'peer-3', origin: 'https://peer-3.example', hmacSecret: 'secret',
      status: 'active', lastSyncedAt: 42_000, createdAt: Date.now(),
    }).run();

    vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
      new Response('internal error', { status: 500 })
    );

    await syncPeerMutationLog('peer-3', 'ensure_peered');
    const row = testDb.select().from(schema.federationPeers)
      .where(eq(schema.federationPeers.id, 'peer-3')).get();
    expect(row?.lastSyncedAt).toBe(42_000);
  });

  it('does nothing when peer is not active', async () => {
    const { syncPeerMutationLog } = await import('./federationPeerActivation.js');
    testDb.insert(schema.federationPeers).values({
      id: 'peer-4', origin: 'https://peer-4.example', hmacSecret: 'secret',
      status: 'pending', lastSyncedAt: 0, createdAt: Date.now(),
    }).run();

    const fetchSpy = vi.spyOn(globalThis, 'fetch');
    await syncPeerMutationLog('peer-4', 'health_check_recovery');
    expect(fetchSpy).not.toHaveBeenCalled();
  });

  it('advances sinceTimestamp within a pass using data.checkpoint when hasMore is true', async () => {
    const { syncPeerMutationLog } = await import('./federationPeerActivation.js');
    testDb.insert(schema.federationPeers).values({
      id: 'peer-5', origin: 'https://peer-5.example', hmacSecret: 'secret',
      status: 'active', lastSyncedAt: 100, createdAt: Date.now(),
    }).run();

    let call = 0;
    const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async () => {
      call++;
      // First DM call: one event, hasMore=true, checkpoint advances to 2500
      // Second DM call: empty, hasMore=false, ends the DM pass
      // Remaining calls (friend, profile): empty/done immediately
      if (call === 1) {
        return new Response(
          JSON.stringify({
            events: [{ eventType: 'create', messageId: 'm1', timestamp: 200, encryptionVersion: 0 }],
            hasMore: true,
            checkpoint: 2500,
          }),
          { status: 200 },
        );
      }
      return new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: call === 2 ? 2500 : 100 }), { status: 200 });
    });

    await syncPeerMutationLog('peer-5', 'health_check_recovery');

    // Call 1: DM pass, since=100 (peer.lastSyncedAt)
    // Call 2: DM pass continuation, since=2500 (advanced by previous checkpoint)
    // Call 3: friend pass, since=100 (re-seeded from peer.lastSyncedAt)
    // Call 4: profile pass, since=100 (re-seeded from peer.lastSyncedAt)
    expect(fetchSpy).toHaveBeenCalledTimes(4);
    const bodies = fetchSpy.mock.calls.map(c => JSON.parse(c[1]?.body as string) as { sinceTimestamp: number; contextType?: string });
    expect(bodies[0]?.sinceTimestamp).toBe(100);
    expect(bodies[0]?.contextType).toBeUndefined();
    expect(bodies[1]?.sinceTimestamp).toBe(2500);  // advanced by checkpoint from call 1
    expect(bodies[1]?.contextType).toBeUndefined();
    expect(bodies[2]?.sinceTimestamp).toBe(100);   // friend pass re-seeds from peer.lastSyncedAt
    expect(bodies[2]?.contextType).toBe('friend');
    expect(bodies[3]?.sinceTimestamp).toBe(100);   // profile pass re-seeds from peer.lastSyncedAt
    expect(bodies[3]?.contextType).toBe('profile');
  });

  it('skips a poison-pill event, logs it, and advances past it to process subsequent events', async () => {
    const { syncPeerMutationLog } = await import('./federationPeerActivation.js');
    testDb.insert(schema.federationPeers).values({
      id: 'peer-poison', origin: 'https://peer-poison.example', hmacSecret: 'secret',
      status: 'active', lastSyncedAt: 100, createdAt: Date.now(),
    }).run();

    // Return a single batch of 3 events on the DM pass, then empty on friend + profile passes.
    let fetchCall = 0;
    vi.spyOn(globalThis, 'fetch').mockImplementation(async () => {
      fetchCall++;
      if (fetchCall === 1) {
        return new Response(JSON.stringify({
          events: [
            { eventType: 'create', messageId: 'good-1', timestamp: 200, encryptionVersion: 0 },
            { eventType: 'create', messageId: 'poison', timestamp: 300, encryptionVersion: 0 },
            { eventType: 'create', messageId: 'good-2', timestamp: 400, encryptionVersion: 0 },
          ],
          hasMore: false,
          checkpoint: 400,
        }), { status: 200 });
      }
      return new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: 100 }), { status: 200 });
    });

    // Grab the top-level mock and override implementation per-call:
    // good-1: resolves, poison: throws, good-2: resolves.
    const processMock = vi.mocked(federationRouteMock.processRelayEvents);
    processMock.mockClear();
    processMock.mockResolvedValueOnce(undefined as never);  // good-1
    processMock.mockRejectedValueOnce(new Error('simulated processor failure'));  // poison
    processMock.mockResolvedValueOnce(undefined as never);  // good-2

    const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {});

    const before = Date.now();
    await syncPeerMutationLog('peer-poison', 'health_check_recovery');

    // Verify processRelayEvents was called per-event: 3 calls for the 3 DM events.
    expect(processMock).toHaveBeenCalledTimes(3);

    // Verify error logged for the poison event.
    expect(errorSpy).toHaveBeenCalled();
    const errorMessages = errorSpy.mock.calls.map(c => String(c[0] ?? ''));
    expect(errorMessages.some(m => m.includes('poison'))).toBe(true);
    expect(errorMessages.some(m => m.includes('simulated processor failure'))).toBe(true);

    // Verify lastSyncedAt advanced despite the poison event (the critical property).
    const row = testDb.select().from(schema.federationPeers)
      .where(eq(schema.federationPeers.id, 'peer-poison')).get();
    expect(row?.lastSyncedAt).toBeGreaterThanOrEqual(before);
  });
});

describe('onPeerActivated', () => {
  beforeEach(() => {
    sqlite = new Database(':memory:');
    testDb = drizzle(sqlite, { schema });
    applyMigrations(sqlite);
    vi.restoreAllMocks();
  });

  it('runs resetOutboxBackoff and syncPeerMutationLog once, even under concurrent calls', async () => {
    const { onPeerActivated } = await import('./federationPeerActivation.js');

    testDb.insert(schema.federationPeers).values({
      id: 'peer-x', origin: 'https://peer-x.example', hmacSecret: 'secret',
      status: 'active', lastSyncedAt: 0, createdAt: Date.now(),
    }).run();

    let fetchCount = 0;
    vi.spyOn(globalThis, 'fetch').mockImplementation(async () => {
      fetchCount++;
      // Deliberately slow to let the second concurrent call share the in-flight promise.
      await new Promise(r => setTimeout(r, 20));
      return new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: 0 }), { status: 200 });
    });

    const p1 = onPeerActivated('peer-x', 'health_check_recovery');
    const p2 = onPeerActivated('peer-x', 'accept_new');
    await Promise.all([p1, p2]);

    // Three fetch calls for the three sync passes (dm, friend, profile) — not six.
    expect(fetchCount).toBe(3);
  });

  it('swallows errors from syncPeerMutationLog so the handler does not throw', async () => {
    const { onPeerActivated } = await import('./federationPeerActivation.js');

    testDb.insert(schema.federationPeers).values({
      id: 'peer-err', origin: 'https://peer-err.example', hmacSecret: 'secret',
      status: 'active', lastSyncedAt: 0, createdAt: Date.now(),
    }).run();

    vi.spyOn(globalThis, 'fetch').mockImplementation(async () => {
      throw new Error('network down');
    });

    await expect(onPeerActivated('peer-err', 'ensure_peered')).resolves.toBeUndefined();
  });
});

describe('onPeerDeactivated', () => {
  beforeEach(() => {
    sqlite = new Database(':memory:');
    testDb = drizzle(sqlite, { schema });
    applyMigrations(sqlite);
    vi.clearAllMocks();
  });

  it('evicts federated calls for the peer origin with peer_transient_failure on network_threshold', async () => {
    seedPeer('peer-net', 'unreachable');
    testDb.update(schema.federationPeers)
      .set({ instanceName: 'NetPeer' })
      .where(eq(schema.federationPeers.id, 'peer-net'))
      .run();

    const { onPeerDeactivated } = await import('./federationPeerActivation.js');
    await onPeerDeactivated('peer-net', 'network_threshold');

    const { connectionManager } = await import('../ws/handler.js');
    expect(connectionManager.evictFederatedCallsForHost).toHaveBeenCalledWith(
      'https://peer-net.example',
      { reason: 'peer_transient_failure', peerLabel: 'NetPeer' },
    );
  });

  it('maps rejected status to peer_rejected reason', async () => {
    seedPeer('peer-rej', 'rejected');
    const { onPeerDeactivated } = await import('./federationPeerActivation.js');
    await onPeerDeactivated('peer-rej', 'remote_rejected');

    const { connectionManager } = await import('../ws/handler.js');
    expect(connectionManager.evictFederatedCallsForHost).toHaveBeenCalledWith(
      'https://peer-rej.example',
      { reason: 'peer_rejected', peerLabel: undefined },
    );
  });

  it('maps revoked status to peer_rejected reason', async () => {
    seedPeer('peer-rev', 'revoked');
    const { onPeerDeactivated } = await import('./federationPeerActivation.js');
    await onPeerDeactivated('peer-rev', 'admin_revoked');

    const { connectionManager } = await import('../ws/handler.js');
    expect(connectionManager.evictFederatedCallsForHost).toHaveBeenCalledWith(
      'https://peer-rev.example',
      { reason: 'peer_rejected', peerLabel: undefined },
    );
  });

  it('broadcasts federation_peers_changed to admins', async () => {
    seedPeer('peer-broad', 'unreachable');
    const { onPeerDeactivated } = await import('./federationPeerActivation.js');
    await onPeerDeactivated('peer-broad', 'network_threshold');

    const { connectionManager } = await import('../ws/handler.js');
    expect(connectionManager.sendToAdmins).toHaveBeenCalledWith({ type: 'federation_peers_changed' });
  });

  it('aborts silently when the peer row is missing', async () => {
    const { onPeerDeactivated } = await import('./federationPeerActivation.js');
    await expect(onPeerDeactivated('peer-missing', 'network_threshold')).resolves.toBeUndefined();

    const { connectionManager } = await import('../ws/handler.js');
    expect(connectionManager.evictFederatedCallsForHost).not.toHaveBeenCalled();
  });

  it('deduplicates concurrent calls for the same peerId', async () => {
    seedPeer('peer-x', 'unreachable');
    const { onPeerDeactivated } = await import('./federationPeerActivation.js');
    const { connectionManager } = await import('../ws/handler.js');

    const p1 = onPeerDeactivated('peer-x', 'network_threshold');
    const p2 = onPeerDeactivated('peer-x', 'network_threshold');
    await Promise.all([p1, p2]);

    // Exactly one eviction call, not two
    expect(connectionManager.evictFederatedCallsForHost).toHaveBeenCalledTimes(1);
  });

  it('uses a dedup map SEPARATE from onPeerActivated', async () => {
    seedPeer('peer-flap', 'active');
    const { onPeerActivated, onPeerDeactivated } = await import('./federationPeerActivation.js');

    // Simulate an activation already in flight — spawn onPeerActivated then
    // immediately kick off a deactivation for the same peer id. The latter
    // must NOT be swallowed as a dedup hit against the activation.
    const actPromise = onPeerActivated('peer-flap', 'ensure_peered');

    // Mark peer non-active now — otherwise the deactivation utility's
    // own status guard would skip it.
    testDb.update(schema.federationPeers)
      .set({ status: 'unreachable' })
      .where(eq(schema.federationPeers.id, 'peer-flap'))
      .run();

    const deactPromise = onPeerDeactivated('peer-flap', 'network_threshold');
    await Promise.all([actPromise, deactPromise]);

    const { connectionManager } = await import('../ws/handler.js');
    expect(connectionManager.evictFederatedCallsForHost).toHaveBeenCalled();
  });
});