Shade/packages/shade-transport-bridge/tests/bridge.test.ts

/**
 * Bridge integration tests — exercises real Bun.serve + InboxServer +
 * createBridgeRoutes against actual SSE / long-poll / WS clients.
 *
 * The acceptance criteria from V3.7 we cover here:
 *  1. "Send 100 small messages" passes on all three transports.
 *  2. WS-blocked client falls through to SSE without message loss.
 *  3. Long-poll uses no more than one outstanding request per client.
 */

import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
import { Hono } from 'hono';
import { SubtleCryptoProvider } from '@shade/crypto-web';
import {
  createInboxRoutes,
  createBridgeRoutes,
  InboxServerEvents,
  MemoryInboxStore,
  computeMsgId,
  type InboxStore,
} from '@shade/inbox-server';
import { signPayload } from '@shade/server';
import { generateIdentityKeyPair, toBase64 } from '@shade/core';

import {
  SseBridge,
  LongPollBridge,
  WsBridge,
  FallbackBridgeTransport,
  signBridgeQuery,
  PresenceBridge,
  type IncomingMessage,
  type PresenceChange,
} from '../src/index.js';

const crypto = new SubtleCryptoProvider();

interface Harness {
  server: ReturnType<typeof Bun.serve>;
  baseUrl: string;
  store: InboxStore;
  events: InboxServerEvents;
  bob: Awaited<ReturnType<typeof generateIdentityKeyPair>>;
  alice: Awaited<ReturnType<typeof generateIdentityKeyPair>>;
}

async function bootstrap(opts: { mountWs?: boolean } = {}): Promise<Harness> {
  const store = new MemoryInboxStore();
  const events = new InboxServerEvents();
  // V4.8.4 — share a BridgeDeliveryLog between bridge + inbox routes so
  // the inbox-fetch path filters out blobs the bridge already pushed.
  // Mirrors the wiring in `@shade/server/standalone.ts`.
  const bridge = createBridgeRoutes({
    store,
    crypto,
    events,
    longPollTimeoutMs: 1_000,
    longPollMaxTimeoutMs: 2_000,
    heartbeatIntervalMs: 200,
    fallbackPollIntervalMs: 50,
  });
  const inboxApp = createInboxRoutes(store, crypto, {
    events,
    disableRateLimit: true,
    bridgeDeliveryLog: bridge.bridgeDeliveryLog,
  });
  const app = new Hono();
  app.route('/', inboxApp);
  app.route('/', bridge.app);

  const port = 19000 + Math.floor(Math.random() * 1000);
  const server = Bun.serve({
    port,
    fetch: (req, srv) => app.fetch(req, srv),
    websocket: opts.mountWs === false ? undefined : (bridge.websocket as any),
  });

  // Register Bob.
  const bob = await generateIdentityKeyPair(crypto);
  const alice = await generateIdentityKeyPair(crypto);
  const regBody = await signPayload(crypto, bob.signingPrivateKey, {
    address: 'bob',
    signingKey: toBase64(bob.signingPublicKey),
  });
  const regRes = await fetch(`http://localhost:${port}/v1/inbox/register`, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify(regBody),
  });
  expect(regRes.status).toBe(200);

  return {
    server,
    baseUrl: `http://localhost:${port}`,
    store,
    events,
    bob,
    alice,
  };
}

async function putBlob(harness: Harness, ciphertext: Uint8Array): Promise<string> {
  const msgId = await computeMsgId(ciphertext);
  const body = await signPayload(crypto, harness.alice.signingPrivateKey, {
    senderSigningKey: toBase64(harness.alice.signingPublicKey),
    msgId,
    ciphertext: toBase64(ciphertext),
  });
  const res = await fetch(`${harness.baseUrl}/v1/inbox/bob`, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify(body),
  });
  expect(res.status).toBe(200);
  return msgId;
}

function rand(n: number): Uint8Array {
  const b = new Uint8Array(n);
  globalThis.crypto.getRandomValues(b);
  return b;
}

function bobAuth(harness: Harness) {
  return {
    crypto,
    signingPrivateKey: harness.bob.signingPrivateKey,
    address: 'bob',
  };
}

async function waitFor(predicate: () => boolean, timeoutMs = 4000): Promise<void> {
  const start = Date.now();
  while (Date.now() - start < timeoutMs) {
    if (predicate()) return;
    await new Promise((r) => setTimeout(r, 25));
  }
  throw new Error(`waitFor timeout: predicate still false after ${timeoutMs}ms`);
}

describe('signBridgeQuery', () => {
  test('produces deterministic shape', async () => {
    const id = await generateIdentityKeyPair(crypto);
    const qs = await signBridgeQuery({
      crypto,
      signingPrivateKey: id.signingPrivateKey,
      address: 'foo',
      kind: 'stream',
      since: 42,
    });
    expect(qs.get('address')).toBe('foo');
    expect(qs.get('kind')).toBe('stream');
    expect(qs.get('since')).toBe('42');
    expect(qs.get('signedAt')).toMatch(/^\d+$/);
    expect(qs.get('signature')).toBeTruthy();
  });
});

describe('SseBridge — send 100 small messages', () => {
  let h: Harness;
  beforeAll(async () => {
    h = await bootstrap();
  });
  afterAll(() => {
    h.server.stop(true);
  });

  test('all messages reach the client over SSE', async () => {
    const received: IncomingMessage[] = [];
    const bridge = new SseBridge({
      baseUrl: h.baseUrl,
      auth: bobAuth(h),
      initialBackoffMs: 100,
      maxBackoffMs: 500,
    });
    await bridge.connect({
      onMessage: (m) => {
        received.push(m);
      },
    });

    const sent: string[] = [];
    for (let i = 0; i < 100; i++) {
      const ct = rand(32 + (i % 8));
      sent.push(await putBlob(h, ct));
    }
    await waitFor(() => received.length === 100, 5_000);
    expect(received.length).toBe(100);
    const ids = new Set(received.map((m) => m.msgId));
    expect(ids.size).toBe(100);
    for (const id of sent) expect(ids.has(id)).toBe(true);
    await bridge.disconnect();
  });
});

describe('LongPollBridge — send 100 small messages', () => {
  let h: Harness;
  beforeAll(async () => {
    h = await bootstrap();
  });
  afterAll(() => {
    h.server.stop(true);
  });

  test('all messages reach the client over long-poll', async () => {
    const received: IncomingMessage[] = [];
    const bridge = new LongPollBridge({
      baseUrl: h.baseUrl,
      auth: bobAuth(h),
      pollTimeoutMs: 500,
      requestTimeoutMs: 1_500,
      errorBackoffMs: 50,
    });
    await bridge.connect({
      onMessage: (m) => {
        received.push(m);
      },
    });
    const sent: string[] = [];
    for (let i = 0; i < 100; i++) {
      const ct = rand(48);
      sent.push(await putBlob(h, ct));
    }
    await waitFor(() => received.length === 100, 8_000);
    expect(received.length).toBe(100);
    const ids = new Set(received.map((m) => m.msgId));
    for (const id of sent) expect(ids.has(id)).toBe(true);
    await bridge.disconnect();
  });

  test('one outstanding request at a time', async () => {
    let inflight = 0;
    let peak = 0;
    const wrapped: typeof fetch = async (input, init) => {
      inflight++;
      peak = Math.max(peak, inflight);
      try {
        return await fetch(input as any, init as any);
      } finally {
        inflight--;
      }
    };
    const bridge = new LongPollBridge({
      baseUrl: h.baseUrl,
      auth: bobAuth(h),
      pollTimeoutMs: 200,
      requestTimeoutMs: 500,
      errorBackoffMs: 50,
      fetch: wrapped,
    });
    const received: IncomingMessage[] = [];
    await bridge.connect({ onMessage: (m) => void received.push(m) });
    // Let the loop run a few times; nothing else should be pumping the
    // bridge concurrently.
    await new Promise((r) => setTimeout(r, 1_500));
    expect(peak).toBeLessThanOrEqual(1);
    await bridge.disconnect();
  });
});

describe('WsBridge — send 100 small messages', () => {
  let h: Harness;
  beforeAll(async () => {
    h = await bootstrap();
  });
  afterAll(() => {
    h.server.stop(true);
  });

  test('all messages reach the client over WS', async () => {
    const received: IncomingMessage[] = [];
    const bridge = new WsBridge({
      baseUrl: h.baseUrl,
      auth: bobAuth(h),
      connectTimeoutMs: 2_000,
      disableAutoReconnect: true,
    });
    await bridge.connect({
      onMessage: (m) => {
        received.push(m);
      },
    });
    const sent: string[] = [];
    for (let i = 0; i < 100; i++) {
      const ct = rand(40);
      sent.push(await putBlob(h, ct));
    }
    await waitFor(() => received.length === 100, 5_000);
    expect(received.length).toBe(100);
    const ids = new Set(received.map((m) => m.msgId));
    for (const id of sent) expect(ids.has(id)).toBe(true);
    await bridge.disconnect();
  });
});

describe('FallbackBridgeTransport', () => {
  let h: Harness;
  beforeAll(async () => {
    h = await bootstrap();
  });
  afterAll(() => {
    h.server.stop(true);
  });

  test('falls through WS → SSE when WS endpoint is blocked', async () => {
    const received: IncomingMessage[] = [];
    const ws = new WsBridge({
      baseUrl: 'http://127.0.0.1:1', // unreachable on purpose
      auth: bobAuth(h),
      connectTimeoutMs: 500,
      disableAutoReconnect: true,
    });
    const sse = new SseBridge({
      baseUrl: h.baseUrl,
      auth: bobAuth(h),
      initialBackoffMs: 100,
      maxBackoffMs: 500,
    });
    const lp = new LongPollBridge({
      baseUrl: h.baseUrl,
      auth: bobAuth(h),
      pollTimeoutMs: 500,
      requestTimeoutMs: 1_500,
      errorBackoffMs: 50,
    });
    const fallback = new FallbackBridgeTransport([ws, sse, lp]);
    await fallback.connect({
      onMessage: (m) => {
        received.push(m);
      },
    });
    expect(fallback.activeKind).toBe('sse');
    expect(fallback.attempts).toEqual(['ws', 'sse']);

    const sent: string[] = [];
    for (let i = 0; i < 10; i++) {
      const ct = rand(32);
      sent.push(await putBlob(h, ct));
    }
    await waitFor(() => received.length === 10, 5_000);
    expect(received.length).toBe(10);
    const ids = new Set(received.map((m) => m.msgId));
    for (const id of sent) expect(ids.has(id)).toBe(true);
    await fallback.disconnect();
  });

  test('falls through SSE → long-poll when SSE endpoint returns 502', async () => {
    // Build a server where SSE explicitly returns 502; long-poll passes
    // through. We build a small Hono shim that wraps the bridge app.
    const port = h.server.port;
    const altPort = port + 1;

    const store = h.store; // share data with the main harness
    const events = h.events;
    const bridge = createBridgeRoutes({
      store,
      crypto,
      events,
      longPollTimeoutMs: 500,
      longPollMaxTimeoutMs: 1_000,
      heartbeatIntervalMs: 200,
      fallbackPollIntervalMs: 50,
    });
    const inboxApp = createInboxRoutes(store, crypto, { events, disableRateLimit: true });
    const wrapped = new Hono();
    wrapped.get('/v1/bridge/stream', (c) => c.text('bad gateway', 502));
    wrapped.route('/', bridge.app);
    wrapped.route('/', inboxApp);

    const altServer = Bun.serve({
      port: altPort,
      fetch: (req, srv) => wrapped.fetch(req, srv),
      websocket: bridge.websocket as any,
    });
    try {
      const altUrl = `http://localhost:${altPort}`;
      const received: IncomingMessage[] = [];
      const sse = new SseBridge({
        baseUrl: altUrl,
        auth: bobAuth(h),
        initialBackoffMs: 100,
        maxBackoffMs: 200,
        disableAutoReconnect: true,
      });
      const lp = new LongPollBridge({
        baseUrl: altUrl,
        auth: bobAuth(h),
        pollTimeoutMs: 200,
        requestTimeoutMs: 600,
        errorBackoffMs: 50,
      });
      const fallback = new FallbackBridgeTransport([sse, lp]);
      await fallback.connect({
        onMessage: (m) => {
          received.push(m);
        },
      });
      expect(fallback.activeKind).toBe('long-poll');
      // Push a message via the original PUT path.
      const ct = rand(32);
      await putBlob(h, ct);
      await waitFor(() => received.length >= 1, 4_000);
      expect(received.length).toBeGreaterThanOrEqual(1);
      await fallback.disconnect();
    } finally {
      altServer.stop(true);
    }
  });

  test('throws when every bridge fails', async () => {
    const ws = new WsBridge({
      baseUrl: 'http://127.0.0.1:1',
      auth: bobAuth(h),
      connectTimeoutMs: 200,
      disableAutoReconnect: true,
    });
    const sse = new SseBridge({
      baseUrl: 'http://127.0.0.1:1',
      auth: bobAuth(h),
      initialBackoffMs: 50,
      maxBackoffMs: 100,
      disableAutoReconnect: true,
    });
    const lp = new LongPollBridge({
      baseUrl: 'http://127.0.0.1:1',
      auth: bobAuth(h),
      pollTimeoutMs: 100,
      requestTimeoutMs: 300,
      errorBackoffMs: 50,
    });
    const fallback = new FallbackBridgeTransport([ws, sse, lp]);
    await expect(
      fallback.connect({ onMessage: () => undefined }),
    ).rejects.toThrow();
    await fallback.disconnect();
  });
});

describe('Bridge auth — rejects bad signatures', () => {
  let h: Harness;
  beforeAll(async () => {
    h = await bootstrap();
  });
  afterAll(() => {
    h.server.stop(true);
  });

  test('SSE rejects when signature is wrong', async () => {
    const eve = await generateIdentityKeyPair(crypto);
    const sse = new SseBridge({
      baseUrl: h.baseUrl,
      auth: { crypto, signingPrivateKey: eve.signingPrivateKey, address: 'bob' },
      initialBackoffMs: 50,
      maxBackoffMs: 100,
      disableAutoReconnect: true,
    });
    await expect(sse.connect({ onMessage: () => undefined })).rejects.toThrow();
    await sse.disconnect();
  });

  test('long-poll rejects when address is unregistered', async () => {
    const newId = await generateIdentityKeyPair(crypto);
    const lp = new LongPollBridge({
      baseUrl: h.baseUrl,
      auth: { crypto, signingPrivateKey: newId.signingPrivateKey, address: 'unregistered' },
      pollTimeoutMs: 200,
      requestTimeoutMs: 500,
      errorBackoffMs: 50,
    });
    await expect(lp.connect({ onMessage: () => undefined })).rejects.toThrow();
    await lp.disconnect();
  });
});

describe('Bridge cursor resume', () => {
  let h: Harness;
  beforeAll(async () => {
    h = await bootstrap();
  });
  afterAll(() => {
    h.server.stop(true);
  });

  test('SSE picks up after disconnect using Last-Event-ID cursor', async () => {
    const first: IncomingMessage[] = [];
    const sseA = new SseBridge({
      baseUrl: h.baseUrl,
      auth: bobAuth(h),
      initialBackoffMs: 100,
      maxBackoffMs: 200,
      disableAutoReconnect: true,
    });
    await sseA.connect({ onMessage: (m) => first.push(m) });
    await putBlob(h, rand(20));
    await putBlob(h, rand(24));
    await waitFor(() => first.length === 2, 3_000);
    const cursor = sseA.getCursor();
    await sseA.disconnect();

    // Push more while disconnected.
    await putBlob(h, rand(28));
    await putBlob(h, rand(32));

    const second: IncomingMessage[] = [];
    const sseB = new SseBridge({
      baseUrl: h.baseUrl,
      auth: bobAuth(h),
      startCursor: cursor,
      initialBackoffMs: 100,
      maxBackoffMs: 200,
      disableAutoReconnect: true,
    });
    await sseB.connect({ onMessage: (m) => second.push(m) });
    await waitFor(() => second.length === 2, 3_000);
    expect(second.length).toBe(2);
    // No overlap with the first batch.
    const firstIds = new Set(first.map((m) => m.msgId));
    for (const m of second) expect(firstIds.has(m.msgId!)).toBe(false);
    await sseB.disconnect();
  });
});

describe('Bridges — default fetch is bound to globalThis', () => {
  // Regression: browsers' `fetch` is a WebIDL bound operation that throws
  // "Illegal invocation" when called via `this.fetchFn(...)`. Constructors
  // for LongPollBridge / SseBridge must `bind(globalThis)`.
  function installStrictFetch(): { restore: () => void; getReceiver: () => unknown } {
    const realFetch = globalThis.fetch;
    let observedReceiver: unknown = 'unset';
    function strictFetch(this: unknown, _input: unknown, _init?: unknown): Promise<Response> {
      observedReceiver = this;
      if (this !== globalThis) {
        throw new TypeError("Failed to execute 'fetch' on 'Window': Illegal invocation");
      }
      return Promise.resolve(
        new Response('{"blobs":[]}', {
          status: 200,
          headers: { 'content-type': 'application/json' },
        }),
      );
    }
    Object.defineProperty(globalThis, 'fetch', {
      configurable: true,
      writable: true,
      value: strictFetch,
    });
    return {
      restore: () =>
        Object.defineProperty(globalThis, 'fetch', {
          configurable: true,
          writable: true,
          value: realFetch,
        }),
      getReceiver: () => observedReceiver,
    };
  }

  test('LongPollBridge default path passes globalThis as `this`', async () => {
    const { restore, getReceiver } = installStrictFetch();
    try {
      const id = await generateIdentityKeyPair(crypto);
      const bridge = new LongPollBridge({
        baseUrl: 'http://example.invalid',
        auth: { crypto, signingPrivateKey: id.signingPrivateKey, address: 'foo' },
        pollTimeoutMs: 100,
        requestTimeoutMs: 200,
        errorBackoffMs: 50,
        disableLoop: true,
      });
      await bridge.connect({ onMessage: () => {} });
      expect(getReceiver()).toBe(globalThis);
      await bridge.disconnect();
    } finally {
      restore();
    }
  });

  test('SseBridge default path passes globalThis as `this`', async () => {
    // For SSE we only need to know the very first fetch was bound; the
    // 200-with-empty-stream response will let openOnce return cleanly,
    // and disableAutoReconnect prevents an infinite reconnect loop.
    const realFetch = globalThis.fetch;
    let observedReceiver: unknown = 'unset';
    function strictFetch(this: unknown, _input: unknown, _init?: unknown): Promise<Response> {
      observedReceiver = this;
      if (this !== globalThis) {
        throw new TypeError("Failed to execute 'fetch' on 'Window': Illegal invocation");
      }
      // Empty SSE-shaped body: stream closes immediately.
      return Promise.resolve(
        new Response('', {
          status: 200,
          headers: { 'content-type': 'text/event-stream' },
        }),
      );
    }
    Object.defineProperty(globalThis, 'fetch', {
      configurable: true,
      writable: true,
      value: strictFetch,
    });
    try {
      const id = await generateIdentityKeyPair(crypto);
      const bridge = new SseBridge({
        baseUrl: 'http://example.invalid',
        auth: { crypto, signingPrivateKey: id.signingPrivateKey, address: 'foo' },
        initialBackoffMs: 50,
        maxBackoffMs: 100,
        disableAutoReconnect: true,
      });
      await bridge.connect({ onMessage: () => {} });
      expect(observedReceiver).toBe(globalThis);
      await bridge.disconnect();
    } finally {
      Object.defineProperty(globalThis, 'fetch', {
        configurable: true,
        writable: true,
        value: realFetch,
      });
    }
  });
});

// ─── V4.7 — presence events ────────────────────────────────────────

async function registerAddress(
  baseUrl: string,
  address: string,
  identity: Awaited<ReturnType<typeof generateIdentityKeyPair>>,
): Promise<void> {
  const body = await signPayload(crypto, identity.signingPrivateKey, {
    address,
    signingKey: toBase64(identity.signingPublicKey),
  });
  const res = await fetch(`${baseUrl}/v1/inbox/register`, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify(body),
  });
  expect(res.status).toBe(200);
}

describe('Presence — server emits peer_connected / peer_disconnected', () => {
  let h: Harness;
  beforeAll(async () => {
    h = await bootstrap();
    // bootstrap registers bob; alice needs registration too so her bridge
    // can authenticate against /v1/bridge/ws.
    await registerAddress(h.baseUrl, 'alice', h.alice);
  });
  afterAll(() => {
    h.server.stop(true);
  });

  test('open + close a WsBridge → events fire on the inbox event bus (acceptance 1)', async () => {
    const seen: Array<{ name: string; address: string; bridgeKind: string }> = [];
    const off = h.events.on((e) => {
      if (e.name === 'inbox.peer_connected' || e.name === 'inbox.peer_disconnected') {
        seen.push({
          name: e.name,
          address: e.data.address,
          bridgeKind: e.data.bridgeKind,
        });
      }
    });
    try {
      const ws = new WsBridge({
        baseUrl: h.baseUrl,
        auth: { crypto, signingPrivateKey: h.alice.signingPrivateKey, address: 'alice' },
        connectTimeoutMs: 2_000,
        disableAutoReconnect: true,
      });
      await ws.connect({ onMessage: () => {} });
      await waitFor(() => seen.some((s) => s.name === 'inbox.peer_connected'), 2_000);
      await ws.disconnect();
      await waitFor(() => seen.some((s) => s.name === 'inbox.peer_disconnected'), 2_000);
      expect(seen[0]).toEqual({ name: 'inbox.peer_connected', address: 'alice', bridgeKind: 'ws' });
      expect(seen[seen.length - 1]).toEqual({
        name: 'inbox.peer_disconnected',
        address: 'alice',
        bridgeKind: 'ws',
      });
    } finally {
      off();
    }
  });
});

describe('PresenceBridge — subscribe to remote presence changes', () => {
  let h: Harness;
  beforeAll(async () => {
    h = await bootstrap();
    await registerAddress(h.baseUrl, 'alice', h.alice);
  });
  afterAll(() => {
    h.server.stop(true);
  });

  test('online → offline → online over a single subscription (acceptance 2A + 4)', async () => {
    const presence = new PresenceBridge({
      baseUrl: h.baseUrl,
      crypto,
      signingPrivateKey: h.bob.signingPrivateKey,
      address: 'bob',
      initialBackoffMs: 50,
      maxBackoffMs: 200,
    });

    const changes: PresenceChange[] = [];
    const sub = await presence.subscribe({
      watch: ['alice'],
      onPresenceChange: (e) => {
        changes.push(e);
      },
    });

    try {
      // Initial snapshot — alice not yet connected.
      await waitFor(() => changes.length >= 1, 2_000);
      expect(changes[0]!.address).toBe('alice');
      expect(changes[0]!.status).toBe('offline');
      expect(changes[0]!.via).toBeUndefined();

      // Alice opens a WsBridge → bob must see online (acceptance 2A).
      const aliceWs = new WsBridge({
        baseUrl: h.baseUrl,
        auth: { crypto, signingPrivateKey: h.alice.signingPrivateKey, address: 'alice' },
        connectTimeoutMs: 2_000,
        disableAutoReconnect: true,
      });
      await aliceWs.connect({ onMessage: () => {} });
      await waitFor(
        () => changes.some((c) => c.status === 'online' && c.address === 'alice'),
        2_000,
      );
      const onlineFrame = changes.find((c) => c.status === 'online' && c.address === 'alice')!;
      expect(onlineFrame.via).toBe('ws');

      // Alice's bridge drops → bob must see offline.
      await aliceWs.disconnect();
      await waitFor(
        () =>
          changes.filter((c) => c.address === 'alice' && c.status === 'offline').length >= 2,
        2_000,
      );

      // Reconnect: alice reopens → bob sees online again (acceptance 4).
      const aliceWs2 = new WsBridge({
        baseUrl: h.baseUrl,
        auth: { crypto, signingPrivateKey: h.alice.signingPrivateKey, address: 'alice' },
        connectTimeoutMs: 2_000,
        disableAutoReconnect: true,
      });
      await aliceWs2.connect({ onMessage: () => {} });
      await waitFor(
        () => changes.filter((c) => c.address === 'alice' && c.status === 'online').length >= 2,
        2_000,
      );
      await aliceWs2.disconnect();
    } finally {
      await sub.unsubscribe();
    }
  });

  test('subscription on [alice] does not leak carol (acceptance 3)', async () => {
    const carol = await generateIdentityKeyPair(crypto);
    await registerAddress(h.baseUrl, 'carol', carol);

    const presence = new PresenceBridge({
      baseUrl: h.baseUrl,
      crypto,
      signingPrivateKey: h.bob.signingPrivateKey,
      address: 'bob',
      initialBackoffMs: 50,
      maxBackoffMs: 200,
    });

    const changes: PresenceChange[] = [];
    const sub = await presence.subscribe({
      watch: ['alice'],
      onPresenceChange: (e) => {
        changes.push(e);
      },
    });
    try {
      // Drain the initial snapshot for alice.
      await waitFor(() => changes.length >= 1, 2_000);
      const baseline = changes.length;

      // Carol opens a bridge — bob's alice-only subscription must NOT see her.
      const carolWs = new WsBridge({
        baseUrl: h.baseUrl,
        auth: { crypto, signingPrivateKey: carol.signingPrivateKey, address: 'carol' },
        connectTimeoutMs: 2_000,
        disableAutoReconnect: true,
      });
      await carolWs.connect({ onMessage: () => {} });
      // Give the server time to emit + filter.
      await new Promise((r) => setTimeout(r, 250));
      await carolWs.disconnect();
      await new Promise((r) => setTimeout(r, 100));

      const newFrames = changes.slice(baseline);
      for (const f of newFrames) {
        expect(f.address).toBe('alice');
        expect(f.address).not.toBe('carol');
      }
    } finally {
      await sub.unsubscribe();
    }
  });

  test('addPeer / removePeer mutate the watched set via reconnect', async () => {
    const carol = await generateIdentityKeyPair(crypto);
    await registerAddress(h.baseUrl, 'carol-2', carol);

    const presence = new PresenceBridge({
      baseUrl: h.baseUrl,
      crypto,
      signingPrivateKey: h.bob.signingPrivateKey,
      address: 'bob',
      initialBackoffMs: 50,
      maxBackoffMs: 200,
    });

    const changes: PresenceChange[] = [];
    const sub = await presence.subscribe({
      watch: ['alice'],
      onPresenceChange: (e) => {
        changes.push(e);
      },
    });
    try {
      await waitFor(() => changes.length >= 1, 2_000);
      expect(sub.watching()).toEqual(['alice']);

      // Add carol-2 — reconnect should deliver a fresh snapshot that
      // includes the new address.
      await sub.addPeer('carol-2');
      await waitFor(() => changes.some((c) => c.address === 'carol-2'), 2_000);
      expect(sub.watching().sort()).toEqual(['alice', 'carol-2']);

      const carolWs = new WsBridge({
        baseUrl: h.baseUrl,
        auth: { crypto, signingPrivateKey: carol.signingPrivateKey, address: 'carol-2' },
        connectTimeoutMs: 2_000,
        disableAutoReconnect: true,
      });
      await carolWs.connect({ onMessage: () => {} });
      await waitFor(
        () => changes.some((c) => c.address === 'carol-2' && c.status === 'online'),
        2_000,
      );
      await carolWs.disconnect();

      // removePeer → carol-2 events must stop arriving.
      await sub.removePeer('carol-2');
      const baseline = changes.filter((c) => c.address === 'carol-2').length;
      const carolWs2 = new WsBridge({
        baseUrl: h.baseUrl,
        auth: { crypto, signingPrivateKey: carol.signingPrivateKey, address: 'carol-2' },
        connectTimeoutMs: 2_000,
        disableAutoReconnect: true,
      });
      await carolWs2.connect({ onMessage: () => {} });
      await new Promise((r) => setTimeout(r, 250));
      await carolWs2.disconnect();
      await new Promise((r) => setTimeout(r, 100));
      const after = changes.filter((c) => c.address === 'carol-2').length;
      expect(after).toBe(baseline);
      expect(sub.watching()).toEqual(['alice']);
    } finally {
      await sub.unsubscribe();
    }
  });
});

// ─── V4.8 — sender-fingerprint propagation on bridge push ──────────

describe('Sender attribution — bridge push surfaces IncomingMessage.from', () => {
  test('SSE push carries from = 8-byte hex of SHA-256(senderSigningKey)', async () => {
    const h = await bootstrap();
    try {
      const received: IncomingMessage[] = [];
      const bridge = new SseBridge({
        baseUrl: h.baseUrl,
        auth: bobAuth(h),
        initialBackoffMs: 50,
        maxBackoffMs: 200,
        disableAutoReconnect: true,
      });
      await bridge.connect({ onMessage: (m) => received.push(m) });
      try {
        await putBlob(h, rand(48));
        await waitFor(() => received.length === 1, 5_000);
        const fp = received[0]!.from;
        expect(fp).toMatch(/^[0-9a-f]{16}$/);
        const digest = await globalThis.crypto.subtle.digest(
          'SHA-256',
          h.alice.signingPublicKey as unknown as ArrayBuffer,
        );
        const expected = Array.from(new Uint8Array(digest).slice(0, 8), (b) =>
          b.toString(16).padStart(2, '0'),
        ).join('');
        expect(fp).toBe(expected);
      } finally {
        await bridge.disconnect();
      }
    } finally {
      h.server.stop(true);
    }
  });

  test('WS push carries from likewise', async () => {
    const h = await bootstrap();
    try {
      const received: IncomingMessage[] = [];
      const bridge = new WsBridge({
        baseUrl: h.baseUrl,
        auth: bobAuth(h),
        connectTimeoutMs: 2_000,
        disableAutoReconnect: true,
      });
      await bridge.connect({ onMessage: (m) => received.push(m) });
      try {
        await putBlob(h, rand(48));
        await waitFor(() => received.length === 1, 5_000);
        expect(received[0]!.from).toMatch(/^[0-9a-f]{16}$/);
      } finally {
        await bridge.disconnect();
      }
    } finally {
      h.server.stop(true);
    }
  });

  test('long-poll push carries from likewise', async () => {
    const h = await bootstrap();
    try {
      const received: IncomingMessage[] = [];
      const bridge = new LongPollBridge({
        baseUrl: h.baseUrl,
        auth: bobAuth(h),
        pollTimeoutMs: 500,
        requestTimeoutMs: 1_500,
        errorBackoffMs: 50,
      });
      await bridge.connect({ onMessage: (m) => received.push(m) });
      try {
        await putBlob(h, rand(48));
        await waitFor(() => received.length === 1, 5_000);
        expect(received[0]!.from).toMatch(/^[0-9a-f]{16}$/);
      } finally {
        await bridge.disconnect();
      }
    } finally {
      h.server.stop(true);
    }
  });
});

// ─── V4.8.4 — cross-channel dedup via shared BridgeDeliveryLog ────────

describe('BridgeDeliveryLog — bridge push suppresses subsequent inbox-poll for the same msgId', () => {
  test('WS push then /v1/inbox/:addr/fetch: fetch returns 0 blobs but advances cursor', async () => {
    // Reproduces the Prism FR `cross-channel-duplicate-fanout-v4.8.2`
    // (re-verified on 4.8.3): a single inbox.send was being delivered
    // both via the WS bridge AND via the next inbox-poll cycle, the
    // duplicate dispatch tripping on already-consumed prekeys.
    // V4.8.4's shared BridgeDeliveryLog records every successful
    // bridge push and the inbox-fetch route filters those msgIds out
    // for the grace window — so a recipient that runs both a bridge
    // and a poll cycle observes exactly one delivery.
    const h = await bootstrap();
    try {
      const received: IncomingMessage[] = [];
      const bridge = new WsBridge({
        baseUrl: h.baseUrl,
        auth: bobAuth(h),
        connectTimeoutMs: 2_000,
        disableAutoReconnect: true,
      });
      await bridge.connect({ onMessage: (m) => received.push(m) });
      try {
        const msgId = await putBlob(h, rand(48));
        await waitFor(() => received.length === 1, 2_000);
        // Give the bridge handler a tick to record the push in the log
        // (it happens after the await on ws.send returns).
        await new Promise((r) => setTimeout(r, 50));

        // Now do a regular inbox-fetch as if the recipient's
        // `Inbox.pollOnce` cycle fired. With V4.8.4 wiring, the
        // bridge-pushed msgId is filtered out.
        const body = await signPayload(crypto, h.bob.signingPrivateKey, {
          address: 'bob',
          sinceCursor: 0,
        });
        const res = await fetch(`${h.baseUrl}/v1/inbox/bob/fetch`, {
          method: 'POST',
          headers: { 'content-type': 'application/json' },
          body: JSON.stringify(body),
        });
        const fetchJson = (await res.json()) as { blobs: unknown[]; cursor: number };
        expect(fetchJson.blobs.length).toBe(0);
        // Cursor advances past the suppressed blob so the next poll
        // doesn't re-fetch the same range and stay stuck.
        expect(fetchJson.cursor).toBeGreaterThan(0);
        expect(received[0]!.msgId).toBe(msgId);
      } finally {
        await bridge.disconnect();
      }
    } finally {
      h.server.stop(true);
    }
  });

  test('SSE push also records into the log (parity with WS)', async () => {
    const h = await bootstrap();
    try {
      const received: IncomingMessage[] = [];
      const bridge = new SseBridge({
        baseUrl: h.baseUrl,
        auth: bobAuth(h),
        initialBackoffMs: 50,
        maxBackoffMs: 200,
        disableAutoReconnect: true,
      });
      await bridge.connect({ onMessage: (m) => received.push(m) });
      try {
        await putBlob(h, rand(48));
        await waitFor(() => received.length === 1, 2_000);
        await new Promise((r) => setTimeout(r, 50));
        const body = await signPayload(crypto, h.bob.signingPrivateKey, {
          address: 'bob',
          sinceCursor: 0,
        });
        const res = await fetch(`${h.baseUrl}/v1/inbox/bob/fetch`, {
          method: 'POST',
          headers: { 'content-type': 'application/json' },
          body: JSON.stringify(body),
        });
        const fetchJson = (await res.json()) as { blobs: unknown[] };
        expect(fetchJson.blobs.length).toBe(0);
      } finally {
        await bridge.disconnect();
      }
    } finally {
      h.server.stop(true);
    }
  });

  test('a non-bridge-pushed msgId is still returned by inbox-fetch', async () => {
    // Negative control: blobs that the bridge never pushed (e.g. the
    // bridge wasn't connected when the put landed) must still come
    // through the inbox-fetch path. The filter is bridge-delivered-
    // specific, not a blanket suppression.
    const h = await bootstrap();
    try {
      // No bridge connected.
      const msgId = await putBlob(h, rand(48));
      const body = await signPayload(crypto, h.bob.signingPrivateKey, {
        address: 'bob',
        sinceCursor: 0,
      });
      const res = await fetch(`${h.baseUrl}/v1/inbox/bob/fetch`, {
        method: 'POST',
        headers: { 'content-type': 'application/json' },
        body: JSON.stringify(body),
      });
      const fetchJson = (await res.json()) as { blobs: Array<{ msgId: string }> };
      expect(fetchJson.blobs.length).toBe(1);
      expect(fetchJson.blobs[0]!.msgId).toBe(msgId);
    } finally {
      h.server.stop(true);
    }
  });
});

// ─── V4.8.2 — per-connection msgId dedup (Prism FR: duplicate fan-out) ─

describe('Bridge dedup — single PUT yields exactly one push per connection', () => {
  test('WS: storming inbox.blob_stored does not duplicate frames for one msgId', async () => {
    const h = await bootstrap();
    try {
      const received: IncomingMessage[] = [];
      const bridge = new WsBridge({
        baseUrl: h.baseUrl,
        auth: bobAuth(h),
        connectTimeoutMs: 2_000,
        disableAutoReconnect: true,
      });
      await bridge.connect({ onMessage: (m) => received.push(m) });
      try {
        // One real PUT + replay the inbox.blob_stored event ten times to
        // simulate any future code path (or external bug) that double-
        // fires the trigger. The cursor in flushTo would already cover
        // the happy case, but the per-connection LRU is the explicit
        // dedup gate that survives even if cursor logic regresses.
        const msgId = await putBlob(h, rand(48));
        for (let i = 0; i < 10; i++) {
          h.events.emit('inbox.blob_stored', {
            address: 'bob',
            msgId,
            bytes: 48,
            ttlSeconds: 60,
          });
        }
        await waitFor(() => received.length >= 1, 2_000);
        // Give any stragglers a chance to arrive and inflate the count.
        await new Promise((r) => setTimeout(r, 250));
        expect(received.length).toBe(1);
        expect(received[0]!.msgId).toBe(msgId);
      } finally {
        await bridge.disconnect();
      }
    } finally {
      h.server.stop(true);
    }
  });

  test('SSE: same dedup contract', async () => {
    const h = await bootstrap();
    try {
      const received: IncomingMessage[] = [];
      const bridge = new SseBridge({
        baseUrl: h.baseUrl,
        auth: bobAuth(h),
        initialBackoffMs: 50,
        maxBackoffMs: 200,
        disableAutoReconnect: true,
      });
      await bridge.connect({ onMessage: (m) => received.push(m) });
      try {
        const msgId = await putBlob(h, rand(48));
        for (let i = 0; i < 10; i++) {
          h.events.emit('inbox.blob_stored', {
            address: 'bob',
            msgId,
            bytes: 48,
            ttlSeconds: 60,
          });
        }
        await waitFor(() => received.length >= 1, 2_000);
        await new Promise((r) => setTimeout(r, 250));
        expect(received.length).toBe(1);
        expect(received[0]!.msgId).toBe(msgId);
      } finally {
        await bridge.disconnect();
      }
    } finally {
      h.server.stop(true);
    }
  });
});