NestJs Event Store

NestJS CQRS module with support EventStore.org and NATS Streaming. It requires @nestjs/cqrs.

Installation

$ yarn add @juicycleff/nestjs-event-store
$ yarn add node-nats-streaming node-eventstore-client

Description

This module aims to bridge the gap between NestJs and popular event store brokers like Event Store and NATS Streaming with support for kafka coming.

Event Store

It supports all different subscription strategies in EventStore.Org, such as Volatile, CatchUp and Persistent subscriptions fairly easily. There is support for a storage adapter interface for storing catchup events type last checkpoint position, so the checkpoint can be read on start up; The adapter interface is very slim and easy and can be assigned preferably using the EventStoreModule.registerFeatureAsync method. Adapter data store examples coming soon.

Note: if your event broker type is Event Store then featureStreamName should look like '$ce-user', then you should name your domain argument should be user without $ce, for example.

export class UserCreatedEvent implements IEvent {
    constructor(
        public readonly user: any // This what im talking about.
    )  { }
}

The way this works is we group the event based the first argument in the constructor name and this argument name must be a substring of featureStreamName. I'm sorry you can't pass you your own unique name at the moment, but I will add support for it

NATS Streaming

It supports all both durable/persistent subscription with shared subscription and volatile. It does not have the limitations of Event Store stated above.

Note: if your event broker type is NATS then featureStreamName should look like 'user'.

Setup from versions from v3.1.15

Setup NATS

Setup root app module for NATS

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    EventStoreModule.register({
      type: 'nats',
      groupId: 'groupId',
      clusterId: 'clusterId',
      clientId: 'clientId', // Optional (Auto generated with uuid)
      options: {
        url: 'nats://localhost:4222',
        reconnect: true,
        maxReconnectAttempts: -1,
      },
    }),
  ]
})
export class AppModule {}

Setup async root app module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';
import { EventStoreConfigService } from './eventstore-config.service';

@Module({
  imports: [
    EventStoreModule.registerAsync({
      type: 'nats',
      useClass: EventStoreConfigService
    }),
  ]
})
export class AppModule {}

Setup feature module

import { Module } from '@nestjs/common';
import { UsersController } from './users.controller';
import { AccountEventHandlers, UserLoggedInEvent } from '@ultimatebackend/core';
import { AccountSagas } from '../common';
import { EventStoreModule, EventStoreSubscriptionType } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    EventStoreModule.registerFeature({
      featureStreamName: 'user',
      type: 'nats',
      subscriptions: [
        {
          type: EventStoreSubscriptionType.Persistent,
          stream: 'account',
          durableName: 'svc-user',
        }
      ],
      eventHandlers: {
        UserLoggedInEvent: (data) => new UserLoggedInEvent(data),
      },
    })
  ],
  providers: [...AccountEventHandlers, AccountSagas],
  controllers: [UsersController],
})
export class UsersModule {}

Setup EventStore

Setup root app module for EventStore

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    EventStoreModule.register({
      type: 'event-store',
      tcpEndpoint: {
        host: 'localhost',
        port: 1113,
      },
      options: {
        maxRetries: 1000, // Optional
        maxReconnections: 1000,  // Optional
        reconnectionDelay: 1000,  // Optional
        heartbeatInterval: 1000,  // Optional
        heartbeatTimeout: 1000,  // Optional
        defaultUserCredentials: {
          password: 'admin',
          username: 'chnageit',
        },
      },
    }),
  ]
})
export class AppModule {}

Setup async root app module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';
import { EventStoreConfigService } from './eventstore-config.service';

@Module({
  imports: [
    EventStoreModule.registerAsync({
      type: 'event-store',
      useClass: EventStoreConfigService
    }),
  ]
})
export class AppModule {}

Setup feature module

import { Module } from '@nestjs/common';
import { CommandBus, CqrsModule, EventBus } from '@nestjs/cqrs';
import { EventStoreModule, EventStore, EventStoreSubscriptionType } from '@juicycleff/nestjs-event-store';

import {
  UserCommandHandlers,
  UserLoggedInEvent,
  UserEventHandlers,
  UserQueryHandlers,
} from '../cqrs';
import { UserSagas } from './sagas';
import { MongoStore } from './mongo-eventstore-adapter';

@Module({
  imports: [
    CqrsModule,
    EventStoreModule.registerFeature({
      featureStreamName: '$ce-user',
      type: 'event-store',
      store: MongoStore, // Optional mongo store for persisting catchup events position for microservices to mitigate failures. Must implement IAdapterStore
      subscriptions: [
        {
          type: EventStoreSubscriptionType.CatchUp,
          stream: '$ce-user',
          resolveLinkTos: true, // Default is true (Optional)
          lastCheckpoint: 13, // Default is 0 (Optional)
        },
      ],
      eventHandlers: {
        UserLoggedInEvent: (data) => new UserLoggedInEvent(data),
      },
    }),
  ],
  
  providers: [
    UserSagas,
    ...UserQueryHandlers,
    ...UserCommandHandlers,
    ...UserEventHandlers,
  ],
})
export class UserModule {}

Setup from versions from v3.0.0 to 3.0.5

Setup root app module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    EventStoreModule.register({
      tcpEndpoint: {
        host: process.env.ES_TCP_HOSTNAME || AppConfig.eventstore?.hostname,
        port: parseInt(process.env.ES_TCP_PORT, 10) || AppConfig.eventstore?.tcpPort,
      },
      options: {
        maxRetries: 1000, // Optional
        maxReconnections: 1000,  // Optional
        reconnectionDelay: 1000,  // Optional
        heartbeatInterval: 1000,  // Optional
        heartbeatTimeout: 1000,  // Optional
        defaultUserCredentials: {
          password: AppConfig.eventstore?.tcpPassword,
          username: AppConfig.eventstore?.tcpUsername,
        },
      },
    }),
  ]
})
export class AppModule {}

Setup async root app module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';
import { EventStoreConfigService } from './eventstore-config.service';

@Module({
  imports: [
    EventStoreModule.registerAsync({
      useClass: EventStoreConfigService
    }),
  ]
})
export class AppModule {}

Setup module

Note featureStreamName field is not important if you're subscription type is persistent'

Setup feature module

import { Module } from '@nestjs/common';
import { CommandBus, CqrsModule, EventBus } from '@nestjs/cqrs';
import { EventStoreModule, EventStore, EventStoreSubscriptionType } from '@juicycleff/nestjs-event-store';

import {
  UserCommandHandlers,
  UserCreatedEvent,
  UserEventHandlers,
  UserQueryHandlers,
} from '../cqrs';
import { UserSagas } from './sagas';
import { MongoStore } from './mongo-eventstore-adapter';

@Module({
  imports: [
    CqrsModule,
    EventStoreModule.registerFeature({
      featureStreamName: '$ce-user',
      store: MongoStore, // Optional mongo store for persisting catchup events position for microservices to mitigate failures. Must implement IAdapterStore
      subscriptions: [
        {
          type: EventStoreSubscriptionType.CatchUp,
          stream: '$ce-user',
          resolveLinkTos: true, // Default is true (Optional)
          lastCheckpoint: 13, // Default is 0 (Optional)
        },
        {
          type: EventStoreSubscriptionType.Volatile,
          stream: '$ce-user',
        },
        {
          type: EventStoreSubscriptionType.Persistent,
          stream: '$ce-user',
          persistentSubscriptionName: 'steamName',
          resolveLinkTos: true,  // Default is true (Optional)
        },
      ],
      eventHandlers: {
        UserLoggedInEvent: (data) => new UserLoggedInEvent(data),
        UserRegisteredEvent: (data) => new UserRegisteredEvent(data),
        EmailVerifiedEvent: (data) => new EmailVerifiedEvent(data),
      },
    }),
  ],
  
  providers: [
    UserSagas,
    ...UserQueryHandlers,
    ...UserCommandHandlers,
    ...UserEventHandlers,
  ],
})
export class UserModule {}

Setup async feature module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';
import { EventStoreFeatureService } from './user-eventstore-feature.service';

@Module({
  imports: [
    EventStoreModule.registerFeatureAsync({
      useClass: EventStoreFeatureService
    }),
  ]
})
export class AppModule {}

Setup from versions below v2.0.0

Setup root app module

import { Module } from '@nestjs/common';
import { NestjsEventStoreModule } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    NestjsEventStoreModule.forRoot({
      http: {
        port: parseInt(process.env.ES_HTTP_PORT, 10),
        protocol: process.env.ES_HTTP_PROTOCOL,
      },
      tcp: {
        credentials: {
          password: process.env.ES_TCP_PASSWORD,
          username: process.env.ES_TCP_USERNAME,
        },
        hostname: process.env.ES_TCP_HOSTNAME,
        port: parseInt(process.env.ES_TCP_PORT, 10),
        protocol: process.env.ES_TCP_PROTOCOL,
      },
    }),
  ]
})
export class AppModule {}

Setup module

import { Module } from '@nestjs/common';
import { CommandBus, CqrsModule, EventBus } from '@nestjs/cqrs';
import { NestjsEventStoreModule, EventStore } from '@juicycleff/nestjs-event-store';

import {
  UserCommandHandlers,
  UserCreatedEvent,
  UserEventHandlers,
  UserQueryHandlers,
} from '../cqrs';
import { UserSagas } from './sagas';

@Module({
  imports: [
    CqrsModule,
    NestjsEventStoreModule.forFeature({
      name: 'user',
      resolveLinkTos: false,
    }),
  ],
  
  providers: [
    UserSagas,
    ...UserQueryHandlers,
    ...UserCommandHandlers,
    ...UserEventHandlers,
  ],
})
export class UserModule {
  constructor(
    private readonly command$: CommandBus,
    private readonly event$: EventBus,
    private readonly eventStore: EventStore,
  ) {}

  onModuleInit(): any {
    this.eventStore.setEventHandlers(this.eventHandlers);
    this.eventStore.bridgeEventsTo((this.event$ as any).subject$);
    this.event$.publisher = this.eventStore;
  }

  eventHandlers = {
    UserCreatedEvent: (data) => new UserCreatedEvent(data),
  };
}

Notice

2.0.0 release inspired by nestjs-eventstore

License

This project is MIT licensed.