Add TimelineRedactionQueue. (#949)

This just batches the redactions going on within a single room to the
same `/messages` request. And blocks the next request for that room
from happening until the current request has completed.

This is necessary because `/messages` pagination is really heavy on
homeservers as it can force them to backfill.

There is also some really strange bug in Synapse that is triggered by
Draupnir's behaviour
element-hq/synapse#18829.

Author: Gnuxie

package.json

@@ -53,7 +53,7 @@
     "@sinclair/typebox": "0.34.13",
     "@the-draupnir-project/interface-manager": "4.2.5",
     "@the-draupnir-project/matrix-basic-types": "1.4.0",
-    "@the-draupnir-project/mps-interface-adaptor": "^0.4.1",
+    "@the-draupnir-project/mps-interface-adaptor": "0.5.0",
     "better-sqlite3": "^9.4.3",
     "body-parser": "^1.20.2",
     "config": "^3.3.9",
@@ -63,8 +63,8 @@
     "jsdom": "^24.0.0",
     "matrix-appservice-bridge": "^10.3.1",
     "matrix-bot-sdk": "npm:@vector-im/matrix-bot-sdk@^0.7.1-element.6",
-    "matrix-protection-suite": "npm:@gnuxie/matrix-protection-suite@3.11.0",
-    "matrix-protection-suite-for-matrix-bot-sdk": "npm:@gnuxie/matrix-protection-suite-for-matrix-bot-sdk@3.11.0",
+    "matrix-protection-suite": "npm:@gnuxie/matrix-protection-suite@3.12.0",
+    "matrix-protection-suite-for-matrix-bot-sdk": "npm:@gnuxie/matrix-protection-suite-for-matrix-bot-sdk@3.12.0",
     "pg": "^8.8.0",
     "yaml": "^2.3.2"
   },

src/Draupnir.ts

@@ -80,6 +80,7 @@ import {
   MatrixReactionHandler,
   sendMatrixEventsFromDeadDocument,
 } from "@the-draupnir-project/mps-interface-adaptor";
+import { TimelineRedactionQueue } from "./queues/TimelineRedactionQueue";
 
 const log = new Logger("Draupnir");
 
@@ -135,6 +136,12 @@ export class Draupnir implements Client, MatrixAdaptorContext {
 
   private readonly JSInterfaceDispatcher: JSInterfaceCommandDispatcher<BasicInvocationInformation> =
     makeDraupnirJSCommandDispatcher(this);
+
+  public readonly timelineRedactionQueue = new TimelineRedactionQueue(
+    this.clientPlatform.toRoomMessages(),
+    this.clientPlatform.toRoomEventRedacter()
+  );
+
   private constructor(
     public readonly client: MatrixSendClient,
     public readonly clientUserID: StringUserID,

src/protections/RedactionSynchronisation.ts

@@ -4,8 +4,6 @@
 
 import {
   AbstractProtection,
-  ActionException,
-  ActionExceptionKind,
   ActionResult,
   Capability,
   CapabilityMethodSchema,
@@ -41,10 +39,9 @@ import {
   StringRoomID,
   StringUserID,
 } from "@the-draupnir-project/matrix-basic-types";
-import { Result } from "@gnuxie/typescript-result";
+import { isError, Result } from "@gnuxie/typescript-result";
 import { Type } from "@sinclair/typebox";
 import { revisionRulesMatchingUser } from "../commands/unban/UnbanUsers";
-import { redactUserMessagesIn } from "../utils";
 
 export type RedactionSynchronisationProtectionCapabilitiesSet = {
   consequences: RedactionSynchronisationConsequences;
@@ -106,23 +103,20 @@ describeCapabilityProvider({
       requiredStatePermissions: [],
       requiredEventPermissions: [],
       async redactMessagesIn(userIDOrGlob, reason, roomIDs) {
-        const redactionResult = await redactUserMessagesIn(
-          draupnir.client,
-          draupnir.managementRoomOutput,
-          userIDOrGlob,
-          roomIDs
-        ).then(
-          (_) => Ok(undefined),
-          (error) =>
-            ActionException.Result(
-              `Error redacting messages for ${userIDOrGlob}`,
-              {
-                exception: error,
-                exceptionKind: ActionExceptionKind.Unknown,
-              }
+        const redactionResults = await Promise.all(
+          roomIDs.map((roomID) =>
+            draupnir.timelineRedactionQueue.enqueueRedaction(
+              userIDOrGlob,
+              roomID
             )
+          )
         );
-        return redactionResult;
+        const firstError = redactionResults.find((result) => isError(result));
+        if (firstError) {
+          return firstError;
+        } else {
+          return Ok(undefined);
+        }
       },
       async rejectInvite(roomID, _sender, target, reason) {
         return await draupnir.clientPlatform

src/queues/TimelineRedactionQueue.ts

@@ -0,0 +1,98 @@
+// SPDX-FileCopyrightText: 2025 Gnuxie <[email protected]>
+//
+// SPDX-License-Identifier: AFL-3.0
+
+import { isError, Ok, Result } from "@gnuxie/typescript-result";
+import {
+  MatrixGlob,
+  StringEventID,
+  StringRoomID,
+  StringUserID,
+} from "@the-draupnir-project/matrix-basic-types";
+import {
+  KeyedBatchQueue,
+  Logger,
+  RoomEventRedacter,
+  RoomMessages,
+} from "matrix-protection-suite";
+
+const log = new Logger("TimelineRedactionQueue");
+
+export class TimelineRedactionQueue {
+  private readonly batchProcessor = this.processBatch.bind(this);
+  private readonly queue = new KeyedBatchQueue<StringRoomID, StringUserID>(
+    this.batchProcessor
+  );
+
+  public constructor(
+    private readonly roomMessages: RoomMessages,
+    private readonly roomEventRedacter: RoomEventRedacter,
+    // FIXME: We really should have a way of adding a limit to the batch enqueue operation...
+    // It doesn't really seem possible though. Unless the maximum limit anyone enqueues is chosen.
+    // Yeah that sounds like how it would have to work. Meanwhile, it doesn't really matter since
+    // the current behaviour is constrained to 1000.
+    private readonly limit = 1000
+  ) {
+    // nothing to do.
+  }
+  private async processBatch(
+    roomID: StringRoomID,
+    userIDs: StringUserID[]
+  ): Promise<Result<void>> {
+    const globUserIDs = userIDs.filter(
+      (userID) => userID.includes("*") || userID.includes("?")
+    );
+    const globsToTest = globUserIDs.map((userID) => new MatrixGlob(userID));
+    const isGlobInUsers = globsToTest.length !== 0;
+    const usersToTest = userIDs.filter(
+      (userID) => !globUserIDs.includes(userID)
+    );
+    const paginator = this.roomMessages.toRoomMessagesIterator(roomID, {
+      direction: "backwards",
+      limit: this.limit,
+      ...(isGlobInUsers ? {} : { filter: { senders: userIDs } }),
+    });
+    const eventsToRedact: StringEventID[] = [];
+    const paginationResult = await paginator.forEachItem({
+      forEachItemCB: (event) => {
+        if (
+          // Always add users when there are no globs, since events are filtered by sender.
+          !isGlobInUsers ||
+          usersToTest.includes(event.sender) ||
+          globsToTest.some((glob) => glob.test(event.sender))
+        ) {
+          eventsToRedact.push(event.event_id);
+        }
+      },
+      totalItemLimit: this.limit,
+    });
+    if (isError(paginationResult)) {
+      return paginationResult.elaborate(
+        `Failed to paginate /messages in ${roomID} to begin redaction`
+      );
+    }
+    // TODO: It would be good if we had a way of throttling these requests
+    // per draupnir and in general but y'know.
+    for (const eventID of eventsToRedact) {
+      const redactResult = await this.roomEventRedacter.redactEvent(
+        roomID,
+        eventID
+      );
+      if (isError(redactResult)) {
+        log.error(
+          `Error while trying to redact messages for in ${roomID}:`,
+          eventID,
+          redactResult.error
+        );
+      }
+    }
+    return Ok(undefined);
+  }
+
+  public async enqueueRedaction(
+    userID: StringUserID,
+    roomID: StringRoomID
+  ): Promise<Result<void>> {
+    return await this.queue.enqueue(roomID, userID);
+  }
+}

test/integration/commands/commandUtils.ts

@@ -22,6 +22,8 @@ import {
   TextMessageContent,
   Value,
   StringEventIDSchema,
+  ReactionContent,
+  hasOwn,
 } from "matrix-protection-suite";
 import { Type } from "@sinclair/typebox";
 import { Draupnir } from "../../../src/Draupnir";
@@ -145,14 +147,19 @@ export async function getFirstReaction(
   const addEvent = function (roomId: string, event: RoomEvent) {
     if (roomId !== targetRoom) return;
     if (!Value.Check(ReactionEvent, event)) return;
+    if (Object.keys(event.content).length === 0) return;
     reactionEvents.push(event);
   };
   let targetCb;
   try {
     matrix.on("room.event", addEvent);
     const targetEventId = await targetEventThunk();
     for (const event of reactionEvents) {
-      const relates_to = event.content?.["m.relates_to"];
+      const relates_to = (
+        hasOwn(event.content, "m.relates_to")
+          ? event.content["m.relates_to"]
+          : undefined
+      ) as ReactionContent["m.relates_to"];
       if (
         relates_to?.event_id === targetEventId &&
         relates_to.key === reactionKey
@@ -164,7 +171,11 @@ export async function getFirstReaction(
       targetCb = function (roomId: string, event: RoomEvent) {
         if (roomId !== targetRoom) return;
         if (!Value.Check(ReactionEvent, event)) return;
-        const relates_to = event.content["m.relates_to"];
+        const relates_to = (
+          hasOwn(event.content, "m.relates_to")
+            ? event.content["m.relates_to"]
+            : undefined
+        ) as ReactionContent["m.relates_to"];
         if (
           relates_to?.event_id === targetEventId &&
           relates_to.key === reactionKey

yarn.lock

@@ -327,10 +327,10 @@
     "@gnuxie/typescript-result" "^1.0.0"
     glob-to-regexp "^0.4.1"
 
-"@the-draupnir-project/mps-interface-adaptor@^0.4.1":
-  version "0.4.1"
-  resolved "https://registry.yarnpkg.com/@the-draupnir-project/mps-interface-adaptor/-/mps-interface-adaptor-0.4.1.tgz#673b20e3af916d3587f440fedfd6ea94b60d368e"
-  integrity sha512-nHyQuC7nrJn7RtNpthEWL/6uNK5HPS2UDBT74Ml00FofzqSohAyNpi9BeCi9zkMjpdmVWvsWiEqn2Iv8fc/UjQ==
+"@the-draupnir-project/mps-interface-adaptor@0.5.0":
+  version "0.5.0"
+  resolved "https://registry.yarnpkg.com/@the-draupnir-project/mps-interface-adaptor/-/mps-interface-adaptor-0.5.0.tgz#de0bf229a51c906f4b53f72f6a8ce0be33399434"
+  integrity sha512-7lSDeHGsf/UtXFxJ1OYBt8ScU36bSJAdxLxM4tBBDNe2EWDlyOd1oBaa7OfEcf+/4btBI0T+Np5eseCqVfLo0Q==
 
 "@tsconfig/node10@^1.0.7":
   version "1.0.11"
@@ -2598,18 +2598,18 @@ matrix-appservice@^2.0.0:
     request-promise "^4.2.6"
     sanitize-html "^2.11.0"
 
-"matrix-protection-suite-for-matrix-bot-sdk@npm:@gnuxie/matrix-protection-suite-for-matrix-bot-sdk@3.11.0":
-  version "3.11.0"
-  resolved "https://registry.yarnpkg.com/@gnuxie/matrix-protection-suite-for-matrix-bot-sdk/-/matrix-protection-suite-for-matrix-bot-sdk-3.11.0.tgz#0f972cd69a918422aca5b05f0f917766b2899e13"
-  integrity sha512-fNT0rdGXN8VLZtkqS5s4wySYZPJ3LaqsU/CP3iS4WK7XQ6EXT1n3UM4chxmwL5YhQITRW34PehTx7ScdKpL+4w==
+"matrix-protection-suite-for-matrix-bot-sdk@npm:@gnuxie/matrix-protection-suite-for-matrix-bot-sdk@3.12.0":
+  version "3.12.0"
+  resolved "https://registry.yarnpkg.com/@gnuxie/matrix-protection-suite-for-matrix-bot-sdk/-/matrix-protection-suite-for-matrix-bot-sdk-3.12.0.tgz#74394ad099fb0d60709d817065c8221154b2132f"
+  integrity sha512-hdTLuwvhFg+qcZeOJCrKzhg5Tpn2X6VMRAj2uKQkHdx6KF+2MwiZoOrWCZ1BlssFXOyZwIvTLULeDU7Wa1JNPQ==
   dependencies:
     "@gnuxie/typescript-result" "^1.0.0"
     await-lock "^2.2.2"
 
-"matrix-protection-suite@npm:@gnuxie/matrix-protection-suite@3.11.0":
-  version "3.11.0"
-  resolved "https://registry.yarnpkg.com/@gnuxie/matrix-protection-suite/-/matrix-protection-suite-3.11.0.tgz#2ed31e8bd0c68b89927117090f049b5905c9ed82"
-  integrity sha512-Y/on4KzPN8413dNv1zLBP1LaD9ijgZSR7qXPJUZRZHqezEruQp1bg8q2YIOdzsrI5dk/TK2ftTnEC/lbz2aWYA==
+"matrix-protection-suite@npm:@gnuxie/matrix-protection-suite@3.12.0":
+  version "3.12.0"
+  resolved "https://registry.yarnpkg.com/@gnuxie/matrix-protection-suite/-/matrix-protection-suite-3.12.0.tgz#d89d73b8ff90d0eb7b9c8f9d1771acf042e97bc7"
+  integrity sha512-f/nuzHYaCB/uVwN7UZy4BatcIvmQL/pTzRE4O2TGMrdv2fDQPsSvc3HZyJTJI/t8iU5g+k/58tk2TyJmLzMS6A==
   dependencies:
     "@gnuxie/typescript-result" "^1.0.0"
     await-lock "^2.2.2"