T-HUB/apps/archive-bot/src/matrix.ts

import type { Logger } from "pino";
import { isMatrixEvent, parseMxc, type MatrixEvent, type RoomEncryption } from "./events.js";
import { PermanentArchiveError, parseContentLength } from "./inspector.js";

export class MatrixHttpError extends Error {
  constructor(
    readonly status: number,
    readonly errcode: string | null,
    message: string,
  ) {
    super(message);
  }
}

export interface JoinedRoomSync {
  timeline: { events: MatrixEvent[]; limited: boolean; prevBatch: string | null };
  stateEvents: MatrixEvent[];
}

export interface SyncResult {
  nextBatch: string;
  joined: Map<string, JoinedRoomSync>;
  invited: string[];
}

export interface MediaDownload {
  body: ReadableStream<Uint8Array>;
  contentLength: number | null;
}

/** The narrow Matrix surface the processor needs; faked in tests. */
export interface MatrixPort {
  readonly userId: string;
  sync(since: string | null, signal: AbortSignal): Promise<SyncResult>;
  /** Older events between `from` (a prev_batch) back to `to` (the previous sync token), oldest first. */
  backfill(roomId: string, from: string, to: string, signal: AbortSignal): Promise<MatrixEvent[]>;
  roomEncryption(roomId: string, signal: AbortSignal): Promise<RoomEncryption>;
  joinRoom(roomId: string, signal: AbortSignal): Promise<void>;
  downloadMedia(mxcUri: string, signal: AbortSignal): Promise<MediaDownload>;
  sendNotice(roomId: string, txnId: string, content: Record<string, unknown>, signal: AbortSignal): Promise<void>;
  /** Phase-B export gate: current `m.room.power_levels` for a mapped room. */
  roomPowerLevels(roomId: string, signal: AbortSignal): Promise<Record<string, unknown>>;
  /** Every room the account has joined (encryption lint input). */
  joinedRooms(signal: AbortSignal): Promise<string[]>;
}

const SYNC_TIMEOUT_MS = 30_000;
const MAX_REDIRECTS = 3;
const BACKFILL_PAGES = 20;
const TIMELINE_TYPES = ["m.room.message", "m.room.redaction", "m.room.encrypted", "m.room.encryption"];

/** Client-side filter: mapped rooms only, minimal timeline types, no presence/ephemeral noise. */
export function syncFilter(roomIds: readonly string[]): Record<string, unknown> {
  return {
    presence: { not_types: ["*"] },
    account_data: { not_types: ["*"] },
    room: {
      rooms: [...roomIds],
      account_data: { not_types: ["*"] },
      ephemeral: { not_types: ["*"] },
      state: { types: ["m.room.encryption"], lazy_load_members: true },
      timeline: { limit: 50, types: TIMELINE_TYPES },
    },
  };
}

/** Authenticated media requires spec v1.11+ (or the MSC3916 stable flag on older advertisers). */
export function supportsAuthenticatedMedia(versions: unknown): boolean {
  if (typeof versions !== "object" || versions === null) return false;
  const record = versions as { versions?: unknown; unstable_features?: unknown };
  const list = Array.isArray(record.versions) ? record.versions.filter((v): v is string => typeof v === "string") : [];
  const modern = list.some((v) => {
    const match = /^v1\.(\d+)$/.exec(v);
    return match !== null && Number(match[1]) >= 11;
  });
  const features = record.unstable_features;
  const flagged = typeof features === "object" && features !== null
    && (features as Record<string, unknown>)["org.matrix.msc3916.stable"] === true;
  return modern || flagged;
}

/**
 * Redirect policy for media downloads. Bearer auth is only ever sent to the
 * homeserver origin; other hosts must be explicitly allowlisted (e.g. the R2
 * endpoint) and are fetched without credentials. Downgrades to http are refused.
 */
export function resolveRedirect(
  current: URL,
  location: string,
  homeserver: URL,
  allowedHosts: readonly string[],
): { url: URL; sendAuth: boolean } {
  const url = new URL(location, current);
  if (url.origin === homeserver.origin) return { url, sendAuth: true };
  if (url.protocol !== "https:") throw new PermanentArchiveError(`refusing non-https media redirect to ${url.host}`);
  if (!allowedHosts.includes(url.hostname.toLowerCase())) {
    throw new PermanentArchiveError(`media redirect to unapproved host ${url.hostname}`);
  }
  return { url, sendAuth: false };
}

function events(value: unknown): MatrixEvent[] {
  return Array.isArray(value) ? value.filter(isMatrixEvent) : [];
}

function record(value: unknown): Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value) ? (value as Record<string, unknown>) : {};
}

export function parseSync(body: unknown): SyncResult {
  const root = record(body);
  if (typeof root.next_batch !== "string") throw new Error("sync response missing next_batch");
  const rooms = record(root.rooms);
  const joined = new Map<string, JoinedRoomSync>();
  for (const [roomId, raw] of Object.entries(record(rooms.join))) {
    const room = record(raw);
    const timeline = record(room.timeline);
    joined.set(roomId, {
      timeline: {
        events: events(timeline.events),
        limited: timeline.limited === true,
        prevBatch: typeof timeline.prev_batch === "string" ? timeline.prev_batch : null,
      },
      stateEvents: events(record(room.state).events),
    });
  }
  return { nextBatch: root.next_batch, joined, invited: Object.keys(record(rooms.invite)) };
}

export interface MatrixClientOptions {
  homeserver: URL;
  user: string;
  password: string;
  deviceId: string;
  mediaRedirectHosts: string[];
  roomIds: string[];
  logger: Logger;
}

/** Plain client-server sync-loop client (not an appservice). No crypto: encrypted content is never touched. */
export class MatrixClient implements MatrixPort {
  private accessToken: string | null = null;
  private resolvedUserId = "";
  private readonly filter: string;

  constructor(private readonly opts: MatrixClientOptions) {
    this.filter = JSON.stringify(syncFilter(opts.roomIds));
  }

  get userId(): string {
    return this.resolvedUserId;
  }

  async login(signal: AbortSignal): Promise<void> {
    const versions = await this.request("GET", "/_matrix/client/versions", { signal, auth: false });
    if (!supportsAuthenticatedMedia(versions)) {
      throw new Error("homeserver does not support authenticated media (v1.11); refusing to start");
    }
    const body = record(await this.request("POST", "/_matrix/client/v3/login", {
      signal,
      auth: false,
      json: {
        type: "m.login.password",
        identifier: { type: "m.id.user", user: this.opts.user },
        password: this.opts.password,
        device_id: this.opts.deviceId,
        initial_device_display_name: "Hub archive bot",
      },
    }));
    if (typeof body.access_token !== "string" || typeof body.user_id !== "string") {
      throw new Error("login response missing access_token/user_id");
    }
    this.accessToken = body.access_token;
    this.resolvedUserId = body.user_id;
  }

  async sync(since: string | null, signal: AbortSignal): Promise<SyncResult> {
    const query = new URLSearchParams({ filter: this.filter, timeout: String(since === null ? 0 : SYNC_TIMEOUT_MS) });
    if (since !== null) query.set("since", since);
    const timeout = AbortSignal.any([signal, AbortSignal.timeout(SYNC_TIMEOUT_MS + 30_000)]);
    return parseSync(await this.request("GET", `/_matrix/client/v3/sync?${query.toString()}`, { signal: timeout }));
  }

  async backfill(roomId: string, from: string, to: string, signal: AbortSignal): Promise<MatrixEvent[]> {
    const collected: MatrixEvent[] = [];
    let cursor: string | null = from;
    for (let page = 0; page < BACKFILL_PAGES && cursor !== null; page += 1) {
      const query = new URLSearchParams({
        from: cursor,
        to,
        dir: "b",
        limit: "100",
        filter: JSON.stringify({ types: TIMELINE_TYPES }),
      });
      const body = record(await this.request(
        "GET",
        `/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/messages?${query.toString()}`,
        { signal },
      ));
      const chunk = events(body.chunk);
      collected.push(...chunk);
      cursor = chunk.length > 0 && typeof body.end === "string" ? body.end : null;
    }
    if (cursor !== null) this.opts.logger.warn({ roomId }, "matrix: backfill page budget exhausted; older gap events skipped");
    return collected.reverse();
  }

  async roomEncryption(roomId: string, signal: AbortSignal): Promise<RoomEncryption> {
    try {
      const body = record(await this.request(
        "GET",
        `/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/state/m.room.encryption/`,
        { signal },
      ));
      // Any present encryption state (even an unknown algorithm) means encrypted.
      return Object.keys(body).length > 0 ? "encrypted" : "unknown";
    } catch (err) {
      if (err instanceof MatrixHttpError && err.status === 404 && err.errcode === "M_NOT_FOUND") return "unencrypted";
      if (err instanceof MatrixHttpError && (err.status === 403 || err.status === 404)) return "unknown";
      throw err;
    }
  }

  async joinRoom(roomId: string, signal: AbortSignal): Promise<void> {
    await this.request("POST", `/_matrix/client/v3/join/${encodeURIComponent(roomId)}`, { signal, json: {} });
  }

  async sendNotice(roomId: string, txnId: string, content: Record<string, unknown>, signal: AbortSignal): Promise<void> {
    await this.request(
      "PUT",
      `/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(txnId)}`,
      { signal, json: content },
    );
  }

  async roomPowerLevels(roomId: string, signal: AbortSignal): Promise<Record<string, unknown>> {
    const body = record(await this.request(
      "GET",
      `/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/state/m.room.power_levels/`,
      { signal },
    ));
    return body;
  }

  async joinedRooms(signal: AbortSignal): Promise<string[]> {
    const body = record(await this.request("GET", "/_matrix/client/v3/joined_rooms", { signal }));
    const joined = body.joined_rooms;
    return Array.isArray(joined) ? joined.filter((r): r is string => typeof r === "string") : [];
  }

  /** Authenticated media endpoint only; the URL is derived from a validated mxc URI, never user-supplied HTTP. */
  async downloadMedia(mxcUri: string, signal: AbortSignal): Promise<MediaDownload> {
    const { serverName, mediaId } = parseMxc(mxcUri);
    let url = new URL(
      `/_matrix/client/v1/media/download/${encodeURIComponent(serverName)}/${encodeURIComponent(mediaId)}?allow_redirect=true`,
      this.opts.homeserver,
    );
    let sendAuth = true;
    for (let hop = 0; hop <= MAX_REDIRECTS; hop += 1) {
      const headers: Record<string, string> = {};
      if (sendAuth) headers.authorization = `Bearer ${this.token()}`;
      const res = await fetch(url, { headers, redirect: "manual", signal });
      if (res.status >= 300 && res.status < 400) {
        const location = res.headers.get("location");
        await res.body?.cancel();
        if (location === null) throw new Error("media redirect without Location");
        ({ url, sendAuth } = resolveRedirect(url, location, this.opts.homeserver, this.opts.mediaRedirectHosts));
        continue;
      }
      if (!res.ok || res.body === null) {
        const error = await this.toError(res);
        // Missing/forbidden/too-large media will not appear on retry.
        if (res.status === 404 || res.status === 403 || res.status === 413) throw new PermanentArchiveError(error.message);
        throw error;
      }
      return { body: res.body, contentLength: parseContentLength(res.headers.get("content-length")) };
    }
    throw new PermanentArchiveError("too many media redirects");
  }

  private token(): string {
    if (this.accessToken === null) throw new Error("matrix client not logged in");
    return this.accessToken;
  }

  private async request(
    method: string,
    path: string,
    opts: { signal: AbortSignal; auth?: boolean; json?: unknown },
  ): Promise<unknown> {
    const headers: Record<string, string> = { accept: "application/json" };
    if (opts.auth !== false) headers.authorization = `Bearer ${this.token()}`;
    if (opts.json !== undefined) headers["content-type"] = "application/json";
    const res = await fetch(new URL(path, this.opts.homeserver), {
      method,
      headers,
      body: opts.json === undefined ? undefined : JSON.stringify(opts.json),
      redirect: "error",
      signal: opts.signal,
    });
    if (!res.ok) throw await this.toError(res);
    return res.json() as Promise<unknown>;
  }

  private async toError(res: Response): Promise<MatrixHttpError> {
    let errcode: string | null = null;
    let message = `matrix HTTP ${res.status}`;
    try {
      const body = record(await res.json());
      if (typeof body.errcode === "string") errcode = body.errcode;
      if (typeof body.error === "string") message = `${message}: ${body.error}`;
    } catch {
      // Non-JSON error body; status is enough.
    }
    return new MatrixHttpError(res.status, errcode, message);
  }
}