EXP-585 Kafka metrics manager (#308)

* EXP-585 Using metrics manager

* Lint fix

* EXP-585 Improving id resolving logic

* Fixing tests

* EXP-585 Adjusting processed message

* EXP-585 Adding publisher tests

* EXP-585 Adding consumer tests

* EXP-585 Fixing timestamp issue

* release prepare

* Coverage fix

Author: CarlosGamero

packages/kafka/lib/AbstractKafkaConsumer.ts

@@ -89,6 +89,7 @@ export abstract class AbstractKafkaConsumer<
     })
 
     const logDetails = { origin: this.constructor.name, groupId: this.options.groupId }
+    /* v8 ignore start */
     this.consumer.on('consumer:group:join', (_) =>
       this.logger.debug(logDetails, 'Consumer is joining a group'),
     )
@@ -101,6 +102,7 @@ export abstract class AbstractKafkaConsumer<
     this.consumer.on('consumer:group:rebalance', (_) =>
       this.logger.debug(logDetails, 'Group is rebalancing'),
     )
+    /* v8 ignore stop */
   }
 
   async init(): Promise<void> {
@@ -135,6 +137,8 @@ export abstract class AbstractKafkaConsumer<
     // message.value can be undefined if the message is not JSON-serializable
     if (!message.value) return this.commitMessage(message)
 
+    const messageProcessingStartTimestamp = Date.now()
+
     const handler = this.handlerContainer.resolveHandler(message.topic, message.value)
     // if there is no handler for the message, we ignore it (simulating subscription)
     if (!handler) return this.commitMessage(message)
@@ -150,9 +154,9 @@ export abstract class AbstractKafkaConsumer<
         message: stringValueSerializer(message.value),
       })
       this.handleMessageProcessed({
-        topic: message.topic,
-        message: message.value,
+        message: message,
         processingResult: { status: 'error', errorReason: 'invalidMessage' },
+        messageProcessingStartTimestamp,
       })
 
       return this.commitMessage(message)
@@ -180,15 +184,15 @@ export abstract class AbstractKafkaConsumer<
 
     if (consumed) {
       this.handleMessageProcessed({
-        topic: message.topic,
-        message: validatedMessage,
+        message: message,
         processingResult: { status: 'consumed' },
+        messageProcessingStartTimestamp,
       })
     } else {
       this.handleMessageProcessed({
-        topic: message.topic,
-        message: validatedMessage,
+        message: message,
         processingResult: { status: 'error', errorReason: 'handlerError' },
+        messageProcessingStartTimestamp,
       })
     }
 
@@ -223,9 +227,7 @@ export abstract class AbstractKafkaConsumer<
       )
       await message.commit()
     } catch (error) {
-      if (error instanceof ResponseError) {
-        return this.handleResponseErrorOnCommit(error)
-      }
+      if (error instanceof ResponseError) return this.handleResponseErrorOnCommit(error)
       throw error
     }
   }

packages/kafka/lib/AbstractKafkaPublisher.ts

@@ -95,6 +95,8 @@ export abstract class AbstractKafkaPublisher<
     requestContext?: RequestContext,
     options?: KafkaMessageOptions,
   ): Promise<void> {
+    const messageProcessingStartTimestamp = Date.now()
+
     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
@@ -109,15 +111,15 @@ export abstract class AbstractKafkaPublisher<
         [this.resolveHeaderRequestIdField()]: requestContext?.reqId ?? '',
       }
 
-      // biome-ignore lint/style/noNonNullAssertion: Should always exist due to lazy init
-      await this.producer!.send({
-        messages: [{ ...options, topic, value: parsedMessage, headers }],
+      const kafkaMessage = { ...options, topic, value: parsedMessage, headers }
+      await this.producer?.send({
+        messages: [kafkaMessage],
       })
 
       this.handleMessageProcessed({
-        message: parsedMessage,
+        message: kafkaMessage,
         processingResult: { status: 'published' },
-        topic,
+        messageProcessingStartTimestamp,
       })
     } catch (error) {
       const errorDetails = {

packages/kafka/lib/AbstractKafkaService.ts

@@ -5,14 +5,16 @@ import {
   resolveGlobalErrorLogObject,
   stringValueSerializer,
 } from '@lokalise/node-core'
+import type { MakeRequired, MayOmit } from '@lokalise/universal-ts-utils/node'
 import {
   type HandlerSpy,
   type HandlerSpyParams,
+  type MessageMetricsManager,
   type MessageProcessingResult,
   type PublicHandlerSpy,
   resolveHandlerSpy,
 } from '@message-queue-toolkit/core'
-import type { BaseOptions } from '@platformatic/kafka'
+import type { BaseOptions, Message } from '@platformatic/kafka'
 import type {
   KafkaConfig,
   KafkaDependencies,
@@ -34,20 +36,29 @@ export type BaseKafkaOptions = {
   logMessages?: boolean
 } & Omit<BaseOptions, keyof KafkaConfig> // Exclude properties that are already in KafkaConfig
 
+type ProcessedMessage = MayOmit<
+  Pick<Message<string, object, string, string>, 'topic' | 'value' | 'timestamp'>,
+  'timestamp'
+>
+
 export abstract class AbstractKafkaService<
   TopicsConfig extends TopicConfig[],
   KafkaOptions extends BaseKafkaOptions,
 > {
   protected readonly errorReporter: ErrorReporter
   protected readonly logger: CommonLogger
+  protected readonly messageMetricsManager?: MessageMetricsManager<
+    SupportedMessageValues<TopicsConfig>
+  >
 
-  protected readonly options: KafkaOptions
+  protected readonly options: MakeRequired<KafkaOptions, 'messageIdField'>
   protected readonly _handlerSpy?: HandlerSpy<SupportedMessageValues<TopicsConfig>>
 
   constructor(dependencies: KafkaDependencies, options: KafkaOptions) {
     this.logger = dependencies.logger
     this.errorReporter = dependencies.errorReporter
-    this.options = options
+    this.messageMetricsManager = dependencies.messageMetricsManager
+    this.options = { ...options, messageIdField: options.messageIdField ?? 'id' }
 
     this._handlerSpy = resolveHandlerSpy(options)
   }
@@ -65,40 +76,54 @@ export abstract class AbstractKafkaService<
 
   protected resolveMessageType(message: SupportedMessageValues<TopicsConfig>): string | undefined {
     if (!this.options.messageTypeField) return undefined
-    return message[this.options.messageTypeField]
+    return message[this.options.messageTypeField] as string | undefined
   }
 
   protected resolveMessageId(message: SupportedMessageValues<TopicsConfig>): string | undefined {
-    if (!this.options.messageIdField) return undefined
-    return message[this.options.messageIdField]
+    // @ts-expect-error
+    return message[this.options.messageIdField] as string | undefined
   }
 
   protected resolveHeaderRequestIdField(): string {
     return this.options.headerRequestIdField ?? 'x-request-id'
   }
 
   protected handleMessageProcessed(params: {
-    message: SupportedMessageValues<TopicsConfig>
+    message: ProcessedMessage
     processingResult: MessageProcessingResult
-    topic: string
+    messageProcessingStartTimestamp: number
   }) {
-    const { message, processingResult, topic } = params
-    const messageId = this.resolveMessageId(message)
+    const { message, processingResult } = params
+    const messageId = this.resolveMessageId(message.value)
+    const messageType = this.resolveMessageType(message.value)
 
-    this._handlerSpy?.addProcessedMessage({ message, processingResult }, messageId)
+    this._handlerSpy?.addProcessedMessage({ message: message.value, processingResult }, messageId)
 
     if (this.options.logMessages) {
       this.logger.debug(
         {
-          message: stringValueSerializer(message),
-          topic,
+          message: stringValueSerializer(message.value),
+          topic: message.topic,
           processingResult,
           messageId,
-          messageType: this.resolveMessageType(message),
+          messageType,
         },
         `Finished processing message ${messageId}`,
       )
     }
+
+    if (this.messageMetricsManager) {
+      this.messageMetricsManager.registerProcessedMessage({
+        message: message.value,
+        processingResult,
+        queueName: message.topic,
+        messageId: messageId ?? 'unknown',
+        messageType: messageType ?? 'unknown',
+        messageTimestamp: message.timestamp ? Number(message.timestamp) : undefined,
+        messageProcessingStartTimestamp: params.messageProcessingStartTimestamp,
+        messageProcessingEndTimestamp: Date.now(),
+      })
+    }
   }
 
   protected handlerError(error: unknown, context: Record<string, unknown> = {}): void {

packages/kafka/lib/types.ts

@@ -3,7 +3,7 @@ import type { ConnectionOptions } from '@platformatic/kafka'
 import type { ZodSchema } from 'zod'
 import type z from 'zod/v3'
 
-export type KafkaDependencies = Omit<QueueDependencies, 'messageMetricsManager'>
+export type KafkaDependencies = QueueDependencies
 
 export type KafkaConfig = {
   bootstrapBrokers: string[]

packages/kafka/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@message-queue-toolkit/kafka",
-    "version": "0.2.3",
+    "version": "0.3.0",
     "engines": {
         "node": ">= 22.14.0"
     },