didi-lot2-backend/backend/services/orchestration-layer/agent-v3/scripts/verify-rabbitmq-cluster.ts

/**
 * RabbitMQ cluster readiness verification.
 *
 * Runs checks against the RabbitMQ instance configured via env vars
 * (RABBITMQ_HOST/PORT/USER/PASS/VHOST). Intended to be run:
 *   - Pre-cutover: confirm cluster accessible + vhost + privileges
 *   - Post-cutover: confirm expected topology created
 *   - CI / healthcheck: block container start until broker is reachable
 *
 * Exit codes:
 *   0 — all checks pass
 *   1 — fatal issue (connectivity, auth, missing vhost)
 *   2 — warning (topology partially initialized — normal pre-workers)
 *
 * Usage:
 *   RABBITMQ_HOST=10.11.50.100 RABBITMQ_PORT=16672 \
 *     RABBITMQ_USER=didi RABBITMQ_PASS=... RABBITMQ_VHOST=/didi \
 *     npx ts-node --transpile-only scripts/verify-rabbitmq-cluster.ts
 */

import amqp from 'amqplib';
import {
  getRabbitMQUrl,
  getRabbitMQConfig,
  EXCHANGE_NAME,
  QUEUE,
  MEDIA_QUEUE,
  ANALYSIS_COMPONENTS,
  type PlanType,
} from '../src/shared/queue/constants';

const PLAN_TYPES: PlanType[] = [1, 2, 3, 4, 5, 6];

type Severity = 'OK' | 'WARN' | 'FAIL';
interface Check { name: string; severity: Severity; detail: string; }

const results: Check[] = [];
let fatal = 0, warn = 0;

function record(name: string, severity: Severity, detail: string) {
  results.push({ name, severity, detail });
  if (severity === 'FAIL') fatal++;
  else if (severity === 'WARN') warn++;
}

async function main() {
  const cfg = getRabbitMQConfig();
  const target = `${cfg.host}:${cfg.port}${cfg.vhost}`;

  console.log(`\n${'='.repeat(70)}`);
  console.log(`RabbitMQ cluster readiness — ${target} (user: ${cfg.user})`);
  console.log('='.repeat(70));

  // -------------------------------------------------------------------------
  // 1. Connect + auth
  // -------------------------------------------------------------------------
  let conn: any = null;
  try {
    conn = await amqp.connect(getRabbitMQUrl());
    record('connectivity', 'OK', `amqp.connect → ${target}`);
  } catch (err: any) {
    // Map common errors to actionable messages
    let hint = err.message;
    if (err.code === 'ENOTFOUND') hint = `DNS resolution failed for ${cfg.host} — check network`;
    else if (err.code === 'ECONNREFUSED') hint = `port ${cfg.port} refused — broker down?`;
    else if (/ACCESS_REFUSED/i.test(err.message)) hint = `auth failed — check user/pass`;
    else if (/NOT_ALLOWED.*vhost/i.test(err.message)) hint = `vhost '${cfg.vhost}' missing or no access`;

    record('connectivity', 'FAIL', hint);
    await finalize(null);
    return;
  }

  // -------------------------------------------------------------------------
  // 2. Channel creation
  // -------------------------------------------------------------------------
  let ch: any = null;
  try {
    ch = await conn.createChannel();
    record('channel', 'OK', 'createChannel → ready');
  } catch (err: any) {
    record('channel', 'FAIL', `cannot create channel: ${err.message}`);
    await finalize(conn);
    return;
  }

  // -------------------------------------------------------------------------
  // 3. Exchange declare (idempotent — won't break existing)
  // -------------------------------------------------------------------------
  try {
    await ch.assertExchange(EXCHANGE_NAME, 'topic', { durable: true });
    record('exchange:analysis', 'OK', `exchange '${EXCHANGE_NAME}' (topic, durable) asserted`);
  } catch (err: any) {
    record('exchange:analysis', 'FAIL', `cannot assert exchange: ${err.message}`);
  }

  // -------------------------------------------------------------------------
  // 4. Test queue write privilege (create a temp queue, then delete)
  // -------------------------------------------------------------------------
  try {
    const tempQueue = `didi.verify.${Date.now()}.${Math.random().toString(36).slice(2, 8)}`;
    await ch.assertQueue(tempQueue, { durable: false, autoDelete: true, exclusive: true });
    await ch.deleteQueue(tempQueue);
    record('privileges:write', 'OK', 'can create + delete queues on vhost');
  } catch (err: any) {
    record('privileges:write', 'FAIL', `no write privilege on vhost '${cfg.vhost}': ${err.message}`);
  }

  // -------------------------------------------------------------------------
  // 5. Expected topology — check which queues already exist
  //    (All 31 queues are OK if workers have started; 0 is OK pre-cutover)
  // -------------------------------------------------------------------------
  const expectedQueues: string[] = [];
  for (const comp of ANALYSIS_COMPONENTS) {
    for (const plan of PLAN_TYPES) {
      expectedQueues.push(QUEUE.queueName(comp, plan));
    }
  }
  for (const plan of PLAN_TYPES) {
    expectedQueues.push(MEDIA_QUEUE.queueName(plan));
  }
  expectedQueues.push(QUEUE.RESULTS_QUEUE);

  let foundQueues = 0;
  for (const q of expectedQueues) {
    try {
      // checkQueue throws if queue doesn't exist (and poisons the channel!)
      // Use a new channel per check to avoid cascade failures
      const probeCh = await conn.createChannel();
      probeCh.on('error', () => { /* swallow */ });
      try {
        await probeCh.checkQueue(q);
        foundQueues++;
      } catch {
        // Queue doesn't exist yet — normal pre-cutover
      }
      try { await probeCh.close(); } catch { /* ignore */ }
    } catch {
      // Channel creation failed — broker issue
      break;
    }
  }

  if (foundQueues === expectedQueues.length) {
    record('topology', 'OK', `all ${expectedQueues.length} expected queues present`);
  } else if (foundQueues === 0) {
    record('topology', 'WARN', `0/${expectedQueues.length} queues — normal if workers not started yet (will auto-create)`);
  } else {
    record('topology', 'WARN', `${foundQueues}/${expectedQueues.length} queues present (partial — may be mid-cutover)`);
  }

  // -------------------------------------------------------------------------
  // 6. Cleanup
  // -------------------------------------------------------------------------
  try { await ch.close(); } catch { /* ignore */ }
  await finalize(conn);
}

async function finalize(conn: any) {
  if (conn) {
    try { await conn.close(); } catch { /* ignore */ }
  }

  console.log('');
  for (const r of results) {
    const badge =
      r.severity === 'OK' ? '  ✓ ' :
      r.severity === 'WARN' ? '  ⚠ ' : '  ✘ ';
    console.log(`${badge} [${r.severity.padEnd(4)}] ${r.name.padEnd(28)} ${r.detail}`);
  }

  console.log('');
  console.log('='.repeat(70));
  const passed = results.filter(r => r.severity === 'OK').length;
  console.log(`Results: ${passed} OK · ${warn} WARN · ${fatal} FAIL`);
  console.log('='.repeat(70));

  if (fatal > 0) {
    console.log(`\n✘ NOT READY — ${fatal} fatal issue(s).\n`);
    process.exit(1);
  }
  if (warn > 0) {
    console.log(`\n⚠ READY with ${warn} warning(s) — topology will auto-create when workers start.\n`);
    process.exit(0);  // warnings are expected pre-cutover
  }
  console.log(`\n✓ CLUSTER READY — safe to cutover.\n`);
  process.exit(0);
}

main().catch((err) => {
  console.error('Unhandled error:', err);
  process.exit(1);
});