hivecast-sdk/packages/federation/tests/integration/inbound-routing.spec.ts

// packages/federation/tests/integration/inbound-routing.spec.ts
//
// FP-21-3: FederationPeer inbound → $inbox
//
// Predicate: FederationPeer converts inbound env.to.path (old per-op format
// like 'daemon.accepts.$introspect') into the actor's $inbox topic with op
// extracted into the MxEnvelope.
//
// 1. Inbound daemon.accepts.$introspect → actor's $inbox with op='$introspect'
// 2. Inbound daemon.accepts.scan → actor's $inbox with op='scan'
// 3. Reply reaches sender's ingress topic with matching correlationId
// 4. Two concurrent requests each get their own correct reply

import { describe, test, beforeEach, afterEach } from 'node:test';
import assert from 'node:assert/strict';
import {
  FederationPeer,
  ExactInstanceResolver,
} from '@open-matrix/federation/federation_peer';
import { InMemoryFilterBroker } from '@open-matrix/federation/transport/inmem';
import {
  getMxIngressTopic,
  getInternalTopic,
  createMxEnvV1,
  serializeMxEnvV1,
} from '../../src';

function delay(ms: number): Promise<void> {
  return new Promise(r => setTimeout(r, ms));
}

const REALM = 'test-realm';
const CLIENT_REALM = 'client-realm';

describe('FP-21-3: FederationPeer inbound → $inbox', () => {
  let backbone: InMemoryFilterBroker;
  let localBus: InMemoryFilterBroker;
  let peer: FederationPeer;

  beforeEach(async () => {
    backbone = new InMemoryFilterBroker();
    localBus = new InMemoryFilterBroker();
    peer = new FederationPeer({
      localRealm: REALM,
      localBus,
      backbone,
      internalTimeoutMs: 2000,
    });
    await peer.start();
  });

  afterEach(() => {
    peer.stop();
  });

  test('routes daemon.accepts.$introspect to $inbox with op=$introspect', async () => {
    let received: Record<string, unknown> | null = null;

    // Subscribe to the $inbox topic (what a real actor would listen on)
    localBus.subscribe('daemon/$inbox', (_t, p) => {
      received = JSON.parse(p);
    });

    // Send inbound command with old per-op path format
    const cmd = createMxEnvV1({
      msgType: 'Cmd',
      toRealm: REALM,
      toPath: 'daemon.accepts.$introspect',
      fromRealm: CLIENT_REALM,
      fromPath: '_client',
      correlationId: 'test-corr-1',
      replyToRealm: CLIENT_REALM,
      replyToPath: '_replies.emits.1',
      body: { depth: 1 },
    });

    backbone.publish(getMxIngressTopic(REALM, '00'), serializeMxEnvV1(cmd));
    await delay(50);

    assert.ok(received, 'Should receive on $inbox topic');
    const r = received as Record<string, unknown>;
    assert.equal(r.op, '$introspect', 'op should be $introspect');
    assert.deepEqual(r.payload, { depth: 1 }, 'payload should match body');
    assert.equal(r.correlationId, 'test-corr-1', 'correlationId should match');
    assert.ok(r.replyTo, 'Should have replyTo');
  });

  test('routes daemon.accepts.scan to $inbox with op=scan', async () => {
    let received: Record<string, unknown> | null = null;

    localBus.subscribe('daemon/$inbox', (_t, p) => {
      received = JSON.parse(p);
    });

    const cmd = createMxEnvV1({
      msgType: 'Cmd',
      toRealm: REALM,
      toPath: 'daemon.accepts.scan',
      fromRealm: CLIENT_REALM,
      fromPath: '_client',
      correlationId: 'test-corr-2',
      replyToRealm: CLIENT_REALM,
      replyToPath: '_replies.emits.2',
      body: { path: '/home' },
    });

    backbone.publish(getMxIngressTopic(REALM, '00'), serializeMxEnvV1(cmd));
    await delay(50);

    assert.ok(received, 'Should receive on $inbox topic');
    const r = received as Record<string, unknown>;
    assert.equal(r.op, 'scan', 'op should be scan');
    assert.deepEqual(r.payload, { path: '/home' }, 'payload should match body');
  });

  test('reply reaches sender with matching correlationId', async () => {
    // Set up handler on $inbox that sends reply
    localBus.subscribe('component/$inbox', (_t, p) => {
      const req = JSON.parse(p);
      localBus.publish(req.replyTo, JSON.stringify({
        correlationId: req.correlationId,
        ok: true,
        result: { type: 'TestComponent', mount: 'component' },
      }));
    });

    // Capture reply on backbone
    let reply: Record<string, unknown> | null = null;
    backbone.subscribe(getMxIngressTopic(CLIENT_REALM, '00'), (_t, p) => {
      reply = JSON.parse(p);
    });

    const cmd = createMxEnvV1({
      msgType: 'Cmd',
      toRealm: REALM,
      toPath: 'component.accepts.$introspect',
      fromRealm: CLIENT_REALM,
      fromPath: '_client',
      correlationId: 'reply-test-corr',
      replyToRealm: CLIENT_REALM,
      replyToPath: '_replies.emits.reply-test',
      body: { depth: 0 },
    });

    backbone.publish(getMxIngressTopic(REALM, '00'), serializeMxEnvV1(cmd));
    await delay(100);

    assert.ok(reply, 'Should receive reply on backbone');
    const r = reply as Record<string, unknown>;
    assert.equal(r.msgType, 'Result', 'reply should be Result type');
    assert.equal(r.ok, true, 'reply should be ok');
    assert.equal(r.correlationId, 'reply-test-corr', 'correlationId should match');

    const body = r.body as { type: string; mount: string };
    assert.equal(body.type, 'TestComponent', 'reply body should match');
    assert.equal(body.mount, 'component', 'reply body should match');
  });

  test('two concurrent requests get their own correct replies', async () => {
    // Handler on $inbox dispatches by op
    localBus.subscribe('svc/$inbox', (_t, p) => {
      const req = JSON.parse(p);
      let result: unknown;
      if (req.op === '$introspect') {
        result = { type: 'ServiceActor', mount: 'svc' };
      } else if (req.op === 'ping') {
        result = { message: 'pong', echo: req.payload };
      } else {
        result = { error: 'unknown op' };
      }
      localBus.publish(req.replyTo, JSON.stringify({
        correlationId: req.correlationId,
        ok: true,
        result,
      }));
    });

    // Capture replies
    const replies: Record<string, unknown>[] = [];
    backbone.subscribe(getMxIngressTopic(CLIENT_REALM, '00'), (_t, p) => {
      replies.push(JSON.parse(p));
    });

    // Send two concurrent requests with different ops
    const cmd1 = createMxEnvV1({
      msgType: 'Cmd',
      toRealm: REALM,
      toPath: 'svc.accepts.$introspect',
      fromRealm: CLIENT_REALM,
      fromPath: '_client',
      correlationId: 'concurrent-1',
      replyToRealm: CLIENT_REALM,
      replyToPath: '_replies.emits.c1',
      body: { depth: 1 },
    });

    const cmd2 = createMxEnvV1({
      msgType: 'Cmd',
      toRealm: REALM,
      toPath: 'svc.accepts.ping',
      fromRealm: CLIENT_REALM,
      fromPath: '_client',
      correlationId: 'concurrent-2',
      replyToRealm: CLIENT_REALM,
      replyToPath: '_replies.emits.c2',
      body: { msg: 'hello' },
    });

    backbone.publish(getMxIngressTopic(REALM, '00'), serializeMxEnvV1(cmd1));
    backbone.publish(getMxIngressTopic(REALM, '00'), serializeMxEnvV1(cmd2));
    await delay(100);

    assert.equal(replies.length, 2, 'Should receive 2 replies');

    // Find replies by correlationId
    const r1 = replies.find(r => r.correlationId === 'concurrent-1');
    const r2 = replies.find(r => r.correlationId === 'concurrent-2');

    assert.ok(r1, 'Should have reply for concurrent-1');
    assert.ok(r2, 'Should have reply for concurrent-2');

    const body1 = (r1 as Record<string, unknown>).body as { type: string };
    assert.equal(body1.type, 'ServiceActor', 'introspect reply should have correct type');

    const body2 = (r2 as Record<string, unknown>).body as { message: string };
    assert.equal(body2.message, 'pong', 'ping reply should have pong');
  });
});