fix: await processEvent before acking RabbitMQ message to restore backpressure

Unawaited func() caused channel.ack() to fire synchronously before any
HTTP requests were made, making AMQP_PREFETCH_MESSAGES completely
ineffective and creating unbounded concurrent connections under load.

Author: Uxio0

README.md

@@ -291,7 +291,7 @@ All configuration is done through environment variables. See `.env.sample` for a
 | `HTTP_TIMEOUT` | No | `1000` | Webhook HTTP client timeout in milliseconds |
 | `HTTP_MAX_REDIRECTS` | No | `0` | Max redirects followed when dispatching webhooks |
 | `DB_HEALTH_CHECK_TIMEOUT` | No | `5000` | Database health check timeout in milliseconds |
-| `AMQP_PREFETCH_MESSAGES` | No | `10` | RabbitMQ prefetch message count |
+| `AMQP_PREFETCH_MESSAGES` | No | `100` | RabbitMQ prefetch message count |
 | `WEBHOOK_AUTO_DISABLE` | No | `false` | Auto-disable webhooks that exceed the failure threshold |
 | `WEBHOOK_FAILURE_THRESHOLD` | No | `90` | Failure rate percentage (0–100) above which a webhook is auto-disabled |
 | `WEBHOOK_HEALTH_MINUTES_WINDOW` | No | `60` | Rolling window in minutes used to compute per-webhook failure rates |
@@ -316,4 +316,4 @@ This repository contains code developed under two different ownership and licens
 
 Users who require a purely MIT-licensed codebase should base their work on the `sef-mit-final` tag. The historical MIT-licensed code remains MIT and is not retroactively relicensed.
 
-For details, see `LICENSE` and `NOTICE`.
+For details, see `LICENSE` and `NOTICE`.

src/datasources/queue/queue.provider.ts

@@ -64,7 +64,7 @@ export class QueueProvider implements OnApplicationShutdown {
    *          at the same time
    */
   getPrefetchMessages(): number {
-    const value = Number(this.configService.get('AMQP_PREFETCH_MESSAGES', 10));
+    const value = Number(this.configService.get('AMQP_PREFETCH_MESSAGES', 100));
     this.logger.log(`AMQP_PREFETCH_MESSAGES=${value}`);
     return value;
   }
@@ -149,10 +149,15 @@ export class QueueProvider implements OnApplicationShutdown {
       );
       const consumer = await channel.consume(
         this.getQueueName(),
-        (message: ConsumeMessage) => {
+        async (message: ConsumeMessage) => {
           if (message.content) {
-            func(message.content.toString());
-            channel.ack(message);
+            try {
+              await func(message.content.toString());
+            } catch (error) {
+              this.logger.error(`Error processing message: ${error.message}`);
+            } finally {
+              channel.ack(message);
+            }
           }
         },
         {