/**
 * pg.ts — pooled Postgres + LISTEN/NOTIFY helpers.
 *
 * The pool is for short-lived queries inside workers. The LISTEN connection
 * is a single dedicated client kept alive for the lifetime of the process.
 */
import pg from "pg";
import { env } from "./env";

export const pool = new pg.Pool({
  connectionString: env.DATABASE_URL,
  max: 4,
  idleTimeoutMillis: 30_000,
  allowExitOnIdle: false,
});

export async function query<T = unknown>(sql: string, params: unknown[] = []): Promise<T[]> {
  const r = await pool.query(sql, params as unknown[]);
  return r.rows as T[];
}

export async function queryOne<T = unknown>(sql: string, params: unknown[] = []): Promise<T | null> {
  const rows = await query<T>(sql, params);
  return rows[0] ?? null;
}

/**
 * Dedicated long-lived connection that issues LISTEN <channel> and yields
 * notification payloads as an async iterator.
 */
export async function listen(
  channel: string,
  onNotify: (payload: string | undefined) => void,
): Promise<pg.Client> {
  const client = new pg.Client({ connectionString: env.DATABASE_URL });
  await client.connect();
  await client.query(`LISTEN ${channel}`);
  client.on("notification", (msg) => {
    if (msg.channel !== channel) return;
    try {
      onNotify(msg.payload);
    } catch (e) {
      console.error(`[listen ${channel}] handler threw: ${(e as Error).message}`);
    }
  });
  client.on("error", (e) => {
    console.error(`[listen ${channel}] connection error: ${e.message}`);
  });
  return client;
}

export async function close(): Promise<void> {
  await pool.end().catch(() => undefined);
}
