event-bus.ts · Source

event-bus.tsapps/api/src/common/event-bus.tsraw ← common2,329 B← Back to the app
1import { Inject, Injectable } from "@nestjs/common";
2import { EventEmitter } from "node:events";
3import { v4 as uuid } from "uuid";
4import { EVENTS_STREAM, EVENTS_STREAM_MAXLEN } from "../constants.js";
5import { logger } from "./logger.js";
6import { RedisService } from "./redis.service.js";
7 
8export interface DomainEventPayload {
9  workspaceId?: string;
10  [key: string]: unknown;
11}
12 
13/**
14 * Domain event bus.
15 *
16 * `publish(event, payload)` appends to the Redis stream `factory:events`
17 * (XADD, MAXLEN ~ 10000) and mirrors it on an in-process EventEmitter so
18 * local consumers and tests can listen without a live Redis.
19 *
20 * Workers (BullMQ side) consume the stream for search indexing, webhooks,
21 * automation runs and notification fan-out. Publishing must never fail a
22 * request: Redis outages log a warning and drop the stream write.
23 */
24@Injectable()
25export class EventBus {
26  private readonly local = new EventEmitter();
27  static readonly localStream = "local";
28 
29  constructor(@Inject(RedisService) private readonly redis: RedisService) {
30    this.local.setMaxListeners(200);
31  }
32 
33  async publish(event: string, payload: DomainEventPayload = {}): Promise<void> {
34    const body = {
35      event,
36      payload: JSON.stringify(payload),
37      ts: new Date().toISOString(),
38      id: uuid()
39    };
40    // Local mirror first (tests, in-process consumers).
41    try {
42      this.local.emit(EventBus.localStream, body);
43      this.local.emit(event, body);
44    } catch (err) {
45      logger.warn(`event bus local emit failed for ${event}: ${(err as Error).message}`);
46    }
47    // Stream append (best effort).
48    try {
49      await this.redis.getRedis().xadd(
50        EVENTS_STREAM,
51        "MAXLEN",
52        "~",
53        String(EVENTS_STREAM_MAXLEN),
54        "*",
55        "event",
56        body.event,
57        "payload",
58        body.payload,
59        "ts",
60        body.ts
61      );
62    } catch (err) {
63      logger.warn(`event bus stream publish failed for ${event}: ${(err as Error).message}`);
64    }
65  }
66 
67  /** Subscribe to local events (tests / in-process). Returns unsubscribe. */
68  onLocal(event: string | typeof EventBus.localStream, handler: (body: { event: string; payload: string; ts: string; id: string }) => void): () => void {
69    this.local.on(event, handler);
70    return () => this.local.off(event, handler);
71  }
72}
73