From 6f6ed97206a54b0e60b7c75ab6239c26b66b917f Mon Sep 17 00:00:00 2001 From: Gustavo Date: Mon, 3 Aug 2026 13:07:03 -0300 Subject: [PATCH] Add worker safety rails; split evaluateCampaignHealth into pure read + mutating apply Two issues in the campaign worker path: 1. runCampaignQueue had none of the safety rails run.mjs's automatic sender has: no outreach.ativo gate, no business-hours window, no jitter between sends. A campaign left "rodando" would send around the clock even with outreach disabled or outside business hours, and would hammer the same WhatsApp number back-to-back. Added the same outreach.ativo/horaInicio/ horaFim checks (duplicated, not imported, to keep the modules independent) plus a randomized 3-8s jitter between real send attempts. Sleep delay is injectable via an optional sleepMs param so tests can pass sleepMs: 0. 2. evaluateCampaignHealth mutated state (via applyPause) on every call, including from the read-only GET /api/campaigns/:id/saude route -- so simply opening the campaign detail panel in the UI could silently pause a healthy-looking campaign. Split it into a pure evaluateCampaignHealth (computes {pause, reason}, never writes) and a new applyCampaignHealth (evaluates, then persists the pause if warranted). The worker now calls applyCampaignHealth; the GET /saude route keeps calling the pure evaluateCampaignHealth. Tests: worker sends nothing with outreach.ativo=false or outside business hours; evaluateCampaignHealth confirmed not to mutate status even when it would return pause:true; existing mutation assertions moved onto applyCampaignHealth. --- src/outreach/campaign-metrics.mjs | 20 ++++++++-- src/outreach/campaign-worker.mjs | 49 ++++++++++++++++++++++-- tests/campaign-metrics.test.mjs | 32 +++++++++++++++- tests/campaign-worker.test.mjs | 62 +++++++++++++++++++++++++++---- 4 files changed, 149 insertions(+), 14 deletions(-) diff --git a/src/outreach/campaign-metrics.mjs b/src/outreach/campaign-metrics.mjs index aa7d0c8..6357512 100644 --- a/src/outreach/campaign-metrics.mjs +++ b/src/outreach/campaign-metrics.mjs @@ -16,6 +16,12 @@ function applyPause(db, campaignId, reason) { // 10% erro, 20 contatos/10% negativo+opt-out, 50 contatos/5% resposta), // adaptados as tabelas da Comercial via JOIN por outreach_campaign_queue // (nao por um valor de origem dedicado). +// +// PURA/read-only: so calcula e retorna {pause, reason}, nunca escreve no +// banco. Quem precisa da mutacao (pausar de verdade quando pause=true) deve +// chamar applyCampaignHealth, nao esta funcao — assim uma rota GET (como +// /api/campaigns/:id/saude, chamada so de abrir o painel na UI) nunca pausa +// uma campanha como efeito colateral de uma leitura. export function evaluateCampaignHealth(db, campaignId, now = new Date()) { const campaign = db.prepare('SELECT status FROM outreach_campaigns WHERE id = ?').get(campaignId); if (!campaign || campaign.status !== 'rodando') return { pause: false, reason: null }; @@ -36,7 +42,6 @@ export function evaluateCampaignHealth(db, campaignId, now = new Date()) { WHERE q.campaign_id = ? AND s.status = 'erro' AND date(s.sent_at, 'localtime') = ?` ).get(campaignId, todayKey).total); if (failureCount / attemptsToday > 0.10) { - applyPause(db, campaignId, 'high_failure_rate'); return { pause: true, reason: 'high_failure_rate' }; } } @@ -54,7 +59,6 @@ export function evaluateCampaignHealth(db, campaignId, now = new Date()) { AND ls.stage IN ('sem_interesse', 'opt_out')` ).get(campaignId).total); if (negativeCount / contacts > 0.10) { - applyPause(db, campaignId, 'high_negative_rate'); return { pause: true, reason: 'high_negative_rate' }; } } @@ -67,7 +71,6 @@ export function evaluateCampaignHealth(db, campaignId, now = new Date()) { WHERE q.campaign_id = ? AND q.status IN ('sent','error')` ).get(campaignId).total); if (respondedCount / contacts < 0.05) { - applyPause(db, campaignId, 'low_reply_rate'); return { pause: true, reason: 'low_reply_rate' }; } } @@ -75,6 +78,17 @@ export function evaluateCampaignHealth(db, campaignId, now = new Date()) { return { pause: false, reason: null }; } +// MUTANTE: chama a avaliacao pura acima e, se ela mandar pausar, persiste a +// pausa de verdade via applyPause. E isto que o worker deve chamar apos cada +// envio real — e so o worker, nunca uma rota de leitura. +export function applyCampaignHealth(db, campaignId, now = new Date()) { + const resultado = evaluateCampaignHealth(db, campaignId, now); + if (resultado.pause) { + applyPause(db, campaignId, resultado.reason); + } + return resultado; +} + // Numero de leads distintos com resposta por variante, via subquery // correlacionada (nao um JOIN direto) para nao contar duplicado quando um // lead tem mais de um outreach_stage_events. diff --git a/src/outreach/campaign-worker.mjs b/src/outreach/campaign-worker.mjs index 0794f4f..cf7b4eb 100644 --- a/src/outreach/campaign-worker.mjs +++ b/src/outreach/campaign-worker.mjs @@ -1,6 +1,28 @@ import { createZproClient } from './zpro-client.mjs'; import { isOptedOut } from './campaign-store.mjs'; -import { evaluateCampaignHealth } from './campaign-metrics.mjs'; +import { applyCampaignHealth } from './campaign-metrics.mjs'; +import { getSetting } from '../db.mjs'; + +// Mesmo intervalo aleatorio entre envios do run.mjs (INTERVALO_MIN_MS/ +// INTERVALO_MAX_MS), duplicado aqui de proposito — os dois workers devem se +// comportar de forma consistente contra o mesmo numero de WhatsApp, mas sao +// modulos independentes (nao importar um do outro). +const SLEEP_MIN_MS = 3000; +const SLEEP_MAX_MS = 8000; + +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function randomSleepMs() { + return SLEEP_MIN_MS + Math.random() * (SLEEP_MAX_MS - SLEEP_MIN_MS); +} + +// Mesma logica de comparacao de string HH:MM do run.mjs. +function withinBusinessHours(horaInicio, horaFim, now = new Date()) { + const hhmm = now.toTimeString().slice(0, 5); + return hhmm >= horaInicio && hhmm <= horaFim; +} function todayAttempts(db, campaignId) { return Number(db.prepare( @@ -16,10 +38,26 @@ function todayAttempts(db, campaignId) { // (zproClient.sendText) e sincrono: ou resolve (aceito pelo Z-PRO) ou lanca // erro. Por isso o resultado ja e gravado direto como 'enviado'/'erro', sem // um estado intermediario "reservado" nem logica de recuperacao pos-crash. -export async function runCampaignQueue({ db, zproClient, now = new Date(), log = () => {} }) { +export async function runCampaignQueue({ db, zproClient, now = new Date(), log = () => {}, sleepMs = null }) { const client = zproClient || createZproClient(db); const summary = { enviados: 0, erros: 0, pulados: 0 }; + // Mesmas duas travas de seguranca do disparo automatico (run.mjs): + // interruptor geral e janela de horario comercial. Sem elas, o worker de + // campanha mandaria mensagem 24h/dia mesmo com o operador tendo desligado + // o outreach nas Configuracoes. + if (getSetting(db, 'outreach.ativo', 'false') !== 'true') { + log('campanhas: desativado (outreach.ativo=false)'); + return summary; + } + + const horaInicio = getSetting(db, 'outreach.horaInicio', '08:00'); + const horaFim = getSetting(db, 'outreach.horaFim', '18:00'); + if (!withinBusinessHours(horaInicio, horaFim, now)) { + log(`campanhas: fora do horario comercial (${horaInicio}-${horaFim})`); + return summary; + } + const campaigns = db.prepare(`SELECT id FROM outreach_campaigns WHERE status = 'rodando'`).all(); for (const { id: campaignId } of campaigns) { const campaign = db.prepare('SELECT * FROM outreach_campaigns WHERE id = ?').get(campaignId); @@ -77,7 +115,12 @@ export async function runCampaignQueue({ db, zproClient, now = new Date(), log = ).run(queueStatus, sendId, row.id); attemptsToday += 1; - const health = evaluateCampaignHealth(db, campaignId, now); + // Jitter entre tentativas reais de envio (nao entre pulos/cancelamentos, + // que nao batem no Z-PRO) — mesmo padrao de run.mjs, pra nao parecer + // disparo automatizado agressivo pro mesmo numero de WhatsApp. + await sleep(sleepMs != null ? sleepMs : randomSleepMs()); + + const health = applyCampaignHealth(db, campaignId, now); if (health.pause) { log(`campanha #${campaignId}: pausada automaticamente (${health.reason})`); break; diff --git a/tests/campaign-metrics.test.mjs b/tests/campaign-metrics.test.mjs index 4882d7a..cae20ed 100644 --- a/tests/campaign-metrics.test.mjs +++ b/tests/campaign-metrics.test.mjs @@ -2,7 +2,7 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import { openDb, createLeadInserter } from '../src/db.mjs'; import { createCampaign, setCampaignStatus } from '../src/outreach/campaign-store.mjs'; -import { evaluateCampaignHealth, variantStats } from '../src/outreach/campaign-metrics.mjs'; +import { evaluateCampaignHealth, applyCampaignHealth, variantStats } from '../src/outreach/campaign-metrics.mjs'; function inserirLead(db, dedup) { const inserir = createLeadInserter(db); @@ -63,11 +63,41 @@ test('evaluateCampaignHealth pausa por high_failure_rate quando erro > 10% com 1 const resultado = evaluateCampaignHealth(db, campanha.id); assert.equal(resultado.pause, true); assert.equal(resultado.reason, 'high_failure_rate'); + // evaluateCampaignHealth e read-only: so retorna o veredito, nao mexe no banco. + const semMutacao = db.prepare('SELECT status FROM outreach_campaigns WHERE id = ?').get(campanha.id); + assert.equal(semMutacao.status, 'rodando'); + + // A mutacao (pausar de verdade) so acontece via applyCampaignHealth. + const aplicado = applyCampaignHealth(db, campanha.id); + assert.equal(aplicado.pause, true); + assert.equal(aplicado.reason, 'high_failure_rate'); const atualizada = db.prepare('SELECT status FROM outreach_campaigns WHERE id = ?').get(campanha.id); assert.equal(atualizada.status, 'pausado'); db.close(); }); +test('evaluateCampaignHealth nunca muda outreach_campaigns.status, mesmo quando retornaria pause:true (GET /saude nao deve pausar como efeito colateral de uma leitura)', () => { + const db = openDb(':memory:'); + const campanha = campanhaRodando(db); + const variant = db.prepare('SELECT id FROM outreach_campaign_variants WHERE campaign_id = ?').get(campanha.id); + for (let i = 0; i < 8; i += 1) { + const lead = inserirLead(db, `puro-ok${i}`); + simularEnvio(db, campanha.id, variant.id, lead.id, lead.whatsapp, 'enviado'); + } + for (let i = 0; i < 2; i += 1) { + const lead = inserirLead(db, `puro-err${i}`); + simularEnvio(db, campanha.id, variant.id, lead.id, lead.whatsapp, 'erro'); + } + + for (let i = 0; i < 5; i += 1) { + const resultado = evaluateCampaignHealth(db, campanha.id); + assert.equal(resultado.pause, true, 'deveria continuar sinalizando pause:true a cada chamada'); + } + const aindaRodando = db.prepare('SELECT status FROM outreach_campaigns WHERE id = ?').get(campanha.id); + assert.equal(aindaRodando.status, 'rodando', 'chamar evaluateCampaignHealth repetidamente nao deveria ter pausado a campanha'); + db.close(); +}); + test('evaluateCampaignHealth nao pausa por falha com menos de 10 tentativas no dia', () => { const db = openDb(':memory:'); const campanha = campanhaRodando(db); diff --git a/tests/campaign-worker.test.mjs b/tests/campaign-worker.test.mjs index c769003..8ae5f02 100644 --- a/tests/campaign-worker.test.mjs +++ b/tests/campaign-worker.test.mjs @@ -1,10 +1,20 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { openDb, createLeadInserter } from '../src/db.mjs'; +import { openDb, createLeadInserter, setSetting } from '../src/db.mjs'; import { createCampaign, setCampaignStatus, addOptOut } from '../src/outreach/campaign-store.mjs'; import { generateCampaignQueue, listQueue, approveQueueRows } from '../src/outreach/campaign-queue.mjs'; import { runCampaignQueue } from '../src/outreach/campaign-worker.mjs'; +// Testes de banco em memoria nao seedam settings.outreach.* (fica no fallback +// 'false'), entao por padrao o worker nao faria nada — precisa ligar +// explicitamente e abrir a janela de horario pro dia todo pra nao depender do +// horario real em que o teste roda. +function ativarWorker(db) { + setSetting(db, 'outreach.ativo', 'true'); + setSetting(db, 'outreach.horaInicio', '00:00'); + setSetting(db, 'outreach.horaFim', '23:59'); +} + function inserirLead(db, dedup, whatsapp) { const inserir = createLeadInserter(db); inserir({ @@ -43,10 +53,11 @@ function fakeZproClient(overrides = {}) { test('runCampaignQueue envia linhas aprovadas via sendText e marca sent', async () => { const db = openDb(':memory:'); + ativarWorker(db); const campanha = campanhaAprovada(db); const zproClient = fakeZproClient(); - const resumo = await runCampaignQueue({ db, zproClient }); + const resumo = await runCampaignQueue({ db, zproClient, sleepMs: 0 }); assert.equal(resumo.enviados, 1); assert.equal(zproClient.enviados.length, 1); assert.equal(zproClient.enviados[0].message, 'Ola Empresa'); @@ -62,6 +73,7 @@ test('runCampaignQueue envia linhas aprovadas via sendText e marca sent', async test('runCampaignQueue registra erro quando o Z-PRO falha, sem travar o restante', async () => { const db = openDb(':memory:'); + ativarWorker(db); const campanha = campanhaAprovada(db, 2); let chamada = 0; const zproClient = fakeZproClient({ @@ -72,7 +84,7 @@ test('runCampaignQueue registra erro quando o Z-PRO falha, sem travar o restante } }); - const resumo = await runCampaignQueue({ db, zproClient }); + const resumo = await runCampaignQueue({ db, zproClient, sleepMs: 0 }); assert.equal(resumo.enviados, 1); assert.equal(resumo.erros, 1); const erro = listQueue(db, campanha.id, 'error')[0]; @@ -83,6 +95,7 @@ test('runCampaignQueue registra erro quando o Z-PRO falha, sem travar o restante test('runCampaignQueue respeita o limite diario da campanha', async () => { const db = openDb(':memory:'); + ativarWorker(db); for (let i = 0; i < 3; i += 1) inserirLead(db, `l${i}`, `551199999000${i}`); const campanha = createCampaign(db, { nome: 'C', produto: 'ERP', segmentos: ['x'], variantes: [{ code: 'A', mensagem: 'oi' }], @@ -93,7 +106,7 @@ test('runCampaignQueue respeita o limite diario da campanha', async () => { approveQueueRows(db, campanha.id, listQueue(db, campanha.id, 'preview').map((r) => r.id)); const zproClient = fakeZproClient(); - const resumo = await runCampaignQueue({ db, zproClient }); + const resumo = await runCampaignQueue({ db, zproClient, sleepMs: 0 }); assert.equal(resumo.enviados, 1); assert.equal(zproClient.enviados.length, 1); assert.equal(listQueue(db, campanha.id, 'approved').length, 2); @@ -102,12 +115,13 @@ test('runCampaignQueue respeita o limite diario da campanha', async () => { test('runCampaignQueue pula e cancela linha de numero que entrou em opt-out depois da fila gerada', async () => { const db = openDb(':memory:'); + ativarWorker(db); const campanha = campanhaAprovada(db); addOptOut(db, '5511999990000', 'pediu pra parar'); db.prepare(`UPDATE outreach_campaign_queue SET numero = '5511999990000' WHERE campaign_id = ?`).run(campanha.id); const zproClient = fakeZproClient(); - const resumo = await runCampaignQueue({ db, zproClient }); + const resumo = await runCampaignQueue({ db, zproClient, sleepMs: 0 }); assert.equal(resumo.pulados, 1); assert.equal(zproClient.enviados.length, 0); const cancelada = listQueue(db, campanha.id, 'cancelled')[0]; @@ -117,6 +131,7 @@ test('runCampaignQueue pula e cancela linha de numero que entrou em opt-out depo test('runCampaignQueue so processa campanhas com status rodando', async () => { const db = openDb(':memory:'); + ativarWorker(db); inserirLead(db, 'l1', '5511999990001'); const campanha = createCampaign(db, { nome: 'C', produto: 'ERP', segmentos: ['x'], variantes: [{ code: 'A', mensagem: 'oi' }] @@ -126,7 +141,7 @@ test('runCampaignQueue so processa campanhas com status rodando', async () => { // campanha continua em 'rascunho' — nunca foi iniciada const zproClient = fakeZproClient(); - const resumo = await runCampaignQueue({ db, zproClient }); + const resumo = await runCampaignQueue({ db, zproClient, sleepMs: 0 }); assert.equal(resumo.enviados, 0); assert.equal(zproClient.enviados.length, 0); db.close(); @@ -134,6 +149,7 @@ test('runCampaignQueue so processa campanhas com status rodando', async () => { test('runCampaignQueue interrompe a campanha quando a saude aciona auto-pausa', async () => { const db = openDb(':memory:'); + ativarWorker(db); const campanha = campanhaAprovada(db, 10); db.prepare(`UPDATE outreach_campaigns SET limite_diario_atual = 10 WHERE id = ?`).run(campanha.id); let chamada = 0; @@ -145,9 +161,41 @@ test('runCampaignQueue interrompe a campanha quando a saude aciona auto-pausa', } }); - await runCampaignQueue({ db, zproClient }); + await runCampaignQueue({ db, zproClient, sleepMs: 0 }); const atualizada = db.prepare('SELECT status FROM outreach_campaigns WHERE id = ?').get(campanha.id); // 10 tentativas, 2 erros = 20% > 10% -> deveria pausar antes de esgotar a fila assert.equal(atualizada.status, 'pausado'); db.close(); }); + +test('runCampaignQueue nao envia nada quando outreach.ativo esta desligado', async () => { + const db = openDb(':memory:'); + // Nao chama ativarWorker: settings.outreach.ativo fica no fallback 'false'. + const campanha = campanhaAprovada(db); + const zproClient = fakeZproClient(); + + const resumo = await runCampaignQueue({ db, zproClient, sleepMs: 0 }); + assert.equal(resumo.enviados, 0); + assert.equal(zproClient.enviados.length, 0); + assert.equal(listQueue(db, campanha.id, 'approved').length, 1); + db.close(); +}); + +test('runCampaignQueue nao envia nada fora do horario comercial configurado', async () => { + const db = openDb(':memory:'); + setSetting(db, 'outreach.ativo', 'true'); + setSetting(db, 'outreach.horaInicio', '08:00'); + setSetting(db, 'outreach.horaFim', '18:00'); + const campanha = campanhaAprovada(db); + const zproClient = fakeZproClient(); + + // 03:00 (horario local), bem fora da janela 08:00-18:00 configurada acima. + const foraDoHorario = new Date(); + foraDoHorario.setHours(3, 0, 0, 0); + + const resumo = await runCampaignQueue({ db, zproClient, now: foraDoHorario, sleepMs: 0 }); + assert.equal(resumo.enviados, 0); + assert.equal(zproClient.enviados.length, 0); + assert.equal(listQueue(db, campanha.id, 'approved').length, 1); + db.close(); +});