first commit

libs/api/shared/infrastructure/src/lib/event-store/deserializers/event.deserializer.ts

@@ -0,0 +1,26 @@
+import { Injectable, Type } from '@nestjs/common';
+import {
+  EventClsRegistry,
+  SerializableEvent,
+} from '@task-stream/api/shared/domain';
+
+@Injectable()
+export class EventDeserializer {
+  deserialize<T>(event: SerializableEvent): SerializableEvent<T> {
+    const eventCls = EventClsRegistry.get(event.type);
+    return {
+      streamId: event.streamId,
+      type: event.type,
+      position: event.position,
+      data: this.instantiateSerializedEvent(eventCls, event.data),
+    };
+  }
+
+  /** Rebuilds the event with its class prototype, without running the constructor. */
+  instantiateSerializedEvent<T extends Type>(
+    eventCls: T,
+    data: Record<string, unknown>,
+  ) {
+    return Object.assign(Object.create(eventCls.prototype), data);
+  }
+}

libs/api/shared/infrastructure/src/lib/event-store/entities/event.entity.ts

@@ -0,0 +1,40 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  Index,
+  PrimaryGeneratedColumn,
+  Unique,
+} from 'typeorm';
+
+/**
+ * Event store (write model). Append-only: one row per domain event.
+ * `UNIQUE(stream_id, position)` gives optimistic concurrency per aggregate.
+ */
+@Entity('events')
+@Unique('uq_events_stream_position', ['streamId', 'position'])
+@Index('idx_events_unpublished', ['id'], { where: '"published_at" IS NULL' })
+export class EventEntity {
+  /** Global order of the store, used to relay events in sequence. */
+  @PrimaryGeneratedColumn('increment', { type: 'bigint' })
+  id!: string;
+
+  @Column({ name: 'stream_id', length: 64 })
+  streamId!: string;
+
+  @Column({ type: 'int' })
+  position!: number;
+
+  @Column()
+  type!: string;
+
+  @Column({ type: 'jsonb' })
+  data!: Record<string, unknown>;
+
+  @CreateDateColumn({ name: 'created_at', type: 'timestamptz' })
+  createdAt!: Date;
+
+  /** When the relay pushed it to the Redis stream. */
+  @Column({ name: 'published_at', type: 'timestamptz', nullable: true })
+  publishedAt!: Date | null;
+}

libs/api/shared/infrastructure/src/lib/event-store/event-store.relay.ts

@@ -0,0 +1,103 @@
+import {
+  Injectable,
+  Logger,
+  OnApplicationBootstrap,
+  OnModuleDestroy,
+} from '@nestjs/common';
+import { SerializableEvent } from '@task-stream/api/shared/domain';
+import { Subscription, debounceTime, interval, merge } from 'rxjs';
+import { DataSource } from 'typeorm';
+import { EventStreamPublisher } from '../event-stream/event-stream.publisher';
+import { EventEntity } from './entities/event.entity';
+import { EventStoreSignal } from './event-store.signal';
+
+const BATCH_SIZE = 200;
+const POLL_INTERVAL_MS = 1000;
+
+/**
+ * Forwards new events from the Postgres event store to the Redis stream, in
+ * store order. `FOR UPDATE SKIP LOCKED` lets several instances relay without
+ * duplicates; delivery is at-least-once.
+ */
+@Injectable()
+export class EventStoreRelay
+  implements OnApplicationBootstrap, OnModuleDestroy
+{
+  private readonly logger = new Logger(EventStoreRelay.name);
+  private subscription?: Subscription;
+  private draining?: Promise<void>;
+  private again = false;
+
+  constructor(
+    private readonly dataSource: DataSource,
+    private readonly publisher: EventStreamPublisher,
+    private readonly signal: EventStoreSignal,
+  ) {}
+
+  onApplicationBootstrap(): void {
+    this.subscription = merge(
+      this.signal.appended$.pipe(debounceTime(10)),
+      interval(POLL_INTERVAL_MS),
+    ).subscribe(() => void this.drain());
+  }
+
+  /** Runs before connections are closed: stop polling and let the current batch finish. */
+  async onModuleDestroy(): Promise<void> {
+    this.subscription?.unsubscribe();
+    await this.draining;
+  }
+
+  drain(): Promise<void> {
+    if (this.draining) {
+      this.again = true;
+      return this.draining;
+    }
+    this.draining = this.drainLoop().finally(() => (this.draining = undefined));
+    return this.draining;
+  }
+
+  private async drainLoop(): Promise<void> {
+    try {
+      do {
+        this.again = false;
+        while ((await this.relayBatch()) === BATCH_SIZE);
+      } while (this.again);
+    } catch (error) {
+      this.logger.error(`Event relay failed: ${(error as Error).message}`);
+    }
+  }
+
+  private relayBatch(): Promise<number> {
+    return this.dataSource.transaction(async (manager) => {
+      const rows = await manager
+        .getRepository(EventEntity)
+        .createQueryBuilder('e')
+        .where('e.published_at IS NULL')
+        .orderBy('e.id', 'ASC')
+        .limit(BATCH_SIZE)
+        .setLock('pessimistic_write')
+        .setOnLocked('skip_locked')
+        .getMany();
+      if (rows.length === 0) return 0;
+
+      await this.publisher.publish(
+        rows.map(
+          (row): SerializableEvent => ({
+            streamId: row.streamId,
+            type: row.type,
+            position: row.position,
+            data: row.data,
+          }),
+        ),
+      );
+      await manager
+        .getRepository(EventEntity)
+        .createQueryBuilder()
+        .update()
+        .set({ publishedAt: () => 'now()' })
+        .whereInIds(rows.map((r) => r.id))
+        .execute();
+      return rows.length;
+    });
+  }
+}

libs/api/shared/infrastructure/src/lib/event-store/event-store.signal.ts

@@ -0,0 +1,13 @@
+import { Injectable } from '@nestjs/common';
+import { Subject } from 'rxjs';
+
+/** In-process nudge so the relay forwards events right after they are stored. */
+@Injectable()
+export class EventStoreSignal {
+  private readonly subject = new Subject<void>();
+  readonly appended$ = this.subject.asObservable();
+
+  notify(): void {
+    this.subject.next();
+  }
+}

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

@@ -0,0 +1,100 @@
+import {
+  Inject,
+  Injectable,
+  Logger,
+  OnApplicationBootstrap,
+  OnApplicationShutdown,
+  Type,
+} from '@nestjs/common';
+import { DiscoveryService } from '@nestjs/core';
+import { IEventHandler } from '@nestjs/cqrs';
+import { SerializableEvent } from '@task-stream/api/shared/domain';
+import { hostname } from 'node:os';
+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) (@nestjs/cqrs). */
+const EVENTS_HANDLER_METADATA = '__eventsHandler__';
+
+/**
+ * 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).
+ */
+@Injectable()
+export class EventsBridge
+  implements OnApplicationBootstrap, OnApplicationShutdown
+{
+  private readonly logger = new Logger(EventsBridge.name);
+  private consumer?: EventStreamConsumer;
+  private readonly handlers = new Map<Type, IEventHandler[]>();
+
+  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,
+  ) {}
+
+  async onApplicationBootstrap(): Promise<void> {
+    if (!this.enabled) {
+      this.logger.log('Events bridge disabled on this instance');
+      return;
+    }
+    this.registerEventHandlers();
+    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)`,
+    );
+  }
+
+  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 async handleEvent(stored: SerializableEvent): Promise<void> {
+    const { data: event } = this.eventDeserializer.deserialize(stored);
+    const eventCls = Object.getPrototypeOf(event).constructor as Type;
+    for (const handler of this.handlers.get(eventCls) ?? []) {
+      await handler.handle(event);
+    }
+  }
+}

libs/api/shared/infrastructure/src/lib/event-store/postgres-event-store.ts

@@ -0,0 +1,74 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { EventStore } from '@task-stream/api/shared/application';
+import {
+  ConcurrencyError,
+  SerializableEvent,
+} from '@task-stream/api/shared/domain';
+import { DataSource } from 'typeorm';
+import { EventEntity } from './entities/event.entity';
+import { EventDeserializer } from './deserializers/event.deserializer';
+import { EventStoreSignal } from './event-store.signal';
+
+const UNIQUE_VIOLATION = '23505';
+
+@Injectable()
+export class PostgresEventStore extends EventStore {
+  private readonly logger = new Logger(PostgresEventStore.name);
+
+  constructor(
+    private readonly dataSource: DataSource,
+    private readonly deserializer: EventDeserializer,
+    private readonly signal: EventStoreSignal,
+  ) {
+    super();
+  }
+
+  async persist(
+    eventOrEvents: SerializableEvent | SerializableEvent[],
+  ): Promise<void> {
+    const events = Array.isArray(eventOrEvents)
+      ? eventOrEvents
+      : [eventOrEvents];
+    if (events.length === 0) return;
+
+    try {
+      await this.dataSource.transaction((manager) =>
+        manager.save(
+          EventEntity,
+          events.map((e) =>
+            manager.create(EventEntity, {
+              streamId: e.streamId,
+              position: e.position,
+              type: e.type,
+              data: e.data as Record<string, unknown>,
+            }),
+          ),
+        ),
+      );
+      this.logger.debug(`Events inserted successfully to the event store`);
+    } catch (error) {
+      if ((error as { code?: string }).code === UNIQUE_VIOLATION) {
+        this.logger.warn(
+          `Events could not be persisted. Aggregate ${events[0].streamId} is stale.`,
+        );
+        throw new ConcurrencyError('Aggregate', events[0].streamId);
+      }
+      throw error;
+    }
+    this.signal.notify();
+  }
+
+  async getEventsByStreamId(streamId: string): Promise<SerializableEvent[]> {
+    const rows = await this.dataSource
+      .getRepository(EventEntity)
+      .find({ where: { streamId }, order: { position: 'ASC' } });
+    return rows.map((row) =>
+      this.deserializer.deserialize({
+        streamId: row.streamId,
+        type: row.type,
+        position: row.position,
+        data: row.data,
+      }),
+    );
+  }
+}

libs/api/shared/infrastructure/src/lib/event-store/publishers/event-store.publisher.ts

@@ -0,0 +1,54 @@
+import { Injectable, OnApplicationBootstrap } from '@nestjs/common';
+import { EventBus, IEvent, IEventPublisher } from '@nestjs/cqrs';
+import { EventStore } from '@task-stream/api/shared/application';
+import { VersionedAggregateRoot } from '@task-stream/api/shared/domain';
+import { EventSerializer } from '../serializers/event.serializer';
+
+/**
+ * Replaces the EventBus publisher: `aggregate.commit()` appends the events
+ * to the event store instead of dispatching them in memory. Handlers see
+ * them later through the EventsBridge.
+ */
+@Injectable()
+export class EventStorePublisher
+  implements OnApplicationBootstrap, IEventPublisher
+{
+  constructor(
+    private readonly eventStore: EventStore,
+    private readonly eventBus: EventBus,
+    private readonly eventSerializer: EventSerializer,
+  ) {}
+
+  onApplicationBootstrap() {
+    this.eventBus.publisher = this;
+  }
+
+  publish<T extends IEvent = IEvent>(
+    event: T,
+    dispatcher: VersionedAggregateRoot,
+  ) {
+    return this.publishAll([event], dispatcher);
+  }
+
+  publishAll<T extends IEvent = IEvent>(
+    events: T[],
+    dispatcher: VersionedAggregateRoot,
+  ) {
+    const serializableEvents = events
+      .map((event) => this.eventSerializer.serialize(event, dispatcher))
+      .map((serializableEvent, index) => {
+        const position = dispatcher.version.value + index + 1;
+        // The stored event knows its own position: handlers use it to version read models.
+        return {
+          ...serializableEvent,
+          position,
+          data: { ...serializableEvent.data, aggregateVersion: position },
+        };
+      });
+
+    const persisted = this.eventStore.persist(serializableEvents);
+    // Lets commitAndWait() await the write and advances the aggregate version.
+    dispatcher.trackCommit(persisted, serializableEvents.length);
+    return persisted;
+  }
+}

libs/api/shared/infrastructure/src/lib/event-store/serializers/event.serializer.ts

@@ -0,0 +1,48 @@
+import { Injectable } from '@nestjs/common';
+import {
+  SerializableEvent,
+  VersionedAggregateRoot,
+} from '@task-stream/api/shared/domain';
+
+@Injectable()
+export class EventSerializer {
+  serialize<T>(
+    event: T,
+    dispatcher: VersionedAggregateRoot,
+  ): SerializableEvent<T> {
+    const eventType = (event as object).constructor?.name;
+    if (!eventType) {
+      throw new Error('Incompatible event type');
+    }
+
+    const aggregateId = dispatcher.id;
+    return {
+      streamId: aggregateId,
+      position: dispatcher.version.value + 1,
+      type: eventType,
+      data: this.toJSON(event) as SerializableEvent<T>['data'],
+    };
+  }
+
+  private toJSON<T>(data: T): unknown {
+    if (typeof data !== 'object' || data === null) {
+      return data;
+    }
+
+    if ('toJSON' in data && typeof data.toJSON === 'function') {
+      return data.toJSON();
+    }
+
+    if (Array.isArray(data)) {
+      return data.map((item) => this.toJSON(item));
+    }
+
+    return Object.entries(data).reduce<Record<string, unknown>>(
+      (acc, [key, value]) => {
+        acc[key] = this.toJSON(value);
+        return acc;
+      },
+      {},
+    );
+  }
+}