stack/packages/comms/src/matrix-client.ts

/**
 * Minimal Matrix Client-Server API client for the P1 presence slice.
 *
 * Deliberately tiny: only the calls presence needs (whoami, set native
 * presence, send a timeline event, read recent timeline). Auth is a single
 * bearer token; an optional `actAsUserId` enables Application-Service
 * masquerade (`?user_id=`) so the P1 provisioner can drive several virtual
 * agents with one as_token in dev (RFC-001 §2.2 step 4/Appendix A). Agents
 * holding their own access_token simply omit `actAsUserId`.
 *
 * `fetch` is injectable for unit tests.
 */

import crypto from 'node:crypto';

import type {
  HeartbeatObservation,
  MatrixPresence,
  PresenceHeartbeatContent,
  PresenceStatus,
} from './types.js';

export interface MatrixClientConfig {
  /** Client-Server API base, e.g. https://matrix.localhost:8448 */
  homeserverUrl: string;
  /** Bearer token (a per-agent access_token, or an as_token for masquerade). */
  accessToken: string;
  /** If set, all calls masquerade as this MXID via ?user_id= (AS mode). */
  actAsUserId?: string;
}

export class MatrixError extends Error {
  constructor(
    readonly status: number,
    readonly errcode: string | undefined,
    message: string,
  ) {
    super(message);
    this.name = 'MatrixError';
  }
}

type FetchLike = typeof fetch;

/** Map our authoritative liveness state to the native Matrix presence EDU. */
export function toMatrixPresence(status: PresenceStatus): MatrixPresence {
  switch (status) {
    case 'online':
      return 'online';
    case 'away':
      return 'unavailable';
    case 'offline':
      return 'offline';
  }
}

export class MinimalMatrixClient {
  private readonly fetchImpl: FetchLike;

  constructor(
    private readonly cfg: MatrixClientConfig,
    fetchImpl?: FetchLike,
  ) {
    this.fetchImpl = fetchImpl ?? fetch;
  }

  private async request(
    method: string,
    path: string,
    options: { query?: Record<string, string>; body?: unknown } = {},
  ): Promise<Record<string, unknown>> {
    const url = new URL(this.cfg.homeserverUrl.replace(/\/$/, '') + path);
    if (this.cfg.actAsUserId) {
      url.searchParams.set('user_id', this.cfg.actAsUserId);
    }
    for (const [k, v] of Object.entries(options.query ?? {})) {
      url.searchParams.set(k, v);
    }
    const res = await this.fetchImpl(url, {
      method,
      headers: {
        Authorization: `Bearer ${this.cfg.accessToken}`,
        'Content-Type': 'application/json',
      },
      body: options.body === undefined ? undefined : JSON.stringify(options.body),
    });
    const text = await res.text();
    const data = (text ? JSON.parse(text) : {}) as Record<string, unknown>;
    if (!res.ok) {
      throw new MatrixError(
        res.status,
        typeof data.errcode === 'string' ? data.errcode : undefined,
        `${method} ${path} -> ${res.status}: ${text.slice(0, 300)}`,
      );
    }
    return data;
  }

  /** GET /account/whoami — resolves the acting MXID. */
  async whoami(): Promise<string> {
    const data = await this.request('GET', '/_matrix/client/v3/account/whoami');
    if (typeof data.user_id !== 'string') {
      throw new MatrixError(500, undefined, 'whoami returned no user_id');
    }
    return data.user_id;
  }

  /**
   * Set the native Matrix presence EDU (so Element shows the right dot for
   * humans). NOT the authoritative liveness signal — the heartbeat is.
   */
  async setPresence(userId: string, status: PresenceStatus, statusMsg?: string): Promise<void> {
    const user = encodeURIComponent(userId);
    await this.request('PUT', `/_matrix/client/v3/presence/${user}/status`, {
      body: {
        presence: toMatrixPresence(status),
        ...(statusMsg ? { status_msg: statusMsg } : {}),
      },
    });
  }

  /** Send an arbitrary timeline event; returns its event_id. */
  async sendEvent(
    roomId: string,
    eventType: string,
    content: Record<string, unknown>,
  ): Promise<string> {
    const room = encodeURIComponent(roomId);
    const txn = `mosaic-comms-${crypto.randomUUID()}`;
    const data = await this.request(
      'PUT',
      `/_matrix/client/v3/rooms/${room}/send/${encodeURIComponent(eventType)}/${txn}`,
      { body: content },
    );
    if (typeof data.event_id !== 'string') {
      throw new MatrixError(500, undefined, 'send returned no event_id');
    }
    return data.event_id;
  }

  /** Post a `mosaic.presence` heartbeat (m.room.message carrier) to the room. */
  async sendHeartbeat(roomId: string, content: PresenceHeartbeatContent): Promise<string> {
    return this.sendEvent(roomId, 'm.room.message', content as unknown as Record<string, unknown>);
  }

  /** Join a room (by id or alias). Idempotent on the server. */
  async joinRoom(roomIdOrAlias: string): Promise<string> {
    const data = await this.request(
      'POST',
      `/_matrix/client/v3/join/${encodeURIComponent(roomIdOrAlias)}`,
      { body: {} },
    );
    if (typeof data.room_id !== 'string') {
      throw new MatrixError(500, undefined, 'join returned no room_id');
    }
    return data.room_id;
  }

  /**
   * Read recent `mosaic.presence` heartbeats from a room and reduce them to the
   * latest observation per agent. Walks the timeline backwards (most-recent
   * first) and keeps, per slug, the beat with the highest seq.
   *
   * `lastSeenTs` uses the server's `origin_server_ts` (honest "when we last
   * heard from it"), falling back to the agent-stamped envelope `ts`.
   */
  async readHeartbeats(roomId: string, limit = 200): Promise<HeartbeatObservation[]> {
    const room = encodeURIComponent(roomId);
    const data = await this.request('GET', `/_matrix/client/v3/rooms/${room}/messages`, {
      query: { dir: 'b', limit: String(limit) },
    });
    const chunk = Array.isArray(data.chunk) ? (data.chunk as Array<Record<string, unknown>>) : [];
    const bySlug = new Map<string, HeartbeatObservation>();

    for (const ev of chunk) {
      const content = ev.content as Record<string, unknown> | undefined;
      if (!content || content.msgtype !== 'mosaic.presence') continue;
      const agent = content.agent as Record<string, unknown> | undefined;
      const slug = agent && typeof agent.slug === 'string' ? agent.slug : undefined;
      const mxid =
        agent && typeof agent.mxid === 'string'
          ? agent.mxid
          : typeof ev.sender === 'string'
            ? ev.sender
            : undefined;
      if (!slug || !mxid) continue;

      const seq = typeof content.seq === 'number' ? content.seq : 0;
      const serverTs = typeof ev.origin_server_ts === 'number' ? ev.origin_server_ts : undefined;
      const envelopeTs = typeof content.ts === 'number' ? content.ts : undefined;
      const lastSeenTs = serverTs ?? envelopeTs ?? 0;
      const assertedStatus =
        content.status === 'online' || content.status === 'away' || content.status === 'offline'
          ? (content.status as PresenceStatus)
          : 'offline';

      const prev = bySlug.get(slug);
      if (!prev || seq > prev.lastSeq) {
        bySlug.set(slug, { slug, mxid, lastSeenTs, lastSeq: seq, assertedStatus });
      }
    }
    return [...bySlug.values()];
  }
}