send history in MUCs

changelog.d/227.feature

@@ -0,0 +1 @@
+Send room history to XMPP clients when they join.

src/GatewayHandler.ts

@@ -12,6 +12,8 @@ import { BifrostRemoteUser } from "./store/BifrostRemoteUser";
 
 const log = Logging.get("GatewayHandler");
 
+const HISTORY_SAFE_ENUMS = ['shared', 'world_readable'];
+
 /**
  * Responsible for handling querys & events on behalf of a gateway style bridge.
  * The gateway system in the bridge is complex, so pull up a a pew and let's dig in.
@@ -58,10 +60,11 @@ export class GatewayHandler {
         }
         const promise = (async () => {
             log.debug(`Getting state for ${roomId}`);
-            const state = await intent.roomState(roomId, false);
+            const state = await intent.roomState(roomId);
             log.debug(`Got state for ${roomId}`);
             const nameEv = state.find((e) => e.type === "m.room.name");
             const topicEv = state.find((e) => e.type === "m.room.topic");
+            const historyVis = state.find((e) => e.type === "m.room.history_visibility");
             const bot = this.bridge.getBot();
             const membership = state.filter((e) => e.type === "m.room.member").map((e: WeakEvent) => (
                 {
@@ -73,6 +76,8 @@ export class GatewayHandler {
                 }
             ))
             const room: IGatewayRoom = {
+                // Default to private
+                allowHistory: HISTORY_SAFE_ENUMS.includes(historyVis?.content?.history_visibility || 'joined'),
                 name: nameEv ? nameEv.content.name : "",
                 topic: topicEv ? topicEv.content.topic : "",
                 roomId,

src/bifrost/Gateway.ts

@@ -26,6 +26,7 @@ export interface IGatewayRoom {
     topic: string;
     avatar?: string;
     roomId: string;
+    allowHistory: boolean;
     membership: {
         sender: string;
         stateKey: string;

src/xmppjs/HistoryManager.ts

@@ -0,0 +1,121 @@
+import { Element } from "@xmpp/xml";
+import { JID } from "@xmpp/jid";
+
+export interface IHistoryLimits {
+    maxchars?: number,
+    maxstanzas?: number,
+    seconds?: number,
+    since?: Date,
+}
+
+/**
+ * Abstraction of a history storage backend.
+ */
+interface IHistoryStorage {
+    // add a message to the history of a given room
+    addMessage: (chatName: string, message: Element, jid: JID) => unknown;
+    // get the history of a room.  The storage should apply the limits that it
+    // wishes too, and remove the limits that it applied from the `limits`
+    // parameter.  The returned list of Elements must include the Delayed
+    // Delivery information.
+    getHistory: (chatName: string, limits: IHistoryLimits) => Promise<Element[]>;
+}
+
+/**
+ * Store room history in memory.
+ */
+export class MemoryStorage implements IHistoryStorage {
+    private history: Map<string, Element[]>;
+    constructor(public maxHistory: number) {
+        this.history = new Map();
+    }
+
+    addMessage(chatName: string, message: Element, jid: JID): void {
+        if (!this.history.has(chatName)) {
+            this.history.set(chatName, []);
+        }
+        const currRoomHistory = this.history.get(chatName);
+
+        // shallow-copy the message, and add the timestamp
+        const copiedMessage = new Element(message.name, message.attrs);
+        copiedMessage.append(message.children as Element[]);
+        copiedMessage.attr("from", jid.toString());
+        copiedMessage.append(new Element("delay", {
+            xmlns: "urn:xmpp:delay",
+            from: chatName,
+            stamp: (new Date()).toISOString(),
+        }));
+
+        currRoomHistory.push(copiedMessage);
+
+        while (currRoomHistory.length > this.maxHistory) {
+            currRoomHistory.shift();
+        }
+    }
+
+    async getHistory(chatName: string, limits: IHistoryLimits): Promise<Element[]> {
+        return this.history.get(chatName) || [];
+    }
+}
+
+// TODO: make a class that stores in PostgreSQL so that we don't lose history
+// when we restart
+
+/**
+ * Manage room history for a MUC
+ */
+export class HistoryManager {
+    constructor(
+        private storage: IHistoryStorage,
+    ) {}
+
+    addMessage(chatName: string, message: Element, jid: JID): unknown {
+        return this.storage.addMessage(chatName, message, jid);
+    }
+
+    async getHistory(chatName: string, limits: IHistoryLimits): Promise<Element[]> {
+        if (limits.seconds) {
+            const since = new Date(Date.now() - limits.seconds * 1000);
+            if (limits.since === undefined || limits.since < since) {
+                limits.since = since;
+            }
+            delete limits.seconds;
+        }
+        let history: Element[] = await this.storage.getHistory(chatName, limits);
+
+        // index of the first history element that we will keep after applying
+        // the limits
+        let idx = 0;
+
+        if ("maxstanzas" in limits && history.length > limits.maxstanzas) {
+            idx = history.length - limits.maxstanzas;
+        }
+
+        if ("since" in limits) {
+            // FIXME: binary search would be better than linear search
+            for (; idx < history.length; idx++) {
+                try {
+                    const ts = history[idx].getChild("delay", "urn:xmpp:delay")?.attr("stamp");
+                    if (new Date(ts) >= limits.since) {
+                        break;
+                    }
+                } catch {}
+            }
+        }
+
+        if ("maxchars" in limits) {
+            let numChars = 0;
+            let i = history.length;
+            for (; i > idx && numChars < limits.maxchars; i--) {
+                numChars += history[i - 1].toString().length;
+            }
+            idx = i;
+        }
+
+        if (idx > 0) {
+            history = history.slice(idx);
+        }
+
+        return history;
+    }
+}

src/xmppjs/XJSGateway.ts

@@ -1,10 +1,17 @@
 import { XmppJsInstance, XMPP_PROTOCOL } from "./XJSInstance";
 import { Element, x } from "@xmpp/xml";
+import parse from "@xmpp/xml/lib/parse";
 import { jid, JID } from "@xmpp/jid";
 import { Logging } from "matrix-appservice-bridge";
 import { IConfigBridge } from "../Config";
 import { IBasicProtocolMessage } from "..//MessageFormatter";
-import { IGatewayJoin, IUserStateChanged, IStoreRemoteUser, IUserInfo } from "../bifrost/Events";
+import {
+    IGatewayJoin,
+    IUserStateChanged,
+    IStoreRemoteUser,
+    IUserInfo,
+    IReceivedImMsg
+} from "../bifrost/Events";
 import { IGatewayRoom } from "../bifrost/Gateway";
 import { PresenceCache } from "./PresenceCache";
 import { XHTMLIM } from "./XHTMLIM";
@@ -17,6 +24,7 @@ import { XMPPStatusCode } from "./XMPPConstants";
 import { AutoRegistration } from "../AutoRegistration";
 import { GatewayStateResolve } from "./GatewayStateResolve";
 import { MatrixMembershipEvent } from "../MatrixTypes";
+import { IHistoryLimits, HistoryManager, MemoryStorage } from "./HistoryManager";
 
 const log = Logging.get("XmppJsGateway");
 
@@ -31,15 +39,15 @@ export interface RemoteGhostExtraData {
  * and XMPP.
  */
 export class XmppJsGateway implements IGateway {
-    // For storing room history, should be clipped at MAX_HISTORY per room.
-    private roomHistory: Map<string, [Element]>;
+    // For storing room history
+    private roomHistory: HistoryManager;
     // For storing requests to be responded to, like joins
     private stanzaCache: Map<string, Element>; // id -> stanza
     private presenceCache: PresenceCache;
     // Storing every XMPP user and their anonymous.
     private members: GatewayMUCMembership;
     constructor(private xmpp: XmppJsInstance, private registration: AutoRegistration, private config: IConfigBridge) {
-        this.roomHistory = new Map();
+        this.roomHistory = new HistoryManager(new MemoryStorage(50));
         this.stanzaCache = new Map();
         this.members = new GatewayMUCMembership();
         this.presenceCache = new PresenceCache(true);
@@ -163,6 +171,18 @@ export class XmppJsGateway implements IGateway {
                 "groupchat",
             )
         );
+
+        // add the message to the room history
+        const historyStanza = new StzaMessage(
+            from.anonymousJid.toString(),
+            "",
+            msg,
+            "groupchat",
+        );
+        if (room.allowHistory) {
+            this.roomHistory.addMessage(chatName, parse(historyStanza.xml), from.anonymousJid);
+        }
+
         return this.xmpp.xmppSendBulk(msgs);
     }
 
@@ -201,6 +221,16 @@ export class XmppJsGateway implements IGateway {
             return false;
         }
         stanza.attrs.from = preserveFrom;
+        try {
+            // TODO: Currently we have no way to determine if this room has private history,
+            // so we may be adding more strain to the cache than nessacery.
+            this.roomHistory.addMessage(
+                chatName, stanza,
+                member.anonymousJid,
+            );
+        } catch (ex) {
+            log.warn(`Failed to add message for ${chatName} to history cache`);
+        }
         return true;
     }
 
@@ -418,13 +448,52 @@ export class XmppJsGateway implements IGateway {
         );
 
         // 4. Room history
-        log.debug("Emitting history");
-        const history: Element[] = this.roomHistory.get(room.roomId) || [];
-        history.forEach((e) => {
-            e.attrs.to = stanza.attrs.from;
-            // TODO: Add delay info to this.
-            this.xmpp.xmppWriteToStream(e);
-        });
+        if (room.allowHistory) {
+            log.debug("Emitting history");
+            const historyLimits: IHistoryLimits = {};
+            const historyRequest = stanza.getChild("x", "http://jabber.org/protocol/muc")?.getChild("history");
+            if (historyRequest !== undefined) {
+                const getIntValue = (str) => {
+                    if (!/^\d+$/.test(str)) {
+                        throw new Error("Not a number");
+                    }
+                    return parseInt(str);
+                };
+                const getDateValue = (str) => {
+                    const val = new Date(str);
+                    // TypeScript doesn't like giving a Date to isNaN, even though it
+                    // works.  And it doesn't like converting directly to number.
+                    if (isNaN(val as unknown as number)) {
+                        throw new Error("Not a date");
+                    }
+                    return val;
+                };
+                const getHistoryParam = (name: string, parser: (str: string) => any): void => {
+                    const param = historyRequest.getAttr(name);
+                    if (param !== undefined) {
+                        try {
+                            historyLimits[name] = parser(param);
+                        } catch (e) {
+                            log.debug(`Invalid ${name} in history management: "${param}" (${e})`);
+                        }
+                    }
+                };
+                getHistoryParam("maxchars", getIntValue);
+                getHistoryParam("maxstanzas", getIntValue);
+                getHistoryParam("seconds", getIntValue);
+                getHistoryParam("since", getDateValue);
+            } else {
+                // default to 20 stanzas if the client doesn't specify
+                historyLimits.maxstanzas = 20;
+            }
+            const history: Element[] = await this.roomHistory.getHistory(chatName, historyLimits);
+            history.forEach((e) => {
+                e.attrs.to = stanza.attrs.from;
+                this.xmpp.xmppWriteToStream(e);
+            });
+        } else {
+            log.debug("Not emitting history, room does not have visibility turned on");
+        }
 
         log.debug("Emitting subject");
         // 5. The room subject

test/xmppjs/test_HistoryManager.ts

@@ -0,0 +1,49 @@
+import { expect } from "chai";
+import { MemoryStorage, HistoryManager } from "../../src/xmppjs/HistoryManager";
+import { Element } from "@xmpp/xml";
+import { JID } from "@xmpp/jid";
+
+describe("HistoryManager", () => {
+    describe("MemoryStorage", () => {
+        it("should return filtered history", async () => {
+            const historyManager = new HistoryManager(new MemoryStorage(20));
+            historyManager.addMessage(
+                "[email protected]", new Element("stanza1"),
+                new JID("room1", "example.org", "user1"),
+            );
+            historyManager.addMessage(
+                "[email protected]", new Element("stanza2"),
+                new JID("room1", "example.org", "user1"),
+            );
+            historyManager.addMessage(
+                "[email protected]", new Element("stanza3"),
+                new JID("room1", "example.org", "user1"),
+            );
+            historyManager.addMessage(
+                "[email protected]", new Element("stanza1"),
+                new JID("room1", "example.org", "user1"),
+            );
+
+            const unfilteredRoom1 = await historyManager.getHistory("[email protected]", {});
+            expect(unfilteredRoom1.length).to.equal(3);
+            const unfilteredRoom2 = await historyManager.getHistory("[email protected]", {});
+            expect(unfilteredRoom2.length).to.equal(1);
+
+            const maxStanzasRoom1 = await historyManager.getHistory("[email protected]", {
+                maxstanzas: 2,
+            });
+            expect(maxStanzasRoom1.length).to.equal(2);
+            const maxStanzasRoom2 = await historyManager.getHistory("[email protected]", {
+                maxstanzas: 2,
+            });
+            expect(maxStanzasRoom2.length).to.equal(1);
+
+            // each stanza will be about 40 characters, so maxchars 50 should
+            // only give us one stanza
+            const maxCharsRoom1 = await historyManager.getHistory("[email protected]", {
+                maxchars: 50,
+            });
+            expect(maxCharsRoom1.length).to.equal(1);
+        });
+    });
+});

test/xmppjs/test_XJSGateway.ts

@@ -73,6 +73,7 @@ describe("XJSGateway", () => {
                 topic: "GatewayTopic",
                 roomId: "!foo:bar",
                 membership: [],
+                allowHistory: true,
             };
             try {
                 await gw.onRemoteJoin(null, "myjoinid", room, "@_xmpp_foo:bar");
@@ -93,6 +94,7 @@ describe("XJSGateway", () => {
                     createMember("@_xmpp_baz:bar", "Baz"),
                     createMember("@leavy:bar", "Leavy", "leave"),
                 ],
+                allowHistory: true,
             };
             gw.handleStanza(
                 x("presence", {
@@ -150,6 +152,7 @@ describe("XJSGateway", () => {
                     createMember("@_xmpp_baz:bar", "Baz"),
                     createMember("@leavy:bar", "Leavy", "leave"),
                 ],
+                allowHistory: true,
             };
             gw.handleStanza(
                 x("presence", {
@@ -178,6 +181,7 @@ describe("XJSGateway", () => {
                 topic: "GatewayTopic",
                 roomId: "!foo:bar",
                 membership,
+                allowHistory: true,
             };
             gw.handleStanza(
                 x("presence", {
@@ -202,6 +206,7 @@ describe("XJSGateway", () => {
                 topic: "GatewayTopic",
                 roomId: "!foo:bar",
                 membership: [],
+                allowHistory: true,
             };
             gw.handleStanza(
                 x("presence", {