Initial import of verae-fleet from zapier monorepo

src/health.js

@@ -0,0 +1,118 @@
+/**
+ * Probe a replica: HTTP /health or TCP connect.
+ * @module health
+ */
+
+import http from 'node:http';
+import net from 'node:net';
+import fs from 'node:fs';
+import path from 'node:path';
+
+/**
+ * @param {number|{ host?: string, port: number, path?: string, timeoutMs?: number }} portOrOpts
+ */
+export function httpProbe(portOrOpts, pathname = '/health', timeoutMs = 800) {
+  const opts =
+    typeof portOrOpts === 'object'
+      ? portOrOpts
+      : { port: portOrOpts, path: pathname, timeoutMs };
+  const host = opts.host || '127.0.0.1';
+  const port = opts.port;
+  const pathName = opts.path || pathname;
+  const tmo = opts.timeoutMs || timeoutMs;
+  return new Promise((resolve) => {
+    const req = http.get(
+      { host, port, path: pathName, timeout: tmo },
+      (res) => {
+        const chunks = [];
+        res.on('data', (c) => chunks.push(c));
+        res.on('end', () => {
+          let body = {};
+          try {
+            body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}');
+          } catch {
+            body = {};
+          }
+          resolve({
+            ok: res.statusCode >= 200 && res.statusCode < 300 && body.ok !== false,
+            statusCode: res.statusCode,
+            paused: Boolean(body.paused),
+            body,
+          });
+        });
+      },
+    );
+    req.on('error', (err) => resolve({ ok: false, error: err.message, paused: false }));
+    req.on('timeout', () => {
+      req.destroy();
+      resolve({ ok: false, error: 'timeout', paused: false });
+    });
+  });
+}
+
+export function tcpProbe(host, port, timeoutMs = 400) {
+  return new Promise((resolve) => {
+    const sock = net.connect({ host, port });
+    const done = (ok, error) => {
+      try {
+        sock.destroy();
+      } catch {
+        /* ignore */
+      }
+      resolve({ ok, error, paused: false });
+    };
+    sock.setTimeout(timeoutMs);
+    sock.on('connect', () => done(true));
+    sock.on('error', (err) => done(false, err.message));
+    sock.on('timeout', () => done(false, 'timeout'));
+  });
+}
+
+/**
+ * @param {number|{ host?: string, port: number, path?: string, body?: object }} portOrOpts
+ * @param {string} [pathname]
+ */
+export function postControl(portOrOpts, pathname = '/') {
+  const opts = typeof portOrOpts === 'object' ? portOrOpts : { port: portOrOpts, path: pathname };
+  const host = opts.host || '127.0.0.1';
+  const port = opts.port;
+  const pathName = opts.path || pathname;
+  const body = opts.body ? JSON.stringify(opts.body) : '';
+  return new Promise((resolve, reject) => {
+    const req = http.request(
+      {
+        host,
+        port,
+        path: pathName,
+        method: 'POST',
+        timeout: opts.timeoutMs || 2000,
+        headers: body ? { 'content-type': 'application/json', 'content-length': Buffer.byteLength(body) } : {},
+      },
+      (res) => {
+        const chunks = [];
+        res.on('data', (c) => chunks.push(c));
+        res.on('end', () => {
+          try {
+            resolve(JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'));
+          } catch {
+            resolve({});
+          }
+        });
+      },
+    );
+    req.on('error', reject);
+    if (body) req.write(body);
+    req.end();
+  });
+}
+
+export function heartbeatFresh(stateDir, maxAgeMs) {
+  const file = path.join(stateDir, 'heartbeat.json');
+  try {
+    const st = fs.statSync(file);
+    const body = JSON.parse(fs.readFileSync(file, 'utf8'));
+    return { ok: Date.now() - st.mtimeMs < maxAgeMs && body.ok !== false, body, ageMs: Date.now() - st.mtimeMs };
+  } catch (err) {
+    return { ok: false, error: err.message };
+  }
+}