Refactor how we listen for matrix events. (#446)

* Refactor Matrix event listener in Mjolnir and ManagedMjolnir.

closes https://github.com/matrix-org/mjolnir/issues/411.

Issue #411 says that we have to be careful about room.join,
but this was before we figured how to make matrix-appservice-bridge
echo events sent by its own intents.

* Remove MatrixClientListener since it isn't actually needed.

* Protect which config values can be used for ManagedMjolnirs.

* Introduce MatrixSendClient

so listeners aren't accidentally added to a MatrixClient instead
of MatrixEmitter.

* doc

* Move provisioned mjolnir config to src/config.

This just aids maintance so whenever someone goes to change the config
of the bot they will see this and update it.

* doc for matrix intent listener.

src/ManagementRoomOutput.ts

@@ -15,8 +15,9 @@ limitations under the License.
 */
 
 import * as Sentry from "@sentry/node";
-import { extractRequestError, LogLevel, LogService, MatrixClient, MessageType, Permalinks, TextualMessageEventContent, UserID } from "matrix-bot-sdk";
+import { extractRequestError, LogLevel, LogService, MessageType, Permalinks, TextualMessageEventContent, UserID } from "matrix-bot-sdk";
 import { IConfig } from "./config";
+import { MatrixSendClient } from "./MatrixEmitter";
 import { htmlEscape } from "./utils";
 
 const levelToFn = {
@@ -33,7 +34,7 @@ export default class ManagementRoomOutput {
 
     constructor(
         private readonly managementRoomId: string,
-        private readonly client: MatrixClient,
+        private readonly client: MatrixSendClient,
         private readonly config: IConfig,
     ) {
 

src/MatrixEmitter.ts

@@ -0,0 +1,55 @@
+/*
+Copyright 2019-2022 The Matrix.org Foundation C.I.C.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import EventEmitter from "events";
+import { MatrixClient } from "matrix-bot-sdk";
+
+/**
+ * This is an interface created in order to keep the event listener
+ * Mjolnir uses for new events generic.
+ * Used to provide a unified API for messages received from matrix-bot-sdk (using GET /sync)
+ * when we're in single bot mode and messages received from matrix-appservice-bridge (using pushed /transaction)
+ * when we're in appservice mode.
+ */
+export declare interface MatrixEmitter extends EventEmitter {
+    on(event: 'room.event', listener: (roomId: string, mxEvent: any) => void ): this
+    emit(event: 'room.event', roomId: string, mxEvent: any): boolean
+
+    on(event: 'room.message', listener: (roomId: string, mxEvent: any) => void ): this
+    emit(event: 'room.message', roomId: string, mxEvent: any): boolean
+
+    on(event: 'room.invite', listener: (roomId: string, mxEvent: any) => void ): this
+    emit(event: 'room.invite', roomId: string, mxEvent: any): boolean
+
+    on(event: 'room.join', listener: (roomId: string, mxEvent: any) => void ): this
+    emit(event: 'room.join', roomId: string, mxEvent: any): boolean
+
+    on(event: 'room.leave', listener: (roomId: string, mxEvent: any) => void ): this
+    emit(event: 'room.leave', roomId: string, mxEvent: any): boolean
+
+    on(event: 'room.archived', listener: (roomId: string, mxEvent: any) => void ): this
+    emit(event: 'room.archived', roomId: string, mxEvent: any): boolean
+
+    start(): Promise<void>;
+    stop(): void;
+}
+
+/**
+ * A `MatrixClient` without the properties of `MatrixEmitter`.
+ * This is in order to enforce listeners are added to `MatrixEmitter`s
+ * rather than on the matrix-bot-sdk version of the matrix client.
+ */
+export type MatrixSendClient = Omit<MatrixClient, keyof MatrixEmitter>;

src/Mjolnir.ts

@@ -18,7 +18,6 @@ import {
     extractRequestError,
     LogLevel,
     LogService,
-    MatrixClient,
     MembershipEvent,
     Permalinks,
 } from "matrix-bot-sdk";
@@ -39,6 +38,7 @@ import ManagementRoomOutput from "./ManagementRoomOutput";
 import { ProtectionManager } from "./protections/ProtectionManager";
 import { RoomMemberManager } from "./RoomMembers";
 import ProtectedRoomsConfig from "./ProtectedRoomsConfig";
+import { MatrixEmitter, MatrixSendClient } from "./MatrixEmitter";
 
 export const STATE_NOT_STARTED = "not_started";
 export const STATE_CHECKING_PERMISSIONS = "checking_permissions";
@@ -88,15 +88,15 @@ export class Mjolnir {
 
     /**
      * Adds a listener to the client that will automatically accept invitations.
-     * @param {MatrixClient} client
+     * @param {MatrixSendClient} client
      * @param options By default accepts invites from anyone.
      * @param {string} options.managementRoom The room to report ignored invitations to if `recordIgnoredInvites` is true.
      * @param {boolean} options.recordIgnoredInvites Whether to report invites that will be ignored to the `managementRoom`.
      * @param {boolean} options.autojoinOnlyIfManager Whether to only accept an invitation by a user present in the `managementRoom`.
      * @param {string} options.acceptInvitesFromSpace A space of users to accept invites from, ignores invites form users not in this space.
      */
-    private static addJoinOnInviteListener(mjolnir: Mjolnir, client: MatrixClient, options: { [key: string]: any }) {
+    private static addJoinOnInviteListener(mjolnir: Mjolnir, client: MatrixSendClient, options: { [key: string]: any }) {
-        client.on("room.invite", async (roomId: string, inviteEvent: any) => {
+        mjolnir.matrixEmitter.on("room.invite", async (roomId: string, inviteEvent: any) => {
             const membershipEvent = new MembershipEvent(inviteEvent);
 
             const reportInvite = async () => {
@@ -130,17 +130,16 @@ export class Mjolnir {
                     });
                 if (!spaceUserIds.includes(membershipEvent.sender)) return reportInvite(); // ignore invite
             }
-
             return client.joinRoom(roomId);
         });
     }
 
     /**
      * Create a new Mjolnir instance from a client and the options in the configuration file, ready to be started.
-     * @param {MatrixClient} client The client for Mjolnir to use.
+     * @param {MatrixSendClient} client The client for Mjolnir to use.
      * @returns A new Mjolnir instance that can be started without further setup.
      */
-    static async setupMjolnirFromConfig(client: MatrixClient, config: IConfig): Promise<Mjolnir> {
+    static async setupMjolnirFromConfig(client: MatrixSendClient, matrixEmitter: MatrixEmitter, config: IConfig): Promise<Mjolnir> {
         const policyLists: PolicyList[] = [];
         const joinedRooms = await client.getJoinedRooms();
 
@@ -152,15 +151,16 @@ export class Mjolnir {
         }
 
         const ruleServer = config.web.ruleServer ? new RuleServer() : null;
-        const mjolnir = new Mjolnir(client, await client.getUserId(), managementRoomId, config, policyLists, ruleServer);
+        const mjolnir = new Mjolnir(client, await client.getUserId(), matrixEmitter, managementRoomId, config, policyLists, ruleServer);
         await mjolnir.managementRoomOutput.logMessage(LogLevel.INFO, "index", "Mjolnir is starting up. Use !mjolnir to query status.");
         Mjolnir.addJoinOnInviteListener(mjolnir, client, config);
         return mjolnir;
     }
 
     constructor(
-        public readonly client: MatrixClient,
+        public readonly client: MatrixSendClient,
         private readonly clientUserId: string,
+        public readonly matrixEmitter: MatrixEmitter,
         public readonly managementRoomId: string,
         public readonly config: IConfig,
         private policyLists: PolicyList[],
@@ -171,9 +171,9 @@ export class Mjolnir {
 
         // Setup bot.
 
-        client.on("room.event", this.handleEvent.bind(this));
+        matrixEmitter.on("room.event", this.handleEvent.bind(this));
 
-        client.on("room.message", async (roomId, event) => {
+        matrixEmitter.on("room.message", async (roomId, event) => {
             if (roomId !== this.managementRoomId) return;
             if (!event['content']) return;
 
@@ -208,11 +208,11 @@ export class Mjolnir {
             }
         });
 
-        client.on("room.join", (roomId: string, event: any) => {
+        matrixEmitter.on("room.join", (roomId: string, event: any) => {
             LogService.info("Mjolnir", `Joined ${roomId}`);
             return this.resyncJoinedRooms();
         });
-        client.on("room.leave", (roomId: string, event: any) => {
+        matrixEmitter.on("room.leave", (roomId: string, event: any) => {
             LogService.info("Mjolnir", `Left ${roomId}`);
             return this.resyncJoinedRooms();
         });
@@ -234,7 +234,7 @@ export class Mjolnir {
             this.reportPoller = new ReportPoller(this, this.reportManager);
         }
         // Setup join/leave listener
-        this.roomJoins = new RoomMemberManager(this.client);
+        this.roomJoins = new RoomMemberManager(this.matrixEmitter);
         this.taskQueue = new ThrottlingQueue(this, config.backgroundDelayMS);
 
         this.protectionManager = new ProtectionManager(this);
@@ -303,7 +303,7 @@ export class Mjolnir {
             }
 
             // Start the bot.
-            await this.client.start();
+            await this.matrixEmitter.start();
 
             this.currentState = STATE_SYNCING;
             if (this.config.syncOnStartup) {
@@ -331,7 +331,7 @@ export class Mjolnir {
      */
     public stop() {
         LogService.info("Mjolnir", "Stopping Mjolnir...");
-        this.client.stop();
+        this.matrixEmitter.stop();
         this.webapis.stop();
         this.reportPoller?.stop();
     }

src/ProtectedRoomsConfig.ts

@@ -15,8 +15,9 @@ limitations under the License.
 */
 
 import AwaitLock from 'await-lock';
-import { extractRequestError, LogService, MatrixClient, Permalinks } from "matrix-bot-sdk";
+import { extractRequestError, LogService, Permalinks } from "matrix-bot-sdk";
 import { IConfig } from "./config";
+import { MatrixSendClient } from './MatrixEmitter';
 const PROTECTED_ROOMS_EVENT_TYPE = "org.matrix.mjolnir.protected_rooms";
 
 /**
@@ -32,7 +33,7 @@ export default class ProtectedRoomsConfig {
     /** This is to prevent clobbering the account data for the protected rooms if several rooms are explicitly protected concurrently. */
     private accountDataLock = new AwaitLock();
 
-    constructor(private readonly client: MatrixClient) {
+    constructor(private readonly client: MatrixSendClient) {
 
     }
 

src/ProtectedRoomsSet.ts

@@ -14,10 +14,11 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import { LogLevel, LogService, MatrixClient, MatrixGlob, Permalinks, UserID } from "matrix-bot-sdk";
+import { LogLevel, LogService, MatrixGlob, Permalinks, UserID } from "matrix-bot-sdk";
 import { IConfig } from "./config";
 import ErrorCache, { ERROR_KIND_FATAL, ERROR_KIND_PERMISSION } from "./ErrorCache";
 import ManagementRoomOutput from "./ManagementRoomOutput";
+import { MatrixSendClient } from "./MatrixEmitter";
 import AccessControlUnit, { Access } from "./models/AccessControlUnit";
 import { RULE_ROOM, RULE_SERVER, RULE_USER } from "./models/ListRule";
 import PolicyList, { ListRuleChange } from "./models/PolicyList";
@@ -88,7 +89,7 @@ export class ProtectedRoomsSet {
     private readonly accessControlUnit = new AccessControlUnit([]);
 
     constructor(
-        private readonly client: MatrixClient,
+        private readonly client: MatrixSendClient,
         private readonly clientUserId: string,
         private readonly managementRoomId: string,
         private readonly managementRoomOutput: ManagementRoomOutput,

src/RoomMembers.ts

@@ -1,4 +1,4 @@
-import { MatrixClient } from "matrix-bot-sdk";
+import { MatrixEmitter } from "./MatrixEmitter";
 
 enum Action {
     Join,
@@ -154,7 +154,7 @@ class RoomMembers {
 export class RoomMemberManager {
     private perRoom: Map<string /* room id */, RoomMembers> = new Map();
     private readonly cbHandleEvent;
-    constructor(private client: MatrixClient) {
+    constructor(private client: MatrixEmitter) {
         // Listen for join events.
         this.cbHandleEvent = this.handleEvent.bind(this);
         client.on("room.event", this.cbHandleEvent);

src/appservice/MjolnirManager.ts

@@ -1,12 +1,16 @@
 import { Mjolnir } from "../Mjolnir";
-import { Request, WeakEvent, BridgeContext, Bridge, Intent } from "matrix-appservice-bridge";
+import { Request, WeakEvent, BridgeContext, Bridge, Intent, Logger } from "matrix-appservice-bridge";
-import { IConfig, read as configRead } from "../config";
+import { getProvisionedMjolnirConfig } from "../config";
 import PolicyList from "../models/PolicyList";
 import { Permalinks, MatrixClient } from "matrix-bot-sdk";
 import { DataStore } from "./datastore";
 import { AccessControl } from "./AccessControl";
 import { Access } from "../models/AccessControlUnit";
 import { randomUUID } from "crypto";
+import EventEmitter from "events";
+import { MatrixEmitter } from "../MatrixEmitter";
+
+const log = new Logger('MjolnirManager');
 
 /**
  * The MjolnirManager is responsible for:
@@ -38,18 +42,6 @@ export class MjolnirManager {
         return mjolnirManager;
     }
 
-    /**
-     * Gets the default config to give the newly provisioned mjolnirs.
-     * @param managementRoomId A room that has been created to serve as the mjolnir's management room for the owner.
-     * @returns A config that can be directly used by the new mjolnir.
-     */
-    private getDefaultMjolnirConfig(managementRoomId: string): IConfig {
-        let config = configRead();
-        config.managementRoom = managementRoomId;
-        config.protectedRooms = [];
-        return config;
-    }
-
     /**
      * Creates a new mjolnir for a user.
      * @param requestingUserId The user that is requesting this mjolnir and who will own it.
@@ -58,10 +50,17 @@ export class MjolnirManager {
      * @returns A new managed mjolnir.
      */
     public async makeInstance(requestingUserId: string, managementRoomId: string, client: MatrixClient): Promise<ManagedMjolnir> {
+        const intentListener = new MatrixIntentListener(await client.getUserId());
         const managedMjolnir = new ManagedMjolnir(
             requestingUserId,
-            await Mjolnir.setupMjolnirFromConfig(client, this.getDefaultMjolnirConfig(managementRoomId))
+            await Mjolnir.setupMjolnirFromConfig(
+                client,
+                intentListener,
+                getProvisionedMjolnirConfig(managementRoomId)
+            ),
+            intentListener,
         );
+        await managedMjolnir.start();
         this.mjolnirs.set(await client.getUserId(), managedMjolnir);
         return managedMjolnir;
     }
@@ -170,7 +169,11 @@ export class MjolnirManager {
                     mjolnirRecord.owner,
                     mjolnirRecord.management_room,
                     mjIntent.matrixClient,
-                );
+                ).catch((e: any) => {
+                    log.error(`Could not start mjolnir ${mjolnirRecord.local_part} for ${mjolnirRecord.owner}:`, e);
+                    // Don't await, we don't want to clobber initialization if this fails.
+                    mjIntent.matrixClient.sendNotice(mjolnirRecord.management_room, `Your mjolnir could not be started. Please alert the administrator`);
+                });
             }
         }
     }
@@ -180,25 +183,11 @@ export class ManagedMjolnir {
     public constructor(
         public readonly ownerId: string,
         private readonly mjolnir: Mjolnir,
+        private readonly matrixEmitter: MatrixIntentListener,
     ) { }
 
     public async onEvent(request: Request<WeakEvent>) {
-        // Emulate the client syncing.
+        this.matrixEmitter.handleEvent(request.getData());
-        // https://github.com/matrix-org/mjolnir/issues/411
-        const mxEvent = request.getData();
-        if (mxEvent['type'] !== undefined) {
-            this.mjolnir.client.emit('room.event', mxEvent.room_id, mxEvent);
-            if (mxEvent.type === 'm.room.message') {
-                this.mjolnir.client.emit('room.message', mxEvent.room_id, mxEvent);
-            }
-            // TODO: We need to figure out how to inform the mjolnir of `room.join`.
-            // https://github.com/matrix-org/mjolnir/issues/411
-        }
-        if (mxEvent['type'] === 'm.room.member') {
-            if (mxEvent['content']['membership'] === 'invite' && mxEvent.state_key === await this.mjolnir.client.getUserId()) {
-                this.mjolnir.client.emit('room.invite', mxEvent.room_id, mxEvent);
-            }
-        }
     }
 
     public async joinRoom(roomId: string) {
@@ -223,4 +212,63 @@ export class ManagedMjolnir {
     public get managementRoomId(): string {
         return this.mjolnir.managementRoomId;
     }
+
+    /**
+     * Intended to be called by the MjolnirManager to make sure the mjolnir is ready to listen to events.
+     * This managed mjolnir should not be informed of any events via `onEvent` until `start` is called.
+     */
+    public async start(): Promise<void> {
+        await this.mjolnir.start();
+    }
+}
+
+/**
+ * This is used to listen for events intended for a single mjolnir that resides in the appservice.
+ * This exists entirely because the Mjolnir class was previously designed only to receive events
+ * from a syncing matrix-bot-sdk MatrixClient. Since appservices provide a transactional push
+ * api for all users on the appservice, almost the opposite of sync, we needed to create an
+ * interface for both. See `MatrixEmitter`.
+ */
+export class MatrixIntentListener extends EventEmitter implements MatrixEmitter {
+    constructor(private readonly mjolnirId: string) {
+        super()
+    }
+
+    public handleEvent(mxEvent: WeakEvent) {
+        // These are ordered to be the same as matrix-bot-sdk's MatrixClient
+        // They shouldn't need to be, but they are just in case it matters.
+        if (mxEvent['type'] === 'm.room.member' &&  mxEvent.state_key === this.mjolnirId) {
+            if (mxEvent['content']['membership'] === 'leave') {
+                this.emit('room.leave', mxEvent.room_id, mxEvent);
+            }
+            if (mxEvent['content']['membership'] === 'invite') {
+                this.emit('room.invite', mxEvent.room_id, mxEvent);
+            }
+            if (mxEvent['content']['membership'] === 'join') {
+                this.emit('room.join', mxEvent.room_id, mxEvent);
+            }
+        }
+        if (mxEvent.type === 'm.room.message') {
+            this.emit('room.message', mxEvent.room_id, mxEvent);
+        }
+        if (mxEvent.type === 'm.room.tombstone' && mxEvent.state_key === '') {
+            this.emit('room.archived', mxEvent.room_id, mxEvent);
+        }
+        this.emit('room.event', mxEvent.room_id, mxEvent);
+
+    }
+
+    /**
+     * To be called by `Mjolnir`.
+     */
+    public async start() {
+        // Nothing to do.
+    }
+
+    /**
+     * To be called by `Mjolnir`.
+     */
+    public stop() {
+        // Nothing to do.
+    }
 }

src/config.ts

@@ -16,7 +16,7 @@ limitations under the License.
 
 import * as fs from "fs";
 import { load } from "js-yaml";
-import { MatrixClient } from "matrix-bot-sdk";
+import { MatrixClient, LogService } from "matrix-bot-sdk";
 import Config from "config";
 
 /**
@@ -180,6 +180,10 @@ const defaultConfig: IConfig = {
     },
 };
 
+export function getDefaultConfig(): IConfig {
+    return Config.util.cloneDeep(defaultConfig);
+}
+
 /**
  * Grabs an explicit path provided for mjolnir's config from an arguments vector if provided, otherwise returns undefined.
  * @param argv An arguments vector sourced from `process.argv`.
@@ -209,3 +213,40 @@ export function read(): IConfig {
         return config;
     }
 }
+
+/**
+ * Provides a config for each newly provisioned mjolnir in appservice mode.
+ * @param managementRoomId A room that has been created to serve as the mjolnir's management room for the owner.
+ * @returns A config that can be directly used by the new mjolnir.
+ */
+export function getProvisionedMjolnirConfig(managementRoomId: string): IConfig {
+    // These are keys that are allowed to be configured for provisioned mjolnirs.
+    // We need a restricted set so that someone doesn't accidentally enable webservers etc
+    // on every created Mjolnir, which would result in very confusing error messages.
+    const allowedKeys = [
+        "commands",
+        "verboseLogging",
+        "logLevel",
+        "syncOnStartup",
+        "verifyPermissionsOnStartup",
+        "fasterMembershipChecks",
+        "automaticallyRedactForReasons",
+        "protectAllJoinedRooms",
+        "backgroundDelayMS",
+    ];
+    const configTemplate = read(); // we use the standard bot config as a template for every provisioned mjolnir.
+    const unusedKeys = Object.keys(configTemplate).filter(key => !allowedKeys.includes(key));
+    if (unusedKeys.length > 0) {
+        LogService.warn("config", "The config provided for provisioned mjolnirs contains keys which are not used by the appservice.", unusedKeys);
+    }
+    const config = Config.util.extendDeep(
+        getDefaultConfig(),
+        allowedKeys.reduce((existingConfig: any, key: string) => {
+            return { ...existingConfig, [key]: configTemplate[key as keyof IConfig] }
+        }, {})
+    );
+
+    config.managementRoom = managementRoomId;
+    config.protectedRooms = [];
+    return config;
+}

src/index.ts

@@ -67,7 +67,7 @@ import { initializeSentry, patchMatrixClient } from "./utils";
         patchMatrixClient();
         config.RUNTIME.client = client;
 
-        bot = await Mjolnir.setupMjolnirFromConfig(client, config);
+        bot = await Mjolnir.setupMjolnirFromConfig(client, client, config);
     } catch (err) {
         console.error(`Failed to setup mjolnir from the config ${config.dataPath}: ${err}`);
         throw err;

src/models/PolicyList.ts

@@ -14,9 +14,10 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import { extractRequestError, LogService, MatrixClient, RoomCreateOptions, UserID } from "matrix-bot-sdk";
+import { extractRequestError, LogService, RoomCreateOptions, UserID } from "matrix-bot-sdk";
 import { EventEmitter } from "events";
 import { ALL_RULE_TYPES, EntityType, ListRule, Recommendation, ROOM_RULE_TYPES, RULE_ROOM, RULE_SERVER, RULE_USER, SERVER_RULE_TYPES, USER_RULE_TYPES } from "./ListRule";
+import { MatrixSendClient } from "../MatrixEmitter";
 
 export const SHORTCODE_EVENT_TYPE = "org.matrix.mjolnir.shortcode";
 
@@ -104,7 +105,7 @@ class PolicyList extends EventEmitter {
      * @param roomRef A sharable/clickable matrix URL that refers to the room.
      * @param client A matrix client that is used to read the state of the room when `updateList` is called.
      */
-    constructor(public readonly roomId: string, public readonly roomRef: string, private client: MatrixClient) {
+    constructor(public readonly roomId: string, public readonly roomRef: string, private client: MatrixSendClient) {
         super();
         this.batcher = new UpdateBatcher(this);
     }
@@ -118,7 +119,7 @@ class PolicyList extends EventEmitter {
      * @returns The room id for the newly created policy list.
      */
     public static async createList(
-        client: MatrixClient,
+        client: MatrixSendClient,
         shortcode: string,
         invite: string[],
         createRoomOptions: RoomCreateOptions = {}

src/protections/ProtectionManager.ts

@@ -64,7 +64,7 @@ export class ProtectionManager {
      */
     public async start() {
         this.mjolnir.reportManager.on("report.new", this.handleReport.bind(this));
-        this.mjolnir.client.on("room.event", this.handleEvent.bind(this));
+        this.mjolnir.matrixEmitter.on("room.event", this.handleEvent.bind(this));
         for (const protection of PROTECTIONS) {
             try {
                 await this.registerProtection(protection);

src/queues/EventRedactionQueue.ts

@@ -18,6 +18,7 @@ import { ERROR_KIND_FATAL } from "../ErrorCache";
 import { RoomUpdateError } from "../models/RoomUpdateError";
 import { redactUserMessagesIn } from "../utils";
 import ManagementRoomOutput from "../ManagementRoomOutput";
+import { MatrixSendClient } from "../MatrixEmitter";
 
 export interface QueuedRedaction {
     /** The room which the redaction will take place in. */
@@ -27,7 +28,7 @@ export interface QueuedRedaction {
      * Called by the EventRedactionQueue.
      * @param client A MatrixClient to use to carry out the redaction.
      */
-    redact(client: MatrixClient, managementRoom: ManagementRoomOutput): Promise<void>
+    redact(client: MatrixSendClient, managementRoom: ManagementRoomOutput): Promise<void>
     /**
      * Used to test whether the redaction is the equivalent to another redaction.
      * @param redaction Another QueuedRedaction to test if this redaction is an equivalent to.
@@ -107,7 +108,7 @@ export class EventRedactionQueue {
      * @param limitToRoomId If the roomId is provided, only redactions for that room will be processed.
      * @returns A description of any errors encountered by each QueuedRedaction that was processed.
      */
-    public async process(client: MatrixClient, managementRoom: ManagementRoomOutput, limitToRoomId?: string): Promise<RoomUpdateError[]> {
+    public async process(client: MatrixSendClient, managementRoom: ManagementRoomOutput, limitToRoomId?: string): Promise<RoomUpdateError[]> {
         const errors: RoomUpdateError[] = [];
         const redact = async (currentBatch: QueuedRedaction[]) => {
             for (const redaction of currentBatch) {

src/report/ReportManager.ts

@@ -79,7 +79,7 @@ export class ReportManager extends EventEmitter {
     constructor(public mjolnir: Mjolnir) {
         super();
         // Configure bot interactions.
-        mjolnir.client.on("room.event", async (roomId, event) => {
+        mjolnir.matrixEmitter.on("room.event", async (roomId, event) => {
             try {
                 switch (event["type"]) {
                     case "m.reaction": {

src/utils.ts

@@ -17,7 +17,6 @@ limitations under the License.
 import {
     LogLevel,
     LogService,
-    MatrixClient,
     MatrixGlob,
     getRequestFn,
     setRequestFn,
@@ -29,6 +28,7 @@ import * as _ from '@sentry/tracing'; // Performing the import activates tracing
 
 import ManagementRoomOutput from "./ManagementRoomOutput";
 import { IConfig } from "./config";
+import { MatrixSendClient } from "./MatrixEmitter";
 
 // Define a few aliases to simplify parsing durations.
 
@@ -81,7 +81,7 @@ export function isTrueJoinEvent(event: any): boolean {
  * @param limit The number of messages to redact from most recent first. If the limit is reached then no further messages will be redacted.
  * @param noop Whether to operate in noop mode.
  */
-export async function redactUserMessagesIn(client: MatrixClient, managementRoom: ManagementRoomOutput, userIdOrGlob: string, targetRoomIds: string[], limit = 1000, noop = false) {
+export async function redactUserMessagesIn(client: MatrixSendClient, managementRoom: ManagementRoomOutput, userIdOrGlob: string, targetRoomIds: string[], limit = 1000, noop = false) {
     for (const targetRoomId of targetRoomIds) {
         await managementRoom.logMessage(LogLevel.DEBUG, "utils#redactUserMessagesIn", `Fetching sent messages for ${userIdOrGlob} in ${targetRoomId} to redact...`, targetRoomId);
 
@@ -101,7 +101,7 @@ export async function redactUserMessagesIn(client: MatrixClient, managementRoom:
 /**
  * Gets all the events sent by a user (or users if using wildcards) in a given room ID, since
  * the time they joined.
- * @param {MatrixClient} client The client to use.
+ * @param {MatrixSendClient} client The client to use.
  * @param {string} sender The sender. A matrix user id or a wildcard to match multiple senders e.g. *.example.com.
  * Can also be used to generically search the sender field e.g. *bob* for all events from senders with "bob" in them.
  * See `MatrixGlob` in matrix-bot-sdk.
@@ -114,7 +114,7 @@ export async function redactUserMessagesIn(client: MatrixClient, managementRoom:
  * The callback will only be called if there are any relevant events.
  * @returns {Promise<void>} Resolves when either: the limit has been reached, no relevant events could be found or there is no more timeline to paginate.
  */
-export async function getMessagesByUserIn(client: MatrixClient, sender: string, roomId: string, limit: number, cb: (events: any[]) => void): Promise<void> {
+export async function getMessagesByUserIn(client: MatrixSendClient, sender: string, roomId: string, limit: number, cb: (events: any[]) => void): Promise<void> {
     const isGlob = sender.includes("*");
     const roomEventFilter = {
         rooms: [roomId],

test/integration/banListTest.ts

@@ -9,6 +9,7 @@ import { Mjolnir } from "../../src/Mjolnir";
 import { ALL_RULE_TYPES, Recommendation, RULE_SERVER, RULE_USER, SERVER_RULE_TYPES } from "../../src/models/ListRule";
 import AccessControlUnit, { Access, EntityAccess } from "../../src/models/AccessControlUnit";
 import { randomUUID } from "crypto";
+import { MatrixSendClient } from "../../src/MatrixEmitter";
 
 /**
  * Create a policy rule in a policy room.
@@ -20,7 +21,7 @@ import { randomUUID } from "crypto";
  * @param template The template to use for the policy rule event.
  * @returns The event id of the newly created policy rule.
  */
-async function createPolicyRule(client: MatrixClient, policyRoomId: string, policyType: string, entity: string, reason: string, template = { recommendation: 'm.ban' }, stateKey = `rule:${entity}`) {
+async function createPolicyRule(client: MatrixSendClient, policyRoomId: string, policyType: string, entity: string, reason: string, template = { recommendation: 'm.ban' }, stateKey = `rule:${entity}`) {
     return await client.sendStateEvent(policyRoomId, policyType, stateKey, {
         entity,
         reason,
@@ -37,7 +38,7 @@ async function createPolicyRule(client: MatrixClient, policyRoomId: string, poli
  * @param stateKey The key for the rule.
  * @returns The event id of the void rule that was created to override the old one.
  */
-async function removePolicyRule(client: MatrixClient, policyRoomId: string, policyType: string, entity: string, stateKey = `rule:${entity}`) {
+async function removePolicyRule(client: MatrixSendClient, policyRoomId: string, policyType: string, entity: string, stateKey = `rule:${entity}`) {
     return await client.sendStateEvent(policyRoomId, policyType, stateKey, {});
 }
 

test/integration/commands/commandUtils.ts

@@ -1,29 +1,30 @@
 import { MatrixClient } from "matrix-bot-sdk";
 import { strict as assert } from "assert";
 import * as crypto from "crypto";
+import { MatrixEmitter } from "../../../src/MatrixEmitter";
 
 /**
  * Returns a promise that resolves to the first event replying to the event produced by targetEventThunk.
- * @param client A MatrixClient that is already in the targetRoom. We will use it to listen for the event produced by targetEventThunk.
+ * @param matrix A MatrixEmitter from a MatrixClient that is already in the targetRoom. We will use it to listen for the event produced by targetEventThunk.
  * This function assumes that the start() has already been called on the client.
  * @param targetRoom The room to listen for the reply in.
  * @param targetEventThunk A function that produces an event ID when called. This event ID is then used to listen for a reply.
  * @returns The replying event.
  */
-export async function getFirstReply(client: MatrixClient, targetRoom: string, targetEventThunk: () => Promise<string>): Promise<any> {
+export async function getFirstReply(matrix: MatrixEmitter, targetRoom: string, targetEventThunk: () => Promise<string>): Promise<any> {
-    return getNthReply(client, targetRoom, 1, targetEventThunk);
+    return getNthReply(matrix, targetRoom, 1, targetEventThunk);
 }
 
 /**
  * Returns a promise that resolves to the nth event replying to the event produced by targetEventThunk.
- * @param client A MatrixClient that is already in the targetRoom. We will use it to listen for the event produced by targetEventThunk.
+ * @param matrix A MatrixEmitter from a MatrixClient that is already in the targetRoom. We will use it to listen for the event produced by targetEventThunk.
  * This function assumes that the start() has already been called on the client.
  * @param targetRoom The room to listen for the reply in.
  * @param n The number of events to wait for. Must be >= 1.
  * @param targetEventThunk A function that produces an event ID when called. This event ID is then used to listen for a reply.
  * @returns The replying event.
  */
-export async function getNthReply(client: MatrixClient, targetRoom: string, n: number, targetEventThunk: () => Promise<string>): Promise<any> {
+export async function getNthReply(matrix: MatrixEmitter, targetRoom: string, n: number, targetEventThunk: () => Promise<string>): Promise<any> {
     if (Number.isNaN(n) || !Number.isInteger(n) || n <= 0) {
         throw new TypeError(`Invalid number of events ${n}`);
     }
@@ -35,7 +36,7 @@ export async function getNthReply(client: MatrixClient, targetRoom: string, n: n
     };
     let targetCb;
     try {
-        client.on('room.event', addEvent)
+        matrix.on('room.event', addEvent)
         const targetEventId = await targetEventThunk();
         if (typeof targetEventId !== 'string') {
             throw new TypeError();
@@ -61,12 +62,12 @@ export async function getNthReply(client: MatrixClient, targetRoom: string, n: n
                     }
                 }
             }
-            client.on('room.event', targetCb);
+            matrix.on('room.event', targetCb);
         });
     } finally {
-        client.removeListener('room.event', addEvent);
+        matrix.removeListener('room.event', addEvent);
         if (targetCb) {
-            client.removeListener('room.event', targetCb);
+            matrix.removeListener('room.event', targetCb);
         }
     }
 }
@@ -74,14 +75,14 @@ export async function getNthReply(client: MatrixClient, targetRoom: string, n: n
 
 /**
  * Returns a promise that resolves to an event that is reacting to the event produced by targetEventThunk.
- * @param client A MatrixClient that is already in the targetRoom that can be started to listen for the event produced by targetEventThunk.
+ * @param matrix A MatrixEmitter for a MatrixClient that is already in the targetRoom that can be started to listen for the event produced by targetEventThunk.
  * This function assumes that the start() has already been called on the client.
  * @param targetRoom The room to listen for the reaction in.
  * @param reactionKey The reaction key to wait for.
  * @param targetEventThunk A function that produces an event ID when called. This event ID is then used to listen for a reaction.
  * @returns The reaction event.
  */
-export async function getFirstReaction(client: MatrixClient, targetRoom: string, reactionKey: string, targetEventThunk: () => Promise<string>): Promise<any> {
+export async function getFirstReaction(matrix: MatrixEmitter, targetRoom: string, reactionKey: string, targetEventThunk: () => Promise<string>): Promise<any> {
     let reactionEvents: any[] = [];
     const addEvent = function (roomId: string, event: any) {
         if (roomId !== targetRoom) return;
@@ -90,7 +91,7 @@ export async function getFirstReaction(client: MatrixClient, targetRoom: string,
     };
     let targetCb;
     try {
-        client.on('room.event', addEvent)
+        matrix.on('room.event', addEvent)
         const targetEventId = await targetEventThunk();
         for (let event of reactionEvents) {
             const relates_to = event.content['m.relates_to'];
@@ -107,12 +108,12 @@ export async function getFirstReaction(client: MatrixClient, targetRoom: string,
                     resolve(event)
                 }
             }
-            client.on('room.event', targetCb);
+            matrix.on('room.event', targetCb);
         });
     } finally {
-        client.removeListener('room.event', addEvent);
+        matrix.removeListener('room.event', addEvent);
         if (targetCb) {
-            client.removeListener('room.event', targetCb);
+            matrix.removeListener('room.event', targetCb);
         }
     }
 }

test/integration/mjolnirSetupUtils.ts

@@ -84,7 +84,7 @@ export async function makeMjolnir(config: IConfig): Promise<Mjolnir> {
     await overrideRatelimitForUser(config.homeserverUrl, await client.getUserId());
     patchMatrixClient();
     await ensureAliasedRoomExists(client, config.managementRoom);
-    let mj = await Mjolnir.setupMjolnirFromConfig(client, config);
+    let mj = await Mjolnir.setupMjolnirFromConfig(client, client, config);
     globalClient = client;
     globalMjolnir = mj;
     return mj;

test/integration/roomMembersTest.ts

@@ -310,7 +310,7 @@ describe("Test: Testing RoomMemberManager", function() {
 
         // Initially, the command should show that same result.
         for (let roomId of roomIds) {
-            const reply = await getFirstReply(mjolnir.client, mjolnir.managementRoomId, () => {
+            const reply = await getFirstReply(mjolnir.matrixEmitter, mjolnir.managementRoomId, () => {
                 const command = `!mjolnir status joins ${roomId}`;
                 return this.moderator.sendMessage(mjolnir.managementRoomId, { msgtype: 'm.text', body: command });
             });
@@ -328,7 +328,7 @@ describe("Test: Testing RoomMemberManager", function() {
             const roomId = roomIds[i];
             const joined = manager.getUsersInRoom(roomId, start, 100);
             assert.equal(joined.length, SAMPLE_SIZE / 2 /* half of the users */ + 1 /* mjolnir */, "We should now see all joined users in the room");
-            const reply = await getFirstReply(mjolnir.client, mjolnir.managementRoomId, () => {
+            const reply = await getFirstReply(mjolnir.matrixEmitter, mjolnir.managementRoomId, () => {
                 const command = `!mjolnir status joins ${roomId}`;
                 return this.moderator.sendMessage(mjolnir.managementRoomId, { msgtype: 'm.text', body: command });
             });
@@ -361,7 +361,7 @@ describe("Test: Testing RoomMemberManager", function() {
 
         for (let i = 0; i < roomIds.length; ++i) {
             const roomId = roomIds[i];
-            const reply = await getFirstReply(mjolnir.client, mjolnir.managementRoomId, () => {
+            const reply = await getFirstReply(mjolnir.matrixEmitter, mjolnir.managementRoomId, () => {
                 const command = `!mjolnir status joins ${roomId}`;
                 return this.moderator.sendMessage(mjolnir.managementRoomId, { msgtype: 'm.text', body: command });
             });
@@ -684,7 +684,7 @@ describe("Test: Testing RoomMemberManager", function() {
             assert.ok(joined.length >= 2 * SAMPLE_SIZE, `In experiment ${experiment.name}, we should have seen ${2 * SAMPLE_SIZE} users, saw ${joined.length}`);
 
             // Run experiment.
-            await getNthReply(mjolnir.client, mjolnir.managementRoomId, experiment.n, async () => {
+            await getNthReply(mjolnir.matrixEmitter, mjolnir.managementRoomId, experiment.n, async () => {
                 const command = experiment.command(roomId, roomAlias);
                 let result = await this.moderator.sendMessage(mjolnir.managementRoomId, { msgtype: 'm.text', body: command });
                 return result;