Recover + fixes (#62)

* 2.x Recovered

* Added 2.1.1 changes

.github/workflows/publish_workflow.yml

@@ -19,7 +19,7 @@ jobs:
       - name: Setup node version
         uses: actions/setup-node@v1
         with:
-          node-version: 12
+          node-version: 14
           registry-url: https://registry.npmjs.org/
 
       - name: Install

examples/README.md

@@ -19,7 +19,7 @@ yarn install
 ## Start
 ```
 docker run -d -p 2113:2113 -p 1113:1113 -e EVENSTORE_RUN_PROJECTIONS=System --name geteventstore eventstore/eventstore
-yarn start
+yarn start // or yarn start:<cqrs | eventstore | subscription | write>
 ```
 
 ## Usage 

examples/package.json

@@ -4,34 +4,42 @@
   "description": "Nest CQRS module usage example",
   "license": "MIT",
   "scripts": {
-    "start": "ts-node src/main.ts",
-    "start:dev": "nest build --webpack --webpackPath webpack-hmr.config.js"
+    "start": "ts-node-dev src/bin/full.ts",
+    "start:cqrs": "ts-node-dev src/bin/cqrs.ts",
+    "start:eventstore": "ts-node-dev src/bin/full.ts",
+    "start:write": "ts-node-dev src/bin/write.ts",
+    "start:subscription": "ts-node-dev src/bin/subscription.ts",
+    "start:projection": "ts-node-dev src/bin/projection.ts"
   },
   "dependencies": {
     "@godaddy/terminus": "^4.4.1",
     "@nestjs/common": "^7.0.0",
     "@nestjs/core": "^7.0.0",
-    "@nestjs/cqrs": "^7.0.0",
+    "@nestjs/cqrs": "^7.0.1",
     "@nestjs/platform-express": "^6.0.0",
     "@nestjs/terminus": "^7.0.1",
+    "class-transformer": "^0.4.0",
+    "class-validator": "^0.13.1",
     "cli-color": "^1.4.0",
     "global": "^4.4.0",
+    "nestjs-context": "^0.1.1",
     "nestjs-geteventstore": "^1.6.0",
     "nestjs-pino-stackdriver": "^1.0.0",
-    "reflect-metadata": "^0.1.12",
-    "rxjs": "^6.4.0"
+    "reflect-metadata": "^0.1.13",
+    "rxjs": "^7.0.0"
   },
   "devDependencies": {
-    "dot-prop": ">=4.2.1",
-    "serialize-javascript": ">=3.1.0",
     "@compodoc/compodoc": "^1.1.11",
     "@types/express": "^4.16.1",
     "@types/node": "^11.9.4",
+    "dot-prop": ">=4.2.1",
     "nodemon": "^1.18.10",
     "prettier": "^1.12.1",
+    "serialize-javascript": ">=3.1.0",
     "start-server-webpack-plugin": "^2.2.5",
     "ts-loader": "^7.0.4",
     "ts-node": "^8.0.2",
+    "ts-node-dev": "^1.1.6",
     "typescript": "^3.3.3",
     "webpack": "^4.43.0",
     "webpack-node-externals": "^1.7.2"

examples/src/sentry.exception-filter.ts → examples/src/all-exception.filter.ts

@@ -2,7 +2,7 @@ import { ExceptionFilter, Catch, ArgumentsHost } from '@nestjs/common';
 import { Request, Response } from 'express';
 
 @Catch()
-export class SentryExceptionFilter implements ExceptionFilter {
+export class AllExceptionFilter implements ExceptionFilter {
   catch(exception: Error, host: ArgumentsHost) {
     const ctx = host.switchToHttp();
     const response = ctx.getResponse<Response>();

examples/src/main.ts → examples/src/bin/cqrs.ts

@@ -1,20 +1,20 @@
 import { NestFactory } from '@nestjs/core';
-import { EventStoredApplicationModule } from './app.module';
-import { SentryExceptionFilter } from './sentry.exception-filter';
-import { Logger } from 'nestjs-pino-stackdriver/dist';
+import { AllExceptionFilter } from '../all-exception.filter';
+import { Logger } from 'nestjs-pino-stackdriver';
+import { HeroesGameModule } from '../heroes/heroes.module';
 
 declare const module: any;
 
 async function bootstrap() {
-  const app = await NestFactory.create(EventStoredApplicationModule, {
+  const app = await NestFactory.create(HeroesGameModule, {
     logger: new Logger(),
   });
 
   if (module.hot) {
     module.hot.accept();
     module.hot.dispose(() => app.close());
   }
-  app.useGlobalFilters(new SentryExceptionFilter());
+  app.useGlobalFilters(new AllExceptionFilter());
   await app.listen(3000, () =>
     console.log('Application is listening on port 3000.'),
   );

examples/src/bin/full.ts

@@ -0,0 +1,23 @@
+import { NestFactory } from '@nestjs/core';
+import { AllExceptionFilter } from '../all-exception.filter';
+import { Logger } from 'nestjs-pino-stackdriver';
+import { EventStoreHeroesModule } from '../heroes/event-store-heroes.module';
+
+declare const module: any;
+
+async function bootstrap() {
+  const app = await NestFactory.create(EventStoreHeroesModule.register(), {
+    logger: new Logger(),
+  });
+
+  if (module.hot) {
+    module.hot.accept();
+    module.hot.dispose(() => app.close());
+  }
+  app.useGlobalFilters(new AllExceptionFilter());
+  await app.listen(3000, () =>
+    console.log('Application is listening on port 3000.'),
+  );
+}
+
+bootstrap();

examples/src/bin/projection.ts

@@ -0,0 +1,26 @@
+import { NestFactory } from '@nestjs/core';
+import { AllExceptionFilter } from '../all-exception.filter';
+import { Logger } from 'nestjs-pino-stackdriver';
+import { EventStoreHeroesModule } from '../heroes/event-store-heroes.module';
+
+declare const module: any;
+
+async function bootstrap() {
+  const app = await NestFactory.create(
+    EventStoreHeroesModule.registerProjection(),
+    {
+      logger: new Logger(),
+    },
+  );
+
+  if (module.hot) {
+    module.hot.accept();
+    module.hot.dispose(() => app.close());
+  }
+  app.useGlobalFilters(new AllExceptionFilter());
+  await app.listen(3000, () =>
+    console.log('Application is listening on port 3000.'),
+  );
+}
+
+bootstrap();

examples/src/bin/subscription.ts

@@ -0,0 +1,26 @@
+import { NestFactory } from '@nestjs/core';
+import { AllExceptionFilter } from '../all-exception.filter';
+import { Logger } from 'nestjs-pino-stackdriver';
+import { EventStoreHeroesModule } from '../heroes/event-store-heroes.module';
+
+declare const module: any;
+
+async function bootstrap() {
+  const app = await NestFactory.create(
+    EventStoreHeroesModule.registerSubscriptions(),
+    {
+      logger: new Logger(),
+    },
+  );
+
+  if (module.hot) {
+    module.hot.accept();
+    module.hot.dispose(() => app.close());
+  }
+  app.useGlobalFilters(new AllExceptionFilter());
+  await app.listen(3000, () =>
+    console.log('Application is listening on port 3000.'),
+  );
+}
+
+bootstrap();

examples/src/bin/write.ts

@@ -0,0 +1,26 @@
+import { NestFactory } from '@nestjs/core';
+import { AllExceptionFilter } from '../all-exception.filter';
+import { Logger } from 'nestjs-pino-stackdriver';
+import { EventStoreHeroesModule } from '../heroes/event-store-heroes.module';
+
+declare const module: any;
+
+async function bootstrap() {
+  const app = await NestFactory.create(
+    EventStoreHeroesModule.registerWriteBus(),
+    {
+      logger: new Logger(),
+    },
+  );
+
+  if (module.hot) {
+    module.hot.accept();
+    module.hot.dispose(() => app.close());
+  }
+  app.useGlobalFilters(new AllExceptionFilter());
+  await app.listen(3000, () =>
+    console.log('Application is listening on port 3000.'),
+  );
+}
+
+bootstrap();

examples/src/heroes/aggregates/hero.aggregate.ts

@@ -2,30 +2,29 @@ import { HeroFoundItemEvent } from '../events/impl/hero-found-item.event';
 import { HeroKilledDragonEvent } from '../events/impl/hero-killed-dragon.event';
 import { HeroDropItemEvent } from '../events/impl/hero-drop-item.event';
 import { HeroDamagedEnemyEvent } from '../events/impl/hero-damaged-enemy.event';
-import { EventStoreAggregateRoot } from '../../../../src/cqrs/event-store.aggregate-root';
+import { EventStoreAggregateRoot } from '../../../../src';
 
 export class Hero extends EventStoreAggregateRoot {
-  constructor(private id) {
+  constructor(private readonly id) {
     super();
-    this.streamConfig = {
-      streamName: `hero-${id}`,
-    };
+    // comment this line to test correlation-id auto-generated stream
+    this.streamName = `hero-${id}`;
   }
 
   damageEnemy(dragonId: string, hitPoint: number) {
-    // logic
-    this.apply(
+    return this.apply(
       new HeroDamagedEnemyEvent({
         heroId: this.id,
+        // comment dragonId if you want to test validation,
         dragonId,
         hitPoint: hitPoint,
-      }),
+      } as any),
     );
   }
 
   killEnemy(dragonId: string) {
     // logic
-    this.apply(
+    return this.apply(
       new HeroKilledDragonEvent({
         heroId: this.id,
         dragonId,
@@ -35,7 +34,7 @@ export class Hero extends EventStoreAggregateRoot {
 
   addItem(itemId: string) {
     // logic
-    this.apply(
+    return this.apply(
       new HeroFoundItemEvent({
         heroId: this.id,
         itemId,
@@ -44,10 +43,10 @@ export class Hero extends EventStoreAggregateRoot {
   }
 
   dropItem(itemId: string) {
-    this.apply(
+    return this.apply(
       new HeroDropItemEvent({
         heroId: this.id,
-        itemId: 'string',
+        itemId,
       }),
     );
   }

examples/src/heroes/commands/handlers/drop-ancient-item.handler.ts

@@ -2,27 +2,26 @@ import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
 import * as clc from 'cli-color';
 import { HeroRepository } from '../../repository/hero.repository';
 import { DropAncientItemCommand } from '../impl/drop-ancient-item.command';
-import { EventStorePublisher } from '../../../../../src/';
+import { WriteEventBus } from '../../../../../src/';
 
 @CommandHandler(DropAncientItemCommand)
 export class DropAncientItemHandler
   implements ICommandHandler<DropAncientItemCommand> {
   constructor(
     private readonly repository: HeroRepository,
-    private readonly publisher: EventStorePublisher,
+    private readonly publisher: WriteEventBus,
   ) {}
 
   async execute(command: DropAncientItemCommand) {
     console.log(clc.yellowBright('Async DropAncientItemCommand...'));
 
     const { heroId, itemId } = command;
-    // Use default aggregate config
-    const hero = this.publisher.mergeObjectContext(
-      await this.repository.findOneById(+heroId),
+    const hero = await this.repository.findOneById(+heroId);
+    hero.autoCommit = true;
+    hero.addPublisher((events, expectedVersion, streamName) =>
+      this.publisher.publishAll(events, expectedVersion, streamName),
     );
     hero.addItem(itemId);
     hero.dropItem(itemId);
-
-    hero.commit();
   }
 }

examples/src/heroes/commands/handlers/kill-dragon.handler.ts

@@ -2,13 +2,13 @@ import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
 import * as clc from 'cli-color';
 import { HeroRepository } from '../../repository/hero.repository';
 import { KillDragonCommand } from '../impl/kill-dragon.command';
-import { DAY, EventStorePublisher, ExpectedVersion } from '../../../../../src/';
+import { ExpectedVersion, WriteEventBus } from '../../../../../src';
 
 @CommandHandler(KillDragonCommand)
 export class KillDragonHandler implements ICommandHandler<KillDragonCommand> {
   constructor(
     private readonly repository: HeroRepository,
-    private readonly publisher: EventStorePublisher,
+    private readonly publisher: WriteEventBus,
   ) {}
 
   async execute(command: KillDragonCommand) {
@@ -19,41 +19,25 @@ export class KillDragonHandler implements ICommandHandler<KillDragonCommand> {
         `KillDragonCommand... for hero ${heroId} on enemy ${dragonId}`,
       ),
     );
-    // build aggregate by fetching data from database
-    // add publish capacity to the aggregate root
-    const hero = this.publisher.mergeObjectContext(
-      await this.repository.findOneById(+heroId),
-    );
-    // Use custom stream only for this process
-    await hero.setStreamConfig({
-      streamName: `dragon_killed-${dragonId}`,
-      // Bug if the stream is not new when writing
-      expectedVersion: ExpectedVersion.NoStream,
-      // Set retention rules for this new stream
-      metadata: {
-        $maxAge: 2 * DAY,
-        $maxCount: 5,
-      },
-    });
-
-    hero.damageEnemy(dragonId, 2);
-    hero.damageEnemy(dragonId, -8);
-    hero.damageEnemy(dragonId, 10);
-    hero.damageEnemy(dragonId, 10);
-    hero.damageEnemy(dragonId, -1);
-    hero.damageEnemy(dragonId, 10);
-    hero.damageEnemy(dragonId, 10);
-    hero.damageEnemy(dragonId, 10);
-    await hero.commit();
+    // build aggregate by fetching data from database && add publisher
+    // const hero = (await this.repository.findOneById(+heroId)).addPublisher(
+    //   this.publisher.publishAll.bind(this.publisher),
+    // );
+    const hero = (
+      await this.repository.findOneById(+heroId)
+    ).addPublisher<WriteEventBus>(this.publisher);
 
-    // Change stream for final event
-    await hero.setStreamConfig({
-      streamName: `hero-${heroId}`,
-      // It must be a new stream
-      expectedVersion: ExpectedVersion.NoStream,
-    });
-    hero.killEnemy(dragonId);
-    await hero.commit();
+    await hero.damageEnemy(dragonId, 2);
+    await hero.damageEnemy(dragonId, -8);
+    await hero.damageEnemy(dragonId, 10);
+    await hero.damageEnemy(dragonId, 10);
+    await hero.damageEnemy(dragonId, -1);
+    await hero.damageEnemy(dragonId, 10);
+    await hero.damageEnemy(dragonId, 10);
+    await hero.damageEnemy(dragonId, 10);
+    await hero.commit(ExpectedVersion.NoStream);
+    await hero.killEnemy(dragonId);
+    await hero.commit(ExpectedVersion.StreamExists);
 
     return command;
   }