transactions for lql/migrate

[pyramation]

it('getClient', async () => {
  try {
    await client.withTransaction(async (client: PoolClient) => {
      const result = await client.query('SELECT 1');
      expect(result.rows[0]['?column?'] || result.rows[0].count).toBe(1);
    });
  } catch (e) {
    if (e instanceof AggregateError) {
      for (const err of e.errors) {
        console.error('AggregateError item:', err);
      }
    } else {
      console.error('Test failure:', e);
    }
    throw e;
  }
});

import { Pool, PoolClient } from 'pg';

export class Database {
  private pool: Pool;

  constructor(databaseUrl: string) {
    this.pool = new Pool({ connectionString: databaseUrl });

    process.on('SIGTERM', async () => {
      await this.shutdown();
    });
  }

  /**
   * Executes a callback function within a database transaction.
   */
  async withTransaction(fn: (client: PoolClient) => Promise<void>): Promise<void> {
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');
      try {
        await fn(client);
        await client.query('COMMIT');
      } catch (e) {
        console.error('Error during transaction:', e);
        await client.query('ROLLBACK');
        throw e;
      }
    } finally {
      client.release();
    }
  }

  /**
   * Shuts down the connection pool.
   */
  async shutdown(): Promise<void> {
    await this.pool.end();
  }
}