import test from 'node:test'; import assert from 'node:assert/strict'; import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; import { Worker } from 'node:worker_threads'; import { addOptOut, addSharedNumberOverride, openDb, recoverPendingOutreach, removeSharedNumberOverride, reserveOutreachSend, saveWhatsAppVerification, setSetting } from '../src/db.mjs'; import { seedCampaigns } from '../src/outreach/campaign-store.mjs'; async function workerModule() { return import('../src/outreach/campaign-worker.mjs'); } function setup() { const db = openDb(':memory:'); seedCampaigns(db); const campaign = db.prepare('SELECT * FROM outreach_campaigns ORDER BY id LIMIT 1').get(); db.prepare("UPDATE outreach_campaigns SET status = 'running' WHERE id = ?").run(campaign.id); return { db, campaign }; } function addQueue(db, campaign, { suffix, status = 'scheduled', scheduledAt = '2026-07-29T12:00:00.000Z', confirmed = true }) { const numero = `55119999${String(suffix).padStart(5, '0')}`; const leadId = Number(db.prepare( `INSERT INTO leads (dedup_key, empresa, whatsapp, segmento, telefone_valido) VALUES (?, ?, ?, 'restaurantes', 1)` ).run(`worker-${suffix}`, `Empresa ${suffix}`, numero).lastInsertRowid); if (confirmed) saveWhatsAppVerification(db, { numero, status: 'confirmado' }); const variantId = db.prepare( 'SELECT id FROM outreach_campaign_variants WHERE campaign_id = ? ORDER BY id LIMIT 1' ).get(campaign.id).id; const queueId = Number(db.prepare( `INSERT INTO outreach_campaign_queue (campaign_id, lead_id, variant_id, produto, numero, segmento, rendered_message, selection_reason, status, scheduled_at) VALUES (?, ?, ?, ?, ?, 'restaurantes', ?, 'teste', ?, ?)` ).run( campaign.id, leadId, variantId, campaign.produto, numero, `Mensagem ${suffix}`, status, scheduledAt ).lastInsertRowid); return { queueId, leadId, numero }; } function whatsappFake({ state = 'READY', beforeProvider = async () => {}, outcome = 'success' } = {}) { let providerCalls = 0; return { get providerCalls() { return providerCalls; }, getStatus: () => ({ state }), sendText: async (args) => { await beforeProvider(args); const context = await args.beforeSend(); providerCalls += 1; if (outcome === 'success') { const result = { messageId: `message-${providerCalls}` }; await args.afterSend(result, null, context); return result; } const error = Object.assign(new Error(outcome), { deliveryUnknown: outcome === 'unknown' }); await args.afterSend(null, error, context); throw error; } }; } test('worker processa somente o primeiro item scheduled vencido', async () => { const { runDueCampaignQueue } = await workerModule(); assert.equal(typeof runDueCampaignQueue, 'function'); const { db, campaign } = setup(); const first = addQueue(db, campaign, { suffix: 1 }); const second = addQueue(db, campaign, { suffix: 2 }); const future = addQueue(db, campaign, { suffix: 3, scheduledAt: '2026-07-29T12:01:00.000Z' }); const approved = addQueue(db, campaign, { suffix: 4, status: 'approved', scheduledAt: null }); const whatsapp = whatsappFake(); const summary = await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.deepEqual(summary, { tentados: 1, enviados: 1, erros: 0, pulados: 0 }); assert.equal(whatsapp.providerCalls, 1); const statuses = new Map(db.prepare( 'SELECT id, status FROM outreach_campaign_queue ORDER BY id' ).all().map((row) => [row.id, row.status])); assert.equal(statuses.get(first.queueId), 'sent'); assert.equal(statuses.get(second.queueId), 'scheduled'); assert.equal(statuses.get(future.queueId), 'scheduled'); assert.equal(statuses.get(approved.queueId), 'approved'); db.close(); }); test('worker cancela item permanentemente inelegivel e continua no proximo vencido', async () => { const { runDueCampaignQueue } = await workerModule(); const { db, campaign } = setup(); const poisoned = addQueue(db, campaign, { suffix: 5, confirmed: false }); const eligible = addQueue(db, campaign, { suffix: 6 }); const whatsapp = whatsappFake(); const summary = await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.deepEqual(summary, { tentados: 2, enviados: 1, erros: 0, pulados: 1 }); assert.deepEqual( { ...db.prepare('SELECT status, cancel_reason FROM outreach_campaign_queue WHERE id = ?').get(poisoned.queueId) }, { status: 'cancelled', cancel_reason: 'whatsapp_not_confirmed' } ); assert.equal(db.prepare('SELECT status FROM outreach_campaign_queue WHERE id = ?').get(eligible.queueId).status, 'sent'); assert.equal(whatsapp.providerCalls, 1); db.close(); }); test('worker mantem itens agendados e para no primeiro bloqueio transitorio de limite', async () => { const { runDueCampaignQueue } = await workerModule(); const { db, campaign } = setup(); const first = addQueue(db, campaign, { suffix: 7 }); const second = addQueue(db, campaign, { suffix: 8 }); db.prepare('UPDATE outreach_campaigns SET hourly_limit = 0 WHERE id = ?').run(campaign.id); const whatsapp = whatsappFake(); const summary = await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.deepEqual(summary, { tentados: 1, enviados: 0, erros: 1, pulados: 0 }); assert.deepEqual( db.prepare('SELECT status FROM outreach_campaign_queue WHERE id IN (?, ?) ORDER BY id').all(first.queueId, second.queueId).map((row) => row.status), ['scheduled', 'scheduled'] ); assert.equal(whatsapp.providerCalls, 0); db.close(); }); test('worker mantem dois itens agendados e para quando conexao cai antes da reserva', async () => { const { runDueCampaignQueue } = await workerModule(); const { db, campaign } = setup(); const first = addQueue(db, campaign, { suffix: 12 }); const second = addQueue(db, campaign, { suffix: 13 }); let state = 'READY'; let providerCalls = 0; const whatsapp = { getStatus: () => ({ state }), sendText: async (args) => { state = 'DISCONNECTED'; await args.beforeSend(); providerCalls += 1; } }; const summary = await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.deepEqual(summary, { tentados: 1, enviados: 0, erros: 1, pulados: 0 }); assert.deepEqual( db.prepare('SELECT status FROM outreach_campaign_queue WHERE id IN (?, ?) ORDER BY id').all(first.queueId, second.queueId).map((row) => row.status), ['scheduled', 'scheduled'] ); assert.equal(providerCalls, 0); db.close(); }); test('worker cancela numero rejeitado no lookup antes de beforeSend e continua', async () => { const { runDueCampaignQueue } = await workerModule(); const { db, campaign } = setup(); const rejected = addQueue(db, campaign, { suffix: 9 }); const eligible = addQueue(db, campaign, { suffix: 11 }); let calls = 0; const whatsapp = { getStatus: () => ({ state: 'READY' }), sendText: async (args) => { calls += 1; if (calls === 1) { throw Object.assign(new Error('numero nao cadastrado'), { status: 'nao_cadastrado', registered: false }); } const context = await args.beforeSend(); const result = { messageId: 'second-message' }; await args.afterSend(result, null, context); return result; } }; const summary = await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.deepEqual(summary, { tentados: 2, enviados: 1, erros: 0, pulados: 1 }); assert.deepEqual( { ...db.prepare('SELECT status, cancel_reason FROM outreach_campaign_queue WHERE id = ?').get(rejected.queueId) }, { status: 'cancelled', cancel_reason: 'whatsapp_not_confirmed' } ); assert.equal(db.prepare('SELECT status FROM outreach_campaign_queue WHERE id = ?').get(eligible.queueId).status, 'sent'); db.close(); }); test('reserva atomica vincula fila e send sem consumir limite manual', async () => { const { reserveCampaignQueueSend } = await workerModule(); const { db, campaign } = setup(); const item = addQueue(db, campaign, { suffix: 10 }); setSetting(db, 'outreach.mensagensPorDia', '1'); reserveOutreachSend(db, { numero: '5511999999999', mensagem: 'manual' }); setSetting(db, 'outreach.ativo', 'true'); assert.throws( () => reserveOutreachSend(db, { leadId: item.leadId, produto: campaign.produto, numero: item.numero, mensagem: 'atalho automatico', origem: 'automatico' }), /fila supervisionada/i ); const reserved = reserveCampaignQueueSend(db, { queueId: item.queueId, now: new Date('2026-07-29T12:00:00.000Z') }); assert.deepEqual(reserved, { queueId: item.queueId, sendId: 2 }); assert.deepEqual( { ...db.prepare('SELECT status, outreach_send_id FROM outreach_campaign_queue WHERE id = ?').get(item.queueId) }, { status: 'reserved', outreach_send_id: 2 } ); assert.deepEqual( { ...db.prepare('SELECT lead_id, produto, numero, mensagem_enviada, status, origem FROM outreach_sends WHERE id = 2').get() }, { lead_id: item.leadId, produto: campaign.produto, numero: item.numero, mensagem_enviada: 'Mensagem 10', status: 'pendente', origem: 'automatico' } ); db.close(); }); test('reserva aplica limite diario da campanha e quatro tentativas na hora rolante', async () => { const { reserveCampaignQueueSend } = await workerModule(); const { db, campaign } = setup(); const due = addQueue(db, campaign, { suffix: 20 }); db.prepare('UPDATE outreach_campaigns SET approved_daily_limit = 10, hourly_limit = 4 WHERE id = ?').run(campaign.id); for (let index = 0; index < 4; index += 1) { const old = addQueue(db, campaign, { suffix: 21 + index }); const sendId = Number(db.prepare( `INSERT INTO outreach_sends (lead_id, produto, numero, mensagem_enviada, status, origem, sent_at) VALUES (?, ?, ?, 'anterior', 'erro', 'automatico', ?)` ).run(old.leadId, campaign.produto, old.numero, `2026-07-29 11:${String(10 + index).padStart(2, '0')}:00`).lastInsertRowid); db.prepare("UPDATE outreach_campaign_queue SET status = 'reserved', outreach_send_id = ? WHERE id = ?").run(sendId, old.queueId); db.prepare("UPDATE outreach_campaign_queue SET status = 'error' WHERE id = ?").run(old.queueId); } assert.throws( () => reserveCampaignQueueSend(db, { queueId: due.queueId, now: new Date('2026-07-29T12:00:00.000Z') }), /limite.*hora/i ); assert.equal(db.prepare('SELECT status FROM outreach_campaign_queue WHERE id = ?').get(due.queueId).status, 'scheduled'); db.prepare("UPDATE outreach_sends SET sent_at = '2026-07-29 10:59:59'").run(); db.prepare('UPDATE outreach_campaigns SET approved_daily_limit = 10 WHERE id = ?').run(campaign.id); for (let index = 0; index < 6; index += 1) { const old = addQueue(db, campaign, { suffix: 30 + index }); const sendId = Number(db.prepare( `INSERT INTO outreach_sends (lead_id, produto, numero, mensagem_enviada, status, origem, sent_at) VALUES (?, ?, ?, 'anterior', 'erro', 'automatico', '2026-07-29 09:00:00')` ).run(old.leadId, campaign.produto, old.numero).lastInsertRowid); db.prepare("UPDATE outreach_campaign_queue SET status = 'reserved', outreach_send_id = ? WHERE id = ?").run(sendId, old.queueId); db.prepare("UPDATE outreach_campaign_queue SET status = 'error' WHERE id = ?").run(old.queueId); } assert.throws( () => reserveCampaignQueueSend(db, { queueId: due.queueId, now: new Date('2026-07-29T12:00:00.000Z') }), /limite diario/i ); db.close(); }); test('duas conexoes nao reservam a mesma fila nem ultrapassam o limite diario', async () => { const { reserveCampaignQueueSend } = await workerModule(); const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'campaign-worker-')); const dbPath = path.join(dir, 'worker.sqlite'); const db = openDb(dbPath); const { campaign } = (() => { seedCampaigns(db); const seeded = db.prepare('SELECT * FROM outreach_campaigns ORDER BY id LIMIT 1').get(); db.prepare("UPDATE outreach_campaigns SET status = 'running', approved_daily_limit = 10 WHERE id = ?").run(seeded.id); return { campaign: seeded }; })(); const first = addQueue(db, campaign, { suffix: 70 }); const second = addQueue(db, campaign, { suffix: 71 }); const other = openDb(dbPath); const sameQueue = await Promise.allSettled([ Promise.resolve().then(() => reserveCampaignQueueSend(db, { queueId: first.queueId, now: new Date('2026-07-29T12:00:00.000Z') })), Promise.resolve().then(() => reserveCampaignQueueSend(other, { queueId: first.queueId, now: new Date('2026-07-29T12:00:00.000Z') })) ]); assert.deepEqual(sameQueue.map((result) => result.status).sort(), ['fulfilled', 'rejected']); assert.equal(db.prepare('SELECT COUNT(*) AS total FROM outreach_sends').get().total, 1); db.prepare('UPDATE outreach_campaigns SET approved_daily_limit = 10 WHERE id = ?').run(campaign.id); for (let index = 0; index < 9; index += 1) { const item = addQueue(db, campaign, { suffix: 80 + index }); const sendId = Number(db.prepare( `INSERT INTO outreach_sends (lead_id, produto, numero, mensagem_enviada, status, origem, sent_at) VALUES (?, ?, ?, 'anterior', 'erro', 'automatico', '2026-07-29T08:00:00.000Z')` ).run(item.leadId, campaign.produto, item.numero).lastInsertRowid); db.prepare("UPDATE outreach_campaign_queue SET status = 'reserved', outreach_send_id = ? WHERE id = ?").run(sendId, item.queueId); db.prepare("UPDATE outreach_campaign_queue SET status = 'error' WHERE id = ?").run(item.queueId); } assert.throws( () => reserveCampaignQueueSend(other, { queueId: second.queueId, now: new Date('2026-07-29T12:00:00.000Z') }), /limite diario/i ); assert.equal(db.prepare('SELECT COUNT(*) AS total FROM outreach_sends').get().total, 10); assert.equal(db.prepare('SELECT status FROM outreach_campaign_queue WHERE id = ?').get(second.queueId).status, 'scheduled'); other.close(); db.close(); fs.rmSync(dir, { recursive: true, force: true }); }); test('segunda conexao aguarda lock e reconta o limite apos commit concorrente', async () => { const { reserveCampaignQueueSend } = await workerModule(); const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'campaign-cap-lock-')); const dbPath = path.join(dir, 'worker.sqlite'); const db = openDb(dbPath); seedCampaigns(db); const campaign = db.prepare('SELECT * FROM outreach_campaigns ORDER BY id LIMIT 1').get(); db.prepare("UPDATE outreach_campaigns SET status = 'running', approved_daily_limit = 10 WHERE id = ?").run(campaign.id); const committing = addQueue(db, campaign, { suffix: 90 }); const blocked = addQueue(db, campaign, { suffix: 91 }); for (let index = 0; index < 9; index += 1) { const item = addQueue(db, campaign, { suffix: 920 + index }); const sendId = Number(db.prepare( `INSERT INTO outreach_sends (lead_id, produto, numero, mensagem_enviada, status, origem, sent_at) VALUES (?, ?, ?, 'anterior', 'erro', 'automatico', '2026-07-29T08:00:00.000Z')` ).run(item.leadId, campaign.produto, item.numero).lastInsertRowid); db.prepare("UPDATE outreach_campaign_queue SET status = 'reserved', outreach_send_id = ? WHERE id = ?").run(sendId, item.queueId); db.prepare("UPDATE outreach_campaign_queue SET status = 'error' WHERE id = ?").run(item.queueId); } const other = openDb(dbPath); const worker = new Worker(` const { parentPort, workerData } = require('node:worker_threads'); const { DatabaseSync } = require('node:sqlite'); const db = new DatabaseSync(workerData.dbPath); db.exec('PRAGMA busy_timeout = 5000; BEGIN IMMEDIATE;'); const inserted = db.prepare( "INSERT INTO outreach_sends (lead_id, produto, numero, mensagem_enviada, status, canal, origem, sent_at) VALUES (?, ?, ?, 'concorrente', 'pendente', 'local', 'automatico', '2026-07-29T12:00:00.000Z')" ).run(workerData.leadId, workerData.produto, workerData.numero); db.prepare("UPDATE outreach_campaign_queue SET status = 'reserved', outreach_send_id = ? WHERE id = ?") .run(Number(inserted.lastInsertRowid), workerData.queueId); parentPort.postMessage('locked'); setTimeout(() => { db.exec('COMMIT'); db.close(); parentPort.postMessage('committed'); }, 150); `, { eval: true, workerData: { dbPath, queueId: committing.queueId, leadId: committing.leadId, numero: committing.numero, produto: campaign.produto } }); const workerExit = new Promise((resolve, reject) => { worker.once('exit', (code) => code === 0 ? resolve() : reject(new Error(`worker encerrou com ${code}`))); worker.once('error', reject); }); try { await new Promise((resolve, reject) => { worker.once('message', resolve); worker.once('error', reject); }); const startedAt = Date.now(); assert.throws( () => reserveCampaignQueueSend(other, { queueId: blocked.queueId, now: new Date('2026-07-29T12:00:00.000Z') }), /limite diario/i ); assert.ok(Date.now() - startedAt >= 75, 'segunda conexao nao aguardou o lock concorrente'); assert.equal(db.prepare('SELECT COUNT(*) AS total FROM outreach_sends').get().total, 10); assert.equal(db.prepare('SELECT status FROM outreach_campaign_queue WHERE id = ?').get(blocked.queueId).status, 'scheduled'); await workerExit; } finally { await worker.terminate(); other.close(); db.close(); fs.rmSync(dir, { recursive: true, force: true }); } }); test('beforeSend revalida campanha, READY, confirmacao, opt-out, resposta e dedupe', async () => { const { runDueCampaignQueue } = await workerModule(); const cases = [ ['campanha pausada', null, ({ db, campaign }) => db.prepare("UPDATE outreach_campaigns SET status = 'paused' WHERE id = ?").run(campaign.id)], ['fila deixou de estar agendada', null, ({ db, item }) => db.prepare("UPDATE outreach_campaign_queue SET status = 'cancelled' WHERE id = ?").run(item.queueId)], ['numero deixou de estar confirmado', 'whatsapp_not_confirmed', ({ db, item }) => saveWhatsAppVerification(db, { numero: item.numero, status: 'nao_cadastrado' })], ['opt-out equivalente', 'opt_out', ({ db, item }) => addOptOut(db, item.numero, 'pare antes do envio')], ['resposta inbound', 'inbound_reply', ({ db, item }) => db.prepare( 'INSERT INTO outreach_inbound_replies (provider_message_id, lead_id, numero, received_at) VALUES (?, ?, ?, ?)' ).run('inbound-1', item.leadId, item.numero, '2026-07-29T11:59:00.000Z')], ['envio ativo do mesmo produto', 'duplicate_product_send', ({ db, campaign, item }) => db.prepare( `INSERT INTO outreach_sends (lead_id, produto, numero, mensagem_enviada, status, origem) VALUES (?, ?, ?, 'duplicado', 'desconhecido', 'automatico')` ).run(item.leadId, campaign.produto, item.numero)], ['numero virou compartilhado', 'shared_number', ({ db, item }) => db.prepare( `INSERT INTO leads (dedup_key, empresa, whatsapp, segmento, telefone_valido) VALUES ('worker-compartilhado', 'Outro lead com o mesmo telefone', ?, 'restaurantes', 1)` ).run(item.numero)] ]; for (const [name, cancelReason, mutate] of cases) { const { db, campaign } = setup(); const item = addQueue(db, campaign, { suffix: 100 + cases.findIndex(([candidate]) => candidate === name) }); let state = 'READY'; const whatsapp = whatsappFake({ beforeProvider: async () => mutate({ db, campaign, item }), get state() { return state; } }); if (name === 'campanha pausada') state = 'READY'; const summary = await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.equal(whatsapp.providerCalls, 0, name); assert.equal(summary.enviados, 0, name); assert.equal(db.prepare('SELECT COUNT(*) total FROM outreach_sends').get().total, name === 'envio ativo do mesmo produto' ? 1 : 0, name); if (cancelReason) { assert.deepEqual( { ...db.prepare('SELECT status, cancel_reason FROM outreach_campaign_queue WHERE id = ?').get(item.queueId) }, { status: 'cancelled', cancel_reason: cancelReason }, name ); } db.close(); } const { db, campaign } = setup(); addQueue(db, campaign, { suffix: 199 }); let state = 'READY'; const whatsapp = { getStatus: () => ({ state }), sendText: async (args) => { state = 'DISCONNECTED'; await args.beforeSend(); throw new Error('provider nao deveria ser chamado'); } }; await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.equal(db.prepare('SELECT COUNT(*) total FROM outreach_sends').get().total, 0); db.close(); }); // Cenario real do operador: numero compartilhado liberado em Configuracoes, // item entra na fila e e agendado, operador reconsidera e clica "Bloquear de // novo". Sem a revalidacao de ultima milha o envio ja agendado sairia mesmo // assim, porque a checagem de compartilhamento so acontecia na selecao. test('libera, agenda e rebloqueia numero compartilhado cancela o envio ja agendado', async () => { const { runDueCampaignQueue } = await workerModule(); for (const [name, reblock, expected] of [ ['override mantido', false, { tentados: 1, enviados: 1, erros: 0, pulados: 0 }], ['override removido antes do envio', true, { tentados: 1, enviados: 0, erros: 0, pulados: 1 }] ]) { const { db, campaign } = setup(); const item = addQueue(db, campaign, { suffix: 600 }); db.prepare( `INSERT INTO leads (dedup_key, empresa, whatsapp, segmento, telefone_valido) VALUES ('worker-600-b', 'Cliente da mesma contabilidade', ?, 'restaurantes', 1)` ).run(item.numero); addSharedNumberOverride(db, item.numero, 'contabilidade confirmada'); if (reblock) assert.equal(removeSharedNumberOverride(db, item.numero), true, name); const whatsapp = whatsappFake(); const summary = await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.deepEqual(summary, expected, name); assert.equal(whatsapp.providerCalls, reblock ? 0 : 1, name); assert.deepEqual( { ...db.prepare('SELECT status, cancel_reason FROM outreach_campaign_queue WHERE id = ?').get(item.queueId) }, reblock ? { status: 'cancelled', cancel_reason: 'shared_number' } : { status: 'sent', cancel_reason: null }, name ); assert.equal(db.prepare('SELECT COUNT(*) total FROM outreach_sends').get().total, reblock ? 0 : 1, name); db.close(); } }); // Par assimetrico: 557533334444 (local '33334444', fora da faixa 6-9) e // 5575933334444. equivalentPhoneNumbers so relaciona o segundo com o primeiro, // entao antes da chave canonica o lead de 12 digitos passava batido aqui. test('ultima milha bloqueia os dois lados de um par 12/13 digitos assimetrico', async () => { const { runDueCampaignQueue } = await workerModule(); for (const [queued, outro] of [ ['557533334444', '5575933334444'], ['5575933334444', '557533334444'] ]) { const { db, campaign } = setup(); const leadId = Number(db.prepare( `INSERT INTO leads (dedup_key, empresa, whatsapp, segmento, telefone_valido) VALUES ('assimetrico-fila', 'Lead na fila', ?, 'restaurantes', 1)` ).run(queued).lastInsertRowid); db.prepare( `INSERT INTO leads (dedup_key, empresa, whatsapp, segmento, telefone_valido) VALUES ('assimetrico-outro', 'Lead com o mesmo telefone', ?, 'restaurantes', 1)` ).run(outro); saveWhatsAppVerification(db, { numero: queued, status: 'confirmado' }); const variantId = db.prepare( 'SELECT id FROM outreach_campaign_variants WHERE campaign_id = ? ORDER BY id LIMIT 1' ).get(campaign.id).id; const queueId = Number(db.prepare( `INSERT INTO outreach_campaign_queue (campaign_id, lead_id, variant_id, produto, numero, segmento, rendered_message, selection_reason, status, scheduled_at) VALUES (?, ?, ?, ?, ?, 'restaurantes', 'Mensagem', 'teste', 'scheduled', '2026-07-29T12:00:00.000Z')` ).run(campaign.id, leadId, variantId, campaign.produto, queued).lastInsertRowid); const whatsapp = whatsappFake(); const summary = await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.deepEqual(summary, { tentados: 1, enviados: 0, erros: 0, pulados: 1 }, queued); assert.equal(whatsapp.providerCalls, 0, queued); assert.deepEqual( { ...db.prepare('SELECT status, cancel_reason FROM outreach_campaign_queue WHERE id = ?').get(queueId) }, { status: 'cancelled', cancel_reason: 'shared_number' }, queued ); db.close(); } }); // Mesmo par assimetrico, agora para opt-out: addOptOut/isOptedOut usavam // equivalentPhoneNumbers diretamente, entao um opt-out registrado numa forma // do numero podia nao bloquear o envio quando a fila tinha a outra forma. test('ultima milha bloqueia opt-out registrado na outra forma de um par 12/13 digitos assimetrico', async () => { const { runDueCampaignQueue } = await workerModule(); for (const [queued, optedOutForm] of [ ['557533334444', '5575933334444'], ['5575933334444', '557533334444'] ]) { const { db, campaign } = setup(); const leadId = Number(db.prepare( `INSERT INTO leads (dedup_key, empresa, whatsapp, segmento, telefone_valido) VALUES ('assimetrico-optout', 'Lead na fila', ?, 'restaurantes', 1)` ).run(queued).lastInsertRowid); saveWhatsAppVerification(db, { numero: queued, status: 'confirmado' }); const variantId = db.prepare( 'SELECT id FROM outreach_campaign_variants WHERE campaign_id = ? ORDER BY id LIMIT 1' ).get(campaign.id).id; const queueId = Number(db.prepare( `INSERT INTO outreach_campaign_queue (campaign_id, lead_id, variant_id, produto, numero, segmento, rendered_message, selection_reason, status, scheduled_at) VALUES (?, ?, ?, ?, ?, 'restaurantes', 'Mensagem', 'teste', 'scheduled', '2026-07-29T12:00:00.000Z')` ).run(campaign.id, leadId, variantId, campaign.produto, queued).lastInsertRowid); addOptOut(db, optedOutForm, 'pediu para nao receber mais'); const whatsapp = whatsappFake(); const summary = await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.equal(summary.enviados, 0, queued); assert.equal(whatsapp.providerCalls, 0, queued); assert.deepEqual( { ...db.prepare('SELECT status, cancel_reason FROM outreach_campaign_queue WHERE id = ?').get(queueId) }, { status: 'cancelled', cancel_reason: 'opt_out' }, queued ); db.close(); } }); test('afterSend finaliza fila e historico uma unica vez por resultado', async () => { const { runDueCampaignQueue } = await workerModule(); for (const [outcome, queueStatus, sendStatus] of [ ['success', 'sent', 'aceito'], ['unknown', 'unknown', 'desconhecido'], ['rejected', 'error', 'erro'] ]) { const { db, campaign } = setup(); const item = addQueue(db, campaign, { suffix: `30${outcome.length}` }); db.exec(` CREATE TABLE finalize_audit (queue_status TEXT); CREATE TRIGGER audit_queue_finalize AFTER UPDATE OF status ON outreach_campaign_queue WHEN OLD.status = 'reserved' BEGIN INSERT INTO finalize_audit VALUES (NEW.status); END; `); const whatsapp = whatsappFake({ outcome }); await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.equal(db.prepare('SELECT status FROM outreach_campaign_queue WHERE id = ?').get(item.queueId).status, queueStatus); assert.equal(db.prepare('SELECT status FROM outreach_sends').get().status, sendStatus); assert.deepEqual(db.prepare('SELECT queue_status FROM finalize_audit').all().map((row) => row.queue_status), [queueStatus]); db.close(); } }); test('resposta do provedor sem messageId fica unknown e nunca tenta a mesma fila novamente', async () => { const { runDueCampaignQueue } = await workerModule(); const { db, campaign } = setup(); const item = addQueue(db, campaign, { suffix: 399 }); let providerCalls = 0; const whatsapp = { getStatus: () => ({ state: 'READY' }), sendText: async (args) => { const context = await args.beforeSend(); providerCalls += 1; const error = Object.assign(new Error('WhatsApp nao retornou o identificador da mensagem'), { code: 'WHATSAPP_MESSAGE_ID_MISSING', deliveryUnknown: true }); await args.afterSend(null, error, context); throw error; } }; await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); await runDueCampaignQueue({ db, whatsapp, now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.equal(providerCalls, 1); assert.equal(db.prepare('SELECT status FROM outreach_campaign_queue WHERE id = ?').get(item.queueId).status, 'unknown'); assert.equal(db.prepare('SELECT status FROM outreach_sends').get().status, 'desconhecido'); db.close(); }); test('restart pausa campanhas, desliga legado, preserva scheduled e recupera reserved pelo send', async () => { const { pauseCampaignsAfterRestart, recoverReservedCampaignQueue } = await workerModule(); const { db, campaign } = setup(); setSetting(db, 'outreach.ativo', 'true'); const scheduled = addQueue(db, campaign, { suffix: 401 }); const mappings = [ ['pendente', 'unknown'], ['aceito', 'unknown'], ['enviado', 'sent'], ['entregue', 'sent'], ['lido', 'sent'], ['desconhecido', 'unknown'], ['erro', 'error'] ]; const reserved = []; for (const [index, [sendStatus]] of mappings.entries()) { const item = addQueue(db, campaign, { suffix: 410 + index }); const sendId = Number(db.prepare( `INSERT INTO outreach_sends (lead_id, produto, numero, mensagem_enviada, status, origem) VALUES (?, ?, ?, 'reservada', ?, 'automatico')` ).run(item.leadId, campaign.produto, item.numero, sendStatus).lastInsertRowid); db.prepare("UPDATE outreach_campaign_queue SET status = 'reserved', outreach_send_id = ? WHERE id = ?").run(sendId, item.queueId); reserved.push(item); } assert.equal(pauseCampaignsAfterRestart(db), 1); assert.deepEqual( { ...db.prepare('SELECT status, pause_reason FROM outreach_campaigns WHERE id = ?').get(campaign.id) }, { status: 'paused', pause_reason: 'server_restart' } ); assert.equal(db.prepare("SELECT value FROM settings WHERE key = 'outreach.ativo'").get().value, 'false'); assert.equal(db.prepare('SELECT status FROM outreach_campaign_queue WHERE id = ?').get(scheduled.queueId).status, 'scheduled'); assert.equal(recoverReservedCampaignQueue(db), mappings.length); assert.deepEqual( reserved.map((item) => db.prepare('SELECT status FROM outreach_campaign_queue WHERE id = ?').get(item.queueId).status), mappings.map(([, queueStatus]) => queueStatus) ); db.close(); }); test('restart reconcilia fila sent quando recovery converte aceite sem ACK em desconhecido', async () => { const { recoverReservedCampaignQueue, runDueCampaignQueue } = await workerModule(); const { db, campaign } = setup(); const item = addQueue(db, campaign, { suffix: 499 }); await runDueCampaignQueue({ db, whatsapp: whatsappFake(), now: new Date('2026-07-29T12:00:00.000Z'), log: () => {} }); assert.deepEqual( { ...db.prepare( `SELECT queue.status AS queue_status, sends.status AS send_status FROM outreach_campaign_queue queue JOIN outreach_sends sends ON sends.id = queue.outreach_send_id WHERE queue.id = ?` ).get(item.queueId) }, { queue_status: 'sent', send_status: 'aceito' } ); assert.equal(recoverPendingOutreach(db), 1); assert.equal(recoverReservedCampaignQueue(db), 1); assert.deepEqual( { ...db.prepare( `SELECT queue.status AS queue_status, sends.status AS send_status FROM outreach_campaign_queue queue JOIN outreach_sends sends ON sends.id = queue.outreach_send_id WHERE queue.id = ?` ).get(item.queueId) }, { queue_status: 'unknown', send_status: 'desconhecido' } ); db.close(); });