Add support for passing full event definitions

packages/amqp/lib/AbstractAmqpConsumer.ts

@@ -9,7 +9,7 @@
   QueueConsumerOptions,
   TransactionObservabilityManager,
 } from '@message-queue-toolkit/core'
 import {
   isMessageError,
   parseMessage,
   HandlerContainer,
@@ -100,11 +100,8 @@
       ? options.locatorConfig.queueName
       : options.creationConfig!.queueName
 
-    const messageSchemas = options.handlers.map((entry) => entry.schema)
-    this.messageSchemaContainer = new MessageSchemaContainer<MessagePayloadType>({
-      messageSchemas,
-      messageTypeField: options.messageTypeField,
-    })
+    this.messageSchemaContainer = this.resolveConsumerMessageSchemaContainer(options)
+
     this.handlerContainer = new HandlerContainer<
       MessagePayloadType,
       ExecutionContext,

packages/amqp/lib/AbstractAmqpPublisher.ts

@@ -8,7 +8,7 @@
   QueuePublisherOptions,
   SyncPublisher,
 } from '@message-queue-toolkit/core'
 import { objectToBuffer, MessageSchemaContainer } from '@message-queue-toolkit/core'
 import type { ZodSchema } from 'zod'
 
 import type { AMQPDependencies } from './AbstractAmqpService'
@@ -49,11 +49,7 @@
   ) {
     super(dependencies, options)
 
-    const messageSchemas = options.messageSchemas
-    this.messageSchemaContainer = new MessageSchemaContainer<MessagePayloadType>({
-      messageSchemas,
-      messageTypeField: options.messageTypeField,
-    })
+    this.messageSchemaContainer = this.resolvePublisherMessageSchemaContainer(options)
     this.exchange = options.exchange
   }
 

packages/core/lib/queues/AbstractQueueService.ts

@@ -2,6 +2,7 @@ import { types } from 'node:util'
 
 import type { ErrorReporter, ErrorResolver, Either } from '@lokalise/node-core'
 import { resolveGlobalErrorLogObject } from '@lokalise/node-core'
+import type { CommonEventDefinition } from '@message-queue-toolkit/schemas'
 import type { ZodSchema, ZodType } from 'zod'
 
 import type { MessageInvalidFormatError, MessageValidationError } from '../errors/Errors'
@@ -13,12 +14,14 @@ import { toDatePreprocessor } from '../utils/toDateProcessor'
 import type {
   BarrierCallback,
   BarrierResult,
+  MessageHandlerConfig,
   Prehandler,
   PrehandlerResult,
   PreHandlingOutputs,
 } from './HandlerContainer'
 import type { HandlerSpy, PublicHandlerSpy } from './HandlerSpy'
 import { resolveHandlerSpy } from './HandlerSpy'
+import { MessageSchemaContainer } from './MessageSchemaContainer'
 
 export type Deserializer<MessagePayloadType extends object> = (
   message: unknown,
@@ -90,6 +93,37 @@ export abstract class AbstractQueueService<
     this.isInitted = false
   }
 
+  protected resolveConsumerMessageSchemaContainer(options: {
+    handlers: MessageHandlerConfig<MessagePayloadSchemas, ExecutionContext, PrehandlerOutput>[]
+    messageTypeField: string
+  }) {
+    const messageSchemas = options.handlers.map((entry) => entry.schema)
+    // @ts-expect-error This should no longer be necessary in upcoming TypeScript updates, filter will narrow down the type
+    const messageDefinitions: CommonEventDefinition[] = options.handlers
+      .map((entry) => entry.definition)
+      .filter((entry) => entry !== undefined)
+
+    return new MessageSchemaContainer<MessagePayloadSchemas>({
+      messageSchemas,
+      messageDefinitions,
+      messageTypeField: options.messageTypeField,
+    })
+  }
+
+  protected resolvePublisherMessageSchemaContainer(options: {
+    messageSchemas: readonly ZodSchema<MessagePayloadSchemas>[]
+    messageTypeField: string
+  }) {
+    const messageSchemas = options.messageSchemas
+    const messageDefinitions: readonly CommonEventDefinition[] = []
+
+    return new MessageSchemaContainer<MessagePayloadSchemas>({
+      messageSchemas,
+      messageDefinitions,
+      messageTypeField: options.messageTypeField,
+    })
+  }
+
   protected abstract resolveSchema(
     message: MessagePayloadSchemas,
   ): Either<Error, ZodSchema<MessagePayloadSchemas>>
@@ -128,14 +162,10 @@ export abstract class AbstractQueueService<
 
   protected handleError(err: unknown, context?: Record<string, unknown>) {
     const logObject = resolveGlobalErrorLogObject(err)
-    if (logObject === 'string') {
-      this.logger.error(context, logObject)
-    } else if (typeof logObject === 'object') {
-      this.logger.error({
-        ...logObject,
-        ...context,
-      })
-    }
+    this.logger.error({
+      ...logObject,
+      ...context,
+    })
     if (types.isNativeError(err)) {
       this.errorReporter.report({ error: err, context })
     }

packages/core/lib/queues/HandlerContainer.ts

@@ -1,4 +1,6 @@
 import type { Either } from '@lokalise/node-core'
+import type { CommonEventDefinition } from '@message-queue-toolkit/schemas'
+import { isCommonEventDefinition } from '@message-queue-toolkit/schemas'
 import type { ZodSchema } from 'zod'
 
 import type { DoNotProcessMessageError } from '../errors/DoNotProcessError'
@@ -70,6 +72,7 @@ export class MessageHandlerConfig<
   const BarrierOutput = unknown,
 > {
   public readonly schema: ZodSchema<MessagePayloadSchema>
+  public readonly definition?: CommonEventDefinition
   public readonly handler: Handler<
     MessagePayloadSchema,
     ExecutionContext,
@@ -98,8 +101,10 @@ export class MessageHandlerConfig<
       PrehandlerOutput,
       BarrierOutput
     >,
+    eventDefinition?: CommonEventDefinition,
   ) {
     this.schema = schema
+    this.definition = eventDefinition
     this.handler = handler
     this.messageLogFormatter = options?.messageLogFormatter ?? defaultLogFormatter
     this.preHandlerBarrier = options?.preHandlerBarrier
@@ -125,7 +130,7 @@ export class MessageHandlerConfigBuilder<
   }
 
   addConfig<MessagePayloadSchema extends MessagePayloadSchemas, const BarrierOutput>(
-    schema: ZodSchema<MessagePayloadSchema>,
+    schema: ZodSchema<MessagePayloadSchema> | CommonEventDefinition,
     handler: Handler<MessagePayloadSchema, ExecutionContext, PrehandlerOutput, BarrierOutput>,
     options?: HandlerConfigOptions<
       MessagePayloadSchema,
@@ -134,6 +139,12 @@ export class MessageHandlerConfigBuilder<
       BarrierOutput
     >,
   ) {
+    const resolvedSchema: ZodSchema<MessagePayloadSchema> = isCommonEventDefinition(schema)
+      ? // @ts-ignore
+        (schema.consumerSchema as ZodSchema<MessagePayloadSchema>)
+      : schema
+    const definition = isCommonEventDefinition(schema) ? schema : undefined
+
     this.configs.push(
       // @ts-ignore
       new MessageHandlerConfig<
@@ -142,10 +153,11 @@ export class MessageHandlerConfigBuilder<
         PrehandlerOutput,
         BarrierOutput
       >(
-        schema,
+        resolvedSchema,
         // @ts-ignore
         handler,
         options,
+        definition,
       ),
     )
     return this
@@ -165,6 +177,7 @@ export type Handler<
   message: MessagePayloadSchemas,
   context: ExecutionContext,
   preHandlingOutputs: PreHandlingOutputs<PrehandlerOutput, BarrierOutput>,
+  definition?: CommonEventDefinition,
 ) => Promise<Either<'retryLater', 'success'>>
 
 export type HandlerContainerOptions<

packages/core/lib/queues/MessageSchemaContainer.ts

@@ -1,18 +1,22 @@
 import type { Either } from '@lokalise/node-core'
+import type { CommonEventDefinition } from '@message-queue-toolkit/schemas'
 import type { ZodSchema } from 'zod'
 
 export type MessageSchemaContainerOptions<MessagePayloadSchemas extends object> = {
+  messageDefinitions: readonly CommonEventDefinition[]
   messageSchemas: readonly ZodSchema<MessagePayloadSchemas>[]
   messageTypeField: string
 }
 
 export class MessageSchemaContainer<MessagePayloadSchemas extends object> {
+  public readonly messageDefinitions: Record<string, CommonEventDefinition>
   private readonly messageSchemas: Record<string, ZodSchema<MessagePayloadSchemas>>
   private readonly messageTypeField: string
 
   constructor(options: MessageSchemaContainerOptions<MessagePayloadSchemas>) {
     this.messageTypeField = options.messageTypeField
     this.messageSchemas = this.resolveSchemaMap(options.messageSchemas)
+    this.messageDefinitions = this.resolveDefinitionsMap(options.messageDefinitions ?? [])
   }
 
   public resolveSchema(
@@ -42,4 +46,17 @@ export class MessageSchemaContainer<MessagePayloadSchemas extends object> {
       {} as Record<string, ZodSchema<MessagePayloadSchemas>>,
     )
   }
+
+  private resolveDefinitionsMap(
+    supportedDefinitions: readonly CommonEventDefinition[],
+  ): Record<string, CommonEventDefinition> {
+    return supportedDefinitions.reduce(
+      (acc, definition) => {
+        // @ts-ignore
+        acc[definition.publisherSchema.shape[this.messageTypeField].value] = definition
+        return acc
+      },
+      {} as Record<string, CommonEventDefinition>,
+    )
+  }
 }

packages/core/package.json

@@ -25,7 +25,7 @@
     "prepublishOnly": "npm run build:release"
   },
   "dependencies": {
-    "@lokalise/node-core": "^9.17.0",
+    "@lokalise/node-core": "^9.21.0",
     "@message-queue-toolkit/schemas": "^1.0.0",
     "fast-equals": "^5.0.1",
     "toad-cache": "^3.7.0",

packages/schemas/lib/events/eventTypes.ts

@@ -8,6 +8,10 @@ import type { CONSUMER_BASE_EVENT_SCHEMA, PUBLISHER_BASE_EVENT_SCHEMA } from './
 export type EventTypeNames<EventDefinition extends CommonEventDefinition> =
   CommonEventDefinitionConsumerSchemaType<EventDefinition>['type']
 
+export function isCommonEventDefinition(entity: unknown): entity is CommonEventDefinition {
+  return (entity as CommonEventDefinition).publisherSchema !== undefined
+}
+
 export type CommonEventDefinition = {
   consumerSchema: ZodObject<
     Omit<(typeof CONSUMER_BASE_EVENT_SCHEMA)['shape'], 'payload'> & { payload: ZodTypeAny }
@@ -16,6 +20,13 @@ export type CommonEventDefinition = {
     Omit<(typeof PUBLISHER_BASE_EVENT_SCHEMA)['shape'], 'payload'> & { payload: ZodTypeAny }
   >
   schemaVersion?: string
+
+  //
+  // Metadata used for automated documentation generation
+  //
+  producedBy?: readonly string[] // Service ids for all the producers of this event.
+  domain?: string // Domain of the event
+  tags?: readonly string[] // Free-form tags for the event
 }
 
 export type CommonEventDefinitionConsumerSchemaType<T extends CommonEventDefinition> = z.infer<

packages/sns/lib/sns/AbstractSnsPublisher.ts

@@ -8,8 +8,8 @@ import type {
   MessageInvalidFormatError,
   MessageValidationError,
   QueuePublisherOptions,
+  MessageSchemaContainer,
 } from '@message-queue-toolkit/core'
-import { MessageSchemaContainer } from '@message-queue-toolkit/core'
 
 import type { SNSCreationConfig, SNSDependencies, SNSQueueLocatorType } from './AbstractSnsService'
 import { AbstractSnsService } from './AbstractSnsService'
@@ -36,11 +36,7 @@ export abstract class AbstractSnsPublisher<MessagePayloadType extends object>
   constructor(dependencies: SNSDependencies, options: SNSPublisherOptions<MessagePayloadType>) {
     super(dependencies, options)
 
-    const messageSchemas = options.messageSchemas
-    this.messageSchemaContainer = new MessageSchemaContainer<MessagePayloadType>({
-      messageSchemas,
-      messageTypeField: options.messageTypeField,
-    })
+    this.messageSchemaContainer = this.resolvePublisherMessageSchemaContainer(options)
   }
 
   async publish(message: MessagePayloadType, options: SNSMessageOptions = {}): Promise<void> {

packages/sns/lib/sns/SnsPublisherManager.spec.ts

@@ -4,6 +4,7 @@ import { enrichMessageSchemaWithBase } from '@message-queue-toolkit/core'
 import type { AwilixContainer } from 'awilix'
 import z from 'zod'
 
+import { SnsSqsEntityConsumer } from '../../test/consumers/SnsSqsEntityConsumer'
 import type {
   Dependencies,
   TestEventPublishPayloadsType,
@@ -13,7 +14,6 @@ import { registerDependencies, TestEvents } from '../../test/utils/testContext'
 
 import { CommonSnsPublisher } from './CommonSnsPublisherFactory'
 import type { SnsPublisherManager } from './SnsPublisherManager'
-import { FakeConsumer } from './fakes/FakeConsumer'
 
 describe('SnsPublisherManager', () => {
   let diContainer: AwilixContainer<Dependencies>
@@ -23,7 +23,7 @@ describe('SnsPublisherManager', () => {
   >
 
   beforeAll(async () => {
-    diContainer = await registerDependencies()
+    diContainer = await registerDependencies({}, false)
     publisherManager = diContainer.cradle.publisherManager
   })
 
@@ -34,13 +34,8 @@ describe('SnsPublisherManager', () => {
   describe('publish', () => {
     it('publishes to a correct publisher', async () => {
       // Given
-      const fakeConsumer = new FakeConsumer(
-        diContainer.cradle,
-        'queue',
-        TestEvents.created.snsTopic,
-        TestEvents.created.consumerSchema,
-      )
-      await fakeConsumer.start()
+      const consumer = new SnsSqsEntityConsumer(diContainer.cradle)
+      await consumer.start()
 
       // When
       const publishedMessage = await publisherManager.publish(TestEvents.created.snsTopic, {
@@ -54,7 +49,7 @@ describe('SnsPublisherManager', () => {
         .handlerSpy(TestEvents.created.snsTopic)
         .waitForMessageWithId(publishedMessage.id)
 
-      const consumerResult = await fakeConsumer.handlerSpy.waitForMessageWithId(publishedMessage.id)
+      const consumerResult = await consumer.handlerSpy.waitForMessageWithId(publishedMessage.id)
       const publishedMessageResult = await handlerSpyPromise
 
       expect(consumerResult.processingResult).toBe('consumed')
@@ -75,7 +70,7 @@ describe('SnsPublisherManager', () => {
         type: 'entity.created',
       })
 
-      await fakeConsumer.close()
+      await consumer.close()
     })
 
     it('message publishing is type-safe', async () => {