Matrix: share reusable client bootstrap

extensions/matrix/src/matrix/actions/client.ts

@@ -1,18 +1,8 @@
-import { getMatrixRuntime } from "../../runtime.js";
-import type { CoreConfig } from "../../types.js";
-import { getActiveMatrixClient } from "../active-client.js";
-import {
-  createMatrixClient,
-  isBunRuntime,
-  resolveMatrixAuth,
-  resolveMatrixAuthContext,
-} from "../client.js";
+import { ensureMatrixNodeRuntime, resolveRuntimeMatrixClient } from "../client-bootstrap.js";
 import type { MatrixActionClient, MatrixActionClientOpts } from "./types.js";
 
 export function ensureNodeRuntime() {
-  if (isBunRuntime()) {
-    throw new Error("Matrix support requires Node (bun runtime not supported)");
-  }
+  ensureMatrixNodeRuntime();
 }
 
 async function ensureActionClientReadiness(
@@ -32,44 +22,16 @@ async function ensureActionClientReadiness(
 export async function resolveActionClient(
   opts: MatrixActionClientOpts = {},
 ): Promise<MatrixActionClient> {
-  ensureNodeRuntime();
-  if (opts.client) {
-    await ensureActionClientReadiness(opts.client, opts.readiness, {
-      createdForOneOff: false,
-    });
-    return { client: opts.client, stopOnDone: false };
-  }
-  const cfg = getMatrixRuntime().config.loadConfig() as CoreConfig;
-  const authContext = resolveMatrixAuthContext({
-    cfg,
+  return await resolveRuntimeMatrixClient({
+    client: opts.client,
+    timeoutMs: opts.timeoutMs,
     accountId: opts.accountId,
+    onResolved: async (client, context) => {
+      await ensureActionClientReadiness(client, opts.readiness, {
+        createdForOneOff: context.createdForOneOff,
+      });
+    },
   });
-  const active = getActiveMatrixClient(authContext.accountId);
-  if (active) {
-    await ensureActionClientReadiness(active, opts.readiness, {
-      createdForOneOff: false,
-    });
-    return { client: active, stopOnDone: false };
-  }
-  const auth = await resolveMatrixAuth({
-    cfg,
-    accountId: authContext.accountId,
-  });
-  const client = await createMatrixClient({
-    homeserver: auth.homeserver,
-    userId: auth.userId,
-    accessToken: auth.accessToken,
-    password: auth.password,
-    deviceId: auth.deviceId,
-    encryption: auth.encryption,
-    localTimeoutMs: opts.timeoutMs,
-    accountId: auth.accountId,
-    autoBootstrapCrypto: false,
-  });
-  await ensureActionClientReadiness(client, opts.readiness, {
-    createdForOneOff: true,
-  });
-  return { client, stopOnDone: true };
 }
 
 export type MatrixActionClientStopMode = "stop" | "persist";

extensions/matrix/src/matrix/client-bootstrap.ts

@@ -1,32 +1,68 @@
-import { createMatrixClient } from "./client.js";
-import type { MatrixAuth } from "./client/types.js";
+import { getMatrixRuntime } from "../runtime.js";
+import type { CoreConfig } from "../types.js";
+import { getActiveMatrixClient } from "./active-client.js";
+import {
+  createMatrixClient,
+  isBunRuntime,
+  resolveMatrixAuth,
+  resolveMatrixAuthContext,
+} from "./client.js";
+import type { MatrixClient } from "./sdk.js";
 
-type MatrixCryptoPrepare = {
-  prepare: (rooms?: string[]) => Promise<void>;
+export type ResolvedRuntimeMatrixClient = {
+  client: MatrixClient;
+  stopOnDone: boolean;
 };
 
-type MatrixBootstrapClient = Awaited<ReturnType<typeof createMatrixClient>>;
+type MatrixResolvedClientHook = (
+  client: MatrixClient,
+  context: { createdForOneOff: boolean },
+) => Promise<void> | void;
 
-export async function createPreparedMatrixClient(opts: {
-  auth: Pick<MatrixAuth, "accountId" | "homeserver" | "userId" | "accessToken" | "encryption">;
-  timeoutMs?: number;
-}): Promise<MatrixBootstrapClient> {
-  const client = await createMatrixClient({
-    homeserver: opts.auth.homeserver,
-    userId: opts.auth.userId,
-    accessToken: opts.auth.accessToken,
-    encryption: opts.auth.encryption,
-    localTimeoutMs: opts.timeoutMs,
-    accountId: opts.auth.accountId,
-  });
-  if (opts.auth.encryption && client.crypto) {
-    try {
-      const joinedRooms = await client.getJoinedRooms();
-      await (client.crypto as MatrixCryptoPrepare).prepare(joinedRooms);
-    } catch {
-      // Ignore crypto prep failures for one-off requests.
-    }
+export function ensureMatrixNodeRuntime() {
+  if (isBunRuntime()) {
+    throw new Error("Matrix support requires Node (bun runtime not supported)");
   }
-  await client.start();
-  return client;
+}
+
+export async function resolveRuntimeMatrixClient(opts: {
+  client?: MatrixClient;
+  timeoutMs?: number;
+  accountId?: string | null;
+  onResolved?: MatrixResolvedClientHook;
+}): Promise<ResolvedRuntimeMatrixClient> {
+  ensureMatrixNodeRuntime();
+  if (opts.client) {
+    await opts.onResolved?.(opts.client, { createdForOneOff: false });
+    return { client: opts.client, stopOnDone: false };
+  }
+
+  const cfg = getMatrixRuntime().config.loadConfig() as CoreConfig;
+  const authContext = resolveMatrixAuthContext({
+    cfg,
+    accountId: opts.accountId,
+  });
+  const active = getActiveMatrixClient(authContext.accountId);
+  if (active) {
+    await opts.onResolved?.(active, { createdForOneOff: false });
+    return { client: active, stopOnDone: false };
+  }
+
+  const auth = await resolveMatrixAuth({
+    cfg,
+    accountId: authContext.accountId,
+  });
+  const client = await createMatrixClient({
+    homeserver: auth.homeserver,
+    userId: auth.userId,
+    accessToken: auth.accessToken,
+    password: auth.password,
+    deviceId: auth.deviceId,
+    encryption: auth.encryption,
+    localTimeoutMs: opts.timeoutMs,
+    accountId: auth.accountId,
+    autoBootstrapCrypto: false,
+  });
+  await opts.onResolved?.(client, { createdForOneOff: true });
+  return { client, stopOnDone: true };
 }

extensions/matrix/src/matrix/send.test.ts

@@ -35,6 +35,7 @@ const runtimeStub = {
 } as unknown as PluginRuntime;
 
 let sendMessageMatrix: typeof import("./send.js").sendMessageMatrix;
+let sendTypingMatrix: typeof import("./send.js").sendTypingMatrix;
 let voteMatrixPoll: typeof import("./actions/polls.js").voteMatrixPoll;
 
 const makeClient = () => {
@@ -56,6 +57,7 @@ describe("sendMessageMatrix media", () => {
   beforeAll(async () => {
     setMatrixRuntime(runtimeStub);
     ({ sendMessageMatrix } = await import("./send.js"));
+    ({ sendTypingMatrix } = await import("./send.js"));
     ({ voteMatrixPoll } = await import("./actions/polls.js"));
   });
 
@@ -202,6 +204,7 @@ describe("sendMessageMatrix threads", () => {
   beforeAll(async () => {
     setMatrixRuntime(runtimeStub);
     ({ sendMessageMatrix } = await import("./send.js"));
+    ({ sendTypingMatrix } = await import("./send.js"));
     ({ voteMatrixPoll } = await import("./actions/polls.js"));
   });
 
@@ -376,3 +379,26 @@ describe("voteMatrixPoll", () => {
     });
   });
 });
+
+describe("sendTypingMatrix", () => {
+  beforeAll(async () => {
+    setMatrixRuntime(runtimeStub);
+    ({ sendTypingMatrix } = await import("./send.js"));
+  });
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    setMatrixRuntime(runtimeStub);
+  });
+
+  it("normalizes room-prefixed targets before sending typing state", async () => {
+    const setTyping = vi.fn().mockResolvedValue(undefined);
+    const client = {
+      setTyping,
+    } as unknown as import("./sdk.js").MatrixClient;
+
+    await sendTypingMatrix("room:!room:example", true, undefined, client);
+
+    expect(setTyping).toHaveBeenCalledWith("!room:example", true, 30_000);
+  });
+});

extensions/matrix/src/matrix/send.ts

@@ -3,7 +3,7 @@ import { getMatrixRuntime } from "../runtime.js";
 import { buildPollStartContent, M_POLL_START } from "./poll-types.js";
 import { buildMatrixReactionContent } from "./reaction-common.js";
 import type { MatrixClient } from "./sdk.js";
-import { resolveMatrixClient, resolveMediaMaxBytes } from "./send/client.js";
+import { resolveMediaMaxBytes, withResolvedMatrixClient } from "./send/client.js";
 import {
   buildReplyRelation,
   buildTextContent,
@@ -63,118 +63,116 @@ export async function sendMessageMatrix(
   if (!trimmedMessage && !opts.mediaUrl) {
     throw new Error("Matrix send requires text or media");
   }
-  const { client, stopOnDone } = await resolveMatrixClient({
-    client: opts.client,
-    timeoutMs: opts.timeoutMs,
-    accountId: opts.accountId,
-  });
-  try {
-    const roomId = await resolveMatrixRoomId(client, to);
-    const cfg = getCore().config.loadConfig();
-    const tableMode = getCore().channel.text.resolveMarkdownTableMode({
-      cfg,
-      channel: "matrix",
+  return await withResolvedMatrixClient(
+    {
+      client: opts.client,
+      timeoutMs: opts.timeoutMs,
       accountId: opts.accountId,
-    });
-    const convertedMessage = getCore().channel.text.convertMarkdownTables(
-      trimmedMessage,
-      tableMode,
-    );
-    const textLimit = getCore().channel.text.resolveTextChunkLimit(cfg, "matrix");
-    const chunkLimit = Math.min(textLimit, MATRIX_TEXT_LIMIT);
-    const chunkMode = getCore().channel.text.resolveChunkMode(cfg, "matrix", opts.accountId);
-    const chunks = getCore().channel.text.chunkMarkdownTextWithMode(
-      convertedMessage,
-      chunkLimit,
-      chunkMode,
-    );
-    const threadId = normalizeThreadId(opts.threadId);
-    const relation = threadId
-      ? buildThreadRelation(threadId, opts.replyToId)
-      : buildReplyRelation(opts.replyToId);
-    const sendContent = async (content: MatrixOutboundContent) => {
-      const eventId = await client.sendMessage(roomId, content);
-      return eventId;
-    };
+    },
+    async (client) => {
+      const roomId = await resolveMatrixRoomId(client, to);
+      const cfg = getCore().config.loadConfig();
+      const tableMode = getCore().channel.text.resolveMarkdownTableMode({
+        cfg,
+        channel: "matrix",
+        accountId: opts.accountId,
+      });
+      const convertedMessage = getCore().channel.text.convertMarkdownTables(
+        trimmedMessage,
+        tableMode,
+      );
+      const textLimit = getCore().channel.text.resolveTextChunkLimit(cfg, "matrix");
+      const chunkLimit = Math.min(textLimit, MATRIX_TEXT_LIMIT);
+      const chunkMode = getCore().channel.text.resolveChunkMode(cfg, "matrix", opts.accountId);
+      const chunks = getCore().channel.text.chunkMarkdownTextWithMode(
+        convertedMessage,
+        chunkLimit,
+        chunkMode,
+      );
+      const threadId = normalizeThreadId(opts.threadId);
+      const relation = threadId
+        ? buildThreadRelation(threadId, opts.replyToId)
+        : buildReplyRelation(opts.replyToId);
+      const sendContent = async (content: MatrixOutboundContent) => {
+        const eventId = await client.sendMessage(roomId, content);
+        return eventId;
+      };
 
-    let lastMessageId = "";
-    if (opts.mediaUrl) {
-      const maxBytes = resolveMediaMaxBytes(opts.accountId);
-      const media = await getCore().media.loadWebMedia(opts.mediaUrl, maxBytes);
-      const uploaded = await uploadMediaMaybeEncrypted(client, roomId, media.buffer, {
-        contentType: media.contentType,
-        filename: media.fileName,
-      });
-      const durationMs = await resolveMediaDurationMs({
-        buffer: media.buffer,
-        contentType: media.contentType,
-        fileName: media.fileName,
-        kind: media.kind ?? "unknown",
-      });
-      const baseMsgType = resolveMatrixMsgType(media.contentType, media.fileName);
-      const { useVoice } = resolveMatrixVoiceDecision({
-        wantsVoice: opts.audioAsVoice === true,
-        contentType: media.contentType,
-        fileName: media.fileName,
-      });
-      const msgtype = useVoice ? MsgType.Audio : baseMsgType;
-      const isImage = msgtype === MsgType.Image;
-      const imageInfo = isImage
-        ? await prepareImageInfo({
-            buffer: media.buffer,
-            client,
-            encrypted: Boolean(uploaded.file),
-          })
-        : undefined;
-      const [firstChunk, ...rest] = chunks;
-      const body = useVoice ? "Voice message" : (firstChunk ?? media.fileName ?? "(file)");
-      const content = buildMediaContent({
-        msgtype,
-        body,
-        url: uploaded.url,
-        file: uploaded.file,
-        filename: media.fileName,
-        mimetype: media.contentType,
-        size: media.buffer.byteLength,
-        durationMs,
-        relation,
-        isVoice: useVoice,
-        imageInfo,
-      });
-      const eventId = await sendContent(content);
-      lastMessageId = eventId ?? lastMessageId;
-      const textChunks = useVoice ? chunks : rest;
-      const followupRelation = threadId ? relation : undefined;
-      for (const chunk of textChunks) {
-        const text = chunk.trim();
-        if (!text) {
-          continue;
-        }
-        const followup = buildTextContent(text, followupRelation);
-        const followupEventId = await sendContent(followup);
-        lastMessageId = followupEventId ?? lastMessageId;
-      }
-    } else {
-      for (const chunk of chunks.length ? chunks : [""]) {
-        const text = chunk.trim();
-        if (!text) {
-          continue;
-        }
-        const content = buildTextContent(text, relation);
+      let lastMessageId = "";
+      if (opts.mediaUrl) {
+        const maxBytes = resolveMediaMaxBytes(opts.accountId);
+        const media = await getCore().media.loadWebMedia(opts.mediaUrl, maxBytes);
+        const uploaded = await uploadMediaMaybeEncrypted(client, roomId, media.buffer, {
+          contentType: media.contentType,
+          filename: media.fileName,
+        });
+        const durationMs = await resolveMediaDurationMs({
+          buffer: media.buffer,
+          contentType: media.contentType,
+          fileName: media.fileName,
+          kind: media.kind ?? "unknown",
+        });
+        const baseMsgType = resolveMatrixMsgType(media.contentType, media.fileName);
+        const { useVoice } = resolveMatrixVoiceDecision({
+          wantsVoice: opts.audioAsVoice === true,
+          contentType: media.contentType,
+          fileName: media.fileName,
+        });
+        const msgtype = useVoice ? MsgType.Audio : baseMsgType;
+        const isImage = msgtype === MsgType.Image;
+        const imageInfo = isImage
+          ? await prepareImageInfo({
+              buffer: media.buffer,
+              client,
+              encrypted: Boolean(uploaded.file),
+            })
+          : undefined;
+        const [firstChunk, ...rest] = chunks;
+        const body = useVoice ? "Voice message" : (firstChunk ?? media.fileName ?? "(file)");
+        const content = buildMediaContent({
+          msgtype,
+          body,
+          url: uploaded.url,
+          file: uploaded.file,
+          filename: media.fileName,
+          mimetype: media.contentType,
+          size: media.buffer.byteLength,
+          durationMs,
+          relation,
+          isVoice: useVoice,
+          imageInfo,
+        });
         const eventId = await sendContent(content);
         lastMessageId = eventId ?? lastMessageId;
+        const textChunks = useVoice ? chunks : rest;
+        const followupRelation = threadId ? relation : undefined;
+        for (const chunk of textChunks) {
+          const text = chunk.trim();
+          if (!text) {
+            continue;
+          }
+          const followup = buildTextContent(text, followupRelation);
+          const followupEventId = await sendContent(followup);
+          lastMessageId = followupEventId ?? lastMessageId;
+        }
+      } else {
+        for (const chunk of chunks.length ? chunks : [""]) {
+          const text = chunk.trim();
+          if (!text) {
+            continue;
+          }
+          const content = buildTextContent(text, relation);
+          const eventId = await sendContent(content);
+          lastMessageId = eventId ?? lastMessageId;
+        }
       }
-    }
 
-    return {
-      messageId: lastMessageId || "unknown",
-      roomId,
-    };
-  } finally {
-    if (stopOnDone) {
-      client.stop();
-    }
-  }
+      return {
+        messageId: lastMessageId || "unknown",
+        roomId,
+      };
+    },
+  );
 }
 
 export async function sendPollMatrix(
@@ -188,30 +186,27 @@ export async function sendPollMatrix(
   if (!poll.options?.length) {
     throw new Error("Matrix poll requires options");
   }
-  const { client, stopOnDone } = await resolveMatrixClient({
-    client: opts.client,
-    timeoutMs: opts.timeoutMs,
-    accountId: opts.accountId,
-  });
+  return await withResolvedMatrixClient(
+    {
+      client: opts.client,
+      timeoutMs: opts.timeoutMs,
+      accountId: opts.accountId,
+    },
+    async (client) => {
+      const roomId = await resolveMatrixRoomId(client, to);
+      const pollContent = buildPollStartContent(poll);
+      const threadId = normalizeThreadId(opts.threadId);
+      const pollPayload = threadId
+        ? { ...pollContent, "m.relates_to": buildThreadRelation(threadId) }
+        : pollContent;
+      const eventId = await client.sendEvent(roomId, M_POLL_START, pollPayload);
 
-  try {
-    const roomId = await resolveMatrixRoomId(client, to);
-    const pollContent = buildPollStartContent(poll);
-    const threadId = normalizeThreadId(opts.threadId);
-    const pollPayload = threadId
-      ? { ...pollContent, "m.relates_to": buildThreadRelation(threadId) }
-      : pollContent;
-    const eventId = await client.sendEvent(roomId, M_POLL_START, pollPayload);
-
-    return {
-      eventId: eventId ?? "unknown",
-      roomId,
-    };
-  } finally {
-    if (stopOnDone) {
-      client.stop();
-    }
-  }
+      return {
+        eventId: eventId ?? "unknown",
+        roomId,
+      };
+    },
+  );
 }
 
 export async function sendTypingMatrix(
@@ -220,18 +215,17 @@ export async function sendTypingMatrix(
   timeoutMs?: number,
   client?: MatrixClient,
 ): Promise<void> {
-  const { client: resolved, stopOnDone } = await resolveMatrixClient({
-    client,
-    timeoutMs,
-  });
-  try {
-    const resolvedTimeoutMs = typeof timeoutMs === "number" ? timeoutMs : 30_000;
-    await resolved.setTyping(roomId, typing, resolvedTimeoutMs);
-  } finally {
-    if (stopOnDone) {
-      resolved.stop();
-    }
-  }
+  await withResolvedMatrixClient(
+    {
+      client,
+      timeoutMs,
+    },
+    async (resolved) => {
+      const resolvedRoom = await resolveMatrixRoomId(resolved, roomId);
+      const resolvedTimeoutMs = typeof timeoutMs === "number" ? timeoutMs : 30_000;
+      await resolved.setTyping(resolvedRoom, typing, resolvedTimeoutMs);
+    },
+  );
 }
 
 export async function sendReadReceiptMatrix(
@@ -242,17 +236,10 @@ export async function sendReadReceiptMatrix(
   if (!eventId?.trim()) {
     return;
   }
-  const { client: resolved, stopOnDone } = await resolveMatrixClient({
-    client,
-  });
-  try {
+  await withResolvedMatrixClient({ client }, async (resolved) => {
     const resolvedRoom = await resolveMatrixRoomId(resolved, roomId);
     await resolved.sendReadReceipt(resolvedRoom, eventId.trim());
-  } finally {
-    if (stopOnDone) {
-      resolved.stop();
-    }
-  }
+  });
 }
 
 export async function reactMatrixMessage(
@@ -262,18 +249,16 @@ export async function reactMatrixMessage(
   opts?: MatrixClient | MatrixClientResolveOpts,
 ): Promise<void> {
   const clientOpts = normalizeMatrixClientResolveOpts(opts);
-  const { client: resolved, stopOnDone } = await resolveMatrixClient({
-    client: clientOpts.client,
-    timeoutMs: clientOpts.timeoutMs,
-    accountId: clientOpts.accountId ?? undefined,
-  });
-  try {
-    const resolvedRoom = await resolveMatrixRoomId(resolved, roomId);
-    const reaction = buildMatrixReactionContent(messageId, emoji);
-    await resolved.sendEvent(resolvedRoom, EventType.Reaction, reaction);
-  } finally {
-    if (stopOnDone) {
-      resolved.stop();
-    }
-  }
+  await withResolvedMatrixClient(
+    {
+      client: clientOpts.client,
+      timeoutMs: clientOpts.timeoutMs,
+      accountId: clientOpts.accountId ?? undefined,
+    },
+    async (resolved) => {
+      const resolvedRoom = await resolveMatrixRoomId(resolved, roomId);
+      const reaction = buildMatrixReactionContent(messageId, emoji);
+      await resolved.sendEvent(resolvedRoom, EventType.Reaction, reaction);
+    },
+  );
 }

extensions/matrix/src/matrix/send/client.ts

@@ -1,21 +1,17 @@
 import { getMatrixRuntime } from "../../runtime.js";
 import type { CoreConfig } from "../../types.js";
 import { resolveMatrixAccountConfig } from "../accounts.js";
-import { getActiveMatrixClient } from "../active-client.js";
 import {
-  createMatrixClient,
-  isBunRuntime,
-  resolveMatrixAuth,
-  resolveMatrixAuthContext,
-} from "../client.js";
+  ensureMatrixNodeRuntime,
+  resolveRuntimeMatrixClient,
+  type ResolvedRuntimeMatrixClient,
+} from "../client-bootstrap.js";
 import type { MatrixClient } from "../sdk.js";
 
 const getCore = () => getMatrixRuntime();
 
 export function ensureNodeRuntime() {
-  if (isBunRuntime()) {
-    throw new Error("Matrix support requires Node (bun runtime not supported)");
-  }
+  ensureMatrixNodeRuntime();
 }
 
 export function resolveMediaMaxBytes(accountId?: string | null): number | undefined {
@@ -33,34 +29,36 @@ export async function resolveMatrixClient(opts: {
   timeoutMs?: number;
   accountId?: string | null;
 }): Promise<{ client: MatrixClient; stopOnDone: boolean }> {
-  ensureNodeRuntime();
-  if (opts.client) {
-    return { client: opts.client, stopOnDone: false };
-  }
-  const cfg = getCore().config.loadConfig() as CoreConfig;
-  const authContext = resolveMatrixAuthContext({
-    cfg,
+  return await resolveRuntimeMatrixClient({
+    client: opts.client,
+    timeoutMs: opts.timeoutMs,
     accountId: opts.accountId,
+    onResolved: async (client, context) => {
+      if (context.createdForOneOff) {
+        await client.prepareForOneOff();
+      }
+    },
   });
-  const active = getActiveMatrixClient(authContext.accountId);
-  if (active) {
-    return { client: active, stopOnDone: false };
-  }
-  const auth = await resolveMatrixAuth({
-    cfg,
-    accountId: authContext.accountId,
-  });
-  const client = await createMatrixClient({
-    homeserver: auth.homeserver,
-    userId: auth.userId,
-    accessToken: auth.accessToken,
-    password: auth.password,
-    deviceId: auth.deviceId,
-    encryption: auth.encryption,
-    localTimeoutMs: opts.timeoutMs,
-    accountId: auth.accountId,
-    autoBootstrapCrypto: false,
-  });
-  await client.prepareForOneOff();
-  return { client, stopOnDone: true };
+}
+
+export function stopResolvedMatrixClient(resolved: ResolvedRuntimeMatrixClient): void {
+  if (resolved.stopOnDone) {
+    resolved.client.stop();
+  }
+}
+
+export async function withResolvedMatrixClient<T>(
+  opts: {
+    client?: MatrixClient;
+    timeoutMs?: number;
+    accountId?: string | null;
+  },
+  run: (client: MatrixClient) => Promise<T>,
+): Promise<T> {
+  const resolved = await resolveMatrixClient(opts);
+  try {
+    return await run(resolved.client);
+  } finally {
+    stopResolvedMatrixClient(resolved);
+  }
 }