`agents/ingest-agent/server.js`

87 lines
#!/usr/bin/env node
/**
 * Ingest agent — HTTP control plane for bulk importers.
 *
 *   GET /health            — { ok, agent, db, jobs: [recent] }
 *   GET /jobs              — recent scrape_jobs rows
 *   POST /run/:importer    — async-fire one of: npi | mbc | hcai | hrsa | cms-care-compare | dca
 *
 * Importers themselves live in importers/<name>.js and run as child processes.
 * This server is intentionally thin — heavy work is one-shot Node scripts.
 */
const path = require('path');
const { spawn } = require('node:child_process');
require('dotenv').config({ path: path.resolve(__dirname, '../../.env') });
const express = require('express');
const { pool, query } = require('../shared/db');

const PORT = Number(process.env.PORT_INGEST || process.env.PORT || 9870);
const app = express();
app.use(express.json());

// LAN/tailnet auth gate (skip /health for watchdog probes)
const BASIC_AUTH = process.env.BASIC_AUTH || 'admin:DWSecure2024!';
const AUTH_HEADER = 'Basic ' + Buffer.from(BASIC_AUTH).toString('base64');
app.use((req, res, next) => {
  if (req.path === '/health' || req.path === '/healthz') return next();
  if (req.get('authorization') === AUTH_HEADER) return next();
  res.set('WWW-Authenticate', 'Basic realm="mac2-pm2"');
  res.status(401).send('auth required');
});

const IMPORTERS = ['npi', 'mbc', 'hcai', 'hrsa', 'cms-care-compare', 'dca', 'cms-hospitals'];

app.get('/health', async (_req, res) => {
  try {
    const r = await query(`
      SELECT id, job_label, status, records_inserted, records_updated, started_at, finished_at
        FROM scrape_jobs ORDER BY id DESC LIMIT 5
    `, []);
    res.json({ ok: true, agent: 'pd-ingest', db: 'doctor_professional_directory', jobs: r.rows });
  } catch (e) {
    res.status(500).json({ ok: false, error: e.message });
  }
});

app.get('/jobs', async (_req, res, next) => {
  try {
    const r = await query(`
      SELECT j.*, s.source_name FROM scrape_jobs j
      LEFT JOIN sources s ON s.id = j.source_id
      ORDER BY j.id DESC LIMIT 50
    `, []);
    res.json({ count: r.rowCount, rows: r.rows });
  } catch (e) { next(e); }
});

// SECURITY: dropped req.body.env passthrough (was caller-controlled env merge
// into spawned child; could overwrite NODE_PATH / DB / Stripe keys). Added
// shared-secret token check. INGEST_RUN_TOKEN must be set in pm2 env.
const INGEST_RUN_TOKEN = process.env.INGEST_RUN_TOKEN || '';
app.post('/run/:importer', (req, res) => {
  if (!INGEST_RUN_TOKEN) {
    return res.status(503).json({ error: 'INGEST_RUN_TOKEN not configured on server' });
  }
  const auth = String(req.get('authorization') || '');
  if (auth !== `Bearer ${INGEST_RUN_TOKEN}`) {
    return res.status(401).json({ error: 'unauthorized' });
  }
  const name = String(req.params.importer);
  if (!IMPORTERS.includes(name)) {
    return res.status(400).json({ error: `unknown importer; allowed: ${IMPORTERS.join(', ')}` });
  }
  const script = path.join(__dirname, 'importers', `${name}.js`);
  const child = spawn('node', [script], {
    cwd: path.resolve(__dirname, '../..'),
    env: process.env,   // NO caller-supplied env merge
    detached: true,
    stdio: 'ignore',
  });
  child.unref();
  res.json({ ok: true, started: name, pid: child.pid });
});

app.listen(PORT, '0.0.0.0', () => {
  console.log(`[pd-ingest] listening on http://127.0.0.1:${PORT}`);
});