hivecast-sdk/packages/federation/tests/unit/cross_realm_client.spec.ts

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