import { PrismaClient, Lead } from '@b2bcall/database'; const RESERVATION_TIMEOUT_SECONDS = 90; /** * Reserva e recuperação de leads (agente.md seção 35). A transição * READY -> RESERVED é atômica via `FOR UPDATE SKIP LOCKED`: dois workers * nunca conseguem reservar o mesmo lead, e nenhum fica bloqueado esperando * a fila de lock do outro — simplesmente pega o próximo disponível. */ export class LeadRepository { constructor(private readonly prisma: PrismaClient) {} async reserveNextLead(campaignId: string, workerId: string): Promise { // READY (nunca tentado) ou BUSY/NO_ANSWER/FAILED cuja janela de retry já // passou (agente.md seção 79) — o status continua informativo até aqui, // é o par (status retentável + next_attempt_at) que decide elegibilidade. // $queryRaw NÃO passa pelo mapeamento camelCase do Prisma (isso só // acontece nos métodos gerados do Client) — sem os aliases explícitos // abaixo, campos como phone_normalized voltam com esse nome mesmo, // quebrando silenciosamente qualquer código que espere phoneNormalized. const rows = await this.prisma.$queryRaw` UPDATE leads SET status = 'RESERVED', reserved_at = now(), reserved_by = ${workerId} WHERE id = ( SELECT id FROM leads WHERE campaign_id = ${campaignId} AND status IN ('READY', 'BUSY', 'NO_ANSWER', 'FAILED') AND (next_attempt_at IS NULL OR next_attempt_at <= now()) ORDER BY next_attempt_at ASC NULLS FIRST, created_at ASC FOR UPDATE SKIP LOCKED LIMIT 1 ) RETURNING id, campaign_id AS "campaignId", import_id AS "importId", name, phone, phone_normalized AS "phoneNormalized", status, attempt_count AS "attemptCount", last_attempt_at AS "lastAttemptAt", next_attempt_at AS "nextAttemptAt", last_result AS "lastResult", reserved_at AS "reservedAt", reserved_by AS "reservedBy", custom_fields AS "customFields", created_at AS "createdAt", updated_at AS "updatedAt" `; return rows[0] ?? null; } // Promove leads NEW -> READY (leads recém-importados começam em NEW para // permitir uma etapa de higienização futura antes de entrarem na fila de // discagem; por ora promovemos todos imediatamente ao iniciar a campanha). async promoteNewLeads(campaignId: string): Promise { const result = await this.prisma.lead.updateMany({ where: { campaignId, status: 'NEW' }, data: { status: 'READY' }, }); return result.count; } // Recupera leads cujo worker morreu antes de originar de fato (agente.md // seção 35: "implementar timeout de reservation para recuperar leads // caso um worker morra"). async releaseExpiredReservations(): Promise { const result = await this.prisma.$executeRaw` UPDATE leads SET status = 'READY', reserved_at = NULL, reserved_by = NULL WHERE status = 'RESERVED' AND reserved_at < now() - (${RESERVATION_TIMEOUT_SECONDS}::text || ' seconds')::interval `; return Number(result); } async countReadyLeads(campaignId: string): Promise { return this.prisma.lead.count({ where: { campaignId, status: { in: ['READY', 'BUSY', 'NO_ANSWER', 'FAILED'] }, OR: [{ nextAttemptAt: null }, { nextAttemptAt: { lte: new Date() } }], }, }); } async hasRemainingWork(campaignId: string): Promise { // BUSY/NO_ANSWER/FAILED contam como trabalho restante até baterem // max_attempts (aí viram MAX_ATTEMPTS, terminal) — excluir esses três // faria a campanha ser marcada COMPLETED com leads ainda pendentes de // retry. const count = await this.prisma.lead.count({ where: { campaignId, status: { in: ['NEW', 'READY', 'RESERVED', 'DIALING', 'RINGING', 'ANSWERED', 'CALLBACK', 'BUSY', 'NO_ANSWER', 'FAILED'], }, }, }); return count > 0; } }