diff --git a/src/db/postgres/PostgresInterface.ts b/src/db/postgres/PostgresInterface.ts index 27776ff..0f123c4 100644 --- a/src/db/postgres/PostgresInterface.ts +++ b/src/db/postgres/PostgresInterface.ts @@ -408,6 +408,14 @@ export class PostgresInterface implements BackendDbInterface { `); } + async deleteSession(token: `as_${string}`): Promise { + this.#checkReadonly(); + await this.#db.run(sql` + DELETE FROM "sessions" + WHERE "token" = ${{ text: token }} + `); + } + async #getApikeyByToken( token: `at_${string}`, db: WrappedPgClient diff --git a/src/db/postgres/PostgresMigrator.ts b/src/db/postgres/PostgresMigrator.ts index 01c85cd..a4da392 100644 --- a/src/db/postgres/PostgresMigrator.ts +++ b/src/db/postgres/PostgresMigrator.ts @@ -5,7 +5,7 @@ import type { } from "../types/Migrator.js"; import { init, migrations } from "./migrations/index.js"; import { pgToDate } from "./cast.js"; -import { WrappedPool } from "./pool.js"; +import { rollbackQuietly, WrappedPgTx, WrappedPool } from "./pool.js"; import { sql, toPositional } from "./sql.js"; export class PostgresMigrator implements Migrator { @@ -97,6 +97,7 @@ export class PostgresMigrator implements Migrator { for (const migration of toApply) { console.log(`Applying migration ${migration.id} (${migration.name})`); const client = await this.#pool._pool.connect(); + const tx = new WrappedPgTx(client, false); try { await client.query("BEGIN"); for (const part of migration.parts) { @@ -104,7 +105,7 @@ export class PostgresMigrator implements Migrator { if ("sql" in part) { await client.query(part.sql); } else { - await part.apply(this.#pool); + await part.apply(tx); } } const recordSql = sql` @@ -114,7 +115,7 @@ export class PostgresMigrator implements Migrator { await client.query(toPositional(recordSql._sql), recordSql._vars); await client.query("COMMIT"); } catch (e) { - await client.query("ROLLBACK"); + await rollbackQuietly(client); throw e; } finally { client.release(); diff --git a/src/db/postgres/pool.ts b/src/db/postgres/pool.ts index 397519a..85fe5e2 100644 --- a/src/db/postgres/pool.ts +++ b/src/db/postgres/pool.ts @@ -29,12 +29,22 @@ async function rethrow(fn: () => Promise): Promise { } } -class WrappedPgTx implements WrappedPgClient { - #client: pg.PoolClient; +// Rollback on a broken connection can itself throw; the original error is +// the one worth surfacing. +export async function rollbackQuietly( + client: pg.PoolClient +): Promise { + try { + await client.query("ROLLBACK"); + } catch {} +} + +abstract class WrappedPgBase implements WrappedPgClient { + #queryable: pg.Pool | pg.PoolClient; #readonly: boolean; - constructor(client: pg.PoolClient, readonly_: boolean) { - this.#client = client; + constructor(queryable: pg.Pool | pg.PoolClient, readonly_: boolean) { + this.#queryable = queryable; this.#readonly = readonly_; } @@ -42,10 +52,14 @@ class WrappedPgTx implements WrappedPgClient { return this.#readonly; } - async destroy(): Promise {} + abstract destroy(): Promise; + abstract multi(fn: (tx: WrappedPgClient) => Promise): Promise; async all(stmt: SqlCode): Promise { - const result = await this.#client.query(toPositional(stmt._sql), stmt._vars); + const result = await this.#queryable.query( + toPositional(stmt._sql), + stmt._vars + ); return result.rows as R[]; } @@ -56,72 +70,61 @@ class WrappedPgTx implements WrappedPgClient { } async run(stmt: SqlCode): Promise<{ changes: number }> { - const result = await this.#client.query(toPositional(stmt._sql), stmt._vars); + const result = await this.#queryable.query( + toPositional(stmt._sql), + stmt._vars + ); return { changes: result.rowCount ?? 0 }; } - multi(_fn: (tx: WrappedPgClient) => Promise): Promise { - throw new Error("Nested transactions not supported"); - } - rethrow = rethrow; } -export class WrappedPool implements WrappedPgClient { +export class WrappedPgTx extends WrappedPgBase { + constructor(client: pg.PoolClient, readonly_: boolean) { + super(client, readonly_); + } + + async destroy(): Promise {} + + multi(_fn: (tx: WrappedPgClient) => Promise): Promise { + throw new Error("Nested transactions not supported"); + } +} + +export class WrappedPool extends WrappedPgBase { #pool: pg.Pool; - #readonly: boolean; constructor(connectionStringOrPool: string | pg.Pool, readonly_ = false) { - this.#pool = + const pool = typeof connectionStringOrPool === "string" ? new pg.Pool({ connectionString: connectionStringOrPool }) : connectionStringOrPool; - this.#readonly = readonly_; + super(pool, readonly_); + this.#pool = pool; } get _pool(): pg.Pool { return this.#pool; } - get readonly(): boolean { - return this.#readonly; - } - async destroy(): Promise { await this.#pool.end(); } - async all(stmt: SqlCode): Promise { - const result = await this.#pool.query(toPositional(stmt._sql), stmt._vars); - return result.rows as R[]; - } - - async get(stmt: SqlCode): Promise { - const rows = await this.all(stmt); - if (rows.length > 1) throw new Error("Multiple results"); - return rows[0] ?? null; - } - - async run(stmt: SqlCode): Promise<{ changes: number }> { - const result = await this.#pool.query(toPositional(stmt._sql), stmt._vars); - return { changes: result.rowCount ?? 0 }; - } - async multi(fn: (tx: WrappedPgClient) => Promise): Promise { const client = await this.#pool.connect(); - const tx = new WrappedPgTx(client, this.#readonly); + const tx = new WrappedPgTx(client, this.readonly); try { await client.query("BEGIN"); const result = await fn(tx); await client.query("COMMIT"); return result; } catch (e) { - await client.query("ROLLBACK"); + await rollbackQuietly(client); throw e; } finally { client.release(); } } - - rethrow = rethrow; } diff --git a/src/db/postgres/sql.ts b/src/db/postgres/sql.ts index 65a9e6a..653812a 100644 --- a/src/db/postgres/sql.ts +++ b/src/db/postgres/sql.ts @@ -73,6 +73,9 @@ export function calcUpdates(updater: { }; } +// Rewrites every literal `?` into a numbered placeholder, so queries must +// not contain Postgres's JSONB `?` / `?|` / `?&` operators (use +// `jsonb_exists`, `jsonb_exists_any`, `jsonb_exists_all` instead). export function toPositional(sql: string): string { let i = 0; return sql.replace(/\?/g, () => `$${++i}`);