Use matrix-appservice-bridge PostgresStore implementation

src/datastore/postgres/schema/index.ts

@@ -0,0 +1,20 @@
+import { SchemaUpdateFunction } from "matrix-appservice-bridge";
+import v1 from './v1';
+import v2 from './v2';
+import v3 from './v3';
+import v4 from './v4';
+import v5 from './v5';
+import v6 from './v6';
+import v7 from './v7';
+import v8 from './v8';
+
+export default [
+    v1,
+    v2,
+    v3,
+    v4,
+    v5,
+    v6,
+    v7,
+    v8,
+] as SchemaUpdateFunction[];

src/datastore/postgres/schema/v1.ts

@@ -1,8 +1,8 @@
-import { PoolClient } from "pg";
+import { SchemaUpdateFunction } from 'matrix-appservice-bridge';
 
-export async function runSchema(connection: PoolClient) {
+const updateFn: SchemaUpdateFunction = async (sql) => {
     // Create schema
-    await connection.query(`
+    await sql`
     CREATE TABLE schema (
         version	INTEGER UNIQUE NOT NULL
     );
@@ -59,5 +59,8 @@ export async function runSchema(connection: PoolClient) {
         count INTEGER
     );
 
-    INSERT INTO ipv6_counter VALUES (0);`);
-}
+    INSERT INTO ipv6_counter VALUES (0);`;
+
+};
+
+export default updateFn;

src/datastore/postgres/schema/v2.ts

@@ -3,9 +3,18 @@ import { PoolClient } from "pg";
 export async function runSchema(connection: PoolClient) {
     // Create schema
     await connection.query(`
+    `);
+}
+import { SchemaUpdateFunction } from 'matrix-appservice-bridge';
+
+const updateFn: SchemaUpdateFunction = async (sql) => {
+    // Create schema
+    await sql`
     CREATE TABLE last_seen (
         user_id	TEXT UNIQUE NOT NULL,
         ts BIGINT NOT NULL
-    );
-    `);
-}
+    );`;
+
+};
+
+export default updateFn;

src/datastore/postgres/schema/v3.ts

@@ -1,11 +1,11 @@
-import { PoolClient } from "pg";
+import { SchemaUpdateFunction } from 'matrix-appservice-bridge';
 
-export async function runSchema(connection: PoolClient) {
-    // Create schema
-    await connection.query(`
+const updateFn: SchemaUpdateFunction = async (sql) => {
+    await sql`
     CREATE TABLE room_visibility (
         room_id	TEXT UNIQUE NOT NULL,
         visibility BOOLEAN NOT NULL
-    );
-    `);
-}
+    );`
+};
+
+export default updateFn;

src/datastore/postgres/schema/v4.ts

@@ -1,11 +1,11 @@
-import { PoolClient } from "pg";
+import { SchemaUpdateFunction } from 'matrix-appservice-bridge';
 
-export async function runSchema(connection: PoolClient) {
-    // Create schema
-    await connection.query(`
+const updateFn: SchemaUpdateFunction = async (sql) => {
+    await sql`
     CREATE TABLE deactivated_users (
         user_id	TEXT UNIQUE NOT NULL,
         ts BIGINT NOT NULL
-    );
-    `);
-}
+    );`
+};
+
+export default updateFn;

src/datastore/postgres/schema/v5.ts

@@ -1,7 +1,7 @@
-import { PoolClient } from "pg";
+import { SchemaUpdateFunction } from 'matrix-appservice-bridge';
 
-export async function runSchema(connection: PoolClient) {
-    await connection.query(`
-    CREATE INDEX client_config_domain_username_idx ON client_config (domain, (config->>'username'));
-    `);
-}
+const updateFn: SchemaUpdateFunction = async (sql) => {
+    await sql`CREATE INDEX client_config_domain_username_idx ON client_config (domain, (config->>'username'));`
+};
+
+export default updateFn;

src/datastore/postgres/schema/v6.ts

@@ -1,7 +1,7 @@
-import { PoolClient } from "pg";
+import { SchemaUpdateFunction } from 'matrix-appservice-bridge';
 
-export async function runSchema(connection: PoolClient) {
-    await connection.query(`
-    DROP INDEX client_config_domain_username_idx;
-    `);
-}
+const updateFn: SchemaUpdateFunction = async (sql) => {
+    await sql`DROP INDEX client_config_domain_username_idx;`
+};
+
+export default updateFn;

src/datastore/postgres/schema/v7.ts

@@ -1,10 +1,11 @@
-import { PoolClient } from "pg";
+import { SchemaUpdateFunction } from 'matrix-appservice-bridge';
 
-export async function runSchema(connection: PoolClient) {
-    await connection.query(`
+const updateFn: SchemaUpdateFunction = async (sql) => {
+    await sql`
     CREATE TABLE user_activity (
         user_id TEXT UNIQUE,
         data JSON
-    );
-    `);
-}
+    );`
+};
+
+export default updateFn;

src/datastore/postgres/schema/v8.ts

@@ -1,59 +1,47 @@
-import { PoolClient } from "pg";
-import { Logging } from "matrix-appservice-bridge";
+import { Logger, SchemaUpdateFunction } from "matrix-appservice-bridge";
 
-const log = Logging.get('postgres/schema/v8');
+const log = new Logger('postgres/schema/v8');
 
-function domainSetToValues(domains: string[], count: number): [string, Array<string|number|null>] {
-    let res = "";
-    const values: Array<string|number|null> = [];
-    // VALUES ...
-    for (let index = 0; index < domains.length; index++) {
-        const i = index * 3;
-        if (res) {
-            res += ", ";
-        }
-        res += `($${i+1}, $${i+2}, $${i+3})`;
-        values.push(count, "*", domains[index]);
-    }
-    return [res, values];
-}
-
-export async function runSchema(connection: PoolClient) {
-
-    await connection.query(`
+const updateFn: SchemaUpdateFunction = async (sql) => {
+    await sql`
         ALTER TABLE ipv6_counter
         ADD COLUMN homeserver TEXT,
         ADD COLUMN server TEXT;
         ALTER TABLE ipv6_counter
         ADD CONSTRAINT cons_ipv6_counter_unique UNIQUE(homeserver, server);
-    `);
-
+    `;
 
     // Migrate data.
-    const existingCounterRes = await connection.query<{count: string}>("SELECT count FROM ipv6_counter;");
-    const existingCounter = existingCounterRes && parseInt(existingCounterRes.rows[0].count, 10);
+    const existingCounterRes = await sql<{count: string}[]>`SELECT count FROM ipv6_counter;`;
+    const existingCounter = existingCounterRes && parseInt(existingCounterRes?.[0].count, 10);
 
     // If we have a counter value
     if (existingCounter) {
-        const serverConfigsRes = await connection.query<{domain: string}>(
-            "SELECT DISTINCT domain FROM client_config WHERE config->>'ipv6' IS NOT NULL;"
-        );
-        if (serverConfigsRes.rowCount === 0) {
+        const serverConfigsRes = await sql<{domain: string}[]>`
+            SELECT DISTINCT domain FROM client_config WHERE config->>'ipv6' IS NOT NULL;
+        `;
+        if (serverConfigsRes?.length === 0) {
             // No servers to migrate?
             throw Error("No client_configs found with ipv6 addresses, but counter was found");
         }
-        else if (serverConfigsRes.rowCount > 1) {
+        else if (serverConfigsRes?.length > 1) {
             log.warn("More than one IPv6 server configured, starting both ipv6 counters from the same value.");
         }
         // Because we cannot determine which IRC network(s) are using the existing counter
         // (owing to a bug where we treated the counter as global across all networks), this assumes
         // that both networks start from the same counter value.
-        const [statement, values] = domainSetToValues(serverConfigsRes.rows.map(d => d.domain), existingCounter);
-        await connection.query(`INSERT INTO ipv6_counter (count, homeserver, server) VALUES ${statement}`, values);
+        const domains = serverConfigsRes.flatMap(r => ({
+            count: existingCounter,
+            domain: r.domain,
+            homeserver: '*'
+        }));
+        await sql`INSERT INTO ipv6_counter ${sql(domains, 'count', 'domain', 'homeserver')})`;
     }
 
-    await connection.query(`
+    await sql`
         DELETE FROM ipv6_counter WHERE server IS NULL;
         ALTER TABLE ipv6_counter ALTER COLUMN server SET NOT NULL;
-    `);
-}
+    `;
+};
+
+export default updateFn;