Kafka publisher (#284)

* dep update

* Adding initial AbstractKafkaService

* Fixing types

* Exposing handler spy resolver

* Avoid abstact queue service

* Adding AbstractKafkaPublisher

* Adjusting test context + adding test schemas

* Publisher multi topic support

* Fixing init issue + adjusting test context

* Adding test publisher + fixing some issue on publish

* Adding publisher tests

* Improving publish error handling

* Improving test coverage

* Coverage improvements

* Narrowing down publish type

* Test updated

* Improving types

* Removing redundant config

* Improving coverage

* Removing unused type

* Lint fix

* Removing not needed override

* Moving topics config

* MessageSchemaContainer making messageTypeField optional

* Publisher messageTypeField can be undefined

* Adding missing tests

* Error fix

* Test fix

* Reverting message definition change

* Lint fix

Author: CarlosGamero

packages/core/lib/index.ts

@@ -5,6 +5,7 @@ export type {
   TransactionObservabilityManager,
   SchemaMap,
   ExtraParams,
+  MessageProcessingResult,
 } from './types/MessageQueueTypes.ts'
 
 export {
@@ -42,8 +43,13 @@ export type {
   Prehandler,
   PreHandlingOutputs,
 } from './queues/HandlerContainer.ts'
-export { HandlerSpy } from './queues/HandlerSpy.ts'
-export type { SpyResultInput, HandlerSpyParams, PublicHandlerSpy } from './queues/HandlerSpy.ts'
+export {
+  HandlerSpy,
+  resolveHandlerSpy,
+  type SpyResultInput,
+  type HandlerSpyParams,
+  type PublicHandlerSpy,
+} from './queues/HandlerSpy.ts'
 
 export { MessageSchemaContainer } from './queues/MessageSchemaContainer.ts'
 export type { MessageSchemaContainerOptions } from './queues/MessageSchemaContainer.ts'

packages/core/lib/queues/AbstractQueueService.ts

@@ -167,9 +167,9 @@ export abstract class AbstractQueueService<
       .filter((entry) => entry !== undefined)
 
     return new MessageSchemaContainer<MessagePayloadSchemas>({
+      messageTypeField: options.messageTypeField,
       messageSchemas,
       messageDefinitions,
-      messageTypeField: options.messageTypeField,
     })
   }
 
@@ -178,12 +178,11 @@ export abstract class AbstractQueueService<
     messageTypeField: string
   }) {
     const messageSchemas = options.messageSchemas
-    const messageDefinitions: readonly CommonEventDefinition[] = []
 
     return new MessageSchemaContainer<MessagePayloadSchemas>({
-      messageSchemas,
-      messageDefinitions,
       messageTypeField: options.messageTypeField,
+      messageSchemas,
+      messageDefinitions: [],
     })
   }
 

packages/core/lib/queues/MessageSchemaContainer.ts

@@ -5,15 +5,23 @@ import type { ZodSchema } from 'zod/v3'
 export type MessageSchemaContainerOptions<MessagePayloadSchemas extends object> = {
   messageDefinitions: readonly CommonEventDefinition[]
   messageSchemas: readonly ZodSchema<MessagePayloadSchemas>[]
-  messageTypeField: string
+  messageTypeField?: string
 }
 
+const NO_MESSAGE_TYPE = 'NO_MESSAGE_TYPE'
+
 export class MessageSchemaContainer<MessagePayloadSchemas extends object> {
   public readonly messageDefinitions: Record<string, CommonEventDefinition>
   private readonly messageSchemas: Record<string, ZodSchema<MessagePayloadSchemas>>
-  private readonly messageTypeField: string
+  private readonly messageTypeField?: string
 
   constructor(options: MessageSchemaContainerOptions<MessagePayloadSchemas>) {
+    if (options.messageTypeField === undefined && options.messageSchemas.length > 1) {
+      throw new Error(
+        'if messageTypeField is not provided, messageSchemas must have a single schema',
+      )
+    }
+
     this.messageTypeField = options.messageTypeField
     this.messageSchemas = this.resolveSchemaMap(options.messageSchemas)
     this.messageDefinitions = this.resolveDefinitionsMap(options.messageDefinitions ?? [])
@@ -23,20 +31,27 @@ export class MessageSchemaContainer<MessagePayloadSchemas extends object> {
     // biome-ignore lint/suspicious/noExplicitAny: This is expected
     message: Record<string, any>,
   ): Either<Error, ZodSchema<MessagePayloadSchemas>> {
+    if (!this.messageTypeField) {
+      return this.messageSchemas[NO_MESSAGE_TYPE]
+        ? { result: this.messageSchemas[NO_MESSAGE_TYPE] }
+        : { error: new Error('Unsupported message') }
+    }
+
     const schema = this.messageSchemas[message[this.messageTypeField]]
     if (!schema) {
-      return {
-        error: new Error(`Unsupported message type: ${message[this.messageTypeField]}`),
-      }
-    }
-    return {
-      result: schema,
+      return { error: new Error(`Unsupported message type: ${message[this.messageTypeField]}`) }
     }
+    return { result: schema }
   }
 
   private resolveSchemaMap(
     supportedSchemas: readonly ZodSchema<MessagePayloadSchemas>[],
   ): Record<string, ZodSchema<MessagePayloadSchemas>> {
+    if (!this.messageTypeField) {
+      if (!supportedSchemas[0]) return {}
+      return { [NO_MESSAGE_TYPE]: supportedSchemas[0] }
+    }
+
     return supportedSchemas.reduce(
       (acc, schema) => {
         // @ts-ignore
@@ -50,6 +65,11 @@ export class MessageSchemaContainer<MessagePayloadSchemas extends object> {
   private resolveDefinitionsMap(
     supportedDefinitions: readonly CommonEventDefinition[],
   ): Record<string, CommonEventDefinition> {
+    if (!this.messageTypeField) {
+      if (!supportedDefinitions[0]) return {}
+      return { [NO_MESSAGE_TYPE]: supportedDefinitions[0] }
+    }
+
     return supportedDefinitions.reduce(
       (acc, definition) => {
         // @ts-ignore

packages/core/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@message-queue-toolkit/core",
-    "version": "21.2.0",
+    "version": "21.2.1",
     "private": false,
     "license": "MIT",
     "description": "Useful utilities, interfaces and base classes for message queue handling. Supports AMQP and SQS with a common abstraction on top currently",

packages/kafka/lib/AbstractKafkaPublisher.ts

@@ -0,0 +1,115 @@
+import { InternalError, stringValueSerializer } from '@lokalise/node-core'
+import { MessageSchemaContainer } from '@message-queue-toolkit/core'
+import {
+  type MessageToProduce,
+  type ProduceOptions,
+  Producer,
+  jsonSerializer,
+  stringSerializer,
+} from '@platformatic/kafka'
+import { AbstractKafkaService, type BaseKafkaOptions } from './AbstractKafkaService.js'
+import type {
+  KafkaDependencies,
+  SupportedMessageValuesInput,
+  SupportedMessageValuesInputForTopic,
+  SupportedTopics,
+  TopicConfig,
+} from './types.js'
+
+export type KafkaPublisherOptions<TopicsConfig extends TopicConfig[]> = BaseKafkaOptions &
+  Omit<ProduceOptions<string, object, string, object>, 'serializers'> & {
+    topicsConfig: TopicsConfig
+  }
+
+export type KafkaMessageOptions = Omit<
+  MessageToProduce<string, object, string, object>,
+  'topic' | 'value'
+>
+
+export abstract class AbstractKafkaPublisher<
+  TopicsConfig extends TopicConfig[],
+> extends AbstractKafkaService<TopicsConfig, KafkaPublisherOptions<TopicsConfig>> {
+  private readonly topicsConfig: TopicsConfig
+  private readonly schemaContainers: Record<
+    string,
+    MessageSchemaContainer<SupportedMessageValuesInput<TopicsConfig>>
+  >
+
+  private producer?: Producer<string, object, string, object>
+
+  constructor(dependencies: KafkaDependencies, options: KafkaPublisherOptions<TopicsConfig>) {
+    super(dependencies, options)
+
+    this.topicsConfig = options.topicsConfig
+    if (this.topicsConfig.length === 0) throw new Error('At least one topic must be defined')
+
+    this.schemaContainers = {}
+    for (const { topic, schemas } of this.topicsConfig) {
+      this.schemaContainers[topic] = new MessageSchemaContainer({
+        messageSchemas: schemas,
+        messageTypeField: this.options.messageTypeField,
+        messageDefinitions: [],
+      })
+    }
+  }
+
+  init(): Promise<void> {
+    if (this.producer) return Promise.resolve()
+
+    this.producer = new Producer({
+      ...this.options.kafka,
+      ...this.options,
+      serializers: {
+        key: stringSerializer,
+        value: jsonSerializer,
+        headerKey: stringSerializer,
+        headerValue: jsonSerializer,
+      },
+    })
+
+    return Promise.resolve()
+  }
+
+  async close(): Promise<void> {
+    await this.producer?.close()
+    this.producer = undefined
+  }
+
+  async publish<Topic extends SupportedTopics<TopicsConfig>>(
+    topic: Topic,
+    message: SupportedMessageValuesInputForTopic<TopicsConfig, Topic>,
+    options?: KafkaMessageOptions,
+  ): Promise<void> {
+    const schemaResult = this.schemaContainers[topic]?.resolveSchema(message)
+    if (!schemaResult) throw new Error(`Message schemas not found for topic: ${topic}`)
+    if (schemaResult.error) throw schemaResult.error
+
+    await this.init() // lazy init
+
+    try {
+      const parsedMessage = schemaResult.result.parse(message)
+
+      // biome-ignore lint/style/noNonNullAssertion: Should always exist due to lazy init
+      await this.producer!.send({ messages: [{ ...options, topic, value: parsedMessage }] })
+
+      this.handleMessageProcessed({
+        message: parsedMessage,
+        processingResult: { status: 'published' },
+        topic,
+      })
+    } catch (error) {
+      const errorDetails = {
+        topic,
+        publisher: this.constructor.name,
+        message: stringValueSerializer(message),
+      }
+      this.handlerError(error, errorDetails)
+      throw new InternalError({
+        message: `Error while publishing to Kafka: ${(error as Error).message}`,
+        errorCode: 'KAFKA_PUBLISH_ERROR',
+        cause: error,
+        details: errorDetails,
+      })
+    }
+  }
+}

packages/kafka/lib/AbstractKafkaService.ts

@@ -0,0 +1,101 @@
+import { types } from 'node:util'
+import {
+  type CommonLogger,
+  type ErrorReporter,
+  resolveGlobalErrorLogObject,
+  stringValueSerializer,
+} from '@lokalise/node-core'
+import type { HandlerSpy, HandlerSpyParams } from '@message-queue-toolkit/core'
+import {
+  type MessageProcessingResult,
+  type PublicHandlerSpy,
+  resolveHandlerSpy,
+} from '@message-queue-toolkit/core'
+import type { BaseOptions } from '@platformatic/kafka'
+import type {
+  KafkaConfig,
+  KafkaDependencies,
+  SupportedMessageValues,
+  TopicConfig,
+} from './types.js'
+
+export type BaseKafkaOptions = {
+  kafka: KafkaConfig
+  messageTypeField?: string
+  messageIdField?: string
+  handlerSpy?: HandlerSpy<object> | HandlerSpyParams | boolean
+  logMessages?: boolean
+} & Omit<BaseOptions, keyof KafkaConfig> // Exclude properties that are already in KafkaConfig
+
+export abstract class AbstractKafkaService<
+  TopicsConfig extends TopicConfig[],
+  KafkaOptions extends BaseKafkaOptions,
+> {
+  protected readonly errorReporter: ErrorReporter
+  protected readonly logger: CommonLogger
+
+  protected readonly options: KafkaOptions
+  protected readonly _handlerSpy?: HandlerSpy<SupportedMessageValues<TopicsConfig>>
+
+  constructor(dependencies: KafkaDependencies, options: KafkaOptions) {
+    this.logger = dependencies.logger
+    this.errorReporter = dependencies.errorReporter
+    this.options = options
+
+    this._handlerSpy = resolveHandlerSpy(options)
+  }
+
+  abstract init(): Promise<void>
+  abstract close(): Promise<void>
+
+  get handlerSpy(): PublicHandlerSpy<SupportedMessageValues<TopicsConfig>> {
+    if (this._handlerSpy) return this._handlerSpy
+
+    throw new Error(
+      'HandlerSpy was not instantiated, please pass `handlerSpy` parameter during creation.',
+    )
+  }
+
+  protected resolveMessageType(
+    message: SupportedMessageValues<TopicsConfig> | null | undefined,
+  ): string | undefined {
+    if (!this.options.messageTypeField || !message) return undefined
+    return message[this.options.messageTypeField]
+  }
+
+  protected resolveMessageId(
+    message: SupportedMessageValues<TopicsConfig> | null | undefined,
+  ): string | undefined {
+    if (!this.options.messageIdField || !message) return undefined
+    return message[this.options.messageIdField]
+  }
+
+  protected handleMessageProcessed(params: {
+    message: SupportedMessageValues<TopicsConfig> | null
+    processingResult: MessageProcessingResult
+    topic: string
+  }) {
+    const { message, processingResult, topic } = params
+    const messageId = this.resolveMessageId(message)
+
+    this._handlerSpy?.addProcessedMessage({ message, processingResult }, messageId)
+
+    if (this.options.logMessages) {
+      this.logger.debug(
+        {
+          message: stringValueSerializer(message),
+          topic,
+          processingResult,
+          messageId,
+          messageType: this.resolveMessageType(message),
+        },
+        `Finished processing message ${messageId}`,
+      )
+    }
+  }
+
+  protected handlerError(error: unknown, context: Record<string, unknown> = {}): void {
+    this.logger.error({ ...resolveGlobalErrorLogObject(error), ...context })
+    if (types.isNativeError(error)) this.errorReporter.report({ error, context })
+  }
+}

packages/kafka/lib/index.ts

@@ -0,0 +1,2 @@
+export * from './types.ts'
+export * from './AbstractKafkaPublisher.js'

packages/kafka/lib/test-confluentic.spec.ts

@@ -2,13 +2,13 @@ import { randomUUID } from 'node:crypto'
 import { KafkaJS } from '@confluentinc/kafka-javascript'
 import { waitAndRetry } from '@lokalise/universal-ts-utils/node'
 import {} from '@platformatic/kafka'
-import { type TestContext, registerDependencies } from '../test/testContext.ts'
+import { type TestContext, createTestContext } from '../test/utils/testContext.ts'
 
 describe('Test confluentic-kafka', () => {
   let testContext: TestContext
 
   beforeAll(async () => {
-    testContext = await registerDependencies()
+    testContext = await createTestContext()
   })
 
   afterAll(async () => {
@@ -25,7 +25,7 @@ describe('Test confluentic-kafka', () => {
 
     const kafka = new KafkaJS.Kafka({
       'client.id': clientId,
-      'bootstrap.servers': testContext.cradle.kafkaConfig.brokers.join(','),
+      'bootstrap.servers': testContext.cradle.kafkaConfig.bootstrapBrokers.join(','),
     })
 
     // Topics can be created from producers, but as we will first connect a consumer, we need to create the topic first