hivecast-sdk/packages/federation/tests/integration/router_e2e.spec.ts

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  FederationPeer, TabInstanceResolver,
  InMemoryFilterBroker,
} from '@open-matrix/federation';
import { getInternalTopic } from '@open-matrix/federation/topics';

/**
 * END-TO-END test for FederationPeer with InMemoryFilterBroker.
 *
 * This test verifies the FULL flow that FlowPad uses:
 * 1. Component subscribes via transport adapter (rewriteLocalTopic)
 * 2. Component publishes via transport adapter (rewriteLocalTopic + route)
 * 3. Router routes locally (rewriteLocalTopic again)
 * 4. InMemoryFilterBroker matches subscription → handler called
 *
 * If this test passes, the protocol works end-to-end.
 * If this test fails, the browser will fail too.
 */

test('E2E: FederationPeer full pub/sub flow with tab isolation', async () => {
  // ═══════════════════════════════════════════════════════════════════════════
  // Setup - mirrors what createBrowserRouter() does
  // ═══════════════════════════════════════════════════════════════════════════
  const localBroker = new InMemoryFilterBroker();

  const localRealm = 'flowpad.tab-abc123';
  const router = new FederationPeer({
    localRealm,
    localBus: localBroker,
    instanceResolver: new TabInstanceResolver(localRealm),
  });

  // ═══════════════════════════════════════════════════════════════════════════
  // Transport adapter - mirrors wrapRouterAsTransport() in setup.ts
  // ═══════════════════════════════════════════════════════════════════════════
  const rewriteLocalTopic = (topic: string): string => {
    return router.rewriteLocalTopic(topic);
  };

  const adapter = {
    publish: (topic: string, payload: unknown) => {
      const payloadStr = typeof payload === 'string' ? payload : JSON.stringify(payload);
      const rewrittenTopic = rewriteLocalTopic(topic);
      console.log(`[E2E] publish: ${topic} → ${rewrittenTopic}`);
      router.route(rewrittenTopic, payloadStr);
    },
    subscribe: (topic: string, handler: (payload: any, meta?: any) => void) => {
      const rewrittenTopic = rewriteLocalTopic(topic);
      console.log(`[E2E] subscribe: ${topic} → ${rewrittenTopic}`);
      return localBroker.subscribe(rewrittenTopic, (t: string, p: string) => {
        const meta = { topic: t, ts: Date.now() };
        try {
          handler(JSON.parse(p), meta);
        } catch {
          handler(p, meta);
        }
      });
    },
  };

  // ═══════════════════════════════════════════════════════════════════════════
  // Test - component subscribes and publishes
  // ═══════════════════════════════════════════════════════════════════════════
  const received: any[] = [];

  // Component subscribes to db.accepts.Query (base realm, no tab suffix)
  // This should be rewritten to flowpad.tab-abc123.db.accepts.Query
  adapter.subscribe('flowpad.db.accepts.Query', (payload, meta) => {
    console.log(`[E2E] handler received:`, payload, meta);
    received.push({ payload, meta });
  });

  // Component publishes to the same topic
  // This should be rewritten and routed locally
  adapter.publish('flowpad.db.accepts.Query', { table: 'products' });

  // Wait a tick for async routing
  await new Promise(resolve => setTimeout(resolve, 10));

  // ═══════════════════════════════════════════════════════════════════════════
  // Verify
  // ═══════════════════════════════════════════════════════════════════════════
  console.log(`[E2E] Received ${received.length} messages`);
  assert.equal(received.length, 1, 'Should receive exactly 1 message');
  assert.deepEqual(received[0].payload, { table: 'products' });
  assert.ok(received[0].meta.topic.includes('flowpad.tab-abc123'), 'Topic should include tab realm');
});

test('E2E: FederationPeer request/reply flow with tab isolation', async () => {
  // Setup
  const localBroker = new InMemoryFilterBroker();

  const localRealm = 'flowpad.tab-xyz789';
  const router = new FederationPeer({
    localRealm,
    localBus: localBroker,
    instanceResolver: new TabInstanceResolver(localRealm),
  });

  // Set up a responder on the local bus using browser local bus format ().
  // TabInstanceResolver returns semantic path for browser local bus.
  const browserTopic = `${localRealm}.db.accepts.Query`;
  localBroker.subscribe(browserTopic, (_topic: string, payload: string) => {
    const request = JSON.parse(payload);
    // Simulate response by publishing to replyTo
    if (request.replyTo) {
      localBroker.publish(request.replyTo, JSON.stringify({
        ok: true,
        result: [{ id: 1, name: 'Product 1' }],
      }));
    }
  });

  // Make a request using router.invoke
  const result = await router.invoke<any[]>(
    'flowpad/db.accepts.Query',
    { table: 'products' },
    1000
  );

  assert.ok(Array.isArray(result), 'Result should be an array');
  assert.equal(result[0].name, 'Product 1');
});

test('E2E: Multiple subscriptions on same topic all receive messages', async () => {
  const localBroker = new InMemoryFilterBroker();

  const router = new FederationPeer({
    localRealm: 'flowpad.tab-multi',
    localBus: localBroker,
    instanceResolver: new TabInstanceResolver('flowpad.tab-multi'),
  });

  const rewriteLocalTopic = (topic: string) => router.rewriteLocalTopic(topic);

  const adapter = {
    publish: (topic: string, payload: unknown) => {
      const payloadStr = typeof payload === 'string' ? payload : JSON.stringify(payload);
      const rewrittenTopic = rewriteLocalTopic(topic);
      router.route(rewrittenTopic, payloadStr);
    },
    subscribe: (topic: string, handler: (payload: any, meta?: any) => void) => {
      const rewrittenTopic = rewriteLocalTopic(topic);
      return localBroker.subscribe(rewrittenTopic, (t: string, p: string) => {
        try {
          handler(JSON.parse(p), { topic: t });
        } catch {
          handler(p, { topic: t });
        }
      });
    },
  };

  const received1: any[] = [];
  const received2: any[] = [];

  // Two subscribers on same topic
  adapter.subscribe('flowpad.tree.emits.stateChanged', (payload) => {
    received1.push(payload);
  });

  adapter.subscribe('flowpad.tree.emits.stateChanged', (payload) => {
    received2.push(payload);
  });

  // Publish
  adapter.publish('flowpad.tree.emits.stateChanged', { value: 42 });

  await new Promise(resolve => setTimeout(resolve, 10));

  assert.equal(received1.length, 1, 'First subscriber should receive');
  assert.equal(received2.length, 1, 'Second subscriber should receive');
  assert.deepEqual(received1[0], { value: 42 });
  assert.deepEqual(received2[0], { value: 42 });
});