CQRS và event sourcing: thêm read model trước, chỉ lưu event khi lịch sử chính là sản phẩm Nâng cao

Vấn đề

Một order service giữ bảng orders và order_items. Ba nhóm cùng đọc: dashboard vận hành (số đơn theo status mỗi giờ), search của support và trang “đơn của tôi”. Mỗi bên cần một hình dạng khác, nên query join năm bảng và tranh tài nguyên với checkout.

Bộ phận audit hỏi ngày càng nhiều: ai huỷ đơn 101, tổng tiền trước khi áp coupon là bao nhiêu. Một lệnh UPDATE đã ghi đè câu trả lời.

Hai pattern thường được đề xuất cùng lúc. Khi nào read model riêng giải quyết được chuyện đọc chậm, và khi nào lưu event thay cho state mới đáng chi phí?

Chọn gì? (tóm tắt trước)

Tình huống	Chọn	Điều kiện
Đọc chậm, nhưng một schema quan hệ vẫn đủ	CRUD: index, materialized view, read replica	Sửa query trước khi thêm model
Dashboard và search cần hình dạng hoặc kho khác	CQRS read model nuôi bằng outbox hoặc CDC	Màn hình chịu được trễ vài giây; projector idempotent
Audit nghĩa là “ai đã đổi gì”	CRUD + bảng lịch sử ghi trong cùng transaction	Lịch sử là bản ghi phụ, không phải nguồn sự thật
Rule nghiệp vụ phụ thuộc lịch sử; câu hỏi mới về quá khứ liên tục xuất hiện; dữ liệu kiểu sổ cái	Event sourcing, chỉ trong bounded context đó	Team gánh được schema evolution, erasure và rebuild projection

Khái niệm cốt lõi

CQRS (Command Query Responsibility Segregation) dùng một model để thay đổi dữ liệu và một model khác để đọc. Greg Young mô tả nó là tách một object thành hai: một bên chứa command, một bên chứa query. Định nghĩa này không đòi event, broker hay database thứ hai.

Event sourcing lưu mỗi thay đổi thành một event bất biến trong stream append-only của từng aggregate. State hiện tại là phép fold: từ đơn rỗng, áp OrderPlaced, ItemAdded, OrderPaid theo thứ tự version. Event log là nguồn sự thật; bảng chỉ là dữ liệu dẫn xuất.

	CRUD	CQRS không ES	Event sourcing (+ CQRS)
Nguồn sự thật	Row hiện tại	Row hiện tại	Event stream
Read model	Chính các bảng đó	Projection từ outbox/CDC	Projection từ event log
Lịch sử	Bị ghi đè nếu không sao lưu	Bị ghi đè nếu không sao lưu	Đầy đủ theo thiết kế
Concurrency	Row lock hoặc cột version	Như CRUD	Expected version theo stream

Fowler cảnh báo với đa số hệ thống, CQRS “adds risky complexity”. Azure gọi event sourcing là tốn kém khi chuyển vào hay chuyển ra: “for most systems and most parts of a system, traditional data management is sufficient”.

Đang tải diagram…

Với CQRS đơn thuần, phía ghi vẫn giữ orders và ghi một domain event vào outbox trong cùng transaction, tránh dual write. Với event sourcing, event chính là lệnh ghi; không còn row nào phải đồng bộ.

Cách hoạt động

Demo (examples/src/cqrs-event-sourcing/) là một event store nằm trong một bảng Postgres 17; dòng quan trọng nhất là constraint:

CREATE TABLE es_events (
  global_position bigserial PRIMARY KEY,
  stream_id       text  NOT NULL,          -- 'order-101'
  version         int   NOT NULL,          -- 1, 2, 3... trong từng stream
  data            jsonb NOT NULL,          -- cùng type, schema_version
  tx_id           xid8  NOT NULL DEFAULT pg_current_xact_id(),
  CONSTRAINT es_events_stream_version_key UNIQUE (stream_id, version)
);

Append = đọc max(version), so với version mà command đã load (expected version), insert expected + 1: optimistic locking theo stream, giống thiết kế quan hệ của Greg Young năm 2010.

PayOrder append version 4 ngay; read model chỉ thấy sau khi projector chạy

Phóng to sơ đồ

Web app → Command handlerPayOrder

◀ Trước▶ PhátTiếp ▶↺ Từ đầuBước 1/6

Khách thanh toán order-101. Command nói ý định, không nói cột nào cần sửa.

Bước	Phía ghi (es_events)	Phía đọc (es_order_summary)
1–2	version 3, fold ra placed	version 3
3–4	OrderPaid = version 4	version 3: cũ so với minVersion=4
5–6	không đổi	áp v4 và checkpoint trong một transaction: paid

Hai bên ghi, một version

Riêng phép kiểm max(version) vẫn bị race: hai session cùng đọc được 3. Unique index mới là thứ phân xử.

INSERT version 4 thứ hai chờ ở unique index rồi lỗi 23505

◀ Trước▶ PhátTiếp ▶↺ Từ đầuBước 1/6

Hai session cùng load order-101 ở version 3, status placed. Không ai giữ lock.

1. Hai session cùng fold order-101 ở version 3.
2. web-checkout insert version 4, chưa commit.
3. support-backoffice insert version 4 và chờ (Lock: transactionid).
4. web-checkout commit; support nhận 23505 → ConcurrencyError.
5. Support load lại và chạy lại rule: đơn đã paid thì refund, không huỷ.

Chi tiết vận hành

• Snapshot lưu state đã fold sau mỗi N event (demo dùng 3); load = snapshot + các event sau đó. Khi apply() đổi, xoá và tạo lại snapshot.
• Thứ tự projection: global_position được cấp lúc insert nhưng chỉ hiện ra lúc commit, nên đọc theo global_position > checkpoint có thể bỏ sót event vĩnh viễn (output bên dưới).
• Projector trong demo chỉ đọc tx_id < pg_snapshot_xmin(pg_current_snapshot()), sắp theo (tx_id, global_position). Mọi transaction dưới xmin đã kết thúc, nên phía sau checkpoint không thể mọc thêm event. Cái giá: một transaction dài bất kỳ trên server, kể cả ở database khác, làm projection đứng.
• Checkpoint và row read model commit cùng nhau; crash thì batch chạy lại vô hại. Projector nhận qua broker cần idempotent consumer.
• Rebuild = truncate read model, xoá checkpoint, replay. Đo thời gian trên dữ liệu cỡ production.
• Kafka không phải event store: không có append kèm expected version theo stream. Dùng Kafka để phát event ra ngoài.

Read-your-writes khi phía đọc là eventual consistency

Người dùng nhận ra độ trễ chủ yếu ngay sau thao tác của chính họ. Các cách, rẻ nhất trước:

1. Trả về version mới; client tự hiển thị kết quả của mình (optimistic UI).
2. Query kèm minVersion: chờ projection một chút, quá hạn thì đọc phía ghi.
3. Fold aggregate của chính người dùng từ stream; danh sách và search vẫn đọc projection.
4. Cập nhật một view quan trọng ngay trong transaction append, trả giá bằng độ trễ ghi.

Schema evolution và erasure

Event đã lưu không bao giờ đổi, nên hình dạng cũ sống mãi. ItemAdded v1 trong demo lưu price: "12.50"; một upcaster đổi nó thành v2 priceMinor: 1250 lúc đọc. Quy tắc tương thích cho consumer: event schema evolution.

Quyền xoá dữ liệu theo GDPR Article 17 va chạm với tính bất biến. Azure nêu hai cách: giữ dữ liệu cá nhân ngoài event và tham chiếu bằng ID, hoặc crypto-shredding: mã hoá field bằng key riêng cho từng người rồi xoá key.

Crypto-shredding làm ciphertext không đọc được; nó không xoá ciphertext. Như vậy có đủ cho yêu cầu xoá hay không là quyết định pháp lý. Plaintext trong read model, cache, log và backup cần quy trình xoá riêng.

Ví dụ thực tế

Read model cho search, event sourcing chỉ cho sổ cái thanh toán

Search và dashboard thành projection nuôi bằng outbox sẵn có; checkout giữ nguyên orders. Trang lịch sử đơn poll kèm minVersion sau khi bấm “Huỷ”.

Sau đó audit hỏi “các khoản refund như đã biết vào từng ngày”, tức temporal query. Chỉ sổ cái thanh toán chuyển sang event sourcing, và event chỉ mang customerId, không mang email.

Chạy demo (Postgres 17.11 từ examples/docker-compose.yml, port 5433):

docker compose -f examples/docker-compose.yml up -d postgres
npx --prefix examples ts-node examples/src/cqrs-event-sourcing/demo.ts
npx --prefix examples ts-node examples/src/cqrs-event-sourcing/gap-demo.ts

Output thật, đã rút gọn (ciphertext khác nhau mỗi lần chạy):

== 3. Concurrency: customer pays while support cancels, both loaded at version 3 ==
support session while web is uncommitted: {"wait_event_type":"Lock","wait_event":"transactionid","state":"active"}
web-checkout: COMMIT version 4 (OrderPaid)
support-backoffice: ConcurrencyError: order-101: duplicate key value violates unique constraint "es_events_stream_version_key" (Key (stream_id, version)=(order-101, 4) already exists.)
retry after reload at version 4: cannot cancel an order in status paid; start a refund instead

== 4. Snapshot + read-your-writes ==
command returned version 5; read model has: status=placed items=3 total_minor=22400 email=lan@example.com version=3
read model caught up after 1 projector pass(es): status=shipped items=3 total_minor=22400 email=lan@example.com version=5

== 6. Erasure request for cust-7: crypto-shred, then rebuild the projection ==
rebuild replayed 5 events -> status=shipped items=3 total_minor=22400 email=[erased] version=5
event 1 still holds ciphertext: QzOdz/ffJ+GQIKUQ...

gap-demo.ts: vì sao đọc theo global_position bỏ sót event

naive reader sees positions: [ 2 ] -> checkpoint = 2
xmin-guarded reader sees positions: [] (waits for the open transaction)
after slow COMMIT, naive "> 2" sees: [] -> position 1 is skipped forever
xmin-guarded reader now sees (tx order): [ 1, 2 ]

Position 1 thuộc về một transaction commit sau position 2.

Còn thiếu cho production: khử trùng lặp command, metric độ trễ, KMS thật thay cho bảng key.

Trade-offs

Event sourcing mang lại

• Lịch sử và ý định đầy đủ: audit, temporal query, debug bằng replay.
• Read model mới dựng từ event cũ mà không đụng phía ghi.
• Concurrency theo từng stream với expected version rõ ràng.

Event sourcing đòi trả

• Mọi hình dạng event đã lưu phải được hỗ trợ mãi mãi.
• Không còn SQL tuỳ ý trên state hiện tại; projection trễ và phải rebuild được.
• Erasure và sửa dữ liệu sai cần cách làm thiết kế sẵn.
• Tốn kém cả khi áp dụng lẫn khi bỏ.

CQRS không event sourcing chỉ tốn thêm một projector, giám sát độ trễ và idempotency, đồng thời giữ được SQL trên nguồn sự thật.

Checklist production

1. Liệt kê màn hình nào cần hình dạng nào và chịu được dữ liệu cũ bao lâu.
2. Bắt đầu bằng CQRS trên bảng sẵn có; chỉ áp event sourcing cho bounded context mà lịch sử quyết định nghiệp vụ.
3. Bắt buộc UNIQUE (stream_id, version) và map vi phạm thành conflict có thể retry bằng cách load lại và quyết lại.
4. Đọc projection theo thứ tự an toàn với commit, lưu checkpoint cùng row read model, alert độ trễ projection tính bằng giây.
5. Đánh version mọi loại event; test upcaster với fixture của từng version đã lưu.
6. Quyết định dữ liệu cá nhân nằm đâu trước event đầu tiên; xin pháp chế duyệt crypto-shredding.
7. Diễn tập và đo thời gian rebuild toàn bộ projection trên dữ liệu cỡ production.

Sai lầm thường gặp

Retry conflict bằng cách gửi lại đúng các event cũ

✕ Hay gặp

catch (e) {
  if (e instanceof ConcurrencyError) await append(stream, actualVersion, sameEvents);
}

✓ Nên viết

catch (e) {
  if (e instanceof ConcurrencyError) return handle(command); // load lại, fold, kiểm rule lại
}

Vì sao

Quyết định đã dựa trên state cũ; rule phải chạy lại.

Cái bẫy thứ hai là coi event log như log thay đổi cột.

Sai lầm thường gặp

OrderStatusChanged { status: "cancelled" } ở khắp nơi. Event dạng state mất ý định, projection không phân biệt được khách tự huỷ với huỷ do gian lận. Azure khuyên dùng event mang ý định: OrderCancelledByCustomer { reason }.

Cách hiểu sai	Hệ quả	Cần làm
“CQRS nghĩa là event sourcing”	Dựng event store chỉ để đọc nhanh hơn	Làm projection từ outbox trước
“Audit bắt buộc event sourcing”	Viết lại cả service cho một bảng lịch sử	Bảng lịch sử hoặc CDC sang audit log
“Kafka topic = event store”	Hai bên cùng append event mâu thuẫn	Event store có expected version; Kafka để phát ra

Tự kiểm tra

1. Vì sao phép kiểm max(version) cần unique constraint đứng sau?
2. Projector đọc theo global_position > checkpoint. Thứ tự commit nào làm mất event?
3. Sau ConcurrencyError của CancelOrder, vì sao lần retry có thể kết thúc mà không append gì?

Tóm tắt

Tóm tắt

• CQRS tách model đọc và ghi; event sourcing biến event log thành nguồn sự thật. Đó là hai quyết định riêng.
• Đọc chậm thường cần projection nuôi bằng outbox hoặc CDC, không cần event store.
• Event sourcing đáng giá khi lịch sử quyết định nghiệp vụ; áp theo từng bounded context.
• Event store cần append kèm expected version, projection theo thứ tự an toàn với commit và rebuild đã đo giờ; lên kế hoạch upcasting và erasure trước event đầu tiên.

Tham khảo

• Azure — CQRS pattern: một kho hay hai kho, đồng bộ qua outbox, khi nào không nên dùng.
• Azure — Event Sourcing pattern: concurrency, snapshot, upcasting, crypto-shredding, broker khác event store.
• Martin Fowler — CQRS: vì sao đa số hệ thống nên giữ CRUD.
• Martin Fowler — Event Sourcing: rebuild, temporal query, replay với external gateway.
• Greg Young — CQRS Documents: định nghĩa gốc, lưu event trên DB quan hệ, rolling snapshot.
• PostgreSQL 17 — System Information Functions: pg_current_xact_id, ý nghĩa xmin của snapshot.
• PostgreSQL 17 — Sequence functions: sequence có lỗ hổng; gap-demo.ts cho thấy nó cũng không theo thứ tự commit.
• GDPR Article 17: quyền được xoá dữ liệu.