task-stream/libs/api/shared/infrastructure/src/lib/event-store/events-bridge.ts

import {
  Inject,
  Injectable,
  Logger,
  OnApplicationBootstrap,
  OnApplicationShutdown,
  Type,
} from '@nestjs/common';
import { DiscoveryService } from '@nestjs/core';
import {
  CommandBus,
  ICommand,
  IEvent,
  IEventHandler,
  ISaga,
} from '@nestjs/cqrs';
import {
  ConcurrencyError,
  DomainError,
  SerializableEvent,
} from '@task-stream/api/shared/domain';
import { hostname } from 'node:os';
import { lastValueFrom, of, toArray } from 'rxjs';
import { EventStreamConsumer } from '../event-stream/event-stream.consumer';
import {
  EVENT_STREAM_OPTIONS,
  EventStreamOptions,
} from '../event-stream/event-stream.options';
import { RedisConnectionFactory } from '../redis/redis-connection.factory';
import { EventDeserializer } from './deserializers/event.deserializer';

export const EVENTS_BRIDGE_ENABLED = Symbol('EVENTS_BRIDGE_ENABLED');

/** Metadata written by @EventsHandler(...events) and @Saga() (@nestjs/cqrs). */
const EVENTS_HANDLER_METADATA = '__eventsHandler__';
const SAGA_METADATA = '__saga__';

/** Attempts of a saga command that keeps hitting concurrent writes. */
const CONCURRENCY_ATTEMPTS = 5;

/**
 * Redis stream → @EventsHandler. Plays the role of the event-store change
 * stream: every stored event is deserialized and handed to the event
 * handlers registered for its class (read-model projections, realtime...).
 *
 * Unlike `eventBus.subject$.next()`, handlers are awaited one event at a
 * time, so events of an aggregate are projected in order, and the entry is
 * acknowledged in Redis only once every handler succeeded (otherwise it
 * stays pending and is retried).
 *
 * It also runs the `@Saga()`s (the in-memory EventBus never sees stored
 * events, so it would not): after the handlers, each saga gets the event
 * alone (`of(event)`), and the commands it emits are executed one by one,
 * awaited. Commands must be idempotent (an event may be handled twice).
 * A command refused by the domain (not found, business rule...) is logged
 * and skipped, so it cannot block the stream; concurrent writes are retried.
 */
@Injectable()
export class EventsBridge
  implements OnApplicationBootstrap, OnApplicationShutdown
{
  private readonly logger = new Logger(EventsBridge.name);
  private consumer?: EventStreamConsumer;
  private readonly handlers = new Map<Type, IEventHandler[]>();
  private readonly sagas: { name: string; saga: ISaga }[] = [];

  constructor(
    private readonly connections: RedisConnectionFactory,
    @Inject(EVENT_STREAM_OPTIONS) private readonly stream: EventStreamOptions,
    @Inject(EVENTS_BRIDGE_ENABLED) private readonly enabled: boolean,
    private readonly eventDeserializer: EventDeserializer,
    private readonly discovery: DiscoveryService,
    private readonly commandBus: CommandBus,
  ) {}

  async onApplicationBootstrap(): Promise<void> {
    if (!this.enabled) {
      this.logger.log('Events bridge disabled on this instance');
      return;
    }
    this.registerEventHandlers();
    this.registerSagas();
    this.consumer = new EventStreamConsumer(
      this.connections.create('events-bridge'),
      {
        streamKey: this.stream.streamKey,
        group: 'events-bridge',
        consumer: hostname(),
      },
      (event) => this.handleEvent(event),
    );
    await this.consumer.start();
    this.logger.log(
      `Bridging ${this.stream.streamKey} to ${this.handlers.size} event type(s) and ${this.sagas.length} saga(s)`,
    );
  }

  async onApplicationShutdown(): Promise<void> {
    await this.consumer?.stop();
  }

  private registerEventHandlers(): void {
    for (const wrapper of this.discovery.getProviders()) {
      if (!wrapper.metatype || !wrapper.instance) continue;
      const events: Type[] | undefined = Reflect.getMetadata(
        EVENTS_HANDLER_METADATA,
        wrapper.metatype,
      );
      for (const eventCls of events ?? []) {
        this.handlers.set(eventCls, [
          ...(this.handlers.get(eventCls) ?? []),
          wrapper.instance as IEventHandler,
        ]);
      }
    }
  }

  private registerSagas(): void {
    for (const wrapper of this.discovery.getProviders()) {
      if (!wrapper.metatype || !wrapper.instance) continue;
      const keys: string[] =
        Reflect.getMetadata(SAGA_METADATA, wrapper.metatype) ?? [];
      const instance = wrapper.instance as Record<string, ISaga>;
      for (const key of keys) {
        this.sagas.push({
          name: `${wrapper.metatype.name}.${key}`,
          saga: instance[key].bind(instance),
        });
      }
    }
  }

  private async handleEvent(stored: SerializableEvent): Promise<void> {
    const { data: event } = this.eventDeserializer.deserialize(stored);
    const eventCls = Object.getPrototypeOf(event).constructor as Type;
    // Projections first: sagas may read the read model they just updated.
    for (const handler of this.handlers.get(eventCls) ?? []) {
      await handler.handle(event);
    }
    for (const { name, saga } of this.sagas) {
      const commands = await lastValueFrom(
        saga(of(event as IEvent)).pipe(toArray()),
      );
      for (const command of commands) {
        await this.execute(name, event as IEvent, command);
      }
    }
  }

  private async execute(
    saga: string,
    event: IEvent,
    command: ICommand,
  ): Promise<void> {
    for (let attempt = 1; ; attempt++) {
      try {
        await this.commandBus.execute(command);
        return;
      } catch (error) {
        if (
          error instanceof ConcurrencyError &&
          attempt < CONCURRENCY_ATTEMPTS
        ) {
          await new Promise((resolve) => setTimeout(resolve, 50 * attempt));
          continue;
        }
        if (
          error instanceof DomainError &&
          !(error instanceof ConcurrencyError)
        ) {
          this.logger.warn(
            `${saga}: ${command.constructor.name} after ${event.constructor.name} skipped: ${error.message}`,
          );
          return;
        }
        throw error;
      }
    }
  }
}