Claude Code事件驱动架构实战：消息队列到异步Worker全流程

claude --version
# 输出类似: 1.x.x

docker run -d --name redis-eda \
  -p 6379:6379 \
  -v redis-data:/data \
  redis:7-alpine redis-server --appendonly yes

graph LR
    subgraph 传统同步模型
        A[客户端] -->|HTTP请求| B[API服务]
        B -->|同步调用| C[邮件服务]
        B -->|同步调用| D[报表服务]
        B -->|等待...| E[返回响应]
        style E fill:#f96,stroke:#333
    end

    subgraph 事件驱动模型
        F[客户端] -->|HTTP请求| G[API服务]
        G -->|发布事件| H[事件总线]
        H -->|异步| I[邮件Worker]
        H -->|异步| J[报表Worker]
        G -->|立即返回| K[响应✓]
        style K fill:#9f6,stroke:#333
    end

mkdir event-driven-platform && cd event-driven-platform
claude init

# Event-Driven Platform

## Architecture
- Event-driven architecture using BullMQ + Redis
- TypeScript strict mode
- Pattern: Producer → Queue → Worker → Event Bus

## Tech Stack
- Runtime: Node.js 20+ with ESM modules
- Queue: BullMQ 5.x backed by Redis 7
- Validation: Zod schemas
- Logging: Pino structured logging
- Testing: Vitest

## Project Structure

## Rules
- 所有事件必须使用Zod schema验证
- Worker必须实现优雅关闭（graceful shutdown）
- 使用structured logging，禁止console.log
- 错误处理使用Result模式，不抛异常

claude "安装以下依赖：bullmq ioredis zod pino pino-pretty dotenv，以及开发依赖 typescript @types/node tsx vitest"

claude "创建 src/shared/redis.ts，使用ioredis建立Redis连接，支持环境变量配置，包含连接池和错误处理。同时创建 src/shared/types.ts 定义基础的Event和JobResult泛型类型。"

import Redis, { type RedisOptions } from 'ioredis';
import { logger } from './logger.js';

const getRedisConfig = (): RedisOptions => ({
  host: process.env.REDIS_HOST ?? 'localhost',
  port: parseInt(process.env.REDIS_PORT ?? '6379'),
  password: process.env.REDIS_PASSWORD,
  maxRetriesPerRequest: null, // BullMQ要求
  enableReadyCheck: true,
  retryStrategy: (times) => Math.min(times * 200, 5000),
});

// 创建连接池 - BullMQ需要专用连接
export const createRedis = (label: string): Redis => {
  const redis = new Redis(getRedisConfig());

  redis.on('error', (err) =>
    logger.error({ err, label }, 'Redis connection error')
  );
  redis.on('connect', () =>
    logger.info({ label }, 'Redis connected')
  );

  return redis;
};

import { z } from 'zod';

// 基础事件类型
export interface BaseEvent<T = unknown> {
  id: string;
  type: string;
  timestamp: number;
  payload: T;
  correlationId: string;
}

// 任务结果
export type JobResult<T = unknown> =
  | { success: true; data: T }
  | { success: false; error: string; retryable: boolean };

// 事件处理器签名
export type EventHandler<T> = (
  event: BaseEvent<T>
) => Promise<JobResult>;

claude "创建 src/events/schemas.ts，用Zod定义三种事件：
1. EmailEvent - 包含 to, subject, body, template?
2. ReportEvent - 包含 reportType(daily/weekly/monthly), dateRange, recipients[]
3. WebhookEvent - 包含 url, method, headers, body, retryCount
每个schema都要包含uuid和timestamp的自动生成。"

import { z } from 'zod';

const uuid = () => crypto.randomUUID();
const now = () => Date.now();

export const EmailEventSchema = z.object({
  id: z.string().uuid().default(uuid),
  type: z.literal('email.send'),
  timestamp: z.number().default(now),
  correlationId: z.string().uuid(),
  payload: z.object({
    to: z.string().email(),
    subject: z.string().min(1).max(200),
    body: z.string().min(1),
    template: z.enum(['welcome', 'notification', 'report']).optional(),
  }),
});

export const ReportEventSchema = z.object({
  id: z.string().uuid().default(uuid),
  type: z.literal('report.generate'),
  timestamp: z.number().default(now),
  correlationId: z.string().uuid(),
  payload: z.object({
    reportType: z.enum(['daily', 'weekly', 'monthly']),
    dateRange: z.object({
      start: z.string().datetime(),
      end: z.string().datetime(),
    }),
    recipients: z.array(z.string().email()).min(1),
  }),
});

export const WebhookEventSchema = z.object({
  id: z.string().uuid().default(uuid),
  type: z.literal('webhook.dispatch'),
  timestamp: z.number().default(now),
  correlationId: z.string().uuid(),
  payload: z.object({
    url: z.string().url(),
    method: z.enum(['POST', 'PUT', 'PATCH']),
    headers: z.record(z.string()).default({}),
    body: z.unknown(),
    retryCount: z.number().int().min(0).max(5).default(0),
  }),
});

// 类型推导
export type EmailEvent = z.infer<typeof EmailEventSchema>;
export type ReportEvent = z.infer<typeof ReportEventSchema>;
export type WebhookEvent = z.infer<typeof WebhookEventSchema>;

// 事件注册表 - 用于动态路由
export const EventRegistry = {
  'email.send': EmailEventSchema,
  'report.generate': ReportEventSchema,
  'webhook.dispatch': WebhookEventSchema,
} as const;

claude "创建 src/queues/index.ts，定义三个BullMQ队列：email-queue、report-queue、webhook-queue。
配置合理的默认值：并发数3、尝试次数3、指数退避。
然后创建 src/producers/event-producer.ts，实现类型安全的事件发布接口。"

import { Queue, type QueueOptions } from 'bullmq';
import { createRedis } from '../shared/redis.js';

const defaultQueueOptions: QueueOptions = {
  connection: createRedis('queue'),
  defaultJobOptions: {
    removeOnComplete: { count: 1000 },   // 保留最近1000条成功记录
    removeOnFail: { count: 5000 },        // 保留最近5000条失败记录
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 1000,  // 1s → 2s → 4s
    },
  },
};

export const emailQueue = new Queue('email-queue', {
  ...defaultQueueOptions,
  limiter: { max: 10, duration: 1000 }, // 每秒最多10封
});

export const reportQueue = new Queue('report-queue', {
  ...defaultQueueOptions,
  defaultJobOptions: {
    ...defaultQueueOptions.defaultJobOptions,
    attempts: 2, // 报表生成昂贵，少重试
  },
});

export const webhookQueue = new Queue('webhook-queue', {
  ...defaultQueueOptions,
  defaultJobOptions: {
    ...defaultQueueOptions.defaultJobOptions,
    backoff: {
      type: 'exponential',
      delay: 5000, // Webhook退避更保守
    },
  },
});

export const queues = { emailQueue, reportQueue, webhookQueue };

import { queues } from '../queues/index.js';
import { EventRegistry } from '../events/schemas.js';
import { logger } from '../shared/logger.js';
import type { JobResult } from '../shared/types.js';

type EventType = keyof typeof EventRegistry;

export class EventProducer {
  async publish<T extends EventType>(
    type: T,
    rawPayload: z.infer<typeof EventRegistry[T]>
  ): Promise<JobResult<{ jobId: string }>> {
    const schema = EventRegistry[type];

    // Zod验证 - 第一道防线
    const parsed = schema.safeParse(rawPayload);
    if (!parsed.success) {
      logger.warn({ type, errors: parsed.error.flatten() }, 'Event validation failed');
      return {
        success: false,
        error: `Validation: ${parsed.error.message}`,
        retryable: false,
      };
    }

    const event = parsed.data;
    const queueMap = {
      'email.send': queues.emailQueue,
      'report.generate': queues.reportQueue,
      'webhook.dispatch': queues.webhookQueue,
    };

    const queue = queueMap[type];
    const job = await queue.add(type, event, {
      jobId: event.id,
      // 事件元数据存入job，方便Dashboard追踪
      opts: { priority: this.getPriority(type) },
    });

    logger.info(
      { eventType: type, jobId: job.id, correlationId: event.correlationId },
      'Event published'
    );

    return { success: true, data: { jobId: job.id! } };
  }

  private getPriority(type: EventType): number {
    // 数字越小优先级越高
    const priorities: Record<EventType, number> = {
      'webhook.dispatch': 1,  // Webhook实时性要求最高
      'email.send': 5,
      'report.generate': 10,  // 报表可以稍慢
    };
    return priorities[type];
  }
}

export const producer = new EventProducer();

claude "创建 src/workers/email-worker.ts，实现邮件发送Worker。
要求：
1. 使用Worker类，并发度3
2. 模拟邮件发送（随机延迟+成功率95%）
3. 实现优雅关闭（SIGTERM/SIGINT）
4. 成功/失败都要写structured log
5. 失败时返回retryable信息"

import { Worker, type Job } from 'bullmq';
import { createRedis } from '../shared/redis.js';
import { logger } from '../shared/logger.js';
import { EmailEventSchema } from '../events/schemas.js';
import type { JobResult } from '../shared/types.js';

// 模拟邮件发送API
async function sendEmail(
  to: string,
  subject: string,
  body: string
): Promise<{ delivered: boolean; messageId: string }> {
  // 模拟网络延迟
  await new Promise((r) => setTimeout(r, 100 + Math.random() * 400));

  // 95%成功率
  if (Math.random() > 0.95) {
    throw new Error('SMTP connection refused');
  }

  return {
    delivered: true,
    messageId: `msg-${Date.now()}-${Math.random().toString(36).slice(2)}`,
  };
}

const processor = async (job: Job): Promise<JobResult> => {
  const { data } = job;

  // 运行时验证
  const parsed = EmailEventSchema.safeParse(data);
  if (!parsed.success) {
    return {
      success: false,
      error: `Invalid event data: ${parsed.error.message}`,
      retryable: false,
    };
  }

  const event = parsed.data;
  logger.info(
    { jobId: job.id, to: event.payload.to, attempt: job.attemptsStarted },
    'Processing email job'
  );

  try {
    const result = await sendEmail(
      event.payload.to,
      event.payload.subject,
      event.payload.body
    );

    logger.info(
      { jobId: job.id, messageId: result.messageId },
      'Email sent successfully'
    );

    return { success: true, data: result };
  } catch (err) {
    const errorMsg = err instanceof Error ? err.message : 'Unknown error';
    logger.warn({ jobId: job.id, error: errorMsg }, 'Email send failed');

    return {
      success: false,
      error: errorMsg,
      retryable: true, // SMTP错误可重试
    };
  }
};

// 创建Worker实例
const worker = new Worker('email-queue', processor, {
  connection: createRedis('email-worker'),
  concurrency: 3,
  limiter: { max: 5, duration: 1000 },
});

// 进度更新
worker.on('progress', (job, progress) => {
  logger.debug({ jobId: job.id, progress }, 'Job progress');
});

// 完成事件
worker.on('completed', (job) => {
  logger.info({ jobId: job.id }, 'Job completed');
});

// 失败事件
worker.on('failed', (job, err) => {
  logger.error({ jobId: job?.id, error: err.message }, 'Job failed permanently');
});

// 优雅关闭
const shutdown = async (signal: string) => {
  logger.info({ signal }, 'Shutting down email worker...');
  await worker.close(true); // true = 等待当前任务完成
  process.exit(0);
};

process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));

logger.info('Email worker started');

claude "创建 src/events/event-bus.ts，实现一个EventBus类：
1. 基于Redis Pub/Sub的跨进程事件广播
2. 支持subscribe/unsubscribe/publish
3. 内置事件日志（写入Redis Stream做审计追踪）
4. 提供getEventHistory方法查询历史"

import Redis from 'ioredis';
import { createRedis } from '../shared/redis.js';
import { logger } from '../shared/logger.js';
import type { BaseEvent, EventHandler } from '../shared/types.js';

type Listener<T = unknown> = {
  eventType: string;
  handler: EventHandler<T>;
};

export class EventBus {
  private publisher: Redis;
  private subscriber: Redis;
  private listeners: Map<string, Listener[]> = new Map();
  private streamKey = 'event-bus:history';

  constructor() {
    this.publisher = createRedis('event-bus-pub');
    this.subscriber = createRedis('event-bus-sub');

    this.subscriber.on('message', (channel, message) => {
      this.handleMessage(channel, message);
    });
  }

  async publish<T>(event: BaseEvent<T>): Promise<void> {
    const channel = `events:${event.type}`;
    const payload = JSON.stringify(event);

    // 广播给订阅者
    await this.publisher.publish(channel, payload);

    // 写入审计流
    await this.publisher.xadd(
      this.streamKey,
      '*',
      'id', event.id,
      'type', event.type,
      'payload', payload,
      'correlationId', event.correlationId
    );

    logger.info(
      { eventType: event.type, eventId: event.id, channel },
      'Event broadcasted'
    );
  }

  subscribe<T>(eventType: string, handler: EventHandler<T>): () => void {
    const channel = `events:${eventType}`;

    if (!this.listeners.has(channel)) {
      this.listeners.set(channel, []);
      this.subscriber.subscribe(channel);
    }

    const listener: Listener<T> = { eventType, handler };
    this.listeners.get(channel)!.push(listener as Listener);

    logger.info({ eventType, channel }, 'Subscribed to event');

    // 返回取消订阅函数
    return () => {
      const listeners = this.listeners.get(channel);
      if (listeners) {
        const idx = listeners.findIndex((l) => l.handler === handler);
        if (idx !== -1) listeners.splice(idx, 1);
        if (listeners.length === 0) {
          this.subscriber.unsubscribe(channel);
          this.listeners.delete(channel);
        }
      }
    };
  }

  private async handleMessage(channel: string, raw: string): Promise<void> {
    try {
      const event = JSON.parse(raw);
      const listeners = this.listeners.get(channel) ?? [];

      await Promise.allSettled(
        listeners.map(async ({ handler }) => {
          try {
            await handler(event);
          } catch (err) {
            logger.error(
              { err, eventType: event.type, eventId: event.id },
              'Event handler failed'
            );
          }
        })
      );
    } catch (err) {
      logger.error({ err, channel }, 'Failed to process event message');
    }
  }

  // 查询事件历史
  async getHistory(count = 50): Promise<BaseEvent[]> {
    const entries = await this.publisher.xrange(
      this.streamKey,
      '-',
      '+',
      'COUNT',
      count
    );

    return entries.map(([, fields]) => {
      const obj = Object.fromEntries(
        fields.reduce<string[]>((acc, val, i) => {
          if (i % 2 === 0) acc.push(val);
          else acc[acc.length - 1] = val;
          return acc;
        }, []).reduce<[string, string][]>((acc, _, i, arr) => {
          if (i % 2 === 0) acc.push([arr[i], arr[i + 1]]);
          return acc;
        }, [])
      );
      return JSON.parse(obj.payload);
    });
  }

  async disconnect(): Promise<void> {
    await this.subscriber.unsubscribe();
    await this.publisher.quit();
    await this.subscriber.quit();
  }
}

export const eventBus = new EventBus();

graph TB
    subgraph 客户端
        CLI[CLI/API调用]
        DASH[监控Dashboard]
    end

    subgraph 事件层
        EP[EventProducer<br/>Zod验证]
        EB[EventBus<br/>Redis Pub/Sub]
    end

    subgraph 队列层 - BullMQ
        EQ[email-queue]
        RQ[report-queue]
        WQ[webhook-queue]
    end

    subgraph Worker层
        EW[EmailWorker<br/>并发3]
        RW[ReportWorker<br/>并发2]
        WW[WebhookWorker<br/>并发5]
    end

    subgraph 存储
        REDIS[(Redis 7<br/>队列+Pub/Sub+Streams)]
    end

    CLI -->|发布事件| EP
    EP -->|验证+入队| EQ
    EP -->|验证+入队| RQ
    EP -->|验证+入队| WQ
    EP -->|广播| EB

    EB -->|审计日志| REDIS
    EQ -->|消费| EW
    RQ -->|消费| RW
    WQ -->|消费| WW

    EW -->|状态更新| REDIS
    RW -->|状态更新| REDIS
    WW -->|状态更新| REDIS

    DASH -->|读取状态| REDIS

    style EP fill:#6366f1,color:#fff
    style EB fill:#8b5cf6,color:#fff
    style REDIS fill:#dc2626,color:#fff
    style DASH fill:#059669,color:#fff

claude "创建 src/monitoring/api.ts，使用Node.js原生http模块（不引入Express）创建监控API：
- GET /stats - 各队列的等待/活跃/完成/失败数
- GET /history - 最近事件历史
- GET /health - 健康检查（Redis连接状态）
端口从环境变量MONITOR_PORT读取，默认3001"

import { createServer } from 'node:http';
import { queues } from '../queues/index.js';
import { createRedis } from '../shared/redis.js';
import { logger } from '../shared/logger.js';

const PORT = parseInt(process.env.MONITOR_PORT ?? '3001');

const healthRedis = createRedis('monitor');

async function getStats() {
  const [email, report, webhook] = await Promise.all([
    queues.emailQueue.getJobCounts(),
    queues.reportQueue.getJobCounts(),
    queues.webhookQueue.getJobCounts(),
  ]);

  return {
    timestamp: new Date().toISOString(),
    queues: {
      email: email,
      report: report,
      webhook: webhook,
    },
  };
}

async function getHealth() {
  try {
    await healthRedis.ping();
    return { status: 'healthy', redis: 'connected', uptime: process.uptime() };
  } catch {
    return { status: 'unhealthy', redis: 'disconnected' };
  }
}

const server = createServer(async (req, res) => {
  res.setHeader('Content-Type', 'application/json');
  res.setHeader('Access-Control-Allow-Origin', '*');

  try {
    if (req.url === '/stats') {
      const stats = await getStats();
      res.writeHead(200);
      res.end(JSON.stringify(stats));
    } else if (req.url === '/health') {
      const health = await getHealth();
      res.writeHead(health.status === 'healthy' ? 200 : 503);
      res.end(JSON.stringify(health));
    } else {
      res.writeHead(404);
      res.end(JSON.stringify({ error: 'Not found' }));
    }
  } catch (err) {
    logger.error({ err }, 'Monitor API error');
    res.writeHead(500);
    res.end(JSON.stringify({ error: 'Internal error' }));
  }
});

server.listen(PORT, () => {
  logger.info({ port: PORT }, 'Monitor API started');
});

claude "创建 src/monitoring/dashboard.ts，用Node.js http模块提供一个单页面HTML dashboard。
页面功能：
1. 每2秒自动刷新队列状态（调用/stats API）
2. 用CSS Grid展示三个队列的状态卡片
3. 每个卡片显示：等待中、活跃、完成、失败的数字
4. 使用纯CSS动画，数字变化时有闪烁效果
5. 深色主题，适合运维大屏"

// src/monitoring/dashboard.ts 的关键HTML模板
const dashboardHTML = `
<!DOCTYPE html>
<html lang="zh-CN">
<head>
  <meta charset="UTF-8">
  <title>Event-Driven Platform Dashboard</title>
  <style>
    * { margin: 0; box-sizing: border-box; }
    body { background: #0f172a; color: #e2e8f0; font-family: system-ui; padding: 2rem; }
    .grid { display: grid; grid-template-columns: repeat(3, 1fr); gap: 1.5rem; }
    .card { background: #1e293b; border-radius: 12px; padding: 1.5rem;
            border: 1px solid #334155; transition: border-color 0.3s; }
    .card:hover { border-color: #6366f1; }
    .card h2 { font-size: 1.1rem; margin-bottom: 1rem; color: #94a3b8; }
    .metric { display: flex; justify-content: space-between; padding: 0.5rem 0;
              border-bottom: 1px solid #334155; }
    .metric:last-child { border: none; }
    .metric .value { font-weight: 700; font-size: 1.3rem; font-variant-numeric: tabular-nums; }
    .waiting .value { color: #facc15; }
    .active .value { color: #22d3ee; }
    .completed .value { color: #4ade80; }
    .failed .value { color: #f87171; }
    @keyframes pulse { 0%,100% { opacity: 1; } 50% { opacity: 0.5; } }
    .flash { animation: pulse 0.5s ease; }
    .header { text-align: center; margin-bottom: 2rem; }
    .header h1 { font-size: 1.5rem; color: #f8fafc; }
    .header .ts { color: #64748b; font-size: 0.85rem; }
  </style>
</head>
<body>
  <div class="header">
    <h1>Event-Driven Platform Monitor</h1>
    <div class="ts" id="timestamp">--</div>
  </div>
  <div class="grid" id="queues"></div>
  <script>
    const API = location.port === '3002' ? 'http://localhost:3001' : '';
    async function refresh() {
      try {
        const res = await fetch(API + '/stats');
        const data = await res.json();
        document.getElementById('timestamp').textContent = data.timestamp;
        const container = document.getElementById('queues');
        const labels = { waiting: '等待中', active: '处理中',
                         completed: '已完成', failed: '已失败' };
        const cls = { waiting: 'waiting', active: 'active',
                      completed: 'completed', failed: 'failed' };
        container.innerHTML = Object.entries(data.queues).map(([name, counts]) => \`
          <div class="card">
            <h2>\${name.toUpperCase().replace('-', ' ')}</h2>
            \${Object.entries(labels).map(([key, label]) => \`
              <div class="metric \${cls[key]}">
                <span>\${label}</span>
                <span class="value flash">\${counts[key] ?? 0}</span>
              </div>
            \`).join('')}
          </div>
        \`).join('');
      } catch(e) { console.error('Fetch failed:', e); }
    }
    refresh();
    setInterval(refresh, 2000);
  </script>
</body>
</html>
`;

claude "创建 src/index.ts 作为系统入口，功能：
1. 加载dotenv配置
2. 启动所有Worker
3. 启动监控API和Dashboard
4. 提供一个交互式CLI，可以手动发送测试事件
5. 演示模式：启动后自动发送一批测试事件"

import 'dotenv/config';
import { producer } from './producers/event-producer.js';
import { eventBus } from './events/event-bus.js';
import { logger } from './shared/logger.js';
import './workers/email-worker.js';
import './monitoring/api.js';

async function demo() {
  logger.info('=== Event-Driven Platform Demo ===');

  // 订阅所有事件做审计日志
  const eventTypes = ['email.send', 'report.generate', 'webhook.dispatch'];
  const unsubscribers = eventTypes.map((type) =>
    eventBus.subscribe(type, async (event) => {
      logger.info(
        { eventType: type, eventId: event.id },
        '[Audit] Event received via bus'
      );
      return { success: true, data: 'logged' };
    })
  );

  // 发送一批测试事件
  const correlationId = crypto.randomUUID();

  // 1. 邮件事件
  const emailResult = await producer.publish('email.send', {
    id: crypto.randomUUID(),
    type: 'email.send',
    timestamp: Date.now(),
    correlationId,
    payload: {
      to: '[email protected]',
      subject: '欢迎使用事件驱动平台',
      body: '这是一封由BullMQ队列异步处理的邮件。',
      template: 'welcome',
    },
  });
  logger.info({ result: emailResult }, 'Email event published');

  // 2. 报表事件
  const reportResult = await producer.publish('report.generate', {
    id: crypto.randomUUID(),
    type: 'report.generate',
    timestamp: Date.now(),
    correlationId,
    payload: {
      reportType: 'daily',
      dateRange: {
        start: new Date(Date.now() - 86400000).toISOString(),
        end: new Date().toISOString(),
      },
      recipients: ['[email protected]'],
    },
  });
  logger.info({ result: reportResult }, 'Report event published');

  // 3. Webhook事件
  const webhookResult = await producer.publish('webhook.dispatch', {
    id: crypto.randomUUID(),
    type: 'webhook.dispatch',
    timestamp: Date.now(),
    correlationId,
    payload: {
      url: 'https://httpbin.org/post',
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: { event: 'test', correlationId },
      retryCount: 0,
    },
  });
  logger.info({ result: webhookResult }, 'Webhook event published');

  logger.info('Demo events published. Workers are processing...');
  logger.info('Dashboard: http://localhost:3002');
  logger.info('API: http://localhost:3001/stats');
}

demo().catch((err) => {
  logger.fatal({ err }, 'Failed to start platform');
  process.exit(1);
});

# 终端1：启动Redis（如果还没启动）
docker start redis-eda

# 终端2：启动系统
npx tsx src/index.ts

[INFO] Redis connected (label="queue")
[INFO] Redis connected (label="email-worker")
[INFO] Redis connected (label="event-bus-pub")
[INFO] Redis connected (label="event-bus-sub")
[INFO] Email worker started
[INFO] Monitor API started (port=3001)
[INFO] === Event-Driven Platform Demo ===
[INFO] Event published (eventType="email.send", jobId="...")
[INFO] Event published (eventType="report.generate", jobId="...")
[INFO] Event published (eventType="webhook.dispatch", jobId="...")
[INFO] Processing email job (jobId="...", to="[email protected]")
[INFO] Email sent successfully (jobId="...", messageId="msg-...")
[INFO] Job completed (jobId="...")

# 检查队列统计
curl http://localhost:3001/stats | jq .

# 输出：
# {
#   "timestamp": "2026-05-19T10:30:00.000Z",
#   "queues": {
#     "email": { "waiting": 0, "active": 0, "completed": 1, "failed": 0 },
#     "report": { "waiting": 0, "active": 1, "completed": 0, "failed": 0 },
#     "webhook": { "waiting": 0, "active": 0, "completed": 1, "failed": 0 }
#   }
# }

# 健康检查
curl http://localhost:3001/health | jq .
# { "status": "healthy", "redis": "connected", "uptime": 42.5 }

claude "在 src/workers/ 下创建 dead-letter-worker.ts，监听所有队列的failed事件，
将失败任务写入 Redis Sorted Set（按时间排序），
并提供查询接口。同时在 monitoring/api.ts 中添加 GET /dead-letters 端点。"

// Worker失败时自动触发
worker.on('failed', async (job, err) => {
  if (job && job.attemptsMade >= job.opts?.attempts!) {
    // 写入死信队列
    await deadLetterRedis.zadd(
      'dead-letters',
      Date.now(),
      JSON.stringify({
        jobId: job.id,
        queueName: 'email-queue',
        data: job.data,
        error: err.message,
        failedAt: new Date().toISOString(),
      })
    );
    logger.error({ jobId: job.id }, 'Job moved to dead letter queue');
  }
});

// 延迟发送 - 5分钟后
await emailQueue.add('email.send', eventPayload, {
  delay: 5 * 60 * 1000,
});

// 定时调度 - 使用BullMQ的Repeatable API
await reportQueue.add(
  'report.generate',
  { reportType: 'daily', /* ... */ },
  {
    repeat: { pattern: '0 6 * * *' }, // 每天早上6点
    jobId: 'daily-report',
  }
);

graph TD
    IN[事件入口] --> P{优先级判断}
    P -->|urgent| H[高优先级队列<br/>priority=1]
    P -->|normal| M[普通队列<br/>priority=5]
    P -->|batch| L[批量队列<br/>priority=10]
    H --> W[Worker<br/>concurrency=5]
    M --> W
    L --> W
    W -->|超限| RL[速率限制器<br/>10 req/s]
    RL --> EXT[外部API]

    style H fill:#ef4444,color:#fff
    style M fill:#f59e0b,color:#fff
    style L fill:#6b7280,color:#fff
    style RL fill:#8b5cf6,color:#fff

// 在Producer中动态设置优先级
async publishWithPriority(
  type: EventType,
  payload: unknown,
  priority: 'urgent' | 'normal' | 'batch'
) {
  const priorityMap = { urgent: 1, normal: 5, batch: 10 };
  return queue.add(type, payload, {
    priority: priorityMap[priority],
  });
}

claude "为 event-producer.ts 编写Vitest单元测试。
要求：
1. Mock BullMQ的Queue类
2. 测试正常发布流程
3. 测试Zod验证失败的场景
4. 测试优先级分配逻辑
5. 使用beforeEach/afterEach管理Mock"

import { describe, it, expect, vi, beforeEach } from 'vitest';
import { producer } from '../event-producer.js';

// Mock BullMQ
vi.mock('../../queues/index.js', () => ({
  queues: {
    emailQueue: { add: vi.fn() },
    reportQueue: { add: vi.fn() },
    webhookQueue: { add: vi.fn() },
  },
}));

import { queues } from '../../queues/index.js';

describe('EventProducer', () => {
  beforeEach(() => {
    vi.clearAllMocks();
  });

  it('should publish valid email event', async () => {
    const mockAdd = vi.mocked(queues.emailQueue.add).mockResolvedValue({
      id: 'job-123',
    } as any);

    const result = await producer.publish('email.send', {
      id: crypto.randomUUID(),
      type: 'email.send',
      timestamp: Date.now(),
      correlationId: crypto.randomUUID(),
      payload: {
        to: '[email protected]',
        subject: 'Test',
        body: 'Test body',
      },
    });

    expect(result.success).toBe(true);
    if (result.success) {
      expect(result.data.jobId).toBe('job-123');
    }
    expect(mockAdd).toHaveBeenCalledOnce();
  });

  it('should reject invalid email', async () => {
    const result = await producer.publish('email.send', {
      id: crypto.randomUUID(),
      type: 'email.send',
      timestamp: Date.now(),
      correlationId: crypto.randomUUID(),
      payload: {
        to: 'not-an-email',
        subject: 'Test',
        body: 'Test body',
      },
    });

    expect(result.success).toBe(false);
    if (!result.success) {
      expect(result.retryable).toBe(false);
      expect(result.error).toContain('Validation');
    }
  });
});

npx vitest run
#  ✓ event-producer.test.ts (2 tests) 15ms
#  Tests  2 passed (2)
#  Duration  320ms

# docker-compose.yml
version: '3.8'

services:
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes --maxmemory 512mb --maxmemory-policy allkeys-lru
    volumes:
      - redis-data:/data
    healthcheck:
      test: ['CMD', 'redis-cli', 'ping']
      interval: 5s
      retries: 5

  email-worker:
    build: .
    command: npx tsx src/workers/email-worker.ts
    depends_on:
      redis: { condition: service_healthy }
    environment:
      REDIS_HOST: redis
    deploy:
      replicas: 2  # 水平扩展
      resources:
        limits: { memory: 256M }

  report-worker:
    build: .
    command: npx tsx src/workers/report-worker.ts
    depends_on:
      redis: { condition: service_healthy }
    environment:
      REDIS_HOST: redis

  monitor:
    build: .
    command: npx tsx src/monitoring/api.ts
    depends_on:
      redis: { condition: service_healthy }
    ports:
      - '3001:3001'
    environment:
      REDIS_HOST: redis

volumes:
  redis-data:

// 添加到 monitoring/api.ts 的告警逻辑
async function checkAlerts(): Promise<string[]> {
  const alerts: string[] = [];
  const emailCounts = await queues.emailQueue.getJobCounts();

  if (emailCounts.waiting > 1000) {
    alerts.push('email-queue堆积超过1000，需要扩容Worker');
  }
  if (emailCounts.failed > 100) {
    alerts.push('email-queue失败数超过100，检查SMTP服务');
  }

  const deadLetterCount = await deadLetterRedis.zcard('dead-letters');
  if (deadLetterCount > 50) {
    alerts.push(`死信队列中有${deadLetterCount}条消息待处理`);
  }

  return alerts;
}

sequenceDiagram
    participant C as 客户端
    participant P as Producer
    participant Z as Zod验证
    participant Q as BullMQ Queue
    participant W as Worker
    participant BUS as EventBus
    participant R as Redis
    participant M as Monitor

    C->>P: publish('email.send', payload)
    P->>Z: safeParse(payload)

    alt 验证失败
        Z-->>P: ❌ Validation Error
        P-->>C: {success: false, retryable: false}
    else 验证通过
        Z-->>P: ✅ Parsed Event
        P->>Q: queue.add(type, event)
        Q->>R: LPUSH job data
        P->>BUS: broadcast(event)
        BUS->>R: XADD audit stream
        BUS->>R: PUBLISH events:email.send
        P-->>C: {success: true, jobId: '...'}
    end

    Note over Q,W: 异步处理
    Q->>W: BRPOP job
    W->>W: process(job)
    W->>R: 更新job状态(active)
    W->>W: 执行业务逻辑
    W->>R: 更新job状态(completed)

    M->>R: GET job counts
    R-->>M: stats data
    M-->>C: Dashboard展示