Use a Map for queueing redactions by roomId.

src/Mjolnir.ts

@@ -54,7 +54,15 @@ export class Mjolnir {
     private localpart: string;
     private currentState: string = STATE_NOT_STARTED;
     private protections: IProtection[] = [];
+    /**
+     * This is for users who are not listed on a watchlist,
+     * but have been flagged by the automatic spam detection as suispicous
+     */
     private unlistedUserRedactionQueue = new UnlistedUserRedactionQueue();
+    /**
+     * This is a queue for redactions to process after mjolnir
+     * has finished applying ACL and bans when syncing.
+     */
     private eventRedactionQueue = new EventRedactionQueue();
     private automaticRedactionReasons: MatrixGlob[] = [];
     private protectedJoinedRoomIds: string[] = [];
@@ -140,6 +148,10 @@ export class Mjolnir {
         return this.protections;
     }
 
+    /**
+     * Retruns the handler to flag a user that is not listed on a watchlist
+     * for redaction, removing any future messages that they send.
+     */
     public get unlistedUserRedactionHandler(): UnlistedUserRedactionQueue {
         return this.unlistedUserRedactionQueue;
     }
@@ -486,6 +498,10 @@ export class Mjolnir {
         return errors;
     }
 
+    /**
+     * Sync all the rooms with all the watched lists, banning and applying any changes ACLS.
+     * @param verbose Whether to report any errors to the management room.
+     */
     public async syncLists(verbose = true) {
         for (const list of this.banLists) {
             await list.updateList();
@@ -594,7 +610,7 @@ export class Mjolnir {
                 }
                 return;
             } else if (event['type'] === "m.room.member") {
-                // Only apply bans in the room we're looking at.
+                // Only apply bans and then redactions in the room we're looking at.
                 const banErrors = await applyUserBans(this.banLists, [roomId], this);
                 const redactionErrors = await this.processRedactionQueue(roomId);
                 await this.printActionResult(banErrors);
@@ -671,7 +687,13 @@ export class Mjolnir {
         this.eventRedactionQueue.add(new RedactUserInRoom(userId, roomId));
     }
 
-    public async processRedactionQueue(roomId?: string) {
+    /**
+     * Process all queued redactions, this is usually called at the end of the sync process,
+     * after all users have been banned and ACLs applied.
+     * @param roomId Limit processing to one room only, otherwise process redactions for all rooms.
+     * @returns An array of descriptive errors if any were encountered that can be reported to a management room.
+     */
+    public async processRedactionQueue(roomId?: string): Promise<RoomUpdateError[]> {
         return await this.eventRedactionQueue.process(this.client, roomId);
     }
 }

src/queues/EventRedactionQueue.ts

@@ -20,12 +20,24 @@ import { RoomUpdateError } from "../models/RoomUpdateError";
 import { redactUserMessagesIn } from "../utils";
 
 export interface QueuedRedaction {
-    roomId: string; // The room which the redaction will take place in.
+    /** The room which the redaction will take place in. */
+    readonly roomId: string;
+    /**
+     * Carries out the redaction task and is called by the EventRedactionQueue
+     * when processing this redaction.
+     * @param client A MatrixClient to use to carry out the redaction.
+     */
     redact(client: MatrixClient): Promise<any>
+    /**
+     * 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.
+     */
     redactionEqual(redaction: QueuedRedaction): boolean
-    report(e: any): RoomUpdateError
 }
 
+/**
+ * Redacts all of the messages a user has sent to one room.
+ */
 export class RedactUserInRoom implements QueuedRedaction {
     userId: string;
     roomId: string;
@@ -47,55 +59,79 @@ export class RedactUserInRoom implements QueuedRedaction {
             return false;
         }
     }
-
-    public report(e): RoomUpdateError {
-        const message = e.message || (e.body ? e.body.error : '<no message>');
-        return {
-            roomId: this.roomId,
-            errorMessage: message,
-            errorKind: ERROR_KIND_FATAL,
-        };
-    }
 }
 /**
  * This is a queue for events so that other protections can happen first (e.g. applying room bans to every room).
  */
 export class EventRedactionQueue {
-    private toRedact: Array<QueuedRedaction> = new Array<QueuedRedaction>();
+    private toRedact: Map<string, QueuedRedaction[]> = new Map<string, QueuedRedaction[]>();
 
-    public has(redaction: QueuedRedaction) {
+    /**
-        return this.toRedact.find(r => r.redactionEqual(redaction));
+     * Test whether the redaction is already present in the queue.
+     * @param redaction a QueuedRedaction.
+     * @returns True if the queue already has the redaction, false otherwise.
+     */
+    public has(redaction: QueuedRedaction): boolean {
+        return !!this.toRedact.get(redaction.roomId)?.find(r => r.redactionEqual(redaction));
     }
 
-    public add(redaction: QueuedRedaction) {
+    /**
+     * Adds a QueuedRedaction to the queue to be processed when process is called.
+     * @param redaction A QueuedRedaction to await processing
+     * @returns A boolean that is true if the redaction was added to the queue and false if it was duplicated.
+     */
+    public add(redaction: QueuedRedaction): boolean {
         if (this.has(redaction)) {
-            return;
+            return false;
         } else {
-            this.toRedact.push(redaction);
+            let entry = this.toRedact.get(redaction.roomId);
+            if (entry) {
+                entry.push(redaction);
+            } else {
+                this.toRedact.set(redaction.roomId, [redaction]);
+            }return true;
         }
     }
 
-    public delete(redaction: QueuedRedaction) {
-        this.toRedact = this.toRedact.filter(r => r.redactionEqual(redaction));
-    }
-
     /**
      * Process the redaction queue, carrying out the action of each QueuedRedaction in sequence.
      * @param client The matrix client to use for processing redactions.
-     * @param roomId If the roomId is provided, only redactions for that room will be processed.
+     * @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, roomId?: string): Promise<RoomUpdateError[]> {
+    public async process(client: MatrixClient, limitToRoomId?: string): Promise<RoomUpdateError[]> {
         const errors: RoomUpdateError[] = [];
-        const currentBatch = roomId ? this.toRedact.filter(r => r.roomId === roomId) : this.toRedact;
+        const redact = async (currentBatch: QueuedRedaction[]) => {
-        for (const redaction of currentBatch) {
+            for (const redaction of currentBatch) {
-            try {
+                try {
-                await redaction.redact(client);
+                    await redaction.redact(client);
-            } catch (e) {
+                } catch (e) {
-                errors.push(redaction.report(e));
+                    let roomError: RoomUpdateError;
-            } finally {
+                    if (e.roomId && e.errorMessage && e.errorKind) {
-                // We need to figure out in which circumstances we want to retry here.
+                        roomError = e;
-                this.delete(redaction);
+                    } else {
+                        const message = e.message || (e.body ? e.body.error : '<no message>');
+                        roomError = {
+                            roomId: redaction.roomId,
+                            errorMessage: message,
+                            errorKind: ERROR_KIND_FATAL,
+                        };
+                    }
+                    errors.push(roomError);
+                }
+            }
+        }
+        if (limitToRoomId) {
+            // There might not actually be any queued redactions for this room.
+            let queuedRedactions = this.toRedact.get(limitToRoomId);
+            if (queuedRedactions) {
+                await redact(queuedRedactions);
+                this.toRedact.delete(limitToRoomId);
+            }
+        } else {
+            for (const [roomId, redactions] of this.toRedact) {
+                await redact(redactions);
+                this.toRedact.delete(roomId);
             }
         }
         return errors;

src/queues/UnlistedUserRedactionQueue.ts

@@ -18,8 +18,12 @@ import { logMessage } from "../LogProxy";
 import config from "../config";
 
 /**
- * This is used to redact new events from users who are not banned from a watched list, but have been flagged
+ * This class is a queue of users who have been flagged
  * for redaction by the flooding or image protection.
+ * Specifically any new events sent by a queued user will be redacted.
+ * This does not handle previously sent events, for that see the EventRedactionQueue.
+ * These users are not listed as banned in any watch list and so may continue
+ * to view a room until a moderator can investigate.
  */
 export class UnlistedUserRedactionQueue {
     private usersToRedact: Set<string> = new Set<string>();