Message Queue：
用 Kafka 解耦你的系統

用戶按下「下訂單」
  → 扣庫存（DB 操作）
  → 建立訂單記錄（DB 操作）
  → 發送確認 Email（外部 API）
  → 通知倉庫系統（外部 API）
  → 更新推薦系統（ML 運算）
  → 產生財務報表（複雜運算）
  → 回傳「訂單成功」給用戶

Producer（生產者）→ [Message Queue / Broker] → Consumer（消費者）
                          ↑
                     Topic（主題）
                      ├── Partition 0
                      ├── Partition 1
                      └── Partition 2

Topic: "order-created"
  ├── Partition 0: [msg1, msg4, msg7...]  → Consumer A
  ├── Partition 1: [msg2, msg5, msg8...]  → Consumer B
  └── Partition 2: [msg3, msg6, msg9...]  → Consumer C

Partition 0: [0: "order-123", 1: "order-456", 2: "order-789"]
                                                ↑
                                       Consumer A 已處理到這裡
                                       （committed offset: 2）

import { Kafka } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'order-service',
  brokers: ['kafka:9092'],
});

// ── Producer（訂單服務）────────────────────────────────────────────
const producer = kafka.producer();
await producer.connect();

// 下訂單時發布事件
async function createOrder(orderData) {
  // 1. 存入資料庫（必須先做，確保資料一致性）
  const order = await db.order.create({ data: orderData });

  // 2. 發布事件到 Kafka（非阻塞，發完就回傳）
  await producer.send({
    topic: 'order-created',
    messages: [{
      key: order.id,              // 確保同一訂單的所有事件落在同一 Partition
      value: JSON.stringify({
        orderId: order.id,
        userId: order.userId,
        items: order.items,
        totalAmount: order.total,
        createdAt: new Date().toISOString(),
      }),
    }],
  });

  // 3. 立刻回傳，不等 Email / 倉庫 / 推薦系統完成
  return { success: true, orderId: order.id };
}

// ── Consumer（Email 服務）──────────────────────────────────────────
const emailConsumer = kafka.consumer({ groupId: 'email-service' });
await emailConsumer.connect();
await emailConsumer.subscribe({ topic: 'order-created', fromBeginning: false });

await emailConsumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const order = JSON.parse(message.value.toString());

    try {
      await sendOrderConfirmationEmail(order.userId, {
        orderId: order.orderId,
        items: order.items,
        total: order.totalAmount,
      });
      console.log(`已發送確認 Email，訂單 ${order.orderId}`);
    } catch (err) {
      // 錯誤處理：可以發到 Dead Letter Queue 或記錄 log
      console.error(`Email 發送失敗：${err.message}`, { orderId: order.orderId });
      throw err; // 重新拋出，讓 Kafka 知道這條訊息處理失敗（不 commit offset）
    }
  },
});

// ── Consumer（倉庫服務）────────────────────────────────────────────
const warehouseConsumer = kafka.consumer({ groupId: 'warehouse-service' });
await warehouseConsumer.connect();
await warehouseConsumer.subscribe({ topic: 'order-created', fromBeginning: false });

await warehouseConsumer.run({
  eachMessage: async ({ message }) => {
    const order = JSON.parse(message.value.toString());

    // 通知倉庫系統準備出貨
    await warehouseApi.createPickingTask({
      orderId: order.orderId,
      items: order.items,
      priority: order.totalAmount > 10000 ? 'high' : 'normal',
    });

    console.log(`已通知倉庫，訂單 ${order.orderId}`);
  },
});

// Dead Letter Queue 模式
const consumer = kafka.consumer({
  groupId: 'email-service',
  retry: {
    initialRetryTime: 100,     // 第一次重試等 100ms
    retries: 3,                // 最多重試 3 次
    multiplier: 2,             // 每次等待時間翻倍（100 → 200 → 400ms）
  },
});

// 如果 3 次都失敗，手動送到 DLQ
await emailConsumer.run({
  eachMessage: async ({ message }) => {
    try {
      await processMessage(message);
    } catch (err) {
      // 發到 Dead Letter Queue 供後續人工檢查
      await producer.send({
        topic: 'order-created-dlq',
        messages: [{
          key: message.key,
          value: message.value,
          headers: {
            'original-topic': 'order-created',
            'error-message': err.message,
            'failed-at': new Date().toISOString(),
          },
        }],
      });
    }
  },
});

// 傳統做法：只儲存當前狀態
// orders 表: { id, status: 'shipped', updatedAt }
// → 無法知道「什麼時候從 pending 變成 processing」

// Event Sourcing：儲存所有事件
// order_events 表:
// { orderId, type: 'ORDER_CREATED',   timestamp, payload: {...} }
// { orderId, type: 'PAYMENT_SUCCESS', timestamp, payload: {...} }
// { orderId, type: 'ORDER_SHIPPED',   timestamp, payload: {...} }

// 從事件重建當前狀態
function rebuildOrderState(events) {
  return events.reduce((state, event) => {
    switch (event.type) {
      case 'ORDER_CREATED':
        return { ...state, status: 'pending', ...event.payload };
      case 'PAYMENT_SUCCESS':
        return { ...state, status: 'paid', paidAt: event.timestamp };
      case 'ORDER_SHIPPED':
        return { ...state, status: 'shipped', shippedAt: event.timestamp };
      default:
        return state;
    }
  }, {});
}

// 好處：可以「時光旅行」— 重播到任何時間點
const stateAtYesterday = rebuildOrderState(
  events.filter(e => e.timestamp < yesterday)
);

寫入流程（Command）:
用戶下訂單 → 訂單服務 → 寫入 Write DB → 發布 order-created 事件

同步流程（非同步）:
order-created 事件 → Consumer → 更新 Read DB（針對查詢優化的結構）

讀取流程（Query）:
用戶查訂單列表 → 直接讀 Read DB（可以是 Elasticsearch、Redis、Cache）
                 → 超快，因為是預先計算好的讀取優化結構

// Saga：下訂單的分散式交易
// 正向流程（每步成功才到下一步）
// 下訂單 → 扣庫存 → 扣款 → 建立出貨單

// Choreography 模式（事件驅動）
// 1. 訂單服務：建立訂單 → 發布 "order-pending"
// 2. 庫存服務：收到事件 → 扣庫存 → 發布 "inventory-reserved"
// 3. 支付服務：收到事件 → 扣款 → 發布 "payment-success"
// 4. 出貨服務：收到事件 → 建立出貨單 → 發布 "order-confirmed"

// 補償流程（如果中途失敗）
// 假設扣款失敗：
// 支付服務發布 "payment-failed"
// → 庫存服務收到 → 還原庫存（補償操作）
// → 訂單服務收到 → 標記訂單失敗 → 通知用戶

// Node.js 實作（簡化版）
const sagaSteps = [
  {
    execute: () => inventoryService.reserve(orderId, items),
    compensate: () => inventoryService.release(orderId, items),
  },
  {
    execute: () => paymentService.charge(userId, amount),
    compensate: () => paymentService.refund(userId, amount),
  },
  {
    execute: () => shippingService.create(orderId),
    compensate: () => shippingService.cancel(orderId),
  },
];

async function executeSaga(steps) {
  const completed = [];
  try {
    for (const step of steps) {
      await step.execute();
      completed.push(step);
    }
  } catch (err) {
    // 倒序補償已完成的步驟
    for (const step of completed.reverse()) {
      await step.compensate().catch(console.error);
    }
    throw err;
  }
}