diff --git a/src/outreach/campaign-worker.mjs b/src/outreach/campaign-worker.mjs new file mode 100644 index 0000000..0794f4f --- /dev/null +++ b/src/outreach/campaign-worker.mjs @@ -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; +} diff --git a/tests/campaign-worker.test.mjs b/tests/campaign-worker.test.mjs new file mode 100644 index 0000000..c769003 --- /dev/null +++ b/tests/campaign-worker.test.mjs @@ -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(); +});