/** * 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); });