cfw-autumn/server/src/init.ts

import cluster from "node:cluster";
import http from "node:http";
import os from "node:os";
import { getRequestListener } from "@hono/node-server";
import { client, clientCritical, clientReplica, db } from "./db/initDrizzle.js";
import {
	initPgHealthMonitor,
	shutdownPgHealthMonitor,
} from "./db/pgHealthMonitor.js";
import { startPgPoolMonitor, stopPgPoolMonitor } from "./db/pgPoolMonitor.js";
import { getRedactedDatabaseUrls } from "./db/redactDatabaseUrl.js";
import { createLogger } from "./external/logtail/logtailUtils.js";
import { initDrizzleModules } from "./db/initDrizzle.js";
import {
	startAllEdgeConfigPolling,
	stopAllEdgeConfigPolling,
} from "./internal/misc/edgeConfig/edgeConfigRegistry.js";

// Edge config modules self-register on import
import "./internal/misc/requestBlocks/requestBlockStore.js";
import "./internal/misc/rollouts/rolloutConfigStore.js";
import "./internal/misc/featureFlags/featureFlagStore.js";
import "./internal/misc/customerBlocks/customerBlockStore.js";
import "./internal/misc/edgeConfig/orgLimitsStore.js";
import "./internal/misc/stripeSync/stripeSyncStore.js";
import "./internal/misc/redisV2Cache/redisV2CacheStore.js";
import "./internal/misc/cacheV2Ramp/cacheV2RampStore.js";
import "./internal/misc/jobQueues/jobQueueStore.js";
import { closeStripeSyncEngine } from "@autumn/stripe-sync";
import {
	startRedisMonitor,
	stopRedisMonitor,
	warmupRegionalRedis,
} from "./external/redis/initRedis.js";
import { primeRedisMonitor } from "./external/redis/initUtils/redisAvailability.js";
import {
	primeRedisV2Monitor,
	startRedisV2Monitor,
	stopRedisV2Monitor,
} from "./external/redis/initUtils/redisV2Availability.js";
import { preWarmOrgRedisConnections } from "./external/redis/orgRedisPool.js";
import { createHonoApp } from "./initHono.js";
import { initTelemetry, otelSdk } from "./instrumentation.js";
import { initSentry } from "./sentry.js";
import { configureTrigger } from "./trigger/configureTrigger.js";
import { checkEnvVars } from "./utils/initUtils.js";
import { startMemoryMonitor } from "./utils/memoryMonitor.js";

let shuttingDown = false;

const init = async ({
	startupStartedAt,
	env,
}: {
	startupStartedAt: number;
	env: Env;
}) => {
	const logger = createLogger(env);

	logger.info(getRedactedDatabaseUrls(env), "DB URLs");

	const app = createHonoApp(env);

	initPgHealthMonitor({ client: clientCritical, env });
	startPgPoolMonitor(env);

	void warmupRegionalRedis().catch((error) => {
		logger.warn("[Redis] Warmup failed", { error });
	});
	void preWarmOrgRedisConnections({ db }).catch((error) => {
		logger.warn("[OrgRedis] Warmup failed", { error });
	});

	await startAllEdgeConfigPolling({ logger });
	await Promise.all([primeRedisMonitor(), primeRedisV2Monitor()]);
	startRedisMonitor();
	startRedisV2Monitor();

	const PORT = env.SERVER_PORT ? Number.parseInt(env.SERVER_PORT) : 8080;

	const requestListener = getRequestListener(app.fetch);
	const server = http.createServer(requestListener);

	server.keepAliveTimeout = 120000;
	server.headersTimeout = 120000;

	await new Promise<void>((resolve) => {
		server.listen(PORT, "0.0.0.0", () => {
			const startupDurationMs = Date.now() - startupStartedAt;
			console.log(
				`Server running on port ${PORT} (${startupDurationMs}ms startup)`,
			);
			startMemoryMonitor("server", env, 60_000);
			resolve();
		});
	});
};

export const startNodeServer = async (env: Env) => {
	const logger = createLogger(env);

	initDrizzleModules(env);
	initSentry(env);
	initTelemetry(env);
	configureTrigger(env);
	checkEnvVars(env);

	if (env.NODE_ENV === "development") {
		registerFatalErrorHandlers(logger);
		await init({ startupStartedAt: Date.now(), env });
		registerShutdownHandlers(logger);

		return;
	}

	const numCPUs = os.cpus().length;

	if (cluster.isPrimary) {
		console.log(`Master ${process.pid} is running`);
		console.log("Number of CPUs", numCPUs);

		const numWorkers = 3;

		for (let i = 0; i < numWorkers; i++) {
			cluster.fork();
		}

		cluster.on("exit", (worker, code, signal) => {
			logger.error("WORKER DIED", {
				pid: worker.process.pid,
				code,
				signal,
				exitedAfterDisconnect: worker.exitedAfterDisconnect,
			});
			if (shuttingDown) return;
			cluster.fork();
		});

		registerShutdownHandlers(logger);
	} else {
		registerFatalErrorHandlers(logger);
		await init({ startupStartedAt: Date.now(), env });
		registerShutdownHandlers(logger);
	}
};

function registerFatalErrorHandlers(logger: ReturnType<typeof createLogger>) {
	const exitAfterLog = () => setTimeout(() => process.exit(1), 100);
	const logFatal = (event: string, error: unknown) => {
		logger.error(event, {
			error:
				error instanceof Error
					? { name: error.name, message: error.message, stack: error.stack }
					: error,
		});
	};

	process.on("uncaughtException", (error) => {
		logFatal("WORKER FATAL uncaughtException", error);
		exitAfterLog();
	});
	process.on("unhandledRejection", (reason) => {
		logFatal("WORKER FATAL unhandledRejection", reason);
		exitAfterLog();
	});
}

function registerShutdownHandlers(logger: ReturnType<typeof createLogger>) {
	process.on("SIGTERM", () => gracefulShutdown(logger));
	process.on("SIGINT", () => gracefulShutdown(logger));
	// Do NOT use process.on("exit", ...) for async cleanup!
}

async function gracefulShutdown(logger: ReturnType<typeof createLogger>) {
	shuttingDown = true;
	console.log("Shutting down worker, flushing telemetry and closing DB...");
	try {
		// Flush any buffered OTel spans before shutting down
		if (otelSdk) {
			await otelSdk.shutdown();
		}
		shutdownPgHealthMonitor();
		stopPgPoolMonitor();
		stopRedisMonitor();
		stopRedisV2Monitor();
		stopAllEdgeConfigPolling();
		await Promise.all([
			client.end(),
			clientCritical.end(),
			clientReplica?.end(),
			closeStripeSyncEngine(),
		]);
		console.log("Shutdown complete. Exiting process.");
		process.exit(0);
	} catch (err) {
		console.error("Error during graceful shutdown:", err);
		process.exit(1);
	}
}