Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dd98421b08 | ||
|
|
69312c9080 | ||
|
|
c8697532f9 | ||
|
|
be8122f9e5 | ||
|
|
0a8463d8f4 | ||
|
|
721f15e900 | ||
|
|
877ea9e3f8 | ||
|
|
93b51dcaaa | ||
|
|
eb7c318c13 | ||
|
|
864427b45c | ||
|
|
46aa8ca327 |
@@ -0,0 +1,33 @@
|
||||
name: Deploy to Raspberry Pi
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [dev]
|
||||
|
||||
jobs:
|
||||
deploy:
|
||||
runs-on: home # must match a label on your act_runner
|
||||
steps:
|
||||
- name: Deploy
|
||||
run: |
|
||||
set -euo pipefail
|
||||
|
||||
# act_runner uses bash --noprofile --norc; load nvm/pnpm explicitly
|
||||
export NVM_DIR="/home/nsrb/.nvm"
|
||||
export PNPM_HOME="/home/nsrb/.local/share/pnpm"
|
||||
# shellcheck disable=SC1091
|
||||
[ -s "$NVM_DIR/nvm.sh" ] && . "$NVM_DIR/nvm.sh"
|
||||
export PATH="$PNPM_HOME/bin:$PATH"
|
||||
|
||||
APP_DIR=/home/nsrb/apps/jerapah-flow
|
||||
cd "$APP_DIR"
|
||||
|
||||
git fetch origin dev
|
||||
git checkout dev
|
||||
git pull --ff-only origin dev
|
||||
|
||||
pnpm install --frozen-lockfile
|
||||
pnpm build
|
||||
|
||||
pm2 startOrReload ecosystem.config.cjs --update-env
|
||||
pm2 save
|
||||
@@ -105,8 +105,9 @@ Optional `script.meta.reads = "ctx"` documents expression hosts. `meta.input` /
|
||||
| `pnpm dev:server` | Monolith API/runner only |
|
||||
| `pnpm dev:web` | UI only (proxies `/api` → :8700, `/ops` → :8600) |
|
||||
| `pnpm build` | Production UI build |
|
||||
| `pnpm start` | Monolith: API + worker + built UI |
|
||||
| `pnpm start` | Monolith: API + worker + built UI (serves `dist` on :8700) |
|
||||
| `pnpm start:control` | Control plane only (migrates, manages PM2 children) |
|
||||
| `pnpm start:web` | Production UI on :8500 (`dist` + proxies to control/HTTP) |
|
||||
| `pnpm start:api` | HTTP API + cron enqueue (`JFLOW_ROLE=api`) |
|
||||
| `pnpm start:worker` | BullMQ worker only |
|
||||
| `pnpm migrate` | Apply SQLite migrations |
|
||||
@@ -140,24 +141,45 @@ Desired state is stored in `packages/server/data/control-state.json` (generation
|
||||
| `JFLOW_ROLE` | `all` | `all` (HTTP + cron + worker), `api`, or `worker`. Prefer `pnpm start:api` / `start:worker` under control. |
|
||||
| `JFLOW_CONFIG_GENERATION` | `1` | Set by control/PM2 so children report config generation in heartbeats. |
|
||||
| `JFLOW_CONTROL_PORT` | `8600` | Control ops API port. |
|
||||
| `JFLOW_UI_PORT` | `8500` | Production UI server (`web-server.js`) port. |
|
||||
| `JFLOW_HTTP_PORT` | `8700` | HTTP API port (PM2 children / UI proxy target). |
|
||||
| `JFLOW_LOG_LEVEL` | `debug` | Pino level |
|
||||
| `JFLOW_RETENTION_DAYS` | `30` | Run history prune |
|
||||
| `JFLOW_CORS_ORIGIN` | `http://localhost:8500` | Vite origin in dev |
|
||||
| `PORT` | `8700` | HTTP API port |
|
||||
| `NODE_ENV` | — | Set `production` for secure cookies |
|
||||
| `JFLOW_CORS_ORIGIN` | `http://localhost:8500` | Browser origin (Vite in dev, UI server in prod) |
|
||||
| `PORT` | `8700` | HTTP API port (alias; prefer `JFLOW_HTTP_PORT` under control) |
|
||||
| `NODE_ENV` | — | Set `production` for secure cookies (unless overridden) |
|
||||
| `COOKIE_SECURE` | (from `NODE_ENV`) | `true`/`false` — force Secure cookie flag. Use `false` for plain HTTP LAN access (`http://192.168.x.x`) |
|
||||
|
||||
Workflow runs are **queued** via BullMQ. HTTP and manual triggers return `202 { runId, status: "queued" }` immediately; poll `GET /api/runs/:id` for progress (`queued` → `running` → `success` \| `failed`). Cron remains an in-process producer that enqueues jobs on each tick.
|
||||
|
||||
## Production
|
||||
|
||||
Control-plane topology (same ports as `pnpm dev:pm2`):
|
||||
|
||||
| Process | Port | Role |
|
||||
|---|---|---|
|
||||
| `jflow-web` | **8500** | Built UI + proxies `/api` → :8700, `/ops` + `/api/auth` → :8600 |
|
||||
| `jflow-control` | **8600** | Migrations, Ops API, starts/stops PM2 HTTP + workers |
|
||||
| `jflow-http` | **8700** | API + cron enqueue (managed by control) |
|
||||
| `jflow-worker` | — | BullMQ workers (managed by control) |
|
||||
|
||||
```bash
|
||||
pnpm install
|
||||
pnpm build
|
||||
# Redis must be reachable at REDIS_URL (set REDIS_PASS if Redis requires AUTH)
|
||||
# Recommended: run control (migrates + manages PM2 HTTP/workers)
|
||||
JFLOW_JWT_SECRET=... JFLOW_SECRETS_KEY=... REDIS_URL=redis://127.0.0.1:6379 REDIS_PASS=... NODE_ENV=production pnpm start:control
|
||||
# Or monolith (dev-style):
|
||||
# ... pnpm start
|
||||
# Put secrets in .env (JFLOW_JWT_SECRET, JFLOW_SECRETS_KEY, REDIS_URL, …)
|
||||
# Use in-tree PM2 6.x (same module control.js requires). A global `pm2` 7.x
|
||||
# against a 6.x daemon pegs CPU even when ls shows only 2 fork instances.
|
||||
pnpm start:pm2
|
||||
# UI: http://localhost:8500
|
||||
# If you already mixed versions: pnpm pm2 -- kill && pnpm start:pm2
|
||||
```
|
||||
|
||||
With control, serve the built UI from Vite preview, a reverse proxy, or set `JFLOW_SERVE_UI=1` on the HTTP process.
|
||||
Or without the ecosystem file:
|
||||
|
||||
```bash
|
||||
NODE_ENV=production pnpm start:control # :8600 + PM2 children
|
||||
NODE_ENV=production pnpm start:web # :8500
|
||||
```
|
||||
|
||||
Monolith (no Ops stop/scale): `pnpm build && pnpm start` serves the UI from the API process on :8700. Optional `JFLOW_SERVE_UI=1` on `start:api` does the same when you run HTTP alone — do **not** use that under control-plane mode (stopping HTTP would take down the UI).
|
||||
|
||||
+35
-8
@@ -1,7 +1,12 @@
|
||||
/**
|
||||
* Production PM2 ecosystem (monolith runner).
|
||||
* Use for deployed/single-process starts. For local multi-process Ops UI, use
|
||||
* `pnpm dev:pm2` → packages/server/ecosystem.dev.cjs / control.js instead.
|
||||
* Production PM2 ecosystem (control plane + UI).
|
||||
* Starts always-on processes only; HTTP (:8700) and workers are owned by
|
||||
* control.js via PM2 (same as `pnpm dev:pm2`).
|
||||
*
|
||||
* Use `pnpm start:pm2` (in-tree PM2 6.x). Do not use a global `pm2` 7.x —
|
||||
* a CLI/daemon version mismatch pegs CPU even with instances: 1.
|
||||
*
|
||||
* Prerequisites: `pnpm build` (packages/web/dist), Redis, .env secrets.
|
||||
*/
|
||||
const fs = require("fs");
|
||||
const path = require("path");
|
||||
@@ -28,20 +33,42 @@ function loadEnv(file) {
|
||||
}
|
||||
|
||||
const root = __dirname;
|
||||
const env = {
|
||||
NODE_ENV: "production",
|
||||
...loadEnv(path.join(root, ".env")),
|
||||
};
|
||||
|
||||
module.exports = {
|
||||
apps: [
|
||||
{
|
||||
name: "jerapah-flow",
|
||||
name: "jflow-control",
|
||||
cwd: root,
|
||||
script: "packages/server/runner.js",
|
||||
interpreter: "node",
|
||||
script: "packages/server/control.js",
|
||||
interpreter: process.execPath,
|
||||
instances: 1,
|
||||
exec_mode: "fork",
|
||||
autorestart: true,
|
||||
max_restarts: 8,
|
||||
restart_delay: 2000,
|
||||
env: {
|
||||
...env,
|
||||
JFLOW_CONTROL_PORT: env.JFLOW_CONTROL_PORT ?? "8600",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "jflow-web",
|
||||
cwd: root,
|
||||
script: "packages/server/web-server.js",
|
||||
interpreter: process.execPath,
|
||||
instances: 1,
|
||||
exec_mode: "fork",
|
||||
autorestart: true,
|
||||
max_restarts: 20,
|
||||
env: {
|
||||
NODE_ENV: "production",
|
||||
...loadEnv(path.join(root, ".env")),
|
||||
...env,
|
||||
JFLOW_UI_PORT: env.JFLOW_UI_PORT ?? "8500",
|
||||
JFLOW_CONTROL_PORT: env.JFLOW_CONTROL_PORT ?? "8600",
|
||||
JFLOW_HTTP_PORT: env.JFLOW_HTTP_PORT ?? "8700",
|
||||
},
|
||||
},
|
||||
],
|
||||
|
||||
@@ -14,6 +14,9 @@
|
||||
"start:api": "pnpm --filter @jerapah-flow/server start:api",
|
||||
"start:worker": "pnpm --filter @jerapah-flow/server start:worker",
|
||||
"start:control": "pnpm --filter @jerapah-flow/server start:control",
|
||||
"start:web": "pnpm --filter @jerapah-flow/server start:web",
|
||||
"pm2": "node scripts/pm2.mjs",
|
||||
"start:pm2": "node scripts/pm2.mjs start ecosystem.config.cjs",
|
||||
"migrate": "pnpm --filter @jerapah-flow/server migrate",
|
||||
"test": "pnpm --filter @jerapah-flow/shared test && pnpm --filter @jerapah-flow/web test",
|
||||
"lint": "pnpm --filter @jerapah-flow/web lint"
|
||||
|
||||
+21
-18
@@ -64,13 +64,6 @@ try {
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
try {
|
||||
await connectPm2();
|
||||
} catch (err) {
|
||||
log.error({ err }, "failed to connect to PM2 — is pm2 installed?");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
async function applyDesiredState() {
|
||||
const state = readControlState();
|
||||
await ensureHttp({
|
||||
@@ -98,8 +91,6 @@ async function applyDesiredState() {
|
||||
);
|
||||
}
|
||||
|
||||
await applyDesiredState();
|
||||
|
||||
const server = fastify({ loggerInstance: log });
|
||||
await server.register(cookie);
|
||||
await server.register(jwt, {
|
||||
@@ -484,12 +475,24 @@ process.on("SIGINT", shutdown);
|
||||
process.on("SIGTERM", shutdown);
|
||||
|
||||
const port = Number(process.env.JFLOW_CONTROL_PORT ?? process.env.PORT ?? 8600);
|
||||
server
|
||||
.listen({ host: "0.0.0.0", port })
|
||||
.then(() => {
|
||||
log.info(`Control is running on port ${port}`);
|
||||
})
|
||||
.catch((err) => {
|
||||
log.error({ err }, "failed to start control");
|
||||
process.exit(1);
|
||||
});
|
||||
try {
|
||||
await server.listen({ host: "0.0.0.0", port });
|
||||
log.info(`Control is running on port ${port}`);
|
||||
} catch (err) {
|
||||
log.error({ err }, "failed to start control");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
try {
|
||||
await connectPm2();
|
||||
} catch (err) {
|
||||
log.error({ err }, "failed to connect to PM2 — is pm2 installed?");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
try {
|
||||
await applyDesiredState();
|
||||
} catch (err) {
|
||||
log.error({ err }, "failed to apply desired state");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
const HOP_BY_HOP = new Set([
|
||||
"connection",
|
||||
"keep-alive",
|
||||
"proxy-authenticate",
|
||||
"proxy-authorization",
|
||||
"te",
|
||||
"trailer",
|
||||
"transfer-encoding",
|
||||
"upgrade",
|
||||
"host",
|
||||
"content-length",
|
||||
]);
|
||||
|
||||
const REPLY_SKIP = new Set([
|
||||
"connection",
|
||||
"keep-alive",
|
||||
"transfer-encoding",
|
||||
"content-encoding",
|
||||
"content-length",
|
||||
]);
|
||||
|
||||
/**
|
||||
* Control origin used when the UI or HTTP process proxies to control.
|
||||
* @returns {string}
|
||||
*/
|
||||
export function controlOrigin() {
|
||||
const explicit = process.env.JFLOW_CONTROL_URL?.trim();
|
||||
if (explicit) return explicit.replace(/\/$/, "");
|
||||
const port = Number(process.env.JFLOW_CONTROL_PORT ?? 8600);
|
||||
return `http://127.0.0.1:${port}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* HTTP API origin (workflow triggers + REST).
|
||||
* @returns {string}
|
||||
*/
|
||||
export function httpOrigin() {
|
||||
const explicit = process.env.JFLOW_HTTP_URL?.trim();
|
||||
if (explicit) return explicit.replace(/\/$/, "");
|
||||
const port = Number(process.env.JFLOW_HTTP_PORT ?? process.env.PORT ?? 8700);
|
||||
return `http://127.0.0.1:${port}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Forward the incoming request to `origin`, preserving path + query.
|
||||
* @param {import("fastify").FastifyRequest} req
|
||||
* @param {import("fastify").FastifyReply} reply
|
||||
* @param {string} origin
|
||||
* @param {{ unreachableMessage?: string }} [opts]
|
||||
*/
|
||||
export async function proxyToOrigin(req, reply, origin, opts = {}) {
|
||||
const target = `${origin.replace(/\/$/, "")}${req.raw.url ?? "/"}`;
|
||||
const headers = {};
|
||||
for (const [key, value] of Object.entries(req.headers)) {
|
||||
if (value == null || HOP_BY_HOP.has(key.toLowerCase())) continue;
|
||||
headers[key] = Array.isArray(value) ? value.join(", ") : String(value);
|
||||
}
|
||||
|
||||
const method = req.method.toUpperCase();
|
||||
const hasBody = method !== "GET" && method !== "HEAD";
|
||||
let body;
|
||||
if (hasBody) {
|
||||
if (Buffer.isBuffer(req.body)) body = req.body;
|
||||
else if (typeof req.body === "string") body = req.body;
|
||||
else if (req.body != null) {
|
||||
body = JSON.stringify(req.body);
|
||||
if (!headers["content-type"]) headers["content-type"] = "application/json";
|
||||
}
|
||||
}
|
||||
|
||||
let res;
|
||||
try {
|
||||
res = await fetch(target, { method, headers, body });
|
||||
} catch (err) {
|
||||
const message = opts.unreachableMessage ?? "upstream unreachable";
|
||||
req.log.warn({ err, target }, `proxy: ${message}`);
|
||||
return reply.code(502).send({ error: message });
|
||||
}
|
||||
|
||||
reply.code(res.status);
|
||||
res.headers.forEach((value, key) => {
|
||||
if (REPLY_SKIP.has(key.toLowerCase())) return;
|
||||
reply.header(key, value);
|
||||
});
|
||||
return reply.send(Buffer.from(await res.arrayBuffer()));
|
||||
}
|
||||
|
||||
/**
|
||||
* Forward `/ops/*` to the control plane (same-origin UI in production).
|
||||
* @param {import("fastify").FastifyRequest} req
|
||||
* @param {import("fastify").FastifyReply} reply
|
||||
*/
|
||||
export async function proxyOpsToControl(req, reply) {
|
||||
return proxyToOrigin(req, reply, controlOrigin(), {
|
||||
unreachableMessage: "control plane unreachable",
|
||||
});
|
||||
}
|
||||
@@ -11,6 +11,7 @@
|
||||
"start:api": "node server.js",
|
||||
"start:worker": "node worker.js",
|
||||
"start:control": "node control.js",
|
||||
"start:web": "node web-server.js",
|
||||
"migrate": "node -e \"import('./db.js').then((m) => m.migrate().then(() => process.exit(0)))\"",
|
||||
"test:plugins": "JFLOW_PLUGINS_DIR=./data/plugins-smoke-test JFLOW_DB_PATH=./data/plugins-smoke.db node test/plugins-smoke.js",
|
||||
"test:workflow-history": "JFLOW_WORKFLOWS_DIR=./data/workflow-history-smoke JFLOW_DB_PATH=./data/workflow-history-smoke.db node test/workflow-history-smoke.js",
|
||||
@@ -41,7 +42,7 @@
|
||||
"nodemailer": "^9.0.5",
|
||||
"pino": "^10.3.1",
|
||||
"pino-roll": "^4.0.0",
|
||||
"pm2": "^6.0.13",
|
||||
"pm2": "6.0.14",
|
||||
"rss-parser": "^3.13.0",
|
||||
"ssh2-sftp-client": "^12.1.1",
|
||||
"webdav": "^5.10.0",
|
||||
|
||||
@@ -164,13 +164,57 @@ export async function restartPm2Process(pmId) {
|
||||
return { name: proc.name, pmId: id };
|
||||
}
|
||||
|
||||
/**
|
||||
* PM2 injects these into process.env of a managed app. Spreading them into
|
||||
* `pm2.start({ env })` overwrites `name` / `pm_exec_path` so God restarts
|
||||
* jflow-control instead of launching http/worker (EADDRINUSE :8600 loop).
|
||||
*/
|
||||
const PM2_META_KEYS = new Set([
|
||||
"name",
|
||||
"namespace",
|
||||
"exec_mode",
|
||||
"exec_interpreter",
|
||||
"instances",
|
||||
"instance_var",
|
||||
"node_app_instance",
|
||||
"unique_id",
|
||||
"status",
|
||||
"username",
|
||||
"windowsHide",
|
||||
"merge_logs",
|
||||
"vizion",
|
||||
"vizion_running",
|
||||
"autostart",
|
||||
"autorestart",
|
||||
"automation",
|
||||
"km_link",
|
||||
]);
|
||||
|
||||
/**
|
||||
* @param {NodeJS.ProcessEnv} env
|
||||
* @returns {NodeJS.ProcessEnv}
|
||||
*/
|
||||
export function withoutPm2Meta(env) {
|
||||
/** @type {NodeJS.ProcessEnv} */
|
||||
const out = {};
|
||||
for (const [key, val] of Object.entries(env)) {
|
||||
if (val == null) continue;
|
||||
if (PM2_META_KEYS.has(key)) continue;
|
||||
if (key.startsWith("pm_") || key.startsWith("axm_") || key.startsWith("PM2_")) {
|
||||
continue;
|
||||
}
|
||||
out[key] = val;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared env for child processes.
|
||||
* @param {{ generation: number }} opts
|
||||
*/
|
||||
export function childEnv(opts) {
|
||||
return {
|
||||
...process.env,
|
||||
...withoutPm2Meta(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",
|
||||
|
||||
@@ -36,7 +36,9 @@ import {
|
||||
resolveFailureTriggerConfig,
|
||||
} from "./trigger-failure.js";
|
||||
import { enqueueWorkflowJob } from "./workflow-queue.js";
|
||||
import { ensureInitialRevision } from "./workflow-history.js";
|
||||
import { ensureInitialRevision, recordRevision } from "./workflow-history.js";
|
||||
import { workflowIdFromFile } from "./workflow-normalize.js";
|
||||
import { publishReload } from "./control-bus.js";
|
||||
|
||||
/**
|
||||
* @typedef {{ owner: string, file: string, workflow: any }} WorkflowEntry
|
||||
@@ -553,6 +555,45 @@ export function createRegistry(server, opts = {}) {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist `enabled: false` for a workflow and reload registries across processes.
|
||||
* @param {string} owner
|
||||
* @param {string} file
|
||||
* @param {string} key
|
||||
*/
|
||||
async function disableWorkflowForConsecutiveFailures(owner, file, key) {
|
||||
const content = fsStore.readWorkflowYaml(owner, file);
|
||||
if (content == null) {
|
||||
throw new Error(`workflow file missing for ${key}`);
|
||||
}
|
||||
const doc = yaml.parseDocument(content);
|
||||
if (doc.errors?.length) {
|
||||
throw new Error(doc.errors[0]?.message ?? "invalid yaml");
|
||||
}
|
||||
const parsed = doc.toJSON();
|
||||
if (parsed?.enabled === false) {
|
||||
log.debug({ workflow: key }, "workflow already disabled");
|
||||
return;
|
||||
}
|
||||
doc.set("enabled", false);
|
||||
const nextContent = String(doc);
|
||||
fsStore.writeWorkflowYaml(owner, file, nextContent);
|
||||
await recordRevision({
|
||||
workflowId: workflowIdFromFile(file),
|
||||
owner,
|
||||
file,
|
||||
content: nextContent,
|
||||
reason: "disable-on-consecutive-failures",
|
||||
});
|
||||
reregister();
|
||||
try {
|
||||
await publishReload({ type: "workflows" });
|
||||
} catch {
|
||||
// Redis may be briefly unavailable; local reload already applied.
|
||||
}
|
||||
log.warn({ workflow: key }, "disabled workflow after consecutive failures");
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {{
|
||||
* key: string,
|
||||
@@ -589,6 +630,17 @@ export function createRegistry(server, opts = {}) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (failureConfig.disableOnConsecutiveFailures) {
|
||||
const entry = workflows.get(opts.key);
|
||||
if (entry) {
|
||||
await disableWorkflowForConsecutiveFailures(entry.owner, entry.file, opts.key);
|
||||
} else {
|
||||
log.warn({ workflow: opts.key }, "cannot disable missing workflow entry");
|
||||
}
|
||||
}
|
||||
|
||||
if (!failureConfig.workflowName) return;
|
||||
|
||||
const destKey = resolveWorkflowTriggerKey(opts.owner, failureConfig.workflowName);
|
||||
const alertData = buildFailureAlertData({
|
||||
sourceKey: opts.key,
|
||||
|
||||
@@ -76,6 +76,7 @@ function triggerSummary(owner, workflow, nameById) {
|
||||
schedule: t?.schedule ?? null,
|
||||
onConsecutiveFailures: t?.onConsecutiveFailures ?? null,
|
||||
onFailureWorkflow: t?.onFailureWorkflow ?? null,
|
||||
disableOnConsecutiveFailures: t?.disableOnConsecutiveFailures === true,
|
||||
auth: isHttp ? authLabel(t?.auth, nameById) : null,
|
||||
};
|
||||
});
|
||||
|
||||
@@ -47,12 +47,18 @@ export function resolveFailureTriggerConfig(workflow, owner, runtimeTrigger) {
|
||||
|
||||
const threshold = Number(spec.onConsecutiveFailures);
|
||||
const workflowName = onFailureWorkflowName(spec);
|
||||
if (!Number.isFinite(threshold) || threshold < 1 || workflowName.length === 0) {
|
||||
const disableOnConsecutiveFailures = isDisableOnConsecutiveFailures(spec);
|
||||
if (
|
||||
!Number.isFinite(threshold) ||
|
||||
threshold < 1 ||
|
||||
(workflowName.length === 0 && !disableOnConsecutiveFailures)
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
threshold: Math.floor(threshold),
|
||||
workflowName,
|
||||
workflowName: workflowName.length > 0 ? workflowName : null,
|
||||
disableOnConsecutiveFailures,
|
||||
};
|
||||
}
|
||||
return null;
|
||||
@@ -66,6 +72,13 @@ function onFailureWorkflowName(trigger) {
|
||||
return typeof value === "string" ? value.trim() : "";
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {Record<string, unknown>} trigger
|
||||
*/
|
||||
export function isDisableOnConsecutiveFailures(trigger) {
|
||||
return trigger?.disableOnConsecutiveFailures === true;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {unknown} workflow
|
||||
*/
|
||||
@@ -81,12 +94,13 @@ export async function validateWorkflowFailureTriggers(workflow) {
|
||||
const hasThreshold =
|
||||
trigger.onConsecutiveFailures != null && trigger.onConsecutiveFailures !== "";
|
||||
const hasWorkflow = onFailureWorkflowName(trigger).length > 0;
|
||||
const hasDisable = isDisableOnConsecutiveFailures(trigger);
|
||||
|
||||
if (!hasThreshold && !hasWorkflow) continue;
|
||||
if (!hasThreshold && !hasWorkflow && !hasDisable) continue;
|
||||
|
||||
if (!hasThreshold || !hasWorkflow) {
|
||||
if (!hasThreshold || (!hasWorkflow && !hasDisable)) {
|
||||
const err = new Error(
|
||||
"onConsecutiveFailures and onFailureWorkflow must both be set on a trigger",
|
||||
"onConsecutiveFailures requires onFailureWorkflow and/or disableOnConsecutiveFailures",
|
||||
);
|
||||
err.statusCode = 400;
|
||||
throw err;
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
/**
|
||||
* Production UI server (:8500).
|
||||
* Serves packages/web/dist and proxies /api, /ops, /admin, /u like Vite in dev.
|
||||
* Always-on — survives Ops stop of jflow-http.
|
||||
*/
|
||||
import fs from "fs";
|
||||
import fastify from "fastify";
|
||||
import fastifyStatic from "@fastify/static";
|
||||
import { log } from "./logger.js";
|
||||
import { WEB_DIST } from "./paths.js";
|
||||
import {
|
||||
controlOrigin,
|
||||
httpOrigin,
|
||||
proxyToOrigin,
|
||||
} from "./ops-proxy.js";
|
||||
|
||||
if (!fs.existsSync(WEB_DIST)) {
|
||||
log.error(
|
||||
{ WEB_DIST },
|
||||
"web dist missing — run `pnpm build` before starting the UI server",
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const port = Number(process.env.JFLOW_UI_PORT ?? 8500);
|
||||
const control = controlOrigin();
|
||||
const http = httpOrigin();
|
||||
|
||||
const server = fastify({ loggerInstance: log });
|
||||
|
||||
/**
|
||||
* @param {import("fastify").FastifyRequest} req
|
||||
* @param {import("fastify").FastifyReply} reply
|
||||
*/
|
||||
async function proxyApi(req, reply) {
|
||||
const url = req.raw.url ?? "";
|
||||
// Match Vite: /api/auth → control (login works when HTTP is stopped).
|
||||
if (url === "/api/auth" || url.startsWith("/api/auth/") || url.startsWith("/api/auth?")) {
|
||||
return proxyToOrigin(req, reply, control, {
|
||||
unreachableMessage: "control plane unreachable",
|
||||
});
|
||||
}
|
||||
return proxyToOrigin(req, reply, http, {
|
||||
unreachableMessage: "HTTP API unreachable",
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {import("fastify").FastifyRequest} req
|
||||
* @param {import("fastify").FastifyReply} reply
|
||||
*/
|
||||
async function proxyOps(req, reply) {
|
||||
return proxyToOrigin(req, reply, control, {
|
||||
unreachableMessage: "control plane unreachable",
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {import("fastify").FastifyRequest} req
|
||||
* @param {import("fastify").FastifyReply} reply
|
||||
*/
|
||||
async function proxyHttp(req, reply) {
|
||||
return proxyToOrigin(req, reply, http, {
|
||||
unreachableMessage: "HTTP API unreachable",
|
||||
});
|
||||
}
|
||||
|
||||
server.all("/api", proxyApi);
|
||||
server.all("/api/*", proxyApi);
|
||||
server.all("/ops", proxyOps);
|
||||
server.all("/ops/*", proxyOps);
|
||||
server.all("/admin", proxyHttp);
|
||||
server.all("/admin/*", proxyHttp);
|
||||
server.all("/u", proxyHttp);
|
||||
server.all("/u/*", proxyHttp);
|
||||
|
||||
await server.register(fastifyStatic, {
|
||||
root: WEB_DIST,
|
||||
wildcard: false,
|
||||
});
|
||||
|
||||
server.setNotFoundHandler((req, reply) => {
|
||||
const url = req.raw.url ?? "";
|
||||
if (
|
||||
url.startsWith("/api") ||
|
||||
url.startsWith("/u/") ||
|
||||
url.startsWith("/admin") ||
|
||||
url.startsWith("/ops")
|
||||
) {
|
||||
return reply.code(404).send({ error: "not found" });
|
||||
}
|
||||
return reply.sendFile("index.html");
|
||||
});
|
||||
|
||||
async function shutdown() {
|
||||
try {
|
||||
await server.close();
|
||||
} catch (err) {
|
||||
log.error({ err }, "web-server shutdown error");
|
||||
}
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
process.on("SIGINT", shutdown);
|
||||
process.on("SIGTERM", shutdown);
|
||||
|
||||
try {
|
||||
await server.listen({ host: "0.0.0.0", port });
|
||||
log.info(
|
||||
{ port, WEB_DIST, control, http },
|
||||
"UI server listening (static + proxy)",
|
||||
);
|
||||
} catch (err) {
|
||||
log.error({ err }, "failed to start UI server");
|
||||
process.exit(1);
|
||||
}
|
||||
@@ -143,10 +143,15 @@ function TriggerCardView({
|
||||
|
||||
function triggerSummary(trigger, owner) {
|
||||
const type = trigger?.type;
|
||||
const failure =
|
||||
trigger.onConsecutiveFailures && trigger.onFailureWorkflow
|
||||
? ` · onFailure@${trigger.onFailureWorkflow}`
|
||||
: "";
|
||||
/** @type {string[]} */
|
||||
const failureParts = [];
|
||||
if (trigger.onConsecutiveFailures && trigger.onFailureWorkflow) {
|
||||
failureParts.push(`onFailure@${trigger.onFailureWorkflow}`);
|
||||
}
|
||||
if (trigger.disableOnConsecutiveFailures) {
|
||||
failureParts.push("auto-disable");
|
||||
}
|
||||
const failure = failureParts.length ? ` · ${failureParts.join(", ")}` : "";
|
||||
if (type === "HTTP") {
|
||||
return `${trigger.method || "POST"} ${namespacedPath(owner || "owner", trigger.path || "/")}${failure}`;
|
||||
}
|
||||
@@ -383,6 +388,20 @@ function FailureAlertFields({ trigger, disabled, onChange, alertDestinations })
|
||||
) : null}
|
||||
</FormSelect>
|
||||
</Field>
|
||||
<label className="label cursor-pointer justify-start gap-3 py-0">
|
||||
<input
|
||||
type="checkbox"
|
||||
className="checkbox checkbox-sm"
|
||||
checked={Boolean(trigger.disableOnConsecutiveFailures)}
|
||||
disabled={disabled}
|
||||
onChange={(e) =>
|
||||
onChange({ ...trigger, disableOnConsecutiveFailures: e.target.checked })
|
||||
}
|
||||
/>
|
||||
<span className="label-text">
|
||||
Disable this workflow when the consecutive failure threshold is reached
|
||||
</span>
|
||||
</label>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -171,6 +171,7 @@ export function newHttpTrigger() {
|
||||
unauthorized: null,
|
||||
onConsecutiveFailures: "",
|
||||
onFailureWorkflow: "",
|
||||
disableOnConsecutiveFailures: false,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -186,6 +187,7 @@ export function newCronTrigger() {
|
||||
unauthorized: null,
|
||||
onConsecutiveFailures: "",
|
||||
onFailureWorkflow: "",
|
||||
disableOnConsecutiveFailures: false,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -201,6 +203,7 @@ export function newWorkflowTrigger() {
|
||||
unauthorized: null,
|
||||
onConsecutiveFailures: "",
|
||||
onFailureWorkflow: "",
|
||||
disableOnConsecutiveFailures: false,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -303,9 +306,16 @@ function normalizeTrigger(raw) {
|
||||
"unauthorized",
|
||||
"onConsecutiveFailures",
|
||||
"onFailureWorkflow",
|
||||
"disableOnConsecutiveFailures",
|
||||
])
|
||||
: type === "cron"
|
||||
? new Set(["type", "schedule", "onConsecutiveFailures", "onFailureWorkflow"])
|
||||
? new Set([
|
||||
"type",
|
||||
"schedule",
|
||||
"onConsecutiveFailures",
|
||||
"onFailureWorkflow",
|
||||
"disableOnConsecutiveFailures",
|
||||
])
|
||||
: new Set(["type"]);
|
||||
/** @type {Record<string, unknown>} */
|
||||
const extra = {};
|
||||
@@ -323,6 +333,7 @@ function normalizeTrigger(raw) {
|
||||
? ""
|
||||
: String(raw.onConsecutiveFailures),
|
||||
onFailureWorkflow: readOnFailureWorkflow(raw),
|
||||
disableOnConsecutiveFailures: raw.disableOnConsecutiveFailures === true,
|
||||
auth: Array.isArray(raw.auth) ? raw.auth : null,
|
||||
response: typeof raw.response === "string" ? raw.response : "",
|
||||
unauthorized: raw.unauthorized ?? null,
|
||||
@@ -377,6 +388,9 @@ function dumpFailureTriggerFields(t, out) {
|
||||
if (typeof t.onFailureWorkflow === "string" && t.onFailureWorkflow.trim()) {
|
||||
out.onFailureWorkflow = t.onFailureWorkflow.trim();
|
||||
}
|
||||
if (t.disableOnConsecutiveFailures === true) {
|
||||
out.disableOnConsecutiveFailures = true;
|
||||
}
|
||||
}
|
||||
|
||||
function dumpTrigger(t) {
|
||||
|
||||
Generated
+1
-1
@@ -77,7 +77,7 @@ importers:
|
||||
specifier: ^4.0.0
|
||||
version: 4.0.0
|
||||
pm2:
|
||||
specifier: ^6.0.13
|
||||
specifier: 6.0.14
|
||||
version: 6.0.14(supports-color@7.2.0)
|
||||
rss-parser:
|
||||
specifier: ^3.13.0
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
#!/usr/bin/env node
|
||||
/**
|
||||
* Run the same PM2 binary control.js `require("pm2")` uses.
|
||||
* A global `pm2` 7.x talking to an in-memory 6.x daemon pegs CPU on start.
|
||||
*/
|
||||
import { spawn, spawnSync } from "node:child_process";
|
||||
import { createRequire } from "node:module";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..");
|
||||
const require = createRequire(
|
||||
path.join(root, "packages/server/package.json"),
|
||||
);
|
||||
const pm2Root = path.dirname(require.resolve("pm2/package.json"));
|
||||
const pm2Bin = path.join(pm2Root, "bin/pm2");
|
||||
const args = process.argv.slice(2);
|
||||
|
||||
function run(pm2Args, opts = {}) {
|
||||
return spawnSync(process.execPath, [pm2Bin, ...pm2Args], {
|
||||
cwd: root,
|
||||
encoding: "utf8",
|
||||
...opts,
|
||||
});
|
||||
}
|
||||
|
||||
const cmd = args[0];
|
||||
if (cmd === "start" || cmd === "restart" || cmd === "reload") {
|
||||
const probe = run(["ls"], { stdio: ["ignore", "pipe", "pipe"] });
|
||||
const text = `${probe.stdout ?? ""}${probe.stderr ?? ""}`;
|
||||
const mem = text.match(/In memory PM2 version:\s*(\S+)/);
|
||||
const loc = text.match(/Local PM2 version:\s*(\S+)/);
|
||||
if (mem && loc && mem[1] !== loc[1]) {
|
||||
console.error(
|
||||
`[jflow] PM2 daemon ${mem[1]} != CLI ${loc[1]}. Killing the daemon so control.js and the CLI share one version.`,
|
||||
);
|
||||
run(["kill"], { stdio: "inherit" });
|
||||
}
|
||||
}
|
||||
|
||||
const child = spawn(process.execPath, [pm2Bin, ...args], {
|
||||
cwd: root,
|
||||
stdio: "inherit",
|
||||
});
|
||||
child.on("exit", (code, signal) => {
|
||||
if (signal) process.kill(process.pid, signal);
|
||||
process.exit(code ?? 1);
|
||||
});
|
||||
Reference in New Issue
Block a user