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( '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 }); });