Event Listeners

Event Listeners

Domain events emitted via entity.commit() flow through the NestJS CQRS EventBus. Listeners are @EventsHandler classes that react to these events. Multiple handlers for the same event run in parallel via Promise.allSettled — if one fails, others continue.

This skill covers WHERE listeners live, WHAT they do, and WHEN you actually need one.

---

Decision Tree

Do you need to react to a domain event?
│
├─ Is it a side effect within the SAME bounded context?
│  └─ YES → Same-BC Listener (see Section 1)
│     Examples: update Redis projection, write audit log, invalidate cache
│
├─ Does ANOTHER bounded context need to react?
│  └─ YES → Cross-BC Listener (see Section 2)
│     Examples: Billing creates invoice when Order is created
│
├─ Does the event need to leave the process?
│  └─ YES → Bridge Listener (see Section 3)
│     Examples: WebSocket broadcast, RabbitMQ publish, send email, call webhook
│
└─ Is the side effect simple and only 1 consumer exists?
   └─ YES → Consider putting it in the command handler directly (no listener needed)

---

When NOT to Create a Listener (anti-over-engineering)

| Situation | Do this instead | |---|---| | Only 1 side effect, simple and synchronous | Put it in the command handler after entity.commit() | | Side effect is part of the core business transaction | Keep it in the use case / handler — not a separate listener | | /infrastructure/listeners/`. Reacts to events from its OWN bounded context.

// infrastructure/listeners/order-created-projection.handler.ts
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { Inject, Logger } from '@nestjs/common';

import { OrderCreatedEvent } from '../../domain/events/order-created.event';
import { REDIS_READ_MODEL_TOKEN } from '../../application/ports/read-model.port';
import type { ReadModelPort } from '../../application/ports/read-model.port';

@EventsHandler(OrderCreatedEvent)
export class OrderCreatedProjectionHandler implements IEventHandler {
  private readonly logger = new Logger(OrderCreatedProjectionHandler.name);

  constructor(
    @Inject(REDIS_READ_MODEL_TOKEN)
    private readonly readModel: ReadModelPort,
  ) {}

  async handle(event: OrderCreatedEvent): Promise {
    try {
      await this.readModel.upsert(`order:${event.aggregateId}`, {
        id: event.aggregateId,
        total: event.total,
        status: event.status,
        organizationId: event.organizationId,
        updatedAt: event.occurredOn.toISOString(),
      });
    } catch (error) {
      // Log but never re-throw — don't break the event chain
      this.logger.error(`[OrderCreatedProjection] Failed:`, error);
    }
  }
}

Common same-BC listeners:

• Projection updater — writes denormalized view to Redis/MongoDB
• Audit log — records domain change for compliance
• Cache invalidation — clears stale cache entries
• Counter update — increments/decrements aggregate counters

See references/same-bc-listeners.md for full templates.

---

Section 2: Cross-BC Listener

Cross-BC communication uses the Integration Event Port pattern with an Anti-Corruption Layer (ACL):

• Domain Event = INTERNAL to the BC. Never leaves the bounded context.
• Integration Event = PUBLIC contract between BCs. Always via port/adapter.
• ACL = Adapter in the consuming BC that translates external model to internal model.

The 3-part flow:

OrderBC (emitter)                      InvoicingBC (consumer)
  domain/events/order-paid.event.ts
  application/ports/
    order-integration-events.port.ts  ←  implemented by →  InvoiceIntegrationAdapter
  infrastructure/listeners/
    order-paid-integration.handler.ts                         ↓ dispatches CreateInvoiceCommand

Part 1 — Port definition (in emitting BC):

// orders/application/ports/order-integration-events.port.ts
export const ORDER_INTEGRATION_EVENTS_TOKEN = Symbol('OrderIntegrationEvents');

export interface OrderIntegrationEventsPort {
  orderPaid(payload: {
    orderId: string;
    organizationId: string;
    customerId: string;
    total: number;
    currency: string;
  }): Promise;
}

Part 2 — Publisher handler (in emitting BC):

// orders/infrastructure/listeners/order-paid-integration.handler.ts
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { Inject, Logger } from '@nestjs/common';
import { OrderPaidEvent } from '../../domain/events/order-paid.event';
import {
  ORDER_INTEGRATION_EVENTS_TOKEN,
  type OrderIntegrationEventsPort,
} from '../../application/ports/order-integration-events.port';

@EventsHandler(OrderPaidEvent)
export class OrderPaidIntegrationHandler implements IEventHandler {
  private readonly logger = new Logger(OrderPaidIntegrationHandler.name);

  constructor(
    @Inject(ORDER_INTEGRATION_EVENTS_TOKEN)
    private readonly integrationEvents: OrderIntegrationEventsPort,
  ) {}

  async handle(event: OrderPaidEvent): Promise {
    try {
      await this.integrationEvents.orderPaid({
        orderId: event.aggregateId,
        organizationId: event.organizationId,
        customerId: event.customerId,
        total: event.total,
        currency: event.currency,
      });
    } catch (error) {
      this.logger.error(
        `Failed to publish order.paid integration event for ${event.aggregateId}`,
        error instanceof Error ? error.stack : String(error),
      );
    }
  }
}

Part 3 — Consumer adapter (ACL in consuming BC):

// invoicing/infrastructure/adapters/order-integration-events.adapter.ts
import { Injectable } from '@nestjs/common';
import { CommandBus } from '@nestjs/cqrs';
import { type OrderIntegrationEventsPort } from '@/enterprise/orders/application/ports/order-integration-events.port';
import { CreateInvoiceCommand } from '../../application/commands/create-invoice.command';

@Injectable()
export class OrderIntegrationEventsAdapter implements OrderIntegrationEventsPort {
  constructor(private readonly commandBus: CommandBus) {}

  async orderPaid(payload: {
    orderId: string;
    organizationId: string;
    customerId: string;
    total: number;
    currency: string;
  }): Promise {
    // ACL: translate Orders BC model to Invoicing BC model
    await this.commandBus.execute(
      new CreateInvoiceCommand(
        payload.organizationId,
        payload.orderId,
        payload.customerId,
        payload.total,
        payload.currency,
      ),
    );
  }
}

Module wiring — emitter BC provides the token:

// orders/orders.module.ts
{
  provide: ORDER_INTEGRATION_EVENTS_TOKEN,
  useClass: OrderIntegrationEventsAdapter,  // wired from InvoicingModule via DI
}

Key rules:

1. Orders BC NEVER imports from Invoicing BC
2. Invoicing BC implements the port that Orders BC defines
3. The ACL adapter is the only place that knows both models
4. try/catch mandatory in the publisher handler — cross-BC failure must not affect emitting BC

3 approaches — choose by deployment context:

| Approach | When to use | |---|---| | Integration Event Port (this pattern) | Modular monolith — recommended | | String events via EventEmitter2 | Simpler, but loses type-safety | | Message broker (RabbitMQ/Kafka) | Microservices — maximum decoupling |

See references/cross-bc-listeners.md for full templates and testing patterns.

---

Section 3: Bridge Listener

Transforms a domain event into an external output: WebSocket broadcast, RabbitMQ message, email, webhook.

// infrastructure/listeners/order-created-broadcast.handler.ts
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { Inject, Logger } from '@nestjs/common';

import { OrderCreatedEvent } from '../../domain/events/order-created.event';
import { WS_GATEWAY_TOKEN, type WsGatewayPort } from '@/shared/ports/ws-gateway.port';

@EventsHandler(OrderCreatedEvent)
export class OrderCreatedBroadcastHandler implements IEventHandler {
  private readonly logger = new Logger(OrderCreatedBroadcastHandler.name);

  constructor(
    @Inject(WS_GATEWAY_TOKEN) private readonly gateway: WsGatewayPort,
  ) {}

  async handle(event: OrderCreatedEvent): Promise {
    try {
      this.gateway.emitToOrganization(event.organizationId, 'order:created', {
        id: event.aggregateId,
        total: event.total,
        status: event.status,
      });
    } catch (error) {
      this.logger.error(`[OrderCreatedBroadcast] Failed:`, error);
    }
  }
}

Bridge types:

• WebSocket — WsGatewayPort.emitToOrganization() (see websocket-broadcasting skill)
• Message broker — publish to RabbitMQ/Kafka exchange for microsservice consumption
• Email — send via EmailPort (notification listeners)
• Webhook — POST to external URL via HttpPort

See references/bridge-listeners.md for full templates.

---

Strategy + Gateway Pattern (3+ consumers)

When 3+ listeners react to the same event and share pre-processing (validation, enrichment), use the Gateway pattern with Strategy fan-out.

When to use: 3+ consumers for the same event that share common logic. When NOT to use: { supports(event: T): boolean; execute(event: T): Promise; }

// Gateway handler — single @EventsHandler, fans out to strategies @EventsHandler(OrderCreatedEvent) export class OrderCreatedGateway implements IEventHandler { private readonly logger = new Logger(OrderCreatedGateway.name);

constructor( @Inject('ORDERCREATEDSTRATEGIES') private readonly strategies: EventListenerStrategy[], ) {}

async handle(event: OrderCreatedEvent): Promise { const applicable = this.strategies.filter(s => s.supports(event));

const results = await Promise.allSettled( applicable.map(strategy => strategy.execute(event)), );

for (const result of results) { if (result.status === 'rejected') { this.logger.error([OrderCreatedGateway] Strategy failed:, result.reason); } } } }

// Strategy implementations @Injectable() export class OrderProjectionStrategy implements EventListenerStrategy { supports(): boolean { return true; } async execute(event: OrderCreatedEvent): Promise { / update Redis / } }

@Injectable() export class OrderBroadcastStrategy implements EventListenerStrategy { supports(): boolean { return true; } async execute(event: OrderCreatedEvent): Promise { / WS emit / } }

@Injectable() export class OrderInvoiceStrategy implements EventListenerStrategy { supports(event: OrderCreatedEvent): boolean { return event.total > 0; // Only for paid orders } async execute(event: OrderCreatedEvent): Promise { / create invoice / } }

// Module wiring { provide: 'ORDERCREATEDSTRATEGIES', useFactory: ( projection: OrderProjectionStrategy, broadcast: OrderBroadcastStrategy, invoice: OrderInvoiceStrategy, ): EventListenerStrategy[] => [projection, broadcast, invoice], inject: [OrderProjectionStrategy, OrderBroadcastStrategy, OrderInvoiceStrategy], }

**Benefits over separate handlers:**
- Centralized error isolation (`Promise.allSettled`)
- Shared pre-processing (fetch once, pass to all)
- Dynamic enable/disable via `supports()`
- Single `@EventsHandler` registration per event

**Trade-off:** More indirection. Only use when the benefits outweigh the simplicity of separate handlers.

---

## Listener Registration

All listeners are registered in the CONSUMING module's `providers` array:

```typescript
// In the module that CONSUMES the event
@Module({
  imports: [CqrsModule],
  providers: [
    // Same-BC listeners
    OrderCreatedProjectionHandler,
    OrderUpdatedCacheInvalidator,

    // Cross-BC listeners (events imported from other BCs)
    PaymentReceivedInvoiceHandler,

    // Bridge listeners
    OrderCreatedBroadcastHandler,

    // Strategy gateway (if using Strategy pattern)
    OrderCreatedGateway,
    OrderProjectionStrategy,
    OrderBroadcastStrategy,
    OrderInvoiceStrategy,
    {
      provide: 'ORDER_CREATED_STRATEGIES',
      useFactory: (...strategies) => strategies,
      inject: [OrderProjectionStrategy, OrderBroadcastStrategy, OrderInvoiceStrategy],
    },
  ],
})
export class OrdersModule {}

---

Testing Listeners

describe('OrderCreatedProjectionHandler', () => {
  let handler: OrderCreatedProjectionHandler;
  let readModel: MockReadModelPort;

  beforeEach(() => {
    readModel = { upsert: vi.fn(), get: vi.fn(), delete: vi.fn() };
    handler = new OrderCreatedProjectionHandler(readModel);
  });

  it('should update read model on order created', async () => {
    const event = new OrderCreatedEvent('order-1', 'org-1', 1000, 'PENDING');

    await handler.handle(event);

    expect(readModel.upsert).toHaveBeenCalledWith('order:order-1', {
      id: 'order-1',
      total: 1000,
      status: 'PENDING',
      organizationId: 'org-1',
      updatedAt: expect.any(String),
    });
  });

  it('should not throw on read model failure', async () => {
    readModel.upsert.mockRejectedValue(new Error('Redis down'));
    const event = new OrderCreatedEvent('order-1', 'org-1', 1000, 'PENDING');

    // Must not throw
    await expect(handler.handle(event)).resolves.toBeUndefined();
  });
});

---

References

| File | Content | |------|---------| | references/same-bc-listeners.md | Projection updater, audit log, cache invalidation, counter update | | references/cross-bc-listeners.md | Event import rules, CommandBus dispatch, module wiring, circular dep avoidance | | references/bridge-listeners.md | WS broadcast, RabbitMQ publish, email notification, webhook POST |

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: Softtor
• Source: Softtor/nestjs-hexagonal
• License: MIT
• Homepage: https://github.com/Softtor/nestjs-hexagonal#readme

Install and usage instructions live in the source repository linked above.