export interface SendNotificationParams {
  recipientId: string;
  channel: "email" | "sms" | "in_app";
  templateId: string;
  variables: Record<string, string | number>;
}

export async function sendNotification(
  params: SendNotificationParams
): Promise<{ deliveredAt: string }> {
  // 业务实现：读取数据库、调用下游通道接口等
  console.log(`[Service] 处理通知分发: ${params.channel} -> ${params.recipientId}`);

  return {
    deliveredAt: new Date().toISOString(),
  };
}

import { Worker, type Job, type WorkerOptions } from "bullmq";
import { sendNotification } from "./notificationService";
import type { NotificationJobData } from "./notificationQueue";

// 薄层任务处理器
export async function processNotificationJob(
  job: Job<NotificationJobData>
): Promise<{ deliveredAt: string }> {
  const { recipientId, channel, templateId, variables } = job.data;

  if (!recipientId || !channel) {
    throw new Error("任务参数校验失败：缺失 recipientId 或 channel");
  }

  await job.updateProgress(20);

  // 委托给领域服务层
  const result = await sendNotification({
    recipientId,
    channel,
    templateId,
    variables,
  });

  await job.updateProgress(100);
  return result;
}

// 消费者工厂
export function createNotificationWorker(options?: Partial<WorkerOptions>) {
  const worker = new Worker<NotificationJobData, { deliveredAt: string }>(
    "notification-dispatch",
    processNotificationJob,
    {
      connection: {
        host: process.env.REDIS_HOST ?? "127.0.0.1",
        port: Number(process.env.REDIS_PORT ?? 6379),
        maxRetriesPerRequest: null, // BullMQ 官方要求：避免阻塞命令在断连时抛出错误
      },
      concurrency: 5, // 控制单个 Worker 实例的并发处理上限
      ...options,
    }
  );

  worker.on("completed", (job, result) => {
    console.log(`[Worker] 任务 ${job.id} 处理完成，时间: ${result.deliveredAt}`);
  });

  worker.on("failed", (job, err) => {
    console.error(`[Worker] 任务 ${job?.id} 第 ${job?.attemptsMade} 次执行失败: ${err.message}`);
  });

  return worker;
}

import { Queue, type ConnectionOptions, type JobsOptions } from "bullmq";

export interface NotificationJobData {
  recipientId: string;
  channel: "email" | "sms" | "in_app";
  templateId: string;
  variables: Record<string, string | number>;
  timestamp: string;
}

const connection: ConnectionOptions = {
  host: process.env.REDIS_HOST ?? "127.0.0.1",
  port: Number(process.env.REDIS_PORT ?? 6379),
  maxRetriesPerRequest: null,
};

const defaultJobOptions: JobsOptions = {
  attempts: 3,
  backoff: {
    type: "exponential",
    delay: 2000, // 失败后按 2s, 4s, 8s 指数退避重试
  },
  removeOnComplete: {
    count: 1000, // 仅保留最新 1000 条完成记录
  },
  removeOnFail: {
    count: 5000, // 保留最近 5000 条失败记录供排查
  },
};

export const notificationQueue = new Queue<NotificationJobData>(
  "notification-dispatch",
  {
    connection,
    defaultJobOptions,
  }
);

export async function enqueueNotification(
  data: NotificationJobData,
  eventId?: string
): Promise<string> {
  // 若传入业务事件 ID 则作为幂等键，避免相同事件重复入队
  const jobId = eventId ? `notify_${data.channel}_${eventId}_${data.recipientId}` : undefined;

  const job = await notificationQueue.add("send-notification", data, {
    jobId,
  });

  return job.id!;
}

// 延迟执行：15 分钟后触发
await notificationQueue.add(
  "timeout-reminder",
  reminderData,
  { delay: 15 * 60 * 1000 }
);

// 周期性调度：每小时整点执行（基于 Cron 表达式）
await notificationQueue.add(
  "hourly-digest",
  digestData,
  {
    repeat: {
      pattern: "0 * * * *",
    },
  }
);

const productionJobOptions: JobsOptions = {
  removeOnComplete: {
    age: 24 * 3600, // 自动清理超过 24 小时的完成记录
    count: 5000,    // 最多保留最新 5000 条
  },
  removeOnFail: {
    age: 7 * 24 * 3600, // 失败任务保留 7 天以供排查追溯
  },
};

// scripts/start-worker.ts
import { createNotificationWorker } from "../workers/notificationWorker";

console.log("[Worker] 启动后台消费者守护进程...");

// 批量实例化需要监听的 Worker
const notificationWorker = createNotificationWorker();

// 注册退出信号处理逻辑
async function gracefulShutdown(signal: string) {
  console.log(`[Worker] 收到退出信号 [${signal}]，开始平滑关闭...`);
  try {
    // 停止提取新任务，并等待正在执行中的任务处理完毕
    await Promise.allSettled([
      notificationWorker.close(),
    ]);
    console.log("[Worker] 所有消费者已安全关闭");
    process.exit(0);
  } catch (err) {
    console.error("[Worker] 关闭期间发生异常:", err);
    process.exit(1);
  }
}

process.on("SIGTERM", () => gracefulShutdown("SIGTERM"));
process.on("SIGINT", () => gracefulShutdown("SIGINT"));

import express from "express";
import { createBullBoard } from "@bull-board/api";
import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
import { ExpressAdapter } from "@bull-board/express";
import { notificationQueue } from "./notificationQueue";

const app = express();
const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath("/admin/queues");

createBullBoard({
  queues: [new BullMQAdapter(notificationQueue)],
  serverAdapter,
});

app.use("/admin/queues", serverAdapter.getRouter());

app.listen(3001, () => {
  console.log("[Dashboard] 监控面板已启动: http://127.0.0.1:3001/admin/queues");
});