refactor: code improvements and update import paths

Author: sgomez

src/decorators/inject-repository.decorator.ts

@@ -1,6 +1,6 @@
 import { Inject, Type } from '@nestjs/common';
 
-import { getRepositoryToken } from '../utils/repository';
+import { getRepositoryToken } from '../utils';
 
 export const InjectAggregateRepository = (
   aggregate: Type<unknown>,

src/domain/models/event.ts

@@ -16,9 +16,9 @@ export class Event<P = unknown> implements IEvent {
   private _payload: P;
   private _metadata: Metadata;
 
-  public constructor(aggregateId: string, payload: P) {
+  public constructor(aggregateId: string, payload?: P) {
     this.eventId = uuid.v4();
-    this._payload = Object.assign({}, payload);
+    this._payload = { ...payload };
     this.eventType = Object.getPrototypeOf(this).constructor.name;
     this._metadata = {
       _aggregate_id: aggregateId,

src/eventstore-core.module.ts

@@ -10,7 +10,7 @@ import { CqrsModule, EventBus } from '@nestjs/cqrs';
 import { ExplorerService } from '@nestjs/cqrs/dist/services/explorer.service';
 import { MongooseModule } from '@nestjs/mongoose';
 
-import { KEYS, KeySchema, KeyService } from './crypto';
+import { KEYS, KeySchema } from './crypto';
 import { Event } from './domain';
 import { EventStore } from './eventstore';
 import { EventStoreCli } from './eventstore.cli';
@@ -20,11 +20,8 @@ import {
   EVENTSTORE_SETTINGS_TOKEN,
 } from './eventstore.constants';
 import { EventStoreMapper } from './eventstore.mapper';
-import {
-  ConfigService,
-  EventStoreModuleAsyncOptions,
-} from './interfaces/eventstore-module.interface';
-import { ProjectionsService, TransformerService } from './services';
+import { ConfigService, EventStoreModuleAsyncOptions } from './interfaces';
+import { KeyService, ProjectionsService, TransformerService } from './services';
 
 @Global()
 @Module({

src/eventstore.cli.ts

@@ -1,33 +1,40 @@
-import { EventStoreDBClient, FORWARDS, START } from '@eventstore/db-client';
-import { Inject } from '@nestjs/common';
+import {
+  EventStoreDBClient,
+  FORWARDS,
+  ResolvedEvent,
+  START,
+} from '@eventstore/db-client';
+import { Inject, Logger } from '@nestjs/common';
 import { Command, Console } from 'nestjs-console';
 
+import { Event } from './domain';
 import { Config } from './eventstore.config';
 import { EVENTSTORE_SETTINGS_TOKEN } from './eventstore.constants';
 import { EventStoreMapper } from './eventstore.mapper';
 import { ProjectionsService } from './services';
 
 @Console()
 export class EventStoreCli {
-  private client: EventStoreDBClient;
-  private category: string;
+  private readonly category: string;
+  private readonly client: EventStoreDBClient;
+  private readonly logger = new Logger(EventStoreCli.name);
+  private readonly eventHandlers;
 
   constructor(
     private readonly mapper: EventStoreMapper,
-    private readonly projections: ProjectionsService,
-    @Inject(EVENTSTORE_SETTINGS_TOKEN) private readonly config: Config,
+    projections: ProjectionsService,
+    @Inject(EVENTSTORE_SETTINGS_TOKEN) config: Config,
   ) {
     this.client = EventStoreDBClient.connectionString(config.connection);
     this.category = config.category;
+    this.eventHandlers = projections.eventHandlers();
   }
 
   @Command({
     command: 'eventstore:readmodel:restore',
     description: 'Restore read model',
   })
   async restore(): Promise<void> {
-    const eventHandlers = this.projections.eventHandlers();
-
     let position: any = START;
     const MAX_COUNT = 1000;
 
@@ -46,27 +53,35 @@ export class EventStoreCli {
         break;
       }
 
-      const lastResolvedEvent = resolvedEvents[resolvedEvents.length - 1];
-      position = this.incrementRevision(lastResolvedEvent.link.revision);
+      await this.handleResolvedEvents(resolvedEvents);
+      position = this.calculateNextPosition(resolvedEvents);
+    }
 
-      const events = await this.mapper.resolvedEventsToDomainEvents(
-        resolvedEvents,
-      );
+    this.logger.log('Projections have been restored!');
+    process.exit(0);
+  }
 
-      for (const event of events) {
-        const key = event.constructor.name;
+  private async handleResolvedEvents(resolvedEvents: ResolvedEvent[]) {
+    for (const resolvedEvent of resolvedEvents) {
+      const event = await this.mapper.resolvedEventToDomainEvent(resolvedEvent);
 
-        for (const eventHandler of eventHandlers[key]) {
-          await eventHandler.handle(event);
-        }
-      }
+      if (!event) continue;
+
+      await this.handleEvent(event);
     }
+  }
 
-    console.log('View db has been restored!');
-    process.exit(0);
+  private async handleEvent(event: Event) {
+    const key = event.constructor.name;
+    for (const eventHandler of this.eventHandlers[key]) {
+      await eventHandler.handle(event);
+    }
   }
 
-  private incrementRevision(revision: bigint): bigint {
+  private calculateNextPosition(resolvedEvents: ResolvedEvent[]): bigint {
+    const lastResolvedEvent = resolvedEvents[resolvedEvents.length - 1];
+    const revision = lastResolvedEvent.link.revision;
+
     return BigInt(Number(revision) + 1);
   }
 }

src/eventstore.mapper.ts

@@ -1,14 +1,16 @@
-import { ResolvedEvent } from '@eventstore/db-client';
-import { Inject, Injectable } from '@nestjs/common';
+import { JSONType, ResolvedEvent } from '@eventstore/db-client';
+import { Inject, Injectable, Logger } from '@nestjs/common';
 
-import { KeyService } from './crypto';
 import { Event, Metadata } from './domain';
+import { KeyNotFoundError } from './errors';
 import { Config } from './eventstore.config';
 import { EVENTSTORE_SETTINGS_TOKEN } from './eventstore.constants';
-import { TransformerService } from './services';
+import { KeyService, TransformerService } from './services';
 
 @Injectable()
 export class EventStoreMapper {
+  private readonly logger = new Logger(EventStoreMapper.name);
+
   constructor(
     @Inject(EVENTSTORE_SETTINGS_TOKEN) private readonly config: Config,
     private readonly transformers: TransformerService,
@@ -17,39 +19,48 @@ export class EventStoreMapper {
 
   public async resolvedEventToDomainEvent(
     resolvedEvent: ResolvedEvent,
-  ): Promise<Event> | undefined {
+  ): Promise<Event> | null {
     if (
       resolvedEvent.event === undefined ||
       resolvedEvent.event.type.startsWith('$') ||
       !resolvedEvent.event.streamId.startsWith(this.config.category)
     ) {
-      return undefined;
+      return null;
     }
 
-    const metadata = resolvedEvent.event.metadata as Metadata;
-    let payload = resolvedEvent.event.data;
+    try {
+      const metadata = resolvedEvent.event.metadata as Metadata;
+      const payload = await this.extractPayload(resolvedEvent);
+      const transformer =
+        this.transformers.getTransformerToEvent(resolvedEvent);
 
-    if (metadata._aggregate_encrypted) {
-      payload = await this.keyService.decryptPayload(
-        metadata._aggregate_id,
-        metadata._encrypted_payload,
-      );
-    }
+      const event = transformer?.(
+        new Event(metadata._aggregate_id, payload),
+      ).withMetadata(metadata);
+
+      return event;
+    } catch (error) {
+      if (error instanceof KeyNotFoundError) {
+        this.logger.error(
+          `Error during decrypting ${resolvedEvent.event.type}: ${error.message}`,
+        );
 
-    const event = this.transformers.repo[resolvedEvent.event.type]?.(
-      new Event(metadata._aggregate_id, payload),
-    ).withMetadata(metadata);
+        return null;
+      }
 
-    return event;
+      throw error;
+    }
   }
 
-  public async resolvedEventsToDomainEvents(
-    resolvedEvents: ResolvedEvent[],
-  ): Promise<Event[]> {
-    return await Promise.all(
-      resolvedEvents
-        .map((resolvedEvent) => this.resolvedEventToDomainEvent(resolvedEvent))
-        .filter((event) => event !== undefined),
-    );
+  private async extractPayload(
+    resolvedEvent: ResolvedEvent,
+  ): Promise<JSONType | Uint8Array> {
+    const metadata = resolvedEvent.event.metadata as Metadata;
+    return metadata._aggregate_encrypted
+      ? await this.keyService.decryptPayload(
+          metadata._aggregate_id,
+          metadata._encrypted_payload,
+        )
+      : resolvedEvent.event.data;
   }
 }

src/eventstore.module.ts

@@ -3,14 +3,13 @@ import { CqrsModule, EventPublisher } from '@nestjs/cqrs';
 import { ConsoleModule } from 'nestjs-console';
 
 import { AggregateRepository } from './aggregate.repository';
-import { KeyService } from './crypto';
 import { AggregateRoot } from './domain';
 import { EventStore } from './eventstore';
 import { Config } from './eventstore.config';
 import { EventStoreCoreModule } from './eventstore-core.module';
 import { EventStoreModuleAsyncOptions, TransformerRepo } from './interfaces';
-import { EVENT_STORE_TRANSFORMERS_TOKEN } from './services/transformer.service';
-import { getRepositoryToken } from './utils/repository';
+import { EVENT_STORE_TRANSFORMERS_TOKEN, KeyService } from './services';
+import { getRepositoryToken } from './utils';
 
 @Module({})
 export class EventStoreModule {

src/eventstore.ts

@@ -14,11 +14,11 @@ import { IEventPublisher, IMessageSource } from '@nestjs/cqrs';
 import { Subject } from 'rxjs';
 import { v4 as uuid } from 'uuid';
 
-import { KeyService } from './crypto';
 import { AggregateRoot, Event } from './domain';
 import { Config } from './eventstore.config';
 import { EVENTSTORE_SETTINGS_TOKEN } from './eventstore.constants';
 import { EventStoreMapper } from './eventstore.mapper';
+import { KeyService } from './services';
 
 @Injectable()
 export class EventStore
@@ -84,12 +84,12 @@ export class EventStore
       entity.loadFromHistory(events);
 
       return entity;
-    } catch (err) {
-      if (err?.type === ErrorType.STREAM_NOT_FOUND) {
+    } catch (error) {
+      if (error?.type === ErrorType.STREAM_NOT_FOUND) {
         return null;
       }
 
-      this.logger.debug(err);
+      this.logger.debug(error);
     }
 
     return null;
@@ -111,8 +111,8 @@ export class EventStore
           resolveLinkTos: true,
         })
         .on('data', onEvent);
-    } catch (err) {
-      this.logger.debug(err);
+    } catch (error) {
+      this.logger.debug(error);
     }
   }
 }