Event-Driven Architecture

What's an event?

An event is a record that something happened: "a user registered", "a password was reset". Instead of one piece of code directly calling another to make the next thing happen, it just announces the event, and whoever cares reacts.

That indirection is the whole point. The code that does something doesn't need to know who's interested or what they'll do. You can add a new reaction (send a welcome email, update a projection, notify another team's service) without touching the code that emitted the event. Things stay decoupled.

This matters most at the seams between bounded contexts. Each context is its own little system. Several may share a process today, but any of them should be splittable into its own microservice later. They collaborate only through events, never by importing each other or sharing a database, so that split is always possible.

Two kinds of events

There are two, and they exist for different reasons. Getting the distinction right is the key to everything else on this page: one kind stays inside a context, the other is how contexts talk across the boundary.

Domain events: internal to one context

A domain event is emitted by an aggregate when something meaningful happens in the business. It's handled in the same context, in the same process, and it never leaves. Because it stays internal, it can carry rich domain objects (value objects, entities, whatever the handlers need):

// auth context — in-process only, never serialized over the wire
export class UserRegistered_DomainEvent extends Base_DomainEvent {
  constructor(
    public readonly userId: Id,        // domain value objects, not strings
    public readonly email: Email,
    public readonly role: UserRole,
  ) {
    super(userId);
  }
}

Use them to keep a single context's own logic decoupled: an aggregate records what happened, and one or more handlers in that same context react to it.

Integration events: the contract between contexts

An integration event is the public contract that crosses a boundary, between bounded contexts, and between services. It travels over Kafka, so it must be easily serializable: plain, primitive fields only, no domain objects. It's also versioned, because other systems depend on its shape and you need to be able to evolve it without breaking them:

// the public "a user was created" contract — crosses context/service boundaries
export class UserCreated_IntegrationEvent extends Base_IntegrationEvent {
  public readonly userId: string;      // plain strings: serializable, stable
  public readonly email: string;
  public readonly role: string;
 
  constructor(props: UserCreated_IntegrationEventProps) {
    //                              topic            name                 version
    super(props, Topics.USERS.topic, Topics.USERS.events.USER_CREATED, '1.0');
    this.userId = props.userId;
    this.email = props.email;
    this.role = props.role;
    this.validate();
  }
}

The rule of thumb: a domain event says "this happened inside me." An integration event says "here's something I'm announcing to everyone else, in a shape they can rely on."

The handlers

Each kind of event has its own kind of handler, and they live in different layers because they answer to different sides of the boundary.

Domain-event handlers

A domain-event handler reacts to a domain event in the same context, in the same process. It lives in application/domain-event-handlers/<event>/, extends Base_DomainEventHandler(SomeEvent), and is named <Event>_<Action>_DomainEventHandler, one handler per reaction, so an event with three consequences has three small handlers instead of one big one:

// …/domain-event-handlers/user-registered/user-registered_create-email-verification.domain-event-handler.ts
export class UserRegistered_CreateEmailVerification_DomainEventHandler extends Base_DomainEventHandler(
  UserRegistered_DomainEvent,
) {
  constructor(@Inject(COMMAND_BUS) private readonly commandBus: ICommandBus) {
    super();
  }
 
  async handleEvent(event: UserRegistered_DomainEvent) {
    await this.commandBus.execute(
      new CreateEmailVerification_Command({
        userId: event.aggregateId.toValue(),
        email: event.email.toValue(),
        emailAlreadyVerified: event.emailVerified || event.authProvider === 'google',
      }),
    );
  }
}

Handlers run when the command handler that mutated the aggregate calls sendDomainEvents(), which it does after its transaction has committed, so an in-process reaction never acts on a change that was rolled back. The handler above shows the most common shape: dispatching a follow-up command through the command bus (that command then writes UserCreated to the outbox in its own transaction). Other same-service side effects, such as updating a read model, belong here too; the generated example context's GreetingCreated_Log_DomainEventHandler simply logs the new greeting.

Domain-event handlers are not how events leave the service. Anything other contexts or services need is written to the outbox by the command handler, inside the same transaction as the state change (see the outbox).

Integration-event handlers

An integration-event handler is the consume side of the public contract: it reacts to an integration event that arrived over Kafka from another context or service. Because it's an entry point into the context (like a controller, just event-driven instead of HTTP-driven), it lives in interfaces/integration-events/ and is registered with a decorator:

// interfaces/integration-events/user-created.integration-event-handler.ts
@IntegrationEventHandler(UserCreated_IntegrationEvent)
export class UserCreated_IntegrationEventHandler {
  constructor(@Inject(COMMAND_BUS) private readonly commandBus: ICommandBus) {}
 
  async handleEvent(event: UserCreated_IntegrationEvent) {
    await this.commandBus.execute(new CreateGreeting_Command(`Welcome, ${event.email}!`, event.name));
  }
}

The decorator is all the registration there is. The Kafka listener subscribes to the topics that have handlers, and the inbox deduplicates each event before calling handleEvent (see the inbox). Keep the handler thin: translate the event into one of the context's own commands and let the application layer do the work. Make it idempotent: if any handler for an event fails, the inbox retries the event and runs its handlers again.

Getting it reliable

Publishing an integration event sounds simple: change some data, then send a message to Kafka. But that naive approach has two classic failure modes:

• On the publish side: if you save to the database and then send to Kafka, a crash in between means the change happened but the event never went out (lost event). Send first and save second, and you can publish an event for a change that ends up rolling back (phantom event).
• On the consume side: Kafka guarantees at-least-once delivery, so the same message can arrive twice. Processing it twice (charging a card, sending an email again) is a bug.

ForgeStack solves both with two patterns: the transactional outbox on the way out and the idempotent inbox on the way in. The rest of this page is how they fit together.

The flow end to end

1. A command mutates an aggregate, which records DOMAIN events.
2. In ONE DB transaction, the command handler saves the change and writes
   the INTEGRATION event to the OUTBOX.
3. After the commit, it publishes the domain events to their in-process
   handlers.
4. A relay reads unprocessed outbox rows and PUBLISHES them to Kafka.
5. Consumers receive the message into their INBOX (which dedupes it).
6. The inbox dispatches it to the registered handlers, retrying on failure.

Step 2 commits atomically, so you can never publish an event for a change that rolled back, nor lose one for a change that committed. Step 3 comes after the commit for the same reason: in-process reactions only see changes that actually happened.

The outbox (publish side)

The command handler that changes the state writes the integration event to the outbox inside the same transaction, through the base class's sendIntegrationEvent(event, context). Deleting a user is the worked example: the removal and the UserDeleted outbox row commit together or not at all, and the in-process domain events go out only after the commit:

// …/commands/delete-user/delete-user.command-handler.ts (essence)
async handle(command: DeleteUser_Command) {
  const user = await this.userRepository.findById(new Id(command.userIdToDelete));
  if (!user) throw new NotFoundException('User', command.userIdToDelete);
 
  user.delete(); // records the in-process domain events
 
  const integrationEvent = new UserDeleted_IntegrationEvent({
    id: Id.random().toValue(),
    occurredOn: new Date(),
    userId: user.id.toValue(),
    email: user.email.toValue(), // the row is gone by the time consumers run
  });
 
  await Transaction.run(async (context) => {
    await this.userRepository.remove(user.id, context);
    await this.sendIntegrationEvent(integrationEvent, context); // same transaction
  });
 
  await this.sendDomainEvents(user); // only after the commit
}

sendIntegrationEvent wraps the event in an OutboxEvent, which snapshots it as a JSON payload and tracks retry state.

The OutboxEvent aggregate owns its lifecycle (retry counting, max retries, and the processed flag) as behaviour, not setters:

// libs/common/src/outbox/domain/outbox.aggregate.ts
export class OutboxEvent extends SharedAggregate {
  canRetry(): boolean {
    return this.retryCount.isLessThan(this.maxRetries) && !this.isProcessed();
  }
  markAsProcessed(): void {
    this.processedAt = OutboxProcessedAt.now();
  }
  incrementRetry(): void {
    if (!this.canRetry()) throw new DomainValidationException(/* … */);
    this.retryCount = this.retryCount.increment();
  }
}

The generated example context follows the same rule: CreateGreeting_CommandHandler builds GreetingCreated_IntegrationEvent itself, saves the greeting and calls sendIntegrationEvent inside one Transaction.run, and sends its domain events after the commit. Don't save an OutboxEvent from a domain-event handler: that write would not be part of the command's transaction, so the event could be lost, or announced for a change that rolled back.

Kafka (transport)

The publisher sends the serialised event to its topic, keyed by event id:

// libs/kafka/src/kafka.integration-event-publisher.ts
async publish(topic: string, message: string) {
  const producer = this.kafkaService.getProducer();
  const key = JSON.parse(message).id;            // dedupe key
  await producer.send({ topic, messages: [{ key, value: message, timestamp: `${Date.now()}` }] });
}

The listener subscribes to the topics that have registered handlers, and on each message routes it through the inbox:

// libs/kafka/src/kafka.integration-event-listener.ts — eachMessage
await consumer.run({
  eachMessage: async ({ topic, message }) => {
    await this.handleMessage(topic, this.parseMessage(message));
  },
});

The inbox (consume side)

The inbox makes sure a redelivered message is not processed twice. Incoming messages are recorded as InboxEvents with a status that transitions through a small state machine; duplicates are detected and dropped:

// libs/common/src/inbox/domain/inbox.aggregate.ts
export class InboxEvent extends SharedAggregate {
  canProcess(): boolean { return this.status.isPending(); }
 
  markAsProcessing(): void {
    if (!this.canProcess()) throw new DomainValidationException(/* not pending */);
    this.status = InboxStatusVO.processing();
  }
  markAsProcessed(): void {
    if (!this.isProcessing()) throw new DomainValidationException(/* not processing */);
    this.status = InboxStatusVO.processed();
    this.processedAt = InboxProcessedAt.now();
  }
  markAsDuplicate(): void {
    this.status = InboxStatusVO.duplicate();
    this.processedAt = InboxProcessedAt.now();
  }
}

When the listener has an inbox service, handleMessage hands the message to it; the inbox stores, dedupes, and then dispatches to the registered handler. Without an inbox service, the listener dispatches directly (useful in tests).

Anatomy of an integration event

Every integration event extends Base_IntegrationEvent, which standardises the envelope (topic, name, version, id, occurredOn, tracing metadata) and the serialisation that makes it safe to send over the wire:

// libs/common/src/integration-events/events/base.integration-event.ts
export abstract class Base_IntegrationEvent {
  readonly id: string;
  readonly occurredOn: Date;
 
  constructor(props, readonly topic: string, readonly name: string, readonly version: string) {
    this.id = props.id;
    this.occurredOn = props.occurredOn;
  }
 
  toJSON(): Record<string, unknown> {
    return {
      topic: this.topic, name: this.name, version: this.version,
      id: this.id, occurredOn: this.occurredOn.toISOString(),
      metadata: TracingService.getTraceMetadata() || {},
      ...this.toEventJSON(),
    };
  }
 
  protected abstract toEventJSON(): Record<string, unknown>;
}

A concrete event defines its payload and how to (de)serialise it: toEventJSON to send, fromJSON to reconstruct on the other side:

// libs/common/src/integration-events/events/user-deleted.integration-event.ts
export class UserDeleted_IntegrationEvent extends Base_IntegrationEvent {
  public readonly userId: string;
  public readonly email: string;
 
  constructor(props: UserDeleted_IntegrationEventProps) {
    super(props, Topics.USERS.topic, Topics.USERS.events.USER_DELETED, '1.0');
    this.userId = props.userId;
    this.email = props.email;
    this.validate();
  }
 
  protected toEventJSON() {
    return { userId: this.userId, email: this.email };
  }
 
  static fromJSON(json: Record<string, unknown>): UserDeleted_IntegrationEvent {
    Base_IntegrationEvent.validateJson(json);
    return new UserDeleted_IntegrationEvent({
      id: json.id as string,
      occurredOn: new Date(json.occurredOn as string),
      userId: json.userId as string,
      email: json.email as string,
    });
  }
}

The metadata carries the current trace context, so a consumer's spans link back to the producer's trace in Tempo.

Adding your own event

1. Define an integration event extending Base_IntegrationEvent (with toEventJSON / fromJSON).
2. In the producing command handler, build it and write it with sendIntegrationEvent(event, context) inside the same Transaction.run as the state change.
3. In the consuming context, register a handler for the topic/event name; the inbox delivers it.

Next: Testing.