From 1d63f47000116beb28477a16f748d452e5d8bd11 Mon Sep 17 00:00:00 2001 From: menachem mendel kuperman Date: Wed, 23 Sep 2026 15:16:23 +0300 Subject: [PATCH 1/2] feat(message): queue transiently limited text sends --- .github/workflows/check_code_quality.yml | 3 + docs/README.md | 2 +- docs/outbound-automation-safety.md | 83 ++- docs/runtime-routes.json | 690 +++++++++++++++++- package-lock.json | 4 +- package.json | 4 +- .../migration.sql | 19 + prisma/mysql-schema.prisma | 21 + .../migration.sql | 22 + prisma/postgresql-schema.prisma | 21 + prisma/psql_bouncer-schema.prisma | 21 + scripts/runtime-openapi-overrides.mjs | 88 ++- src/api/controllers/sendMessage.controller.ts | 8 + .../whatsapp/whatsapp.baileys.service.ts | 158 +++- src/api/routes/sendMessage.router.ts | 21 + .../outbound-message-queue.service.ts | 241 ++++++ src/api/services/outbound-safety.service.ts | 50 +- test/outbound-message-queue.postgres.test.ts | 55 ++ test/outbound-message-queue.test.ts | 154 ++++ 19 files changed, 1625 insertions(+), 40 deletions(-) create mode 100644 prisma/mysql-migrations/20260923010000_outbound_message_queue/migration.sql create mode 100644 prisma/postgresql-migrations/20260923010000_outbound_message_queue/migration.sql create mode 100644 src/api/services/outbound-message-queue.service.ts create mode 100644 test/outbound-message-queue.postgres.test.ts create mode 100644 test/outbound-message-queue.test.ts diff --git a/.github/workflows/check_code_quality.yml b/.github/workflows/check_code_quality.yml index a73a4101d6..969c8e168f 100644 --- a/.github/workflows/check_code_quality.yml +++ b/.github/workflows/check_code_quality.yml @@ -78,6 +78,9 @@ jobs: - name: Apply PostgreSQL migrations run: npm run db:deploy + - name: Test durable outbound queue + run: npm run test:outbound-queue && npm run test:outbound-queue:postgres + - name: Test archive lifecycle run: npm run test:archive diff --git a/docs/README.md b/docs/README.md index cd98be38ee..82916e91b4 100644 --- a/docs/README.md +++ b/docs/README.md @@ -61,7 +61,7 @@ Eligible reads use a persistent, instance-scoped database snapshot before contac See the [local-first read-through reference](./local-first-reads.md) and [complete instance-settings reference](./settings.md) for the method matrix, every setting, identity rules, invalidation behavior, migration sequence, examples, and rollback guidance. -Automated outbound sends can use the per-instance [automation safety and pacing policy](./outbound-automation-safety.md). It provides content-preserving typing indicators, rolling minute/daily limits, quiet hours, suppression and optional recipient allowlisting, configurable exact/near-duplicate blocking, failure circuits, and persistent audit records. It never rewrites message text or attempts to conceal automation. +Automated outbound sends can use the per-instance [automation safety and pacing policy](./outbound-automation-safety.md). It provides content-preserving typing indicators, rolling minute/daily limits, quiet hours, suppression and optional recipient allowlisting, configurable exact/near-duplicate blocking, failure circuits, persistent audit records, and a durable cooldown queue for text messages. It never rewrites message text or attempts to conceal automation. Username-only contacts and server-issued LIDs are documented in [WhatsApp username and LID identities](./username-lid-identities.md), including send/moderation examples, validation codes, compatibility, and test commands. diff --git a/docs/outbound-automation-safety.md b/docs/outbound-automation-safety.md index 55afbe7103..d1a6726fa2 100644 --- a/docs/outbound-automation-safety.md +++ b/docs/outbound-automation-safety.md @@ -15,8 +15,11 @@ This feature is for traffic safety, recipient protection, and predictable user e 7. Calculate a bounded typing-indicator duration from visible text or caption length. A request's explicit `delay` remains authoritative when it is longer. 8. Send the message without changing its content. 9. Mark the audit row `SENT` or `FAILED`. +10. For `sendText`, persist transiently blocked work in the internal queue instead of requiring the caller to retry. -Rejected requests return HTTP `429` with a machine-readable policy code and, when meaningful, `retryAfterSeconds`. Quiet-hour requests are rejected rather than silently held, so callers retain control of scheduling and idempotency. +For `POST /message/sendText/{instanceName}`, transient cooldown decisions return a successful queued response. The queue stores the original text request, survives restarts, and re-runs the current safety policy immediately before delivery. Other send methods continue to return HTTP `429` when blocked. + +Permanent decisions still return HTTP `429`. These include invalid destinations, suppression, allowlist rejection, quiet hours, and deliberate duplicate blocking. The API never queues a request that policy says must not be sent. ```json { @@ -36,6 +39,82 @@ Rejected requests return HTTP `429` with a machine-readable policy code and, whe `retryAfterSeconds` is omitted for permanent decisions such as suppression or a missing allowlist entry. The API does not return a `Retry-After` HTTP header; automation should read the JSON field. +## Internal text-message queue + +The queue is used for these transient block codes: + +- `concurrency_limit` +- `failure_circuit_open` +- `instance_rate_limit` +- `instance_daily_limit` +- `recipient_rate_limit` +- `recipient_daily_limit` +- `outreach_recipient_limit` +- `minimum_interval` + +An accepted queued response looks like this: + +```json +{ + "queued": true, + "queueId": "cm...", + "status": "PENDING", + "reason": "outreach_recipient_limit", + "position": 1, + "scheduledAt": "2026-09-24T11:00:00.000Z", + "retryAfterSeconds": 86400 +} +``` + +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. + +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. + +### Inspect the queue + +`GET /message/outboundQueue/{instanceName}?limit=100` + +```json +{ + "pendingCount": 2, + "processingCount": 0, + "failedCount": 0, + "nextSendAt": "2026-09-24T11:00:00.000Z", + "estimatedEmptyAt": "2026-09-24T11:00:01.000Z", + "items": [ + { + "id": "cm...", + "recipient": "15551234567@s.whatsapp.net", + "status": "PENDING", + "reason": "outreach_recipient_limit", + "attempts": 0, + "requestedAt": "2026-09-23T11:00:00.000Z", + "scheduledAt": "2026-09-24T11:00:00.000Z", + "lastError": null + } + ] +} +``` + +`nextSendAt` and `estimatedEmptyAt` are estimates. They can move when policy settings, recipient history, failures, or additional queued work change. + +### Clear the queue + +`DELETE /message/outboundQueue/{instanceName}` + +```json +{ + "cleared": true, + "deletedCount": 2, + "processingAtClearTime": 0, + "note": "All queued messages were deleted before delivery." +} +``` + +Clear removes pending and failed entries immediately. If `processingAtClearTime` is non-zero, a request already handed to WhatsApp may complete and cannot be recalled. + Use `GET /settings/outbound-audit/{instanceName}?limit=100&status=BLOCKED&recipient=...` to inspect the newest audit rows. `limit` is capped at 500 and filters are optional. ## Complete settings structure @@ -154,6 +233,6 @@ Policy block codes are `recipient_suppressed`, `recipient_not_allowed`, `quiet_h - Rolling limits and duplicate/failure checks use PostgreSQL or MySQL audit rows and therefore survive restarts. The concurrency count is process-local; use conservative rolling limits when running multiple API replicas. - Similarity comparison normalizes Unicode compatibility forms, letter case, and repeated whitespace, then compares a 64-bit character-trigram SimHash. The displayed percentage is an approximate fingerprint similarity, not an edit-distance guarantee. Exact SHA-256 equality is always blocked regardless of the configured percentage. - Audit rows created before the similarity migration have no similarity fingerprint. They still participate in exact matching, but near-duplicate matching starts with sends recorded after the upgrade. -- Automatic retries are intentionally not performed after an uncertain WhatsApp send because retrying without a confirmed idempotency key can duplicate a message. Workflow callers should retry only clearly rejected pre-send requests. +- The internal queue retries only messages rejected before WhatsApp delivery by a queueable safety decision. It does not automatically retry an uncertain transport send, because doing so without a confirmed idempotency key can duplicate a message. - The settings migration is automatic during normal Evolution API startup. Back up the database before every application upgrade as usual. - See [Defensive detection of automation disguise](./defensive-automation-detection.md) for safe teaching examples and detection guidance. diff --git a/docs/runtime-routes.json b/docs/runtime-routes.json index 8bceaf318a..f54e79aeb9 100644 --- a/docs/runtime-routes.json +++ b/docs/runtime-routes.json @@ -54671,6 +54671,638 @@ "x-source": "src/api/routes/label.router.ts" } }, + { + "method": "delete", + "path": "/message/outboundQueue/{instanceName}", + "source": "src/api/routes/sendMessage.router.ts", + "schemaName": "null", + "multipart": false, + "runtimeStatus": "OK", + "operation": { + "tags": [ + "Messages" + ], + "summary": "Clear the outbound text-message queue", + "description": "Deletes queued messages that have not been delivered. A message already handed to WhatsApp cannot be recalled.", + "operationId": "delete_message_outboundQueue", + "parameters": [ + { + "name": "instanceName", + "in": "path", + "required": true, + "description": "Evolution API instance name.", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Queue clear result.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": false, + "properties": { + "cleared": { + "type": "boolean" + }, + "deletedCount": { + "type": "integer" + }, + "processingAtClearTime": { + "type": "integer" + }, + "note": { + "type": "string" + } + }, + "required": [ + "cleared", + "deletedCount", + "processingAtClearTime", + "note" + ] + } + } + } + }, + "400": { + "description": "Invalid request.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 400, + "error": "Bad Request", + "message": "Invalid request." + } + } + } + }, + "401": { + "description": "Missing or invalid API key.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 401, + "error": "Unauthorized", + "message": "Missing or invalid API key." + } + } + } + }, + "404": { + "description": "Instance or resource not found.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 404, + "error": "Not Found", + "message": "Resource not found." + } + } + } + }, + "422": { + "description": "Request validation failed.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 422, + "error": "Unprocessable Entity", + "message": [ + "A required field is missing." + ] + } + } + } + }, + "429": { + "description": "Request rate limit exceeded.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 429, + "error": "Too Many Requests", + "message": "Request rate limit exceeded." + } + } + } + }, + "500": { + "description": "Unexpected server error.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 500, + "error": "Internal Server Error", + "message": "Unexpected server error." + } + } + } + } + }, + "x-validation-schema": "null", + "x-source": "src/api/routes/sendMessage.router.ts" + } + }, + { + "method": "get", + "path": "/message/outboundQueue/{instanceName}", + "source": "src/api/routes/sendMessage.router.ts", + "schemaName": "null", + "multipart": false, + "runtimeStatus": "OK", + "operation": { + "tags": [ + "Messages" + ], + "summary": "Inspect the outbound text-message queue", + "description": "Returns complete queue counts, the next estimated send time, the estimated empty time, and scheduled entries. Every message is checked against the current safety policy again before delivery.", + "operationId": "get_message_outboundQueue", + "parameters": [ + { + "name": "instanceName", + "in": "path", + "required": true, + "description": "Evolution API instance name.", + "schema": { + "type": "string" + } + }, + { + "name": "limit", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 500 + } + } + ], + "responses": { + "200": { + "description": "Current outbound text-message queue state.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": false, + "properties": { + "pendingCount": { + "type": "integer" + }, + "processingCount": { + "type": "integer" + }, + "failedCount": { + "type": "integer" + }, + "nextSendAt": { + "type": "string", + "format": "date-time", + "nullable": true + }, + "estimatedEmptyAt": { + "type": "string", + "format": "date-time", + "nullable": true + }, + "items": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "properties": { + "id": { + "type": "string" + }, + "recipient": { + "type": "string" + }, + "status": { + "type": "string", + "enum": [ + "PENDING", + "PROCESSING", + "FAILED" + ] + }, + "reason": { + "type": "string", + "nullable": true + }, + "attempts": { + "type": "integer" + }, + "requestedAt": { + "type": "string", + "format": "date-time" + }, + "scheduledAt": { + "type": "string", + "format": "date-time" + }, + "lastError": { + "type": "string", + "nullable": true + } + }, + "required": [] + } + } + }, + "required": [ + "pendingCount", + "processingCount", + "failedCount", + "items" + ] + } + } + } + }, + "400": { + "description": "Invalid request.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 400, + "error": "Bad Request", + "message": "Invalid request." + } + } + } + }, + "401": { + "description": "Missing or invalid API key.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 401, + "error": "Unauthorized", + "message": "Missing or invalid API key." + } + } + } + }, + "404": { + "description": "Instance or resource not found.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 404, + "error": "Not Found", + "message": "Resource not found." + } + } + } + }, + "422": { + "description": "Request validation failed.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 422, + "error": "Unprocessable Entity", + "message": [ + "A required field is missing." + ] + } + } + } + }, + "429": { + "description": "Request rate limit exceeded.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 429, + "error": "Too Many Requests", + "message": "Request rate limit exceeded." + } + } + } + }, + "500": { + "description": "Unexpected server error.", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true, + "properties": { + "status": { + "type": "integer" + }, + "error": { + "type": "string" + }, + "message": { + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + } + } + }, + "example": { + "status": 500, + "error": "Internal Server Error", + "message": "Unexpected server error." + } + } + } + } + }, + "x-validation-schema": "null", + "x-source": "src/api/routes/sendMessage.router.ts" + } + }, { "method": "post", "path": "/message/sendButtons/{instanceName}", @@ -61051,8 +61683,8 @@ "tags": [ "Messages" ], - "summary": "Create send Text", - "description": "Create send Text for Messages.", + "summary": "Send or queue a text message", + "description": "Sends immediately when Automation Safety permits it. A transient cooldown returns a queued result instead of HTTP 429; permanent policy decisions still fail.", "operationId": "post_message_sendText", "parameters": [ { @@ -61067,24 +61699,58 @@ ], "responses": { "201": { - "description": "Resource created. See the domain documentation for the operation-specific response fields.", + "description": "Message sent or accepted into the durable outbound queue.", "content": { "application/json": { "schema": { "oneOf": [ { "type": "object", - "additionalProperties": true - }, - { - "type": "array", - "items": { - "type": "object", - "additionalProperties": true - } + "additionalProperties": true, + "description": "WhatsApp message result after immediate send." }, { - "type": "string" + "type": "object", + "additionalProperties": false, + "properties": { + "queued": { + "type": "boolean", + "enum": [ + true + ] + }, + "queueId": { + "type": "string" + }, + "status": { + "type": "string", + "enum": [ + "PENDING" + ] + }, + "reason": { + "type": "string" + }, + "position": { + "type": "integer" + }, + "scheduledAt": { + "type": "string", + "format": "date-time" + }, + "retryAfterSeconds": { + "type": "integer" + } + }, + "required": [ + "queued", + "queueId", + "status", + "reason", + "position", + "scheduledAt", + "retryAfterSeconds" + ] } ] } diff --git a/package-lock.json b/package-lock.json index 6337301461..c32bd88ff5 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "evolution-api", - "version": "5.4.3", + "version": "5.5.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "evolution-api", - "version": "5.4.3", + "version": "5.5.0", "hasInstallScript": true, "license": "Apache-2.0", "dependencies": { diff --git a/package.json b/package.json index 2244a5909d..b0f46375de 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "evolution-api", - "version": "5.4.3", + "version": "5.5.0", "description": "Rest api for communication with WhatsApp", "main": "./dist/main.js", "type": "commonjs", @@ -15,6 +15,8 @@ "test:archive": "tsx ./test/archive.test.ts", "test:message-archive": "tsx ./test/message-archive.test.ts", "test:outbound-safety": "tsx ./test/outbound-safety.test.ts", + "test:outbound-queue": "tsx ./test/outbound-message-queue.test.ts", + "test:outbound-queue:postgres": "tsx ./test/outbound-message-queue.postgres.test.ts", "test:outbound-safety:postgres": "tsx ./test/outbound-safety.postgres.test.ts", "test:delete-message": "tsx ./test/message-key.test.ts", "lint": "eslint --fix --ext .ts src", diff --git a/prisma/mysql-migrations/20260923010000_outbound_message_queue/migration.sql b/prisma/mysql-migrations/20260923010000_outbound_message_queue/migration.sql new file mode 100644 index 0000000000..9d195fa14d --- /dev/null +++ b/prisma/mysql-migrations/20260923010000_outbound_message_queue/migration.sql @@ -0,0 +1,19 @@ +CREATE TABLE `OutboundMessageQueue` ( + `id` VARCHAR(191) NOT NULL, + `instanceId` VARCHAR(191) NOT NULL, + `recipient` VARCHAR(100) NOT NULL, + `payload` JSON NOT NULL, + `settingsTemplateId` VARCHAR(100) NULL, + `status` VARCHAR(30) NOT NULL DEFAULT 'PENDING', + `reason` VARCHAR(100) NULL, + `attempts` INTEGER NOT NULL DEFAULT 0, + `scheduledAt` DATETIME(3) NOT NULL, + `requestedAt` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + `lockedAt` DATETIME(3) NULL, + `lastError` TEXT NULL, + `updatedAt` DATETIME(3) NOT NULL, + PRIMARY KEY (`id`), + INDEX `OutboundMessageQueue_instanceId_status_scheduledAt_idx` (`instanceId`, `status`, `scheduledAt`), + INDEX `OutboundMessageQueue_instanceId_recipient_status_idx` (`instanceId`, `recipient`, `status`), + CONSTRAINT `OutboundMessageQueue_instanceId_fkey` FOREIGN KEY (`instanceId`) REFERENCES `Instance`(`id`) ON DELETE CASCADE ON UPDATE CASCADE +) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; diff --git a/prisma/mysql-schema.prisma b/prisma/mysql-schema.prisma index abd0099692..11789957a5 100644 --- a/prisma/mysql-schema.prisma +++ b/prisma/mysql-schema.prisma @@ -115,6 +115,7 @@ model Instance { LocalReadSnapshot LocalReadSnapshot[] IsOnWhatsapp IsOnWhatsapp[] OutboundMessageAudit OutboundMessageAudit[] + OutboundMessageQueue OutboundMessageQueue[] RecipientEngagement RecipientEngagement[] SettingsTemplateBinding SettingsTemplateBinding[] } @@ -367,6 +368,26 @@ model OutboundMessageAudit { @@index([instanceId, recipient, messageHash, requestedAt]) } +model OutboundMessageQueue { + id String @id @default(cuid()) + instanceId String + recipient String @db.VarChar(100) + payload Json @db.Json + settingsTemplateId String? @db.VarChar(100) + status String @default("PENDING") @db.VarChar(30) + reason String? @db.VarChar(100) + attempts Int @default(0) @db.Int + scheduledAt DateTime @db.DateTime(3) + requestedAt DateTime @default(now()) @db.DateTime(3) + lockedAt DateTime? @db.DateTime(3) + lastError String? @db.Text + updatedAt DateTime @updatedAt @db.DateTime(3) + Instance Instance @relation(fields: [instanceId], references: [id], onDelete: Cascade) + + @@index([instanceId, status, scheduledAt]) + @@index([instanceId, recipient, status]) +} + model RecipientEngagement { id String @id @default(cuid()) instanceId String diff --git a/prisma/postgresql-migrations/20260923010000_outbound_message_queue/migration.sql b/prisma/postgresql-migrations/20260923010000_outbound_message_queue/migration.sql new file mode 100644 index 0000000000..f1047be33e --- /dev/null +++ b/prisma/postgresql-migrations/20260923010000_outbound_message_queue/migration.sql @@ -0,0 +1,22 @@ +CREATE TABLE "OutboundMessageQueue" ( + "id" TEXT NOT NULL, + "instanceId" TEXT NOT NULL, + "recipient" VARCHAR(100) NOT NULL, + "payload" JSONB NOT NULL, + "settingsTemplateId" VARCHAR(100), + "status" VARCHAR(30) NOT NULL DEFAULT 'PENDING', + "reason" VARCHAR(100), + "attempts" INTEGER NOT NULL DEFAULT 0, + "scheduledAt" TIMESTAMP NOT NULL, + "requestedAt" TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + "lockedAt" TIMESTAMP, + "lastError" TEXT, + "updatedAt" TIMESTAMP NOT NULL, + CONSTRAINT "OutboundMessageQueue_pkey" PRIMARY KEY ("id"), + CONSTRAINT "OutboundMessageQueue_instanceId_fkey" FOREIGN KEY ("instanceId") REFERENCES "Instance"("id") ON DELETE CASCADE ON UPDATE CASCADE +); + +CREATE INDEX "OutboundMessageQueue_instanceId_status_scheduledAt_idx" + ON "OutboundMessageQueue"("instanceId", "status", "scheduledAt"); +CREATE INDEX "OutboundMessageQueue_instanceId_recipient_status_idx" + ON "OutboundMessageQueue"("instanceId", "recipient", "status"); diff --git a/prisma/postgresql-schema.prisma b/prisma/postgresql-schema.prisma index 20774f95f2..81b8f7cc1e 100644 --- a/prisma/postgresql-schema.prisma +++ b/prisma/postgresql-schema.prisma @@ -115,6 +115,7 @@ model Instance { LocalReadSnapshot LocalReadSnapshot[] IsOnWhatsapp IsOnWhatsapp[] OutboundMessageAudit OutboundMessageAudit[] + OutboundMessageQueue OutboundMessageQueue[] RecipientEngagement RecipientEngagement[] SettingsTemplateBinding SettingsTemplateBinding[] } @@ -368,6 +369,26 @@ model OutboundMessageAudit { @@index([instanceId, recipient, messageHash, requestedAt]) } +model OutboundMessageQueue { + id String @id @default(cuid()) + instanceId String + recipient String @db.VarChar(100) + payload Json @db.JsonB + settingsTemplateId String? @db.VarChar(100) + status String @default("PENDING") @db.VarChar(30) + reason String? @db.VarChar(100) + attempts Int @default(0) @db.Integer + scheduledAt DateTime @db.Timestamp + requestedAt DateTime @default(now()) @db.Timestamp + lockedAt DateTime? @db.Timestamp + lastError String? @db.Text + updatedAt DateTime @updatedAt @db.Timestamp + Instance Instance @relation(fields: [instanceId], references: [id], onDelete: Cascade) + + @@index([instanceId, status, scheduledAt]) + @@index([instanceId, recipient, status]) +} + model RecipientEngagement { id String @id @default(cuid()) instanceId String diff --git a/prisma/psql_bouncer-schema.prisma b/prisma/psql_bouncer-schema.prisma index 672e9825d6..5977bf9d58 100644 --- a/prisma/psql_bouncer-schema.prisma +++ b/prisma/psql_bouncer-schema.prisma @@ -116,6 +116,7 @@ model Instance { LocalReadSnapshot LocalReadSnapshot[] IsOnWhatsapp IsOnWhatsapp[] OutboundMessageAudit OutboundMessageAudit[] + OutboundMessageQueue OutboundMessageQueue[] RecipientEngagement RecipientEngagement[] } @@ -343,6 +344,26 @@ model OutboundMessageAudit { @@index([instanceId, recipient, messageHash, requestedAt]) } +model OutboundMessageQueue { + id String @id @default(cuid()) + instanceId String + recipient String @db.VarChar(100) + payload Json @db.JsonB + settingsTemplateId String? @db.VarChar(100) + status String @default("PENDING") @db.VarChar(30) + reason String? @db.VarChar(100) + attempts Int @default(0) @db.Integer + scheduledAt DateTime @db.Timestamp + requestedAt DateTime @default(now()) @db.Timestamp + lockedAt DateTime? @db.Timestamp + lastError String? @db.Text + updatedAt DateTime @updatedAt @db.Timestamp + Instance Instance @relation(fields: [instanceId], references: [id], onDelete: Cascade) + + @@index([instanceId, status, scheduledAt]) + @@index([instanceId, recipient, status]) +} + model RecipientEngagement { id String @id @default(cuid()) instanceId String diff --git a/scripts/runtime-openapi-overrides.mjs b/scripts/runtime-openapi-overrides.mjs index 840f3bb266..94b8a83ead 100644 --- a/scripts/runtime-openapi-overrides.mjs +++ b/scripts/runtime-openapi-overrides.mjs @@ -22,6 +22,10 @@ const body = (properties, required = []) => ({ 'application/json': { schema: bodySchema(properties, required) }, }, }); +const jsonResponse = (description, schema, example) => ({ + description, + content: { 'application/json': { schema, ...(example ? { example } : {}) } }, +}); const limit = { ...integer, minimum: 1, maximum: 1000 }; const eventsQuery = { limit, @@ -96,6 +100,83 @@ export const overrides = { 'get /settings/outbound-audit/{instanceName}': { parameters: params({ limit: { ...integer, minimum: 1 }, recipient: string, status: string }), }, + 'get /message/outboundQueue/{instanceName}': { + summary: 'Inspect the outbound text-message queue', + description: + 'Returns complete queue counts, the next estimated send time, the estimated empty time, and scheduled entries. Every message is checked against the current safety policy again before delivery.', + parameters: params({ limit: { ...integer, minimum: 1, maximum: 500 } }), + responses: { + 200: jsonResponse( + 'Current outbound text-message queue state.', + bodySchema( + { + pendingCount: integer, + processingCount: integer, + failedCount: integer, + nextSendAt: { ...string, format: 'date-time', nullable: true }, + estimatedEmptyAt: { ...string, format: 'date-time', nullable: true }, + items: { + type: 'array', + items: bodySchema({ + id: string, + recipient: string, + status: { type: 'string', enum: ['PENDING', 'PROCESSING', 'FAILED'] }, + reason: { ...string, nullable: true }, + attempts: integer, + requestedAt: { ...string, format: 'date-time' }, + scheduledAt: { ...string, format: 'date-time' }, + lastError: { ...string, nullable: true }, + }), + }, + }, + ['pendingCount', 'processingCount', 'failedCount', 'items'], + ), + ), + }, + }, + 'post /message/sendText/{instanceName}': { + summary: 'Send or queue a text message', + description: + 'Sends immediately when Automation Safety permits it. A transient cooldown returns a queued result instead of HTTP 429; permanent policy decisions still fail.', + responses: { + 201: jsonResponse('Message sent or accepted into the durable outbound queue.', { + oneOf: [ + { type: 'object', additionalProperties: true, description: 'WhatsApp message result after immediate send.' }, + bodySchema( + { + queued: { type: 'boolean', enum: [true] }, + 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'], + ), + ], + }), + }, + }, + 'delete /message/outboundQueue/{instanceName}': { + summary: 'Clear the outbound text-message queue', + description: + 'Deletes queued messages that have not been delivered. A message already handed to WhatsApp cannot be recalled.', + responses: { + 200: jsonResponse( + 'Queue clear result.', + bodySchema( + { + cleared: boolean, + deletedCount: integer, + processingAtClearTime: integer, + note: string, + }, + ['cleared', 'deletedCount', 'processingAtClearTime', 'note'], + ), + ), + }, + }, 'post /chat/getBase64FromMediaMessage/{instanceName}': { requestBody: body( { @@ -118,6 +199,11 @@ export const applyOverride = (route, operation) => { let x = overrides[`${route.method} ${route.path}`]; if (route.path.startsWith('/archive/')) x = withArchiveHeader(x ?? {}); return Object.keys(x ?? {}).length - ? { ...operation, ...x, parameters: [...(operation.parameters ?? []), ...(x.parameters ?? [])] } + ? { + ...operation, + ...x, + parameters: [...(operation.parameters ?? []), ...(x.parameters ?? [])], + responses: { ...(operation.responses ?? {}), ...(x.responses ?? {}) }, + } : operation; }; diff --git a/src/api/controllers/sendMessage.controller.ts b/src/api/controllers/sendMessage.controller.ts index 64aa1c8468..83a229b176 100644 --- a/src/api/controllers/sendMessage.controller.ts +++ b/src/api/controllers/sendMessage.controller.ts @@ -39,6 +39,14 @@ export class SendMessageController { return await this.waMonitor.waInstances[instanceName].textMessage(data); } + public async outboundQueue({ instanceName }: InstanceDto, limit?: number) { + return await this.waMonitor.waInstances[instanceName].outboundQueueStatus(limit); + } + + public async clearOutboundQueue({ instanceName }: InstanceDto) { + return await this.waMonitor.waInstances[instanceName].clearOutboundQueue(); + } + public async sendMedia({ instanceName }: InstanceDto, data: SendMediaDto, file?: any) { if (isBase64(data?.media) && !data?.fileName && data?.mediatype === 'document') { throw new BadRequestException('For base64 the file name must be informed.'); diff --git a/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts b/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts index ce0e3ca191..957a1f649e 100644 --- a/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts +++ b/src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts @@ -74,6 +74,11 @@ import { resolveMaximumDeleteScope, } from '@api/services/message-key.service'; import { applyMistakesToMessage } from '@api/services/mistakes-generator.service'; +import { + OutboundMessageQueueService, + OutboundQueueCapacityError, + QUEUEABLE_OUTBOUND_SAFETY_CODES, +} from '@api/services/outbound-message-queue.service'; import { OutboundSafetyService } from '@api/services/outbound-safety.service'; import { SettingsTemplateService } from '@api/services/settings-template.service'; import { Events, MessageSubtype, TypeMediaMessage, wa } from '@api/types/wa.types'; @@ -255,6 +260,7 @@ export class BaileysStartupService extends ChannelStartupService { private readonly authoritativeLids = new AuthoritativeLidRegistry(); private messageProcessor = new BaileysMessageProcessor(); private readonly outboundSafety: OutboundSafetyService; + private readonly outboundMessageQueue: OutboundMessageQueueService; private readonly settingsTemplates: SettingsTemplateService; constructor( @@ -268,6 +274,7 @@ export class BaileysStartupService extends ChannelStartupService { ) { super(configService, eventEmitter, prismaRepository, chatwootCache); this.outboundSafety = new OutboundSafetyService(prismaRepository); + this.outboundMessageQueue = new OutboundMessageQueueService(prismaRepository); this.settingsTemplates = new SettingsTemplateService(prismaRepository); this.instance.qrcode = { count: 0 }; this.messageProcessor.mount({ @@ -283,6 +290,8 @@ export class BaileysStartupService extends ChannelStartupService { private endSession = false; private logBaileys = this.configService.get('LOG').BAILEYS; private eventProcessingQueue: Promise = Promise.resolve(); + private outboundQueueTimer?: NodeJS.Timeout; + private outboundQueueProcessing = false; // Cache TTL constants (in seconds) private readonly MESSAGE_CACHE_TTL_SECONDS = 5 * 60; // 5 minutes - avoid duplicate message processing @@ -297,6 +306,7 @@ export class BaileysStartupService extends ChannelStartupService { } public async logoutInstance() { + this.stopOutboundQueueWorker(); this.messageProcessor.onDestroy(); await this.client?.logout('Log out instance: ' + this.instanceName); @@ -456,6 +466,7 @@ export class BaileysStartupService extends ChannelStartupService { } if (connection === 'close') { + this.stopOutboundQueueWorker(); const statusCode = (lastDisconnect?.error as Boom)?.output?.statusCode; const codesToNotReconnect = [DisconnectReason.loggedOut, DisconnectReason.forbidden, 402, 406]; const shouldReconnect = !codesToNotReconnect.includes(statusCode); @@ -497,6 +508,7 @@ export class BaileysStartupService extends ChannelStartupService { } if (connection === 'open') { + this.startOutboundQueueWorker(); this.instance.wuid = this.client.user.id.replace(/:\d+/, ''); try { const profilePic = await this.profilePicture(this.instance.wuid); @@ -3004,28 +3016,142 @@ export class BaileysStartupService extends ChannelStartupService { } // Send Message Controller - public async textMessage(data: SendTextDto, isIntegration = false) { + public async textMessage(data: SendTextDto, isIntegration = false, fromQueue = false) { const text = data.text; if (!text || text.trim().length === 0) { throw new BadRequestException('Text is required'); } - return await this.sendMessageWithTyping( - data.number, - { conversation: data.text }, - { - delay: data?.delay, - presence: 'composing', - quoted: data?.quoted, - linkPreview: data?.linkPreview, - mentionsEveryOne: data?.mentionsEveryOne, - mentioned: data?.mentioned, - settingsTemplateId: data?.settingsTemplateId, - }, - isIntegration, - data.settingsTemplateId, - ); + try { + return await this.sendMessageWithTyping( + data.number, + { conversation: data.text }, + { + delay: data?.delay, + presence: 'composing', + quoted: data?.quoted, + linkPreview: data?.linkPreview, + mentionsEveryOne: data?.mentionsEveryOne, + mentioned: data?.mentioned, + settingsTemplateId: data?.settingsTemplateId, + }, + isIntegration, + data.settingsTemplateId, + ); + } catch (error) { + const safetyError = this.parseOutboundSafetyError(error); + if (fromQueue || !safetyError || !QUEUEABLE_OUTBOUND_SAFETY_CODES.has(safetyError.code)) throw error; + + const recipient = createJid(data.number).toLowerCase(); + const templateSettings = await this.settingsTemplates.resolve( + this.instanceId, + recipient, + data.settingsTemplateId, + ); + const effectiveSafety = templateSettings?.automationSafety ?? this.localSettings.automationSafety; + try { + const queued = await this.outboundMessageQueue.enqueueText( + this.instanceId, + recipient, + data, + safetyError.code, + safetyError.retryAfterSeconds, + effectiveSafety, + isIntegration, + ); + this.startOutboundQueueWorker(); + return queued; + } catch (queueError) { + if (queueError instanceof OutboundQueueCapacityError) { + throw new TooManyRequestsException({ + code: queueError.code, + retryAfterSeconds: queueError.retryAfterSeconds, + message: queueError.message, + }); + } + throw queueError; + } + } + } + + public async outboundQueueStatus(limit = 100) { + return this.outboundMessageQueue.snapshot(this.instanceId, limit); + } + + public async clearOutboundQueue() { + return this.outboundMessageQueue.clear(this.instanceId); + } + + private parseOutboundSafetyError(error: any): { code: string; retryAfterSeconds: number } | null { + if (Number(error?.status) !== 429) return null; + const details = Array.isArray(error?.message) ? error.message[0] : error?.message; + if (!details?.code) return null; + return { + code: String(details.code), + retryAfterSeconds: Math.max(1, Number(details.retryAfterSeconds) || 1), + }; + } + + private startOutboundQueueWorker() { + if (this.outboundQueueTimer) clearTimeout(this.outboundQueueTimer); + this.outboundQueueTimer = setTimeout(() => { + this.outboundQueueTimer = undefined; + void this.processOutboundQueue(); + }, 0); + this.outboundQueueTimer.unref?.(); + } + + private stopOutboundQueueWorker() { + if (this.outboundQueueTimer) clearTimeout(this.outboundQueueTimer); + this.outboundQueueTimer = undefined; + } + + private scheduleNextOutboundQueueRun(nextSendAt: Date | null) { + if (!nextSendAt || this.stateConnection.state !== 'open') return; + const delayMs = Math.min(60_000, Math.max(1000, nextSendAt.getTime() - Date.now())); + if (this.outboundQueueTimer) clearTimeout(this.outboundQueueTimer); + this.outboundQueueTimer = setTimeout(() => { + this.outboundQueueTimer = undefined; + void this.processOutboundQueue(); + }, delayMs); + this.outboundQueueTimer.unref?.(); + } + + private async processOutboundQueue() { + if (this.outboundQueueProcessing || this.stateConnection.state !== 'open') return; + this.outboundQueueProcessing = true; + try { + const queued = await this.outboundMessageQueue.claimNext(this.instanceId); + if (!queued) return; + 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); + } catch (error) { + const safetyError = this.parseOutboundSafetyError(error); + if (safetyError && QUEUEABLE_OUTBOUND_SAFETY_CODES.has(safetyError.code)) { + await this.outboundMessageQueue.reschedule( + queued.id, + safetyError.code, + safetyError.retryAfterSeconds, + JSON.stringify(error), + ); + } else { + await this.outboundMessageQueue.fail(queued.id, error?.message ?? error?.toString() ?? 'Queue send failed'); + } + } + } catch (error) { + this.logger.error(['Outbound queue worker failed', error?.message ?? error]); + } finally { + this.outboundQueueProcessing = false; + try { + const snapshot = await this.outboundMessageQueue.snapshot(this.instanceId, 1); + this.scheduleNextOutboundQueueRun(snapshot.nextSendAt); + } catch (error) { + this.logger.warn(['Could not schedule the next outbound queue run', error?.message ?? error]); + } + } } public async pollMessage(data: SendPollDto) { diff --git a/src/api/routes/sendMessage.router.ts b/src/api/routes/sendMessage.router.ts index cd073dba3d..6b0be3d5b5 100644 --- a/src/api/routes/sendMessage.router.ts +++ b/src/api/routes/sendMessage.router.ts @@ -1,4 +1,5 @@ import { RouterBroker } from '@api/abstract/abstract.router'; +import { InstanceDto } from '@api/dto/instance.dto'; import { SendAudioDto, SendButtonsDto, @@ -61,6 +62,26 @@ export class MessageRouter extends RouterBroker { return res.status(HttpStatus.CREATED).json(response); }) + .get(this.routerPath('outboundQueue'), ...guards, async (req, res) => { + const response = await this.dataValidate({ + request: req, + schema: null, + ClassRef: InstanceDto, + execute: (instance) => sendMessageController.outboundQueue(instance, Number(req.query.limit) || undefined), + }); + + return res.status(HttpStatus.OK).json(response); + }) + .delete(this.routerPath('outboundQueue'), ...guards, async (req, res) => { + const response = await this.dataValidate({ + request: req, + schema: null, + ClassRef: InstanceDto, + execute: (instance) => sendMessageController.clearOutboundQueue(instance), + }); + + return res.status(HttpStatus.OK).json(response); + }) .post(this.routerPath('sendMedia'), ...guards, upload.single('file'), async (req, res) => { const bodyData = req.body; diff --git a/src/api/services/outbound-message-queue.service.ts b/src/api/services/outbound-message-queue.service.ts new file mode 100644 index 0000000000..595634fef8 --- /dev/null +++ b/src/api/services/outbound-message-queue.service.ts @@ -0,0 +1,241 @@ +import { SendTextDto } from '@api/dto/sendMessage.dto'; +import { AutomationSafetySettings } from '@api/dto/settings.dto'; +import { PrismaRepository } from '@api/repository/repository.service'; + +import { normalizeAutomationSafety } from './outbound-safety.service'; + +export const QUEUEABLE_OUTBOUND_SAFETY_CODES = new Set([ + 'concurrency_limit', + 'failure_circuit_open', + 'instance_rate_limit', + 'instance_daily_limit', + 'minimum_interval', + 'outreach_recipient_limit', + 'recipient_rate_limit', + 'recipient_daily_limit', +]); + +export type OutboundQueueSnapshot = { + pendingCount: number; + processingCount: number; + failedCount: number; + nextSendAt: Date | null; + estimatedEmptyAt: Date | null; + items: Array<{ + id: string; + recipient: string; + status: string; + reason: string | null; + attempts: number; + requestedAt: Date; + scheduledAt: Date; + lastError: string | null; + }>; +}; + +export class OutboundQueueCapacityError extends Error { + constructor( + public readonly code: 'outbound_queue_instance_capacity' | 'outbound_queue_recipient_capacity', + public readonly retryAfterSeconds: number, + ) { + super('Outbound queue plus recently sent messages reached the configured daily limit.'); + } +} + +export class OutboundMessageQueueService { + constructor(private readonly repository: PrismaRepository) {} + + public async enqueueText( + instanceId: string, + recipient: string, + payload: SendTextDto, + reason: string, + retryAfterSeconds: number, + rawPolicy?: AutomationSafetySettings | null, + isIntegration = false, + ) { + const policy = normalizeAutomationSafety(rawPolicy); + const now = new Date(); + const dayAgo = new Date(now.getTime() - 86_400_000); + const queuedStatuses = ['PENDING', 'PROCESSING']; + + await this.repository.outboundMessageQueue.deleteMany({ + where: { + instanceId, + status: 'FAILED', + requestedAt: { lt: new Date(now.getTime() - policy.audit.retentionDays * 86_400_000) }, + }, + }); + + 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 requestedSchedule = now.getTime() + Math.max(1, retryAfterSeconds) * 1000; + const afterExistingQueue = lastQueued + ? lastQueued.scheduledAt.getTime() + Math.max(1, policy.rateLimit.minimumIntervalMs) + : requestedSchedule; + const scheduledAt = new Date(Math.max(requestedSchedule, afterExistingQueue)); + + const item = await this.repository.outboundMessageQueue.create({ + data: { + instanceId, + recipient, + payload: JSON.parse(JSON.stringify({ data: payload, isIntegration })), + settingsTemplateId: payload.settingsTemplateId || null, + reason, + scheduledAt, + }, + }); + const position = + (await this.repository.outboundMessageQueue.count({ + where: { + instanceId, + status: { in: queuedStatuses }, + scheduledAt: { lte: item.scheduledAt }, + }, + })) || 1; + + return { + queued: true, + queueId: item.id, + status: item.status, + reason, + position, + scheduledAt: item.scheduledAt, + retryAfterSeconds: Math.max(1, Math.ceil((item.scheduledAt.getTime() - now.getTime()) / 1000)), + }; + } + + public async snapshot(instanceId: string, limit = 100): Promise { + const safeLimit = Math.min(500, Math.max(1, Number(limit) || 100)); + const [pendingCount, processingCount, failedCount, items, nextPending, lastPending] = await Promise.all([ + this.repository.outboundMessageQueue.count({ where: { instanceId, status: 'PENDING' } }), + this.repository.outboundMessageQueue.count({ where: { instanceId, status: 'PROCESSING' } }), + this.repository.outboundMessageQueue.count({ where: { instanceId, status: 'FAILED' } }), + this.repository.outboundMessageQueue.findMany({ + where: { instanceId, status: { in: ['PENDING', 'PROCESSING', 'FAILED'] } }, + orderBy: [{ scheduledAt: 'asc' }, { requestedAt: 'asc' }], + take: safeLimit, + select: { + id: true, + recipient: true, + status: true, + reason: true, + attempts: true, + requestedAt: true, + scheduledAt: true, + lastError: true, + }, + }), + this.repository.outboundMessageQueue.findFirst({ + where: { instanceId, status: { in: ['PENDING', 'PROCESSING'] } }, + orderBy: { scheduledAt: 'asc' }, + select: { scheduledAt: true }, + }), + this.repository.outboundMessageQueue.findFirst({ + where: { instanceId, status: { in: ['PENDING', 'PROCESSING'] } }, + orderBy: { scheduledAt: 'desc' }, + select: { scheduledAt: true }, + }), + ]); + return { + pendingCount, + processingCount, + failedCount, + nextSendAt: nextPending?.scheduledAt ?? null, + estimatedEmptyAt: lastPending?.scheduledAt ?? null, + items, + }; + } + + public async clear(instanceId: string) { + const processingCount = await this.repository.outboundMessageQueue.count({ + where: { instanceId, status: 'PROCESSING' }, + }); + const result = await this.repository.outboundMessageQueue.deleteMany({ where: { instanceId } }); + return { + cleared: true, + deletedCount: result.count, + processingAtClearTime: processingCount, + note: + processingCount > 0 + ? 'Pending rows were deleted. A message already handed to WhatsApp may still complete.' + : 'All queued messages were deleted before delivery.', + }; + } + + public async claimNext(instanceId: string) { + const now = new Date(); + await this.repository.outboundMessageQueue.updateMany({ + where: { + instanceId, + status: 'PROCESSING', + lockedAt: { lt: new Date(now.getTime() - 300_000) }, + }, + data: { status: 'PENDING', lockedAt: null }, + }); + const candidate = await this.repository.outboundMessageQueue.findFirst({ + where: { instanceId, status: 'PENDING', scheduledAt: { lte: now } }, + orderBy: [{ scheduledAt: 'asc' }, { requestedAt: 'asc' }], + }); + if (!candidate) return null; + const claim = await this.repository.outboundMessageQueue.updateMany({ + where: { id: candidate.id, status: 'PENDING' }, + data: { status: 'PROCESSING', lockedAt: now, attempts: { increment: 1 } }, + }); + if (claim.count !== 1) return null; + return this.repository.outboundMessageQueue.findUnique({ where: { id: candidate.id } }); + } + + public async complete(id: string): Promise { + await this.repository.outboundMessageQueue.deleteMany({ where: { id } }); + } + + public async reschedule(id: string, reason: string, retryAfterSeconds: number, lastError?: string): Promise { + await this.repository.outboundMessageQueue.updateMany({ + where: { id }, + data: { + status: 'PENDING', + reason, + scheduledAt: new Date(Date.now() + Math.max(1, retryAfterSeconds) * 1000), + lockedAt: null, + lastError: lastError?.slice(0, 4000) ?? null, + }, + }); + } + + public async fail(id: string, error: string): Promise { + await this.repository.outboundMessageQueue.updateMany({ + where: { id }, + data: { status: 'FAILED', lockedAt: null, lastError: error.slice(0, 4000) }, + }); + } +} diff --git a/src/api/services/outbound-safety.service.ts b/src/api/services/outbound-safety.service.ts index 291805aa9a..2090cc9d82 100644 --- a/src/api/services/outbound-safety.service.ts +++ b/src/api/services/outbound-safety.service.ts @@ -351,46 +351,76 @@ export class OutboundSafetyService { } } if (instanceMinute >= policy.rateLimit.instancePerMinute) { + const oldest = await this.repository.outboundMessageAudit.findFirst({ + where: { instanceId, requestedAt: { gte: minuteAgo }, status: countable }, + orderBy: { requestedAt: 'asc' }, + select: { requestedAt: true }, + }); return this.block( instanceId, normalizedRecipient, messageHash, messageType, 'instance_rate_limit', - 60, + this.retryAfterWindow(oldest?.requestedAt, 60_000, now), recipientCategory, ); } if (instanceDay >= policy.rateLimit.instancePerDay) { + const oldest = await this.repository.outboundMessageAudit.findFirst({ + where: { instanceId, requestedAt: { gte: dayAgo }, status: countable }, + orderBy: { requestedAt: 'asc' }, + select: { requestedAt: true }, + }); return this.block( instanceId, normalizedRecipient, messageHash, messageType, 'instance_daily_limit', - 86400, + this.retryAfterWindow(oldest?.requestedAt, 86_400_000, now), recipientCategory, ); } if (recipientMinute >= policy.rateLimit.recipientPerMinute) { + const oldest = await this.repository.outboundMessageAudit.findFirst({ + where: { + instanceId, + recipient: normalizedRecipient, + requestedAt: { gte: minuteAgo }, + status: countable, + }, + orderBy: { requestedAt: 'asc' }, + select: { requestedAt: true }, + }); return this.block( instanceId, normalizedRecipient, messageHash, messageType, 'recipient_rate_limit', - 60, + this.retryAfterWindow(oldest?.requestedAt, 60_000, now), recipientCategory, ); } if (recipientDay >= policy.rateLimit.recipientPerDay) { + const oldest = await this.repository.outboundMessageAudit.findFirst({ + where: { + instanceId, + recipient: normalizedRecipient, + requestedAt: { gte: dayAgo }, + status: countable, + }, + orderBy: { requestedAt: 'asc' }, + select: { requestedAt: true }, + }); return this.block( instanceId, normalizedRecipient, messageHash, messageType, 'recipient_daily_limit', - 86400, + this.retryAfterWindow(oldest?.requestedAt, 86_400_000, now), recipientCategory, ); } @@ -437,13 +467,18 @@ export class OutboundSafetyService { where: { instanceId, lastOutreachAt: { gte: dayAgo } }, }); if (outreachCount >= policy.outreach.newOrDormantRecipientsPerDay) { + const oldest = await this.repository.recipientEngagement.findFirst({ + where: { instanceId, lastOutreachAt: { gte: dayAgo } }, + orderBy: { lastOutreachAt: 'asc' }, + select: { lastOutreachAt: true }, + }); return this.block( instanceId, normalizedRecipient, messageHash, messageType, 'outreach_recipient_limit', - 86400, + this.retryAfterWindow(oldest?.lastOutreachAt, 86_400_000, now), recipientCategory, ); } @@ -507,6 +542,11 @@ export class OutboundSafetyService { else inFlightByInstance.set(instanceId, active - 1); } + private retryAfterWindow(oldest: Date | null | undefined, windowMs: number, now: Date): number { + if (!oldest) return Math.ceil(windowMs / 1000); + return Math.max(1, Math.ceil((oldest.getTime() + windowMs - now.getTime()) / 1000)); + } + private async block( instanceId: string, recipient: string, diff --git a/test/outbound-message-queue.postgres.test.ts b/test/outbound-message-queue.postgres.test.ts new file mode 100644 index 0000000000..c93bcd0195 --- /dev/null +++ b/test/outbound-message-queue.postgres.test.ts @@ -0,0 +1,55 @@ +import assert from 'node:assert/strict'; + +import { PrismaClient } from '@prisma/client'; + +import { OutboundMessageQueueService } from '../src/api/services/outbound-message-queue.service'; + +async function main() { + if (!process.env.DATABASE_CONNECTION_URI) throw new Error('DATABASE_CONNECTION_URI is required'); + const prisma = new PrismaClient(); + const instance = await prisma.instance.create({ data: { name: `outbound-queue-test-${Date.now()}` } }); + try { + const service = new OutboundMessageQueueService(prisma as any); + const queued = await service.enqueueText( + instance.id, + '15551234567@s.whatsapp.net', + { number: '15551234567', text: 'Persist this exact queued text.' }, + 'outreach_recipient_limit', + 60, + { + enabled: true, + rateLimit: { instancePerDay: 10, recipientPerDay: 5, minimumIntervalMs: 0 }, + }, + ); + assert.equal(queued.queued, true); + + const stored = await prisma.outboundMessageQueue.findUnique({ where: { id: queued.queueId } }); + assert.deepEqual(stored?.payload, { + data: { number: '15551234567', text: 'Persist this exact queued text.' }, + isIntegration: false, + }); + + await prisma.outboundMessageQueue.update({ + where: { id: queued.queueId }, + data: { scheduledAt: new Date(Date.now() - 1000) }, + }); + const claimed = await service.claimNext(instance.id); + assert.equal(claimed?.id, queued.queueId); + assert.equal(claimed?.status, 'PROCESSING'); + assert.equal(claimed?.attempts, 1); + + await service.reschedule(queued.queueId, 'minimum_interval', 30); + assert.equal((await service.snapshot(instance.id)).pendingCount, 1); + + const cleared = await service.clear(instance.id); + assert.equal(cleared.deletedCount, 1); + assert.equal(await prisma.outboundMessageQueue.count({ where: { instanceId: instance.id } }), 0); + } finally { + await prisma.instance.delete({ where: { id: instance.id } }); + assert.equal(await prisma.outboundMessageQueue.count({ where: { instanceId: instance.id } }), 0); + await prisma.$disconnect(); + } + console.log('outbound-message-queue-postgres-tests-pass'); +} + +void main(); diff --git a/test/outbound-message-queue.test.ts b/test/outbound-message-queue.test.ts new file mode 100644 index 0000000000..77b58faa7b --- /dev/null +++ b/test/outbound-message-queue.test.ts @@ -0,0 +1,154 @@ +import assert from 'node:assert/strict'; + +import { + OutboundMessageQueueService, + OutboundQueueCapacityError, + QUEUEABLE_OUTBOUND_SAFETY_CODES, +} from '../src/api/services/outbound-message-queue.service'; + +async function main() { + const rows: any[] = []; + let recentInstanceMessages = 2; + let recentRecipientMessages = 1; + const repository = { + outboundMessageAudit: { + count: async ({ where }: any) => (where.recipient ? recentRecipientMessages : recentInstanceMessages), + }, + outboundMessageQueue: { + count: async ({ where }: any) => + rows.filter( + (row) => + row.instanceId === where.instanceId && + (!where.recipient || row.recipient === where.recipient) && + (!where.status || + row.status === where.status || + (where.status.in && where.status.in.includes(row.status))) && + (!where.scheduledAt?.lte || row.scheduledAt <= where.scheduledAt.lte), + ).length, + findFirst: async ({ where, orderBy }: any) => { + const matches = rows.filter( + (row) => + row.instanceId === where.instanceId && + (!where.status || + row.status === where.status || + (where.status.in && where.status.in.includes(row.status))) && + (!where.scheduledAt?.lte || row.scheduledAt <= where.scheduledAt.lte), + ); + const direction = Array.isArray(orderBy) ? orderBy[0].scheduledAt : orderBy.scheduledAt; + return matches.sort((left, right) => + direction === 'desc' + ? right.scheduledAt.getTime() - left.scheduledAt.getTime() + : left.scheduledAt.getTime() - right.scheduledAt.getTime(), + )[0] ?? null; + }, + findMany: async ({ where, take }: any) => + rows + .filter((row) => where.status.in.includes(row.status)) + .sort((left, right) => left.scheduledAt.getTime() - right.scheduledAt.getTime()) + .slice(0, take), + create: async ({ data }: any) => { + const row = { + id: `queue-${rows.length + 1}`, + status: 'PENDING', + attempts: 0, + requestedAt: new Date(), + lockedAt: null, + lastError: null, + ...data, + }; + rows.push(row); + return row; + }, + updateMany: async ({ where, data }: any) => { + let count = 0; + for (const row of rows) { + if (where.id && row.id !== where.id) continue; + if (where.instanceId && row.instanceId !== where.instanceId) continue; + if (where.status && typeof where.status === 'string' && row.status !== where.status) continue; + row.status = data.status ?? row.status; + row.lockedAt = data.lockedAt === undefined ? row.lockedAt : data.lockedAt; + row.scheduledAt = data.scheduledAt ?? row.scheduledAt; + row.reason = data.reason ?? row.reason; + row.lastError = data.lastError ?? row.lastError; + if (data.attempts?.increment) row.attempts += data.attempts.increment; + count += 1; + } + return { count }; + }, + findUnique: async ({ where }: any) => rows.find((row) => row.id === where.id) ?? null, + deleteMany: async ({ where }: any) => { + const before = rows.length; + for (let index = rows.length - 1; index >= 0; index -= 1) { + if ( + (!where.id || rows[index].id === where.id) && + (!where.instanceId || rows[index].instanceId === where.instanceId) && + (!where.status || rows[index].status === where.status) + ) { + rows.splice(index, 1); + } + } + return { count: before - rows.length }; + }, + }, + } as any; + + const service = new OutboundMessageQueueService(repository); + assert.equal(QUEUEABLE_OUTBOUND_SAFETY_CODES.has('outreach_recipient_limit'), true); + assert.equal(QUEUEABLE_OUTBOUND_SAFETY_CODES.has('recipient_suppressed'), false); + assert.equal(QUEUEABLE_OUTBOUND_SAFETY_CODES.has('duplicate_message'), false); + + const first = await service.enqueueText( + 'instance-1', + '15551234567@s.whatsapp.net', + { number: '15551234567', text: 'First queued message' }, + 'outreach_recipient_limit', + 60, + { enabled: true, rateLimit: { instancePerDay: 10, recipientPerDay: 5, minimumIntervalMs: 1000 } }, + ); + assert.equal(first.queued, true); + assert.equal(first.position, 1); + + const second = await service.enqueueText( + 'instance-1', + '15557654321@s.whatsapp.net', + { number: '15557654321', text: 'Second queued message' }, + 'instance_rate_limit', + 60, + { enabled: true, rateLimit: { instancePerDay: 10, recipientPerDay: 5, minimumIntervalMs: 1000 } }, + ); + assert.equal(second.position, 2); + assert.ok(second.scheduledAt.getTime() > first.scheduledAt.getTime()); + + const snapshot = await service.snapshot('instance-1'); + assert.equal(snapshot.pendingCount, 2); + assert.equal(snapshot.processingCount, 0); + assert.equal(snapshot.nextSendAt?.getTime(), first.scheduledAt.getTime()); + assert.equal(snapshot.estimatedEmptyAt?.getTime(), second.scheduledAt.getTime()); + + recentInstanceMessages = 10; + await assert.rejects( + () => + service.enqueueText( + 'instance-1', + '15550000000@s.whatsapp.net', + { number: '15550000000', text: 'Over capacity' }, + 'instance_daily_limit', + 60, + { enabled: true, rateLimit: { instancePerDay: 10, recipientPerDay: 5 } }, + ), + (error: any) => error instanceof OutboundQueueCapacityError && error.code === 'outbound_queue_instance_capacity', + ); + recentInstanceMessages = 2; + recentRecipientMessages = 1; + + const claimed = await service.claimNext('instance-1'); + assert.equal(claimed, null, 'future work should not be claimed early'); + + const cleared = await service.clear('instance-1'); + assert.equal(cleared.deletedCount, 2); + assert.equal((await service.snapshot('instance-1')).pendingCount, 0); + + console.log('outbound-message-queue-tests-pass'); +} + +void main(); From 87e0e61e3dcad6e3212521e7b8385765eda71ed4 Mon Sep 17 00:00:00 2001 From: menachem mendel kuperman Date: Wed, 23 Sep 2026 15:19:58 +0300 Subject: [PATCH 2/2] test(openapi): account for queue endpoints --- test/review-surfaces.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/review-surfaces.test.ts b/test/review-surfaces.test.ts index 19edc653b2..13129f25e5 100644 --- a/test/review-surfaces.test.ts +++ b/test/review-surfaces.test.ts @@ -7,7 +7,7 @@ const js = fs.readFileSync(`manager/dist${asset.startsWith('/') ? asset : `/${as for (const text of ['Settings templates', 'New template', 'Automation safety & pacing', 'Mistakes generator', 'Minimum letters', 'Chance (%)']) assert(js.includes(text), `embedded Manager contains ${text}`); const routes = JSON.parse(fs.readFileSync('docs/runtime-routes.json', 'utf8')); -assert.equal(routes.length, 202); +assert.equal(routes.length, 204); const sendText = routes.find((route: any) => route.path === '/message/sendText/{instanceName}'); assert.deepEqual(Object.keys(sendText.operation.requestBody.content['application/json'].schema.properties), [ 'number',