fix: address review findings on postgres backend
- PostgresMigrator: run JS migration parts on the transactional client (WrappedPgTx) instead of the pool, preserving migration atomicity - pool.ts: guard ROLLBACK so a failing rollback no longer masks the original error (also applied in the migrator loop) - pool.ts: share all/get/run between WrappedPool and WrappedPgTx via a common base class - sql.ts: document that toPositional precludes JSONB ?/?|/?& operators - PostgresInterface: implement deleteSession, required by BackendDbInterface since the logout-invalidation change on master Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit was merged in pull request #1.
This commit is contained in:
@@ -408,6 +408,14 @@ export class PostgresInterface implements BackendDbInterface {
|
|||||||
`);
|
`);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async deleteSession(token: `as_${string}`): Promise<void> {
|
||||||
|
this.#checkReadonly();
|
||||||
|
await this.#db.run(sql`
|
||||||
|
DELETE FROM "sessions"
|
||||||
|
WHERE "token" = ${{ text: token }}
|
||||||
|
`);
|
||||||
|
}
|
||||||
|
|
||||||
async #getApikeyByToken(
|
async #getApikeyByToken(
|
||||||
token: `at_${string}`,
|
token: `at_${string}`,
|
||||||
db: WrappedPgClient
|
db: WrappedPgClient
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import type {
|
|||||||
} from "../types/Migrator.js";
|
} from "../types/Migrator.js";
|
||||||
import { init, migrations } from "./migrations/index.js";
|
import { init, migrations } from "./migrations/index.js";
|
||||||
import { pgToDate } from "./cast.js";
|
import { pgToDate } from "./cast.js";
|
||||||
import { WrappedPool } from "./pool.js";
|
import { rollbackQuietly, WrappedPgTx, WrappedPool } from "./pool.js";
|
||||||
import { sql, toPositional } from "./sql.js";
|
import { sql, toPositional } from "./sql.js";
|
||||||
|
|
||||||
export class PostgresMigrator implements Migrator {
|
export class PostgresMigrator implements Migrator {
|
||||||
@@ -97,6 +97,7 @@ export class PostgresMigrator implements Migrator {
|
|||||||
for (const migration of toApply) {
|
for (const migration of toApply) {
|
||||||
console.log(`Applying migration ${migration.id} (${migration.name})`);
|
console.log(`Applying migration ${migration.id} (${migration.name})`);
|
||||||
const client = await this.#pool._pool.connect();
|
const client = await this.#pool._pool.connect();
|
||||||
|
const tx = new WrappedPgTx(client, false);
|
||||||
try {
|
try {
|
||||||
await client.query("BEGIN");
|
await client.query("BEGIN");
|
||||||
for (const part of migration.parts) {
|
for (const part of migration.parts) {
|
||||||
@@ -104,7 +105,7 @@ export class PostgresMigrator implements Migrator {
|
|||||||
if ("sql" in part) {
|
if ("sql" in part) {
|
||||||
await client.query(part.sql);
|
await client.query(part.sql);
|
||||||
} else {
|
} else {
|
||||||
await part.apply(this.#pool);
|
await part.apply(tx);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
const recordSql = sql`
|
const recordSql = sql`
|
||||||
@@ -114,7 +115,7 @@ export class PostgresMigrator implements Migrator {
|
|||||||
await client.query(toPositional(recordSql._sql), recordSql._vars);
|
await client.query(toPositional(recordSql._sql), recordSql._vars);
|
||||||
await client.query("COMMIT");
|
await client.query("COMMIT");
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
await client.query("ROLLBACK");
|
await rollbackQuietly(client);
|
||||||
throw e;
|
throw e;
|
||||||
} finally {
|
} finally {
|
||||||
client.release();
|
client.release();
|
||||||
|
|||||||
+42
-39
@@ -29,12 +29,22 @@ async function rethrow<R>(fn: () => Promise<R>): Promise<R> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
class WrappedPgTx implements WrappedPgClient {
|
// Rollback on a broken connection can itself throw; the original error is
|
||||||
#client: pg.PoolClient;
|
// the one worth surfacing.
|
||||||
|
export async function rollbackQuietly(
|
||||||
|
client: pg.PoolClient
|
||||||
|
): Promise<void> {
|
||||||
|
try {
|
||||||
|
await client.query("ROLLBACK");
|
||||||
|
} catch {}
|
||||||
|
}
|
||||||
|
|
||||||
|
abstract class WrappedPgBase implements WrappedPgClient {
|
||||||
|
#queryable: pg.Pool | pg.PoolClient;
|
||||||
#readonly: boolean;
|
#readonly: boolean;
|
||||||
|
|
||||||
constructor(client: pg.PoolClient, readonly_: boolean) {
|
constructor(queryable: pg.Pool | pg.PoolClient, readonly_: boolean) {
|
||||||
this.#client = client;
|
this.#queryable = queryable;
|
||||||
this.#readonly = readonly_;
|
this.#readonly = readonly_;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -42,10 +52,14 @@ class WrappedPgTx implements WrappedPgClient {
|
|||||||
return this.#readonly;
|
return this.#readonly;
|
||||||
}
|
}
|
||||||
|
|
||||||
async destroy(): Promise<void> {}
|
abstract destroy(): Promise<void>;
|
||||||
|
abstract multi<R>(fn: (tx: WrappedPgClient) => Promise<R>): Promise<R>;
|
||||||
|
|
||||||
async all<R>(stmt: SqlCode): Promise<R[]> {
|
async all<R>(stmt: SqlCode): Promise<R[]> {
|
||||||
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[];
|
return result.rows as R[];
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -56,72 +70,61 @@ class WrappedPgTx implements WrappedPgClient {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async run(stmt: SqlCode): Promise<{ changes: number }> {
|
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 };
|
return { changes: result.rowCount ?? 0 };
|
||||||
}
|
}
|
||||||
|
|
||||||
multi<R>(_fn: (tx: WrappedPgClient) => Promise<R>): Promise<R> {
|
|
||||||
throw new Error("Nested transactions not supported");
|
|
||||||
}
|
|
||||||
|
|
||||||
rethrow = rethrow;
|
rethrow = rethrow;
|
||||||
}
|
}
|
||||||
|
|
||||||
export class WrappedPool implements WrappedPgClient {
|
export class WrappedPgTx extends WrappedPgBase {
|
||||||
|
constructor(client: pg.PoolClient, readonly_: boolean) {
|
||||||
|
super(client, readonly_);
|
||||||
|
}
|
||||||
|
|
||||||
|
async destroy(): Promise<void> {}
|
||||||
|
|
||||||
|
multi<R>(_fn: (tx: WrappedPgClient) => Promise<R>): Promise<R> {
|
||||||
|
throw new Error("Nested transactions not supported");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export class WrappedPool extends WrappedPgBase {
|
||||||
#pool: pg.Pool;
|
#pool: pg.Pool;
|
||||||
#readonly: boolean;
|
|
||||||
|
|
||||||
constructor(connectionStringOrPool: string | pg.Pool, readonly_ = false) {
|
constructor(connectionStringOrPool: string | pg.Pool, readonly_ = false) {
|
||||||
this.#pool =
|
const pool =
|
||||||
typeof connectionStringOrPool === "string"
|
typeof connectionStringOrPool === "string"
|
||||||
? new pg.Pool({ connectionString: connectionStringOrPool })
|
? new pg.Pool({ connectionString: connectionStringOrPool })
|
||||||
: connectionStringOrPool;
|
: connectionStringOrPool;
|
||||||
this.#readonly = readonly_;
|
super(pool, readonly_);
|
||||||
|
this.#pool = pool;
|
||||||
}
|
}
|
||||||
|
|
||||||
get _pool(): pg.Pool {
|
get _pool(): pg.Pool {
|
||||||
return this.#pool;
|
return this.#pool;
|
||||||
}
|
}
|
||||||
|
|
||||||
get readonly(): boolean {
|
|
||||||
return this.#readonly;
|
|
||||||
}
|
|
||||||
|
|
||||||
async destroy(): Promise<void> {
|
async destroy(): Promise<void> {
|
||||||
await this.#pool.end();
|
await this.#pool.end();
|
||||||
}
|
}
|
||||||
|
|
||||||
async all<R>(stmt: SqlCode): Promise<R[]> {
|
|
||||||
const result = await this.#pool.query(toPositional(stmt._sql), stmt._vars);
|
|
||||||
return result.rows as R[];
|
|
||||||
}
|
|
||||||
|
|
||||||
async get<R>(stmt: SqlCode): Promise<R | null> {
|
|
||||||
const rows = await this.all<R>(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<R>(fn: (tx: WrappedPgClient) => Promise<R>): Promise<R> {
|
async multi<R>(fn: (tx: WrappedPgClient) => Promise<R>): Promise<R> {
|
||||||
const client = await this.#pool.connect();
|
const client = await this.#pool.connect();
|
||||||
const tx = new WrappedPgTx(client, this.#readonly);
|
const tx = new WrappedPgTx(client, this.readonly);
|
||||||
try {
|
try {
|
||||||
await client.query("BEGIN");
|
await client.query("BEGIN");
|
||||||
const result = await fn(tx);
|
const result = await fn(tx);
|
||||||
await client.query("COMMIT");
|
await client.query("COMMIT");
|
||||||
return result;
|
return result;
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
await client.query("ROLLBACK");
|
await rollbackQuietly(client);
|
||||||
throw e;
|
throw e;
|
||||||
} finally {
|
} finally {
|
||||||
client.release();
|
client.release();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
rethrow = rethrow;
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -73,6 +73,9 @@ export function calcUpdates<T extends object>(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 {
|
export function toPositional(sql: string): string {
|
||||||
let i = 0;
|
let i = 0;
|
||||||
return sql.replace(/\?/g, () => `$${++i}`);
|
return sql.replace(/\?/g, () => `$${++i}`);
|
||||||
|
|||||||
Reference in New Issue
Block a user