Backend | CB-222 NestJS App structure for DDD on CQRS write-side

- Project structure for write-side of CQRS
- Base classes for Aggregates and Domain Events implementations.
- EventStorage library with in-memory and postgres implementations.
- App time source-of-truth by TimeProvider.
- Sample of functionality with inviting applicant to CodersCrew, based on EventStorming.

Author: MateuszNaKodach

backend/.eslintrc.js

@@ -20,5 +20,8 @@ module.exports = {
     '@typescript-eslint/interface-name-prefix': 'off',
     '@typescript-eslint/explicit-function-return-type': 'off',
     '@typescript-eslint/no-explicit-any': 'off',
+    "@typescript-eslint/no-namespace": "off",
+    "@typescript-eslint/no-empty-interface": "off",
+    "@typescript-eslint/no-use-before-define": "off"
   },
 };

backend/.prettierrc

@@ -1,4 +1,4 @@
 {
   "singleQuote": true,
   "trailingComma": "all"
-}
+}

backend/Dockerfile

@@ -1,25 +1,48 @@
-version: "3.7"
-
-services:
-  coders-board:
-    build:
-      args:
-        NODE_ENV: development
-        IMAGE_BASE_DEV: node:alpine
-        IMAGE_BASE_PROD: node:alpine
-      context: .
-      target: development
-
-    volumes:
-      - .:/app
-      - /app/node_modules
-    ports:
-      - 3000:3000
-      - 9229:9229
-    command:
-      - start:debug
-    networks:
-      - coders-board-network
-networks:
-  coders-board-network:
-    name: coders-board-network
+version: '3.7'
+
+services:
+  coders-board:
+    build:
+      args:
+        NODE_ENV: development
+        IMAGE_BASE_DEV: node:alpine
+        IMAGE_BASE_PROD: node:alpine
+      context: .
+      target: development
+
+    container_name: coders-board-nestjs
+    volumes:
+      - .:/app
+      - /app/node_modules
+    ports:
+      - 3000:3000
+      - 9229:9229
+    environment:
+      DATABASE_MODE: in-memory
+      DATABASE_HOST: postgres
+      DATABASE_PORT: 5432
+      DATABASE_USERNAME: postgres
+      DATABASE_PASSWORD: postgres
+    command:
+      - start:debug
+    networks:
+      - coders-board-network
+
+  postgres:
+    image: postgres:12
+    container_name: coders-board-postgres
+    restart: always
+    ports:
+      - 5002:5432
+    environment:
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_USER: postgres
+      POSTGRES_DB: coders-board
+    networks:
+      - coders-board-network
+    #volumes:
+    # - ../pgdata:/var/lib/postgresql/data #keep data outside container
+
+networks:
+  coders-board-network:
+    name: coders-board-network

backend/docker-compose.yml

@@ -0,0 +1,7 @@
+export class EventStreamVersion {
+  constructor(readonly raw: number) {}
+
+  static exactly(raw: number) {
+    return new EventStreamVersion(raw);
+  }
+}

backend/libs/event-sourcing/src/api/event-stream-version.valueobject.ts

@@ -0,0 +1,8 @@
+export interface StorageEventEntry {
+  readonly eventId: string;
+  readonly eventType: string;
+  readonly occurredAt: Date;
+  readonly aggregateId: string;
+  readonly aggregateType: string;
+  readonly payload: any;
+}

backend/libs/event-sourcing/src/api/storage-event-entry.ts

@@ -0,0 +1,13 @@
+import { ModuleMetadata, Type } from '@nestjs/common/interfaces';
+import { EventSourcingModuleConfigFactory } from '@coders-board-library/event-sourcing/event-sourcing.module-config-factory';
+import { EventSourcingModuleConfig } from '@coders-board-library/event-sourcing/event-sourcing.module-config';
+
+export interface EventSourcingModuleAsyncConfig
+  extends Pick<ModuleMetadata, 'imports'> {
+  inject?: any[];
+  useExisting?: Type<EventSourcingModuleConfigFactory>;
+  useClass?: Type<EventSourcingModuleConfigFactory>;
+  useFactory?: (
+    ...args: any[]
+  ) => Promise<EventSourcingModuleConfig> | EventSourcingModuleConfig;
+}

backend/libs/event-sourcing/src/event-sourcing.module-async-config.ts

@@ -0,0 +1,7 @@
+import { EventSourcingModuleConfig } from '@coders-board-library/event-sourcing/event-sourcing.module-config';
+
+export interface EventSourcingModuleConfigFactory {
+  createModuleConfig():
+    | Promise<EventSourcingModuleConfig>
+    | EventSourcingModuleConfig;
+}

backend/libs/event-sourcing/src/event-sourcing.module-config-factory.ts

@@ -0,0 +1,13 @@
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { Time } from './time.type';
+
+export type EventSourcingModuleConfig =
+  | {
+      time: Time;
+      eventStorage: 'in-memory';
+    }
+  | {
+      time: Time;
+      eventStorage: 'typeorm';
+      typeOrmModule: TypeOrmModule;
+    };

backend/libs/event-sourcing/src/event-sourcing.module-config.ts

@@ -0,0 +1,96 @@
+import { DynamicModule, Module, Provider } from '@nestjs/common';
+import { EVENT_STORAGE, EventStorage } from './event-storage/event-storage';
+import { TypeOrmEventStorage } from './event-storage/typeorm/event-storage.typeorm';
+import { InMemoryEventStorage } from './event-storage/in-memory/event-storage.in-memory';
+import { DomainEventEntity } from './event-storage/typeorm/event.typeorm-entity';
+import { EventSourcingModuleConfig } from './event-sourcing.module-config';
+import { Repository } from 'typeorm';
+import { EventSourcingModuleAsyncConfig } from '@coders-board-library/event-sourcing/event-sourcing.module-async-config';
+import { Time } from '@coders-board-library/event-sourcing/time.type';
+import { EventSourcingModuleConfigFactory } from '@coders-board-library/event-sourcing/event-sourcing.module-config-factory';
+
+const EVENT_SOURCING_CONFIG = Symbol();
+
+@Module({})
+export class EventSourcingModule {
+  static register(config: EventSourcingModuleConfig): DynamicModule {
+    const configProvider: Provider = {
+      provide: EVENT_SOURCING_CONFIG,
+      useValue: config,
+    };
+    if (config.eventStorage === 'typeorm') {
+      const optionalImports = [];
+      optionalImports.push(config.typeOrmModule);
+      return {
+        module: EventSourcingModule,
+        imports: [...optionalImports],
+        providers: [
+          {
+            provide: EVENT_STORAGE,
+            useFactory: (typeOrmRepository: Repository<DomainEventEntity>) =>
+              new TypeOrmEventStorage(config.time, typeOrmRepository),
+          },
+        ],
+        exports: [EVENT_STORAGE],
+      };
+    }
+
+    return {
+      module: EventSourcingModule,
+      providers: [
+        configProvider,
+        {
+          provide: EVENT_STORAGE,
+          useFactory: (config: EventSourcingModuleConfig) =>
+            new InMemoryEventStorage(config.time),
+          inject: [EVENT_SOURCING_CONFIG],
+        },
+      ],
+      exports: [EVENT_STORAGE],
+    };
+  }
+
+  static registerAsync(config: EventSourcingModuleAsyncConfig): DynamicModule {
+    return {
+      module: EventSourcingModule,
+      imports: config.imports || [],
+      providers: [
+        this.createAsyncProviders(config),
+        {
+          provide: EVENT_STORAGE,
+          useFactory: (config: EventSourcingModuleConfig) =>
+            new InMemoryEventStorage(config.time),
+          inject: [EVENT_SOURCING_CONFIG],
+        },
+      ],
+      exports: [EVENT_STORAGE],
+    };
+  }
+
+  private static createAsyncProviders(
+    config: EventSourcingModuleAsyncConfig,
+  ): Provider {
+    if (config) {
+      if (config.useFactory) {
+        return {
+          provide: EVENT_SOURCING_CONFIG,
+          useFactory: config.useFactory,
+          inject: config.inject || [],
+        };
+      } else {
+        return {
+          provide: EVENT_SOURCING_CONFIG,
+          useFactory: async (
+            optionsFactory: EventSourcingModuleConfigFactory,
+          ) => await optionsFactory.createModuleConfig(),
+          inject: [config.useExisting || config.useClass],
+        };
+      }
+    } else {
+      return {
+        provide: EVENT_SOURCING_CONFIG,
+        useValue: {},
+      };
+    }
+  }
+}

backend/libs/event-sourcing/src/event-sourcing.module.ts

@@ -0,0 +1,20 @@
+import { EventStreamVersion } from '../api/event-stream-version.valueobject';
+import { StorageEventEntry } from '../api/storage-event-entry';
+
+export const EVENT_STORAGE = 'EventStorage';
+
+export interface EventStorage {
+  store(
+    event: StorageEventEntry,
+    expectedVersion?: EventStreamVersion,
+  ): Promise<void>;
+
+  //TODO: Consider interface change to return stored events and errors
+  // or leave only method for store one event or do storeAll in one transaction
+  storeAll(
+    events: StorageEventEntry[],
+    expectedVersion?: EventStreamVersion,
+  ): Promise<void>;
+
+  readEvents(aggregateId: string, toDate?: Date): Promise<StorageEventEntry[]>;
+}

backend/libs/event-sourcing/src/event-storage/event-storage.ts

@@ -0,0 +1,119 @@
+import { InMemoryEventStorage } from '@coders-board-library/event-sourcing/event-storage/in-memory/event-storage.in-memory';
+import { EventStorage } from '@coders-board-library/event-sourcing/event-storage/event-storage';
+import * as moment from 'moment';
+import { EventStreamVersion } from '@coders-board-library/event-sourcing/api/event-stream-version.valueobject';
+
+const time = {
+  15_00: moment.utc(new Date(2020, 4, 7, 15, 0)).toDate(),
+  15_20: moment.utc(new Date(2020, 4, 7, 15, 20)).toDate(),
+  15_30: moment.utc(new Date(2020, 4, 7, 15, 30)).toDate(),
+  15_40: moment.utc(new Date(2020, 4, 7, 15, 40)).toDate(),
+};
+
+const events = {
+  aggregate1: {
+    id: 'aggregateId1',
+    event1: {
+      eventId: 'eventId1.1',
+      eventType: 'EVENT_TYPE_1',
+      aggregateId: 'aggregateId1',
+      aggregateType: 'aggregateType1',
+      occurredAt: time['1530'],
+      payload: { value: 'value' },
+    },
+    event2: {
+      eventId: 'eventId1.2',
+      eventType: 'EVENT_TYPE_2',
+      aggregateId: 'aggregateId1',
+      aggregateType: 'aggregateType1',
+      occurredAt: time['1540'],
+      payload: {},
+    },
+  },
+  aggregate2: {
+    id: 'aggregateId2',
+    event1: {
+      eventId: 'eventId2.1',
+      eventType: 'EVENT_TYPE_1',
+      aggregateId: 'aggregateId2',
+      aggregateType: 'aggregateType2',
+      occurredAt: time['1530'],
+      payload: { value: 'value' },
+    },
+    event2: {
+      eventId: 'eventId2.2',
+      eventType: 'EVENT_TYPE_2',
+      aggregateId: 'aggregateId2',
+      aggregateType: 'aggregateType2',
+      occurredAt: time['1540'],
+      payload: {},
+    },
+  },
+};
+
+describe('Feature: In memory event storage', () => {
+  let currentDate: Date;
+  let eventStorage: EventStorage;
+
+  beforeEach(() => {
+    eventStorage = new InMemoryEventStorage(() => currentDate);
+  });
+
+  describe('Given: Events to store', () => {
+    describe('When: store the events', () => {
+      beforeEach(() => {
+        eventStorage.store(events.aggregate1.event1);
+        eventStorage.store(events.aggregate2.event1);
+        eventStorage.store(events.aggregate1.event2);
+        eventStorage.store(events.aggregate2.event2);
+      });
+
+      it('Then: The event should be queryable by all event', async () => {
+        currentDate = time['1540'];
+        const stored = await eventStorage.readEvents(events.aggregate1.id);
+        expect(stored).toContain(events.aggregate1.event1);
+        expect(stored).toContain(events.aggregate1.event2);
+      });
+
+      it('Then: The event should be queryable by time', async () => {
+        expect(
+          await eventStorage.readEvents(events.aggregate1.id, time['1520']),
+        ).toStrictEqual([]);
+        expect(
+          await eventStorage.readEvents(events.aggregate1.id, time['1530']),
+        ).toContain(events.aggregate1.event1);
+
+        expect(
+          await eventStorage.readEvents(events.aggregate1.id, time['1540']),
+        ).toContain(events.aggregate1.event1);
+        expect(
+          await eventStorage.readEvents(events.aggregate1.id, time['1540']),
+        ).toContain(events.aggregate1.event2);
+      });
+
+      it('Then: The event cannot be stored twice', async () => {
+        await expect(
+          eventStorage.store(events.aggregate1.event1),
+        ).rejects.toMatch(
+          `Event stream already contains this event with id ${events.aggregate1.event1.eventId}!`,
+        );
+      });
+
+      it('Then: The event cannot be stored if aggregate was modified', async () => {
+        const anotherEvent2 = {
+          eventId: 'eventId1.2_2',
+          eventType: 'EVENT_TYPE_2',
+          aggregateId: 'aggregateId1',
+          aggregateType: 'aggregateType1',
+          occurredAt: time['1540'],
+          payload: {},
+        };
+        await expect(
+          eventStorage.store(anotherEvent2, EventStreamVersion.exactly(1)),
+        ).rejects.toMatch(
+          `Event stream for aggregate was modified! Expected version: 1, but actual is: 2`,
+        );
+      });
+    });
+  });
+});