AP-4870 async event handlers support

packages/core/lib/events/DomainEventEmitter.ts

@@ -1,10 +1,11 @@
-import { InternalError } from '@lokalise/node-core'
+import { type ErrorReporter, InternalError } from '@lokalise/node-core'
 
 import type { MetadataFiller } from '../messages/MetadataFiller'
 import type { HandlerSpy, HandlerSpyParams, PublicHandlerSpy } from '../queues/HandlerSpy'
 import { resolveHandlerSpy } from '../queues/HandlerSpy'
 
 import type { ConsumerMessageMetadataType } from '@message-queue-toolkit/schemas'
+import type { Logger } from '../types/MessageQueueTypes'
 import type { EventRegistry } from './EventRegistry'
 import type {
   AnyEventHandler,
@@ -16,36 +17,51 @@ import type {
   SingleEventHandler,
 } from './eventTypes'
 
+export type DomainEventEmitterDependencies<SupportedEvents extends CommonEventDefinition[]> = {
+  eventRegistry: EventRegistry<SupportedEvents>
+  metadataFiller: MetadataFiller
+  // TODO: make them mandatory is a breaking change, decide if we are fine with that
+  logger?: Logger
+  errorReporter?: ErrorReporter
+}
+
+// TODO: not sure how to call it
+type Handlers<T> = {
+  sync: T[]
+  async: T[]
+}
+
 export class DomainEventEmitter<SupportedEvents extends CommonEventDefinition[]> {
   private readonly eventRegistry: EventRegistry<SupportedEvents>
+  private readonly metadataFiller: MetadataFiller
+  private readonly logger?: Logger
+  private readonly errorReporter?: ErrorReporter
+  private readonly _handlerSpy?: HandlerSpy<
+    CommonEventDefinitionConsumerSchemaType<SupportedEvents[number]>
+  >
 
   private readonly eventHandlerMap: Record<
     string,
-    EventHandler<CommonEventDefinitionPublisherSchemaType<SupportedEvents[number]>>[]
-  > = {}
-  private readonly anyHandlers: AnyEventHandler<SupportedEvents>[] = []
-  private readonly metadataFiller: MetadataFiller
-  private _handlerSpy:
-    | HandlerSpy<CommonEventDefinitionConsumerSchemaType<SupportedEvents[number]>>
-    | undefined
+    Handlers<EventHandler<CommonEventDefinitionPublisherSchemaType<SupportedEvents[number]>>>
+  >
+  private readonly anyHandlers: Handlers<AnyEventHandler<SupportedEvents>>
 
   constructor(
-    {
-      eventRegistry,
-      metadataFiller,
-    }: {
-      eventRegistry: EventRegistry<SupportedEvents>
-      metadataFiller: MetadataFiller
-    },
+    deps: DomainEventEmitterDependencies<SupportedEvents>,
     options: {
       handlerSpy?: HandlerSpy<object> | HandlerSpyParams | boolean
     } = {},
   ) {
-    this.eventRegistry = eventRegistry
-    this.metadataFiller = metadataFiller
+    this.eventRegistry = deps.eventRegistry
+    this.metadataFiller = deps.metadataFiller
+    this.logger = deps.logger
+    this.errorReporter = deps.errorReporter
 
     this._handlerSpy =
       resolveHandlerSpy<CommonEventDefinitionConsumerSchemaType<SupportedEvents[number]>>(options)
+
+    this.eventHandlerMap = {}
+    this.anyHandlers = { sync: [], async: [] }
   }
 
   get handlerSpy(): PublicHandlerSpy<
@@ -64,13 +80,16 @@ export class DomainEventEmitter<SupportedEvents extends CommonEventDefinition[]>
     data: Omit<CommonEventDefinitionPublisherSchemaType<SupportedEvent>, 'type'>,
     precedingMessageMetadata?: Partial<ConsumerMessageMetadataType>,
   ): Promise<Omit<CommonEventDefinitionConsumerSchemaType<SupportedEvent>, 'type'>> {
-    if (!data.timestamp) {
-      data.timestamp = this.metadataFiller.produceTimestamp()
-    }
-    if (!data.id) {
-      data.id = this.metadataFiller.produceId()
+    const eventTypeName = supportedEvent.publisherSchema.shape.type.value
+    if (!this.eventRegistry.isSupportedEvent(eventTypeName)) {
+      throw new InternalError({
+        errorCode: 'UNKNOWN_EVENT',
+        message: `Unknown event ${eventTypeName}`,
+      })
     }
 
+    if (!data.timestamp) data.timestamp = this.metadataFiller.produceTimestamp()
+    if (!data.id) data.id = this.metadataFiller.produceId()
     if (!data.metadata) {
       data.metadata = this.metadataFiller.produceMetadata(
         // @ts-ignore
@@ -79,52 +98,19 @@ export class DomainEventEmitter<SupportedEvents extends CommonEventDefinition[]>
         precedingMessageMetadata ?? {},
       )
     }
-
-    if (!data.metadata.correlationId) {
-      data.metadata.correlationId = this.metadataFiller.produceId()
-    }
-
-    const eventTypeName = supportedEvent.publisherSchema.shape.type.value
-
-    if (!this.eventRegistry.isSupportedEvent(eventTypeName)) {
-      throw new InternalError({
-        errorCode: 'UNKNOWN_EVENT',
-        message: `Unknown event ${eventTypeName}`,
-      })
-    }
-
-    const eventHandlers = this.eventHandlerMap[eventTypeName]
-
-    // No relevant handlers are registered, we can stop processing
-    if (!eventHandlers && this.anyHandlers.length === 0) {
-      // @ts-ignore
-      return data
-    }
+    if (!data.metadata.correlationId) data.metadata.correlationId = this.metadataFiller.produceId()
 
     const validatedEvent = this.eventRegistry
       .getEventDefinitionByTypeName(eventTypeName)
-      .publisherSchema.parse({
-        type: eventTypeName,
-        ...data,
-      })
+      .publisherSchema.parse({ type: eventTypeName, ...data })
 
-    if (eventHandlers) {
-      for (const handler of eventHandlers) {
-        await handler.handleEvent(validatedEvent)
-      }
-    }
-
-    for (const handler of this.anyHandlers) {
-      await handler.handleEvent(validatedEvent)
-    }
+    await this.handleEvent(validatedEvent)
 
     if (this._handlerSpy) {
       this._handlerSpy.addProcessedMessage(
         {
           // @ts-ignore
-          message: {
-            ...validatedEvent,
-          },
+          message: validatedEvent,
           processingResult: 'consumed',
         },
         validatedEvent.id,
@@ -141,8 +127,14 @@ export class DomainEventEmitter<SupportedEvents extends CommonEventDefinition[]>
   public on<EventTypeName extends EventTypeNames<SupportedEvents[number]>>(
     eventTypeName: EventTypeName,
     handler: SingleEventHandler<SupportedEvents, EventTypeName>,
+    asyncHandler = false,
   ) {
-    this.addOnHandler(eventTypeName, handler)
+    if (!this.eventHandlerMap[eventTypeName]) {
+      this.eventHandlerMap[eventTypeName] = { sync: [], async: [] }
+    }
+
+    if (asyncHandler) this.eventHandlerMap[eventTypeName].async.push(handler)
+    else this.eventHandlerMap[eventTypeName].sync.push(handler)
   }
 
   /**
@@ -151,27 +143,28 @@ export class DomainEventEmitter<SupportedEvents extends CommonEventDefinition[]>
   public onMany<EventTypeName extends EventTypeNames<SupportedEvents[number]>>(
     eventTypeNames: EventTypeName[],
     handler: SingleEventHandler<SupportedEvents, EventTypeName>,
+    asyncHandler = false,
   ) {
     for (const eventTypeName of eventTypeNames) {
-      this.on(eventTypeName, handler)
+      this.on(eventTypeName, handler, asyncHandler)
     }
   }
 
   /**
    * Register handler for all events supported by the emitter
    */
-  public onAny(handler: AnyEventHandler<SupportedEvents>) {
-    this.anyHandlers.push(handler)
+  public onAny(handler: AnyEventHandler<SupportedEvents>, asyncHandler = false) {
+    if (asyncHandler) this.anyHandlers.async.push(handler)
+    else this.anyHandlers.sync.push(handler)
   }
 
-  private addOnHandler(
-    eventTypeName: EventTypeNames<SupportedEvents[number]>,
-    handler: EventHandler,
-  ) {
-    if (!this.eventHandlerMap[eventTypeName]) {
-      this.eventHandlerMap[eventTypeName] = []
-    }
-
-    this.eventHandlerMap[eventTypeName].push(handler)
+  private async handleEvent<SupportedEvent extends SupportedEvents[number]>(
+    _event: CommonEventDefinitionPublisherSchemaType<SupportedEvent>,
+  ): Promise<void> {
+    // TODO: implement
+    //const handlers = [...(this.eventHandlerMap[event.type] ?? []), ...this.anyHandlers]
+    //for (const handler of handlers) {
+    //  await handler.handleEvent(event)
+    //}
   }
 }