import fs from "fs"; import fastify from "fastify"; import cookie from "@fastify/cookie"; import cors from "@fastify/cors"; import jwt from "@fastify/jwt"; import fastifyStatic from "@fastify/static"; import { migrate, db } from "./db.js"; import { log, enableLogPersistence, flushLogs } from "./logger.js"; import * as store from "./store.js"; import { createRegistry } from "./registry.js"; import { addApiAuthGuard, COOKIE } from "./src/api/auth.js"; import authPlugin from "./src/api/auth.js"; import usersPlugin from "./src/api/users.js"; import scriptsPluginFactory from "./src/api/scripts.js"; import workflowsPluginFactory from "./src/api/workflows.js"; import runsPlugin from "./src/api/runs.js"; import { queryRunsFromRequest } from "./src/api/run-query.js"; import dashboardPluginFactory from "./src/api/dashboard.js"; import secretsPlugin from "./src/api/secrets.js"; import kvPlugin from "./src/api/kv.js"; import variablesPlugin from "./src/api/variables.js"; import httpPagesPlugin from "./src/api/http-pages.js"; import httpAuthsPlugin from "./src/api/http-auths.js"; import { WEB_DIST } from "./paths.js"; import { resolveSecretsKeyMaterial } from "./secrets.js"; import { closeRedis, createWorkflowQueue, createWorkflowWorker, getRedisUrlForLog, } from "./workflow-queue.js"; import { purgeExpiredTrash } from "./workflow-trash.js"; import { getConfigGeneration, startHeartbeatLoop, subscribeReload, } from "./control-bus.js"; /** * @param {{ * role?: string, * migrate?: boolean, * serveStaticUi?: boolean, * }} [opts] */ export async function startApp(opts = {}) { const role = (opts.role || process.env.JFLOW_ROLE || "all").toLowerCase(); const shouldMigrate = opts.migrate ?? role === "all"; const serveStaticUi = opts.serveStaticUi ?? (role === "all" || role === "api"); if (shouldMigrate) { await migrate(); try { const purged = await purgeExpiredTrash(); if (purged > 0) { log.info({ purged }, "purged expired workflow trash"); } } catch (err) { log.warn({ err }, "workflow trash purge failed"); } } enableLogPersistence(); const jwtSecret = process.env.JFLOW_JWT_SECRET ?? (process.env.NODE_ENV === "production" ? "" : "jflow-dev-secret"); if (!jwtSecret) { log.error("JFLOW_JWT_SECRET is required in production"); process.exit(1); } try { resolveSecretsKeyMaterial(); } catch (err) { log.error(err instanceof Error ? err.message : String(err)); process.exit(1); } const runApi = role === "all" || role === "api"; const runWorker = role === "all" || role === "worker"; log.info( { redis: getRedisUrlForLog(), role, generation: getConfigGeneration(), migrate: shouldMigrate, }, "starting jerapah-flow", ); const workflowQueue = createWorkflowQueue(); try { await workflowQueue.waitUntilReady(); } catch (err) { log.error({ err, redis: getRedisUrlForLog() }, "failed to connect to Redis"); process.exit(1); } const server = fastify({ loggerInstance: log }); await server.register(cookie); await server.register(jwt, { secret: jwtSecret, cookie: { cookieName: COOKIE, signed: false, }, }); await server.register(cors, { origin: process.env.JFLOW_CORS_ORIGIN ?? "http://localhost:8500", credentials: true, }); server.decorate("authenticate", async function authenticate(req, reply) { try { await req.jwtVerify(); } catch { return reply.code(401).send({ error: "unauthorized" }); } }); server.decorate("requireAdmin", async function requireAdmin(req, reply) { if (req.user?.role !== "admin") { return reply.code(403).send({ error: "forbidden" }); } }); const registry = createRegistry(server, { queue: workflowQueue, // Cron + HTTP triggers enqueue jobs; only the API process may own them. enableTriggers: runApi, }); registry.registerWorkflows(); if (runApi) { registry.registerHttpTriggers(); registry.registerCronTriggers(); registry.registerPruneJob(); } /** @type {import("bullmq").Worker | null} */ let workflowWorker = null; if (runWorker) { workflowWorker = createWorkflowWorker(async (job) => { const data = /** @type {{ runId?: string, key?: string, depth?: number }} */ ( job.data ?? {} ); if (typeof data.runId !== "string" || typeof data.key !== "string") { throw new Error("invalid workflow job payload"); } const result = await registry.executeQueuedRun({ runId: data.runId, key: data.key, depth: data.depth ?? 0, }); if (result.status === "failed") { throw new Error(result.error || "workflow failed"); } return result; }); } if (runApi) { await server.register( async (api) => { addApiAuthGuard(api, server); await api.register(authPlugin); await api.register(usersPlugin); await api.register(secretsPlugin); await api.register(variablesPlugin); await api.register(kvPlugin); await api.register(httpPagesPlugin); await api.register(httpAuthsPlugin); await api.register(scriptsPluginFactory(registry)); await api.register(workflowsPluginFactory(registry)); await api.register(runsPlugin); await api.register(dashboardPluginFactory(registry)); }, { prefix: "/api" }, ); server.post( "/admin/workflows/reregister", { onRequest: [server.authenticate] }, async (_req, reply) => { registry.reregister(); try { const { publishReload } = await import("./control-bus.js"); await publishReload({ type: "workflows" }); } catch { // ignore } return reply.send({ message: "Workflows refreshed" }); }, ); server.get( "/admin/runs", { onRequest: [server.authenticate] }, async (req, reply) => { const q = /** @type {Record} */ (req.query ?? {}); return reply.send(await queryRunsFromRequest(q)); }, ); server.get( "/admin/runs/:id", { onRequest: [server.authenticate] }, async (req, reply) => { const { id } = /** @type {{ id: string }} */ (req.params); const run = await store.getRun(id); if (!run) { return reply.code(404).send({ error: "run not found" }); } return reply.send(run); }, ); if (serveStaticUi && fs.existsSync(WEB_DIST)) { 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"); }); } } const stopHeartbeat = startHeartbeatLoop(runApi && runWorker ? "all" : runApi ? "api" : "worker"); const reloadSub = await subscribeReload(async () => { log.info("received reload signal"); registry.reregister(); }); async function shutdown() { try { stopHeartbeat(); await reloadSub.stop(); if (workflowWorker) { await workflowWorker.close(); } await workflowQueue.close(); await closeRedis(); await flushLogs(); await db.destroy(); } catch (err) { log.error({ err }, "shutdown error"); } process.exit(0); } process.on("SIGINT", shutdown); process.on("SIGTERM", shutdown); const port = Number(process.env.PORT ?? 8700); if (runApi) { try { await server.listen({ host: "0.0.0.0", port }); log.info(`Server is running on port ${port}`); } catch (err) { log.error({ err }, "failed to start server"); process.exit(1); } } else { log.info("worker-only mode; HTTP server not started"); } return { role, server, workflowQueue, workflowWorker, registry, shutdown, }; }