Add campaign send worker with daily limit and health auto-pause
This commit is contained in:
Gustavo
@@ -0,0 +1,88 @@
|
||||
import { createZproClient } from './zpro-client.mjs';
|
||||
import { isOptedOut } from './campaign-store.mjs';
|
||||
import { evaluateCampaignHealth } from './campaign-metrics.mjs';
|
||||
|
||||
function todayAttempts(db, campaignId) {
|
||||
return Number(db.prepare(
|
||||
`SELECT COUNT(*) AS total
|
||||
FROM outreach_campaign_queue q
|
||||
JOIN outreach_sends s ON s.id = q.outreach_send_id
|
||||
WHERE q.campaign_id = ? AND date(s.sent_at, 'localtime') = date('now', 'localtime')`
|
||||
).get(campaignId).total);
|
||||
}
|
||||
|
||||
// Diferente da Maquina Pessoal (que reserva o envio antes de mandar, porque a
|
||||
// entrega via Baileys e assincrona/incerta), aqui o envio via Z-PRO
|
||||
// (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 = () => {} }) {
|
||||
const client = zproClient || createZproClient(db);
|
||||
const summary = { enviados: 0, erros: 0, pulados: 0 };
|
||||
|
||||
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);
|
||||
let attemptsToday = todayAttempts(db, campaignId);
|
||||
|
||||
while (attemptsToday < campaign.limite_diario_atual) {
|
||||
const row = db.prepare(
|
||||
`SELECT * FROM outreach_campaign_queue
|
||||
WHERE campaign_id = ? AND status = 'approved'
|
||||
ORDER BY id ASC LIMIT 1`
|
||||
).get(campaignId);
|
||||
if (!row) break;
|
||||
|
||||
if (isOptedOut(db, row.numero)) {
|
||||
db.prepare(
|
||||
`UPDATE outreach_campaign_queue SET status = 'cancelled', cancel_reason = 'opt_out', updated_at = datetime('now') WHERE id = ?`
|
||||
).run(row.id);
|
||||
summary.pulados += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
const jaEnviado = db.prepare(
|
||||
`SELECT 1 FROM outreach_sends WHERE produto = ? AND status = 'enviado'
|
||||
AND (lead_id = ? OR numero = ?) LIMIT 1`
|
||||
).get(campaign.produto, row.lead_id, row.numero);
|
||||
if (jaEnviado) {
|
||||
db.prepare(
|
||||
`UPDATE outreach_campaign_queue SET status = 'cancelled', cancel_reason = 'ja_contatado', updated_at = datetime('now') WHERE id = ?`
|
||||
).run(row.id);
|
||||
summary.pulados += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
const externalKey = `campanha-${campaignId}-${row.id}`;
|
||||
const insert = db.prepare(
|
||||
`INSERT INTO outreach_sends (lead_id, produto, numero, mensagem_enviada, status, erro, zpro_ticket_id, canal, origem)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, 'nao_oficial', 'automatico')`
|
||||
);
|
||||
let sendId;
|
||||
let queueStatus;
|
||||
try {
|
||||
const resposta = await client.sendText({ number: row.numero, message: row.rendered_message, externalKey });
|
||||
const ticketId = resposta?.data?.ticketId ? String(resposta.data.ticketId) : null;
|
||||
sendId = Number(insert.run(row.lead_id, campaign.produto, row.numero, row.rendered_message, 'enviado', null, ticketId).lastInsertRowid);
|
||||
queueStatus = 'sent';
|
||||
summary.enviados += 1;
|
||||
} catch (error) {
|
||||
sendId = Number(insert.run(row.lead_id, campaign.produto, row.numero, row.rendered_message, 'erro', String(error.message || error), null).lastInsertRowid);
|
||||
queueStatus = 'error';
|
||||
summary.erros += 1;
|
||||
log(`campanha #${campaignId}: envio falhou (fila #${row.id}): ${error.message}`);
|
||||
}
|
||||
db.prepare(
|
||||
`UPDATE outreach_campaign_queue SET status = ?, outreach_send_id = ?, updated_at = datetime('now') WHERE id = ?`
|
||||
).run(queueStatus, sendId, row.id);
|
||||
|
||||
attemptsToday += 1;
|
||||
const health = evaluateCampaignHealth(db, campaignId, now);
|
||||
if (health.pause) {
|
||||
log(`campanha #${campaignId}: pausada automaticamente (${health.reason})`);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
return summary;
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { openDb, createLeadInserter } 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';
|
||||
|
||||
function inserirLead(db, dedup, whatsapp) {
|
||||
const inserir = createLeadInserter(db);
|
||||
inserir({
|
||||
dedup_key: dedup, empresa: 'Empresa', legal_name: '', whatsapp, ddd: '11', email: '',
|
||||
cidade: '', uf: '', segmento: 'x', segmento_label: '', interesse: '', cnpj: '', cnae: '',
|
||||
endereco: '', website: '', pains_json: '[]', origem: 't', telefone_valido: 1, cnpj_valido: 0,
|
||||
empresa_valida: 1, score: 50, classificacao: 'bom', motivo_classificacao: '', score_qualidade: 50,
|
||||
score_comercial: 50, prioridade: 1, produto_principal: 'ERP', produtos_json: '[]',
|
||||
capture_run_id: null, captured_at: null, exported_at: null
|
||||
});
|
||||
}
|
||||
|
||||
function campanhaAprovada(db, quantidadeLeads = 1) {
|
||||
for (let i = 0; i < quantidadeLeads; i += 1) {
|
||||
inserirLead(db, `lead-${i}`, `551199999000${i}`);
|
||||
}
|
||||
const campanha = createCampaign(db, {
|
||||
nome: 'C', produto: 'ERP', segmentos: ['x'], variantes: [{ code: 'A', mensagem: 'Ola {empresa}' }],
|
||||
limiteDiarioInicial: 10
|
||||
});
|
||||
setCampaignStatus(db, campanha.id, 'rodando');
|
||||
generateCampaignQueue(db, campanha.id);
|
||||
const fila = listQueue(db, campanha.id, 'preview');
|
||||
approveQueueRows(db, campanha.id, fila.map((r) => r.id));
|
||||
return campanha;
|
||||
}
|
||||
|
||||
function fakeZproClient(overrides = {}) {
|
||||
const enviados = [];
|
||||
return {
|
||||
enviados,
|
||||
sendText: async (args) => { enviados.push(args); return { data: { ticketId: 999 } }; },
|
||||
...overrides
|
||||
};
|
||||
}
|
||||
|
||||
test('runCampaignQueue envia linhas aprovadas via sendText e marca sent', async () => {
|
||||
const db = openDb(':memory:');
|
||||
const campanha = campanhaAprovada(db);
|
||||
const zproClient = fakeZproClient();
|
||||
|
||||
const resumo = await runCampaignQueue({ db, zproClient });
|
||||
assert.equal(resumo.enviados, 1);
|
||||
assert.equal(zproClient.enviados.length, 1);
|
||||
assert.equal(zproClient.enviados[0].message, 'Ola Empresa');
|
||||
|
||||
const fila = listQueue(db, campanha.id, 'sent');
|
||||
assert.equal(fila.length, 1);
|
||||
assert.ok(fila[0].outreach_send_id);
|
||||
const send = db.prepare('SELECT * FROM outreach_sends WHERE id = ?').get(fila[0].outreach_send_id);
|
||||
assert.equal(send.status, 'enviado');
|
||||
assert.equal(send.origem, 'automatico');
|
||||
db.close();
|
||||
});
|
||||
|
||||
test('runCampaignQueue registra erro quando o Z-PRO falha, sem travar o restante', async () => {
|
||||
const db = openDb(':memory:');
|
||||
const campanha = campanhaAprovada(db, 2);
|
||||
let chamada = 0;
|
||||
const zproClient = fakeZproClient({
|
||||
sendText: async (args) => {
|
||||
chamada += 1;
|
||||
if (chamada === 1) throw new Error('Z-PRO falhou');
|
||||
return { data: { ticketId: 1 } };
|
||||
}
|
||||
});
|
||||
|
||||
const resumo = await runCampaignQueue({ db, zproClient });
|
||||
assert.equal(resumo.enviados, 1);
|
||||
assert.equal(resumo.erros, 1);
|
||||
const erro = listQueue(db, campanha.id, 'error')[0];
|
||||
const send = db.prepare('SELECT * FROM outreach_sends WHERE id = ?').get(erro.outreach_send_id);
|
||||
assert.equal(send.status, 'erro');
|
||||
db.close();
|
||||
});
|
||||
|
||||
test('runCampaignQueue respeita o limite diario da campanha', async () => {
|
||||
const db = openDb(':memory:');
|
||||
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' }],
|
||||
limiteDiarioInicial: 1
|
||||
});
|
||||
setCampaignStatus(db, campanha.id, 'rodando');
|
||||
generateCampaignQueue(db, campanha.id);
|
||||
approveQueueRows(db, campanha.id, listQueue(db, campanha.id, 'preview').map((r) => r.id));
|
||||
|
||||
const zproClient = fakeZproClient();
|
||||
const resumo = await runCampaignQueue({ db, zproClient });
|
||||
assert.equal(resumo.enviados, 1);
|
||||
assert.equal(zproClient.enviados.length, 1);
|
||||
assert.equal(listQueue(db, campanha.id, 'approved').length, 2);
|
||||
db.close();
|
||||
});
|
||||
|
||||
test('runCampaignQueue pula e cancela linha de numero que entrou em opt-out depois da fila gerada', async () => {
|
||||
const db = openDb(':memory:');
|
||||
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 });
|
||||
assert.equal(resumo.pulados, 1);
|
||||
assert.equal(zproClient.enviados.length, 0);
|
||||
const cancelada = listQueue(db, campanha.id, 'cancelled')[0];
|
||||
assert.equal(cancelada.cancel_reason, 'opt_out');
|
||||
db.close();
|
||||
});
|
||||
|
||||
test('runCampaignQueue so processa campanhas com status rodando', async () => {
|
||||
const db = openDb(':memory:');
|
||||
inserirLead(db, 'l1', '5511999990001');
|
||||
const campanha = createCampaign(db, {
|
||||
nome: 'C', produto: 'ERP', segmentos: ['x'], variantes: [{ code: 'A', mensagem: 'oi' }]
|
||||
});
|
||||
generateCampaignQueue(db, campanha.id);
|
||||
approveQueueRows(db, campanha.id, listQueue(db, campanha.id, 'preview').map((r) => r.id));
|
||||
// campanha continua em 'rascunho' — nunca foi iniciada
|
||||
|
||||
const zproClient = fakeZproClient();
|
||||
const resumo = await runCampaignQueue({ db, zproClient });
|
||||
assert.equal(resumo.enviados, 0);
|
||||
assert.equal(zproClient.enviados.length, 0);
|
||||
db.close();
|
||||
});
|
||||
|
||||
test('runCampaignQueue interrompe a campanha quando a saude aciona auto-pausa', async () => {
|
||||
const db = openDb(':memory:');
|
||||
const campanha = campanhaAprovada(db, 10);
|
||||
db.prepare(`UPDATE outreach_campaigns SET limite_diario_atual = 10 WHERE id = ?`).run(campanha.id);
|
||||
let chamada = 0;
|
||||
const zproClient = fakeZproClient({
|
||||
sendText: async () => {
|
||||
chamada += 1;
|
||||
if (chamada <= 8) return { data: { ticketId: chamada } };
|
||||
throw new Error('falha simulada');
|
||||
}
|
||||
});
|
||||
|
||||
await runCampaignQueue({ db, zproClient });
|
||||
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();
|
||||
});
|
||||
Reference in New Issue
Block a user