#!/usr/bin/env node /** * Health HTTP + optional NATS subscribe. * Fleet probes /health. Live NATS is off unless NATS_URL is set. */ import http from 'node:http'; import { SUBJECTS } from './subjects.js'; import { handle } from './handle.js'; const port = Number(process.env.FLEET_HEALTH_PORT || process.env.PORT || 13900); const bind = process.env.FLEET_HEALTH_BIND || '127.0.0.1'; const instance = process.env.FLEET_INSTANCE || 'verae-nats-process-0'; let paused = false; let processed = 0; function snapshot() { return { ok: !paused, paused, instance, role: 'verae-nats-process', subjects: { in: SUBJECTS.IN, out: SUBJECTS.OUT, queue: SUBJECTS.QUEUE }, processed, }; } const server = http.createServer(async (req, res) => { const url = new URL(req.url, 'http://127.0.0.1'); const json = (code, obj) => { res.writeHead(code, { 'content-type': 'application/json' }); res.end(JSON.stringify(obj)); }; if (url.pathname === '/health' && req.method === 'GET') { const s = snapshot(); return json(s.ok ? 200 : 503, s); } if (url.pathname === '/pause' && req.method === 'POST') { paused = true; return json(200, snapshot()); } if (url.pathname === '/resume' && req.method === 'POST') { paused = false; return json(200, snapshot()); } if (url.pathname === '/message' && req.method === 'POST') { const chunks = []; for await (const c of req) chunks.push(c); let body = {}; try { body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'); } catch { body = {}; } if (paused) return json(503, snapshot()); try { const out = await handle(body); processed += 1; return json(200, { rttMs: 0, result: out, replyTo: SUBJECTS.reply(out.correlationId) }); } catch (err) { return json(400, { error: err.message }); } } json(404, { error: 'not found' }); }); server.listen(port, bind, () => { process.stdout.write(`${instance} health http://${bind}:${port}/health in=${SUBJECTS.IN}\n`); }); process.on('SIGTERM', () => server.close(() => process.exit(0)));