B2BCall-dialer/apps/predictive-dialer/src/main.ts

import { randomUUID } from "node:crypto";
import Redis from "ioredis";
import { getPrismaClient, withTenantContext } from "@b2bcall/database";
import { FreeSwitchTelephonyProvider } from "@b2bcall/telephony";
import { createLogger } from "@b2bcall/shared";
import { acquireCampaignLock, renewCampaignLock, releaseCampaignLock } from "./redis-primitives";
import { tickCampaign, type TickDeps } from "./tick";
import { startEventListener } from "./event-listener";

const logger = createLogger("b2bcall-predictive-dialer");

const TICK_INTERVAL_MS = 2000;
const LOCK_TTL_MS = 10_000;

function requireEnv(name: string): string {
  const value = process.env[name];
  if (!value) throw new Error(`${name} nao definido no ambiente`);
  return value;
}

/**
 * Real Outbound Safety (agente.md secao 186): as DUAS condições precisam
 * estar explicitamente ligadas — nunca ativado automaticamente. Lido uma
 * vez no boot (não é algo que deveria mudar em runtime sem reiniciar o
 * worker).
 */
function readOutboundSafetyFlags(): { dialerSimulation: boolean; allowRealOutboundCalls: boolean } {
  const dialerSimulation = (process.env.DIALER_SIMULATION ?? "true") !== "false";
  const allowRealOutboundCalls = process.env.ALLOW_REAL_OUTBOUND_CALLS === "true";
  if (dialerSimulation) {
    logger.info("DIALER_SIMULATION=true — nenhuma chamada PSTN real sera originada");
  } else if (!allowRealOutboundCalls) {
    logger.warn(
      "DIALER_SIMULATION=false mas ALLOW_REAL_OUTBOUND_CALLS != true — chamadas reais continuam bloqueadas (secao 186)",
    );
  } else {
    logger.warn("CHAMADAS PSTN REAIS HABILITADAS (DIALER_SIMULATION=false + ALLOW_REAL_OUTBOUND_CALLS=true)");
  }
  return { dialerSimulation, allowRealOutboundCalls };
}

async function main(): Promise<void> {
  const { dialerSimulation, allowRealOutboundCalls } = readOutboundSafetyFlags();
  const redis = new Redis(requireEnv("REDIS_URL"));
  redis.on("error", (err) => logger.error("erro na conexao Redis", { error: String(err) }));

  const provider = new FreeSwitchTelephonyProvider({
    host: requireEnv("ESL_HOST"),
    port: Number(process.env.ESL_PORT ?? 8021),
    password: requireEnv("ESL_PASSWORD"),
    logger: {
      debug: () => {},
      info: (msg) => logger.debug(msg),
      error: (msg, data) => logger.error(msg, { detail: data }),
    },
  });
  provider.connect();
  await provider.waitUntilConnected(10_000);

  const eventSubscriber = startEventListener(requireEnv("REDIS_URL"));

  const workerId = randomUUID();
  const deps: TickDeps = { redis, provider, dialerSimulation, allowRealOutboundCalls };

  let stopped = false;
  const tick = async () => {
    if (stopped) return;
    try {
      await runTick(deps, workerId);
    } catch (err) {
      logger.error("falha no tick", { error: String(err) });
    }
    if (!stopped) setTimeout(tick, TICK_INTERVAL_MS);
  };
  setTimeout(tick, TICK_INTERVAL_MS);

  logger.info("b2bcall-predictive-dialer iniciado", { workerId, tickIntervalMs: TICK_INTERVAL_MS });

  const shutdown = async () => {
    stopped = true;
    logger.info("encerrando b2bcall-predictive-dialer");
    await provider.disconnect();
    redis.disconnect();
    eventSubscriber.disconnect();
    process.exit(0);
  };
  process.on("SIGTERM", shutdown);
  process.on("SIGINT", shutdown);
}

/**
 * Um tick = uma passada por todos os tenants ativos, campanhas RUNNING (ou
 * WAITING_SCHEDULE, que pode voltar a RUNNING dentro da janela — ver
 * tick.ts). Lock por campanha (secao 79) garante que, mesmo rodando mais
 * de um worker deste serviço, só um processa uma dada campanha por vez —
 * renovado durante o processamento, liberado com compare-and-delete no
 * final (nunca libera o lock de outro dono).
 */
async function runTick(deps: TickDeps, workerId: string): Promise<void> {
  const prisma = getPrismaClient();
  const tenants = await prisma.tenant.findMany({ where: { status: "ACTIVE" } });

  for (const tenant of tenants) {
    const campaigns = await withTenantContext(prisma, tenant.id, (tx) =>
      tx.campaign.findMany({
        where: { tenantId: tenant.id, deletedAt: null, status: { in: ["RUNNING", "WAITING_SCHEDULE"] } },
      }),
    );

    for (const campaign of campaigns) {
      const ownerToken = `${workerId}:${randomUUID()}`;
      const locked = await acquireCampaignLock(deps.redis, campaign.id, ownerToken, LOCK_TTL_MS);
      if (!locked) continue;

      const renewTimer = setInterval(() => {
        renewCampaignLock(deps.redis, campaign.id, ownerToken, LOCK_TTL_MS).catch(() => undefined);
      }, LOCK_TTL_MS / 2);

      try {
        await tickCampaign(deps, tenant, campaign);
      } catch (err) {
        logger.error("falha no tick da campanha", { error: String(err), campaignId: campaign.id });
      } finally {
        clearInterval(renewTimer);
        await releaseCampaignLock(deps.redis, campaign.id, ownerToken);
      }
    }
  }
}

main().catch((err) => {
  logger.error("falha fatal ao iniciar b2bcall-predictive-dialer", { error: String(err) });
  process.exit(1);
});