/**
 * Drain pós-fusão: consome `ai_agent.dispatch_requested` do event_log (MESMO
 * banco — a role vendaval_drain e o transporte cross-banco morreram) e enfileira
 * jobs `inbound_turn` na fila durável do harness.
 *
 * Garantias:
 *   - organization_id vem da LINHA do evento (fonte confiável), nunca do payload;
 *   - at-least-once + dedup: claim CAS (pending→processing) + unique
 *     (organization_id, source_event_id) em job_queue com captura de 23505;
 *   - coalescência de rajada: mensagens do MESMO contato dentro da janela de
 *     debounce viram UM job (o turno lê o histórico completo e responde a todas);
 *   - grupos @g.us: skip (regra dura nº 12), exceto quando a chave POR GRUPO
 *     `ia_responde` liberou a IA (#2729) — evento marcado done sem job;
 *   - eventos 'processing' órfãos (crash do worker) voltam a 'pending' por timeout.
 */
import { z } from 'zod';
import type pg from 'pg';

import { loadConversationAgentConfig, type PublishedAgentConfig } from '../../agent/agent-config';
import { insertInboxItem } from '../../db/repository';
import type { Logger } from '../../obs/logger';
import { enqueueJob } from '../../queue/queue';
import { janelaDeRespostaAberta, soltarAdiadoPelaJanela } from '../../pacing/adiados-da-janela';
import { decidirRajada, debounceEfetivo } from './debounce';
import {
  intervaloEfetivo,
  runAfterComIntervalo,
  ultimaRespostaDaIa,
} from './intervalo-entre-respostas';
import { avisoDeEventoMorto, IA_QUE_NAO_RESPONDEU } from '@/lib/event-log/aviso-de-evento-morto';
import { acordarLacoDeEventLog } from '@/lib/event-log/sinal-de-trabalho';
import { canalDesativado } from '@/lib/channels/desativado';
import { TETO_DA_ESPERA_DA_DERIVACAO_MS } from '@/lib/messaging/media/derivable';
import {
  haLeitorDaDescricao,
  LOTE_DE_REABERTURA,
  reabrirLeituraAdiada,
} from '@/lib/messaging/media/leitura-adiada';
import { midiaDerivavelPendente } from './midia-derivavel';
import { haQuemAtendaASessao } from '@/lib/ai/agents/quem-atende-a-sessao';
import { decidirElegibilidadeDaConversa } from '@/lib/ai/elegibilidade/consulta-pg';
import { deveCederTurnoAoRetorno } from '@/lib/followup/ceder-turno-ao-retorno';
import { grupoLiberaIa } from '@/lib/grupos/ia-no-grupo';
import { ehOperante } from '@/lib/organizacao/operante';

const DRAIN_CONSUMER = 'agent-engine';

const dispatchPayloadSchema = z
  .object({
    conversation_id: z.string().uuid(),
    contact_id: z.string().uuid(),
    channel_session_id: z.string().uuid(),
    inbound_message_id: z.string().uuid(),
  })
  .passthrough();

interface EventRow {
  id: string;
  organization_id: string;
  payload: unknown;
  attempts: number;
  created_at: string;
}

export interface DrainKnobs {
  batchSize: number;
  intervalMs: number;
  idleIntervalMs: number;
  /** Janela de coalescência de rajada inbound por contato (0 = sem debounce). */
  debounceMs: number;
  /** Evento 'processing' órfão volta a 'pending' após isto. */
  reapTimeoutMs: number;
  /**
   * Janela de validade da autorização de IA de um contato (gate 'allowlist').
   * Só consultada em canal com `metadata.ai_gate = 'allowlist'`. Ausente nos
   * testes que não exercitam o gate — o default de 21 dias em ms é aplicado.
   */
  allowlistTtlMs?: number;
  /**
   * Relógio da janela de RESPOSTA (#3383). Só é usado na pergunta "a janela
   * está aberta AGORA?" que o drain faz quando achar um job adiado esperando
   * carona — e é justamente aí que o teste precisa fixar a hora (22h11 para a
   * janela 7h–22h fechada) sem depender do relógio de parede do runner.
   * Ausente = `new Date()`.
   */
  clock?: () => Date;
  /**
   * Avisa o laço da fila que há job NOVO — corta a soneca dele (#3688). O drain
   * roda no MESMO worker do laço, então o aviso é uma promessa local, não fila
   * nem polling. Ausente onde não há laço para acordar (CLI, testes): quem
   * enfileira não pode depender de existir alguém dormindo.
   */
  acordarFila?: () => void;
}

/** Default de `allowlistTtlMs` (21 dias) para testes que omitem o knob. */
const ALLOWLIST_TTL_MS_PADRAO = 21 * 24 * 60 * 60 * 1000;

/** Um tick do drain: claima um lote de eventos e os transforma em jobs. */
export async function drainTick(pool: pg.Pool, knobs: DrainKnobs, log: Logger): Promise<number> {
  // Reaper de eventos órfãos, a cada tick. Quem o serve é o parcial
  // `event_log_processing_por_tipo_idx` (event_type where status='processing',
  // migration 0585); o `= any(consumed_by)` é filtro e não usa o GIN.
  await pool.query(
    `update event_log set status = 'pending', updated_at = now()
     where event_type = 'ai_agent.dispatch_requested'
       and status = 'processing'
       and $1 = any(consumed_by)
       and updated_at < now() - make_interval(secs => $2 / 1000.0)`,
    [DRAIN_CONSUMER, knobs.reapTimeoutMs],
  );

  const { rows: events } = await pool.query<EventRow>(
    `update event_log e
     set status = 'processing', attempts = e.attempts + 1,
         consumed_by = array_append(array_remove(coalesce(e.consumed_by, '{}'), $2), $2),
         updated_at = now()
     where e.id in (
       select id from event_log
       where event_type = 'ai_agent.dispatch_requested'
         and status = 'pending'
         and (next_attempt_at is null or next_attempt_at <= now())
       order by created_at
       limit $1
       for update skip locked
     )
     returning e.id, e.organization_id, e.payload, e.attempts, e.created_at`,
    [knobs.batchSize, DRAIN_CONSUMER],
  );

  for (const event of events) {
    try {
      const desfecho = await processEvent(pool, event, knobs, log);
      if (desfecho === 'adiar') {
        // Adiar NÃO é falha: volta a pending com uma espera curta e não gasta
        // o orçamento de tentativas (que existe para erro de verdade).
        await pool.query(
          `update event_log
           set status = 'pending', attempts = greatest(attempts - 1, 0),
               next_attempt_at = now() + make_interval(secs => $2 / 1000.0), updated_at = now()
           where id = $1`,
          [event.id, ESPERA_DERIVACAO_MS],
        );
        continue;
      }
      await pool.query(`update event_log set status = 'done', updated_at = now() where id = $1`, [
        event.id,
      ]);
    } catch (err) {
      const message = (err instanceof Error ? err.message : String(err)).slice(0, 300);
      const terminal = event.attempts >= 5;
      await pool.query(
        `update event_log
         set status = $2, last_error = $3, next_attempt_at = now() + interval '30 seconds',
             updated_at = now()
         where id = $1`,
        [event.id, terminal ? 'dead' : 'pending', message],
      );
      log.error('drain: evento falhou', { event_id: event.id, terminal, error: message });
      if (terminal) await avisarDespachoMorto(pool, event, message, log);
    }
  }
  return events.length;
}

/**
 * O DESPACHO DA IA QUE MORRE AVISA A CENTRAL — como o dreno de handlers já avisa.
 *
 * `lib/event-log/drain.ts` passou a abrir `event_dead` quando desiste de um
 * evento; este dreno marca `dead` o `ai_agent.dispatch_requested` pelo mesmo
 * critério (5 tentativas) e seguia sem avisar ninguém. É o pior dos dois
 * silêncios: o efeito que não aconteceu é a resposta ao cliente.
 *
 * Mesmo texto do outro dreno, mas dedupe POR TÍTULO (`kind_e_titulo`), só
 * enquanto houver um aberto: um `event_dead` de mídia ou de automação aberto não
 * engole este, que é o único que diz que um cliente ficou sem resposta (ver
 * `aviso-de-evento-morto.ts`, "as duas famílias"). SQL de uma instrução
 * (`insertInboxItem`, `insert … where not exists`) em vez de consulta seguida
 * de insert. Mil despachos mortos numa pane abrem um aviso, não mil: medido em
 * `tests/invariants/evento-morto-nao-inunda-a-central.test.ts`; o aviso de
 * outra família aberto não cala este: medido em
 * `tests/invariants/aviso-da-ia-nao-some-atras-de-outro-evento-morto.test.ts`.
 *
 * Fire-and-forget: falhar ao avisar não pode derrubar o tick, que ainda tem o
 * resto do lote para drenar.
 */
async function avisarDespachoMorto(
  pool: pg.Pool,
  event: EventRow,
  motivo: string,
  log: Logger,
): Promise<void> {
  const { title, body } = avisoDeEventoMorto({
    eventType: 'ai_agent.dispatch_requested',
    // `attempts` já foi incrementado no claim: é a contagem com esta tentativa.
    tentativas: event.attempts,
    motivo,
    efeito: IA_QUE_NAO_RESPONDEU,
  });
  try {
    await insertInboxItem(
      pool,
      event.organization_id,
      { kind: 'event_dead', severity: 'critical', title, body },
      'kind_e_titulo',
    );
  } catch (err) {
    log.error('drain: aviso de despacho morto falhou', {
      event_id: event.id,
      error: (err instanceof Error ? err.message : String(err)).slice(0, 300),
    });
  }
}

/** Quanto esperar entre uma checagem e outra da derivação de mídia. */
const ESPERA_DERIVACAO_MS = 4_000;

type DesfechoEvento = 'processado' | 'adiar';

/**
 * O agente desta conversa, para as afinações do dreno: a janela de rajada
 * (#1856), o intervalo mínimo entre respostas e a foto (9108: se ele vê
 * imagens, e quantas mensagens ele lê).
 *
 * A resolução é a de `loadConversationAgentConfig` — o `active_ai_agent_id` da
 * conversa quando há dono explícito, senão o agente publicado da sessão. NÃO é
 * a resolução completa do turno (`resolveTurnAgent`: router, classificador,
 * campanha): numa conversa que o turno entregaria a outro agente sem torná-lo
 * dono, vale a janela do agente da sessão. Por isso a foto só usa esta
 * resolução para o SIM ("ele vê"); o NÃO, ou o `null`, cai na régua da empresa
 * (`haLeitorDaDescricao`), e a espera da foto não depende do agente. Sem agente
 * resolvido, a janela é a da env e não há intervalo.
 *
 * Falha da consulta NÃO derruba o evento: devolve `null`, como a checagem de
 * elegibilidade acima. Afinação não é motivo para retry.
 */
async function agenteDoEvento(
  pool: pg.Pool,
  event: EventRow,
  p: { conversation_id: string; channel_session_id: string },
  log: Logger,
): Promise<PublishedAgentConfig | null> {
  try {
    return await loadConversationAgentConfig(pool, event.organization_id, p.conversation_id, p.channel_session_id);
  } catch (err) {
    log.warn('drain: agente da conversa não resolveu — janela da instalação, sem intervalo, e foto pela régua da empresa', {
      event_id: event.id,
      error: (err instanceof Error ? err.message : String(err)).slice(0, 160),
    });
    return null;
  }
}

async function processEvent(
  pool: pg.Pool,
  event: EventRow,
  knobs: DrainKnobs,
  log: Logger,
): Promise<DesfechoEvento> {
  const parsed = dispatchPayloadSchema.safeParse(event.payload);
  if (!parsed.success) {
    // Payload fora do contrato do ingest — evento é descartável (processed), não
    // retryável: re-tentar não conserta shape.
    log.warn('drain: payload de dispatch fora do contrato — evento descartado', {
      event_id: event.id,
    });
    return 'processado';
  }
  const p = parsed.data;

  // Organização parada (suspensa, redigida, arquivada) não gera turno: o evento
  // é consumido sem job. Vai na MESMA consulta do modo externo — uma ida ao banco
  // por evento, não duas — e vem ANTES do `canAssist`, que desliga o gate.
  const { rows: modeRows } = await pool.query<{ mode: string | null; status: string | null }>(
    `select settings->>'ai_dispatch_mode' as mode, status from organizations where id = $1`,
    [event.organization_id],
  );
  if (!ehOperante(modeRows[0]?.status)) {
    log.info('drain: organização não operante — evento consumido sem job', { event_id: event.id });
    return 'processado';
  }
  // Spec 14: org em modo 'external' tem agente EXTERNO como dono da conversa —
  // o engine não responde por cima. Evento é consumido (done) sem job.
  if (modeRows[0]?.mode === 'external') {
    log.info('drain: org em modo external (spec 14) — evento pulado', { event_id: event.id });
    return 'processado';
  }

  // Canal desativado pelo operador: defesa em profundidade do `pedirDespachoDoAgente`
  // (que já não emite para desativado). Evento antigo em voo ou emit direto cai
  // aqui e é consumido sem job, antes de qualquer custo.
  if (p.channel_session_id) {
    const { rows: canalRows } = await pool.query<{ metadata: unknown }>(
      'select metadata from channel_sessions where organization_id = $1 and id = $2',
      [event.organization_id, p.channel_session_id],
    );
    const meta = canalRows[0]?.metadata;
    if (canalDesativado(meta)) {
      log.info('drain: canal desativado — evento consumido sem job', { event_id: event.id });
      return 'processado';
    }
  }

  // Grupos: skip, com UMA exceção estreita (regra dura nº 12 + #2729) — o
  // evento marcado done sem job continua sendo o padrão de 100% das
  // instalações.
  // A consulta já traz a chave POR GRUPO (`channel_session_groups.ia_responde`)
  // junto do `is_group`, por um LEFT JOIN no MESMO select: é a leitura que o
  // ingest do grupo também faz, e a régua que decide é a MESMA função
  // (`grupoLiberaIa`, `=== true`, falha fechada) — divergência entre as duas
  // leituras é como a régua local do worker já passou a discordar da tela em
  // produção (ver `pos-entrada.ts`). Conversa sem linha na fila (LEFT JOIN sem
  // par) chega com `ia_responde` nulo, que é "não sei se liberaram" — e não sei
  // não responde.
  const { rows: convRows } = await pool.query<{ is_group: boolean; ia_responde: boolean | null }>(
    `select c.is_group, g.ia_responde
       from conversations c
       left join channel_session_groups g
         on g.organization_id = c.organization_id
        and g.channel_session_id = c.channel_session_id
        and g.group_chat_id = c.group_chat_id
      where c.organization_id = $1 and c.id = $2`,
    [event.organization_id, p.conversation_id],
  );
  if (convRows[0]?.is_group !== false) {
    if (!grupoLiberaIa(convRows[0]?.ia_responde)) {
      log.info('drain: conversa de grupo sem a chave da IA liberada — evento pulado', { event_id: event.id });
      return 'processado';
    }
    log.info('drain: grupo liberado pela chave por grupo (#2729) — turno segue o caminho normal', {
      event_id: event.id,
    });
  }

  // Ninguém para atender: NÃO gastar. Sem agente publicado para esta sessão e
  // sem roteador que possa resolver alguém, o turno seguia assim mesmo e caía
  // no caminho genérico — rodando o pipeline inteiro e pagando por ele.
  //
  // Medido nesta VPS com o agente PAUSADO (despublicado pela tela): uma única
  // mensagem gastou 6 chamadas ao LLM, ~2 centavos, e ainda produziu resposta.
  // Multiplicado por toda mensagem que chega, com o agente desligado, é dinheiro
  // saindo sem ninguém ter pedido nada — e "pausei o agente" tem que significar
  // "parou de gastar".
  //
  // Também cobre a instalação recém-feita que ainda não configurou agente
  // nenhum: hoje ela pagaria por cada mensagem recebida.
  //
  // Roteador com membro PUBLICADO ou fallback PUBLICADO continua passando: ali
  // existe quem atenda, e o caminho genérico de "classificou e não bateu" segue
  // valendo.
  //
  // Este parágrafo dizia "roteador COM membros ou COM fallback", e a diferença
  // não é de redação: pausar um agente não apaga a linha dele em
  // `ai_router_members` nem zera `ai_routers.fallback_agent_id`. Medir a
  // EXISTÊNCIA da linha deixava o portão aberto para um roteador cujos membros
  // foram todos pausados — exatamente o caso que o parágrafo acima diz estar
  // cobrindo, entrando pela outra porta.
  //
  // A pergunta mora em `haQuemAtendaASessao` porque o worker de clima faz a
  // MESMA antes de perguntar ao Jev pelos pedidos do cliente: ele só conta um
  // pedido que a regra de hoje deixou passar onde este turno rodaria.
  const haQuem = await haQuemAtendaASessao(pool, event.organization_id, p.channel_session_id);
  if (haQuem === false) {
    log.info('drain: nenhum agente publicado para a sessão — turno pulado (sem gasto)', {
      event_id: event.id,
      channel_session_id: p.channel_session_id,
    });
    return 'processado';
  }

  // ANTI-BACKLOG (toda instalação, sem knob): a mensagem que disparou este
  // evento ainda é a última inbound da conversa? Se já veio inbound mais nova,
  // ESTE evento está superado — a mensagem nova tem o próprio evento, e o turno
  // dela lê o histórico inteiro (esta mensagem inclusa). Sem isto, um worker que
  // ficou parado (deploy, OOM na VPS) acorda e drena o backlog em ordem de
  // `created_at`, disparando um turno para CADA mensagem antiga — a IA
  // respondendo conversa de dias atrás. Vira done, sem job, sem gasto.
  //
  // R7: o desempate. `order by sent_at desc, id desc` cai no `id` — uuid
  // aleatório, não cronológico — sempre que dois inbound compartilham `sent_at`
  // (relógio do provider repetido, ou duas mensagens na mesma janela sem
  // timestamp). "A última" saía por sorteio e podia eleger a ANTIGA, disparando
  // o turno dela. `coalesce(sent_at, created_at)` (defensivo — `sent_at` é
  // `not null default now()` hoje, mas o padrão do repo, ver migration 0027, não
  // confia nisso) com desempate por `created_at` (ordem de INGESTÃO, uma
  // mensagem por webhook) dá recência determinística.
  const { rows: ultimaInbound } = await pool.query<{ id: string }>(
    `select id from messages
     where organization_id = $1 and conversation_id = $2 and direction = 'inbound'
     order by coalesce(sent_at, created_at) desc, created_at desc, id desc
     limit 1`,
    [event.organization_id, p.conversation_id],
  );
  if (ultimaInbound[0] !== undefined && ultimaInbound[0].id !== p.inbound_message_id) {
    log.info('drain: evento superado por inbound mais recente — turno pulado (sem gasto)', {
      event_id: event.id,
      inbound_message_id: p.inbound_message_id,
      ultima_inbound_id: ultimaInbound[0].id,
    });
    return 'processado';
  }

  // UMA VOZ: se o gatilho "cliente voltou" enrollaria neste inbound, o LLM
  // não responde por cima. Fail-open dentro do helper — consulta falha = turno segue.
  if (
    await deveCederTurnoAoRetorno(pool, {
      organizationId: event.organization_id,
      contactId: p.contact_id,
      conversationId: p.conversation_id,
      messageId: p.inbound_message_id,
    })
  ) {
    log.info('drain: turno cedido ao follow-up de retorno — inbound_turn pulado', {
      event_id: event.id,
      contact_id: p.contact_id,
    });
    return 'processado';
  }

  // GATE DE ELEGIBILIDADE (opt-in por canal — `metadata.ai_gate = 'allowlist'`).
  // Num canal 'open' (o default), `decidirElegibilidade` devolve `permite:true`
  // com motivo 'gate_aberto' e nada muda. Num canal 'allowlist', a IA só assume
  // se o CONTATO estiver autorizado por uma origem elegível (Respondi, campanha,
  // automação, retomada manual) e dentro da janela. Bloqueio por allowlist =
  // done, sem job, sem gasto — a conversa fica para atendimento humano.
  //
  // `force_human` / silêncio / dono humano bloqueiam em QUALQUER modo, e é o
  // TURNO quem garante isso — aqui a decisão só se antecipa para não enfileirar.
  //
  // Repare no `!canAssist` do `if` abaixo: com agente assistido publicado no
  // canal o gate é desligado INTEIRO nesta ponta, de propósito (o rascunho é o
  // produto do modo assistido, barrar aqui o mataria). A frase que este
  // comentário trazia — "o turno revalida" — era falsa justamente nesse caso: o
  // ramo assistido de `createInboundTurnHandler` devolvia antes das guardas de
  // `runAgentTurn`. As duas checagens agora vivem dentro daquele ramo
  // (`inbound-turn.ts`, `operationMode === 'assisted'`), e é lá que a defesa em
  // profundidade realmente acontece.
  // This is a capability check, never a selection by priority. The canonical
  // router chooses once in the worker, then automatic eligibility is rechecked.
  const {rows:assistance}=await pool.query<{available:boolean}>(`select exists(
    select 1 from ai_agents a join ai_agent_versions v on v.organization_id=a.organization_id and v.id=a.published_version_id
    where a.organization_id=$1 and a.archived_at is null and a.operation_mode='assisted' and v.status='published'
    and(v.channel_session_id=$2 or exists(select 1 from ai_routers r where r.organization_id=a.organization_id and r.channel_session_id=$2 and r.is_active and(r.fallback_agent_id=a.id or exists(select 1 from ai_router_members m where m.organization_id=r.organization_id and m.router_id=r.id and m.agent_id=a.id))))) as available`,[event.organization_id,p.channel_session_id]);
  const canAssist=assistance[0]?.available===true;
  try {
    const elegib = await decidirElegibilidadeDaConversa(pool, {
      organizationId: event.organization_id,
      conversationId: p.conversation_id,
      agora: new Date(),
      ttlMs: knobs.allowlistTtlMs ?? ALLOWLIST_TTL_MS_PADRAO,
    });
    if (!canAssist && elegib !== null && !elegib.permite) {
      log.info('drain: conversa não elegível para IA — turno pulado (sem gasto)', {
        event_id: event.id,
        conversation_id: p.conversation_id,
        motivo: elegib.motivo,
      });
      return 'processado';
    }
  } catch (err) {
    // Falha da consulta de elegibilidade NÃO derruba o drain e NÃO bloqueia o
    // turno: um lead real pode estar esperando. Degrada para o fluxo antigo
    // (enfileira) — o turno tem a segunda checagem.
    log.warn('drain: checagem de elegibilidade falhou — seguindo para o turno', {
      event_id: event.id,
      error: (err instanceof Error ? err.message : String(err)).slice(0, 160),
    });
  }

  // O agente da conversa: a janela de rajada (lá embaixo) e a foto (aqui).
  const agentConfig = await agenteDoEvento(pool, event, p, log);
  // Ele vê as fotos (9108)? Só decide sozinho quando a resposta é SIM: quem
  // responde o turno é escolhido depois, por `resolveTurnAgent` (roteador,
  // campanha), e pode ser outro agente, que vê.
  const agenteVe = agentConfig !== null && agentConfig.imageVisionEnabled !== false;

  // FOTO ADIADA POR FALTA DE LEITOR (`MEDIA_VISION_WITHOUT_READER=defer`): este
  // é o ponto em que se SABE que um agente vai ler a conversa — depois do portão
  // de quem atende e da elegibilidade. Reabrir aqui pede a descrição, e a espera
  // logo abaixo segura o primeiro turno até ela chegar (âncora no pedido). Falha
  // não segura o cliente: o turno segue com `[image]`, como no teto de hoje, e o
  // cron `media-leitura-adiada` reabre na rodada seguinte.
  //
  // Quem vê: o agente da conversa, se ele vê. Se ele não vê (ou não resolveu,
  // como no número só com roteador), a régua é a mesma que adiou a foto,
  // `haLeitorDaDescricao`: a empresa tem um agente que vê, ou outro leitor. Sem
  // ninguém, a foto fica adiada e o turno não espera por ela (`skipped` é final).
  //
  // Quantas: as que o agente LÊ. O turno lê as últimas `historyMessageWindow`
  // mensagens (até 200), e as N fotos mais novas contêm toda foto entre as N
  // mensagens mais novas; reabrir além disso pagava visão de foto que o turno
  // não alcança. Quando quem vê pode ser outro agente, o lote de sempre. Idade
  // não entra: foto antiga numa conversa que voltou é lida agora.
  try {
    if (agenteVe || (await haLeitorDaDescricao(pool, event.organization_id))) {
      const n = await reabrirLeituraAdiada(pool, event.organization_id, {
        conversationId: p.conversation_id,
        limite: agenteVe ? agentConfig.historyMessageWindow : LOTE_DE_REABERTURA,
      });
      if (n > 0) log.info('drain: leitura de mídia adiada pedida antes do turno', { event_id: event.id, reabertas: n });
    }
  } catch (err) {
    log.warn('drain: não consegui pedir a leitura adiada — turno segue', {
      event_id: event.id,
      error: (err instanceof Error ? err.message : String(err)).slice(0, 160),
    });
  }

  // Mídia ainda virando texto: ESPERAR. Sem isto o turno era despachado no mesmo
  // instante em que a mensagem chegava, enquanto o áudio ainda estava sendo
  // baixado e transcrito — e o cliente recebia "recebi seu áudio, mas não
  // consigo ouvi-lo" segundos ANTES de a transcrição ficar pronta. Medido nesta
  // VPS: dispatch às 20:24:22, derivação só pedida às 20:25:03.
  //
  // ─── A espera olha a CONVERSA, não a mensagem que disparou o evento ────────
  //
  // Antes olhava só `p.inbound_message_id`. Quando o cliente manda a FOTO e,
  // logo depois, a pergunta em TEXTO ("isso é de vocês?"), o turno dispara pelo
  // TEXTO — que não é derivável — e seguia sem esperar a visão da foto. O
  // cliente recebia "me conta o que aparece nela?" sobre uma foto cujo texto
  // derivado o próprio sistema terminou de gerar 3s depois. Medido nesta VPS,
  // 24/09/2026: foto 13:28:16 · texto 13:28:19 · turno enfileirado 13:28:28 ·
  // derivação concluída 13:28:35.
  //
  // O caso que isso quebra é o mais comum de todos: o cliente manda o
  // COMPROVANTE e escreve "já paguei, e vocês estão me cobrando". A evidência e
  // a alegação chegam em mensagens separadas, e o turno precisa das duas.
  //
  // A consulta em si (tipos deriváveis, `media_url is not null`, status fora de
  // `DERIVACAO_TERMINADA`) mora em `./midia-derivavel.ts`, porque o TURNO faz a
  // mesma pergunta antes de montar o prompt (#3836) e as duas não podem
  // discordar. O que fica aqui é a POLÍTICA do despacho: a âncora do teto é a
  // hora DA MÍDIA (`created_at`, quando ela chegou a nós — nunca `sent_at`, o
  // relógio do aparelho, que faria uma foto entregue com atraso nascer "além do
  // teto"; a foto reaberta acima conta do PEDIDO da leitura, também lá), e mídia
  // velha e travada não segura o turno para sempre.
  const midia = await midiaDerivavelPendente(pool, {
    organizationId: event.organization_id,
    conversationId: p.conversation_id,
  });
  if (midia !== null) {
    const esperandoHa = Date.now() - new Date(midia.quando).getTime();
    if (esperandoHa < TETO_DA_ESPERA_DA_DERIVACAO_MS) {
      log.info('drain: mídia ainda sendo transcrita — turno adiado', {
        event_id: event.id,
        tipo: midia.type,
        esperando_ha_ms: esperandoHa,
      });
      // ACORDA o laço do `event_log` — o que roda `media.persist_requested` e
      // `media.derive_requested`, isto é, quem faz a transcrição acontecer.
      //
      // É daqui porque este é o único ponto DENTRO do worker que sabe que a
      // derivação está pendente: os dois laços sobem no mesmo processo
      // (`workers/agent-worker/main.ts`), então um `import` basta — sem
      // LISTEN/NOTIFY, sem conexão nova por worker. O ingest do canal, que
      // escreve `media.persist_requested`, roda em OUTRO processo (a rota HTTP
      // do webhook) e continua limitado pelo teto do recuo; acordar de lá é o
      // que ficaria para um LISTEN/NOTIFY, fora deste PR.
      //
      // Sem isto, o laço do event_log dorme até o teto do recuo (30s por
      // default) e um áudio lento (86s medidos) chega a 116 dos 120s deste
      // turno; com teto maior passa, e o cliente ouve "não consegui ouvir seu
      // áudio", o defeito que fez o teto subir de 45s para 120s.
      acordarLacoDeEventLog();
      return 'adiar';
    }
    log.warn('drain: derivação não concluiu no teto — seguindo sem o texto', {
      event_id: event.id,
      tipo: midia.type,
      esperando_ha_ms: esperandoHa,
    });
  }

  // Coalescência: já existe job PENDING futuro deste contato → esta mensagem
  // entra de carona (o turno lê o histórico completo). Evento vira done.
  //
  // A janela e a exclusão do job em HOLD (`held_run_after` no payload — a lição
  // do #830) moram em ./debounce.ts, com teste próprio.
  // O agente já resolvido lá em cima (uma consulta só): sem ele, a janela da
  // instalação e nenhum intervalo.
  const ajustes = {
    debounceMs: debounceEfetivo(agentConfig?.inboundDebounceMs ?? null, knobs.debounceMs),
    // O intervalo entre respostas vem da MESMA versão do agente (mesma consulta).
    intervaloEntreRespostasMs: intervaloEfetivo(agentConfig?.replyMinIntervalMs ?? null),
  };
  const rajada = await decidirRajada(
    pool,
    { organizationId: event.organization_id, contactId: p.contact_id },
    ajustes.debounceMs,
    Date.now(),
    // Só é chamada quando o candidato à carona está adiado pela janela (#3383),
    // uma consulta por mensagem nesse caso — nenhum custo no caminho comum.
    () =>
      janelaDeRespostaAberta(pool, {
        organizationId: event.organization_id,
        channelSessionId: p.channel_session_id,
        agora: knobs.clock?.() ?? new Date(),
        log,
      }),
  );
  if (rajada.tipo === 'coalescido') {
    log.info('drain: rajada coalescida em job pendente', {
      event_id: event.id,
      job_id: rajada.jobId,
    });
    return 'processado';
  }
  // #3383 — a janela JÁ ABRIU e este job ainda dormia até o horário antigo:
  // ele não recebe a mensagem de carona (acima) e não continua adiado. O MESMO
  // job acorda (`run_after = now()`), sem inserir nada — o turno dele roda uma
  // vez, e a mensagem nova enfileira o próprio job logo abaixo.
  //
  // Falhar aqui NÃO derruba o evento: a mensagem nova continua sendo
  // enfileirada e respondida; o adiado fica para o próximo acordamento (ou
  // para a abertura calculada), com o warn dizendo o que não aconteceu.
  if (rajada.caronaRecusadaEm !== undefined) {
    let acordado = false;
    try {
      acordado = await soltarAdiadoPelaJanela(pool, rajada.caronaRecusadaEm);
    } catch (err) {
      log.warn('drain: não consegui acordar o job adiado com a janela aberta', {
        event_id: event.id,
        job_id: rajada.caronaRecusadaEm,
        error: (err instanceof Error ? err.message : String(err)).slice(0, 160),
      });
    }
    log.info('drain: job adiado pela janela não recebe carona — mensagem enfileirada à parte', {
      event_id: event.id,
      job_id: rajada.caronaRecusadaEm,
      acordado,
    });
  }

  // Intervalo mínimo entre respostas da IA (por agente): o turno só nasce
  // depois de `última resposta da IA + intervalo`. Falha da consulta NÃO derruba
  // o evento — o intervalo é afinação, e o cliente que espera resposta não paga
  // por ela: degrada para o `run_after` da rajada.
  let runAfter = rajada.runAfter;
  if (ajustes.intervaloEntreRespostasMs > 0) {
    try {
      const ultima = await ultimaRespostaDaIa(pool, {
        organizationId: event.organization_id,
        conversationId: p.conversation_id,
      });
      runAfter = runAfterComIntervalo(runAfter, ultima, ajustes.intervaloEntreRespostasMs);
    } catch (err) {
      log.warn('drain: não consegui medir o intervalo entre respostas — turno sem espera extra', {
        event_id: event.id,
        error: (err instanceof Error ? err.message : String(err)).slice(0, 160),
      });
    }
  }
  const { job, deduped } = await enqueueJob(pool, event.organization_id, {
    kind: 'inbound_turn',
    leadId: p.contact_id,
    sourceEventId: event.id,
    payload: {
      conversation_id: p.conversation_id,
      contact_id: p.contact_id,
      channel_session_id: p.channel_session_id,
      inbound_message_id: p.inbound_message_id,
      crm_event_id: event.id,
    },
    ...(runAfter !== undefined ? { runAfter } : {}),
  });
  log.info('drain: job de turno enfileirado', { event_id: event.id, job_id: job.id, deduped });
  // A soneca do laço da fila sai na hora (#3688): sem este aviso o job esperaria
  // até o fim do teto do recuo (quando ligado), e é a PRIMEIRA resposta de um lead que escreveu
  // depois de um período parado que paga a conta. Depois do enfileiramento, e só
  // nele: avisar antes deixaria o laço consultar o relógio e não ver o job.
  knobs.acordarFila?.();
  return 'processado';
}

/**
 * Quantos ciclos OCIOSOS sem um tick CONCLUÍDO contam como "laço parado"
 * (issue #2505).
 *
 * O default da instalação é `idle = 15s`, então 20 ciclos = **5 minutos**. A
 * folga é dupla de propósito: o carimbo é do FIM do tick (um tick de lote cheio
 * demora mais que o intervalo) e o custo de um falso positivo é um `/healthz`
 * vermelho, que o operador investiga — contra o custo do falso negativo, que
 * foi o do caso real: DOIS DIAS de laço travado com o worker dizendo `ok`
 * (#2501), enquanto as mensagens se acumulavam com `attempts=0`.
 */
export const TICKS_PARA_PARADO = 20;

/**
 * O carimbo de vida do laço, lido de FORA dele (hoje: o `/healthz` do worker).
 *
 * `ultimoTickEm = null` significa "o laço ainda não começou" — não é parado,
 * é um worker que acabou de subir; quem nunca deu a primeira volta não pode
 * ser acusado de ter parado. `limiteMs` nasce do `idleIntervalMs` DA INSTALAÇÃO
 * quando o laço começa, então um `.env` com intervalo maior ganha o mesmo
 * número de ciclos de folga, não o mesmo relógio.
 */
interface VigilanciaDoDrain {
  ultimoTickEm: number | null;
  limiteMs: number | null;
}
const vigilancia: VigilanciaDoDrain = { ultimoTickEm: null, limiteMs: null };

/** Resposta do `/healthz` sobre o laço do drain — nomes para a máquina. */
export interface ProntidaoDoDrainDaIa {
  /** Há quanto tempo o último tick concluiu; `null` = o laço não começou. */
  ultimo_tick_ha_ms: number | null;
  /** Limite em vigor (idle da instalação × `TICKS_PARA_PARADO`); `null` antes de começar. */
  limite_ms: number | null;
  /** `true` quando o carimbo passou do limite — o laço travou NO MEIO de um tick. */
  parado: boolean;
}

/**
 * Estado do laço para quem pergunta de fora. Nunca lança: o `/healthz` não pode
 * ser o motivo de o worker cair — quem está degradado não ganha o direito de
 * derrubar o resto.
 *
 * `agoraMs` é injetável para o teste medir o LIMITE sem esperar o relógio.
 */
export function prontidaoDoDrainDaIa(agoraMs: number = Date.now()): ProntidaoDoDrainDaIa {
  const { ultimoTickEm, limiteMs } = vigilancia;
  if (ultimoTickEm === null || limiteMs === null) {
    return { ultimo_tick_ha_ms: null, limite_ms: limiteMs, parado: false };
  }
  const ha = Math.max(0, agoraMs - ultimoTickEm);
  return { ultimo_tick_ha_ms: ha, limite_ms: limiteMs, parado: ha > limiteMs };
}

/** Só para teste: volta ao estado de "o laço ainda não começou". */
export function _reiniciarVigilanciaDoDrain(): void {
  vigilancia.ultimoTickEm = null;
  vigilancia.limiteMs = null;
}

/** Loop do drain — polling com backoff adaptativo (ocioso = tick mais lento). */
export async function runDrainLoop(
  pool: pg.Pool,
  knobs: DrainKnobs,
  log: Logger,
  signal: AbortSignal,
): Promise<void> {
  // O carimbo de vida (#2505): o limite nasce do intervalo ocioso DA INSTALAÇÃO
  // e o carimbo é gravado quando o laço começa e ao FIM de cada volta — um tick
  // que travar deixa o carimbo envelhecer, e é isso que o `/healthz` lê.
  vigilancia.limiteMs = knobs.idleIntervalMs * TICKS_PARA_PARADO;
  vigilancia.ultimoTickEm = Date.now();
  while (!signal.aborted) {
    let drained = 0;
    try {
      drained = await drainTick(pool, knobs, log);
    } catch (err) {
      log.error('drain: tick falhou', {
        error: (err instanceof Error ? err.message : String(err)).slice(0, 300),
      });
    }
    // O laço DEU A VOLTA — sucesso e erro capturado são a mesma prova de vida.
    vigilancia.ultimoTickEm = Date.now();
    if (signal.aborted) break;
    // Lote CHEIO é sinal de backlog: há mais evento esperando do que caberia no
    // lote, e pagar o intervalo antes de voltar só empurra a fila para frente.
    // Ocioso e lote parcial mantêm o ritmo de sempre — este ramo não muda o
    // custo de quem não tem atendimento nenhum.
    const waitMs =
      drained >= knobs.batchSize ? 0 : drained > 0 ? knobs.intervalMs : knobs.idleIntervalMs;
    await new Promise<void>((resolve) => {
      // O listener é REMOVIDO no fim de cada espera. Sem isso, um loop de dias
      // acumula um listener por tick no mesmo AbortSignal — vazamento que só
      // aparece como memória crescendo no worker, sem erro nenhum.
      const finish = (): void => {
        clearTimeout(timer);
        signal.removeEventListener('abort', finish);
        resolve();
      };
      const timer = setTimeout(finish, waitMs);
      signal.addEventListener('abort', finish, { once: true });
    });
  }
}