Autofill metadata on domain event emitter (#194)

kibertoad · CarlosGamero · web-flow · commit 807227afc5e3 · 2024-07-27T01:38:01.000+03:00
Co-authored-by: CarlosGamero &lt;carlos.gamero@lokalise.com&gt;
diff --git a/packages/core/lib/events/DomainEventEmitter.spec.ts b/packages/core/lib/events/DomainEventEmitter.spec.ts
@@ -82,6 +82,41 @@ describe('AutopilotEventEmitter', () => {
     expect(fakeListener.receivedEvents[0]).toMatchObject(expectedCreatedPayload)
   })
 
+  it('emits event to anyListener and populates metadata', async () => {
+    const { eventEmitter } = diContainer.cradle
+    const fakeListener = new FakeListener(diContainer.cradle.eventRegistry.supportedEvents)
+    eventEmitter.onAny(fakeListener)
+
+    const emittedEvent = await eventEmitter.emit(TestEvents.created, {
+      payload: {
+        message: 'msg',
+      },
+    })
+
+    const processedEvent = await eventEmitter.handlerSpy.waitForMessageWithId<
+      ConsumerMessageSchema<typeof TestEvents.created>
+    >(emittedEvent.id)
+
+    expect(processedEvent.message.type).toBe(TestEvents.created.consumerSchema.shape.type.value)
+
+    await waitAndRetry(() => {
+      return fakeListener.receivedEvents.length > 0
+    })
+
+    expect(fakeListener.receivedEvents).toHaveLength(1)
+    expect(fakeListener.receivedEvents[0]).toMatchObject({
+      id: expect.any(String),
+      payload: {
+        message: 'msg',
+      },
+      timestamp: expect.any(String),
+      type: 'entity.created',
+    })
+    expect(fakeListener.receivedEvents[0].metadata).toMatchObject({
+      correlationId: expect.any(String),
+    })
+  })
+
   it('can check spy for messages not being sent', async () => {
     const { eventEmitter } = diContainer.cradle
     const fakeListener = new FakeListener(diContainer.cradle.eventRegistry.supportedEvents)
@@ -127,7 +162,10 @@ describe('AutopilotEventEmitter', () => {
     expect(emitResult.message).toEqual({
       id: expect.any(String),
       metadata: {
-        correlationId: 'dummy',
+        correlationId: createdEventPayload.metadata!.correlationId!,
+        originatedFrom: 'service',
+        producedBy: undefined,
+        schemaVersion: '1',
       },
       payload: {
         message: 'msg',
@@ -150,10 +188,6 @@ describe('AutopilotEventEmitter', () => {
       timestamp: expect.any(String),
       type: 'entity.created',
     })
-    expect(fakeListener.receivedMetadata).toHaveLength(1)
-    expect(fakeListener.receivedMetadata[0]).toMatchObject({
-      correlationId: 'dummy',
-    })
   })
 
   it('emits event to singleListener', async () => {
diff --git a/packages/core/lib/events/DomainEventEmitter.ts b/packages/core/lib/events/DomainEventEmitter.ts
@@ -4,7 +4,7 @@ import type { MetadataFiller } from '../messages/MetadataFiller'
 import type { HandlerSpy, HandlerSpyParams, PublicHandlerSpy } from '../queues/HandlerSpy'
 import { resolveHandlerSpy } from '../queues/HandlerSpy'
 
-import type { PublisherMessageMetadataType } from '@message-queue-toolkit/schemas'
+import type { ConsumerMessageMetadataType } from '@message-queue-toolkit/schemas'
 import type { EventRegistry } from './EventRegistry'
 import type {
   AnyEventHandler,
@@ -62,7 +62,7 @@ export class DomainEventEmitter<SupportedEvents extends CommonEventDefinition[]>
   public async emit<SupportedEvent extends SupportedEvents[number]>(
     supportedEvent: SupportedEvent,
     data: Omit<CommonEventDefinitionPublisherSchemaType<SupportedEvent>, 'type'>,
-    metadata?: PublisherMessageMetadataType,
+    precedingMessageMetadata?: Partial<ConsumerMessageMetadataType>,
   ): Promise<Omit<CommonEventDefinitionConsumerSchemaType<SupportedEvent>, 'type'>> {
     if (!data.timestamp) {
       data.timestamp = this.metadataFiller.produceTimestamp()
@@ -71,6 +71,22 @@ export class DomainEventEmitter<SupportedEvents extends CommonEventDefinition[]>
       data.id = this.metadataFiller.produceId()
     }
 
+    if (!data.metadata) {
+      data.metadata = precedingMessageMetadata
+        ? // @ts-ignore
+          this.metadataFiller.produceMetadata(data, supportedEvent, precedingMessageMetadata)
+        : {
+            correlationId: this.metadataFiller.produceId(),
+            schemaVersion: supportedEvent.schemaVersion,
+            producedBy: this.metadataFiller.produceCurrentServiceId(),
+            originatedFrom: this.metadataFiller.produceCurrentServiceId(),
+          }
+    }
+
+    if (!data.metadata.correlationId) {
+      data.metadata.correlationId = this.metadataFiller.produceId()
+    }
+
     const eventTypeName = supportedEvent.publisherSchema.shape.type.value
 
     if (!this.eventRegistry.isSupportedEvent(eventTypeName)) {
@@ -97,12 +113,12 @@ export class DomainEventEmitter<SupportedEvents extends CommonEventDefinition[]>
 
     if (eventHandlers) {
       for (const handler of eventHandlers) {
-        await handler.handleEvent(validatedEvent, metadata)
+        await handler.handleEvent(validatedEvent)
       }
     }
 
     for (const handler of this.anyHandlers) {
-      await handler.handleEvent(validatedEvent, metadata)
+      await handler.handleEvent(validatedEvent)
     }
 
     if (this._handlerSpy) {
@@ -111,7 +127,6 @@ export class DomainEventEmitter<SupportedEvents extends CommonEventDefinition[]>
           // @ts-ignore
           message: {
             ...validatedEvent,
-            ...(metadata !== undefined ? { metadata } : {}),
           },
           processingResult: 'consumed',
         },
diff --git a/packages/core/lib/events/fakes/FakeListener.ts b/packages/core/lib/events/fakes/FakeListener.ts
@@ -1,21 +1,15 @@
-import type { PublisherMessageMetadataType } from '@message-queue-toolkit/schemas'
 import type { AnyEventHandler, CommonEventDefinition } from '../eventTypes'
 
 export class FakeListener<SupportedEvents extends CommonEventDefinition[]>
   implements AnyEventHandler<SupportedEvents>
 {
   public receivedEvents: SupportedEvents[number]['publisherSchema']['_output'][] = []
-  public receivedMetadata: PublisherMessageMetadataType[] = []
 
   constructor(_supportedEvents: SupportedEvents) {
     this.receivedEvents = []
   }
 
-  handleEvent(
-    event: SupportedEvents[number]['publisherSchema']['_output'],
-    metadata: PublisherMessageMetadataType,
-  ): void | Promise<void> {
+  handleEvent(event: SupportedEvents[number]['publisherSchema']['_output']): void | Promise<void> {
     this.receivedEvents.push(event)
-    this.receivedMetadata.push(metadata)
   }
 }
diff --git a/packages/core/lib/messages/MetadataFiller.spec.ts b/packages/core/lib/messages/MetadataFiller.spec.ts
@@ -53,7 +53,11 @@ describe('MetadataFiller', () => {
       const metadata = filler.produceMetadata(TEST_MESSAGE, EVENT_DEFINITION, providedMetadata)
 
       // Then
-      expect(metadata).toEqual({ ...providedMetadata, producedBy: SERVICE_ID })
+      expect(metadata).toEqual({
+        ...providedMetadata,
+        producedBy: SERVICE_ID,
+        schemaVersion: '0.0.0',
+      })
     })
   })
 })
diff --git a/packages/core/lib/messages/MetadataFiller.ts b/packages/core/lib/messages/MetadataFiller.ts
@@ -23,6 +23,7 @@ export type MetadataFiller<
   produceMetadata(currentMessage: T, eventDefinition: D, precedingMessageMetadata?: M): M
   produceId(): string
   produceTimestamp(): string
+  produceCurrentServiceId(): string
 }
 
 export class CommonMetadataFiller implements MetadataFiller {
@@ -46,6 +47,10 @@ export class CommonMetadataFiller implements MetadataFiller {
       })
   }
 
+  produceCurrentServiceId(): string {
+    return this.serviceId
+  }
+
   produceMetadata(
     _currentMessage: PublisherBaseEventType,
     eventDefinition: Pick<CommonEventDefinition, 'schemaVersion'>,
@@ -54,10 +59,7 @@ export class CommonMetadataFiller implements MetadataFiller {
     return {
       producedBy: this.serviceId,
       originatedFrom: precedingMessageMetadata?.originatedFrom ?? this.serviceId,
-      schemaVersion:
-        precedingMessageMetadata?.schemaVersion ??
-        eventDefinition.schemaVersion ??
-        this.defaultVersion,
+      schemaVersion: eventDefinition.schemaVersion ?? this.defaultVersion,
       correlationId: precedingMessageMetadata?.correlationId ?? this.produceId(),
     }
   }
diff --git a/packages/core/lib/messages/baseMessageSchemas.ts b/packages/core/lib/messages/baseMessageSchemas.ts
@@ -1,8 +1,6 @@
 export {
   PUBLISHER_MESSAGE_METADATA_SCHEMA,
   CONSUMER_MESSAGE_METADATA_SCHEMA,
-  PUBLISHER_MESSAGE_SCHEMA_EXTENSION,
-  CONSUMER_MESSAGE_SCHEMA_EXTENSION,
   PUBLISHER_BASE_MESSAGE_SCHEMA,
   CONSUMER_BASE_MESSAGE_SCHEMA,
   ConsumerBaseMessageType,
diff --git a/packages/schemas/lib/events/baseEventSchemas.ts b/packages/schemas/lib/events/baseEventSchemas.ts
@@ -1,16 +1,55 @@
 import type { ZodLiteral, ZodObject, ZodOptional, ZodString, ZodRawShape } from 'zod'
 import { z } from 'zod'
 
+// External message metadata that describe the context in which the message was created, primarily used for debugging purposes
+export const PUBLISHER_MESSAGE_METADATA_SCHEMA = z
+  .object({
+    schemaVersion: z.optional(z.string().min(1).describe('message schema version')),
+    // this is always set to a service that created the message
+    producedBy: z.optional(z.string().min(1).describe('app/service that produced the message')),
+    // this is always propagated within the message chain. For the first message in the chain it is equal to "producedBy"
+    originatedFrom: z.optional(
+      z
+        .string()
+        .min(1)
+        .describe('app/service that initiated entire workflow that led to creating this message'),
+    ),
+    // this is always propagated within the message chain.
+    correlationId: z.optional(
+      z.string().describe('unique identifier passed to all events in workflow chain'),
+    ),
+  })
+  .describe('external message metadata')
+
+export const CONSUMER_MESSAGE_METADATA_SCHEMA = z
+  .object({
+    schemaVersion: z.string().min(1).describe('message schema version'),
+    // this is always set to a service that created the message
+    producedBy: z.string().min(1).describe('app/service that produced the message'),
+    // this is always propagated within the message chain. For the first message in the chain it is equal to "producedBy"
+    originatedFrom: z
+      .string()
+      .min(1)
+      .describe('app/service that initiated entire workflow that led to creating this message'),
+    // this is always propagated within the message chain.
+    correlationId: z.string().describe('unique identifier passed to all events in workflow chain'),
+  })
+  .describe('external message metadata')
+
 // Base event fields that are typically autogenerated
 export const GENERATED_BASE_EVENT_SCHEMA = z.object({
   id: z.string().describe('event unique identifier'),
   timestamp: z.string().datetime().describe('iso 8601 datetime'),
+  // For internal domain events that did not originate within a message chain metadata field can be omitted, producer should then assume it is initiating a new chain
+  metadata: CONSUMER_MESSAGE_METADATA_SCHEMA,
 })
 
 // Base event fields that are typically autogenerated, marked as optional
 export const OPTIONAL_GENERATED_BASE_EVENT_SCHEMA = z.object({
   id: z.string().describe('event unique identifier').optional(),
   timestamp: z.string().datetime().describe('iso 8601 datetime').optional(),
+  // For internal domain events that did not originate within a message chain metadata field can be omitted, producer should then assume it is initiating a new chain
+  metadata: PUBLISHER_MESSAGE_METADATA_SCHEMA.optional(),
 })
 
 // Base event fields that are always defined manually
diff --git a/packages/schemas/lib/events/eventTypes.ts b/packages/schemas/lib/events/eventTypes.ts
@@ -1,8 +1,6 @@
 import type { ZodObject, ZodTypeAny } from 'zod'
 import type z from 'zod'
 
-import type { PublisherMessageMetadataType } from '../messages/baseMessageSchemas'
-
 import type { CONSUMER_BASE_EVENT_SCHEMA, PUBLISHER_BASE_EVENT_SCHEMA } from './baseEventSchemas'
 
 export type EventTypeNames<EventDefinition extends CommonEventDefinition> =
@@ -40,13 +38,8 @@ export type CommonEventDefinitionPublisherSchemaType<T extends CommonEventDefini
 export type EventHandler<
   EventDefinitionSchema extends
     CommonEventDefinitionPublisherSchemaType<CommonEventDefinition> = CommonEventDefinitionPublisherSchemaType<CommonEventDefinition>,
-  MetadataDefinitionSchema extends
-    Partial<PublisherMessageMetadataType> = Partial<PublisherMessageMetadataType>,
 > = {
-  handleEvent(
-    event: EventDefinitionSchema,
-    metadata?: MetadataDefinitionSchema,
-  ): void | Promise<void>
+  handleEvent(event: EventDefinitionSchema): void | Promise<void>
 }
 
 export type AnyEventHandler<EventDefinitions extends CommonEventDefinition[]> = EventHandler<
diff --git a/packages/schemas/lib/messages/baseMessageSchemas.ts b/packages/schemas/lib/messages/baseMessageSchemas.ts

Original file line number	Diff line number	Diff line change
`@@ -1,21 +1,15 @@`
`1`		`-import type { PublisherMessageMetadataType } from '@message-queue-toolkit/schemas'`
`2`	`1`	`import type { AnyEventHandler, CommonEventDefinition } from '../eventTypes'`
`3`	`2`
`4`	`3`	`export class FakeListener<SupportedEvents extends CommonEventDefinition[]>`
`5`	`4`	`implements AnyEventHandler<SupportedEvents>`
`6`	`5`	`{`
`7`	`6`	`public receivedEvents: SupportedEvents[number]['publisherSchema']['_output'][] = []`
`8`		`- public receivedMetadata: PublisherMessageMetadataType[] = []`
`9`	`7`
`10`	`8`	`constructor(_supportedEvents: SupportedEvents) {`
`11`	`9`	`this.receivedEvents = []`
`12`	`10`	`}`
`13`	`11`
`14`		`- handleEvent(`
`15`		`- event: SupportedEvents[number]['publisherSchema']['_output'],`
`16`		`- metadata: PublisherMessageMetadataType,`
`17`		`- ): void \| Promise<void> {`
	`12`	`+ handleEvent(event: SupportedEvents[number]['publisherSchema']['_output']): void \| Promise<void> {`
`18`	`13`	`this.receivedEvents.push(event)`
`19`		`- this.receivedMetadata.push(metadata)`
`20`	`14`	`}`
`21`	`15`	`}`