apps/backend/src/infrastructure/database/database.service.tsTypeScript
import { Injectable, OnModuleDestroy } from '@nestjs/common';
import { Pool, type PoolClient, type QueryResultRow } from 'pg';
export interface DatabaseQueryClient {
query<T extends QueryResultRow = Record<string, unknown>>(
text: string,
params?: unknown[],
): Promise<T[]>;
}
class PgTransactionClient implements DatabaseQueryClient {
constructor(private readonly client:PoolClient) {}
async query<T extends QueryResultRow = Record<string, unknown>>(
text:string,
params:unknown[]=[],
):Promise<T[]> {
const result=await this.client.query(text,params);
return result.rows as T[];
}
}
@Injectable()
export class DatabaseService implements OnModuleDestroy,DatabaseQueryClient {
private readonly pool = new Pool({
connectionString: process.env.DATABASE_URL,
max: Number(process.env.DATABASE_POOL_MAX ?? 20),
connectionTimeoutMillis:Number(process.env.DATABASE_CONNECT_TIMEOUT_MS??5000),
idleTimeoutMillis:Number(process.env.DATABASE_IDLE_TIMEOUT_MS??30000),
statement_timeout:Number(process.env.DATABASE_STATEMENT_TIMEOUT_MS??15000),
query_timeout:Number(process.env.DATABASE_QUERY_TIMEOUT_MS??20000),
application_name:process.env.DATABASE_APPLICATION_NAME??'gastro-os-backend',
keepAlive:true,
});
async query<T extends QueryResultRow = Record<string, unknown>>(
text:string,
params:unknown[]=[],
):Promise<T[]> {
const result=await this.pool.query(text,params);
return result.rows as T[];
}
async transaction<T>(
callback:(tx:DatabaseQueryClient)=>Promise<T>,
options?:{
isolationLevel?:'READ COMMITTED'|'REPEATABLE READ'|'SERIALIZABLE';
statementTimeoutMs?:number;
lockTimeoutMs?:number;
},
):Promise<T> {
const client=await this.pool.connect();
try {
await client.query('BEGIN');
if(options?.isolationLevel) {
await client.query(`SET TRANSACTION ISOLATION LEVEL ${options.isolationLevel}`);
}
if(options?.statementTimeoutMs) {
await client.query(`SET LOCAL statement_timeout = '${Math.max(1,Math.floor(options.statementTimeoutMs))}ms'`);
}
if(options?.lockTimeoutMs) {
await client.query(`SET LOCAL lock_timeout = '${Math.max(1,Math.floor(options.lockTimeoutMs))}ms'`);
}
const tx=new PgTransactionClient(client);
const result=await callback(tx);
await client.query('COMMIT');
return result;
} catch(error) {
try { await client.query('ROLLBACK'); } catch {}
throw error;
} finally {
client.release();
}
}
async ping(): Promise<void> {
await this.pool.query('select 1');
}
poolStats() {
return {
total:this.pool.totalCount,
idle:this.pool.idleCount,
waiting:this.pool.waitingCount,
max:Number(process.env.DATABASE_POOL_MAX??20),
};
}
async migrationStatus() {
const rows=await this.query<any>(
`select count(*)::int as count,max(applied_at) as latest from schema_migrations`,
);
return {applied:Number(rows[0]?.count??0),latest:rows[0]?.latest??null};
}
async onModuleDestroy(): Promise<void> {
await this.pool.end();
}
}