fix(message): enforce outbound queue capacity and horizon

.env.example

@@ -79,6 +79,9 @@ DATABASE_READ_THROUGH_TTL_SECONDS=3600
 # Optional JSON map, for example: {"fetchStatus":300,"groupMetadata":1800}
 DATABASE_READ_THROUGH_TTL_OVERRIDES={}
 
+# Maximum calculated delay between the current queue tail and a newly queued text.
+OUTBOUND_QUEUE_MAX_TAIL_GAP_SECONDS=120
+
 # PostgreSQL-only, audit-grade WhatsApp archive. The operational tables continue
 # to work independently. Generate the master key with: openssl rand -base64 32
 ARCHIVE_ENABLED=false

docs/outbound-automation-safety.md

@@ -70,7 +70,11 @@ The database is the durable source of truth. The worker runs inside Evolution AP
 
 Unlike the hash-only audit table, the queue must retain the complete `sendText` request and its integration-routing context so it can deliver it later with the same behavior. Successful and cleared rows are deleted immediately. Failed rows remain visible for diagnosis and are removed after the configured `audit.retentionDays` period during subsequent queue activity.
 
-Queue admission counts recent `SENT` audit rows together with pending/processing queue rows. A request returns `outbound_queue_instance_capacity` only when that total reaches `rateLimit.instancePerDay`, or `outbound_queue_recipient_capacity` when the recipient total reaches `rateLimit.recipientPerDay`. In-flight immediate sends remain protected by the ordinary safety reservation and concurrency checks.
+Queue admission counts recent `SENT` audit rows together with pending/processing queue rows. It rejects the new request when that combined total reaches any configured finite capacity: instance per minute/day, recipient per minute/day, or unique new/dormant outreach recipients per day. Pacing controls such as typing delay and `minimumIntervalMs` schedule work but do not reduce queue capacity. In-flight immediate sends remain protected by the ordinary safety reservation and concurrency checks.
+
+Capacity errors return HTTP `429` with a specific code (`outbound_queue_instance_minute_capacity`, `outbound_queue_instance_daily_capacity`, `outbound_queue_recipient_minute_capacity`, `outbound_queue_recipient_daily_capacity`, or `outbound_queue_outreach_daily_capacity`) and include `scope`, `window`, `limit`, `sent`, `queued`, `total`, and an expiry-based `retryAfterSeconds` value.
+
+The queue also has a bounded scheduling horizon. If the calculated send time for a new item is more than `OUTBOUND_QUEUE_MAX_TAIL_GAP_SECONDS` after the current queue tail, the API rejects it with HTTP `429` and code `outbound_queue_wait_too_long`. The response includes `queueTailAt`, `requestedSendAt`, `gapSeconds`, and `maxGapSeconds`. The default maximum gap is 120 seconds. This guard is not applied when the queue is empty.
 
 ### Inspect the queue
 

env.example

@@ -44,6 +44,9 @@ DATABASE_SAVE_IS_ON_WHATSAPP=true
 DATABASE_SAVE_IS_ON_WHATSAPP_DAYS=7
 DATABASE_DELETE_MESSAGE=false
 
+# Maximum calculated delay between the current queue tail and a newly queued text.
+OUTBOUND_QUEUE_MAX_TAIL_GAP_SECONDS=120
+
 # ===========================================
 # REDIS
 # ===========================================

package.json

@@ -1,6 +1,6 @@
 {
   "name": "evolution-api",
-  "version": "5.5.0",
+  "version": "5.5.1",
   "description": "Rest api for communication with WhatsApp",
   "main": "./dist/main.js",
   "type": "commonjs",

src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts

@@ -77,6 +77,7 @@ import { applyMistakesToMessage } from '@api/services/mistakes-generator.service
 import {
   OutboundMessageQueueService,
   OutboundQueueCapacityError,
+  OutboundQueueWaitTooLongError,
   QUEUEABLE_OUTBOUND_SAFETY_CODES,
 } from '@api/services/outbound-message-queue.service';
 import { OutboundSafetyService } from '@api/services/outbound-safety.service';
@@ -274,7 +275,10 @@ export class BaileysStartupService extends ChannelStartupService {
   ) {
     super(configService, eventEmitter, prismaRepository, chatwootCache);
     this.outboundSafety = new OutboundSafetyService(prismaRepository);
-    this.outboundMessageQueue = new OutboundMessageQueueService(prismaRepository);
+    this.outboundMessageQueue = new OutboundMessageQueueService(
+      prismaRepository,
+      this.configService.get('OUTBOUND_QUEUE').MAX_TAIL_GAP_SECONDS * 1000,
+    );
     this.settingsTemplates = new SettingsTemplateService(prismaRepository);
     this.instance.qrcode = { count: 0 };
     this.messageProcessor.mount({
@@ -3063,11 +3067,12 @@ export class BaileysStartupService extends ChannelStartupService {
         this.startOutboundQueueWorker();
         return queued;
       } catch (queueError) {
-        if (queueError instanceof OutboundQueueCapacityError) {
+        if (queueError instanceof OutboundQueueCapacityError || queueError instanceof OutboundQueueWaitTooLongError) {
           throw new TooManyRequestsException({
             code: queueError.code,
             retryAfterSeconds: queueError.retryAfterSeconds,
             message: queueError.message,
+            ...queueError.details,
           });
         }
         throw queueError;
@@ -3124,10 +3129,12 @@ export class BaileysStartupService extends ChannelStartupService {
     try {
       const queued = await this.outboundMessageQueue.claimNext(this.instanceId);
       if (!queued) return;
+      this.logger.info(`Processing outbound queue item ${queued.id} (attempt ${queued.attempts})`);
       try {
         const payload = queued.payload as unknown as { data?: SendTextDto; isIntegration?: boolean } & SendTextDto;
         await this.textMessage(payload.data ?? payload, payload.isIntegration ?? false, true);
         await this.outboundMessageQueue.complete(queued.id);
+        this.logger.info(`Delivered outbound queue item ${queued.id}`);
       } catch (error) {
         const safetyError = this.parseOutboundSafetyError(error);
         if (safetyError && QUEUEABLE_OUTBOUND_SAFETY_CODES.has(safetyError.code)) {
@@ -3137,8 +3144,12 @@ export class BaileysStartupService extends ChannelStartupService {
             safetyError.retryAfterSeconds,
             JSON.stringify(error),
           );
+          this.logger.info(
+            `Rescheduled outbound queue item ${queued.id} after ${safetyError.code} for ${safetyError.retryAfterSeconds}s`,
+          );
         } else {
           await this.outboundMessageQueue.fail(queued.id, error?.message ?? error?.toString() ?? 'Queue send failed');
+          this.logger.error([`Outbound queue item ${queued.id} failed`, error?.message ?? error]);
         }
       }
     } catch (error) {

src/api/services/outbound-message-queue.service.ts

@@ -35,15 +35,53 @@ export type OutboundQueueSnapshot = {
 
 export class OutboundQueueCapacityError extends Error {
   constructor(
-    public readonly code: 'outbound_queue_instance_capacity' | 'outbound_queue_recipient_capacity',
+    public readonly code:
+      | 'outbound_queue_instance_minute_capacity'
+      | 'outbound_queue_instance_daily_capacity'
+      | 'outbound_queue_recipient_minute_capacity'
+      | 'outbound_queue_recipient_daily_capacity'
+      | 'outbound_queue_outreach_daily_capacity',
     public readonly retryAfterSeconds: number,
+    public readonly details: {
+      scope: 'instance' | 'recipient' | 'outreach';
+      window: 'minute' | 'day';
+      limit: number;
+      sent: number;
+      queued: number;
+      total: number;
+    },
   ) {
-    super('Outbound queue plus recently sent messages reached the configured daily limit.');
+    super(
+      `Outbound queue capacity reached for ${details.scope} ${details.window} limit ` +
+        `(${details.sent} sent + ${details.queued} queued = ${details.total}; limit ${details.limit}).`,
+    );
+  }
+}
+
+export class OutboundQueueWaitTooLongError extends Error {
+  public readonly code = 'outbound_queue_wait_too_long';
+
+  constructor(
+    public readonly retryAfterSeconds: number,
+    public readonly details: {
+      queueTailAt: Date;
+      requestedSendAt: Date;
+      gapSeconds: number;
+      maxGapSeconds: number;
+    },
+  ) {
+    super(
+      `Outbound message was not queued because its send time would be ${details.gapSeconds}s after the queue tail; ` +
+        `the configured maximum is ${details.maxGapSeconds}s.`,
+    );
   }
 }
 
 export class OutboundMessageQueueService {
-  constructor(private readonly repository: PrismaRepository) {}
+  constructor(
+    private readonly repository: PrismaRepository,
+    private readonly maxTailGapMs = 120_000,
+  ) {}
 
   public async enqueueText(
     instanceId: string,
@@ -56,6 +94,7 @@ export class OutboundMessageQueueService {
   ) {
     const policy = normalizeAutomationSafety(rawPolicy);
     const now = new Date();
+    const minuteAgo = new Date(now.getTime() - 60_000);
     const dayAgo = new Date(now.getTime() - 86_400_000);
     const queuedStatuses = ['PENDING', 'PROCESSING'];
 
@@ -67,35 +106,127 @@ export class OutboundMessageQueueService {
       },
     });
 
-    const [recentInstanceMessages, recentRecipientMessages, queuedInstance, queuedRecipient, lastQueued] =
-      await Promise.all([
-        this.repository.outboundMessageAudit.count({
-          where: { instanceId, requestedAt: { gte: dayAgo }, status: 'SENT' },
-        }),
-        this.repository.outboundMessageAudit.count({
-          where: {
-            instanceId,
-            recipient,
-            requestedAt: { gte: dayAgo },
-            status: 'SENT',
-          },
-        }),
-        this.repository.outboundMessageQueue.count({ where: { instanceId, status: { in: queuedStatuses } } }),
-        this.repository.outboundMessageQueue.count({
-          where: { instanceId, recipient, status: { in: queuedStatuses } },
-        }),
-        this.repository.outboundMessageQueue.findFirst({
-          where: { instanceId, status: { in: queuedStatuses } },
-          orderBy: { scheduledAt: 'desc' },
-          select: { scheduledAt: true },
-        }),
-      ]);
-
-    if (recentInstanceMessages + queuedInstance >= policy.rateLimit.instancePerDay) {
-      throw new OutboundQueueCapacityError('outbound_queue_instance_capacity', 86_400);
-    }
-    if (recentRecipientMessages + queuedRecipient >= policy.rateLimit.recipientPerDay) {
-      throw new OutboundQueueCapacityError('outbound_queue_recipient_capacity', 86_400);
+    const [
+      instanceMinuteRows,
+      instanceDayRows,
+      recipientMinuteRows,
+      recipientDayRows,
+      queuedInstance,
+      queuedRecipient,
+      queuedOutreachRecipients,
+      recentOutreachRecipients,
+      lastQueued,
+    ] = await Promise.all([
+      this.repository.outboundMessageAudit.findMany({
+        where: { instanceId, requestedAt: { gte: minuteAgo }, status: 'SENT' },
+        orderBy: { requestedAt: 'asc' },
+        select: { requestedAt: true },
+      }),
+      this.repository.outboundMessageAudit.findMany({
+        where: { instanceId, requestedAt: { gte: dayAgo }, status: 'SENT' },
+        orderBy: { requestedAt: 'asc' },
+        select: { requestedAt: true },
+      }),
+      this.repository.outboundMessageAudit.findMany({
+        where: { instanceId, recipient, requestedAt: { gte: minuteAgo }, status: 'SENT' },
+        orderBy: { requestedAt: 'asc' },
+        select: { requestedAt: true },
+      }),
+      this.repository.outboundMessageAudit.findMany({
+        where: { instanceId, recipient, requestedAt: { gte: dayAgo }, status: 'SENT' },
+        orderBy: { requestedAt: 'asc' },
+        select: { requestedAt: true },
+      }),
+      this.repository.outboundMessageQueue.count({ where: { instanceId, status: { in: queuedStatuses } } }),
+      this.repository.outboundMessageQueue.count({
+        where: { instanceId, recipient, status: { in: queuedStatuses } },
+      }),
+      this.repository.outboundMessageQueue.findMany({
+        where: { instanceId, status: { in: queuedStatuses }, reason: 'outreach_recipient_limit' },
+        distinct: ['recipient'],
+        select: { recipient: true },
+      }),
+      this.repository.recipientEngagement.findMany({
+        where: { instanceId, lastOutreachAt: { gte: dayAgo } },
+        orderBy: { lastOutreachAt: 'asc' },
+        select: { recipient: true, lastOutreachAt: true },
+      }),
+      this.repository.outboundMessageQueue.findFirst({
+        where: { instanceId, status: { in: queuedStatuses } },
+        orderBy: { scheduledAt: 'desc' },
+        select: { scheduledAt: true },
+      }),
+    ]);
+
+    const capacity = (
+      code: OutboundQueueCapacityError['code'],
+      scope: OutboundQueueCapacityError['details']['scope'],
+      window: OutboundQueueCapacityError['details']['window'],
+      limit: number,
+      sentRows: Array<{ requestedAt?: Date; lastOutreachAt?: Date | null }>,
+      queued: number,
+    ) => {
+      const sent = sentRows.length;
+      if (sent + queued < limit) return;
+      const oldest = sentRows[0]?.requestedAt ?? sentRows[0]?.lastOutreachAt;
+      const windowMs = window === 'minute' ? 60_000 : 86_400_000;
+      const retryAfterSeconds = oldest
+        ? Math.max(1, Math.ceil((oldest.getTime() + windowMs - now.getTime()) / 1000))
+        : Math.ceil(windowMs / 1000);
+      throw new OutboundQueueCapacityError(code, retryAfterSeconds, {
+        scope,
+        window,
+        limit,
+        sent,
+        queued,
+        total: sent + queued,
+      });
+    };
+
+    capacity(
+      'outbound_queue_instance_minute_capacity',
+      'instance',
+      'minute',
+      policy.rateLimit.instancePerMinute,
+      instanceMinuteRows,
+      queuedInstance,
+    );
+    capacity(
+      'outbound_queue_instance_daily_capacity',
+      'instance',
+      'day',
+      policy.rateLimit.instancePerDay,
+      instanceDayRows,
+      queuedInstance,
+    );
+    capacity(
+      'outbound_queue_recipient_minute_capacity',
+      'recipient',
+      'minute',
+      policy.rateLimit.recipientPerMinute,
+      recipientMinuteRows,
+      queuedRecipient,
+    );
+    capacity(
+      'outbound_queue_recipient_daily_capacity',
+      'recipient',
+      'day',
+      policy.rateLimit.recipientPerDay,
+      recipientDayRows,
+      queuedRecipient,
+    );
+
+    if (reason === 'outreach_recipient_limit' && policy.outreach.enabled) {
+      const queuedRecipients = new Set(queuedOutreachRecipients.map((row) => row.recipient));
+      const queuedOutreach = queuedRecipients.has(recipient) ? queuedRecipients.size - 1 : queuedRecipients.size;
+      capacity(
+        'outbound_queue_outreach_daily_capacity',
+        'outreach',
+        'day',
+        policy.outreach.newOrDormantRecipientsPerDay,
+        recentOutreachRecipients,
+        queuedOutreach,
+      );
     }
 
     const requestedSchedule = now.getTime() + Math.max(1, retryAfterSeconds) * 1000;
@@ -104,6 +235,18 @@ export class OutboundMessageQueueService {
       : requestedSchedule;
     const scheduledAt = new Date(Math.max(requestedSchedule, afterExistingQueue));
 
+    if (lastQueued) {
+      const gapMs = scheduledAt.getTime() - lastQueued.scheduledAt.getTime();
+      if (gapMs > this.maxTailGapMs) {
+        throw new OutboundQueueWaitTooLongError(Math.max(1, Math.ceil(gapMs / 1000)), {
+          queueTailAt: lastQueued.scheduledAt,
+          requestedSendAt: scheduledAt,
+          gapSeconds: Math.ceil(gapMs / 1000),
+          maxGapSeconds: Math.floor(this.maxTailGapMs / 1000),
+        });
+      }
+    }
+
     const item = await this.repository.outboundMessageQueue.create({
       data: {
         instanceId,

src/config/env.config.ts

@@ -75,6 +75,10 @@ export type Archive = {
   S3_PREFIX: string;
 };
 
+export type OutboundQueue = {
+  MAX_TAIL_GAP_SECONDS: number;
+};
+
 export type DeleteData = {
   LOGICAL_MESSAGE_DELETE: boolean;
 };
@@ -452,6 +456,7 @@ export interface Env {
   PROVIDER: ProviderSession;
   DATABASE: Database;
   ARCHIVE: Archive;
+  OUTBOUND_QUEUE: OutboundQueue;
   RABBITMQ: Rabbitmq;
   NATS: Nats;
   SQS: Sqs;
@@ -605,6 +610,9 @@ export class ConfigService {
         CONFIRM_TTL_SECONDS: parseNonNegativeInteger(process.env?.ARCHIVE_CONFIRM_TTL_SECONDS, 900),
         S3_PREFIX: process.env?.ARCHIVE_S3_PREFIX || 'whatsapp-archive',
       },
+      OUTBOUND_QUEUE: {
+        MAX_TAIL_GAP_SECONDS: parseNonNegativeInteger(process.env?.OUTBOUND_QUEUE_MAX_TAIL_GAP_SECONDS, 120),
+      },
       RABBITMQ: {
         ENABLED: process.env?.RABBITMQ_ENABLED === 'true',
         GLOBAL_ENABLED: process.env?.RABBITMQ_GLOBAL_ENABLED === 'true',