Add support for passing full event definitions

packages/amqp/README.md

@@ -2,8 +2,7 @@
 
 The library provides support for both direct exchanges and topic exchanges.
 
-> **_NOTE:_**  Check [README.md](../../README.md) for transport-agnostic library documentation.
->
+> **_NOTE:_** Check [README.md](../../README.md) for transport-agnostic library documentation.
 
 ## Publishers
 
@@ -25,42 +24,44 @@ Example:
 
 ```ts
 export const TEST_AMQP_CONFIG: AmqpConfig = {
-    vhost: '',
-    hostname: 'localhost',
-    username: 'guest',
-    password: 'guest',
-    port: 5672,
-    useTls: false,
+  vhost: '',
+  hostname: 'localhost',
+  username: 'guest',
+  password: 'guest',
+  port: 5672,
+  useTls: false,
 }
 
 const amqpConnectionManager = new AmqpConnectionManager(config, logger)
 await amqpConnectionManager.init()
 
 const publisher = new TestAmqpPublisher(
-    { amqpConnectionManager },
-    {
-        // other amqp options
-    })
+  { amqpConnectionManager },
+  {
+    // other amqp options
+  },
+)
 await publisher.init()
 
 const consumer = new TestAmqpConsumer(
-    { amqpConnectionManager },
-    {
-        // other amqp options
-    })
+  { amqpConnectionManager },
+  {
+    // other amqp options
+  },
+)
 await consumer.start()
 
 // break connection, to simulate unexpected disconnection in production
 await (await amqpConnectionManager.getConnection()).close()
 
-const message = { 
-    // some test message
+const message = {
+  // some test message
 }
 
 // This will fail, but will trigger reconnection within amqpConnectionManager
 publisher.publish(message)
-    
-// eventually connection is reestablished and propagated across all the AMQP services that use same amqpConnectionManager 
+
+// eventually connection is reestablished and propagated across all the AMQP services that use same amqpConnectionManager
 
 // This will succeed and consumer, which also received new connection, will be able to consume it
 publisher.publish(message)

packages/amqp/lib/AbstractAmqpConsumer.ts

@@ -8,13 +8,9 @@ import type {
   QueueConsumer,
   QueueConsumerOptions,
   TransactionObservabilityManager,
-} from '@message-queue-toolkit/core'
-import {
-  isMessageError,
-  parseMessage,
-  HandlerContainer,
   MessageSchemaContainer,
 } from '@message-queue-toolkit/core'
+import { isMessageError, parseMessage, HandlerContainer } from '@message-queue-toolkit/core'
 import type { Connection, Message } from 'amqplib'
 
 import type {
@@ -100,11 +96,8 @@ export abstract class AbstractAmqpConsumer<
       ? 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

@@ -4,11 +4,12 @@ import type {
   BarrierResult,
   CommonCreationConfigType,
   MessageInvalidFormatError,
+  MessageSchemaContainer,
   MessageValidationError,
   QueuePublisherOptions,
   SyncPublisher,
 } from '@message-queue-toolkit/core'
-import { objectToBuffer, MessageSchemaContainer } from '@message-queue-toolkit/core'
+import { objectToBuffer } from '@message-queue-toolkit/core'
 import type { ZodSchema } from 'zod'
 
 import type { AMQPDependencies } from './AbstractAmqpService'
@@ -49,11 +50,7 @@ export abstract class AbstractAmqpPublisher<
   ) {
     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> {