Merge upstream release

.github/workflows/static_analysis.yml

@@ -63,6 +63,6 @@ jobs:
       - run: yarn install
       - uses: nick-invision/retry@v2
         with:
-          max_attempts: 5
-          timeout_minutes: 5
+          max_attempts: 3
+          timeout_minutes: 15
           command: yarn test

package.json

@@ -20,6 +20,7 @@
     "docs": "jsdoc -c jsdoc.json -P package.json -u docs/tutorials",
     "build": "tsc --listEmittedFiles -p tsconfig-release.json",
     "lint": "eslint \"{src,test,examples}/**/*.ts\"",
+    "lint:fix": "eslint \"{src,test,examples}/**/*.ts\" --fix",
     "test": "jest",
     "build:examples": "tsc -p tsconfig-examples.json",
     "example:bot": "yarn build:examples && node lib/examples/bot.js",
@@ -68,6 +69,7 @@
     "lru-cache": "^10.0.1",
     "mkdirp": "^3.0.1",
     "morgan": "^1.10.0",
+    "postgres": "^3.4.1",
     "request": "^2.88.2",
     "request-promise": "^4.2.6",
     "sanitize-html": "^2.11.0"
@@ -76,6 +78,7 @@
     "@babel/core": "^7.23.2",
     "@babel/eslint-parser": "^7.22.15",
     "@babel/eslint-plugin": "^7.22.10",
+    "@testcontainers/postgresql": "^10.2.2",
     "@types/async-lock": "^1.4.1",
     "@types/jest": "^29.5.6",
     "@types/lowdb": "^1.0.14",
@@ -96,6 +99,7 @@
     "matrix-mock-request": "^2.6.0",
     "simple-mock": "^0.8.0",
     "taffydb": "^2.7.3",
+    "testcontainers": "^10.2.2",
     "tmp": "^0.2.1",
     "ts-jest": "^29.1.1",
     "typescript": "^5.2.2"

src/MatrixClient.ts

@@ -728,6 +728,7 @@ export class MatrixClient extends EventEmitter {
                     LogService.info("MatrixClientLite", "Client stop requested - cancelling sync");
                     return;
                 }
+
                 LogService.error("MatrixClientLite", "Error handling sync " + extractRequestError(e));
                 const backoffTime = SYNC_BACKOFF_MIN_MS + Math.random() * (SYNC_BACKOFF_MAX_MS - SYNC_BACKOFF_MIN_MS);
                 LogService.info("MatrixClientLite", `Backing off for ${backoffTime}ms`);

src/appservice/Appservice.ts

@@ -302,12 +302,7 @@ export class Appservice extends EventEmitter {
         this.app.post("/_matrix/app/v1/unstable/org.matrix.msc3984/keys/query", this.onKeysQuery.bind(this));
         this.app.post("/unstable/org.matrix.msc3984/keys/query", this.onKeysQuery.bind(this));
 
-        // Everything else should 404
-        // Technically, according to https://spec.matrix.org/v1.6/application-service-api/#unknown-routes we should
-        // be returning 405 for *known* endpoints with the wrong method.
-        this.app.all("*", (req: express.Request, res: express.Response) => {
-            res.status(404).json({ errcode: "M_UNRECOGNIZED", error: "Endpoint not implemented" });
-        });
+        // We register the 404 handler in the `begin()` function to allow consumers to add their own endpoints.
 
         if (!this.registration.namespaces || !this.registration.namespaces.users || this.registration.namespaces.users.length === 0) {
             throw new Error("No user namespaces in registration");
@@ -381,6 +376,13 @@ export class Appservice extends EventEmitter {
      */
     public begin(): Promise<void> {
         return new Promise<void>((resolve, reject) => {
+            // Per constructor, all other endpoints should 404.
+            // Technically, according to https://spec.matrix.org/v1.6/application-service-api/#unknown-routes we should
+            // be returning 405 for *known* endpoints with the wrong method.
+            this.app.all("*", (req: express.Request, res: express.Response) => {
+                res.status(404).json({ errcode: "M_UNRECOGNIZED", error: "Endpoint not implemented" });
+            });
+
             this.appServer = this.app.listen(this.options.port, this.options.bindAddress, () => resolve());
         }).then(async () => {
             if (this.options.intentOptions?.encryption) {

src/index.ts

@@ -93,6 +93,7 @@ export * from "./storage/MemoryStorageProvider";
 export * from "./storage/SimpleFsStorageProvider";
 export * from "./storage/ICryptoStorageProvider";
 export * from "./storage/RustSdkCryptoStorageProvider";
+export * from "./storage/SimplePostgresStorageProvider";
 
 // Strategies
 export * from "./strategies/AppserviceJoinRoomStrategy";

src/storage/SimplePostgresStorageProvider.ts

@@ -0,0 +1,168 @@
+import * as postgres from "postgres";
+
+import { IStorageProvider } from "./IStorageProvider";
+import { IAppserviceStorageProvider } from "./IAppserviceStorageProvider";
+import { IFilterInfo } from "../IFilter";
+
+/**
+ * A barebones postgresql storage provider. It is not efficient, but it does work.
+ * @category Storage providers
+ */
+export class SimplePostgresStorageProvider implements IStorageProvider, IAppserviceStorageProvider {
+    private readonly db: postgres.Sql;
+    private readonly waitPromise: Promise<void>;
+    private completedTransactions = [];
+
+    /**
+     * Creates a new simple postgresql storage provider.
+     * @param connectionString The `postgres://` connection string to use.
+     * @param trackTransactionsInMemory True (default) to track all received appservice transactions rather than on disk.
+     * @param maxInMemoryTransactions The maximum number of transactions to hold in memory before rotating the oldest out. Defaults to 20.
+     */
+    constructor(connectionString: string, private trackTransactionsInMemory = true, private maxInMemoryTransactions = 20) {
+        this.db = postgres(connectionString);
+
+        this.waitPromise = Promise.all([
+            this.db`
+                CREATE TABLE IF NOT EXISTS bot_metadata (key TEXT NOT NULL PRIMARY KEY, value TEXT);
+            `,
+            this.db`
+                CREATE TABLE IF NOT EXISTS bot_kv (key TEXT NOT NULL PRIMARY KEY, value TEXT);
+            `,
+            this.db`
+                CREATE TABLE IF NOT EXISTS appservice_users (user_id TEXT NOT NULL PRIMARY KEY, registered BOOLEAN NOT NULL);
+            `,
+            this.db`
+                CREATE TABLE IF NOT EXISTS appservice_transactions (txn_id TEXT NOT NULL PRIMARY KEY, completed BOOLEAN NOT NULL);
+            `,
+        ]).then();
+    }
+
+    public async setSyncToken(token: string | null): Promise<any> {
+        await this.waitPromise;
+        return this.db`
+            INSERT INTO bot_metadata (key, value) VALUES ('syncToken', ${token}) 
+            ON CONFLICT (key) DO UPDATE SET value = ${token};
+        `;
+    }
+
+    public async getSyncToken(): Promise<string | null> {
+        await this.waitPromise;
+        return (await this.db`
+            SELECT value FROM bot_metadata WHERE key = 'syncToken';
+        `)[0]?.value;
+    }
+
+    public async setFilter(filter: IFilterInfo): Promise<any> {
+        await this.waitPromise;
+        const filterStr = filter ? JSON.stringify(filter) : null;
+        return this.db`
+            INSERT INTO bot_metadata (key, value) VALUES ('filter', ${filterStr}) 
+            ON CONFLICT (key) DO UPDATE SET value = ${filterStr};
+        `;
+    }
+
+    public async getFilter(): Promise<IFilterInfo> {
+        await this.waitPromise;
+        const value = (await this.db`
+            SELECT value FROM bot_metadata WHERE key = 'filter';
+        `)[0]?.value;
+        return typeof value === "string" ? JSON.parse(value) : value;
+    }
+
+    public async addRegisteredUser(userId: string): Promise<any> {
+        await this.waitPromise;
+        return this.db`
+            INSERT INTO appservice_users (user_id, registered) VALUES (${userId}, TRUE) 
+            ON CONFLICT (user_id) DO UPDATE SET registered = TRUE;
+        `;
+    }
+
+    public async isUserRegistered(userId: string): Promise<boolean> {
+        await this.waitPromise;
+        return !!(await this.db`
+            SELECT registered FROM appservice_users WHERE user_id = ${userId};
+        `)[0]?.registered;
+    }
+
+    public async setTransactionCompleted(transactionId: string): Promise<any> {
+        await this.waitPromise;
+        if (this.trackTransactionsInMemory) {
+            if (this.completedTransactions.indexOf(transactionId) === -1) {
+                this.completedTransactions.push(transactionId);
+            }
+            if (this.completedTransactions.length > this.maxInMemoryTransactions) {
+                this.completedTransactions = this.completedTransactions.reverse().slice(0, this.maxInMemoryTransactions).reverse();
+            }
+            return;
+        }
+
+        return this.db`
+            INSERT INTO appservice_transactions (txn_id, completed) VALUES (${transactionId}, TRUE) 
+            ON CONFLICT (txn_id) DO UPDATE SET completed = TRUE;
+        `;
+    }
+
+    public async isTransactionCompleted(transactionId: string): Promise<boolean> {
+        await this.waitPromise;
+        if (this.trackTransactionsInMemory) {
+            return this.completedTransactions.includes(transactionId);
+        }
+
+        return (await this.db`
+            SELECT completed FROM appservice_transactions WHERE txn_id = ${transactionId};
+        `)[0]?.completed;
+    }
+
+    public async readValue(key: string): Promise<string | null | undefined> {
+        await this.waitPromise;
+        return (await this.db`
+            SELECT value FROM bot_kv WHERE key = ${key};
+        `)[0]?.value;
+    }
+
+    public async storeValue(key: string, value: string): Promise<void> {
+        await this.waitPromise;
+        return this.db`
+            INSERT INTO bot_kv (key, value) VALUES (${key}, ${value}) 
+            ON CONFLICT (key) DO UPDATE SET value = ${value};            
+        `.then();
+    }
+
+    public storageForUser(userId: string): IStorageProvider {
+        return new NamespacedPostgresProvider(userId, this);
+    }
+}
+
+/**
+ * A namespaced storage provider that uses postgres to store information.
+ * @category Storage providers
+ */
+class NamespacedPostgresProvider implements IStorageProvider {
+    constructor(private prefix: string, private parent: SimplePostgresStorageProvider) {
+    }
+
+    public setFilter(filter: IFilterInfo): Promise<any> | void {
+        return this.parent.storeValue(`${this.prefix}_internal_filter`, JSON.stringify(filter));
+    }
+
+    public async getFilter(): Promise<IFilterInfo> {
+        return this.parent.readValue(`${this.prefix}_internal_filter`).then(r => r ? JSON.parse(r) : r);
+    }
+
+    public setSyncToken(token: string | null): Promise<any> | void {
+        return this.parent.storeValue(`${this.prefix}_internal_syncToken`, token ?? "");
+    }
+
+    public async getSyncToken(): Promise<string> {
+        return this.parent.readValue(`${this.prefix}_internal_syncToken`).then(r => r === "" ? null : r);
+    }
+
+    public storeValue(key: string, value: string): Promise<any> | void {
+        return this.parent.storeValue(`${this.prefix}_internal_kv_${key}`, value);
+    }
+
+    public readValue(key: string): string | Promise<string | null | undefined> | null | undefined {
+        return this.parent.readValue(`${this.prefix}_internal_kv_${key}`);
+    }
+}

test/appservice/AppserviceTest.ts

@@ -1988,6 +1988,44 @@ describe('Appservice', () => {
         }
     });
 
+    it('should allow custom endpoints to be added to the express instance', async () => {
+        const port = await getPort();
+        const hsToken = "s3cret_token";
+        const appservice = new Appservice({
+            port: port,
+            bindAddress: '',
+            homeserverName: 'example.org',
+            homeserverUrl: 'https://localhost',
+            registration: {
+                as_token: "",
+                hs_token: hsToken,
+                sender_localpart: "_bot_",
+                namespaces: {
+                    users: [{ exclusive: true, regex: "@_prefix_.*:.+" }],
+                    rooms: [],
+                    aliases: [],
+                },
+            },
+        });
+        appservice.botIntent.ensureRegistered = () => {
+            return null;
+        };
+
+        appservice.expressAppInstance.get("/test", (_, res) => res.sendStatus(200));
+
+        await appservice.begin();
+
+        try {
+            const res = await requestPromise({
+                uri: `http://localhost:${port}/test`,
+                method: "GET",
+            });
+            expect(res).toEqual("OK");
+        } finally {
+            appservice.stop();
+        }
+    });
+
     // TODO: Populate once intent tests are stable
 
     it.skip('should not try decryption if crypto is not possible', async () => {