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.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user