/** * Layer 3: CrossRealmClient Unit Tests * * Tests the cross-realm RPC client using mailbox pattern. */ import test from 'node:test'; import assert from 'node:assert/strict'; import { CrossRealmClient } from '@open-matrix/federation/cross_realm_client'; import { WIRE_PROFILES } from '@open-matrix/federation/wire_profile'; import { deserializeTaggedEnvelope } from '@open-matrix/federation/envelope_codec'; import { InMemoryExactBroker } from '@open-matrix/federation/transport/inmem'; import { getMxIngressTopic, getLegacyIngressTopic } from '@open-matrix/federation/topics'; // ===== Basic CrossRealmClient setup tests ===== test('CrossRealmClient: constructs with required options', () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.mxenv1, }); assert.ok(client); }); test('CrossRealmClient: start subscribes to ingress topics', () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.mxenv1, }); // Track subscriptions const originalSubscribe = transport.subscribe.bind(transport); const capturedTopics: string[] = []; transport.subscribe = (topic: string, handler: any) => { capturedTopics.push(topic); return originalSubscribe(topic, handler); }; client.start(); // Strangler Fig pattern: subscribes to BOTH legacy AND mxenv1 ingress assert.ok(capturedTopics.length >= 1, 'Should subscribe to at least one topic'); // Should include either legacy or mxenv1 format const hasLegacy = capturedTopics.some(t => t.startsWith('public.ingress.')); const hasMxEnv1 = capturedTopics.some(t => t.includes('.$ingress.')); assert.ok(hasLegacy || hasMxEnv1, 'Should subscribe to ingress topics'); }); test('CrossRealmClient: stop unsubscribes', () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.mxenv1, }); client.start(); client.stop(); // Should not throw assert.ok(true); }); // ===== Envelope format tests ===== test('CrossRealmClient: uses MxEnv1 envelope format when configured', async () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.mxenv1, }); client.start(); // Capture published message let publishedPayload: string | null = null; const originalPublish = transport.publish.bind(transport); transport.publish = (topic: string, payload: string) => { if (publishedPayload === null) publishedPayload = payload; // Capture FIRST publish (ingress envelope) return originalPublish(topic, payload); }; // Fire-and-forget invoke (will timeout, but we capture the envelope) const invokePromise = client.invoke('target-realm', 'component.accepts.test', { data: 'hello' }, 100); // Wait a tick for publish to happen await new Promise(r => setTimeout(r, 10)); // Ignore timeout error await invokePromise.catch(() => {}); assert.ok(publishedPayload !== null, 'Should have published a message'); const parsed = deserializeTaggedEnvelope(publishedPayload!); assert.ok(parsed !== null, 'First published message should be a valid MxEnv1 envelope'); assert.equal(parsed.profile, 'mxenv1', 'Should use MxEnv1 profile'); }); test('CrossRealmClient: uses legacy envelope format when configured', async () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.matrix3, }); client.start(); // Capture published message let publishedPayload: string | null = null; const originalPublish = transport.publish.bind(transport); transport.publish = (topic: string, payload: string) => { publishedPayload = payload; return originalPublish(topic, payload); }; // Fire-and-forget invoke const invokePromise = client.invoke('target-realm', 'component.accepts.test', { data: 'hello' }, 100); await new Promise(r => setTimeout(r, 10)); await invokePromise.catch(() => {}); assert.ok(publishedPayload !== null, 'Should have published a message'); const parsed = deserializeTaggedEnvelope(publishedPayload!); assert.ok(parsed !== null); assert.equal(parsed.profile, 'matrix3', 'Should use matrix3 profile'); }); // ===== Topic format tests ===== test('getMxIngressTopic: generates correct MxEnv1 ingress topic', () => { const topic = getMxIngressTopic('flowpad-demo', '00'); assert.equal(topic, 'flowpad-demo.$ingress.00'); }); test('getMxIngressTopic: handles realm with special characters', () => { const topic = getMxIngressTopic('flowpad.tab.abc123', '00'); assert.equal(topic, 'flowpad.tab.abc123.$ingress.00'); }); test('getLegacyIngressTopic: generates correct legacy ingress topic', () => { const topic = getLegacyIngressTopic('flowpad-demo'); // Legacy format uses mxesc1 encoding assert.ok(topic.startsWith('public.ingress.')); assert.ok(topic.includes('flowpad')); }); // ===== Error handling tests ===== test('CrossRealmClient: invoke rejects on timeout', async () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.mxenv1, }); client.start(); // No responder - should timeout await assert.rejects( () => client.invoke('target-realm', 'component.accepts.noResponse', {}, 100), /timeout/i ); }); // ===== Reply handling tests ===== test('CrossRealmClient: resolves on successful mxenv1 reply', async () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.mxenv1, }); client.start(); // Capture the request to get correlationId let capturedRequest: any = null; const originalPublish = transport.publish.bind(transport); transport.publish = (topic: string, payload: string) => { capturedRequest = JSON.parse(payload); return originalPublish(topic, payload); }; // Start invoke const invokePromise = client.invoke<{ result: string }>( 'target-realm', 'component.accepts.greet', { name: 'World' }, 5000 ); // Wait for publish await new Promise(r => setTimeout(r, 50)); assert.ok(capturedRequest !== null, 'Should have captured request'); // Simulate reply from target realm to mxenv1 ingress topic const mxIngressTopic = getMxIngressTopic('flowpad-demo', '00'); const correlationId = capturedRequest.correlationId || capturedRequest.msgId; const reply = { ver: 1, msgType: 'Result', msgId: 'reply-123', timestamp: new Date().toISOString(), to: { realm: 'flowpad-demo', path: `_replies.emits.${correlationId}` }, from: { realm: 'target-realm', path: 'component' }, correlationId: correlationId, body: { result: 'Hello, World!' }, ok: true, }; transport.publish(mxIngressTopic, JSON.stringify(reply)); const result = await invokePromise; assert.ok(result); assert.equal((result as any).result, 'Hello, World!'); }); test('CrossRealmClient: resolves on successful legacy reply', async () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.matrix3, }); client.start(); // Capture the request let capturedRequest: any = null; const originalPublish = transport.publish.bind(transport); transport.publish = (topic: string, payload: string) => { capturedRequest = JSON.parse(payload); return originalPublish(topic, payload); }; // Start invoke const invokePromise = client.invoke<{ result: string }>( 'target-realm', 'component.accepts.greet', { name: 'World' }, 5000 ); // Wait for publish await new Promise(r => setTimeout(r, 50)); assert.ok(capturedRequest !== null, 'Should have captured request'); // Simulate legacy reply const legacyIngressTopic = getLegacyIngressTopic('flowpad-demo'); const correlationId = capturedRequest.correlationId; const reply = { toRealm: 'flowpad-demo', toPath: capturedRequest.replyTo, fromRealm: 'target-realm', correlationId: correlationId, body: { result: 'Hello, World!' }, ok: true, timestamp: new Date().toISOString(), }; transport.publish(legacyIngressTopic, JSON.stringify(reply)); const result = await invokePromise; assert.ok(result); assert.equal((result as any).result, 'Hello, World!'); }); test('CrossRealmClient: rejects on error reply', async () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.mxenv1, }); client.start(); // Capture the request let capturedRequest: any = null; const originalPublish = transport.publish.bind(transport); transport.publish = (topic: string, payload: string) => { capturedRequest = JSON.parse(payload); return originalPublish(topic, payload); }; // Start invoke const invokePromise = client.invoke( 'target-realm', 'component.accepts.failing', {}, 5000 ); // Wait for publish await new Promise(r => setTimeout(r, 50)); // Simulate error reply const mxIngressTopic = getMxIngressTopic('flowpad-demo', '00'); const correlationId = capturedRequest.correlationId || capturedRequest.msgId; const reply = { ver: 1, msgType: 'Result', msgId: 'reply-err', timestamp: new Date().toISOString(), to: { realm: 'flowpad-demo', path: `_replies.emits.${correlationId}` }, from: { realm: 'target-realm', path: 'component' }, correlationId: correlationId, body: null, ok: false, error: { code: 'MX.NOT_FOUND', message: 'Resource not found', }, }; transport.publish(mxIngressTopic, JSON.stringify(reply)); await assert.rejects( invokePromise, /not found/i ); }); // ===== Concurrent invoke tests ===== test('CrossRealmClient: handles concurrent invokes with unique correlationIds', async () => { const transport = new InMemoryExactBroker(); const client = new CrossRealmClient({ localRealm: 'flowpad-demo', transport, profile: WIRE_PROFILES.mxenv1, }); client.start(); // Capture all published messages const capturedEnvelopes: any[] = []; const originalPublish = transport.publish.bind(transport); transport.publish = (topic: string, payload: string) => { capturedEnvelopes.push(JSON.parse(payload)); return originalPublish(topic, payload); }; // Fire multiple concurrent invokes const promises = [ client.invoke('target-realm', 'svc.accepts.a', { id: 1 }, 100).catch(() => {}), client.invoke('target-realm', 'svc.accepts.b', { id: 2 }, 100).catch(() => {}), client.invoke('target-realm', 'svc.accepts.c', { id: 3 }, 100).catch(() => {}), ]; await new Promise(r => setTimeout(r, 50)); await Promise.all(promises); // Each invoke dual-publishes (ingress + local mailbox fallback) assert.equal(capturedEnvelopes.length, 6, 'Each invoke dual-publishes (ingress + local fallback)'); const ids = capturedEnvelopes.map(e => e.correlationId || e.msgId).filter(Boolean); const uniqueIds = new Set(ids); assert.equal(uniqueIds.size, 3, 'Should have 3 unique correlation IDs across 6 envelopes'); }); // ===== Profile behavior tests ===== test('WIRE_PROFILES: mxenv1 subscribes to both ingress types (Strangler Fig)', () => { const profile = WIRE_PROFILES.mxenv1; assert.equal(profile.subscribeLegacyIngress, true); assert.equal(profile.subscribeMxIngress, true); assert.equal(profile.encode, 'mxenv1'); }); test('WIRE_PROFILES: matrix3 subscribes to both ingress types (Strangler Fig)', () => { const profile = WIRE_PROFILES.matrix3; assert.equal(profile.subscribeLegacyIngress, true); assert.equal(profile.subscribeMxIngress, true); assert.equal(profile.encode, 'matrix3'); });