CQRS và Event Sourcing: Khi nào và vì sao nên dùng

Một khách hàng nhấn Chuyển tiền hai lần vì phản hồi đầu tiên quá chậm. Cùng lúc, khoản thanh toán định kỳ cũng trừ tiền từ tài khoản đó. Vấn đề không phải controller cập nhật cột balance nhanh đến đâu. Hệ thống phải chứng minh tiền không tự sinh ra hay bị chi hai lần, giải thích được quyết định đã chấp nhận và vẫn trả lời dashboard mà không phát lại nhiều năm lịch sử.

CQRS và event sourcing thường được giới thiệu như một cặp giải pháp cho bài toán này. Chúng có thể phối hợp rất tốt, nhưng giải quyết hai vấn đề khác nhau và tạo ra hai nhóm chi phí khác nhau. Muốn dùng đúng, trước hết phải tách bạch hai khái niệm.

Hai mẫu kiến trúc, hai quyết định #

Command Query Responsibility Segregation (CQRS) dùng mô hình khác nhau cho thao tác ghi và đọc. Phía command tiếp nhận ý định như OpenAccount hay TransferFunds, bảo vệ bất biến nghiệp vụ rồi trả về thành công hoặc thất bại. Phía query trả dữ liệu đã được định hình cho một màn hình, báo cáo, chỉ mục tìm kiếm hoặc API consumer. Sự phân tách này hoàn toàn có thể nằm trong cùng một tiến trình và dùng chung một cơ sở dữ liệu giao dịch.

Event sourcing lưu các domain event mà aggregate đã chấp nhận làm nguồn sự thật. Thay vì ghi đè một dòng tài khoản, hệ thống nối thêm các sự thật như AccountOpened, FundsDeposited và FundsTransferred. Trạng thái hiện tại được suy ra bằng cách lần lượt áp dụng các sự kiện theo đúng thứ tự. Event log là dữ liệu có thẩm quyền; bảng số dư chỉ là một projection có thể xây dựng lại.

Có thể dùng CQRS mà không event sourcing: write schema chuẩn hóa kết hợp read schema phi chuẩn hóa là thiết kế phổ biến. Cũng có thể event-source một aggregate rồi truy vấn trạng thái đã dựng lại, dù cách đó kém hiệu quả khi tải đọc tăng. Không mẫu nào mặc nhiên đòi hỏi microservice, message broker hay eventual consistency ở mọi nơi.

Aggregate là ranh giới nhất quán #

Aggregate theo event sourcing không mở setter tùy ý. Nó nhận command, kiểm tra invariant trên trạng thái dựng từ event đã commit rồi trả về event mới. Áp dụng event không thực hiện I/O. Tách decide khỏi evolve làm rõ quy tắc và giúp replay có tính tất định.

Xét một tài khoản có loại tiền tệ không đổi, số tiền luôn phải dương và không được rút quá số dư khả dụng. Tiền được biểu diễn bằng đơn vị nhỏ nhất dưới dạng số nguyên để tránh sai số dấu phẩy động.

export type AccountEvent =
  | {
      type: 'AccountOpened';
      accountId: string;
      currency: string;
      occurredAt: string;
    }
  | {
      type: 'FundsDeposited';
      accountId: string;
      amountMinor: number;
      reference: string;
      occurredAt: string;
    }
  | {
      type: 'FundsTransferred';
      accountId: string;
      destinationId: string;
      amountMinor: number;
      transferId: string;
      occurredAt: string;
    };

export type AccountState =
  | { status: 'empty' }
  | {
      status: 'open';
      accountId: string;
      currency: string;
      balanceMinor: number;
      completedTransferIds: ReadonlySet<string>;
    };

export const initialAccount: AccountState = { status: 'empty' };

export function evolve(
  state: AccountState,
  event: AccountEvent,
): AccountState {
  switch (event.type) {
    case 'AccountOpened':
      return {
        status: 'open',
        accountId: event.accountId,
        currency: event.currency,
        balanceMinor: 0,
        completedTransferIds: new Set(),
      };
    case 'FundsDeposited':
      if (state.status !== 'open') throw new Error('Account is not open');
      return { ...state, balanceMinor: state.balanceMinor + event.amountMinor };
    case 'FundsTransferred': {
      if (state.status !== 'open') throw new Error('Account is not open');
      const completedTransferIds = new Set(state.completedTransferIds);
      completedTransferIds.add(event.transferId);
      return {
        ...state,
        balanceMinor: state.balanceMinor - event.amountMinor,
        completedTransferIds,
      };
    }
  }
}

export function rehydrate(events: readonly AccountEvent[]): AccountState {
  return events.reduce(evolve, initialAccount);
}

export function decideTransfer(
  state: AccountState,
  command: {
    destinationId: string;
    amountMinor: number;
    transferId: string;
    now: string;
  },
): AccountEvent[] {
  if (state.status !== 'open') throw new Error('Account is not open');
  if (!Number.isSafeInteger(command.amountMinor) || command.amountMinor <= 0) {
    throw new Error('Transfer amount must be a positive integer');
  }
  if (state.completedTransferIds.has(command.transferId)) return [];
  if (state.balanceMinor < command.amountMinor) {
    throw new Error('Insufficient funds');
  }
  return [{
    type: 'FundsTransferred',
    accountId: state.accountId,
    destinationId: command.destinationId,
    amountMinor: command.amountMinor,
    transferId: command.transferId,
    occurredAt: command.now,
  }];
}

transferId khiến một command được gửi lại trở nên idempotent trong phạm vi aggregate. Quan trọng hơn, invariant về số dư được kiểm tra bên trong một ranh giới nhất quán duy nhất. Một giao dịch đi qua hai tài khoản cần quy trình rõ ràng: ghi nợ aggregate nguồn, sau đó ghi có aggregate đích bằng handler idempotent, kèm cơ chế bù trừ và đối soát cho lỗi một phần. Giả vờ rằng hai stream độc lập là một aggregate nguyên tử chỉ che giấu distributed transaction.

Xử lý command và optimistic concurrency #

Quy trình dựng trạng thái, quyết định rồi append chỉ an toàn nếu không ai thay đổi stream giữa lần đọc và lần ghi. Vì vậy event store append theo expected version. Lệnh append thành công nếu stream vẫn ở đúng version đó; nếu không, store từ chối thao tác. Handler có thể tải lại và thử command lần nữa khi ý định vẫn còn hợp lệ, hoặc trả conflict cho caller.

import {
  decideTransfer,
  rehydrate,
  type AccountEvent,
} from './account.js';

export interface StoredEvent<T> {
  streamId: string;
  streamVersion: number;
  globalPosition: number;
  event: T;
}

export interface EventStore {
  readStream(streamId: string): Promise<readonly StoredEvent<AccountEvent>[]>;
  append(
    streamId: string,
    expectedVersion: number,
    events: readonly AccountEvent[],
  ): Promise<void>;
}

export class ConcurrencyError extends Error {}

export async function transferFunds(
  store: EventStore,
  command: {
    accountId: string;
    destinationId: string;
    amountMinor: number;
    transferId: string;
    now: string;
  },
): Promise<void> {
  const streamId = `account-${command.accountId}`;
  const stored = await store.readStream(streamId);
  const state = rehydrate(stored.map(({ event }) => event));
  const newEvents = decideTransfer(state, command);

  if (newEvents.length === 0) return;
  await store.append(streamId, stored.length, newEvents);
}

Trong production, event envelope còn chứa event ID, schema version, correlation ID, causation ID, actor, tenant và tracing. Event store phải áp đặt khóa duy nhất (streamId, streamVersion) trong cùng transaction với append. Đưa event sang broker bằng transactional outbox hoặc durable subscription, không dùng chuỗi mong manh “commit rồi mới publish”.

Conflict là thông tin nghiệp vụ, không chỉ là nhiễu hạ tầng. Conflict có thể báo rằng số dư khả dụng đã đổi, vì thế tự động lặp lại quyết định cũ là sai. Hãy rehydrate rồi chạy decideTransfer lần nữa để invariant được đánh giá trên các sự thật mới nhất.

Projection biến lịch sử thành dữ liệu hữu dụng #

Stream phía command được cố ý giữ hẹp. Trong khi đó, dashboard muốn tên tài khoản, số dư đã định dạng, hoạt động gần nhất và số giao dịch đang chờ chỉ bằng một lần đọc. Projection tiêu thụ event log có thứ tự và duy trì read model phục vụ đúng nhu cầu đó.

import type { StoredEvent } from './command-handler.js';
import type { AccountEvent } from './account.js';

interface AccountSummary {
  accountId: string;
  currency: string;
  balanceMinor: number;
  lastActivityAt: string;
}

export interface ProjectionStore {
  transaction<T>(work: (tx: ProjectionTransaction) => Promise<T>): Promise<T>;
}

export interface ProjectionTransaction {
  getAccount(accountId: string): Promise<AccountSummary | undefined>;
  putAccount(account: AccountSummary): Promise<void>;
  hasProcessed(eventId: string): Promise<boolean>;
  markProcessed(eventId: string, position: number): Promise<void>;
}

export async function projectAccountSummary(
  db: ProjectionStore,
  eventId: string,
  stored: StoredEvent<AccountEvent>,
): Promise<void> {
  await db.transaction(async (tx) => {
    if (await tx.hasProcessed(eventId)) return;

    const event = stored.event;
    if (event.type === 'AccountOpened') {
      await tx.putAccount({
        accountId: event.accountId,
        currency: event.currency,
        balanceMinor: 0,
        lastActivityAt: event.occurredAt,
      });
    } else {
      const current = await tx.getAccount(event.accountId);
      if (!current) throw new Error(`Missing account ${event.accountId}`);
      const delta = event.type === 'FundsDeposited'
        ? event.amountMinor
        : -event.amountMinor;
      await tx.putAccount({
        ...current,
        balanceMinor: current.balanceMinor + delta,
        lastActivityAt: event.occurredAt,
      });
    }

    await tx.markProcessed(eventId, stored.globalPosition);
  });
}

Projection thường nhận event theo cơ chế at-least-once, nên checkpoint và read model phải commit nguyên tử. Khử trùng lặp bằng event ID giúp retry an toàn. Consumer phải giữ thứ tự khi cần, dừng ở poison event, công khai độ trễ và có quy trình dead-letter hoặc quarantine. Với read-your-own-writes, command có thể trả global position để client chờ projection bắt kịp, hoặc giao diện hiển thị lạc quan từ kết quả vừa chấp nhận.

Replay và tiến hóa schema #

Replay vừa là năng lực mạnh nhất, vừa là cạnh vận hành sắc của event sourcing. Projection mới có thể đọc log từ position 0 để thêm báo cáo mà không migration write model; cơ chế đó cũng sửa read database bị hỏng. Lịch sử lớn cần throttling, phân vùng, chỉ số tiến độ và hạ tầng riêng để replay không làm nghẽn consumer trực tiếp.

Snapshot của aggregate là cache, không phải nguồn sự thật. Có thể lưu trạng thái cùng stream version sau mỗi vài trăm event, tải snapshot tương thích mới nhất rồi fold phần event tiếp theo. Nếu deployment mới không đọc được snapshot, hãy bỏ nó và replay stream.

Event là sự thật vĩnh viễn, nhưng biểu diễn TypeScript của nó không vĩnh viễn. Không sửa payload cũ tại chỗ. Nên ưu tiên bổ sung tương thích ngược, version hóa tên event hoặc envelope khi có thay đổi phá vỡ, và upcast payload cũ sang cấu trúc in-memory hiện tại lúc đọc. Upcaster cần thuần, nối được thành chuỗi và được kiểm thử với fixture lấy từ lịch sử production. Thay đổi ngữ nghĩa khó hơn thay đổi cú pháp: nếu FundsTransferred trước đây có nghĩa là “đã yêu cầu” nhưng nay lại là “đã quyết toán,” hãy tạo event mới thay vì định nghĩa lại sự thật cũ.

Yêu cầu xóa dữ liệu cần được thiết kế trước khi áp dụng event sourcing. Mã hóa bằng khóa riêng cho từng chủ thể cho phép hủy khóa để một số trường cá nhân không còn đọc được; tokenization có thể giữ dữ liệu nhạy cảm bên ngoài event log. Không kỹ thuật nào biện minh cho việc thu thập dữ liệu không cần thiết vào một kho bất biến.

Kiểm thử và các kiểu thất bại #

Mô hình aggregate thuần cho phép viết test ngắn gọn: cho trước các event, khi command đến, kỳ vọng event mới hoặc lỗi. Cần kiểm tra cả luồng đúng lẫn invariant: không đủ tiền, số tiền bằng không, transferId trùng, tài khoản chưa mở và giá trị biên. Property-based testing đặc biệt hữu ích để chứng minh mọi chuỗi command được chấp nhận đều không khiến số dư âm.

Sau đó kiểm thử riêng các điểm nối hạ tầng:

• hai writer dùng cùng expected version chỉ có đúng một append thành công;
• mỗi projection xử lý event theo cách idempotent và cập nhật checkpoint nguyên tử;
• replay toàn bộ log vào database rỗng tạo ra read model giống hệ thống trực tiếp;
• mọi fixture event lịch sử đều upcast và fold thành công;
• handler phục hồi đúng khi crash trước hoặc sau append, publish và commit checkpoint.

Các lỗi phổ biến xuất hiện khi những điểm nối này bị xem nhẹ. Một nhóm có thể gọi các dòng audit mutable là “event,” publish trước khi transaction commit, để projector thực hiện side effect không idempotent, hoặc buộc mọi service phụ thuộc vào một global event schema khổng lồ. Stream dài làm command chậm; aggregate quá nóng liên tục xung đột; projection lag khiến người dùng bối rối; số read model tăng vô hạn biến migration và giám sát thành gánh nặng.

Về vận hành, nhóm phải sở hữu chính sách lưu event, backup, diễn tập restore, subscription offset, công cụ replay, tương thích schema, observability và on-call. Lịch sử giúp điều tra, nhưng sự cố vẫn khó xử lý nếu kỹ sư không thể xem stream, liên kết command với event hay đo projection lag.

Điều cần ghi nhớ #

Hãy chọn CQRS khi invariant của mô hình ghi và hình dạng của mô hình đọc thực sự kéo thiết kế theo hai hướng khác nhau. Hãy chọn event sourcing khi việc lưu quyết định domain thành lịch sử có thứ tự tạo ra giá trị lâu dài: audit theo thời gian, dựng lại trạng thái, bổ sung projection mới, hoặc hỗ trợ quy trình nghiệp vụ phụ thuộc vào chuyện đã xảy ra chứ không chỉ trạng thái hiện tại.

Nên ưu tiên transaction truyền thống khi domain chủ yếu là CRUD, lịch sử ít giá trị nghiệp vụ, nhu cầu đọc nhất quán tức thì chiếm ưu thế, hoặc đội ngũ chưa thể đầu tư công cụ vận hành. Lộ trình áp dụng hợp lý bắt đầu từ một bounded context và một aggregate có giá trị cao, với event contract rõ ràng cùng projection có thể xóa và dựng lại. Hãy đo tỷ lệ conflict, độ dài stream, projection lag, thời gian replay và công sức của operator trước khi mở rộng.

Câu hỏi quyết định không phải CQRS và event sourcing có đủ tinh vi hay không. Câu hỏi là giá trị của lịch sử và việc tách mô hình có bù được chi phí lâu dài hay không. Khi câu trả lời là có, chúng tạo ra các ranh giới rất rõ: command diễn đạt ý định, aggregate bảo vệ invariant, event ghi lại sự thật đã chấp nhận và projection biến sự thật thành dữ liệu hữu dụng. Khi câu trả lời là không, một mô hình giao dịch được thiết kế tốt mới là lựa chọn trưởng thành hơn.