import { randomUUID } from 'node:crypto'; import { PrismaClient, CampaignStatus } from '@b2bcall/database'; import type Redis from 'ioredis'; import type { TelephonyProvider } from '@b2bcall/telephony'; import { normalizePhone } from '@b2bcall/shared'; import { CampaignLock } from './campaign-lock'; import { CpsLimiter } from './cps-limiter'; import { LeadRepository } from './lead-repository'; import { StatsStore } from './stats-store'; import { isWithinSchedule } from './schedule'; import { getLiveCounts } from './live-counts'; import { adjustPacingFactor, calculateCallsToOriginate, updateEwma, type PacingLimits } from './predictive-engine'; import { calculateNextAttemptAt, leadStatusForOutcome, type HangupOutcome } from './retry-rules'; import { DEFAULT_SIMULATION_PROFILE, randomInRange, simulateHangupCause } from './simulation'; import { claimAvailableAgent, releaseAgentAfterCall } from './agent-call-binding'; import { logger } from './logger'; const WORKER_ID = randomUUID(); const DIALER_SIMULATION = process.env.DIALER_SIMULATION === 'true'; export class CampaignWorker { private readonly leadRepo: LeadRepository; private readonly cpsLimiter: CpsLimiter; private readonly statsStore: StatsStore; constructor( private readonly prisma: PrismaClient, private readonly redis: Redis, private readonly telephony: TelephonyProvider, ) { this.leadRepo = new LeadRepository(prisma); this.cpsLimiter = new CpsLimiter(redis); this.statsStore = new StatsStore(redis); } async tick(campaignId: string): Promise { const campaign = await this.prisma.campaign.findUnique({ where: { id: campaignId } }); if (!campaign) return; const lock = new CampaignLock(this.redis, campaignId); const acquired = campaign.status === CampaignStatus.RUNNING ? await lock.acquire() : false; try { if (campaign.status === CampaignStatus.RUNNING && !acquired) { return; // outro worker já controla esta campanha (seção 75) } await this.leadRepo.releaseExpiredReservations(); if (campaign.status !== CampaignStatus.RUNNING) { return; // PAUSED/DRAINING/STOPPED/etc. — nunca origina, mas não mexe no que já está em andamento } if (!isWithinSchedule(campaign)) { logger.debug({ campaignId }, 'Fora da janela de horário (WAITING_SCHEDULE)'); return; } const hasWork = await this.leadRepo.hasRemainingWork(campaignId); if (!hasWork) { await this.prisma.campaign.update({ where: { id: campaignId }, data: { status: CampaignStatus.COMPLETED } }); logger.info({ campaignId }, 'Campanha concluída — sem leads restantes'); return; } await this.leadRepo.promoteNewLeads(campaignId); const trunk = await this.prisma.trunk.findUnique({ where: { id: campaign.trunkId } }); if (!trunk) { logger.error({ campaignId }, 'Tronco da campanha não existe mais'); return; } const limits: PacingLimits = { pacingMin: campaign.pacingMin, pacingMax: campaign.pacingMax, targetAbandonRate: campaign.targetAbandonRate, maxConcurrentCalls: campaign.maxConcurrentCalls, }; const stats = await this.statsStore.load(campaignId, campaign.pacingInitial); const counts = await getLiveCounts(this.prisma, campaignId, campaign.queueId, stats.avgTalkTimeSeconds); const hasActivity = counts.dialingCalls + counts.ringingCalls + counts.connectedWaitingAgent > 0; stats.pacingFactor = adjustPacingFactor(stats, limits, hasActivity); const callsToMake = calculateCallsToOriginate(stats, counts, limits); await this.statsStore.save(campaignId, stats); for (let i = 0; i < callsToMake; i++) { const allowed = await this.cpsLimiter.tryAcquire(campaignId, trunk.id, campaign.maxCps, trunk.maxCps); if (!allowed) break; // limite de CPS atingido neste tick const lead = await this.leadRepo.reserveNextLead(campaignId, WORKER_ID); if (!lead) break; // sem leads prontos agora const normalized = normalizePhone(lead.phone); const isSuppressed = normalized.normalized ? await this.prisma.suppressionEntry.findUnique({ where: { phoneNormalized: normalized.normalized } }) : null; if (isSuppressed) { await this.prisma.lead.update({ where: { id: lead.id }, data: { status: 'DO_NOT_CALL' } }); continue; } await this.originateAttempt(campaign, trunk.name, lead); } } finally { if (acquired) await lock.release(); } } private async originateAttempt( campaign: { id: string; queueId: string; callerId: string | null; context: string; ringTimeoutSeconds: number; maxWaitForAgentSeconds: number; wrapUpTimeSeconds: number; retryRules: unknown; maxAttempts: number; }, trunkName: string, lead: { id: string; phone: string; phoneNormalized: string; attemptCount: number }, ): Promise { const attempt = await this.prisma.dialAttempt.create({ data: { leadId: lead.id, campaignId: campaign.id, state: 'ORIGINATING', calledNumber: lead.phoneNormalized, callerIdUsed: campaign.callerId, }, }); await this.prisma.lead.update({ where: { id: lead.id }, data: { status: 'DIALING', attemptCount: { increment: 1 }, lastAttemptAt: new Date() }, }); if (DIALER_SIMULATION) { this.simulateAttempt(campaign, lead, attempt.id); return; } try { await this.telephony.originate({ channel: `PJSIP/${lead.phoneNormalized}@${trunkName}`, context: campaign.context, exten: 's', priority: 1, callerId: campaign.callerId ?? undefined, timeoutMs: campaign.ringTimeoutSeconds * 1000, variables: { B2BCALL_ATTEMPT_ID: attempt.id }, }); // A resolução final (atendida/ocupada/sem resposta) chega de forma // assíncrona via eventos AMI, processados por // apps/asterisk-events + reconciliação (Fase 7). Aqui só garantimos // que a tentativa não fique presa para sempre se nenhum evento // chegar (rede instável, worker reiniciado, etc.). setTimeout( () => void this.failIfStillPending(attempt.id, campaign.retryRules as Record, campaign.maxAttempts), (campaign.ringTimeoutSeconds + 30) * 1000, ); } catch (err) { logger.error({ err, attemptId: attempt.id }, 'Falha ao originar chamada'); await this.finalizeAttempt(attempt.id, campaign.id, lead.id, 'FAILED', campaign.retryRules as Record, campaign.maxAttempts, lead.attemptCount + 1); } } private async failIfStillPending(attemptId: string, retryRules: Record, maxAttempts: number): Promise { const attempt = await this.prisma.dialAttempt.findUnique({ where: { id: attemptId } }); if (!attempt || attempt.endedAt) return; // já resolvida por um evento real const lead = await this.prisma.lead.findUnique({ where: { id: attempt.leadId } }); if (!lead) return; await this.finalizeAttempt(attemptId, attempt.campaignId, attempt.leadId, 'FAILED', retryRules, maxAttempts, lead.attemptCount); } private simulateAttempt( campaign: { id: string; queueId: string; maxWaitForAgentSeconds: number; wrapUpTimeSeconds: number; retryRules: unknown; maxAttempts: number; }, lead: { id: string; attemptCount: number }, attemptId: string, ): void { const profile = DEFAULT_SIMULATION_PROFILE; const outcome = simulateHangupCause(profile); if (outcome !== 'ANSWERED') { setTimeout( () => void this.finalizeAttempt( attemptId, campaign.id, lead.id, outcome, campaign.retryRules as Record, campaign.maxAttempts, lead.attemptCount + 1, ), randomInRange([1, 3]) * 1000, ); return; } const answerDelayMs = randomInRange(profile.answerDelaySecondsRange) * 1000; setTimeout(() => { void (async () => { await this.prisma.dialAttempt.update({ where: { id: attemptId }, data: { state: 'ANSWERED', answeredAt: new Date() } }); // Atendida != conectada a um agente. Se ninguém está disponível // agora, espera até maxWaitForAgentSeconds antes de considerar // abandono (agente.md seções 32/33) — sem isso, abandonRate nunca // reflete a realidade e o controle de abandono do motor preditivo // fica sem efeito prático (bug corrigido nesta fase). await this.prisma.dialAttempt.update({ where: { id: attemptId }, data: { state: 'QUEUED', queuedAt: new Date() } }); await this.tryConnectOrAbandon(campaign, lead, attemptId, profile, answerDelayMs / 1000); })(); }, answerDelayMs); } private async tryConnectOrAbandon( campaign: { id: string; queueId: string; maxWaitForAgentSeconds: number; wrapUpTimeSeconds: number; maxAttempts: number }, lead: { id: string; attemptCount: number }, attemptId: string, profile: (typeof DEFAULT_SIMULATION_PROFILE), answerDelaySeconds: number, ): Promise { const connect = async (agentId: string) => { await this.updateAnswerStats(campaign.id, answerDelaySeconds, false); const talkTimeSeconds = randomInRange(profile.talkTimeSecondsRange); await this.prisma.dialAttempt.update({ where: { id: attemptId }, data: { state: 'AGENT_CONNECTED', agentConnectedAt: new Date(), agentId }, }); setTimeout(() => { void (async () => { await this.finalizeAttempt(attemptId, campaign.id, lead.id, 'ANSWERED', {}, campaign.maxAttempts, lead.attemptCount + 1, talkTimeSeconds); await releaseAgentAfterCall(this.prisma, agentId, campaign.wrapUpTimeSeconds); })(); }, talkTimeSeconds * 1000); }; const agentId = await claimAvailableAgent(this.prisma, campaign.queueId); if (agentId) { await connect(agentId); return; } // Nenhum agente livre no instante do atendimento — espera até o limite // configurado e reavalia uma vez (poll único, suficiente para o // propósito de simulação sem complicar com um loop de polling real). setTimeout(() => { void (async () => { const retryAgentId = await claimAvailableAgent(this.prisma, campaign.queueId); if (retryAgentId) { await connect(retryAgentId); return; } await this.updateAnswerStats(campaign.id, answerDelaySeconds, true); await this.finalizeAttempt(attemptId, campaign.id, lead.id, 'ABANDONED' as HangupOutcome, {}, campaign.maxAttempts, lead.attemptCount + 1); })(); }, campaign.maxWaitForAgentSeconds * 1000); } // `abandoned` alimenta o EWMA de abandonRate — é este sinal que // `adjustPacingFactor` usa para reduzir o pacing (agente.md seção 33). private async updateAnswerStats(campaignId: string, answerDelaySeconds: number, abandoned: boolean): Promise { const campaign = await this.prisma.campaign.findUnique({ where: { id: campaignId } }); if (!campaign) return; const stats = await this.statsStore.load(campaignId, campaign.pacingInitial); stats.answerProbability = updateEwma(stats.answerProbability, 1); stats.avgAnswerDelaySeconds = updateEwma(stats.avgAnswerDelaySeconds, answerDelaySeconds); stats.abandonRate = updateEwma(stats.abandonRate, abandoned ? 1 : 0, 0.05); await this.statsStore.save(campaignId, stats); } private async finalizeAttempt( attemptId: string, campaignId: string, leadId: string, outcome: HangupOutcome, retryRules: Record, maxAttempts: number, attemptCount: number, talkTimeSeconds?: number, ): Promise { const now = new Date(); await this.prisma.dialAttempt.update({ where: { id: attemptId }, data: { state: outcome === 'ANSWERED' ? 'COMPLETED' : 'FAILED', hangupCause: outcome, endedAt: now, }, }); const campaign = await this.prisma.campaign.findUnique({ where: { id: campaignId } }); const stats = campaign ? await this.statsStore.load(campaignId, campaign.pacingInitial) : null; if (outcome === 'ANSWERED') { await this.prisma.lead.update({ where: { id: leadId }, data: { status: 'COMPLETED', lastResult: 'ANSWERED' } }); if (stats && talkTimeSeconds) { stats.avgTalkTimeSeconds = updateEwma(stats.avgTalkTimeSeconds, talkTimeSeconds); await this.statsStore.save(campaignId, stats); } return; } // Chamada não atendida: também é uma amostra negativa para a taxa de // atendimento (EWMA) — sem isso, answerProbability só reflete os // sucessos e superestima a taxa real. if (stats) { stats.answerProbability = updateEwma(stats.answerProbability, 0); await this.statsStore.save(campaignId, stats); } if (attemptCount >= maxAttempts) { await this.prisma.lead.update({ where: { id: leadId }, data: { status: 'MAX_ATTEMPTS', lastResult: outcome }, }); return; } // Status permanece informativo (BUSY/NO_ANSWER/FAILED, agente.md seção // 26) — não é o status sozinho que controla a elegibilidade para nova // tentativa, é a combinação status-retentável + next_attempt_at <= now // (ver LeadRepository.reserveNextLead). const nextAttemptAt = calculateNextAttemptAt(outcome as Exclude, retryRules, now); await this.prisma.lead.update({ where: { id: leadId }, data: { status: leadStatusForOutcome(outcome), lastResult: outcome, nextAttemptAt, }, }); } }