Observer và domain event: OrderPaid bắn trong process bảo đảm được gì, và khi nào cần outbox Trung cấp

Vấn đề

Khi cổng thanh toán xác nhận đơn 101 (250.000 VND), OrderService.markPaid() commit đơn hàng. Ba team muốn phản ứng: gửi email biên nhận, cộng 250 điểm loyalty, và ghi event analytics purchase.

Gọi thẳng ba service từ markPaid() buộc luồng đơn hàng vào mọi team. Thêm phản ứng thứ tư là phải sửa và test lại code đơn hàng. Observer đảo chiều phụ thuộc: OrderService công bố OrderPaid, mỗi team tự đăng ký listener của mình.

Tách rời như vậy đặt ra các câu hỏi bài này trả lời: khi emit() return, cái gì đã chạy? Listener throw thì sao? Và email sẽ ra sao nếu process chết ngay sau commit?

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

Tình huống	Chọn	Điều kiện
Bên gọi cần kết quả để trả response	Gọi trực tiếp, không dùng event	Event không trả kết quả nghiệp vụ
Phản ứng best-effort trong cùng process (analytics, xoá cache)	Event trong process (EventEmitter, @nestjs/event-emitter)	Mất khi crash cũng chấp nhận được; mỗi listener tự bắt lỗi
Phản ứng phải commit hoặc rollback cùng đơn (điểm nằm cùng DB)	Domain event dispatch trong transaction, hoặc gọi trực tiếp	Cùng database, cùng transaction
Phản ứng không được mất (email biên nhận, service khác)	Integration event qua transactional outbox	Relay + broker + idempotent consumer

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

Observer có hai vai: subject giữ danh sách subscriber và báo cho họ, observer phản ứng. Khi có event bus ở giữa, OrderService publish còn listener subscribe; không bên nào import bên nào.

Một event là sự thật ở thì quá khứ: OrderPaid, không phải SendEmail. Event bất biến và mang đủ thứ listener cần (orderId, customerId, amountMinor). Command yêu cầu một handler làm việc; event có thể có không hoặc nhiều listener.

Domain event và integration event

	Domain event	Integration event
Đối tượng nhận	Cùng bounded context, cùng process	Service khác, bounded context khác
Đường đi	Dispatcher trong bộ nhớ	Broker (Kafka, RabbitMQ), bảng outbox
Thời điểm	Đồng bộ hoặc bất đồng bộ, dispatch trước hoặc sau commit	Luôn bất đồng bộ, chỉ sau khi dữ liệu đã commit
Sống sót qua crash	Không	Có, nếu ghi cùng transaction với dữ liệu
Schema	Nội bộ, đổi theo code	Hợp đồng công khai, có version

Tài liệu của Microsoft coi hai loại là cùng một ý tưởng, khác cách cài đặt: domain event handler thường tạo ra integration event, và integration event chỉ được gửi đi khi entity đã được lưu.

Cách hoạt động

Demo đăng ký listener email, loyalty và analytics trên một Node EventEmitter, rồi cho loyalty throw:

EventEmitter chạy listener ngay trong emit(): loyalty throw, analytics bị bỏ qua, OrderService nhận exception khi đơn đã thanh toán

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

OrderService commit order-101 là đã thanh toán. Từ đây khoản thanh toán là sự thật.

Trạng thái sau từng bước:

1. COMMIT. Đơn 101 đã thanh toán trong database. Chưa có gì khác xảy ra.
2. emit(). OrderService giờ kẹt trong emit(); listener chạy trên call stack của nó.
3. Email. Chạy trước vì đăng ký trước, rồi return.
4. Loyalty throw. Database loyalty đang down.
5. Analytics bị bỏ qua. EventEmitter không bắt lỗi của listener, nên các listener sau không chạy.
6. Exception thoát ra. markPaid() throw sau khi đã commit. Client thấy 500 cho một đơn đã trả tiền và có thể trả lần nữa.

Quy tắc của Node EventEmitter (Node 24, đã kiểm trong demo)

• Listener chạy đồng bộ, theo thứ tự đăng ký; emit() trả true nếu có listener.
• Listener throw sẽ dừng vòng lặp và lỗi lan ra khỏi emit().
• Listener async được khởi chạy nhưng không được await. Promise reject thành unhandled rejection; với cấu hình mặc định của Node 24, process kết thúc với exit code 1.
• new EventEmitter({ captureRejections: true }) chuyển rejection sang event 'error'. Event 'error' không có listener sẽ throw và cũng làm process crash.
• Quá 10 listener cho một event chỉ in cảnh báo “possible memory leak”, không phải lỗi.

1) Listeners run synchronously, in registration order
   email     order-101
   loyalty   +250 points
   analytics tracked
   emit() returned true after all three ran
2) A listener throws synchronously
   email     sent
   emit() threw "loyalty DB down"; analytics never ran
3) An async listener rejects
   plain emitter: emit() returned normally
   plain emitter: unhandledRejection -> ["SMTP timeout"]
   captureRejections: routed to 'error' -> SMTP timeout
4) Isolating publisher: one failure does not stop the others
   email=ok loyalty=failed(loyalty DB down) analytics=ok
5) Process exits before the deferred email handler runs
   order committed, event emitted, email scheduled in memory
   process exits (deploy, OOM, crash): the email is gone, nothing to replay

Demo cài process.on('unhandledRejection') để in được phần 3; không có nó, Node 24 thoát process.

NestJS: @nestjs/event-emitter

Package này bọc eventemitter2. Listener là method của provider:

@Injectable()
export class LoyaltyListener {
  @OnEvent('order.paid')
  async addPoints(event: OrderPaid) {
    await this.points.add(event.customerId, event.amountMinor / 1000);
  }
}

// trong OrderService, sau khi transaction đã commit
this.events.emit('order.paid', new OrderPaid('order-101', 'c-7', 250_000));

Hành vi kiểm bằng cách chạy @nestjs/event-emitter 12.0.1 (eventemitter2 6.4.9) trên Node 24:

Hành vi	Kết quả
Listener throw, option mặc định	Lỗi được log (suppressErrors mặc định true); listener khác vẫn chạy
emit() với listener async	Return trước khi các await của listener xong
@OnEvent(..., { async: true })	Listener bắt đầu sau khi emit() return
suppressErrors: false với emit()	Unhandled rejection (mặc định process thoát)
suppressErrors: false với emitAsync()	Promise trả về reject với lỗi đầu tiên; listener chậm hơn vẫn chạy tiếp

Tài liệu NestJS nêu thêm một bẫy: event emit trước onApplicationBootstrap (trong constructor, onModuleInit) có thể bị lỡ vì listener chưa đăng ký. Hãy await EventEmitterReadinessWatcher.waitUntilReady() trước.

Chi tiết vận hành

• Thứ tự là thứ tự đăng ký, mà trong NestJS lại phụ thuộc thứ tự module và provider. Đừng để listener B dựa vào việc A đã chạy; nếu bắt buộc, đó là một handler, không phải hai.
• Cô lập: bọc từng listener (hoặc bọc publisher, như publishIsolated() trong demo dùng Promise.allSettled) để một lỗi được log kèm event id và không ảnh hưởng listener khác.
• Độ bền: event trong process chỉ nằm trong bộ nhớ. Deploy, OOM kill hay crash giữa commit và lúc listener xong sẽ làm mất event (demo phần 5), không còn gì để chạy lại.
• Emit sau commit: emit trong transaction khiến listener hành động trên dữ liệu có thể còn rollback.
• Request context: listener async sống lâu hơn request; truyền id trong event thay vì đọc state theo request. Xem AsyncLocalStorage.

Với phản ứng bắt buộc phải xảy ra, ghi integration event vào bảng outbox trong cùng transaction với đơn hàng:

Đang tải diagram…

Relay publish ít nhất một lần, nên consumer phải loại trùng theo event id. Chi tiết ở Transactional outbox.

Ví dụ thực tế

Mất email biên nhận sau mỗi lần deploy

Một shop gửi biên nhận từ listener OrderPaid trong process. Mỗi lần deploy, vài khách (minh hoạ: khoảng 0,1% đơn thanh toán trong vài giây cuối) không nhận được biên nhận. Rolling restart giết pod khi call SMTP đang chạy dở.

Team chuyển biên nhận và loyalty sang outbox cùng idempotent consumer theo orderId. Analytics ở lại trong process: mất vài event là chấp nhận được, và nó không còn chung số phận với thanh toán.

npx --prefix examples ts-node examples/src/observer-domain-events/order-paid-demo.ts

Trade-offs

Khi event có ích

• OrderService không đổi khi một team thêm phản ứng.
• Mỗi listener test riêng được với một event mẫu.
• Tên event ghi lại một sự thật nghiệp vụ mà bên khác dựa vào được.

Chi phí

• Luồng điều khiển ngầm: đọc markPaid() không thấy chuyện gì xảy ra tiếp.
• Event trong process không bảo đảm độ bền, retry hay thứ tự qua các lần restart.
• Phải xử lý lỗi tường minh ở từng listener, nếu không một lỗi làm hỏng bên gọi.
• Integration event cần outbox, broker, idempotent consumer và version schema.

Checklist production

1. Phân loại từng phản ứng: cần kết quả, cùng transaction, best-effort, hay không được mất.
2. Chỉ emit sau khi transaction commit; phản ứng không được mất đi qua outbox.
3. Mọi listener tự bắt và log lỗi kèm event id; không listener nào làm hỏng được publisher.
4. Không để listener async thiếu xử lý lỗi; dùng captureRejections kèm listener 'error', hoặc suppressErrors mặc định của NestJS.
5. Không dựa vào thứ tự listener; mỗi phản ứng độc lập là một listener.
6. Event mang id và giá trị, không mang request object hay ORM entity.
7. Consumer của integration event loại trùng theo event id.

Sai lầm thường gặp

Lỗi hay gặp nhất chính là swimlane ở trên: publisher nhận exception của listener.

Emit bên trong transaction

✕ Hay gặp

await db.transaction(async (tx) => {
  await tx.orders.markPaid(orderId);
  this.events.emit('order.paid', event); // email vẫn gửi dù COMMIT thất bại
});

✓ Nên viết

await db.transaction(async (tx) => {
  await tx.orders.markPaid(orderId);
  await tx.outbox.insert(event);         // commit cùng nhau
});
this.events.emit('order.paid', event);   // chỉ cho listener best-effort

Vì sao

Listener có thể hành động trên một đơn về sau bị rollback. Emit sau commit; phản ứng cần bền thì đọc outbox.

Bẫy thứ hai chỉ lộ ra trên production:

Bật suppressErrors: false để có alert

Khi dùng emit(), listener async lỗi sẽ thành unhandled rejection, mặc định làm process Node kết thúc cùng mọi request trong đó. Hãy alert từ catch của chính listener.

Cách hiểu sai	Hệ quả	Cần làm
“emit() là bất đồng bộ”	Listener chậm cộng thêm latency cho request	Listener chạy trong stack của bên gọi; đẩy việc nặng ra sau
“Event là fire-and-forget nên an toàn”	Mất email sau crash	Outbox cho phản ứng không được mất
“Thứ tự listener ổn định, B dùng được kết quả của A”	Vỡ khi đổi thứ tự module	Gộp thành một handler hoặc nối event
“Domain event = integration event”	Schema nội bộ lộ ra service khác	Dịch sang integration event có version

Tự kiểm tra

1. Trong swimlane, vì sao response 500 tệ hơn một lần thanh toán thất bại?
2. Với captureRejections: true mà không có listener 'error', chuyện gì xảy ra khi listener email reject?
3. Trong email, loyalty và analytics, cái nào nên giữ trong process, và vì sao?

🎮 Thử tay

• Chọn design pattern: Observer hay Strategy, và khi nào event cần outbox.

Tóm tắt

Tóm tắt

• Observer tách OrderService khỏi email, loyalty và analytics: nó publish OrderPaid, listener subscribe.
• Node EventEmitter chạy listener đồng bộ theo thứ tự; một throw dừng phần còn lại và lan tới publisher; rejection async cần captureRejections hoặc catch riêng.
• @nestjs/event-emitter mặc định log lỗi listener; emit() không chờ listener async.
• Domain event ở trong process; integration event chỉ rời service sau commit, qua outbox.
• Event trong process biến mất khi crash: việc gì bắt buộc xảy ra thì đi qua outbox và idempotent consumer.

Tham khảo

• Events — Node.js v24: thứ tự listener đồng bộ, event 'error', captureRejections, defaultMaxListeners.
• Events — NestJS: @OnEvent, suppressErrors, wildcard, readiness watcher.
• Domain events: design and implementation — Microsoft Learn: domain event so với integration event, dispatch trước hay sau commit.
• Observer — Refactoring.Guru: vai subject và subscriber.