import { Pool } from 'pg' import type { PoolClient, QueryResult, QueryResultRow } from 'pg' import { runtimeConfig } from '../config/runtime.js' import { logWarn } from '../utils/logger.js' import { assertSqlParamArity } from './sql-arity.js' let poolInstance: Pool | null = null export function getDb(): Pool { if (!poolInstance) { const pool = new Pool({ connectionString: String(runtimeConfig.database?.url || '').trim(), ssl: runtimeConfig.database?.ssl ? { rejectUnauthorized: false } : false, max: Number(runtimeConfig.database?.maxConnections || 10), idleTimeoutMillis: Number(runtimeConfig.database?.idleTimeoutMs || 30_000), connectionTimeoutMillis: Number(runtimeConfig.database?.connectionTimeoutMs || 5_000), statement_timeout: Number(runtimeConfig.database?.statementTimeoutMs || 15_000), }) pool.on('error', (error) => { void logWarn('[db/pool]', '数据库连接池出现空闲连接错误', { error: error instanceof Error ? error.message : String(error), }) }) // 参数个数自检:参数比占位符多时 PostgreSQL 只会回一句难定位的 bind 报错, // 这里在发往数据库之前就把出错的 SQL 片段带出来(见 sql-arity.ts)。 pool.on('connect', (client) => guardSqlParamArity(client)) poolInstance = pool } return poolInstance } export async function query(text: string, params?: unknown[]): Promise> export async function query( text: string, params?: unknown[], ): Promise> export async function query( text: string, params: unknown[] = [], ): Promise> { const pool = getDb() const startedAt = process.hrtime.bigint() try { const result = await pool.query(text, params) logSlowQuery(text, startedAt, result.rowCount ?? 0) return result } catch (error) { logSlowQuery(text, startedAt, 0, error) throw error } } function logSlowQuery(text: string, startedAt: bigint, rowCount: number, error?: unknown) { const thresholdMs = Number(runtimeConfig.database?.slowQueryThresholdMs ?? 500) if (thresholdMs <= 0) return const durationMs = Number(process.hrtime.bigint() - startedAt) / 1_000_000 if (durationMs < thresholdMs && !error) return void logWarn('[db/query]', error ? '数据库 SQL 执行失败' : '数据库慢查询', { durationMs: Math.round(durationMs * 100) / 100, thresholdMs, rowCount, sql: fingerprintSql(text), error: error instanceof Error ? error.message : undefined, }) } /** 只保留可聚合的 SQL 形状,参数值始终不写入日志。 */ function fingerprintSql(text: string) { return String(text || '') .replace(/--[^\r\n]*/g, ' ') .replace(/\/\*[\s\S]*?\*\//g, ' ') .replace(/\$\d+/g, '?') .replace(/\s+/g, ' ') .trim() .slice(0, 500) } /** * 给池里的每条物理连接挂上「参数个数自检」。 * * pg 的 client.query 是原型方法,这里绑一份原方法后挂到实例属性上包装, * 事务里直接调 client.query 的仓库代码同样会被拦到。 */ function guardSqlParamArity(client: PoolClient) { const originalQuery = client.query.bind(client) client.query = ((...args: unknown[]) => { const [first, second] = args if (first && typeof first === 'object') { const config = first as { text?: unknown; values?: unknown } assertSqlParamArity(config.text, config.values) } else if (typeof second !== 'function') { assertSqlParamArity(first, second) } return (originalQuery as unknown as (...inner: unknown[]) => unknown)(...args) }) as unknown as PoolClient['query'] } export async function withTransaction(fn: (client: PoolClient) => Promise): Promise { const client = await getDb().connect() try { await client.query('BEGIN') const result = await fn(client) await client.query('COMMIT') return result } catch (error) { await client.query('ROLLBACK') throw error } finally { client.release() } } export async function closeDb(): Promise { if (!poolInstance) { return } const pool = poolInstance poolInstance = null await pool.end() }