25190541308

Committed 30 Apr 2026 09:36PM UTC coverage: 90.202% (+0.06%) from 90.144%

Build # 25190541308

Build Type

push

github

Committed by

web-flow

Commit Message

Audit triage moderation and budget events (#629)

* fix: preserve pr629 audit cleanup

* fix: throttle triage budget audit spam

Coverage Stats

9685 of 11366 branches covered (85.21%)

Branch coverage included in aggregate %.

74 of 77 new or added lines in 4 files covered. (96.1%)

3 existing lines in 2 files now uncovered.

15329 of 16365 relevant lines covered (93.67%)

177.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.22

/src/db.js

/**
 * Database Module
 * PostgreSQL connection pool and migration runner
 */

import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { runner } from 'node-pg-migrate';
import pg from 'pg';
import { debug, info, error as logError, warn } from './logger.js';

const { Pool } = pg;

/** @type {pg.Pool | null} */
let pool = null;

/** @type {boolean} Re-entrancy guard for initDb */
let initializing = false;

/** @type {ReturnType<typeof setInterval> | null} */
let leakDetectionInterval = null;

/**
 * Selects the SSL configuration for a pg.Pool based on DATABASE_SSL and the connection string.
 *
 * DATABASE_SSL values:
 *   "false" / "off" / "disable" → SSL disabled
 *   "no-verify"                 → SSL enabled but server certificate not verified
 *   "true" / "on" / "require"   → SSL enabled with server certificate verification
 *
 * If DATABASE_SSL is unset, SSL is disabled for local connections and enabled
 * with full certificate verification for non-local connections.
 *
 * @param {string} connectionString - Database connection URL
 * @returns {false|{rejectUnauthorized: boolean}} `false` to disable SSL, or an object with `rejectUnauthorized` indicating whether server certificates must be verified
 */
function getSslConfig(connectionString) {
  let hostname = '';
  let sslMode = '';

  try {
    const connectionUrl = new URL(connectionString);
    hostname = connectionUrl.hostname.toLowerCase();
    sslMode = (connectionUrl.searchParams.get('sslmode') || '').toLowerCase().trim();
  } catch {
    // Ignore malformed URLs and fall back to safe defaults.
  }

  // Explicit sslmode=disable in connection string takes precedence.
  if (sslMode === 'disable' || sslMode === 'off' || sslMode === 'false') {
    return false;
  }

  // Railway internal connections never need SSL.
  if (hostname.includes('railway.internal') || connectionString.includes('railway.internal')) {
    return false;
  }

  const sslEnv = (process.env.DATABASE_SSL || '').toLowerCase().trim();

  if (sslEnv === 'false' || sslEnv === 'off' || sslEnv === 'disable' || sslEnv === '0') {
    return false;
  }

  if (sslEnv === 'no-verify') {
    return { rejectUnauthorized: false };
  }

  if (sslEnv === 'true' || sslEnv === 'on' || sslEnv === 'require' || sslEnv === '1') {
    return { rejectUnauthorized: true };
  }

  // Local development databases commonly run without TLS.
  if (!sslEnv && ['localhost', '127.0.0.1', '::1'].includes(hostname)) {
    return false;
  }

  if (sslEnv) {
    warn('Unrecognized DATABASE_SSL value, using secure default', {
      value: sslEnv,
      source: 'database_ssl',
    });
  }

  // Default: SSL with full verification.
  return { rejectUnauthorized: true };
}

/**
 * Apply pending PostgreSQL schema migrations from the project's migrations directory.
 *
 * @param {string} databaseUrl - Connection string used to run migrations against the database.
 * @returns {Promise<void>}
 */
async function runMigrations(databaseUrl) {
  const __filename = fileURLToPath(import.meta.url);
  const __dirname = path.dirname(__filename);
  const migrationsDir = path.resolve(__dirname, '..', 'migrations');

  await runner({
    databaseUrl,
    dir: migrationsDir,
    direction: 'up',
    migrationsTable: 'pgmigrations',
    // Historical 004_* no-op migrations were run in different orders across environments.
    // Let node-pg-migrate apply pending files without failing on filename order drift.
    checkOrder: false,
    log: (msg) => {
      if (typeof msg === 'string' && msg.includes("Can't determine timestamp")) return;
      info(msg);
    },
  });

  info('Database migrations applied');
}

/**
 * Wrap pool.query with slow query detection.
 * Logs queries that exceed the threshold with timing info.
 *
 * @param {pg.Pool} poolInstance - The pool instance to wrap
 * @param {number} thresholdMs - Slow query threshold in milliseconds
 */
function wrapPoolQuery(poolInstance, thresholdMs) {
  const originalQuery = poolInstance.query.bind(poolInstance);

  poolInstance.query = async function wrappedQuery(...args) {
    const start = Date.now();
    try {
      const result = await originalQuery(...args);
      const duration = Date.now() - start;

      if (duration >= thresholdMs) {
        const queryText = typeof args[0] === 'string' ? args[0] : (args[0]?.text ?? 'unknown');
        warn('Slow query detected', {
          duration_ms: duration,
          threshold_ms: thresholdMs,
          query: queryText.slice(0, 500),
          source: 'slow_query_log',
        });

        // Attempt EXPLAIN (best-effort, fire-and-forget — do not await)
        const explainText = typeof args[0] === 'string' ? args[0] : (args[0]?.text ?? '');
        const explainValues = Array.isArray(args[1]) ? args[1] : (args[0]?.values ?? []);

        if (/^\s*SELECT/i.test(explainText)) {
          originalQuery({
            text: `EXPLAIN ${explainText}`,
            values: explainValues,
          })
            .then((explainResult) => {
              const plan = explainResult.rows.map((r) => Object.values(r)[0]).join('\n');
              warn('Slow query EXPLAIN plan', {
                duration_ms: duration,
                query: queryText.slice(0, 200),
                plan: plan.slice(0, 2000),
                source: 'slow_query_log',
              });
            })
            .catch(() => {});
        }
      }

      return result;
    } catch (err) {
      const duration = Date.now() - start;
      const queryText = typeof args[0] === 'string' ? args[0] : (args[0]?.text ?? 'unknown');
      logError('Query failed', {
        duration_ms: duration,
        query: queryText.slice(0, 500),
        error: err.message,
        source: 'db_query',
      });
      throw err;
    }
  };
}

/**
 * Start the connection leak detection interval.
 * Warns if the pool is near capacity or has waiting requests.
 *
 * @param {pg.Pool} poolInstance - The pool to monitor
 * @param {number} maxSize - Configured pool max size
 */
function startLeakDetection(poolInstance, maxSize) {
  if (leakDetectionInterval) return;

  leakDetectionInterval = setInterval(() => {
    const waiting = poolInstance.waitingCount;
    const total = poolInstance.totalCount;

    if (waiting > 0) {
      warn('Database connection pool has waiting clients', {
        waiting,
        total,
        idle: poolInstance.idleCount,
        max: maxSize,
        source: 'pool_monitor',
      });
    } else {
      const activeCount = poolInstance.totalCount - poolInstance.idleCount;
      if (activeCount >= maxSize * 0.8) {
        warn('Database connection pool nearing capacity', {
          total,
          active: activeCount,
          idle: poolInstance.idleCount,
          waiting,
          max: maxSize,
          utilization_pct: Math.round((activeCount / maxSize) * 100),
          source: 'pool_monitor',
        });
      }
    }
  }, 30_000).unref();
}

/**
 * Stop the connection leak detection interval.
 */
export function stopLeakDetection() {
  if (leakDetectionInterval) {
    clearInterval(leakDetectionInterval);
    leakDetectionInterval = null;
  }
}

/**
 * Get pool statistics.
 *
 * @returns {{ total: number, idle: number, waiting: number } | null} Pool stats or null if not initialized
 */
export function getPoolStats() {
  if (!pool) return null;
  return {
    total: pool.totalCount,
    idle: pool.idleCount,
    waiting: pool.waitingCount,
  };
}

/**
 * Initialize the PostgreSQL connection pool and apply any pending database migrations.
 *
 * @returns {Promise<pg.Pool>} The initialized pg.Pool instance.
 * @throws {Error} If initialization is already in progress.
 * @throws {Error} If the DATABASE_URL environment variable is not set.
 * @throws {Error} If the connection test or migration application fails.
 */
export async function initDb() {
  if (initializing) {
    throw new Error('initDb is already in progress');
  }
  if (pool) return pool;

  initializing = true;
  try {
    const connectionString = process.env.DATABASE_URL;
    if (!connectionString) {
      throw new Error('DATABASE_URL environment variable is not set');
    }

    /** @param {string|undefined} env @param {number} defaultVal */
    const parsePositiveInt = (env, defaultVal) => {
      const val = parseInt(env, 10);
      return Number.isNaN(val) || val < 0 ? defaultVal : val;
    };

    const poolSize = Math.max(1, parsePositiveInt(process.env.PG_POOL_SIZE, 5));
    const idleTimeoutMs = parsePositiveInt(process.env.PG_IDLE_TIMEOUT_MS, 30000);
    const connectionTimeoutMs = parsePositiveInt(process.env.PG_CONNECTION_TIMEOUT_MS, 10000);

    pool = new Pool({
      connectionString,
      max: poolSize,
      idleTimeoutMillis: idleTimeoutMs,
      connectionTimeoutMillis: connectionTimeoutMs,
      ssl: getSslConfig(connectionString),
    });

    // Prevent unhandled pool errors from crashing the process
    pool.on('error', (err) => {
      logError('Unexpected database pool error', { error: err.message, source: 'database_pool' });
    });

    // Pool event listeners for observability
    pool.on('connect', () => {
      debug('Database pool: new client connected', {
        total: pool.totalCount,
        idle: pool.idleCount,
        waiting: pool.waitingCount,
        source: 'pool_events',
      });
    });

    pool.on('acquire', () => {
      debug('Database pool: client acquired', {
        total: pool.totalCount,
        idle: pool.idleCount,
        waiting: pool.waitingCount,
        source: 'pool_events',
      });
    });

    pool.on('remove', () => {
      debug('Database pool: client removed', {
        total: pool.totalCount,
        idle: pool.idleCount,
        waiting: pool.waitingCount,
        source: 'pool_events',
      });
    });

    // Wrap query with slow query logging
    const slowQueryThresholdMs = parsePositiveInt(process.env.PG_SLOW_QUERY_MS, 100);
    wrapPoolQuery(pool, slowQueryThresholdMs);

    try {
      // Test connection
      const client = await pool.connect();
      try {
        await client.query('SELECT NOW()');
        info('Database connected');
      } finally {
        client.release();
      }

      // Run pending migrations
      await runMigrations(connectionString);

      info('Database schema initialized');
    } catch (err) {
      // Clean up the pool so getPool() doesn't return an unusable instance
      await pool.end().catch(() => {});
      pool = null;
      throw err;
    }

    // Start connection leak detection
    startLeakDetection(pool, poolSize);

    return pool;
  } finally {
    initializing = false;
  }
}

/**
 * Get the database pool
 * @returns {pg.Pool} The connection pool
 * @throws {Error} If pool is not initialized
 */
export function getPool() {
  if (!pool) {
    throw new Error('Database not initialized. Call initDb() first.');
  }
  return pool;
}

/**
 * Gracefully close the database pool
 */
export async function closeDb() {
  stopLeakDetection();
  if (pool) {
    try {
      await pool.end();
      info('Database pool closed');
    } catch (err) {
      logError('Error closing database pool', { error: err.message });
    } finally {
      pool = null;
    }
  }
}

1	/**
2	* Database Module
3	* PostgreSQL connection pool and migration runner
4	*/
5
6	import path from 'node:path';
7	import { fileURLToPath } from 'node:url';
8	import { runner } from 'node-pg-migrate';
9	import pg from 'pg';
10	import { debug, info, error as logError, warn } from './logger.js';
11
12	const { Pool } = pg;	85✔
13
14	/** @type {pg.Pool \| null} */
15	let pool = null;	85✔
16
17	/** @type {boolean} Re-entrancy guard for initDb */
18	let initializing = false;	85✔
19
20	/** @type {ReturnType<typeof setInterval> \| null} */
21	let leakDetectionInterval = null;	85✔
22
23	/**
24	* Selects the SSL configuration for a pg.Pool based on DATABASE_SSL and the connection string.
25	*
26	* DATABASE_SSL values:
27	* "false" / "off" / "disable" → SSL disabled
28	* "no-verify" → SSL enabled but server certificate not verified
29	* "true" / "on" / "require" → SSL enabled with server certificate verification
30	*
31	* If DATABASE_SSL is unset, SSL is disabled for local connections and enabled
32	* with full certificate verification for non-local connections.
33	*
34	* @param {string} connectionString - Database connection URL
35	* @returns {false\|{rejectUnauthorized: boolean}} `false` to disable SSL, or an object with `rejectUnauthorized` indicating whether server certificates must be verified
36	*/
37	function getSslConfig(connectionString) {
38	let hostname = '';	27✔
39	let sslMode = '';	27✔
40
41	try {	27✔
42	const connectionUrl = new URL(connectionString);	27✔
43	hostname = connectionUrl.hostname.toLowerCase();	27✔
44	sslMode = (connectionUrl.searchParams.get('sslmode') \|\| '').toLowerCase().trim();	27✔
45	} catch {
46	// Ignore malformed URLs and fall back to safe defaults.
47	}
48
49	// Explicit sslmode=disable in connection string takes precedence.
50	if (sslMode === 'disable' \|\| sslMode === 'off' \|\| sslMode === 'false') {	27✔
51	return false;	1✔
52	}
53
54	// Railway internal connections never need SSL.
55	if (hostname.includes('railway.internal') \|\| connectionString.includes('railway.internal')) {	26✔
56	return false;	1✔
57	}
58
59	const sslEnv = (process.env.DATABASE_SSL \|\| '').toLowerCase().trim();	25✔
60
61	if (sslEnv === 'false' \|\| sslEnv === 'off' \|\| sslEnv === 'disable' \|\| sslEnv === '0') {	27✔
62	return false;	2✔
63	}
64
65	if (sslEnv === 'no-verify') {	23✔
66	return { rejectUnauthorized: false };	1✔
67	}
68
69	if (sslEnv === 'true' \|\| sslEnv === 'on' \|\| sslEnv === 'require' \|\| sslEnv === '1') {	22✔
70	return { rejectUnauthorized: true };	1✔
71	}
72
73	// Local development databases commonly run without TLS.
74	if (!sslEnv && ['localhost', '127.0.0.1', '::1'].includes(hostname)) {	21✔
75	return false;	20✔
76	}
77
78	if (sslEnv) {	1!
79	warn('Unrecognized DATABASE_SSL value, using secure default', {	×
80	value: sslEnv,
81	source: 'database_ssl',
82	});
83	}
84
85	// Default: SSL with full verification.
86	return { rejectUnauthorized: true };	1✔
87	}
88
89	/**
90	* Apply pending PostgreSQL schema migrations from the project's migrations directory.
91	*
92	* @param {string} databaseUrl - Connection string used to run migrations against the database.
93	* @returns {Promise<void>}
94	*/
95	async function runMigrations(databaseUrl) {
96	const __filename = fileURLToPath(import.meta.url);	26✔
97	const __dirname = path.dirname(__filename);	26✔
98	const migrationsDir = path.resolve(__dirname, '..', 'migrations');	26✔
99
100	await runner({	26✔
101	databaseUrl,
102	dir: migrationsDir,
103	direction: 'up',
104	migrationsTable: 'pgmigrations',
105	// Historical 004_* no-op migrations were run in different orders across environments.
106	// Let node-pg-migrate apply pending files without failing on filename order drift.
107	checkOrder: false,
108	log: (msg) => {
109	if (typeof msg === 'string' && msg.includes("Can't determine timestamp")) return;	×
110	info(msg);	×
111	},
112	});
113
114	info('Database migrations applied');	25✔
115	}
116
117	/**
118	* Wrap pool.query with slow query detection.
119	* Logs queries that exceed the threshold with timing info.
120	*
121	* @param {pg.Pool} poolInstance - The pool instance to wrap
122	* @param {number} thresholdMs - Slow query threshold in milliseconds
123	*/
124	function wrapPoolQuery(poolInstance, thresholdMs) {
125	const originalQuery = poolInstance.query.bind(poolInstance);	27✔
126
127	poolInstance.query = async function wrappedQuery(...args) {	27✔
128	const start = Date.now();	4✔
129	try {	4✔
130	const result = await originalQuery(...args);	4✔
131	const duration = Date.now() - start;	3✔
132
133	if (duration >= thresholdMs) {	3✔
134	const queryText = typeof args[0] === 'string' ? args[0] : (args[0]?.text ?? 'unknown');	1!
135	warn('Slow query detected', {	1✔
136	duration_ms: duration,
137	threshold_ms: thresholdMs,
138	query: queryText.slice(0, 500),
139	source: 'slow_query_log',
140	});
141
142	// Attempt EXPLAIN (best-effort, fire-and-forget — do not await)
143	const explainText = typeof args[0] === 'string' ? args[0] : (args[0]?.text ?? '');	1!
144	const explainValues = Array.isArray(args[1]) ? args[1] : (args[0]?.values ?? []);	1!
145
146	if (/^\s*SELECT/i.test(explainText)) {	1!
147	originalQuery({	1✔
148	text: `EXPLAIN ${explainText}`,
149	values: explainValues,
150	})
151	.then((explainResult) => {
UNCOV 152	const plan = explainResult.rows.map((r) => Object.values(r)[0]).join('\n');	×
UNCOV 153	warn('Slow query EXPLAIN plan', {	×
154	duration_ms: duration,
155	query: queryText.slice(0, 200),
156	plan: plan.slice(0, 2000),
157	source: 'slow_query_log',
158	});
159	})
160	.catch(() => {});
161	}
162	}
163
164	return result;	3✔
165	} catch (err) {
166	const duration = Date.now() - start;	1✔
167	const queryText = typeof args[0] === 'string' ? args[0] : (args[0]?.text ?? 'unknown');	1!
168	logError('Query failed', {	1✔
169	duration_ms: duration,
170	query: queryText.slice(0, 500),
171	error: err.message,
172	source: 'db_query',
173	});
174	throw err;	1✔
175	}
176	};
177	}
178
179	/**
180	* Start the connection leak detection interval.
181	* Warns if the pool is near capacity or has waiting requests.
182	*
183	* @param {pg.Pool} poolInstance - The pool to monitor
184	* @param {number} maxSize - Configured pool max size
185	*/
186	function startLeakDetection(poolInstance, maxSize) {
187	if (leakDetectionInterval) return;	25!
188
189	leakDetectionInterval = setInterval(() => {	25✔
190	const waiting = poolInstance.waitingCount;	×
191	const total = poolInstance.totalCount;	×
192
193	if (waiting > 0) {	×
194	warn('Database connection pool has waiting clients', {	×
195	waiting,
196	total,
197	idle: poolInstance.idleCount,
198	max: maxSize,
199	source: 'pool_monitor',
200	});
201	} else {
202	const activeCount = poolInstance.totalCount - poolInstance.idleCount;	×
203	if (activeCount >= maxSize * 0.8) {	×
204	warn('Database connection pool nearing capacity', {	×
205	total,
206	active: activeCount,
207	idle: poolInstance.idleCount,
208	waiting,
209	max: maxSize,
210	utilization_pct: Math.round((activeCount / maxSize) * 100),
211	source: 'pool_monitor',
212	});
213	}
214	}
215	}, 30_000).unref();
216	}
217
218	/**
219	* Stop the connection leak detection interval.
220	*/
221	export function stopLeakDetection() {
222	if (leakDetectionInterval) {	34✔
223	clearInterval(leakDetectionInterval);	25✔
224	leakDetectionInterval = null;	25✔
225	}
226	}
227
228	/**
229	* Get pool statistics.
230	*
231	* @returns {{ total: number, idle: number, waiting: number } \| null} Pool stats or null if not initialized
232	*/
233	export function getPoolStats() {
234	if (!pool) return null;	8✔
235	return {	2✔
236	total: pool.totalCount,
237	idle: pool.idleCount,
238	waiting: pool.waitingCount,
239	};
240	}
241
242	/**
243	* Initialize the PostgreSQL connection pool and apply any pending database migrations.
244	*
245	* @returns {Promise<pg.Pool>} The initialized pg.Pool instance.
246	* @throws {Error} If initialization is already in progress.
247	* @throws {Error} If the DATABASE_URL environment variable is not set.
248	* @throws {Error} If the connection test or migration application fails.
249	*/
250	export async function initDb() {
251	if (initializing) {	30✔
252	throw new Error('initDb is already in progress');	1✔
253	}
254	if (pool) return pool;	29✔
255
256	initializing = true;	28✔
257	try {	28✔
258	const connectionString = process.env.DATABASE_URL;	28✔
259	if (!connectionString) {	28✔
260	throw new Error('DATABASE_URL environment variable is not set');	1✔
261	}
262
263	/** @param {string\|undefined} env @param {number} defaultVal */
264	const parsePositiveInt = (env, defaultVal) => {	27✔
265	const val = parseInt(env, 10);	108✔
266	return Number.isNaN(val) \|\| val < 0 ? defaultVal : val;	108✔
267	};
268
269	const poolSize = Math.max(1, parsePositiveInt(process.env.PG_POOL_SIZE, 5));	27✔
270	const idleTimeoutMs = parsePositiveInt(process.env.PG_IDLE_TIMEOUT_MS, 30000);	27✔
271	const connectionTimeoutMs = parsePositiveInt(process.env.PG_CONNECTION_TIMEOUT_MS, 10000);	27✔
272
273	pool = new Pool({	27✔
274	connectionString,
275	max: poolSize,
276	idleTimeoutMillis: idleTimeoutMs,
277	connectionTimeoutMillis: connectionTimeoutMs,
278	ssl: getSslConfig(connectionString),
279	});
280
281	// Prevent unhandled pool errors from crashing the process
282	pool.on('error', (err) => {	27✔
283	logError('Unexpected database pool error', { error: err.message, source: 'database_pool' });	×
284	});
285
286	// Pool event listeners for observability
287	pool.on('connect', () => {	27✔
288	debug('Database pool: new client connected', {	×
289	total: pool.totalCount,
290	idle: pool.idleCount,
291	waiting: pool.waitingCount,
292	source: 'pool_events',
293	});
294	});
295
296	pool.on('acquire', () => {	27✔
297	debug('Database pool: client acquired', {	×
298	total: pool.totalCount,
299	idle: pool.idleCount,
300	waiting: pool.waitingCount,
301	source: 'pool_events',
302	});
303	});
304
305	pool.on('remove', () => {	27✔
306	debug('Database pool: client removed', {	×
307	total: pool.totalCount,
308	idle: pool.idleCount,
309	waiting: pool.waitingCount,
310	source: 'pool_events',
311	});
312	});
313
314	// Wrap query with slow query logging
315	const slowQueryThresholdMs = parsePositiveInt(process.env.PG_SLOW_QUERY_MS, 100);	27✔
316	wrapPoolQuery(pool, slowQueryThresholdMs);	27✔
317
318	try {	27✔
319	// Test connection
320	const client = await pool.connect();	27✔
321	try {	26✔
322	await client.query('SELECT NOW()');	26✔
323	info('Database connected');	26✔
324	} finally {
325	client.release();	26✔
326	}
327
328	// Run pending migrations
329	await runMigrations(connectionString);	26✔
330
331	info('Database schema initialized');	25✔
332	} catch (err) {
333	// Clean up the pool so getPool() doesn't return an unusable instance
334	await pool.end().catch(() => {});	2✔
335	pool = null;	2✔
336	throw err;	2✔
337	}
338
339	// Start connection leak detection
340	startLeakDetection(pool, poolSize);	25✔
341
342	return pool;	25✔
343	} finally {
344	initializing = false;	28✔
345	}
346	}
347
348	/**
349	* Get the database pool
350	* @returns {pg.Pool} The connection pool
351	* @throws {Error} If pool is not initialized
352	*/
353	export function getPool() {
354	if (!pool) {	75✔
355	throw new Error('Database not initialized. Call initDb() first.');	70✔
356	}
357	return pool;	5✔
358	}
359
360	/**
361	* Gracefully close the database pool
362	*/
363	export async function closeDb() {
364	stopLeakDetection();	34✔
365	if (pool) {	34✔
366	try {	25✔
367	await pool.end();	25✔
368	info('Database pool closed');	24✔
369	} catch (err) {
370	logError('Error closing database pool', { error: err.message });	1✔
371	} finally {
372	pool = null;	25✔
373	}
374	}
375	}

VolvoxLLC / volvox-bot / 25190541308

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous