CQRS và Event Sourcing

Hai khái niệm này hay bị nói chung một câu, nhưng là hai thứ khác nhau. Ta đi từ trang "Đơn hàng của tôi" trong shop-app.

1. Bài toán

Màn hình "My orders" cần hiển thị cho mỗi đơn: tên sản phẩm, ảnh, trạng thái thanh toán, trạng thái giao hàng. Dữ liệu nằm ở 4 service:

order_db       orders, order_items
payment_db     payments
inventory_db   reservations
shipping_db    shipments

Write side (đặt hàng) cần business rule chặt. Read side (xem đơn) chiếm 95% traffic, cần nhanh và cần dữ liệu ghép từ nhiều nơi.

2. Vấn đề

• Mỗi lần mở trang phải gọi 4 service rồi ghép (API composition): chậm, một service chết là trang lỗi.
• Không join được giữa 4 DB khác nhau.
• Thêm index/cột denormalized vào bảng orders để phục vụ read thì làm bẩn model ghi, khóa ghi chậm hơn.

3. Giải pháp: CQRS

CQRS = Command Query Responsibility Segregation: tách model ghi và model đọc.

Commands (PlaceOrder, CancelOrder)
      ↓
Write model (order_db, có business rule) ── events (Outbox/CDC) ──►  Projector
                                                                       │
Queries (GET /my-orders)  ◄──────────  Read model (order_view)  ◄──────┘
                                       denormalized, 1 bảng / Elasticsearch

• Read model / projection: bảng được dựng sẵn đúng hình dạng màn hình cần, cập nhật bằng event.
• Read model là eventually consistent: vừa thanh toán xong, vài trăm ms sau trang mới đổi trạng thái.
• CQRS không bắt buộc 2 database hay Event Sourcing. Bản nhẹ nhất: cùng DB, chỉ tách class command/query.

4. Code từng bước

Bước 1: Read model

CREATE TABLE order_view (
    order_id        VARCHAR(50) PRIMARY KEY,
    user_id         VARCHAR(50) NOT NULL,
    total           NUMERIC(12,2),
    order_status    VARCHAR(20),
    payment_status  VARCHAR(20),
    shipping_status VARCHAR(20),
    last_event_seq  BIGINT NOT NULL DEFAULT 0,
    updated_at      TIMESTAMPTZ
);
CREATE INDEX idx_order_view_user ON order_view (user_id, updated_at DESC);

Bước 2: Projector nghe event từ nhiều service

@Component
@RequiredArgsConstructor
public class OrderViewProjector {
    private final OrderViewRepository views;
    private final InboxRepository inbox;

    @KafkaListener(topics = {"order-events", "payment-events", "shipping-events"},
                   groupId = "order-view-projector")
    @Transactional
    public void on(DomainEvent e) {
        if (inbox.insertIgnore(e.eventId()) == 0) return;         // idempotent
        OrderView v = views.findById(e.orderId()).orElseGet(() -> new OrderView(e.orderId()));
        if (e.sequence() <= v.getLastEventSeq()) return;          // event cũ, bỏ
        switch (e) {
            case OrderPlaced p     -> v.placed(p.userId(), p.total());
            case PaymentCompleted p -> v.setPaymentStatus("PAID");
            case OrderShipped s    -> v.setShippingStatus("SHIPPED");
            default -> { }
        }
        v.setLastEventSeq(e.sequence());
        views.save(v);
    }
}

DomainEvent là sealed interface (Java 21 pattern matching cho switch). Projector phải idempotent vì event đến at-least-once (bài 14.7).

Bước 3: Query side chỉ đọc

@RestController
@RequiredArgsConstructor
class MyOrdersQueryController {
    private final OrderViewRepository views;

    @GetMapping("/my-orders")
    List<OrderView> myOrders(@AuthenticationPrincipal Jwt jwt) {
        return views.findTop20ByUserIdOrderByUpdatedAtDesc(jwt.getSubject());
    }
}

Bước 4: Test

curl -X POST localhost:8081/orders -d '{"cartId":"CART-123"}' -H 'Content-Type: application/json'
curl localhost:8085/my-orders -H "Authorization: Bearer $TOKEN"
# ngay lập tức có thể chưa thấy đơn; ~vài trăm ms sau mới có → eventual consistency

5. Cách hoạt động

        COMMAND SIDE                          QUERY SIDE
POST /orders                              GET /my-orders
     │                                          ▲
     ▼                                          │
order-service ── outbox ──► Kafka ──► Projector ─► order_view
payment-service ─ outbox ──►  │
shipping-service  outbox ──►  ┘
                         (rebuild: reset offset, replay lại toàn bộ topic)

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Nên: read/write khác nhau rõ (tỷ lệ, hình dạng dữ liệu), cần view ghép từ nhiều service, cần search/report riêng.
• Không nên: CRUD đơn giản; màn hình bắt buộc read-after-write (ví dụ vừa đổi mật khẩu); team chưa quen vận hành event pipeline.
• Read-after-write cho chính user vừa ghi: trả kết quả ngay từ command response, hoặc đọc write model cho request đó.

Không. Publish domain event không đồng nghĩa event log là source of truth. Ở shop-app, bảng orders vẫn là sự thật; event chỉ là thông báo.

1. Event Sourcing là gì

Truyền thống (lưu state):        Event Sourcing (lưu chuỗi sự kiện):
Account.balance = 900            AccountOpened   +1000
                                 MoneyDebited    -100
                                        ↓ replay
                                 balance = 900

Ta không UPDATE row. Ta append event bất biến vào event store, state hiện tại được tính bằng cách replay.

2. Code tối giản

CREATE TABLE event_store (
    aggregate_id VARCHAR(50) NOT NULL,
    version      BIGINT      NOT NULL,
    event_type   VARCHAR(100) NOT NULL,
    payload      JSONB       NOT NULL,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (aggregate_id, version)      -- optimistic concurrency
);

public class OrderAggregate {
    private String id;
    private OrderStatus status;
    private long version;

    public static OrderAggregate replay(List<OrderEvent> history) {
        OrderAggregate a = new OrderAggregate();
        history.forEach(a::apply);
        return a;
    }

    public OrderEvent cancel(String reason) {
        if (status == OrderStatus.SHIPPED) throw new IllegalStateException("Đã giao, không hủy được");
        return new OrderCancelled(id, version + 1, reason);   // chỉ sinh event
    }

    private void apply(OrderEvent e) {
        switch (e) {
            case OrderPlaced p    -> { id = p.orderId(); status = OrderStatus.PENDING; }
            case OrderConfirmed c -> status = OrderStatus.CONFIRMED;
            case OrderCancelled c -> status = OrderStatus.CANCELLED;
        }
        version = e.version();
    }
}

Hai request cùng append version = 6 cho ORDER-123: request sau vi phạm primary key → reload và thử lại. Đó là optimistic concurrency.

3. Được gì, mất gì

• Được: audit trail đầy đủ, xem lại state ở bất kỳ thời điểm, dựng read model mới bằng replay, event có sẵn để publish.
• Mất: query state hiện tại phải qua projection (gần như bắt buộc CQRS); event schema evolution (event cũ sống mãi, cần upcaster); xóa dữ liệu cá nhân (GDPR) khó với log bất biến; team cần tư duy khác hẳn CRUD.

Vì vậy Event Sourcing không phải mặc định. Hợp với domain mà lịch sử chính là nghiệp vụ: sổ cái kế toán, ví điện tử, đặt chỗ. Order của shop-app thông thường không cần.

Đúng là vấn đề thật. Giải pháp là snapshot.

event 1 … event 10000 → SNAPSHOT(version=10000, state={...}) → event 10001 … 10037

load(id):
  snap   = snapshotRepo.latest(id)                 // state tại version 10000
  events = eventStore.after(id, snap.version())    // chỉ 37 event
  return OrderAggregate.from(snap).applyAll(events)

• Snapshot chỉ là cache tối ưu, không phải source of truth. Xóa snapshot đi vẫn replay được từ đầu.
• Chụp snapshot mỗi N event (ví dụ 500) hoặc async.
• Đổi cấu trúc class aggregate thì snapshot cũ có thể không dùng được: bỏ và chụp lại.

Read model sai (bug projector)? Sửa code projector, xóa bảng view, replay toàn bộ event từ đầu (reset consumer offset hoặc đọc event store) để dựng lại. Đây là lợi thế lớn: read model có thể vứt đi và dựng lại bất kỳ lúc nào. Trong lúc rebuild, dựng bảng mới song song (order_view_v2) rồi chuyển traffic, tránh trang bị trống.

Q1. CQRS là gì, khi nào nên dùng?

🇻🇳 Tiếng Việt:

CQRS tách model ghi và model đọc. Write model giữ business rule, read model là projection denormalized được cập nhật bằng event, tối ưu cho query. Dùng khi read và write khác nhau rõ về tải hoặc hình dạng dữ liệu, hoặc cần view ghép từ nhiều service. Đổi lại read model eventually consistent và có thêm pipeline phải vận hành, nên CRUD đơn giản thì không cần.

🇬🇧 English:

CQRS separates the write model from the read model. The write model enforces business rules; read models are denormalized projections updated from events and optimized for queries. It's worth it when reads and writes differ a lot in load or shape, or when a view combines data from several services. The trade-off is eventual consistency on reads and an extra pipeline to operate, so simple CRUD doesn't need it.

Q2. Event Sourcing là gì? Khác gì publish event lên Kafka?

🇻🇳 Tiếng Việt:

Event Sourcing lưu chuỗi event bất biến làm source of truth, state hiện tại được tính bằng replay. Publish event lên Kafka chỉ là thông báo thay đổi, còn sự thật vẫn nằm ở bảng state. Event Sourcing cho audit và time travel tốt, nhưng khó ở schema evolution, query và xóa dữ liệu, nên không phải lựa chọn mặc định.

🇬🇧 English:

Event Sourcing stores an append-only sequence of events as the source of truth, and current state is derived by replaying them. Publishing events to Kafka just notifies others; the state table is still the truth. Event Sourcing gives a full audit trail and time travel, but schema evolution, querying and data deletion get harder, so it shouldn't be the default.

Q3. Snapshot dùng để làm gì?

🇻🇳 Tiếng Việt:

Snapshot lưu state của aggregate tại một version để không phải replay toàn bộ lịch sử khi load. Load bằng snapshot gần nhất cộng các event sau đó. Snapshot chỉ là tối ưu hiệu năng, có thể xóa và tạo lại từ event.

🇬🇧 English:

A snapshot stores an aggregate's state at a given version so we don't replay the whole history on every load. We load the latest snapshot plus the events after it. It's only a performance optimization and can be rebuilt from the events.

Cách nhớ ngắn gọn

• CQRS = tách đọc/ghi; Event Sourcing = event là sự thật. Dùng CQRS không cần ES; dùng ES gần như cần CQRS.
• Outbox = reliability pattern, CDC = cơ chế capture log, Event Sourcing = persistence model.
• Read model: idempotent, eventual, vứt đi dựng lại bằng replay.
• Snapshot = cache để replay ngắn. Không phải mặc định cho mọi service.