diff --git a/.env.example b/.env.example index 902958ace6..790ef04ba8 100644 --- a/.env.example +++ b/.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 diff --git a/docs/outbound-automation-safety.md b/docs/outbound-automation-safety.md index d1a6726fa2..0065ab4b73 100644 --- a/docs/outbound-automation-safety.md +++ b/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 diff --git a/env.example b/env.example index 5fe448b82c..0eaac2b5b7 100644 --- a/env.example +++ b/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 # =========================================== diff --git a/package-lock.json b/package-lock.json index c32bd88ff5..294684d44b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "evolution-api", - "version": "5.5.0", + "version": "5.5.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "evolution-api", - "version": "5.5.0", + "version": "5.5.1", "hasInstallScript": true, "license": "Apache-2.0", "dependencies": { diff --git a/package.json b/package.json index b0f46375de..40a9ef8bef 100644 --- a/package.json +++ b/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", diff --git a/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts b/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts index 957a1f649e..e4115f0642 100644 --- a/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts +++ b/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) { diff --git a/src/api/services/outbound-message-queue.service.ts b/src/api/services/outbound-message-queue.service.ts index 595634fef8..3a2a77fddd 100644 --- a/src/api/services/outbound-message-queue.service.ts +++ b/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, diff --git a/src/config/env.config.ts b/src/config/env.config.ts index a16f57484a..f9ef195ef1 100644 --- a/src/config/env.config.ts +++ b/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', diff --git a/test/outbound-message-queue.test.ts b/test/outbound-message-queue.test.ts index 77b58faa7b..eb69846179 100644 --- a/test/outbound-message-queue.test.ts +++ b/test/outbound-message-queue.test.ts @@ -3,6 +3,7 @@ import assert from 'node:assert/strict'; import { OutboundMessageQueueService, OutboundQueueCapacityError, + OutboundQueueWaitTooLongError, QUEUEABLE_OUTBOUND_SAFETY_CODES, } from '../src/api/services/outbound-message-queue.service'; @@ -10,9 +11,20 @@ async function main() { const rows: any[] = []; let recentInstanceMessages = 2; let recentRecipientMessages = 1; + let recentOutreachRecipients = 0; const repository = { outboundMessageAudit: { - count: async ({ where }: any) => (where.recipient ? recentRecipientMessages : recentInstanceMessages), + findMany: async ({ where }: any) => + Array.from({ length: where.recipient ? recentRecipientMessages : recentInstanceMessages }, (_, index) => ({ + requestedAt: new Date(Date.now() - 30_000 + index), + })), + }, + recipientEngagement: { + findMany: async () => + Array.from({ length: recentOutreachRecipients }, (_, index) => ({ + recipient: `outreach-${index}@s.whatsapp.net`, + lastOutreachAt: new Date(Date.now() - 30_000 + index), + })), }, outboundMessageQueue: { count: async ({ where }: any) => @@ -41,11 +53,22 @@ async function main() { : left.scheduledAt.getTime() - right.scheduledAt.getTime(), )[0] ?? null; }, - findMany: async ({ where, take }: any) => - rows + findMany: async ({ where, take, distinct, select }: any) => { + let matches = rows .filter((row) => where.status.in.includes(row.status)) - .sort((left, right) => left.scheduledAt.getTime() - right.scheduledAt.getTime()) - .slice(0, take), + .filter((row) => !where.reason || row.reason === where.reason) + .sort((left, right) => left.scheduledAt.getTime() - right.scheduledAt.getTime()); + if (distinct?.includes('recipient')) { + matches = matches.filter( + (row, index) => matches.findIndex((candidate) => candidate.recipient === row.recipient) === index, + ); + } + if (take) matches = matches.slice(0, take); + if (select?.recipient && Object.keys(select).length === 1) { + return matches.map((row) => ({ recipient: row.recipient })); + } + return matches; + }, create: async ({ data }: any) => { const row = { id: `queue-${rows.length + 1}`, @@ -136,11 +159,75 @@ async function main() { 60, { enabled: true, rateLimit: { instancePerDay: 10, recipientPerDay: 5 } }, ), - (error: any) => error instanceof OutboundQueueCapacityError && error.code === 'outbound_queue_instance_capacity', + (error: any) => + error instanceof OutboundQueueCapacityError && error.code === 'outbound_queue_instance_daily_capacity', ); recentInstanceMessages = 2; recentRecipientMessages = 1; + await assert.rejects( + () => + service.enqueueText( + 'instance-1', + '15550000000@s.whatsapp.net', + { number: '15550000000', text: 'Over minute capacity' }, + 'instance_rate_limit', + 30, + { + enabled: true, + rateLimit: { instancePerMinute: 2, instancePerDay: 100, recipientPerMinute: 10, recipientPerDay: 50 }, + }, + ), + (error: any) => + error instanceof OutboundQueueCapacityError && + error.code === 'outbound_queue_instance_minute_capacity' && + error.details.sent === 2 && + error.details.queued === 2 && + error.details.limit === 2, + ); + + recentInstanceMessages = 0; + recentRecipientMessages = 0; + recentOutreachRecipients = 3; + await assert.rejects( + () => + service.enqueueText( + 'instance-1', + '15551112222@s.whatsapp.net', + { number: '15551112222', text: 'Over outreach capacity' }, + 'outreach_recipient_limit', + 30, + { + enabled: true, + rateLimit: { instancePerMinute: 20, instancePerDay: 100, recipientPerMinute: 10, recipientPerDay: 50 }, + outreach: { enabled: true, newOrDormantRecipientsPerDay: 3 }, + }, + ), + (error: any) => + error instanceof OutboundQueueCapacityError && error.code === 'outbound_queue_outreach_daily_capacity', + ); + recentOutreachRecipients = 0; + + await assert.rejects( + () => + service.enqueueText( + 'instance-1', + '15553334444@s.whatsapp.net', + { number: '15553334444', text: 'Too far beyond the queue tail' }, + 'instance_rate_limit', + 300, + { + enabled: true, + rateLimit: { instancePerMinute: 20, instancePerDay: 100, recipientPerMinute: 10, recipientPerDay: 50 }, + }, + ), + (error: any) => + error instanceof OutboundQueueWaitTooLongError && + error.code === 'outbound_queue_wait_too_long' && + error.details.maxGapSeconds === 120 && + error.details.gapSeconds > 120, + ); + const claimed = await service.claimNext('instance-1'); assert.equal(claimed, null, 'future work should not be claimed early');