/**
 * Worker de roteamento de conversas (G5-02 — AT-03, spec 13 §5).
 *
 * Doutrina: trigger NUNCA faz HTTP. A entrada da conversa na fila emite
 * `conversation.routing_requested` em `event_log` (trigger AFTER INSERT em
 * conversations, migration 0040); ESTE worker (cron TS) drena o evento, resolve
 * o modo da org e atribui — como service_role (admin client bypassa RLS, então
 * TODA query filtra organization_id vindo do payload confiável do evento).
 *
 * Mecânica de claim = mesma do agent-dispatcher: CAS status pending→processing,
 * `consumed_by` marca o consumidor, `next_attempt_at`/`attempts` reenfileiram.
 * A lógica de decisão (manual/round_robin/no-eligible/replay) é pura em
 * lib/routing/decide.ts; aqui só há I/O.
 */
import type { SupabaseClient } from "@supabase/supabase-js";
import { z } from "zod";
import { createAdminClient } from "@/lib/supabase/admin";
import { filtrarCandidatosExcluidos } from "@/lib/escalacao/devolucao-ao-rodizio";
import { logger } from "@/lib/logger";
import { moduloLigado } from "@/lib/instalacao/modulos";
import {
  decideRouting,
  resolverRoteamentoPorNumero,
  type RegraDoNumero,
} from "@/lib/routing/decide";
import { motivoParaNaoRotear, type MotivoDoCaso } from "@/lib/routing/caso-da-conversa";
import { lerEsperaPelaIA, lerQualificacaoDaIA, prazoDaEspera } from "@/lib/routing/qualificacao-da-ia";
import { loadEligibleAttendants, InvalidRoutingChannel, InvalidRoutingSector } from "@/lib/routing/eligibles";
import { routingConfigSchema } from "@/lib/schemas/routing";

export const ROUTING_WORKER_KEY = "worker.routing.v1";
export const ROUTING_EVENT_TYPE = "conversation.routing_requested";

const DEFAULT_BATCH_SIZE = 100;

export type RoutingOutcome =
  | "assigned"
  | "skipped_manual"
  | "skipped_already_assigned"
  | "skipped_unsupported_mode"
  | "requeued_no_eligible"
  | "skipped_invalid_channel"
  | "skipped_invalid_sector"
  | "skipped_conv_missing"
  | "skipped_invalid_payload"
  | "skipped_contato_pessoal"
  /** #3937: contato do tipo `fornecedor` não entra na distribuição automática. */
  | "skipped_contato_fornecedor"
  /** #3717: o caso mais recente da conversa não pede uma pessoa agora. */
  | "skipped_caso_sem_humano"
  /**
   * #3966: o número pediu para esperar a IA qualificar. O pedido NÃO foi
   * consumido — ficou guardado até o prazo máximo, ou até a passagem para humano.
   */
  | "waiting_ia_qualificando"
  | "assign_lost_race"
  | "error";

export interface RoutingSummary {
  batch_size: number;
  outcomes: Record<RoutingOutcome, number>;
  errors: string[];
}

/**
 * A linha como ela sai da VARREDURA (#3878, item 3): sem `payload`/`metadata`.
 *
 * O corpo do evento só existe depois do claim, em `EventRow` — antes disso
 * ninguém pode lê-lo, e é o tipo que impede alguém de "só usar o payload" a
 * partir da varredura e reintroduzir o egress que este corte tirou.
 */
interface EventoDaFila {
  id: string;
  organization_id: string;
  consumed_by: string[];
  attempts: number;
}

interface EventRow extends EventoDaFila {
  payload: Record<string, unknown>;
  metadata: Record<string, unknown> | null;
  claimed_at?: string;
}

const EMPTY_OUTCOMES = (): Record<RoutingOutcome, number> => ({
  assigned: 0,
  skipped_manual: 0,
  skipped_already_assigned: 0,
  skipped_unsupported_mode: 0,
  requeued_no_eligible: 0,
  skipped_invalid_channel: 0,
  skipped_invalid_sector: 0,
  skipped_conv_missing: 0,
  skipped_invalid_payload: 0,
  skipped_contato_pessoal: 0,
  skipped_contato_fornecedor: 0,
  skipped_caso_sem_humano: 0,
  waiting_ia_qualificando: 0,
  assign_lost_race: 0,
  error: 0,
});

export interface RoutingWorkerOptions {
  batchSize?: number;
  /** Relógio injetável (testes) — usado na elegibilidade (horário) e no backoff. */
  now?: Date;
}

export async function runRoutingWorker(opts: RoutingWorkerOptions = {}): Promise<RoutingSummary> {
  const admin = createAdminClient();
  const now = opts.now ?? new Date();
  const batchSize = Math.min(Math.max(opts.batchSize ?? DEFAULT_BATCH_SIZE, 1), 500);
  const summary: RoutingSummary = { batch_size: 0, outcomes: EMPTY_OUTCOMES(), errors: [] };

  // Recupera execução interrompida sem criar outro evento. O timestamp do
  // claim cerca o fechamento: um worker antigo não conclui a nova tentativa.
  const { data: abandoned, error: recoveryError } = await admin
    .from("event_log")
    .select("id, organization_id, updated_at")
    .eq("event_type", ROUTING_EVENT_TYPE)
    .eq("status", "processing")
    .lte("updated_at", new Date(now.getTime() - 300_000).toISOString())
    .limit(batchSize);
  if (recoveryError) {
    summary.errors.push("routing_recovery_failed");
    return summary;
  }
  for (const previous of abandoned ?? []) {
    const { error } = await admin
      .from("event_log")
      .update({ status: "pending", next_attempt_at: now.toISOString() })
      .eq("organization_id", previous.organization_id)
      .eq("id", previous.id)
      .eq("status", "processing")
      .eq("updated_at", previous.updated_at);
    if (error) summary.errors.push("routing_recovery_failed");
  }
  // ─── A VARREDURA NÃO LEVA O CORPO DO EVENTO (#3878, item 3) ────────────────
  //
  // `payload`/`metadata` são o JSONB gordo da linha, e este pull traz até 100
  // eventos por tique só para decidir QUAIS reclamar. O corpo é de quem chega
  // ao `processEvent` — um por claim — e vem no `update ... select` do claim
  // (`claimEvent`), que já acontecia: uma linha por evento processado em vez de
  // 100 por tique, sem consulta a mais.
  const { data: rawEvents, error: pullErr } = await admin
    .from("event_log")
    .select("id, organization_id, consumed_by, attempts")
    .eq("event_type", ROUTING_EVENT_TYPE)
    .eq("status", "pending")
    .or(`next_attempt_at.is.null,next_attempt_at.lte.${now.toISOString()}`)
    .order("created_at", { ascending: true })
    .limit(batchSize);

  if (pullErr) {
    summary.errors.push(`event_log_pull_failed: ${pullErr.message}`);
    return summary;
  }

  const naFila = (rawEvents ?? []).filter(
    (e) => !(Array.isArray(e.consumed_by) && e.consumed_by.includes(ROUTING_WORKER_KEY)),
  ) as EventoDaFila[];

  for (const candidato of naFila) {
    // O claim devolve a linha COM o corpo; quem perde a corrida (outro worker
    // pegou) não custa um byte de `payload`/`metadata`.
    const event = await claimEvent(candidato);
    if (!event) continue;
    summary.batch_size += 1;
    try {
      const outcome = await processEvent(event, now);
      summary.outcomes[outcome] += 1;
    } catch (err) {
      summary.outcomes.error += 1;
      const detail = err instanceof Error ? err.message : String(err);
      summary.errors.push(`${event.id}:${detail}`);
      logger.error("[routing-worker] processEvent threw", {
        event_id: event.id,
        organization_id: event.organization_id,
        error: detail,
      });
      await requeueEvent(event, now, event.attempts + 1, { error: detail.slice(0, 200) });
    }
  }

  return summary;
}

async function processEvent(event: EventRow, now: Date): Promise<RoutingOutcome> {
  const admin = createAdminClient();
  const payload = event.payload ?? {};
  const orgId = strOrNull(payload.organization_id) ?? event.organization_id;
  const conversationId = strOrNull(payload.conversation_id);

  if (
    !conversationId ||
    !z.string().uuid().safeParse(conversationId).success ||
    orgId !== event.organization_id
  ) {
    await markDone(event, "skipped_invalid_payload");
    return "skipped_invalid_payload";
  }

  const { data: conv, error: convError } = await admin
    .from("conversations")
    .select(
      "id, organization_id, contact_id, channel_session_id, assigned_to_user_id, status, metadata, last_inbound_at, sector_id, bot_silenced_until, last_handoff_at, active_ai_agent_id",
    )
    .eq("id", conversationId)
    .eq("organization_id", orgId)
    .maybeSingle();

  if (convError) throw new Error(convError.message);
  if (!conv || !["open", "pending", "claimed", "ai_handling"].includes(conv.status)) {
    await markDone(event, "skipped_conv_missing");
    return "skipped_conv_missing";
  }

  // Contato pessoal não distribui (spec 21, caminho 9): depois da leitura da
  // conversa — o pulo precisa do contato, que só ela tem.
  //
  // #3937: a MESMA leitura, na MESMA ordem, decide o contato FORNECEDOR
  // (`contacts.contact_type`, issue #3005) — promoção de operadora, comunicado
  // de companhia aérea e recado de grupo comercial chegam pelo mesmo número do
  // cliente e ocupavam carga e capacidade do rodízio. Um contato, uma query,
  // duas réguas: pessoal primeiro porque a dele é a mais forte (tira a conversa
  // da operação inteira); fornecedor depois, e só ele sai da DISTRIBUIÇÃO — a
  // conversa continua no Inbox, sem dono, como hoje.
  const { data: contato } = await admin
    .from("contacts")
    .select("is_personal, contact_type")
    .eq("id", conv.contact_id)
    .eq("organization_id", orgId)
    .maybeSingle();
  const contatoDaConversa = contato as
    | { is_personal?: boolean; contact_type?: string | null }
    | null;
  if (contatoDaConversa?.is_personal === true) {
    await markDone(event, "skipped_contato_pessoal");
    return "skipped_contato_pessoal";
  }
  // Fornecedor: `contact_type` nasce 'cliente' no banco (migration 0763), então
  // quem nunca falou de tipo continua distribuindo. Fornecedor E pessoal já saiu
  // acima — a precedência é a de hoje e não há filtro duplo divergente.
  if (contatoDaConversa?.contact_type === "fornecedor") {
    await markDone(event, "skipped_contato_fornecedor");
    return "skipped_contato_fornecedor";
  }

  // O CASO DA CONVERSA (#3717): a pergunta que a fila nunca fazia. Fica AQUI —
  // depois de a conversa existir no pool e de não ser contato pessoal (não
  // gastar query em quem já saiu) —, e antes de qualquer leitura de config ou
  // de candidatos: o caso barra a conversa ANTES de o rodízio montar a lista.
  const motivoDoCaso = await motivoDoCasoParaRoteamento(admin, orgId, {
    id: conversationId,
    last_inbound_at: (conv as { last_inbound_at?: string | null }).last_inbound_at ?? null,
  });

  // organizations.settings.routing → Zod (default manual; knobs = config, não hardcode).
  const { data: org, error: orgError } = await admin
    .from("organizations")
    .select("settings")
    .eq("id", orgId)
    .maybeSingle();
  if (orgError) throw new Error(orgError.message);
  if (!org) throw new Error("routing_organization_missing");
  const settings = (org?.settings ?? {}) as { routing?: unknown };
  const config = routingConfigSchema.parse(settings.routing ?? {});

  // Regra POR NÚMERO (#2331): o canal pode ter modo próprio e um atendente de
  // destino padrão. Número sem regra (policy ausente ou colunas nulas) cai no
  // comportamento da organização — resolverRoteamentoPorNumero é pura e o
  // resultado é o mesmo que o worker já usava quando não havia coluna.
  const regraDoNumero = await carregarRegraDoNumero(admin, orgId, conv.channel_session_id);
  const efetivo = resolverRoteamentoPorNumero(config, regraDoNumero);

  const alreadyAssigned = Boolean(conv.assigned_to_user_id);
  /**
   * (migration 9062) A FILA da conversa. Quando existe, ela manda em QUEM pode
   * receber — é o que transforma "setor que agrupa" em "fila que distribui".
   * Sem fila (`null`), tudo abaixo é o comportamento de sempre, byte a byte.
   */
  // (revisão do mantenedor, 31/10) O `sector_id` PRECISA vir no select acima.
  // Sem ele o cast antigo devolvia sempre `null` e a conversa com fila era
  // distribuída pelo número — a fila existia e não valia nada.
  const sectorIdBruto = strOrNull(conv.sector_id);
  // Módulo `setores` desligado ⇒ a fila não manda: quem desligou o módulo não
  // espera que a conversa pare de ser distribuída pelo número.
  const sectorId = sectorIdBruto && (await moduloLigado(admin, "setores")) ? sectorIdBruto : null;
  /**
   * O destino padrão NÃO depende da lista de elegíveis do modo: é entrega
   * direta. Mas ele não pode furar a fila: se a conversa tem fila e o destino
   * padrão do número não é membro dela, o destino padrão não vale aqui (a fila
   * é a regra mais específica) e o modo distribui normalmente. Sem esta
   * checagem, o claim recusaria (candidate_not_in_sector) e a conversa
   * ficaria girando em backoff para sempre.
   */
  const viaDestinoPadrao = !alreadyAssigned && Boolean(efetivo.defaultAssigneeUserId)
    && (!sectorId || await ehMembroDaFila(admin, orgId, sectorId, efetivo.defaultAssigneeUserId as string));

  // A IA AINDA ESTÁ QUALIFICANDO? (#3966) — só é perguntado quando o NÚMERO
  // ligou a opção (`ai_wait_minutes`, desligada de fábrica; doc 311). Com ela
  // desligada nenhuma leitura a mais acontece e o caminho é o de sempre.
  //
  // Fica DEPOIS do destino padrão do número ("direto para uma pessoa" não
  // espera, #2331) e só vale onde há distribuição automática. A leitura é
  // fail-open (ver `qualificacao-da-ia.ts`).
  //
  // O PRAZO: a espera começa na primeira vez que este pedido é adiado e a marca
  // vai no `metadata` do evento. Venceu sem a IA passar a conversa ⇒ ela segue
  // para a distribuição logo abaixo e a Central avisa UMA vez.
  let esperaPelaIAAte: string | null = null;
  const minutosDeEspera = regraDoNumero?.ai_wait_minutes ?? null;
  if (
    minutosDeEspera !== null &&
    !alreadyAssigned &&
    !motivoDoCaso &&
    !viaDestinoPadrao &&
    (efetivo.mode === "round_robin" || efetivo.mode === "load")
  ) {
    const qualificando = await lerQualificacaoDaIA(admin, orgId, {
      channel_session_id: conv.channel_session_id,
      active_ai_agent_id:
        (conv as { active_ai_agent_id?: string | null }).active_ai_agent_id ?? null,
      bot_silenced_until:
        (conv as { bot_silenced_until?: string | null }).bot_silenced_until ?? null,
      last_handoff_at: (conv as { last_handoff_at?: string | null }).last_handoff_at ?? null,
      assigned_to_user_id: conv.assigned_to_user_id ?? null,
    }, now);
    const marca = lerEsperaPelaIA(event.metadata);
    if (qualificando) {
      const prazo = prazoDaEspera(marca, now, minutosDeEspera);
      if (!prazo.vencida) {
        esperaPelaIAAte = prazo.ate;
        event.metadata = { ...(event.metadata ?? {}), espera_pela_ia: { desde: prazo.desde } };
      } else if (!prazo.avisada && (await avisarEsperaVencida(orgId, conversationId, minutosDeEspera))) {
        event.metadata = {
          ...(event.metadata ?? {}),
          espera_pela_ia: { desde: prazo.desde, avisada: true },
        };
      }
    } else if (marca) {
      // A IA saiu do caminho (passou para humano, foi pausada): a espera
      // acabou. Sem limpar, um episódio novo de qualificação neste MESMO pedido
      // herdaria o relógio antigo e nasceria vencido.
      event.metadata = { ...(event.metadata ?? {}), espera_pela_ia: null };
    }
  }

  let eligibles: Awaited<ReturnType<typeof loadEligibleAttendants>> = [];
  if (
    !alreadyAssigned &&
    esperaPelaIAAte === null &&
    !viaDestinoPadrao &&
    (efetivo.mode === "round_robin" || efetivo.mode === "load")
  ) {
    try {
      eligibles = await loadEligibleAttendants(admin, orgId, now, sectorId
        ? { kind: "sector", sectorId, channelSessionId: conv.channel_session_id }
        : { kind: "conversation_channel", channelSessionId: conv.channel_session_id });
    } catch (error) {
      if (error instanceof InvalidRoutingSector) {
        // A fila sumiu entre o enfileiramento e o processamento. A conversa não
        // perde nada: o `set null` da FK já a deixou sem fila; o próximo evento
        // a distribui pelo caminho de sempre.
        await notice(orgId, conversationId, "invalid_sector");
        await markDone(event, "skipped_invalid_sector");
        return "skipped_invalid_sector";
      }
      if (!(error instanceof InvalidRoutingChannel)) throw error;
      await notice(orgId, conversationId, "invalid_channel");
      await markDone(event, "skipped_invalid_channel");
      return "skipped_invalid_channel";
    }
  }

  // A EXCLUSÃO DE 24 H DA DEVOLUÇÃO AO RODÍZIO (issue #2985). Quem recebeu esta
  // conversa do rodízio e não respondeu no prazo fica fora dela por um dia
  // (`conversations.metadata.devolucao_ao_rodizio`, escrito pelo cron
  // `rodizio-devolucao`). Sem este corte, o cron devolveria a conversa e o
  // próximo tick deste worker a entregaria à MESMA pessoa — numa instalação com
  // um atendente só, sempre ela. A regra da marca é pura e mora em
  // `lib/escalacao/devolucao-ao-rodizio.ts`.
  const elegiveis = filtrarCandidatosExcluidos(eligibles, conv.metadata, now.getTime());

  const action = decideRouting({
    mode: efetivo.mode,
    alreadyAssigned,
    eligibles: elegiveis,
    config,
    // (revisão do mantenedor, 31/10) O destino padrão só entra no decide quando
    // ele PODE receber. `decideRouting` o prefere incondicionalmente (linha 90);
    // passar o destino padrão barrado pela fila fazia o worker pedir um claim que
    // o banco recusa com `candidate_not_in_sector` — a conversa girava a cada
    // minuto, para sempre, sem aviso. Com fila, quem decide é a lista elegível.
    defaultAssigneeUserId: viaDestinoPadrao ? efetivo.defaultAssigneeUserId : null,
    attempts: event.attempts,
    now,
    motivoDoCaso,
    esperaPelaIAAte,
  });

  switch (action.kind) {
    case "assign": {
      // O destino padrão não passou pela régua de elegibilidade (não é
      // rodízio): vai sem snapshot de jornada — a RPC checa disponibilidade e
      // capacidade, mas não compara jornada de quem ninguém escolheu.
      const schedule = viaDestinoPadrao
        ? null
        : (eligibles.find((candidate) => candidate.userId === action.userId)?.scheduleSnapshot ??
          {});
      const { data: result, error } = await admin.rpc("fn_channel_routing_claim", {
        p_org: orgId,
        p_conversation: conversationId,
        p_channel: conv.channel_session_id,
        p_user: action.userId,
        p_reason: "routing",
        p_schedule: schedule,
      });
      if (error) throw new Error(`fn_channel_routing_claim: ${error.message}`);
      if (result === "already_assigned" || result === "conversation_changed") {
        await markDone(event, "assign_lost_race");
        return "assign_lost_race";
      }
      if (result !== "assigned") {
        if (
          ![
            "candidate_revoked",
            "candidate_not_allowed",
            "capacity_changed",
            // (9062) a fila mudou entre a leitura e o claim: reenfileira, não quebra.
            "candidate_not_in_sector",
          ].includes(String(result))
        ) {
          throw new Error("routing_claim_invalid_result");
        }
        await requeueEvent(event, now, Math.min(event.attempts + 1, config.max_retries), {
          reason: result,
        });
        return "requeued_no_eligible";
      }
      const leads = await adotarLeadsDoContato(
        admin,
        orgId,
        strOrNull((conv as { contact_id?: unknown }).contact_id),
        action.userId,
      );
      await markDone(event, "assigned", {
        assigned_to_user_id: action.userId,
        leads_adotados: leads,
      });
      return "assigned";
    }
    case "skip": {
      const outcome: RoutingOutcome =
        action.reason === "manual_mode"
          ? "skipped_manual"
          : action.reason === "already_assigned"
            ? "skipped_already_assigned"
            : action.reason === "caso_aguardando_cliente" || action.reason === "caso_encerrado"
              ? "skipped_caso_sem_humano"
              : "skipped_unsupported_mode";
      if (action.reason.startsWith("unsupported_mode")) {
        logger.warn("[routing-worker] modo não suportado (post-MVP) — no-op", {
          organization_id: orgId,
          mode: config.mode,
        });
      }
      // #3717: o motivo do caso vai para o `metadata` do evento. As duas causas
      // ("esperando o cliente" × "caso concluído") viram o MESMO outcome de
      // contagem — o que se mede é "quantas conversas a fila recusou por causa
      // do caso" —, e a distinção fica no evento para quem for investigar.
      await markDone(
        event,
        outcome,
        outcome === "skipped_caso_sem_humano" ? { motivo_do_caso: action.reason } : {},
      );
      return outcome;
    }
    case "requeue": {
      if (action.attempts >= config.max_retries) await notice(orgId, conversationId, "no_eligible");
      await requeueEvent(
        event,
        now,
        action.attempts,
        { reason: "no_eligible" },
        action.nextAttemptAt,
      );
      return "requeued_no_eligible";
    }
    case "wait_ai": {
      // Um reagendamento só, para o PRAZO — não a cada minuto. O pedido some da
      // varredura até lá, então não ocupa o lote nem gera rodada de trabalho.
      // Não conta tentativa e não avisa: nada falhou. A passagem para humano
      // traz `next_attempt_at` para agora (`fn_request_channel_routing`).
      await requeueEvent(
        event,
        now,
        event.attempts,
        { reason: "ia_qualificando" },
        action.nextAttemptAt,
      );
      return "waiting_ia_qualificando";
    }
  }
}

/** Título fixo: é por ele que quem lê a Central reconhece o aviso. */
const TITULO_DA_ESPERA_VENCIDA = "A IA não passou a conversa no prazo";

/**
 * O aviso do prazo vencido (#3966). Insere direto, sem `on conflict`: quem
 * impede a repetição é a marca `avisada` no `metadata` do pedido. Devolve
 * `false` quando não gravou — o worker então tenta de novo na próxima rodada —,
 * e NUNCA lança: o aviso não pode segurar a conversa fora da distribuição.
 */
async function avisarEsperaVencida(
  orgId: string,
  conversationId: string,
  minutos: number,
): Promise<boolean> {
  const { error } = await createAdminClient().from("agent_inbox_items").insert({
    organization_id: orgId,
    kind: "routing_ia_wait_expired",
    severity: "warn",
    title: TITULO_DA_ESPERA_VENCIDA,
    body:
      `A IA atendeu esta conversa por ${minutos} minutos, o prazo definido para este número, ` +
      `e não a passou para a equipe. A conversa entrou na distribuição. ` +
      `Se isso se repetir, confira o agente de IA deste número e o prazo em Configurações → Atendimento.`,
    ref_kind: "conversation",
    ref_id: conversationId,
  });
  if (error) {
    logger.warn("[routing-worker] aviso de espera vencida não foi gravado", {
      organization_id: orgId,
      conversation_id: conversationId,
      error: error.message,
    });
    return false;
  }
  return true;
}

// --- event lifecycle (mesma mecânica do agent-dispatcher; status ∈ pending|processing|done|dead) ---

async function claimEvent(naFila: EventoDaFila): Promise<EventRow | null> {
  const admin = createAdminClient();
  const claimedAt = new Date().toISOString();
  // O corpo do evento vem DAQUI (#3878, item 3): a varredura acima só trouxe
  // `id/organization_id/consumed_by/attempts`, e este `update ... select` — que
  // já existia para carimbar o claim e ler o `updated_at` — devolve a
  // representação da linha reclamada. Uma linha por evento processado, no mesmo
  // round-trip; nenhum evento que perde o claim paga o JSONB.
  const { data, error } = await admin
    .from("event_log")
    .update({ status: "processing", updated_at: claimedAt })
    .eq("id", naFila.id)
    .eq("organization_id", naFila.organization_id)
    .eq("status", "pending")
    .select("id, updated_at, payload, metadata")
    .maybeSingle();
  if (error) {
    logger.warn("[routing-worker] claim failed", { event_id: naFila.id, error: error.message });
    return null;
  }
  if (!data) return null;
  return {
    ...naFila,
    payload: (data.payload ?? {}) as Record<string, unknown>,
    metadata: (data.metadata ?? null) as Record<string, unknown> | null,
    claimed_at: data.updated_at,
  };
}

async function markDone(
  event: EventRow,
  outcome: RoutingOutcome,
  extra: Record<string, unknown> = {},
): Promise<void> {
  const admin = createAdminClient();
  const consumed = Array.isArray(event.consumed_by) ? event.consumed_by.slice() : [];
  if (!consumed.includes(ROUTING_WORKER_KEY)) consumed.push(ROUTING_WORKER_KEY);
  const { error } = await admin
    .from("event_log")
    .update({
      status: "done",
      consumed_by: consumed,
      metadata: { ...(event.metadata ?? {}), outcome, handled_by: ROUTING_WORKER_KEY, ...extra },
      updated_at: new Date().toISOString(),
    })
    .eq("id", event.id)
    .eq("organization_id", event.organization_id)
    .eq("status", "processing")
    .eq("updated_at", event.claimed_at!);
  if (error)
    logger.warn("[routing-worker] markDone failed", { event_id: event.id, error: error.message });
}

async function requeueEvent(
  event: EventRow,
  now: Date,
  attempts: number,
  extra: Record<string, unknown>,
  nextAttemptAt?: string,
): Promise<void> {
  const admin = createAdminClient();
  const next = nextAttemptAt ?? new Date(now.getTime() + 60_000).toISOString();
  const { error } = await admin
    .from("event_log")
    .update({
      status: "pending",
      attempts,
      next_attempt_at: next,
      metadata: { ...(event.metadata ?? {}), last_requeue: { ...extra, at: now.toISOString() } },
      updated_at: new Date().toISOString(),
    })
    .eq("id", event.id)
    .eq("organization_id", event.organization_id)
    .eq("status", "processing")
    .eq("updated_at", event.claimed_at!);
  if (error)
    logger.warn("[routing-worker] requeue failed", { event_id: event.id, error: error.message });
}

async function notice(orgId: string, conversationId: string, reason: string): Promise<void> {
  const { error } = await createAdminClient().rpc("fn_routing_unassigned_notice", {
    p_org: orgId,
    p_conversation: conversationId,
    p_reason: reason,
  });
  if (error) throw new Error(error.message);
}

function strOrNull(v: unknown): string | null {
  if (typeof v !== "string") return null;
  const t = v.trim();
  return t.length > 0 ? t : null;
}

/**
 * A pessoa atende esta fila? (migration 9062)
 *
 * Pergunta única usada para decidir se o destino padrão do número vale dentro
 * da fila. A mesma regra está no claim (`fn_channel_routing_claim`): aqui é
 * para NÃO montar um claim que o banco vai recusar; lá é a garantia de que
 * nenhum caminho entrega fora da fila.
 */
export async function ehMembroDaFila(
  admin: Pick<SupabaseClient, "from">,
  orgId: string,
  sectorId: string,
  userId: string,
): Promise<boolean> {
  const { data, error } = await admin
    .from("attendant_sector_members")
    .select("user_id")
    .eq("organization_id", orgId)
    .eq("sector_id", sectorId)
    .eq("user_id", userId)
    .maybeSingle();
  if (error) throw new Error(error.message);
  return Boolean(data);
}

/**
 * Lê a regra do NÚMERO (#2331) de `channel_routing_policies`.
 *
 * Ausência de policy = número sem regra própria, e é o estado de TODA
 * instalação existente: as duas colunas nasceram nulas. Devolver `null` (e não
 * um objeto default) é o que faz o worker cair no comportamento da organização.
 */
export async function carregarRegraDoNumero(
  admin: Pick<SupabaseClient, "from">,
  orgId: string,
  channelSessionId: string | null,
): Promise<RegraDoNumero | null> {
  if (!channelSessionId) return null;
  const { data, error } = await admin
    .from("channel_routing_policies")
    .select("distribution_mode, default_assignee_user_id, ai_wait_minutes")
    .eq("organization_id", orgId)
    .eq("channel_session_id", channelSessionId)
    .maybeSingle();
  if (error) throw new Error(error.message);
  if (!data) return null;
  const row = data as {
    distribution_mode: string | null;
    default_assignee_user_id: string | null;
    ai_wait_minutes?: number | null;
  };
  return {
    distribution_mode: (row.distribution_mode ?? null) as RegraDoNumero["distribution_mode"],
    default_assignee_user_id: row.default_assignee_user_id ?? null,
    ai_wait_minutes: row.ai_wait_minutes ?? null,
  };
}

/**
 * Lê o caso MAIS RECENTE da conversa e responde se ele tira a conversa da fila
 * (issue #3717). `null` = a fila pode distribuir.
 *
 * ## Por que a leitura é separada da regra
 *
 * `motivoParaNaoRotear` (`lib/routing/caso-da-conversa.ts`) é puro e não sabe o
 * que é um banco; aqui só existe a pergunta "qual é a linha mais recente?".
 * `agent_cases_conv_idx` (organization_id, conversation_id) existe desde a 0066
 * — não há índice novo a criar.
 *
 * ## Falha de leitura distribui (falha ABERTA)
 *
 * Se o SELECT do caso estourar, o worker segue para a decisão com `null`, e não
 * derruba o evento. Mesma escolha de `operator-turn.ts:686` ("na dúvida o aviso
 * sai, que é o erro barato"): subir a exceção reenfileiraria o evento e deixaria
 * a conversa sem dono — o desfecho caro. O erro sai no log, com a org e a
 * conversa, para não sumir.
 */
export async function motivoDoCasoParaRoteamento(
  admin: Pick<SupabaseClient, "from">,
  orgId: string,
  conversa: { id: string; last_inbound_at?: string | null },
): Promise<MotivoDoCaso | null> {
  const { data, error } = await admin
    .from("agent_cases")
    .select("status, closed_at, updated_at")
    .eq("organization_id", orgId)
    .eq("conversation_id", conversa.id)
    // O MAIS RECENTE, com desempate determinístico: `created_at` sozinho pode
    // empatar (dois casos no mesmo instante) e dois ticks leriam linhas
    // diferentes para a mesma conversa.
    .order("created_at", { ascending: false })
    .order("id", { ascending: false })
    .limit(1)
    .maybeSingle();
  if (error) {
    logger.warn("[routing-worker] não consegui ler o caso da conversa — seguindo para a fila", {
      organization_id: orgId,
      conversation_id: conversa.id,
      error: error.message,
    });
    return null;
  }
  const caso = data as {
    status: string;
    closed_at: string | null;
    updated_at: string | null;
  } | null;
  return motivoParaNaoRotear(
    caso
      ? {
          status: caso.status,
          encerradoEm: caso.closed_at ?? null,
          atualizadoEm: caso.updated_at ?? null,
        }
      : null,
    conversa.last_inbound_at ?? null,
  );
}

/**
 * O rodízio distribui CONVERSA. Sem esta função ele não distribui LEAD — e é o
 * lead que o dono do negócio olha no funil (issue #144).
 *
 * ## Por que isto existe
 *
 * `fn_conversation_assign` grava `conversations.assigned_to_user_id` e, desde a
 * 9208, adota o negócio quando o contato tem UM aberto sem dono. Esta função
 * cobre o contato com mais de um, que a 9208 não toca (por isso
 * `leads_adotados` conta só esses). Enquanto `visibility_mode` era `all` ou
 * `own_and_unassigned`, ninguém notava: o lead sem dono aparecia para todo
 * mundo. No modo `own` — exatamente o que a issue #144 pede — lead sem dono não
 * aparece para NINGUÉM. O atendente receberia a conversa e o funil dele ficaria
 * vazio. A restrição funcionaria e a fila não existiria.
 *
 * ## As três condições, e por que cada uma
 *
 * Adota só o lead que está `open` (negócio fechado não muda de dono), do
 * `contact_id` da conversa, e **sem dono nenhum** — nem humano (`owner_user_id`)
 * nem agente de IA (`owner_agent_id`). Roubar lead com dono transformaria o
 * rodízio numa reatribuição silenciosa a cada mensagem nova do mesmo contato,
 * que é o oposto de "cada um com a sua carteira".
 *
 * `owner_kind` entra junto com `owner_user_id` de propósito: o CHECK aceita
 * kind nulo, então um lead com dono e sem kind some do filtro por dono e das
 * métricas por kind — drift silencioso já registrado em `leads/_handler.ts`.
 *
 * Best-effort por decisão: a conversa JÁ foi atribuída quando chegamos aqui.
 * Deixar uma exceção subir faria o evento voltar para a fila e tentar atribuir
 * de novo uma conversa que já tem dono — troca um funil incompleto por um
 * retry inútil. O que falhou vira log, e a próxima mensagem do contato tenta
 * de novo.
 */
export async function adotarLeadsDoContato(
  admin: ReturnType<typeof createAdminClient>,
  orgId: string,
  contactId: string | null,
  userId: string,
): Promise<number> {
  if (!contactId) return 0;
  try {
    const { data, error } = await admin
      .from("crm_leads")
      .update({
        owner_user_id: userId,
        owner_kind: "user",
        updated_at: new Date().toISOString(),
      })
      .eq("organization_id", orgId)
      .eq("contact_id", contactId)
      .eq("status", "open")
      .is("owner_user_id", null)
      .is("owner_agent_id", null)
      .select("id");
    if (error) {
      logger.warn("[routing-worker] não consegui dar dono ao lead do contato", {
        organization_id: orgId,
        contact_id: contactId,
        error: error.message,
      });
      return 0;
    }
    return (data ?? []).length;
  } catch (err) {
    logger.warn("[routing-worker] adoção de lead lançou", {
      organization_id: orgId,
      contact_id: contactId,
      error: err instanceof Error ? err.message : String(err),
    });
    return 0;
  }
}