oms/oms/services/cron-jobs/index.ts

// oms/services/cron-jobs/index.ts

import cron from "node-cron"; // Import node-cron library

import { initializeClients, clickhouseService } from "../../src/services/clients"; // 从新的文件导入

// Define an interface for a cron job module
interface CronJobModule {
  run: () => Promise<any>; // Expecting a run function that returns a Promise
}

// Define the settings array for cron jobs
// Each element: [name: string, schedule: string, jobModule: CronJobModule]
const settings: [string, string, CronJobModule][] = [
  // 假设这些文件将存在于 oms/services/cron-jobs/ 目录下
  ["done-rate", "10 0 * * *", require("./done-rate2") as CronJobModule], // 每天凌晨0点10分, 统计作品完成率
  ["daily-activity-detector", "50 0 * * *", require("./daily-activity-detector") as CronJobModule], // 每天凌晨0点50分, 检查是否需要生成新的推送消息
  // ["message-sender", "*/5 * * * *", require("./message-sender") as CronJobModule], // 每5分钟运行一次, 已经单独剥离出去了（message-seender-service），定时任务这里取消了
  // ["active-user-daily-notify", "30 18 * * *", require("./active-user-daily-notify") as CronJobModule], // 每天下午6点，开始活跃用户新作品消息推送
  // ["fcm-notify", "30 18 * * *", require("./fcm-notify") as CronJobModule], // 每天下午6点，基于原来的active-user-daily-notify，增加schedule推送，AB测试
  ["daily-notify-at-morning", "0 19 * * *", require("./notify/daily-notify-at-morning") as CronJobModule], // 每天下午7点，对应巴西时间早上8点推送一轮
  ["daily-notify-at-midday", "0 23 * * *", require("./notify/daily-notify-at-midday") as CronJobModule], // 每天下午7点，对应巴西时间中午12点推送一轮
  ["daily-notify-at-evening", "0 3 * * *", require("./notify/daily-notify-at-evening") as CronJobModule], // 每天凌晨3点，对应巴西时间下午4点
  ["daily-notify-at-afternoon", "0 7 * * *", require("./notify/daily-notify-at-afternoon") as CronJobModule], // 每天上午7点，对应巴西时间晚上8点
];

/**
 * Starts all scheduled cron jobs.
 * Includes database connection and task scheduling.
 * @returns Promise<void>
 */
export async function startCronJobs(): Promise<void> {
  console.log("[Cron Jobs] Initializing all scheduled tasks...");

  // 先初始化客户端服务
  await initializeClients();

  // Iterate through settings and schedule each job
  settings.forEach((setting) => {
    const [name, schedule, job] = setting;
    if (!job || typeof job.run !== "function") {
      // Check if job module and run function exist
      console.error(`[Cron Jobs] Job [${name}] is missing a run() function or is not a valid module. Skipping.`);
      return;
    }

    console.log(`[Cron Jobs] Installing job [${name}] to run at '${schedule}'`);

    cron.schedule(schedule, async () => {
      const startTime = new Date().toLocaleString();
      console.log(`[Cron Jobs] Running job [${name}]@'${schedule}' started @ ${startTime}`);
      try {
        const result = await job.run(); // Execute the job's run function
        console.log(`[Cron Jobs] Job [${name}] completed successfully @ ${new Date().toLocaleString()}. Result:`, result);
      } catch (error) {
        console.error(`[Cron Jobs] Job [${name}] failed @ ${new Date().toLocaleString()}. Error:`, error);
      }
    });
  });

  console.log("[Cron Jobs] All cron jobs started.");
}

// If this file is run directly (e.g., using `node dist/services/cron-jobs/index.js`)
if (require.main === module) {
  startCronJobs().catch((error) => {
    console.error("[Cron Jobs] Failed to start cron jobs:", error);
    process.exit(1);
  });

  // Handle graceful shutdown
  process.on("SIGINT", async () => {
    console.log("[Cron Jobs] Shutting down...");
    cron.getTasks().forEach((task) => task.stop()); // Stop all scheduled tasks
    // await disconnectFromDatabase(); // 在退出前断开数据库连接
    console.log("[Cron Jobs] All cron tasks stopped.");
    process.exit(0);
  });
  process.on("SIGTERM", async () => {
    console.log("[Cron Jobs] Shutting down...");
    cron.getTasks().forEach((task) => task.stop());
    // await disconnectFromDatabase(); // 在退出前断开数据库连接
    console.log("[Cron Jobs] All cron tasks stopped.");
    process.exit(0);
  });
}