stack/packages/mosaic/src/fleet/matrix-native-runtime-transport.ts

import { createHash } from 'node:crypto';
import type {
  RuntimeHealth,
  RuntimeMessage,
  RuntimeScope,
  RuntimeSessionState,
  RuntimeStreamEvent,
} from '@mosaicstack/types';

export type MatrixRuntimeTransportErrorCode =
  | 'forbidden'
  | 'invalid_request'
  | 'not_found'
  | 'unavailable';

export class MatrixRuntimeTransportError extends Error {
  constructor(
    readonly code: MatrixRuntimeTransportErrorCode,
    message: string,
  ) {
    super(message);
    this.name = MatrixRuntimeTransportError.name;
  }
}

/** Minimal injectable Matrix fetch surface; it avoids coupling this transport to an SDK. */
export interface MatrixFetchLike {
  (
    url: string,
    init?: { method?: string; headers?: Record<string, string>; body?: string },
  ): Promise<{
    ok: boolean;
    status: number;
    json(): Promise<unknown>;
    text(): Promise<string>;
  }>;
}

/** A server-configured runtime session binding. Callers never select a Matrix room. */
export interface MatrixNativeRuntimeSessionBinding {
  id: string;
  runtimeId: string;
  roomId: string;
  remoteUserId: string;
  parentSessionId?: string;
  state: RuntimeSessionState;
  createdAt: string;
  updatedAt: string;
}

export interface MatrixNativeRuntimeTransportOptions {
  homeserverUrl: string;
  accessToken: string;
  /** Matrix user authenticated by the service token. */
  userId: string;
  sessions: readonly MatrixNativeRuntimeSessionBinding[];
  fetchImpl?: MatrixFetchLike;
  now?: () => Date;
}

interface MatrixSyncEvent {
  type?: string;
  sender?: string;
  origin_server_ts?: number;
  content?: Record<string, unknown>;
}

interface MatrixSyncResponse {
  next_batch?: string;
  rooms?: { join?: Record<string, { timeline?: { events?: MatrixSyncEvent[] } }> };
}

/**
 * Concrete Matrix CS-API transport for the native provider. It authenticates
 * the configured sender before each operation and resolves rooms exclusively
 * from configured bindings, preventing client-selected room/identity routing.
 */
export class MatrixNativeRuntimeTransport {
  private readonly baseUrl: string;
  private readonly fetchImpl: MatrixFetchLike;
  private readonly now: () => Date;
  private readonly bindings: ReadonlyMap<string, MatrixNativeRuntimeSessionBinding>;

  constructor(private readonly options: MatrixNativeRuntimeTransportOptions) {
    this.baseUrl = validatedHomeserverUrl(options.homeserverUrl);
    if (!options.accessToken.trim()) {
      throw new MatrixRuntimeTransportError('invalid_request', 'Matrix access token is required');
    }
    if (!options.userId.trim()) {
      throw new MatrixRuntimeTransportError('invalid_request', 'Matrix user identity is required');
    }
    this.bindings = new Map(
      options.sessions.map((binding) => [binding.id, Object.freeze({ ...binding })]),
    );
    if (this.bindings.size !== options.sessions.length) {
      throw new MatrixRuntimeTransportError(
        'invalid_request',
        'Matrix runtime session IDs must be unique',
      );
    }
    this.fetchImpl = options.fetchImpl ?? (globalThis.fetch as unknown as MatrixFetchLike);
    this.now = options.now ?? (() => new Date());
  }

  async health(_scope: RuntimeScope): Promise<RuntimeHealth> {
    try {
      const versions = await this.request('/_matrix/client/versions', { method: 'GET' });
      if (!versions.ok) {
        return this.healthResult('down', `versions HTTP ${versions.status}`);
      }
      await this.assertIdentity();
      return this.healthResult('healthy');
    } catch (error: unknown) {
      return this.healthResult('down', message(error));
    }
  }

  async listSessions(scope: RuntimeScope): Promise<MatrixNativeRuntimeSessionBinding[]> {
    await this.assertIdentity();
    void scope;
    return [...this.bindings.values()].map((binding) => ({ ...binding }));
  }

  async verifySession(
    sessionId: string,
    scope: RuntimeScope,
  ): Promise<MatrixNativeRuntimeSessionBinding> {
    await this.assertIdentity();
    void scope;
    const binding = this.bindings.get(sessionId);
    if (!binding) {
      throw new MatrixRuntimeTransportError(
        'not_found',
        'Matrix runtime session is not configured',
      );
    }
    return { ...binding };
  }

  async *stream(
    sessionId: string,
    cursor: string | undefined,
    scope: RuntimeScope,
  ): AsyncIterable<RuntimeStreamEvent> {
    const binding = await this.verifySession(sessionId, scope);
    const query = new URLSearchParams({ timeout: '0' });
    if (cursor?.trim()) query.set('since', cursor);
    const response = await this.request(`/_matrix/client/v3/sync?${query.toString()}`, {
      method: 'GET',
      headers: this.authHeaders(),
    });
    if (!response.ok) {
      throw new MatrixRuntimeTransportError(
        'unavailable',
        `Matrix sync failed: HTTP ${response.status}`,
      );
    }
    const payload = (await response.json()) as MatrixSyncResponse;
    const nextCursor = payload.next_batch?.trim() || cursor?.trim() || 'initial';
    const events = payload.rooms?.join?.[binding.roomId]?.timeline?.events ?? [];
    for (const event of events) {
      const normalized = normalizeEvent(event, binding, nextCursor);
      if (normalized) yield normalized;
    }
  }

  async send(sessionId: string, message: RuntimeMessage, scope: RuntimeScope): Promise<void> {
    const binding = await this.verifySession(sessionId, scope);
    const txnId = transactionId('send', binding.id, message.idempotencyKey);
    const response = await this.request(
      `/_matrix/client/v3/rooms/${encodeURIComponent(binding.roomId)}/send/m.room.message/${encodeURIComponent(txnId)}`,
      {
        method: 'PUT',
        headers: this.authHeaders(),
        body: JSON.stringify({
          msgtype: 'm.text',
          body: message.content,
          'mosaic.runtime.v1': metadata(binding, scope, message.idempotencyKey),
        }),
      },
    );
    if (!response.ok) {
      throw new MatrixRuntimeTransportError(
        'unavailable',
        `Matrix message delivery failed: HTTP ${response.status}`,
      );
    }
  }

  async terminate(sessionId: string, approvalRef: string, scope: RuntimeScope): Promise<void> {
    const binding = await this.verifySession(sessionId, scope);
    const txnId = transactionId('terminate', binding.id, `${approvalRef}:${scope.correlationId}`);
    const response = await this.request(
      `/_matrix/client/v3/rooms/${encodeURIComponent(binding.roomId)}/send/mosaic.runtime.terminate/${encodeURIComponent(txnId)}`,
      {
        method: 'PUT',
        headers: this.authHeaders(),
        body: JSON.stringify({
          ...metadata(binding, scope),
          approval_ref: approvalRef,
        }),
      },
    );
    if (!response.ok) {
      throw new MatrixRuntimeTransportError(
        'unavailable',
        `Matrix termination delivery failed: HTTP ${response.status}`,
      );
    }
  }

  private async assertIdentity(): Promise<void> {
    let response: Awaited<ReturnType<MatrixFetchLike>>;
    try {
      response = await this.request('/_matrix/client/v3/account/whoami', {
        method: 'GET',
        headers: this.authHeaders(),
      });
    } catch (error: unknown) {
      throw new MatrixRuntimeTransportError('unavailable', message(error));
    }
    if (!response.ok) {
      throw new MatrixRuntimeTransportError(
        'forbidden',
        `Matrix whoami failed: HTTP ${response.status}`,
      );
    }
    const body = (await response.json()) as { user_id?: unknown };
    if (body.user_id !== this.options.userId) {
      throw new MatrixRuntimeTransportError(
        'forbidden',
        'Matrix whoami identity does not match configuration',
      );
    }
  }

  private request(
    path: string,
    init: { method?: string; headers?: Record<string, string>; body?: string },
  ) {
    return this.fetchImpl(`${this.baseUrl}${path}`, init);
  }

  private authHeaders(): Record<string, string> {
    return {
      Authorization: `Bearer ${this.options.accessToken}`,
      'Content-Type': 'application/json',
    };
  }

  private healthResult(status: RuntimeHealth['status'], detail?: string): RuntimeHealth {
    return { status, checkedAt: this.now().toISOString(), ...(detail ? { detail } : {}) };
  }
}

function metadata(
  binding: MatrixNativeRuntimeSessionBinding,
  scope: RuntimeScope,
  idempotencyKey?: string,
): Record<string, string> {
  return {
    session_id: binding.id,
    runtime_id: binding.runtimeId,
    actor_id: scope.actorId,
    tenant_id: scope.tenantId,
    channel_id: scope.channelId,
    correlation_id: scope.correlationId,
    ...(idempotencyKey ? { idempotency_key: idempotencyKey } : {}),
  };
}

function transactionId(kind: 'send' | 'terminate', sessionId: string, identity: string): string {
  return `mosaic-${kind}-${createHash('sha256').update(`${sessionId}\u0000${identity}`).digest('base64url')}`;
}

function normalizeEvent(
  event: MatrixSyncEvent,
  binding: MatrixNativeRuntimeSessionBinding,
  cursor: string,
): RuntimeStreamEvent | undefined {
  if (event.type !== 'mosaic.runtime.event' || event.sender !== binding.remoteUserId)
    return undefined;
  const payload = event.content?.['mosaic.runtime.v1'];
  if (
    !isRecord(payload) ||
    payload['session_id'] !== binding.id ||
    typeof payload['type'] !== 'string'
  ) {
    return undefined;
  }
  const occurredAt =
    typeof payload['occurred_at'] === 'string'
      ? payload['occurred_at']
      : new Date(event.origin_server_ts ?? 0).toISOString();
  const eventCursor = typeof payload['cursor'] === 'string' ? payload['cursor'] : cursor;
  switch (payload['type']) {
    case 'session.state':
      return isState(payload['state'])
        ? {
            type: 'session.state',
            sessionId: binding.id,
            cursor: eventCursor,
            occurredAt,
            state: payload['state'],
          }
        : undefined;
    case 'message.delta':
      return typeof payload['content'] === 'string'
        ? {
            type: 'message.delta',
            sessionId: binding.id,
            cursor: eventCursor,
            occurredAt,
            content: payload['content'],
          }
        : undefined;
    case 'message.complete':
      return typeof payload['message_id'] === 'string'
        ? {
            type: 'message.complete',
            sessionId: binding.id,
            cursor: eventCursor,
            occurredAt,
            messageId: payload['message_id'],
          }
        : undefined;
    default:
      return undefined;
  }
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === 'object' && value !== null;
}

function isState(value: unknown): value is RuntimeSessionState {
  return (
    value === 'starting' ||
    value === 'active' ||
    value === 'idle' ||
    value === 'stopped' ||
    value === 'failed'
  );
}

function validatedHomeserverUrl(value: string): string {
  let url: URL;
  try {
    url = new URL(value);
  } catch {
    throw new MatrixRuntimeTransportError(
      'invalid_request',
      'Matrix homeserver URL must be absolute',
    );
  }
  if (url.protocol !== 'https:') {
    throw new MatrixRuntimeTransportError(
      'invalid_request',
      'Matrix homeserver URL must use HTTPS',
    );
  }
  return url.toString().replace(/\/$/, '');
}

function message(error: unknown): string {
  return error instanceof Error ? error.message : 'Matrix transport request failed';
}