You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
127 lines
4.2 KiB
TypeScript
127 lines
4.2 KiB
TypeScript
import { AsyncLocalStorage } from 'node:async_hooks'
|
|
import pg from 'pg'
|
|
import type { DB } from './db'
|
|
import { PG_MIGRATIONS } from './migrations-pg'
|
|
|
|
/**
|
|
* Postgres engine behind the async DB interface (D19). Selected by DATABASE_URL.
|
|
* Repos speak `?` placeholders and portable SQL; this adapter owns the dialect:
|
|
* `?` → `$n`, transactions pinned to one pooled client via AsyncLocalStorage
|
|
* (nested calls become savepoints), and bigint/numeric results parsed to JS
|
|
* numbers so integer-paise math is identical on both engines.
|
|
*/
|
|
|
|
// int8 (bigint paise, COUNT(*)) and numeric (SUM over bigint) arrive as strings by
|
|
// default; paise totals sit far below 2^53, so Number is lossless here.
|
|
pg.types.setTypeParser(20, Number)
|
|
pg.types.setTypeParser(1700, Number)
|
|
|
|
/** Rewrite `?` placeholders to `$1..$n`, leaving single-quoted literals untouched. */
|
|
export function toDollarParams(sql: string): string {
|
|
let out = ''
|
|
let n = 0
|
|
let inQuote = false
|
|
for (let i = 0; i < sql.length; i++) {
|
|
const ch = sql[i]!
|
|
if (inQuote) {
|
|
out += ch
|
|
if (ch === "'") {
|
|
if (sql[i + 1] === "'") { out += "'"; i++ } else inQuote = false
|
|
}
|
|
continue
|
|
}
|
|
if (ch === "'") { inQuote = true; out += ch; continue }
|
|
out += ch === '?' ? `$${++n}` : ch
|
|
}
|
|
return out
|
|
}
|
|
|
|
interface TxnStore { client: pg.PoolClient; depth: number }
|
|
|
|
export class PgDb implements DB {
|
|
readonly engine = 'postgres' as const
|
|
private als = new AsyncLocalStorage<TxnStore>()
|
|
constructor(readonly pool: pg.Pool) {}
|
|
|
|
/** Queries inside a transaction ride its pinned client; otherwise the pool. */
|
|
private q(): { query: (sql: string, args?: unknown[]) => Promise<pg.QueryResult> } {
|
|
return this.als.getStore()?.client ?? this.pool
|
|
}
|
|
|
|
async all<T>(sql: string, ...args: unknown[]): Promise<T[]> {
|
|
return (await this.q().query(toDollarParams(sql), args)).rows as T[]
|
|
}
|
|
|
|
async get<T>(sql: string, ...args: unknown[]): Promise<T | undefined> {
|
|
return ((await this.q().query(toDollarParams(sql), args)).rows[0] ?? undefined) as T | undefined
|
|
}
|
|
|
|
async run(sql: string, ...args: unknown[]): Promise<{ changes: number }> {
|
|
const r = await this.q().query(toDollarParams(sql), args)
|
|
return { changes: r.rowCount ?? 0 }
|
|
}
|
|
|
|
async exec(sql: string): Promise<void> {
|
|
await this.q().query(sql)
|
|
}
|
|
|
|
async transaction<T>(fn: () => Promise<T> | T): Promise<T> {
|
|
const store = this.als.getStore()
|
|
if (store !== undefined) {
|
|
// Nested: savepoint on the already-pinned client.
|
|
const sp = `sp_d${++store.depth}`
|
|
await store.client.query(`SAVEPOINT ${sp}`)
|
|
try {
|
|
const out = await fn()
|
|
await store.client.query(`RELEASE SAVEPOINT ${sp}`)
|
|
return out
|
|
} catch (err) {
|
|
await store.client.query(`ROLLBACK TO SAVEPOINT ${sp}`)
|
|
await store.client.query(`RELEASE SAVEPOINT ${sp}`)
|
|
throw err
|
|
} finally {
|
|
store.depth--
|
|
}
|
|
}
|
|
const client = await this.pool.connect()
|
|
try {
|
|
await client.query('BEGIN')
|
|
const out = await this.als.run({ client, depth: 0 }, () => Promise.resolve(fn()))
|
|
await client.query('COMMIT')
|
|
return out
|
|
} catch (err) {
|
|
await client.query('ROLLBACK')
|
|
throw err
|
|
} finally {
|
|
client.release()
|
|
}
|
|
}
|
|
|
|
async close(): Promise<void> {
|
|
await this.pool.end()
|
|
}
|
|
}
|
|
|
|
/** Apply pending numbered migrations, each atomically, recorded in schema_migrations. */
|
|
export async function runPgMigrations(db: PgDb): Promise<string[]> {
|
|
await db.exec(`CREATE TABLE IF NOT EXISTS schema_migrations (id text PRIMARY KEY, applied_at text NOT NULL)`)
|
|
const applied: string[] = []
|
|
for (const m of PG_MIGRATIONS) {
|
|
await db.transaction(async () => {
|
|
const done = await db.get(`SELECT id FROM schema_migrations WHERE id=?`, m.id)
|
|
if (done !== undefined) return
|
|
await db.exec(m.sql)
|
|
await db.run(`INSERT INTO schema_migrations (id, applied_at) VALUES (?, ?)`, m.id, new Date().toISOString())
|
|
applied.push(m.id)
|
|
})
|
|
}
|
|
return applied
|
|
}
|
|
|
|
/** Open the Postgres engine: pool + migrations. Caller runs seedIfEmpty after. */
|
|
export async function openPgDb(url: string): Promise<DB> {
|
|
const db = new PgDb(new pg.Pool({ connectionString: url, max: 10 }))
|
|
await runPgMigrations(db)
|
|
return db
|
|
}
|