From a05069fdc37e1b18c7bd30ba367306bbe3be4f6e Mon Sep 17 00:00:00 2001 From: menachem mendel kuperman Date: Wed, 23 Sep 2026 16:40:47 +0300 Subject: [PATCH] fix(message): distinguish queued delivery state --- docs/outbound-automation-safety.md | 8 +++ docs/runtime-routes.json | 28 ++++++++++ package-lock.json | 4 +- package.json | 2 +- scripts/runtime-openapi-overrides.mjs | 18 ++++++- .../whatsapp/whatsapp.baileys.service.ts | 20 +++++-- .../outbound-message-queue.service.ts | 54 +++++++++++++++++-- test/outbound-message-queue.postgres.test.ts | 6 ++- test/outbound-message-queue.test.ts | 12 ++++- 9 files changed, 137 insertions(+), 15 deletions(-) diff --git a/docs/outbound-automation-safety.md b/docs/outbound-automation-safety.md index 0065ab4b73..174572ab32 100644 --- a/docs/outbound-automation-safety.md +++ b/docs/outbound-automation-safety.md @@ -56,7 +56,11 @@ An accepted queued response looks like this: ```json { + "accepted": true, "queued": true, + "messageWasSent": false, + "deliveryStatus": "PENDING", + "final": false, "queueId": "cm...", "status": "PENDING", "reason": "outreach_recipient_limit", @@ -66,6 +70,8 @@ An accepted queued response looks like this: } ``` +HTTP acceptance is not delivery confirmation. Callers and n8n workflows must use `messageWasSent`, not HTTP success alone. A queued response always has `messageWasSent: false`, `deliveryStatus: "PENDING"`, and `final: false`. + The database is the durable source of truth. The worker runs inside Evolution API and therefore works with or without RabbitMQ. A restart or temporary WhatsApp disconnect does not discard pending rows; processing resumes after the instance reconnects. 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. @@ -76,6 +82,8 @@ Capacity errors return HTTP `429` with a specific code (`outbound_queue_instance 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. +Immediately before actual delivery, Evolution resolves the current settings template again and re-runs every safety rule. This includes changes caused by direct sends using an override template while the item was waiting. A temporary block is returned to `PENDING` only when its recalculated schedule remains inside the configured queue horizon. If the wait is too long, or a permanent rule now rejects the message, the item becomes `FAILED` with the policy code and diagnostic details; it is not treated as sent. + ### Inspect the queue `GET /message/outboundQueue/{instanceName}?limit=100` diff --git a/docs/runtime-routes.json b/docs/runtime-routes.json index f54e79aeb9..bec29a5110 100644 --- a/docs/runtime-routes.json +++ b/docs/runtime-routes.json @@ -61713,12 +61713,36 @@ "type": "object", "additionalProperties": false, "properties": { + "accepted": { + "type": "boolean", + "enum": [ + true + ] + }, "queued": { "type": "boolean", "enum": [ true ] }, + "messageWasSent": { + "type": "boolean", + "enum": [ + false + ] + }, + "deliveryStatus": { + "type": "string", + "enum": [ + "PENDING" + ] + }, + "final": { + "type": "boolean", + "enum": [ + false + ] + }, "queueId": { "type": "string" }, @@ -61743,7 +61767,11 @@ } }, "required": [ + "accepted", "queued", + "messageWasSent", + "deliveryStatus", + "final", "queueId", "status", "reason", diff --git a/package-lock.json b/package-lock.json index 294684d44b..93150c6cfd 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "evolution-api", - "version": "5.5.1", + "version": "5.5.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "evolution-api", - "version": "5.5.1", + "version": "5.5.2", "hasInstallScript": true, "license": "Apache-2.0", "dependencies": { diff --git a/package.json b/package.json index 40a9ef8bef..1be26d3f9b 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "evolution-api", - "version": "5.5.1", + "version": "5.5.2", "description": "Rest api for communication with WhatsApp", "main": "./dist/main.js", "type": "commonjs", diff --git a/scripts/runtime-openapi-overrides.mjs b/scripts/runtime-openapi-overrides.mjs index 94b8a83ead..fc65dc370a 100644 --- a/scripts/runtime-openapi-overrides.mjs +++ b/scripts/runtime-openapi-overrides.mjs @@ -144,7 +144,11 @@ export const overrides = { { type: 'object', additionalProperties: true, description: 'WhatsApp message result after immediate send.' }, bodySchema( { + accepted: { type: 'boolean', enum: [true] }, queued: { type: 'boolean', enum: [true] }, + messageWasSent: { type: 'boolean', enum: [false] }, + deliveryStatus: { type: 'string', enum: ['PENDING'] }, + final: { type: 'boolean', enum: [false] }, queueId: string, status: { type: 'string', enum: ['PENDING'] }, reason: string, @@ -152,7 +156,19 @@ export const overrides = { scheduledAt: { ...string, format: 'date-time' }, retryAfterSeconds: integer, }, - ['queued', 'queueId', 'status', 'reason', 'position', 'scheduledAt', 'retryAfterSeconds'], + [ + 'accepted', + 'queued', + 'messageWasSent', + 'deliveryStatus', + 'final', + 'queueId', + 'status', + 'reason', + 'position', + 'scheduledAt', + 'retryAfterSeconds', + ], ), ], }), diff --git a/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts b/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts index e4115f0642..f8ef6e6497 100644 --- a/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts +++ b/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts @@ -3138,17 +3138,27 @@ export class BaileysStartupService extends ChannelStartupService { } catch (error) { const safetyError = this.parseOutboundSafetyError(error); if (safetyError && QUEUEABLE_OUTBOUND_SAFETY_CODES.has(safetyError.code)) { - await this.outboundMessageQueue.reschedule( + const retry = await this.outboundMessageQueue.rescheduleOrReject( queued.id, safetyError.code, safetyError.retryAfterSeconds, JSON.stringify(error), ); - this.logger.info( - `Rescheduled outbound queue item ${queued.id} after ${safetyError.code} for ${safetyError.retryAfterSeconds}s`, - ); + if (retry.rescheduled) { + this.logger.info( + `Rescheduled outbound queue item ${queued.id} after ${safetyError.code} for ${safetyError.retryAfterSeconds}s`, + ); + } else { + this.logger.warn( + `Rejected outbound queue item ${queued.id} after policy recheck: ${retry.rejection?.message}`, + ); + } } else { - await this.outboundMessageQueue.fail(queued.id, error?.message ?? error?.toString() ?? 'Queue send failed'); + await this.outboundMessageQueue.fail( + queued.id, + error?.message ?? error?.toString() ?? 'Queue send failed', + safetyError?.code ?? 'delivery_failed', + ); this.logger.error([`Outbound queue item ${queued.id} failed`, error?.message ?? error]); } } diff --git a/src/api/services/outbound-message-queue.service.ts b/src/api/services/outbound-message-queue.service.ts index 3a2a77fddd..7054d9a743 100644 --- a/src/api/services/outbound-message-queue.service.ts +++ b/src/api/services/outbound-message-queue.service.ts @@ -267,7 +267,11 @@ export class OutboundMessageQueueService { })) || 1; return { + accepted: true, queued: true, + messageWasSent: false, + deliveryStatus: 'PENDING', + final: false, queueId: item.id, status: item.status, reason, @@ -362,23 +366,65 @@ export class OutboundMessageQueueService { await this.repository.outboundMessageQueue.deleteMany({ where: { id } }); } - public async reschedule(id: string, reason: string, retryAfterSeconds: number, lastError?: string): Promise { + public async rescheduleOrReject( + id: string, + reason: string, + retryAfterSeconds: number, + lastError?: string, + ): Promise<{ rescheduled: boolean; scheduledAt?: Date; rejection?: OutboundQueueWaitTooLongError }> { + const now = new Date(); + const requestedSendAt = new Date(now.getTime() + Math.max(1, retryAfterSeconds) * 1000); + const item = await this.repository.outboundMessageQueue.findUnique({ + where: { id }, + select: { instanceId: true }, + }); + if (!item) throw new Error(`Outbound queue item ${id} was not found.`); + const queueTail = await this.repository.outboundMessageQueue.findFirst({ + where: { instanceId: item.instanceId, status: 'PENDING' }, + orderBy: { scheduledAt: 'desc' }, + select: { scheduledAt: true }, + }); + const anchor = queueTail?.scheduledAt && queueTail.scheduledAt > now ? queueTail.scheduledAt : now; + const gapMs = requestedSendAt.getTime() - anchor.getTime(); + + if (gapMs > this.maxTailGapMs) { + const rejection = new OutboundQueueWaitTooLongError(Math.max(1, Math.ceil(gapMs / 1000)), { + queueTailAt: anchor, + requestedSendAt, + gapSeconds: Math.ceil(gapMs / 1000), + maxGapSeconds: Math.floor(this.maxTailGapMs / 1000), + }); + await this.fail( + id, + JSON.stringify({ + code: rejection.code, + message: rejection.message, + ...rejection.details, + policyReason: reason, + policyError: lastError, + }), + rejection.code, + ); + return { rescheduled: false, rejection }; + } + await this.repository.outboundMessageQueue.updateMany({ where: { id }, data: { status: 'PENDING', reason, - scheduledAt: new Date(Date.now() + Math.max(1, retryAfterSeconds) * 1000), + scheduledAt: requestedSendAt, lockedAt: null, lastError: lastError?.slice(0, 4000) ?? null, }, }); + return { rescheduled: true, scheduledAt: requestedSendAt }; } - public async fail(id: string, error: string): Promise { + public async fail(id: string, error: string, reason?: string): Promise { await this.repository.outboundMessageQueue.updateMany({ where: { id }, - data: { status: 'FAILED', lockedAt: null, lastError: error.slice(0, 4000) }, + data: { status: 'FAILED', reason, lockedAt: null, lastError: error.slice(0, 4000) }, }); } } diff --git a/test/outbound-message-queue.postgres.test.ts b/test/outbound-message-queue.postgres.test.ts index c93bcd0195..054401856f 100644 --- a/test/outbound-message-queue.postgres.test.ts +++ b/test/outbound-message-queue.postgres.test.ts @@ -22,6 +22,9 @@ async function main() { }, ); assert.equal(queued.queued, true); + assert.equal(queued.messageWasSent, false); + assert.equal(queued.deliveryStatus, 'PENDING'); + assert.equal(queued.final, false); const stored = await prisma.outboundMessageQueue.findUnique({ where: { id: queued.queueId } }); assert.deepEqual(stored?.payload, { @@ -38,7 +41,8 @@ async function main() { assert.equal(claimed?.status, 'PROCESSING'); assert.equal(claimed?.attempts, 1); - await service.reschedule(queued.queueId, 'minimum_interval', 30); + const retry = await service.rescheduleOrReject(queued.queueId, 'minimum_interval', 30); + assert.equal(retry.rescheduled, true); assert.equal((await service.snapshot(instance.id)).pendingCount, 1); const cleared = await service.clear(instance.id); diff --git a/test/outbound-message-queue.test.ts b/test/outbound-message-queue.test.ts index eb69846179..ddba4143e8 100644 --- a/test/outbound-message-queue.test.ts +++ b/test/outbound-message-queue.test.ts @@ -129,6 +129,10 @@ async function main() { { enabled: true, rateLimit: { instancePerDay: 10, recipientPerDay: 5, minimumIntervalMs: 1000 } }, ); assert.equal(first.queued, true); + assert.equal(first.accepted, true); + assert.equal(first.messageWasSent, false); + assert.equal(first.deliveryStatus, 'PENDING'); + assert.equal(first.final, false); assert.equal(first.position, 1); const second = await service.enqueueText( @@ -228,8 +232,14 @@ async function main() { error.details.gapSeconds > 120, ); + assert.equal(await service.claimNext('instance-1'), null, 'future work should not be claimed early'); + rows[0].scheduledAt = new Date(Date.now() - 1000); const claimed = await service.claimNext('instance-1'); - assert.equal(claimed, null, 'future work should not be claimed early'); + assert.equal(claimed?.id, first.queueId); + const rejectedRetry = await service.rescheduleOrReject(first.queueId, 'outreach_recipient_limit', 300); + assert.equal(rejectedRetry.rescheduled, false); + assert.equal(rows[0].status, 'FAILED'); + assert.equal(rows[0].reason, 'outbound_queue_wait_too_long'); const cleared = await service.clear('instance-1'); assert.equal(cleared.deletedCount, 2);