`src/jobs/broker_rotate.ts`

52 lines
/**
 * Broker-roster rotation: each run refreshes the STALEST state's licensing
 * roster, so a daily schedule loops through every registered state continuously
 * through the month (5 states ≈ every 5 days each; more states = wider loop —
 * new wave-2 adapters join the rotation automatically once they've run once).
 * Spawns the engine CLI so this file never touches engine.ts (agent territory).
 * `--state xx` overrides selection (testing).
 */
import { execFileSync } from 'node:child_process';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { query, pool } from '../../db/pool.ts';

const __dirname = dirname(fileURLToPath(import.meta.url));
const ROOT = join(__dirname, '..', '..');

async function pickStalest(): Promise<{ source: string; stateArg: string; lastOk: string | null } | null> {
  // every adapter that has EVER ingested brokers is in the rotation; engine arg = prefix before '_'
  const r = await query<{ source: string; last_ok: string | null }>(
    `SELECT b.source, MAX(ir.started_at) FILTER (WHERE ir.status = 'ok')::text AS last_ok
       FROM (SELECT DISTINCT source FROM broker) b
       LEFT JOIN ingest_runs ir ON ir.source = b.source
      GROUP BY b.source
      ORDER BY MAX(ir.started_at) FILTER (WHERE ir.status = 'ok') ASC NULLS FIRST
      LIMIT 1`,
  );
  if (!r.rows.length) return null;
  const source = r.rows[0].source;
  return { source, stateArg: source.split('_')[0], lastOk: r.rows[0].last_ok };
}

async function main() {
  const argIdx = process.argv.indexOf('--state');
  let stateArg: string;
  if (argIdx > -1 && process.argv[argIdx + 1]) {
    stateArg = process.argv[argIdx + 1];
    console.log(`[broker-rotate] override: ${stateArg}`);
  } else {
    const pick = await pickStalest();
    if (!pick) { console.log('[broker-rotate] no broker sources registered yet'); await pool.end(); return; }
    stateArg = pick.stateArg;
    console.log(`[broker-rotate] stalest = ${pick.source} (last ok: ${pick.lastOk ?? 'never'}) -> refreshing '${stateArg}'`);
  }
  await pool.end(); // engine child owns its own pool
  execFileSync('npx', ['tsx', 'src/ingest/brokers/engine.ts', stateArg], {
    cwd: ROOT, stdio: 'inherit', timeout: 60 * 60_000,
  });
  console.log(`[broker-rotate] done: ${stateArg}`);
}

main().catch(e => { console.error('[broker-rotate] FAILED:', e); process.exit(1); });