EpicNext-Cms/scripts/jobs-worker.ts

import "./load-env";
import { Cron } from "croner";
import { and, eq, lt, lte, sql } from "drizzle-orm";
import { env } from "../src/env";
import {
	db,
	PasswordReset,
	WebsiteArticles,
	WebsiteLoginLogs,
} from "../src/lib/db";
import { logger } from "../src/lib/logger";
import { redis } from "../src/lib/redis";
import { emulatorOffline, healthDegraded } from "../src/lib/services/alert";
import { rcon } from "../src/lib/services/rcon";

function captureWorkerError(err: unknown, context: string): void {
	logger.error(context, {
		module: "jobs",
		err: err instanceof Error ? err.message : String(err),
	});
}

/** In-process cooldown so a flapping probe does not spam Discord/email. */
const alertCooldownMs = (env.HEALTH_ALERT_COOLDOWN_MIN ?? 15) * 60_000;
const lastHealthAlertAt = new Map<string, number>();

function canAlert(key: string): boolean {
	const now = Date.now();
	const prev = lastHealthAlertAt.get(key) ?? 0;
	if (now - prev < alertCooldownMs) return false;
	lastHealthAlertAt.set(key, now);
	return true;
}

async function probeHealth(): Promise<{
	database: boolean;
	redis: boolean | null;
	emulator: boolean;
}> {
	const database = await db
		.execute(sql`SELECT 1`)
		.then(() => true)
		.catch(() => false);

	let redisOk: boolean | null = null;
	if (env.REDIS_URL) {
		if (!redis) {
			redisOk = false;
		} else {
			try {
				redisOk = (await redis.ping()) === "PONG";
			} catch {
				redisOk = false;
			}
		}
	}

	const emulator = await rcon.send("ping", null).catch(() => false);
	return {
		database,
		redis: redisOk,
		emulator: Boolean(emulator),
	};
}

async function checkOpsHealth(): Promise<void> {
	try {
		const health = await probeHealth();
		const degraded =
			!health.database || health.redis === false || !health.emulator;
		if (!degraded) return;

		if (!health.emulator && health.database && health.redis !== false) {
			if (canAlert("emulator")) {
				await emulatorOffline("jobs-worker RCON ping failed");
			}
			return;
		}

		if (canAlert("health")) {
			await healthDegraded(health);
		}
	} catch (err) {
		captureWorkerError(err, "Health probe failed");
	}
}

async function backupEmulatorJar(): Promise<void> {
	if (!env.EMULATOR_JAR_PATH || !env.EMULATOR_BACKUP_DIR) return;

	const { copyFileSync, mkdirSync, readdirSync, unlinkSync, existsSync } =
		await import("node:fs");
	const { resolve } = await import("node:path");

	const timestamp = new Date().toISOString().slice(0, 19).replace(/[T:]/g, "-");
	const backupFile = resolve(
		env.EMULATOR_BACKUP_DIR,
		`emulator-${timestamp}.jar`,
	);

	if (!existsSync(env.EMULATOR_BACKUP_DIR)) {
		mkdirSync(env.EMULATOR_BACKUP_DIR, { recursive: true });
	}

	try {
		copyFileSync(env.EMULATOR_JAR_PATH, backupFile);
		logger.info("Backed up emulator JAR", {
			module: "jobs",
			backupFile,
		});

		const keep = env.EMULATOR_BACKUP_KEEP ?? 7;
		const files = readdirSync(env.EMULATOR_BACKUP_DIR)
			.filter((f) => f.startsWith("emulator-") && f.endsWith(".jar"))
			.sort()
			.reverse();

		for (let i = keep; i < files.length; i++) {
			unlinkSync(resolve(env.EMULATOR_BACKUP_DIR, files[i]));
			logger.info("Rotated out old backup", {
				module: "jobs",
				file: files[i],
			});
		}
	} catch (err) {
		captureWorkerError(err, "JAR backup failed");
	}
}

/** Optional mysqldump when DB_BACKUP_DIR is set (host must have mysqldump on PATH). */
async function backupDatabase(): Promise<void> {
	const backupDir = env.DB_BACKUP_DIR;
	if (!backupDir || !env.DATABASE_URL) return;

	const { mkdirSync, readdirSync, unlinkSync, existsSync, createWriteStream } =
		await import("node:fs");
	const { resolve } = await import("node:path");
	const { spawn } = await import("node:child_process");

	let parsed: URL;
	try {
		parsed = new URL(env.DATABASE_URL);
	} catch {
		logger.error("Invalid DATABASE_URL for DB backup", { module: "jobs" });
		return;
	}

	if (!existsSync(backupDir)) {
		mkdirSync(backupDir, { recursive: true });
	}

	const timestamp = new Date().toISOString().slice(0, 19).replace(/[T:]/g, "-");
	const dbName =
		decodeURIComponent(parsed.pathname.replace(/^\//, "")) || "cms";
	const outFile = resolve(backupDir, `db-${dbName}-${timestamp}.sql`);

	const args = [
		`-h${parsed.hostname}`,
		`-P${parsed.port || "3306"}`,
		`-u${decodeURIComponent(parsed.username)}`,
		`--single-transaction`,
		`--routines`,
		`--databases`,
		dbName,
	];
	if (parsed.password) {
		args.splice(3, 0, `-p${decodeURIComponent(parsed.password)}`);
	}

	await new Promise<void>((resolvePromise) => {
		const child = spawn("mysqldump", args, {
			stdio: ["ignore", "pipe", "pipe"],
		});
		const out = createWriteStream(outFile);
		child.stdout.pipe(out);
		let stderr = "";
		child.stderr.on("data", (chunk: Buffer) => {
			stderr += chunk.toString();
		});
		child.on("error", (err) => {
			captureWorkerError(err, "mysqldump spawn failed (is it on PATH?)");
			resolvePromise();
		});
		child.on("close", (code) => {
			out.end();
			if (code !== 0) {
				captureWorkerError(
					new Error(stderr || `mysqldump exit ${code}`),
					"DB backup failed",
				);
			} else {
				logger.info("Backed up database", { module: "jobs", outFile });
				const keep = env.DB_BACKUP_KEEP ?? 7;
				const files = readdirSync(backupDir)
					.filter((f) => f.startsWith("db-") && f.endsWith(".sql"))
					.sort()
					.reverse();
				for (let i = keep; i < files.length; i++) {
					const file = files[i];
					if (file) unlinkSync(resolve(backupDir, file));
				}
			}
			resolvePromise();
		});
	});
}

async function cleanupOldLogs(): Promise<void> {
	try {
		const cutoff = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
		await db
			.delete(WebsiteLoginLogs)
			.where(lt(WebsiteLoginLogs.createdAt, cutoff));
		logger.info("Cleaned up login logs older than 30 days", { module: "jobs" });
	} catch (err) {
		captureWorkerError(err, "Log cleanup failed");
	}
}

async function cleanupOldSessions(): Promise<void> {
	try {
		const cutoff = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000);
		await db.delete(PasswordReset).where(lt(PasswordReset.createdAt, cutoff));
		logger.info("Cleaned up expired password reset tokens", {
			module: "jobs",
		});
	} catch (err) {
		captureWorkerError(err, "Session cleanup failed");
	}
}

async function publishScheduledArticles(): Promise<void> {
	try {
		const now = new Date();
		const result = await db
			.update(WebsiteArticles)
			.set({
				status: "published",
				publishedAt: now,
				updatedAt: now,
			})
			.where(
				and(
					eq(WebsiteArticles.status, "scheduled"),
					lte(WebsiteArticles.publishAt, now),
				),
			);
		const info = result as unknown as {
			affectedRows?: number;
			rowsAffected?: number;
		};
		const published = info.affectedRows ?? info.rowsAffected ?? 0;
		if (published > 0) {
			logger.info(`Published ${published} scheduled article(s)`, {
				module: "jobs",
			});
		}
	} catch (err) {
		captureWorkerError(err, "Scheduled article publish failed");
	}
}

async function main() {
	logger.info("Worker started", { module: "jobs" });

	if (env.EMULATOR_JAR_PATH && env.EMULATOR_BACKUP_DIR) {
		new Cron("0 3 * * *", () => {
			backupEmulatorJar().catch((e) => captureWorkerError(e, "Backup error"));
		});
		logger.info("Scheduled: emulator JAR backup (daily 03:00)", {
			module: "jobs",
		});
	}

	if (env.DB_BACKUP_DIR) {
		new Cron("30 3 * * *", () => {
			backupDatabase().catch((e) => captureWorkerError(e, "DB backup error"));
		});
		logger.info("Scheduled: mysqldump DB backup (daily 03:30)", {
			module: "jobs",
		});
	}

	new Cron("0 4 * * *", () => {
		Promise.all([cleanupOldLogs(), cleanupOldSessions()]).catch((e) =>
			captureWorkerError(e, "Cleanup error"),
		);
	});
	logger.info("Scheduled: old data cleanup (daily 04:00)", { module: "jobs" });

	new Cron("*/5 * * * *", () => {
		checkOpsHealth().catch((e) => captureWorkerError(e, "Health check error"));
	});
	logger.info("Scheduled: ops health probe (every 5 min)", { module: "jobs" });

	new Cron("* * * * *", () => {
		publishScheduledArticles().catch((e) =>
			captureWorkerError(e, "ScheduledArticlePublish"),
		);
	});
	logger.info("Scheduled: publish scheduled articles (every minute)", {
		module: "jobs",
	});

	await Promise.all([
		backupEmulatorJar(),
		cleanupOldLogs(),
		cleanupOldSessions(),
		checkOpsHealth(),
	]);
}

main().catch((err) => {
	captureWorkerError(err, "Fatal");
	process.exit(1);
});