pediatric-ai-scribe-v3/src/utils/auditQueue.js

// ============================================================
// Audit log batcher — collects audit/api/access entries in memory
// and flushes them to Postgres as batched multi-row INSERTs on a
// 1-second interval or when a buffer reaches 50 entries. Reduces
// per-request DB load from "one INSERT per audit call" to "one
// INSERT per ~50 audit calls under load". Transparent to callers.
//
// Trade-off: if the process crashes between flushes, up to ~1s of
// audit entries are lost. The PostgreSQL row is the primary
// destination; Loki is separately pushed fire-and-forget per call
// and is not batched (Loki has its own server-side ingestion that
// handles batching more cheaply).
// ============================================================

var FLUSH_INTERVAL_MS = 1000;
var FLUSH_MAX_BATCH = 50;

function createQueue(tableName, columns) {
  var buffer = [];
  var flushing = false;
  var timer = null;

  // Build "($1,$2,$3,...),($N,$N+1,...),..."  with running placeholder index
  function buildBatchSql(rows) {
    var cols = columns.join(', ');
    var placeholders = [];
    var params = [];
    var k = 1;
    for (var i = 0; i < rows.length; i++) {
      var row = rows[i];
      var tuple = [];
      for (var j = 0; j < columns.length; j++) {
        tuple.push('$' + k++);
        params.push(row[columns[j]] != null ? row[columns[j]] : null);
      }
      placeholders.push('(' + tuple.join(',') + ')');
    }
    return {
      sql: 'INSERT INTO ' + tableName + ' (' + cols + ') VALUES ' + placeholders.join(','),
      params: params
    };
  }

  async function flush() {
    if (flushing || buffer.length === 0) return;
    flushing = true;
    var toFlush = buffer.splice(0, FLUSH_MAX_BATCH);
    try {
      var pool = require('../db/database').pool;
      var q = buildBatchSql(toFlush);
      await pool.query(q.sql, q.params);
    } catch (err) {
      console.error('[auditQueue:' + tableName + '] flush failed:', err && err.message);
      // Don't re-buffer — these are audit entries, not user data.
      // Losing a burst on DB hiccup is acceptable; the alternative is
      // unbounded memory growth during DB outages.
    } finally {
      flushing = false;
      // If more accumulated during the await, schedule another round
      if (buffer.length >= FLUSH_MAX_BATCH) setImmediate(flush);
    }
  }

  function push(row) {
    buffer.push(row);
    if (buffer.length >= FLUSH_MAX_BATCH) setImmediate(flush);
    if (!timer) {
      timer = setInterval(flush, FLUSH_INTERVAL_MS);
      if (typeof timer.unref === 'function') timer.unref();
    }
  }

  // Called from the SIGTERM shutdown path — drains any remaining
  // entries before the process exits.
  async function drain() {
    if (timer) { clearInterval(timer); timer = null; }
    while (buffer.length > 0) {
      await flush();
      if (flushing) { await new Promise(function(r) { setTimeout(r, 20); }); }
    }
  }

  return { push: push, drain: drain, size: function() { return buffer.length; } };
}

var auditQueue = createQueue('audit_log', [
  'user_id', 'action', 'category', 'details',
  'ip_address', 'user_agent', 'model_used', 'tokens_used', 'duration_ms', 'status'
]);

var apiQueue = createQueue('api_log', [
  'user_id', 'endpoint', 'method', 'status_code',
  'request_size', 'response_size', 'model_used',
  'tokens_input', 'tokens_output', 'cost_estimate',
  'duration_ms', 'ip_address', 'error'
]);

var accessQueue = createQueue('access_log', [
  'user_id', 'action', 'ip_address', 'user_agent', 'success'
]);

async function drainAll() {
  try { await auditQueue.drain(); } catch(e) {}
  try { await apiQueue.drain(); } catch(e) {}
  try { await accessQueue.drain(); } catch(e) {}
}

module.exports = {
  audit: auditQueue,
  api: apiQueue,
  access: accessQueue,
  drainAll: drainAll
};