import path from "path"; import pm2 from "pm2"; import { SERVER_ROOT } from "./paths.js"; const REPO_ROOT = path.resolve(SERVER_ROOT, "../.."); export const PM2_HTTP_NAME = "jflow-http"; export const PM2_WORKER_NAME = "jflow-worker"; /** * @returns {Promise} */ export function connectPm2() { return new Promise((resolve, reject) => { pm2.connect((err) => { if (err) reject(err); else resolve(); }); }); } export function disconnectPm2() { try { pm2.disconnect(); } catch { // ignore } } /** * @returns {Promise} */ export function listPm2() { return new Promise((resolve, reject) => { pm2.list((err, list) => { if (err) reject(err); else resolve(list ?? []); }); }); } /** * @param {string} name * @returns {Promise} */ export async function listPm2ByName(name) { const list = await listPm2(); return list.filter((p) => p.name === name); } /** * @param {object} app * @returns {Promise} */ function startPm2App(app) { return new Promise((resolve, reject) => { pm2.start(app, (err) => { if (err) reject(err); else resolve(); }); }); } /** * @param {string} name * @returns {Promise} */ export function stopPm2App(name) { return new Promise((resolve, reject) => { pm2.stop(name, (err) => { if (err) { const msg = err instanceof Error ? err.message : String(err); if (/not found|doesn't exist|process or namespace/i.test(msg)) { resolve(); return; } reject(err); return; } resolve(); }); }); } /** * @param {string} name * @returns {Promise} */ export function deletePm2App(name) { return new Promise((resolve, reject) => { pm2.delete(name, (err) => { if (err) { const msg = err instanceof Error ? err.message : String(err); if (/not found|doesn't exist|process or namespace/i.test(msg)) { resolve(); return; } reject(err); return; } resolve(); }); }); } /** * @param {string} name * @param {number} instances * @returns {Promise} */ export function scalePm2App(name, instances) { return new Promise((resolve, reject) => { pm2.scale(name, instances, (err) => { if (err) reject(err); else resolve(); }); }); } /** * @param {string} name * @returns {Promise} */ export function restartPm2App(name) { return new Promise((resolve, reject) => { pm2.restart(name, (err) => { if (err) reject(err); else resolve(); }); }); } /** * Restart a single PM2 process by id. Only jflow-http / jflow-worker. * @param {number} pmId * @returns {Promise<{ name: string, pmId: number }>} */ export async function restartPm2Process(pmId) { const id = Math.floor(Number(pmId)); if (!Number.isFinite(id) || id < 0) { const err = new Error("invalid pmId"); err.code = "BAD_REQUEST"; throw err; } const list = await listPm2(); const proc = list.find((p) => Number(p.pm_id) === id); if (!proc) { const err = new Error("process not found"); err.code = "NOT_FOUND"; throw err; } if (proc.name !== PM2_HTTP_NAME && proc.name !== PM2_WORKER_NAME) { const err = new Error("process is not a JerapahFlow child"); err.code = "FORBIDDEN"; throw err; } await new Promise((resolve, reject) => { pm2.restart(id, (err) => { if (err) reject(err); else resolve(); }); }); return { name: proc.name, pmId: id }; } /** * Shared env for child processes. * @param {{ generation: number }} opts */ export function childEnv(opts) { return { ...process.env, JFLOW_CONFIG_GENERATION: String(opts.generation), JFLOW_CORS_ORIGIN: process.env.JFLOW_CORS_ORIGIN ?? "http://localhost:8500", PORT: process.env.JFLOW_HTTP_PORT ?? "8700", }; } /** * Ensure HTTP app exists and matches desired running/stopped state. * @param {{ generation: number, running: boolean }} opts */ export async function ensureHttp(opts) { const existing = await listPm2ByName(PM2_HTTP_NAME); if (!opts.running) { if (existing.length) await stopPm2App(PM2_HTTP_NAME); return; } if (existing.length === 0) { await startPm2App({ name: PM2_HTTP_NAME, script: path.join(SERVER_ROOT, "server.js"), cwd: REPO_ROOT, instances: 1, exec_mode: "fork", autorestart: true, max_restarts: 20, env: { ...childEnv(opts), JFLOW_ROLE: "api", }, }); return; } const online = existing.some((p) => p.pm2_env?.status === "online"); if (!online) { await restartPm2App(PM2_HTTP_NAME); } } /** * Ensure worker app has `count` online forks (0 = stopped/deleted). * @param {{ generation: number, count: number }} opts */ export async function ensureWorkers(opts) { const count = Math.max(0, Math.min(32, Math.floor(opts.count))); const existing = await listPm2ByName(PM2_WORKER_NAME); if (count === 0) { if (existing.length) { await stopPm2App(PM2_WORKER_NAME); await deletePm2App(PM2_WORKER_NAME); } return; } if (existing.length === 0) { await startPm2App({ name: PM2_WORKER_NAME, script: path.join(SERVER_ROOT, "worker.js"), cwd: REPO_ROOT, instances: count, exec_mode: "fork", autorestart: true, max_restarts: 50, env: { ...childEnv(opts), JFLOW_ROLE: "worker", }, }); return; } const current = existing.length; if (current !== count) { await scalePm2App(PM2_WORKER_NAME, count); } const stopped = existing.filter((p) => p.pm2_env?.status !== "online"); if (stopped.length) { await restartPm2App(PM2_WORKER_NAME); } } /** * Hard recycle HTTP + workers with updated generation env. * Children must be stopped first for a clean migrate window. * * @param {{ generation: number, http: boolean, workers: number }} opts */ export async function recreateChildren(opts) { await stopPm2App(PM2_HTTP_NAME); await deletePm2App(PM2_HTTP_NAME); await stopPm2App(PM2_WORKER_NAME); await deletePm2App(PM2_WORKER_NAME); if (opts.http) { await ensureHttp({ generation: opts.generation, running: true }); } if (opts.workers > 0) { await ensureWorkers({ generation: opts.generation, count: opts.workers }); } } /** * Summarize PM2 process status for the ops UI. */ export async function describeChildren() { const list = await listPm2(); const http = list.filter((p) => p.name === PM2_HTTP_NAME); const workers = list.filter((p) => p.name === PM2_WORKER_NAME); const mapOne = (p) => ({ name: p.name, pmId: Number(p.pm_id), status: p.pm2_env?.status ?? "unknown", pid: p.pid ?? null, restarts: p.pm2_env?.restart_time ?? 0, uptime: p.pm2_env?.pm_uptime ?? null, generation: Number(p.pm2_env?.JFLOW_CONFIG_GENERATION ?? 0) || null, memory: Number(p.monit?.memory) || 0, cpu: Number(p.monit?.cpu) || 0, }); const httpMapped = http.map(mapOne); const workersMapped = workers.map(mapOne); const all = [...httpMapped, ...workersMapped]; return { http: httpMapped, workers: workersMapped, httpOnline: http.some((p) => p.pm2_env?.status === "online"), workerOnlineCount: workers.filter((p) => p.pm2_env?.status === "online").length, totals: { memory: all.reduce((sum, p) => sum + p.memory, 0), cpu: all.reduce((sum, p) => sum + p.cpu, 0), }, }; } export function getRepoRoot() { return REPO_ROOT; }