claudetools/projects/dataforth-dos/datasheet-pipeline/current-testdatadb/db.js

/**
 * PostgreSQL Database Abstraction Layer
 *
 * Provides a connection pool and helper methods for the TestDataDB app.
 * Replaces better-sqlite3 singleton with pg.Pool.
 *
 * Environment variables (all optional, defaults connect to local PG):
 *   PGHOST (default: localhost)
 *   PGPORT (default: 5432)
 *   PGUSER (default: testdatadb_app)
 *   PGPASSWORD (default: DfTestDB2026!)
 *   PGDATABASE (default: testdatadb)
 */

const { Pool } = require('pg');

const pool = new Pool({
    host: process.env.PGHOST || 'localhost',
    port: parseInt(process.env.PGPORT || '5432', 10),
    user: process.env.PGUSER || 'testdatadb_app',
    password: process.env.PGPASSWORD || 'DfTestDB2026!',
    database: process.env.PGDATABASE || 'testdatadb',
    max: 20,
    idleTimeoutMillis: 30000,
    connectionTimeoutMillis: 5000,
});

pool.on('error', (err) => {
    console.error(`[${new Date().toISOString()}] [PG POOL ERROR] ${err.message}`);
});

/**
 * Convert SQLite-style ? placeholders to PostgreSQL $1, $2, ... placeholders.
 * Skips ? inside single-quoted strings.
 */
function convertPlaceholders(sql) {
    let idx = 0;
    let inString = false;
    let result = '';
    for (let i = 0; i < sql.length; i++) {
        const ch = sql[i];
        if (ch === "'" && (i === 0 || sql[i - 1] !== '\\')) {
            inString = !inString;
            result += ch;
        } else if (ch === '?' && !inString) {
            idx++;
            result += '$' + idx;
        } else {
            result += ch;
        }
    }
    return result;
}

/**
 * Execute a query, return all rows.
 * @param {string} sql - SQL with ? or $N placeholders
 * @param {Array} params - Parameter values
 * @returns {Promise<Array>} rows
 */
async function query(sql, params = []) {
    const pgSql = sql.includes('?') ? convertPlaceholders(sql) : sql;
    const result = await pool.query(pgSql, params);
    return result.rows;
}

/**
 * Execute a query, return the first row or null.
 */
async function queryOne(sql, params = []) {
    const rows = await query(sql, params);
    return rows[0] || null;
}

/**
 * Execute a statement (INSERT/UPDATE/DELETE), return { rowCount }.
 */
async function execute(sql, params = []) {
    const pgSql = sql.includes('?') ? convertPlaceholders(sql) : sql;
    const result = await pool.query(pgSql, params);
    return { rowCount: result.rowCount, rows: result.rows };
}

/**
 * Run a function inside a transaction.
 * The callback receives a client with query/execute helpers.
 * @param {Function} fn - async (client) => result
 * @returns {Promise<*>} result of fn
 */
async function transaction(fn) {
    const client = await pool.connect();
    try {
        await client.query('BEGIN');

        const txClient = {
            async query(sql, params = []) {
                const pgSql = sql.includes('?') ? convertPlaceholders(sql) : sql;
                const result = await client.query(pgSql, params);
                return result.rows;
            },
            async queryOne(sql, params = []) {
                const rows = await txClient.query(sql, params);
                return rows[0] || null;
            },
            async execute(sql, params = []) {
                const pgSql = sql.includes('?') ? convertPlaceholders(sql) : sql;
                const result = await client.query(pgSql, params);
                return { rowCount: result.rowCount, rows: result.rows };
            },
            // Direct pg client access for COPY or other advanced operations
            raw: client,
        };

        const result = await fn(txClient);
        await client.query('COMMIT');
        return result;
    } catch (err) {
        await client.query('ROLLBACK');
        throw err;
    } finally {
        client.release();
    }
}

/**
 * Close the pool (for graceful shutdown).
 */
async function close() {
    await pool.end();
}

/**
 * Get the raw pool (for advanced use like COPY).
 */
function getPool() {
    return pool;
}

module.exports = { query, queryOne, execute, transaction, close, getPool, convertPlaceholders };