COMERCIAL-GTI/tests/campaign-worker.test.mjs

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();
});