hivecast-sdk/packages/federation/tests/integration/comprehensive-scenarios.spec.ts

/**
 * 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<string, unknown> = 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();
    });
  });
});