Add PostgresStore abstraction

.github/workflows/test.yml

@@ -30,4 +30,33 @@ jobs:
         with:
           node-version: "${{ matrix.node-version }}"
       - run: yarn --frozen-lockfile
-      - run: yarn build && yarn test
+      - run: yarn build && yarn test
+  test-postgres:
+    runs-on: ubuntu-20.04
+    container: node:18
+    services:
+      postgres:
+        image: postgres:latest
+        env:
+          POSTGRES_DB: postgres
+          POSTGRES_PASSWORD: postgres_password
+          POSTGRES_PORT: 5432
+          POSTGRES_USER: postgres_user
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    steps:
+      - uses: actions/checkout@v2
+      - name: Use Node.js
+        uses: actions/setup-node@v1
+        with:
+          node-version: 18
+      - run: yarn --frozen-lockfile
+      - run: yarn test
+        env:
+          BRIDGE_TEST_PGDB: "bridge_integtest"
+          BRIDGE_TEST_PGURL: "postgresql://postgres_user:postgres_password@postgres"

changelog.d/442.bugfix

@@ -0,0 +1 @@
+Add implementation of a PostgreSQL datastore for use by other bridges.

package.json

@@ -46,7 +46,8 @@
     "nopt": "^5.0.0",
     "p-queue": "^6.6.2",
     "pkginfo": "^0.4.1",
-    "prom-client": "^14.0.0",
+    "postgres": "^3.3.1",
+    "prom-client": "^14.1.0",
     "uuid": "^8.3.2",
     "winston": "^3.3.3",
     "winston-daily-rotate-file": "^4.5.1"

spec/.eslintrc.json

@@ -1,4 +1,7 @@
 {
-  "strict": "off",
-  "no-var": "off"
-}
+  "rules": {
+    "strict": "off",
+    "no-var": "off",
+    "eol-last": "off"
+  }
+}

spec/helpers/postgres-helper.ts

@@ -0,0 +1,21 @@
+import postgres, { Sql } from 'postgres';
+
+let pgClient: Sql;
+
+export function isPostgresTestingEnabled() {
+    return !!process.env.BRIDGE_TEST_PGURL;
+}
+
+export function initPostgres() {
+    // Setup postgres for the whole process.
+    pgClient = postgres(`${process.env.BRIDGE_TEST_PGURL}/postgres`);
+    process.on("beforeExit", async () => {
+        pgClient.end();
+    })
+}
+
+export async function getPgDatabase() {
+    const pgDb = `${process.env.BRIDGE_TEST_PGDB}_${process.hrtime().join("_")}`;
+    await pgClient`CREATE DATABASE ${pgClient(pgDb)}`;
+    return `${process.env.BRIDGE_TEST_PGURL}/${pgDb}`;
+}

spec/integ/postgres.spec.ts

@@ -0,0 +1,41 @@
+import { PostgresStore } from "../../src";
+import { getPgDatabase, initPostgres, isPostgresTestingEnabled } from "../helpers/postgres-helper";
+
+
+// Only run the tests if we've enabled postgres testing.
+const descr = isPostgresTestingEnabled() ? describe : xdescribe;
+
+class TestPostgresStore extends PostgresStore { }
+
+descr('PostgresStore', () => {
+    let store: TestPostgresStore|undefined;
+    beforeAll(() => {
+        initPostgres();
+    })
+
+    it('can construct a simple database from schema', async () => {
+        store = new TestPostgresStore([], {
+            url: await getPgDatabase(),
+        });
+        await store.ensureSchema();
+    });
+
+    it('can run schema upgrades', async () => {
+        store = new TestPostgresStore([
+            sql => sql.begin(s => [
+                s`CREATE TABLE v1_users (mxid TEXT UNIQUE NOT NULL);`,
+            ]).then(),
+            sql => sql.begin(s => [
+                s`CREATE TABLE v2_rooms (mxid TEXT UNIQUE NOT NULL);`,
+            ]).then(),
+        ], {
+            autocreateSchemaTable: true,
+            url: await getPgDatabase(),
+        });
+        await store.ensureSchema();
+    });
+
+    afterEach(async () => {
+        await store?.destroy();
+    })
+});

src/components/stores/postgres-store.ts

@@ -0,0 +1,153 @@
+import postgres, { PostgresError, PostgresType } from 'postgres';
+import { Logger } from "../..";
+
+const log = new Logger("PostgresStore");
+
+export async function v0Schema(sql: postgres.Sql) {
+    await sql.begin(s => [
+        s`CREATE TABLE schema (version	INTEGER UNIQUE NOT NULL);`,
+        s`INSERT INTO schema VALUES (0);`
+    ]);
+}
+
+export interface PostgresStoreOpts extends postgres.Options<Record<string, PostgresType<unknown>>> {
+    /**
+     * URL to reach the database on.
+     */
+    url?: string;
+    /**
+     * Should the schema table be automatically created (the v0 schema effectively).
+     * Defaults to `true`.
+     */
+    autocreateSchemaTable?: boolean;
+}
+
+export type SchemaUpdateFunction = (sql: postgres.Sql) => Promise<void>|void;
+
+/**
+ * PostgreSQL datastore abstraction which can be inherited by a specialised bridge class.
+ *
+ * @example
+ * class MyBridgeStore extends PostgresStore {
+ *   constructor(myurl) {
+ *     super([schemav1, schemav2, schemav3], { url: myurl });
+ *   }
+ *
+ *   async getData() {
+ *     return this.sql`SELECT * FROM mytable`
+ *   }
+ * }
+ *
+ * // Which can then be used by doing
+ * const store = new MyBridgeStore("postgresql://postgres_user:postgres_password@postgres");
+ * store.ensureSchema();
+ * const data = await store.getData();
+ */
+export abstract class PostgresStore {
+    private hasEnded = false;
+    public readonly sql: postgres.Sql;
+
+    public get latestSchema() {
+        return this.schemas.length;
+    }
+
+    /**
+     * Construct a new store.
+     * @param schemas The set of schema functions to apply to a database. The ordering of this array determines the
+     *                schema number.
+     * @param opts Options to supply to the PostgreSQL client, such as `url`.
+     */
+    constructor(private readonly schemas: SchemaUpdateFunction[], private readonly opts: PostgresStoreOpts) {
+        opts.autocreateSchemaTable = opts.autocreateSchemaTable ?? true;
+        this.sql = opts.url ? postgres(opts.url, opts) : postgres(opts);
+        process.on("beforeExit", () => {
+            // Ensure we clean up on exit
+            this.destroy().catch(ex => {
+                log.warn('Failed to cleanly exit', ex);
+            });
+        })
+    }
+
+    /**
+     * Ensure the database schema is up to date. If you supplied
+     * `autocreateSchemaTable` to `opts` in the constructor, a fresh database
+     * will have a `schema` table created for it.
+     *
+     * @throws If a schema could not be applied cleanly.
+     */
+    public async ensureSchema(): Promise<void> {
+        log.info("Starting database engine");
+        let currentVersion = await this.getSchemaVersion();
+
+        if (currentVersion === -1) {
+            if (this.opts.autocreateSchemaTable) {
+                log.info(`Applying v0 schema (schema table)`);
+                await v0Schema(this.sql);
+                currentVersion = 0;
+            }
+        }
+        else {
+            // We aren't autocreating the schema table, so assume schema 0.
+            currentVersion = 0;
+        }
+
+        // Zero-indexed, so schema 1 would be in slot 0.
+        while (this.schemas[currentVersion]) {
+            log.info(`Updating schema to v${currentVersion + 1}`);
+            const runSchema = this.schemas[currentVersion];
+            try {
+                await runSchema(this.sql);
+                currentVersion++;
+                await this.updateSchemaVersion(currentVersion);
+            }
+            catch (ex) {
+                log.warn(`Failed to run schema v${currentVersion + 1}:`, ex);
+                throw Error("Failed to update database schema");
+            }
+        }
+        log.info(`Database schema is at version v${currentVersion}`);
+    }
+
+    /**
+     * Clean away any resources used by the database. This is automatically
+     * called before the process exits.
+     */
+    public async destroy(): Promise<void> {
+        log.info("Destroy called");
+        if (this.hasEnded) {
+            // No-op if end has already been called.
+            return;
+        }
+        this.hasEnded = true;
+        await this.sql.end();
+        log.info("PostgresSQL connection ended");
+    }
+
+    /**
+     * Update the current schema version.
+     * @param version
+     */
+    protected async updateSchemaVersion(version: number): Promise<void> {
+        log.debug(`updateSchemaVersion: ${version}`);
+        await this.sql`UPDATE schema SET version = ${version};`;
+    }
+
+    /**
+     * Get the current schema version.
+     * @returns The current schema version, or `-1` if no schema table is found.
+     */
+    protected async getSchemaVersion(): Promise<number> {
+        try {
+            const result = await this.sql<{version: number}[]>`SELECT version FROM SCHEMA;`;
+            return result[0].version;
+        }
+        catch (ex) {
+            if (ex instanceof PostgresError && ex.code === "42P01") { // undefined_table
+                log.warn("Schema table could not be found");
+                return -1;
+            }
+            log.error("Failed to get schema version", ex);
+        }
+        throw Error("Couldn't fetch schema version");
+    }
+}

src/index.ts

@@ -39,6 +39,7 @@ export * from "./components/user-bridge-store";
 export * from "./components/user-activity-store";
 export * from "./components/room-bridge-store";
 export * from "./components/event-bridge-store";
+export * from "./components/stores/postgres-store";
 
 // Models
 export * from "./models/rooms/matrix";

yarn.lock

@@ -2619,6 +2619,11 @@ postcss@^8.3.11:
     picocolors "^1.0.0"
     source-map-js "^1.0.2"
 
+postgres@^3.3.1:
+  version "3.3.1"
+  resolved "https://registry.yarnpkg.com/postgres/-/postgres-3.3.1.tgz#1d9b5e8f01ee325df13b6db14f38ae2b8f6fe912"
+  integrity sha512-ak/xXToZYwRvQlZIUtLgPUIggz62eIIbPTgxl/Yl4oTu0TgNOd1CrzTCifsvZ89jBwLvnX6+Ky5frp5HzIBoaw==
+
 prelude-ls@^1.2.1:
   version "1.2.1"
   resolved "https://registry.yarnpkg.com/prelude-ls/-/prelude-ls-1.2.1.tgz#debc6489d7a6e6b0e7611888cec880337d316396"
@@ -2631,10 +2636,10 @@ process-on-spawn@^1.0.0:
   dependencies:
     fromentries "^1.2.0"
 
-prom-client@^14.0.0:
-  version "14.0.1"
-  resolved "https://registry.yarnpkg.com/prom-client/-/prom-client-14.0.1.tgz#bdd9583e02ec95429677c0e013712d42ef1f86a8"
-  integrity sha512-HxTArb6fkOntQHoRGvv4qd/BkorjliiuO2uSWC2KC17MUTKYttWdDoXX/vxOhQdkoECEM9BBH0pj2l8G8kev6w==
+prom-client@^14.1.0:
+  version "14.1.0"
+  resolved "https://registry.yarnpkg.com/prom-client/-/prom-client-14.1.0.tgz#049609859483d900844924df740722c76ed1fdbb"
+  integrity sha512-iFWCchQmi4170omLpFXbzz62SQTmPhtBL35v0qGEVRHKcqIeiexaoYeP0vfZTujxEq3tA87iqOdRbC9svS1B9A==
   dependencies:
     tdigest "^0.1.1"