Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/outbound-automation-safety.md
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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.
Expand All @@ -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`
Expand Down
28 changes: 28 additions & 0 deletions docs/runtime-routes.json
Original file line number Diff line number Diff line change
Expand Up @@ -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"
},
Expand All @@ -61743,7 +61767,11 @@
}
},
"required": [
"accepted",
"queued",
"messageWasSent",
"deliveryStatus",
"final",
"queueId",
"status",
"reason",
Expand Down
4 changes: 2 additions & 2 deletions package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
@@ -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",
Expand Down
18 changes: 17 additions & 1 deletion scripts/runtime-openapi-overrides.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -144,15 +144,31 @@ 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,
position: integer,
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',
],
),
],
}),
Expand Down
20 changes: 15 additions & 5 deletions src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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]);
}
}
Expand Down
54 changes: 50 additions & 4 deletions src/api/services/outbound-message-queue.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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<void> {
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<void> {
public async fail(id: string, error: string, reason?: string): Promise<void> {
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) },
});
}
}
6 changes: 5 additions & 1 deletion test/outbound-message-queue.postgres.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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, {
Expand All @@ -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);
Expand Down
12 changes: 11 additions & 1 deletion test/outbound-message-queue.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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);
Expand Down
Loading