/** * Comprehensive Federation Scenarios * * Tests ALL edge cases and formats: * - Both wire profiles (matrix3 and mxenv1) * - Timeout handling * - Error responses * - Edge cases */ import { describe, it } from 'node:test'; import { strict as assert } from 'node:assert'; import { FederationPeer, ExactInstanceResolver, WIRE_PROFILES, InMemoryFilterBroker, getLegacyIngressTopic, getMxIngressTopic, createLegacyRequest, serializeLegacyEnvelope, createMxEnvV1, serializeMxEnvV1, } from '@open-matrix/federation'; describe('Comprehensive Federation Scenarios', () => { // =========================================================================== // MXENV/1 Format Tests (New Envelope) // =========================================================================== describe('MXENV/1 Wire Format', () => { it('cross-realm request/reply with mxenv1 format', async () => { const REALM_A = 'mxenv-realm-a'; const REALM_B = 'mxenv-realm-b'; const backbone = new InMemoryFilterBroker(); const localBusA = new InMemoryFilterBroker(); const localBusB = new InMemoryFilterBroker(); const gatewayA = new FederationPeer({ localRealm: REALM_A, profile: WIRE_PROFILES.mxenv1, backbone, localBus: localBusA, internalTimeoutMs: 5000, }); const gatewayB = new FederationPeer({ localRealm: REALM_B, profile: WIRE_PROFILES.mxenv1, backbone, localBus: localBusB, internalTimeoutMs: 5000, }); gatewayA.start(); gatewayB.start(); // Handler on Realm B localBusB.subscribe(`service.accepts.greet`, (topic, payload) => { const req = JSON.parse(payload) as { replyTo: string; correlationId: string; params: { name: string }; }; localBusB.publish(req.replyTo, JSON.stringify({ ok: true, correlationId: req.correlationId, result: { greeting: `Hello, ${req.params.name}!` }, })); }); // Listen for reply at Realm A let receivedReply: unknown = null; backbone.subscribe(getMxIngressTopic(REALM_A), (topic, payload) => { const parsed = JSON.parse(payload); if (parsed.msgType === 'Result' || parsed.msgType === 'Nack') { receivedReply = parsed; } }); await new Promise(r => setTimeout(r, 100)); // Send MXENV/1 request from A to B const request = createMxEnvV1({ msgType: 'Cmd', toRealm: REALM_B, toPath: 'service.accepts.greet', fromRealm: REALM_A, fromPath: 'client', replyToRealm: REALM_A, replyToPath: '_inbox', correlationId: 'mxenv-cross-123', body: { name: 'World' }, }); backbone.publish(getMxIngressTopic(REALM_B), serializeMxEnvV1(request)); await new Promise(r => setTimeout(r, 500)); assert.ok(receivedReply, 'Should receive MXENV/1 reply'); assert.strictEqual((receivedReply as { msgType: string }).msgType, 'Result'); assert.deepStrictEqual( (receivedReply as { body: { greeting: string } }).body, { greeting: 'Hello, World!' } ); console.log('✓ MXENV/1 cross-realm: Hello, World!'); gatewayA.stop(); gatewayB.stop(); }); it('dual-profile gateway accepts both matrix3 and mxenv1', async () => { const REALM = 'dual-profile-realm'; const backbone = new InMemoryFilterBroker(); const localBus = new InMemoryFilterBroker(); // Create gateway that subscribes to BOTH mailbox formats const gateway = new FederationPeer({ localRealm: REALM, profile: WIRE_PROFILES.dualRail, // Accepts both formats backbone, localBus, internalTimeoutMs: 5000, }); gateway.start(); let requestCount = 0; localBus.subscribe(`handler.accepts.ping`, (topic, payload) => { requestCount++; const req = JSON.parse(payload) as { replyTo: string; correlationId: string }; localBus.publish(req.replyTo, JSON.stringify({ ok: true, correlationId: req.correlationId, result: { pong: requestCount }, })); }); await new Promise(r => setTimeout(r, 100)); // Send legacy (matrix3) request const legacyReq = createLegacyRequest({ toRealm: REALM, toPath: 'handler.accepts.ping', fromRealm: 'legacy-client', replyTo: '_inbox', correlationId: 'legacy-1', body: {}, }); backbone.publish(getLegacyIngressTopic(REALM), serializeLegacyEnvelope(legacyReq)); await new Promise(r => setTimeout(r, 200)); // Send mxenv1 request const mxReq = createMxEnvV1({ msgType: 'Cmd', toRealm: REALM, toPath: 'handler.accepts.ping', fromRealm: 'mxenv-client', fromPath: 'c', replyToRealm: 'mxenv-client', replyToPath: '_inbox', correlationId: 'mxenv-1', body: {}, }); backbone.publish(getMxIngressTopic(REALM), serializeMxEnvV1(mxReq)); await new Promise(r => setTimeout(r, 200)); assert.strictEqual(requestCount, 2, 'Should receive both legacy and mxenv1 requests'); console.log('✓ Dual-profile gateway accepts both formats'); gateway.stop(); }); }); // =========================================================================== // Timeout Handling // =========================================================================== describe('Timeout Handling', () => { it('returns timeout error when handler does not reply', async () => { const REALM = 'timeout-realm'; const CLIENT_REALM = 'timeout-client'; const backbone = new InMemoryFilterBroker(); const localBus = new InMemoryFilterBroker(); const gateway = new FederationPeer({ localRealm: REALM, profile: WIRE_PROFILES.matrix3, backbone, localBus, internalTimeoutMs: 200, // Very short timeout for testing }); gateway.start(); // Handler that NEVER replies localBus.subscribe(`slow.accepts.cmd`, () => { // Do nothing - simulate hung handler }); // Listen for error reply let errorReply: unknown = null; backbone.subscribe(getLegacyIngressTopic(CLIENT_REALM), (topic, payload) => { errorReply = JSON.parse(payload); }); await new Promise(r => setTimeout(r, 50)); const request = createLegacyRequest({ toRealm: REALM, toPath: 'slow.accepts.cmd', fromRealm: CLIENT_REALM, replyTo: '_inbox', correlationId: 'timeout-test', body: {}, }); backbone.publish(getLegacyIngressTopic(REALM), serializeLegacyEnvelope(request)); // Wait for timeout (200ms) + buffer await new Promise(r => setTimeout(r, 400)); assert.ok(errorReply, 'Should receive timeout error reply'); assert.strictEqual((errorReply as { ok: boolean }).ok, false); assert.ok( (errorReply as { error?: { code?: string } }).error?.code?.includes('TIMEOUT'), 'Error code should indicate timeout' ); console.log('✓ Timeout handling works correctly'); gateway.stop(); }); }); // =========================================================================== // Error Response Handling // =========================================================================== describe('Error Responses', () => { it('forwards handler error back to caller', async () => { const REALM = 'error-realm'; const CLIENT_REALM = 'error-client'; const backbone = new InMemoryFilterBroker(); const localBus = new InMemoryFilterBroker(); const gateway = new FederationPeer({ localRealm: REALM, profile: WIRE_PROFILES.matrix3, backbone, localBus, }); gateway.start(); // Handler that returns an error localBus.subscribe(`failing.accepts.cmd`, (topic, payload) => { const req = JSON.parse(payload) as { replyTo: string; correlationId: string }; localBus.publish(req.replyTo, JSON.stringify({ ok: false, correlationId: req.correlationId, error: { code: 'APP.VALIDATION_ERROR', message: 'Invalid input provided', details: { field: 'email', reason: 'malformed' }, }, })); }); let errorReply: unknown = null; backbone.subscribe(getLegacyIngressTopic(CLIENT_REALM), (topic, payload) => { errorReply = JSON.parse(payload); }); await new Promise(r => setTimeout(r, 50)); const request = createLegacyRequest({ toRealm: REALM, toPath: 'failing.accepts.cmd', fromRealm: CLIENT_REALM, replyTo: '_inbox', correlationId: 'error-test', body: { email: 'not-an-email' }, }); backbone.publish(getLegacyIngressTopic(REALM), serializeLegacyEnvelope(request)); await new Promise(r => setTimeout(r, 300)); assert.ok(errorReply, 'Should receive error reply'); assert.strictEqual((errorReply as { ok: boolean }).ok, false); assert.strictEqual( (errorReply as { error?: { code?: string } }).error?.code, 'APP.VALIDATION_ERROR' ); console.log('✓ Error responses forwarded correctly'); gateway.stop(); }); }); // =========================================================================== // Edge Cases // =========================================================================== describe('Edge Cases', () => { it('handles empty body', async () => { const REALM = 'empty-body-realm'; const backbone = new InMemoryFilterBroker(); const localBus = new InMemoryFilterBroker(); const gateway = new FederationPeer({ localRealm: REALM, profile: WIRE_PROFILES.matrix3, backbone, localBus, }); gateway.start(); let receivedParams: unknown = 'NOT_SET'; localBus.subscribe(`empty.accepts.cmd`, (topic, payload) => { const req = JSON.parse(payload) as { replyTo: string; correlationId: string; params: unknown }; receivedParams = req.params; localBus.publish(req.replyTo, JSON.stringify({ ok: true, correlationId: req.correlationId, })); }); await new Promise(r => setTimeout(r, 50)); const request = createLegacyRequest({ toRealm: REALM, toPath: 'empty.accepts.cmd', fromRealm: 'client', replyTo: '_inbox', correlationId: 'empty-body', body: null, // Empty/null body }); backbone.publish(getLegacyIngressTopic(REALM), serializeLegacyEnvelope(request)); await new Promise(r => setTimeout(r, 200)); assert.strictEqual(receivedParams, null, 'Should receive null params'); console.log('✓ Empty body handled correctly'); gateway.stop(); }); it('handles large payload', async () => { const REALM = 'large-payload-realm'; const backbone = new InMemoryFilterBroker(); const localBus = new InMemoryFilterBroker(); const gateway = new FederationPeer({ localRealm: REALM, profile: WIRE_PROFILES.matrix3, backbone, localBus, }); gateway.start(); let receivedSize = 0; localBus.subscribe(`large.accepts.cmd`, (topic, payload) => { const req = JSON.parse(payload) as { replyTo: string; correlationId: string; params: { data: string } }; receivedSize = req.params.data.length; localBus.publish(req.replyTo, JSON.stringify({ ok: true, correlationId: req.correlationId, result: { receivedBytes: receivedSize }, })); }); await new Promise(r => setTimeout(r, 50)); // Create a 100KB payload const largeData = 'x'.repeat(100 * 1024); const request = createLegacyRequest({ toRealm: REALM, toPath: 'large.accepts.cmd', fromRealm: 'client', replyTo: '_inbox', correlationId: 'large-payload', body: { data: largeData }, }); backbone.publish(getLegacyIngressTopic(REALM), serializeLegacyEnvelope(request)); await new Promise(r => setTimeout(r, 300)); assert.strictEqual(receivedSize, 100 * 1024, 'Should receive full 100KB payload'); console.log('✓ Large payload (100KB) handled correctly'); gateway.stop(); }); it('handles special characters in realm name', async () => { const REALM = 'realm-with.dots_and-dashes'; const backbone = new InMemoryFilterBroker(); const localBus = new InMemoryFilterBroker(); const gateway = new FederationPeer({ localRealm: REALM, profile: WIRE_PROFILES.matrix3, backbone, localBus, }); gateway.start(); let receivedRequest = false; localBus.subscribe(`special.accepts.cmd`, (topic, payload) => { receivedRequest = true; const req = JSON.parse(payload) as { replyTo: string; correlationId: string }; localBus.publish(req.replyTo, JSON.stringify({ ok: true, correlationId: req.correlationId })); }); await new Promise(r => setTimeout(r, 50)); const request = createLegacyRequest({ toRealm: REALM, toPath: 'special.accepts.cmd', fromRealm: 'client', replyTo: '_inbox', correlationId: 'special-realm', body: {}, }); backbone.publish(getLegacyIngressTopic(REALM), serializeLegacyEnvelope(request)); await new Promise(r => setTimeout(r, 200)); assert.ok(receivedRequest, 'Should handle realm with special characters'); console.log('✓ Special characters in realm name handled'); gateway.stop(); }); it('handles concurrent requests from multiple realms', async () => { const TARGET_REALM = 'concurrent-target'; const NUM_CLIENTS = 5; const backbone = new InMemoryFilterBroker(); const localBus = new InMemoryFilterBroker(); const gateway = new FederationPeer({ localRealm: TARGET_REALM, profile: WIRE_PROFILES.matrix3, backbone, localBus, internalTimeoutMs: 5000, }); gateway.start(); let requestCount = 0; localBus.subscribe(`counter.accepts.increment`, (topic, payload) => { requestCount++; const req = JSON.parse(payload) as { replyTo: string; correlationId: string }; localBus.publish(req.replyTo, JSON.stringify({ ok: true, correlationId: req.correlationId, result: { count: requestCount }, })); }); const replies: Map = new Map(); // Subscribe to all client mailboxes for (let i = 0; i < NUM_CLIENTS; i++) { backbone.subscribe(getLegacyIngressTopic(`client-${i}`), (topic, payload) => { const parsed = JSON.parse(payload); if (parsed.correlationId) { replies.set(parsed.correlationId, parsed); } }); } await new Promise(r => setTimeout(r, 100)); // Send concurrent requests from multiple clients for (let i = 0; i < NUM_CLIENTS; i++) { const request = createLegacyRequest({ toRealm: TARGET_REALM, toPath: 'counter.accepts.increment', fromRealm: `client-${i}`, replyTo: '_inbox', correlationId: `concurrent-${i}`, body: { clientId: i }, }); backbone.publish(getLegacyIngressTopic(TARGET_REALM), serializeLegacyEnvelope(request)); } await new Promise(r => setTimeout(r, 1000)); assert.strictEqual(requestCount, NUM_CLIENTS, `Should process all ${NUM_CLIENTS} requests`); assert.strictEqual(replies.size, NUM_CLIENTS, `Should receive all ${NUM_CLIENTS} replies`); console.log(`✓ Concurrent requests from ${NUM_CLIENTS} realms handled correctly`); gateway.stop(); }); }); });