101 lines
3.1 KiB
TypeScript
101 lines
3.1 KiB
TypeScript
import { Pool } from 'pg'
|
|
import type { PoolClient, QueryResult, QueryResultRow } from 'pg'
|
|
|
|
import { runtimeConfig } from '../config/runtime.js'
|
|
import { logWarn } from '../utils/logger.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),
|
|
})
|
|
})
|
|
poolInstance = pool
|
|
}
|
|
|
|
return poolInstance
|
|
}
|
|
|
|
export async function query(text: string, params?: unknown[]): Promise<QueryResult<any>>
|
|
export async function query<T extends QueryResultRow>(
|
|
text: string,
|
|
params?: unknown[],
|
|
): Promise<QueryResult<T>>
|
|
export async function query<T extends QueryResultRow>(
|
|
text: string,
|
|
params: unknown[] = [],
|
|
): Promise<QueryResult<T>> {
|
|
const pool = getDb()
|
|
const startedAt = process.hrtime.bigint()
|
|
try {
|
|
const result = await pool.query<T>(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)
|
|
}
|
|
|
|
export async function withTransaction<T>(fn: (client: PoolClient) => Promise<T>): Promise<T> {
|
|
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<void> {
|
|
if (!poolInstance) {
|
|
return
|
|
}
|
|
|
|
const pool = poolInstance
|
|
poolInstance = null
|
|
await pool.end()
|
|
}
|