triggerdotdev--trigger.dev/apps/webapp/server.ts

import "./sentry.server";

import { createRequestHandler } from "@remix-run/express";
import { broadcastDevReady, logDevReady } from "@remix-run/server-runtime";
import compression from "compression";
import type { Server as EngineServer } from "engine.io";
import express, { type RequestHandler } from "express";
import morgan from "morgan";
import { nanoid } from "nanoid";
import path from "path";
import type { Server as IoServer } from "socket.io";
import type { WebSocketServer } from "ws";
import type { RateLimitMiddleware } from "~/services/apiRateLimit.server";
import { type RunWithHttpContextFunction } from "~/services/httpAsyncStorage.server";
import cluster from "node:cluster";
import os from "node:os";

const ENABLE_CLUSTER = process.env.ENABLE_CLUSTER === "1";
const cpuCount = os.availableParallelism();
const WORKERS =
  Number.parseInt(process.env.WEB_CONCURRENCY || process.env.CLUSTER_WORKERS || "", 10) || cpuCount;
// Must be greater than the upstream load balancer's idle timeout to avoid the
// LB pipelining a request onto a connection Node has already closed (→ 502).
const HTTP_KEEPALIVE_TIMEOUT_MS =
  Number.parseInt(process.env.HTTP_KEEPALIVE_TIMEOUT_MS || "", 10) || 65 * 1000;

function forkWorkers() {
  for (let i = 0; i < WORKERS; i++) {
    cluster.fork();
  }
}

function installPrimarySignalHandlers() {
  let didHandleSigterm = false;
  let didHandleSigint = false;
  let didGracefulExit = false;

  const forward = (signal: NodeJS.Signals) => {
    for (const id in cluster.workers) {
      const w = cluster.workers[id];
      if (w?.process?.pid) {
        try {
          process.kill(w.process.pid, signal);
        } catch {}
      }
    }
  };

  const gracefulExit = () => {
    if (didGracefulExit) return;
    didGracefulExit = true;

    const timeoutMs = Number(process.env.GRACEFUL_SHUTDOWN_TIMEOUT || 30_000);
    // wait for workers to exit, then exit the primary too
    const maybeExit = () => {
      const alive = Object.values(cluster.workers || {}).some((w) => w && !w.isDead());
      if (!alive) process.exit(0);
    };
    setInterval(maybeExit, 1000);
    setTimeout(() => process.exit(0), timeoutMs);
  };

  process.on("SIGTERM", () => {
    if (didHandleSigterm) return;
    didHandleSigterm = true;
    forward("SIGTERM");
    gracefulExit();
  });
  process.on("SIGINT", () => {
    if (didHandleSigint) return;
    didHandleSigint = true;
    forward("SIGINT");
    gracefulExit();
  });
}

if (ENABLE_CLUSTER && cluster.isPrimary) {
  process.title = `node webapp-server primary`;
  console.log(`[cluster] Primary ${process.pid} is starting with ${WORKERS} workers`);
  forkWorkers();

  cluster.on("exit", (worker, code, signal) => {
    const intentional =
      // If we sent "shutdown", the worker will exit with code 0 after closing.
      code === 0 || worker.exitedAfterDisconnect;
    console.log(
      `[cluster] worker ${worker.process.pid} exited (code=${code}, signal=${signal}, intentional=${intentional})`
    );
    // If it wasn't during a shutdown, replace the worker.
    if (!intentional) cluster.fork();
  });

  installPrimarySignalHandlers();
} else {
  const app = express();

  if (process.env.DISABLE_COMPRESSION !== "1") {
    app.use(compression());
  }

  // http://expressjs.com/en/advanced/best-practice-security.html#at-a-minimum-disable-x-powered-by-header
  app.disable("x-powered-by");

  // Remix fingerprints its assets so we can cache forever.
  app.use("/build", express.static("public/build", { immutable: true, maxAge: "1y" }));
  // Stale dev builds can request an old hashed manifest; don't fall through to Remix.
  app.use("/build", (_req, res) => {
    res.status(404).end();
  });

  // Everything else (like favicon.ico) is cached for an hour. You may want to be
  // more aggressive with this caching.
  app.use(express.static("public", { maxAge: "1h" }));

  // On high-volume machine-ingest services (e.g. otel) the per-request access
  // log dominates log volume. HTTP_ACCESS_LOG_DISABLED suppresses successful
  // (2xx) access logs; non-2xx responses are always logged so errors stay visible.
  const suppressSuccessfulAccessLogs = process.env.HTTP_ACCESS_LOG_DISABLED === "1";
  app.use(
    morgan("tiny", {
      skip: (_req, res) =>
        suppressSuccessfulAccessLogs && res.statusCode >= 200 && res.statusCode < 300,
    })
  );

  process.title = ENABLE_CLUSTER
    ? `node webapp-worker-${cluster.isWorker ? cluster.worker?.id : "solo"}`
    : "node webapp-server";

  const MODE = process.env.NODE_ENV;
  const BUILD_DIR = path.join(process.cwd(), "build");
  const build = require(BUILD_DIR);

  const port = process.env.REMIX_APP_PORT || process.env.PORT || 3000;

  if (process.env.HTTP_SERVER_DISABLED !== "true") {
    // Back-compat shim: a previously-deployed client build polls this endpoint after a
    // /build asset 404 and reloads once it reports a newer build id, letting those older
    // tabs recover in a single reload. Temporary — safe to remove once older clients have
    // churned out. Deliberately does NOT set an X-Build-Id response header.
    app.get("/build-version", (_req, res) => {
      res.set("Cache-Control", "no-store");
      res.json({ version: build.assets.version });
    });

    const socketIo: { io: IoServer } | undefined = build.entry.module.socketIo;
    const wss: WebSocketServer | undefined = build.entry.module.wss;
    const apiRateLimiter: RateLimitMiddleware = build.entry.module.apiRateLimiter;
    const engineRateLimiter: RateLimitMiddleware = build.entry.module.engineRateLimiter;
    const runWithHttpContext: RunWithHttpContextFunction = build.entry.module.runWithHttpContext;
    const tenantContextMiddleware: RequestHandler = build.entry.module.tenantContextMiddleware;

    app.use((req, res, next) => {
      // helpful headers:
      res.set("Strict-Transport-Security", `max-age=${60 * 60 * 24 * 365 * 100}`);

      // Add X-Robots-Tag header for test-cloud.trigger.dev
      if (req.hostname !== "cloud.trigger.dev") {
        res.set("X-Robots-Tag", "noindex, nofollow");
      }

      // /clean-urls/ -> /clean-urls. Skip /ph: PostHog ingest endpoints end in
      // a slash, and a 301 would drop sendBeacon POSTs.
      if (req.path.endsWith("/") && req.path.length > 1 && !req.path.startsWith("/ph/")) {
        const query = req.url.slice(req.path.length);
        const safepath = req.path.slice(0, -1).replace(/\/+/g, "/");
        res.redirect(301, safepath + query);
        return;
      }
      next();
    });

    app.use((req, res, next) => {
      // Generate a unique request ID for each request
      const requestId = nanoid();
      const abortController = new AbortController();
      res.on("close", () => abortController.abort());

      runWithHttpContext(
        { requestId, path: req.url, host: req.hostname, method: req.method, abortController },
        next
      );
    });

    if (process.env.DASHBOARD_AND_API_DISABLED !== "true") {
      if (process.env.ALLOW_ONLY_REALTIME_API === "true") {
        // Block all requests that do not start with /realtime
        app.use((req, res, next) => {
          // Make sure /healthcheck is still accessible
          if (!req.url.startsWith("/realtime") && req.url !== "/healthcheck") {
            res.status(404).send("Not Found");
            return;
          }

          next();
        });
      }

      app.use(apiRateLimiter);
      app.use(engineRateLimiter);

      app.use(tenantContextMiddleware);

      app.all(
        "*",
        // @ts-ignore
        createRequestHandler({
          build,
          mode: MODE,
        })
      );
    } else {
      // we need to do the health check here at /healthcheck — forward
      // to the Remix handler so the loader's readiness checks (DB ping,
      // REQUIRE_PLUGINS-gated plugin load) run in this mode too. A
      // static 200 here would silently mask a failed plugin load.
      app.get(
        "/healthcheck",
        // @ts-ignore
        createRequestHandler({
          build,
          mode: MODE,
        })
      );
    }

    const server = app.listen(port, () => {
      console.log(
        `✅ server ready: http://localhost:${port} [NODE_ENV: ${MODE}]${
          ENABLE_CLUSTER && cluster.isWorker ? ` [worker ${cluster.worker?.id}/${process.pid}]` : ""
        }`
      );

      if (MODE === "development") {
        broadcastDevReady(build)
          .then(() => logDevReady(build))
          .catch(console.error);
      }
    });

    server.keepAliveTimeout = HTTP_KEEPALIVE_TIMEOUT_MS;
    // Mitigate against https://github.com/triggerdotdev/trigger.dev/security/dependabot/128
    // by not allowing 2000+ headers to be sent and causing a DoS
    // headers will instead be limited by the maxHeaderSize
    server.maxHeadersCount = 0;

    let didCloseServer = false;

    function closeServer(signal: NodeJS.Signals) {
      if (didCloseServer) return;
      didCloseServer = true;

      server.close((err) => {
        if (err) {
          console.error("Error closing express server:", err);
        } else {
          console.log("Express server closed gracefully.");
        }
      });
    }

    process.on("SIGTERM", closeServer);
    process.on("SIGINT", closeServer);

    socketIo?.io.attach(server);
    server.removeAllListeners("upgrade"); // prevent duplicate upgrades from listeners created by io.attach()

    server.on("upgrade", async (req, socket, head) => {
      console.log(`Attemping to upgrade connection at url ${req.url}`);

      socket.on("error", (err) => {
        console.error("Connection upgrade error:", err);
      });

      const url = new URL(req.url ?? "", "http://localhost");

      // Upgrade socket.io connection
      if (url.pathname.startsWith("/socket.io/")) {
        console.log(`Socket.io client connected, upgrading their connection...`);

        // https://github.com/socketio/socket.io/issues/4693
        (socketIo!.io.engine as EngineServer).handleUpgrade(req, socket, head);
        return;
      }

      // Only upgrade the connecting if the path is `/ws`
      if (url.pathname !== "/ws") {
        // Setting the socket.destroy() error param causes an error event to be emitted which needs to be handled with socket.on("error") to prevent uncaught exceptions.
        socket.destroy(
          new Error(
            "Cannot connect because of invalid path: Please include `/ws` in the path of your upgrade request."
          )
        );
        return;
      }

      console.log(`Client connected, upgrading their connection...`);

      // Handle the WebSocket connection
      wss?.handleUpgrade(req, socket, head, (ws) => {
        wss?.emit("connection", ws, req);
      });
    });
  } else {
    require(BUILD_DIR);
    console.log(`✅ app ready (skipping http server)`);
  }
}