pulsenode/server.ts

import { createServer } from "node:http";
import next from "next";
import { attachWebSocketServer } from "./lib/ws/server";
import { configStore } from "./lib/config/loader";
import { effectiveConfigStore } from "./lib/config/effective";
import { collectorScheduler } from "./lib/collectors/scheduler";

const port = Number(process.env.PORT ?? 3000);
const hostname = process.env.HOSTNAME ?? "0.0.0.0";
const dev = process.env.NODE_ENV !== "production";

const app = next({ dev, hostname, port });
const handle = app.getRequestHandler();

async function main(): Promise<void> {
  await app.prepare();
  const upgradeHandler = app.getUpgradeHandler();

  configStore.load();
  configStore.watch();
  await effectiveConfigStore.start();
  collectorScheduler.start(effectiveConfigStore.get());
  effectiveConfigStore.onUpdate((config) => collectorScheduler.reconcile(config));

  const httpServer = createServer((req, res) => {
    handle(req, res);
  });

  const wss = attachWebSocketServer(httpServer);

  httpServer.on("upgrade", (req, socket, head) => {
    const pathname = new URL(req.url ?? "/", "http://internal").pathname;
    if (pathname === "/ws") {
      wss.handleUpgrade(req, socket, head, (ws) => {
        wss.emit("connection", ws, req);
      });
      return;
    }
    void upgradeHandler(req, socket, head);
  });

  function shutdown(): void {
    collectorScheduler.stop();
    effectiveConfigStore.stop();
    configStore.stop();
    httpServer.close(() => process.exit(0));
    setTimeout(() => process.exit(0), 5_000).unref();
  }

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

  httpServer.listen(port, () => {
    console.log(`> PulseNode listening on http://${hostname}:${port}`);
  });
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});