Message Queue Flow

🎯 Mục tiêu: Tự động hủy đơn hàng hết hạn

Khi khách hàng đặt vé, ghế bị lock trong N giây (mặc định 600s = 10 phút). Nếu quá thời gian mà không thanh toán → phải tự động nhả ghế và hủy đơn.

Không thể dùng setTimeout vì server restart sẽ mất hết. Dùng CloudAMQP (RabbitMQ) để lưu message xuống đĩa, sống sót qua mọi lần restart.

---

🏗️ Cấu trúc AMQP — "Hộp thư trễ hẹn"

Hình dung như một dịch vụ gửi thư tự động:

graph TD
    %% Định nghĩa các Style
    classDef service fill:#e1f5fe,stroke:#01579b,stroke-width:2px;
    classDef queue fill:#fff4dd,stroke:#d4a017,stroke-width:2px;
    classDef exchange fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px;
    classDef cloud fill:#fafafa,stroke:#333,stroke-dasharray: 5 5;

    %% Các thành phần bên ngoài
    P["<b>purchase.service.ts</b><br/>('Tôi vừa tạo đơn #123')"]:::service
    C["<b>consumer.ts (Worker)</b><br/>'Đọc thư → xử lý đơn'"]:::service

    subgraph CloudAMQP ["CloudAMQP (RabbitMQ)"]
        direction TB
        
        DQ["<b>delay-queue</b><br/>(Hộp thư hẹn giờ gửi)<br/><i>Nằm im 10' rồi tự chết</i>"]:::queue
        
        DLX["<b>release-dlx</b><br/>(Bưu cục trung chuyển)"]:::exchange
        
        PQ["<b>process-queue</b><br/>(Hộp thư cho worker)"]:::queue

        %% Luồng bên trong CloudAMQP
        DQ -- "Hết hạn (TTL)" --> DLX
        DLX --> PQ
    end

    %% Luồng kết nối chính
    P -- "publish({orderId:123})" --> DQ
    PQ -- "consume" --> C

    %% Ghi chú thêm
    style CloudAMQP fill:#ffffff,stroke:#333,stroke-width:1px

Loading

Từng thành phần:

Tên	Vai trò	Giải thích dễ hiểu
tixtac.main	Exchange (bưu cục gốc)	Nơi nhận tất cả thư gửi vào
tixtac.order.delay	Queue (hộp thư hẹn giờ)	Thư nằm đây 10 phút rồi tự "chết" (expire). Khi chết → chuyển sang bưu cục tixtac.release-dlx
tixtac.release-dlx	Exchange (bưu cục trung chuyển)	Nhận thư từ hộp hẹn giờ, chuyển tiếp sang tixtac.order.release-process
tixtac.order.release-process	Queue (hộp thư xử lý)	Worker ngồi đợi ở đây, có thư là xử lý ngay
tixtac.order.release-process.retry	Queue (hộp thư thử lại)	Nếu xử lý lỗi → gửi vào đây, đợi 10s rồi tự quay lại release-process
tixtac.order.release-process.dlq	DLQ (hộp thư chết hẳn)	Sau 3 lần thử vẫn lỗi → vứt vào đây, admin phải xem thủ công

---

📦 Message đơn giản hóa

Chỉ chứa đúng 1 thứ:

{ "orderId": 123 }

Tại sao chỉ cần orderId? Vì khi message đến tay worker sau 10 phút, tình trạng đơn hàng có thể đã thay đổi (khách đã thanh toán, hoặc giỏ hàng bị thay thế → re-publish timeout mới). Thay vì tin vào data trong message cũ, worker tự vào DB check lại trạng thái mới nhất.

---

🔄 Flow từng bước

Bước 1: Publish (gửi thư hẹn giờ)

File: purchase.service.ts

await publishOrderTimeout(responseData.order_id);

File: publisher.ts

export async function publishOrderTimeout(orderId: number) {
  const ch = await getChannel();
  const payload = JSON.stringify({ orderId });  // ← chỉ { orderId }
  ch.publish('tixtac.main', 'order-hold', Buffer.from(payload), {
    persistent: true,  // ← ghi xuống đĩa, không sợ restart
  });
}

Gửi message vào exchange tixtac.main với routing key order-hold. Message được route sang tixtac.order.delay.

Bước 2: Delay (nằm chờ)

Queue tixtac.order.delay có config:

• x-message-ttl: seatLockDuration * 1000 — message tự chết sau N giây
• x-dead-letter-exchange: tixtac.release-dlx — khi chết thì chuyển sang đây
• x-dead-letter-routing-key: release-task — với key này

Sau 10 phút, message "chết" → tự động chuyển sang tixtac.release-dlx → route sang tixtac.order.release-process.

Bước 3: Consume (worker xử lý)

File: consumer.ts

ch.consume('tixtac.order.release-process', async (msg) => {
  // 1. Parse message → lấy orderId
  const { orderId } = JSON.parse(msg.content.toString());

  try {
    // 2. Gọi service xử lý
    await orderService.releaseExpiredOrder(orderId);
    // 3. Thành công → ack (xóa message khỏi queue)
    ch.ack(msg);
  } catch (err) {
    // 4. Thất bại → retry hoặc DLQ
    if (retries < 3) {
      ch.sendToQueue('tixtac.order.release-process.retry', ...);
      // Retry queue có TTL 10s, sau 10s message lại quay về release-process
    } else {
      ch.sendToQueue('tixtac.order.release-process.dlq', ...);
      // DLQ — admin phải xem thủ công
    }
    ch.ack(msg);  // ← đã republish thành công thì ack bản gốc
  }
});

Bước 4: ReleaseExpiredOrder (logic chính)

File: order.service.ts

async releaseExpiredOrder(orderId) {
  return db.transaction(async (tx) => {
    const [order] = await tx.select().from(orders).where(...).for('update');

    // Guard 1: Không có đơn → thôi
    if (!order) return { releasedSeatIds: [] };

    // Guard 2: Đã paid/cancelled → thôi (idempotent)
    if (order.status !== 'pending') return { releasedSeatIds: [] };

    // Guard 3: Chưa hết hạn → thôi (message đến sớm do re-publish)
    if (order.expiresAt > now) return { releasedSeatIds: [] };

    // Đã pending + đã hết hạn → release ghế + hủy đơn
    await tx.update(seats).set({ status: 'available', lockedBy: null, ... });
    await tx.update(orders).set({ status: 'cancelled' });
  });
}

---

🧠 Điểm quan trọng

1. Tại sao message chỉ chứa { orderId }?

Vì giữa lúc publish và lúc consume (10 phút sau), đơn hàng có thể đã:

• Được thanh toán (status → paid) → Guard 2 bắt, bỏ qua
• Bị thay thế bởi giỏ hàng mới (re-publish timeout mới với expiresAt mới) → Guard 3 bắt, bỏ qua
• Bị hủy bởi logic khác → Guard 2 bắt, bỏ qua

→ Worker luôn check DB, không tin message.

2. Idempotent (chạy nhiều lần không sao)

Nhờ 3 guard trên, dù message bị deliver 2 lần (network issue) cũng không release ghế 2 lần.

3. Retry với backoff

• Lỗi hệ thống (DB down, network) → retry queue (10s delay, tối đa 3 lần)
• Lỗi nghiệp vụ (đơn đã cancelled) → không throw, chỉ return empty
• Sau 3 lần retry → DLQ (admin phải check)

4. Auto-restart worker

Nếu channel/connection đóng → worker tự reconnect sau:

• Lần 1: 2 giây
• Lần 2: 4 giây
• Lần 3: 8 giây
• ...
• Tối đa: 30 giây