Analytics Event Flow & Materialized View Consumers

Status: Shipped in Phase 1 hardening (2026-04, branches feat/enable-mv-consumers and feat/enable-event-flow).

hmp-analytics-service keeps a set of denormalized read models (occupancy by day, ADR by roomType, F&B revenue by outlet, housekeeping productivity, channel parity drift) that are populated asynchronously from domain events emitted by the operational services. This document describes the event flow and the dispatcher pattern that wires consumers to the message bus.

---

Architecture at a Glance

@hmp/messaging is a thin wrapper over Redis Streams (selected over RabbitMQ/SQS for the demo and small-deployment footprint — see architecture/messaging-evaluation.md). All 13 backend services share the same KafkaService API.

---

Consumer Registration

Each materialized view has a dedicated consumer class that lives in hmp-analytics-service/src/consumers/. Consumers are NestJS providers that implement OnModuleInit and subscribe to topics on startup.

Consumer	Topics Subscribed	Updates
ReservationConsumer	reservation.created, reservation.cancelled, reservation.modified	mv_occupancy_daily, mv_arrivals_daily
FolioConsumer	folio.posted, folio.closed	mv_revenue_daily, mv_adr_by_roomtype
PosConsumer	kot.completed, order.closed	mv_fnb_revenue_daily, mv_outlet_throughput
HkConsumer	task.completed, task.discrepancy	mv_hk_productivity_daily
ChannelConsumer	channel.sync.failed	mv_channel_parity_drift

All five consumers are now registered in ConsumersModule (the C2 fix). Previously they existed but weren't wired into the module graph.

---

EventDispatcherService (the C5 Fix)

Registering consumers wasn't enough — they were never receiving events because nothing was calling kafkaService.subscribe() for them on startup. The C5 fix introduced EventDispatcherService, a single class that owns the wiring:

@Injectable()
export class EventDispatcherService implements OnModuleInit {
  constructor(
    private readonly kafka: KafkaService,
    private readonly reservation: ReservationConsumer,
    private readonly folio: FolioConsumer,
    private readonly pos: PosConsumer,
    private readonly hk: HkConsumer,
    private readonly channel: ChannelConsumer,
  ) {}

  async onModuleInit(): Promise<void> {
    await this.kafka.subscribe('reservation.created', (msg) => this.reservation.onCreated(msg));
    await this.kafka.subscribe('reservation.cancelled', (msg) => this.reservation.onCancelled(msg));
    await this.kafka.subscribe('reservation.modified', (msg) => this.reservation.onModified(msg));
    await this.kafka.subscribe('folio.posted', (msg) => this.folio.onPosted(msg));
    await this.kafka.subscribe('folio.closed', (msg) => this.folio.onClosed(msg));
    await this.kafka.subscribe('kot.completed', (msg) => this.pos.onKotCompleted(msg));
    await this.kafka.subscribe('order.closed', (msg) => this.pos.onOrderClosed(msg));
    await this.kafka.subscribe('task.completed', (msg) => this.hk.onTaskCompleted(msg));
    await this.kafka.subscribe('channel.sync.failed', (msg) => this.channel.onSyncFailed(msg));
  }
}

Nine topic subscriptions, one place to look. If a consumer needs a new topic, it goes here.

---

SKIP_EVENT_SUBSCRIPTIONS Environment Flag

All 13 backend services share the same MessagingModule from @hmp/messaging. By default, the module starts a Redis Streams consumer on boot. For services that should only emit events, this is wasted resources and a source of "no consumer found" log noise.

The SKIP_EVENT_SUBSCRIPTIONS=true env flag disables consumer startup while leaving emit paths intact. The C5 fix to the staging and prod compose files set this on every backend except the api-gateway (which doesn't need either path).

# docker-compose.staging-aws.yml (excerpt)
hmp-reservation-service:
  environment:
    SKIP_EVENT_SUBSCRIPTIONS: "false"  # subscribes via the dispatcher
hmp-frontdesk-service:
  environment:
    SKIP_EVENT_SUBSCRIPTIONS: "false"
# ... all services false except gateway
hmp-api-gateway:
  environment:
    SKIP_EVENT_SUBSCRIPTIONS: "true"   # gateway only proxies HTTP

Identical block in docker-compose.prod-aws.yml.

Naming note: false means "do subscribe". The flag name is the negative because the historical default was true (skip), and we kept that for backwards compatibility with existing dev environments.

---

Test Coverage

The C5 branch added or restored tests covering the dispatcher and consumers:

• 6 tests on the EventDispatcherService (each topic correctly routed; idempotent on re-init; logs subscription failures without crashing)
• 25 tests across the five consumers (event payload parsing, MV upsert, error propagation)
• All 31 tests pass on feat/enable-event-flow

---

Operational Notes

• Backfill: When a new MV is added, run the backfill script in hmp-analytics-service/scripts/backfill-mv.ts to replay historical events. The dispatcher doesn't read from the start of the stream — it only sees events from its consumer-group offset.
• Lag monitoring: Each consumer reports analytics_consumer_lag_seconds via the metrics module. Alert threshold is 60 seconds (configured in hmp-infrastructure/monitoring/alerts/hmp-service-alerts.yaml).
• Replay: To force a full reprocess, delete the consumer group and restart the analytics-service pod. Idempotency is enforced by upsert keys on every MV row.

---

Related Docs

• messaging-evaluation.md — why Redis Streams over Kafka
• cross-service-communication.md — synchronous reads use ServiceClient; analytics writes use this async event flow
• hmp-analytics-service/src/consumers/ — implementation
• hmp-shared-packages/packages/messaging/ — KafkaService wrapper