feat: implement predictive dialing engine
- apps/dialer-worker: motor do discador preditivo completo
- predictive-engine.ts: EWMA de answerProbability/avgTalkTimeSeconds/
abandonRate, previsao de liberacao de agentes, calculo de quantas
chamadas originar. Logica pura, 14 testes unitarios
- cps-limiter.ts: token bucket via script Lua atomico no Redis (dois
buckets independentes campanha+tronco, min() dos dois, seguro com
multiplos workers)
- lead-repository.ts: reserva atomica via FOR UPDATE SKIP LOCKED,
recuperacao de reservas orfas apos queda de worker
- campaign-lock.ts: lock distribuido por campanha (Redis SET NX PX +
token de posse, renovacao/liberacao seguras via Lua)
- schedule.ts: janela de horario da campanha (timezone real via
Intl.DateTimeFormat, dias da semana), 6 testes unitarios
- retry-rules.ts: motor de retentativa por causa de encerramento,
configuravel por campanha, nunca infinito
- simulation.ts + campaign-worker.ts (modo DIALER_SIMULATION): permite
testar o motor inteiro sem tronco de operadora real
- simulation-harness.ts: reproduz em tempo discreto e deterministico o
cenario exato de aceite da secao 66 (20 agentes/10 CPS/30% atendimento/
TMA 180s) — 6 testes validando CPS nunca excedido, concorrencia nunca
excedida, pacing nao diverge
- campaign-worker.ts: orquestra tudo contra Postgres/Redis/Asterisk reais
- docs/PREDICTIVE_DIALER.md: algoritmo documentado, incluindo dois bugs
reais encontrados e corrigidos durante o teste do cenario de aceite
(concorrencia nao contava chamadas em atendimento; pacing subia sem
limite durante periodos ociosos, causando rajada maxima assim que um
agente ficava livre) e limitacoes conhecidas (AMD e wrap-up automatico
via eventos reais ainda pendentes, documentados sem esconder)
Testado ponta a ponta contra containers reais (Postgres/Redis/Asterisk):
campanha completa criada -> agente disponivel via API -> leads importados
-> campanha iniciada -> reserva atomica -> CPS respeitado -> simulacao de
NO_ANSWER (retry agendado) e ANSWERED (AGENT_CONNECTED, EWMA atualizada ao
vivo) -> parada sem derrubar chamadas em andamento.
This commit is contained in:
279
apps/dialer-worker/src/campaign-worker.ts
Normal file
279
apps/dialer-worker/src/campaign-worker.ts
Normal file
@@ -0,0 +1,279 @@
|
||||
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 { 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<void> {
|
||||
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; callerId: string | null; context: string; ringTimeoutSeconds: number; retryRules: unknown; maxAttempts: number },
|
||||
trunkName: string,
|
||||
lead: { id: string; phone: string; phoneNormalized: string; attemptCount: number },
|
||||
): Promise<void> {
|
||||
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<string, number>, 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<string, number>, campaign.maxAttempts, lead.attemptCount + 1);
|
||||
}
|
||||
}
|
||||
|
||||
private async failIfStillPending(attemptId: string, retryRules: Record<string, number>, maxAttempts: number): Promise<void> {
|
||||
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; 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<string, number>,
|
||||
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() } });
|
||||
await this.updateAnswerStats(campaign.id, answerDelayMs / 1000);
|
||||
|
||||
const talkTimeSeconds = randomInRange(profile.talkTimeSecondsRange);
|
||||
await this.prisma.dialAttempt.update({ where: { id: attemptId }, data: { state: 'AGENT_CONNECTED', agentConnectedAt: new Date() } });
|
||||
setTimeout(() => {
|
||||
void this.finalizeAttempt(attemptId, campaign.id, lead.id, 'ANSWERED', {}, campaign.maxAttempts, lead.attemptCount + 1, talkTimeSeconds);
|
||||
}, talkTimeSeconds * 1000);
|
||||
})();
|
||||
}, answerDelayMs);
|
||||
}
|
||||
|
||||
private async updateAnswerStats(campaignId: string, answerDelaySeconds: number): Promise<void> {
|
||||
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);
|
||||
await this.statsStore.save(campaignId, stats);
|
||||
}
|
||||
|
||||
private async finalizeAttempt(
|
||||
attemptId: string,
|
||||
campaignId: string,
|
||||
leadId: string,
|
||||
outcome: HangupOutcome,
|
||||
retryRules: Record<string, number>,
|
||||
maxAttempts: number,
|
||||
attemptCount: number,
|
||||
talkTimeSeconds?: number,
|
||||
): Promise<void> {
|
||||
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<HangupOutcome, 'ANSWERED'>, retryRules, now);
|
||||
await this.prisma.lead.update({
|
||||
where: { id: leadId },
|
||||
data: {
|
||||
status: leadStatusForOutcome(outcome),
|
||||
lastResult: outcome,
|
||||
nextAttemptAt,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user