Add handlerSpy support to AMQP (#63)

packages/amqp/lib/AbstractAmqpBaseConsumer.ts

@@ -118,6 +118,32 @@ export abstract class AbstractAmqpBaseConsumer<
     await this.consume()
   }
 
+  private tryToExtractId(message: Message | null): Either<'abort', string> {
+    if (message === null) {
+      return ABORT_EARLY_EITHER
+    }
+
+    const resolveMessageResult = this.resolveMessage(message)
+    if (isMessageError(resolveMessageResult.error)) {
+      this.handleError(resolveMessageResult.error)
+      return ABORT_EARLY_EITHER
+    }
+    // Empty content for whatever reason
+    if (!resolveMessageResult.result) {
+      return ABORT_EARLY_EITHER
+    }
+
+    // @ts-ignore
+    if (this.messageIdField in resolveMessageResult.result) {
+      return {
+        // @ts-ignore
+        result: resolveMessageResult.result[this.messageIdField],
+      }
+    }
+
+    return ABORT_EARLY_EITHER
+  }
+
   private async consume() {
     await this.channel.consume(this.queueName, (message) => {
       if (message === null) {
@@ -127,6 +153,8 @@ export abstract class AbstractAmqpBaseConsumer<
       const deserializedMessage = this.deserializeMessage(message)
       if (deserializedMessage.error === 'abort') {
         this.channel.nack(message, false, false)
+        const messageId = this.tryToExtractId(message)
+        this.handleMessageProcessed(null, 'invalid_message', messageId.result)
         return
       }
       // @ts-ignore
@@ -146,15 +174,18 @@ export abstract class AbstractAmqpBaseConsumer<
         .then((result) => {
           if (result.error === 'retryLater') {
             this.channel.nack(message, false, true)
+            this.handleMessageProcessed(deserializedMessage.result, 'retryLater')
           }
           if (result.result === 'success') {
             this.channel.ack(message)
+            this.handleMessageProcessed(deserializedMessage.result, 'consumed')
           }
         })
         .catch((err) => {
           // ToDo we need sanity check to stop trying at some point, perhaps some kind of Redis counter
           // If we fail due to unknown reason, let's retry
           this.channel.nack(message, false, true)
+          this.handleMessageProcessed(deserializedMessage.result, 'retryLater')
           this.handleError(err)
         })
         .finally(() => {

packages/amqp/lib/amqpMessageDeserializer.spec.ts

@@ -9,6 +9,7 @@ import { AmqpConsumerErrorResolver } from './errors/AmqpConsumerErrorResolver'
 describe('messageDeserializer', () => {
   it('deserializes valid JSON', () => {
     const messagePayload: PERMISSIONS_MESSAGE_TYPE = {
+      id: '1',
       messageType: 'add',
       userIds: [1],
       permissions: ['perm'],

packages/amqp/test/consumers/AmqpPermissionConsumerMultiSchema.ts

@@ -37,6 +37,7 @@ export class AmqpPermissionConsumerMultiSchema extends AbstractAmqpConsumerMulti
     super(
       dependencies,
       {
+        handlerSpy: true,
         creationConfig: {
           queueName: AmqpPermissionConsumerMultiSchema.QUEUE_NAME,
           queueOptions: {

packages/amqp/test/consumers/AmqpPermissionsConsumer.spec.ts

@@ -86,6 +86,7 @@ describe('PermissionsConsumer', () => {
       void channel.sendToQueue(
         AmqpPermissionConsumer.QUEUE_NAME,
         objectToBuffer({
+          id: '1',
           messageType: 'add',
           userIds,
           permissions: perms,
@@ -113,6 +114,7 @@ describe('PermissionsConsumer', () => {
       userPermissionMap[300] = []
 
       publisher.publish({
+        id: '2',
         messageType: 'add',
         userIds,
         permissions: perms,
@@ -125,6 +127,7 @@ describe('PermissionsConsumer', () => {
         }
 
         publisher.publish({
+          id: '3',
           messageType: 'add',
           userIds,
           permissions: perms,
@@ -150,6 +153,7 @@ describe('PermissionsConsumer', () => {
       channel.sendToQueue(
         AmqpPermissionConsumer.QUEUE_NAME,
         objectToBuffer({
+          id: '4',
           userIds,
           messageType: 'add',
           permissions: perms,
@@ -183,6 +187,7 @@ describe('PermissionsConsumer', () => {
       channel.sendToQueue(
         AmqpPermissionConsumer.QUEUE_NAME,
         objectToBuffer({
+          id: '5',
           userIds,
           messageType: 'add',
           permissions: perms,
@@ -212,6 +217,7 @@ describe('PermissionsConsumer', () => {
       channel.sendToQueue(
         AmqpPermissionConsumer.QUEUE_NAME,
         objectToBuffer({
+          id: '6',
           messageType: 'add',
           permissions: perms,
         } as PERMISSIONS_MESSAGE_TYPE),
@@ -231,7 +237,7 @@ describe('PermissionsConsumer', () => {
       const fakeResolver = consumerErrorResolver as FakeConsumerErrorResolver
       await waitAndRetry(() => fakeResolver.handleErrorCallsCount)
 
-      expect(fakeResolver.handleErrorCallsCount).toBe(1)
+      expect(fakeResolver.handleErrorCallsCount).toBe(2)
     })
   })
 })

packages/amqp/test/consumers/AmqpPermissionsConsumerMultiSchema.spec.ts

@@ -33,16 +33,21 @@ describe('PermissionsConsumerMultiSchema', () => {
       await newConsumer.start()
 
       publisher.publish({
+        id: '1',
         messageType: 'add',
       })
 
-      await waitAndRetry(() => {
-        return logger.loggedMessages.length === 3
-      })
+      await newConsumer.handlerSpy.waitForMessageWithId('1', 'consumed')
 
       expect(logger.loggedMessages.length).toBe(3)
-      expect(logger.loggedMessages[1]).toEqual({ messageType: 'add' })
-      expect(logger.loggedMessages[2]).toEqual({ messageType: 'add' })
+      expect(logger.loggedMessages[1]).toEqual({
+        id: '1',
+        messageType: 'add',
+      })
+      expect(logger.loggedMessages[2]).toEqual({
+        id: '1',
+        messageType: 'add',
+      })
     })
   })
 
@@ -76,12 +81,12 @@ describe('PermissionsConsumerMultiSchema', () => {
       await newConsumer.start()
 
       publisher.publish({
+        id: '3',
         messageType: 'add',
       })
 
-      await waitAndRetry(() => {
-        return newConsumer.addCounter === 1
-      })
+      await newConsumer.handlerSpy.waitForMessageWithId('3', 'retryLater')
+      await newConsumer.handlerSpy.waitForMessageWithId('3', 'consumed')
 
       expect(newConsumer.addCounter).toBe(1)
       expect(barrierCounter).toBe(2)
@@ -104,12 +109,12 @@ describe('PermissionsConsumerMultiSchema', () => {
       await newConsumer.start()
 
       publisher.publish({
+        id: '4',
         messageType: 'add',
       })
 
-      await waitAndRetry(() => {
-        return newConsumer.addCounter === 1
-      })
+      await newConsumer.handlerSpy.waitForMessageWithId('4', 'retryLater')
+      await newConsumer.handlerSpy.waitForMessageWithId('4', 'consumed')
 
       expect(newConsumer.addCounter).toBe(1)
       expect(barrierCounter).toBe(2)
@@ -138,18 +143,21 @@ describe('PermissionsConsumerMultiSchema', () => {
 
     it('Processes messages', async () => {
       publisher.publish({
+        id: '10',
         messageType: 'add',
       })
       publisher.publish({
+        id: '20',
         messageType: 'remove',
       })
       publisher.publish({
+        id: '30',
         messageType: 'remove',
       })
 
-      await waitAndRetry(() => {
-        return consumer.addCounter === 1 && consumer.removeCounter === 2
-      })
+      await consumer.handlerSpy.waitForMessageWithId('10', 'consumed')
+      await consumer.handlerSpy.waitForMessageWithId('20', 'consumed')
+      await consumer.handlerSpy.waitForMessageWithId('30', 'consumed')
 
       expect(consumer.addCounter).toBe(1)
       expect(consumer.removeCounter).toBe(2)
@@ -158,17 +166,21 @@ describe('PermissionsConsumerMultiSchema', () => {
     it('Reconnects if connection is lost', async () => {
       await (await diContainer.cradle.amqpConnectionManager.getConnection()).close()
       publisher.publish({
+        id: '100',
         messageType: 'add',
       })
 
       await waitAndRetry(() => {
         publisher.publish({
+          id: '200',
           messageType: 'add',
         })
 
         return consumer.addCounter > 0
       })
 
+      await consumer.handlerSpy.waitForMessageWithId('200', 'consumed')
+
       expect(consumer.addCounter > 0).toBe(true)
       await consumer.close()
     })

packages/amqp/test/consumers/userConsumerSchemas.ts

@@ -1,16 +1,19 @@
 import z from 'zod'
 
 export const PERMISSIONS_MESSAGE_SCHEMA = z.object({
+  id: z.string(),
   messageType: z.enum(['add', 'remove']),
   userIds: z.array(z.number()).describe('User IDs'),
   permissions: z.array(z.string()).nonempty().describe('List of user permissions'),
 })
 
 export const PERMISSIONS_ADD_MESSAGE_SCHEMA = z.object({
+  id: z.string(),
   messageType: z.literal('add'),
 })
 
 export const PERMISSIONS_REMOVE_MESSAGE_SCHEMA = z.object({
+  id: z.string(),
   messageType: z.literal('remove'),
 })
 

packages/amqp/test/publishers/AmqpPermissionPublisher.spec.ts

@@ -62,6 +62,7 @@ describe('PermissionPublisher', () => {
 
     it('logs a message when logging is enabled', async () => {
       const message = {
+        id: '1',
         userIds,
         messageType: 'add',
         permissions: perms,
@@ -74,6 +75,7 @@ describe('PermissionPublisher', () => {
       })
 
       expect(logger.loggedMessages[1]).toEqual({
+        id: '1',
         messageType: 'add',
         permissions: ['perm1', 'perm2'],
         userIds: [100, 200, 300],
@@ -161,6 +163,7 @@ describe('PermissionPublisher', () => {
       const { permissionPublisher } = diContainer.cradle
 
       const message = {
+        id: '2',
         userIds,
         messageType: 'add',
         permissions: perms,
@@ -186,6 +189,7 @@ describe('PermissionPublisher', () => {
       })
 
       expect(receivedMessage).toEqual({
+        id: '2',
         messageType: 'add',
         permissions: ['perm1', 'perm2'],
         userIds: [100, 200, 300],
@@ -204,6 +208,7 @@ describe('PermissionPublisher', () => {
       await permissionConsumer.start()
 
       const message = {
+        id: '3',
         userIds,
         messageType: 'add',
         permissions: perms,

packages/amqp/vitest.config.ts

@@ -14,7 +14,7 @@ export default defineConfig({
       all: true,
       lines: 90,
       functions: 100,
-      branches: 80,
+      branches: 79,
       statements: 90,
     },
   },