import type { SocialMessage } from "../social/parser";
import { ehCanalDeConversa } from "@/lib/channels/canais-de-conversa";
/**
 * Ingestão do canal intermediado: webhook → contato, conversa, mensagem.
 *
 * A leitura do payload é do módulo puro ao lado (`./webhook.ts`); aqui moram os
 * EFEITOS. A separação não é estética: o que decide (isto é mensagem? de quem?)
 * dá para provar sem banco, e o que escreve fica pequeno o bastante para caber
 * na cabeça.
 *
 * ─── O que este módulo existe para gravar ───────────────────────────────────
 *
 * `conversations.provider_conversation_id`. Sem ele o envio livre não funciona,
 * porque o endereço deste canal não se deriva do contato — e é AQUI, e só aqui,
 * que ele chega. Uma ingestão que grava a mensagem e esquece a thread deixa o
 * inbox mostrando a conversa e o operador sem conseguir responder.
 *
 * ─── Idempotência ───────────────────────────────────────────────────────────
 *
 * O provider reentrega quando não recebe 200 — e reentrega o MESMO evento. A
 * chave é `(organization_id, external_id)` no INSERT da mensagem, com captura
 * do `23505`, exatamente como o canal por QR faz. Sem isso, uma retentativa
 * duplica a mensagem no inbox do cliente.
 */
import type { SupabaseClient } from "@supabase/supabase-js";

import { logger } from "@/lib/logger";
import { audit } from "@/lib/audit";
import { corpoDaLocalizacao } from "@/lib/messaging/localizacao";
import { encontrarContatoPorTelefone } from "@/lib/channels/contato-por-telefone";
import { canonicalPhoneBR } from "@/lib/channels/phone-variants";
import { marcarConversaComMensagem } from "@/lib/channels/marcar-conversa";
import { emitirTelefoneGanho } from "@/lib/contacts/telefone-ganho";

import { extrairAtribuicaoMeta } from "@/lib/channels/atribuicao-de-anuncio-oficial";
import { estamparAtribuicaoDoContato, metadataDoAnuncio } from "@/lib/leads/atribuicao-de-anuncio";
import { ancoraCtwaDaMensagem } from "@/lib/leads/prova-meta-lead-submitted";
import { extrairEEstamparAtribuicaoGoogle } from "@/lib/plataformas-de-anuncio/google/atribuicao";
import {
  pausarIaPorAtendimentoManual,
  type FalhaDaPausa,
  type MotivoDaPausaNaoGravada,
} from "@/lib/escalacao/atendimento-manual";
import {
  ecoDaTransmissao,
  ligarEcoAoDestinatario,
  metadataDoEco,
  type EcoDaTransmissao,
} from "@/lib/campanhas/transmissao";
import { consumirRespostaDoSuporteNoWhatsApp } from "@/lib/escalacao/resposta-whatsapp-ao-caso";

import { aplicarEfeitosPosEntrada } from "../pos-entrada";

import { completarLocalizacao, pedirNovaBuscaDoPino, pinoFicouSemCoordenadas } from "./localizacao";
import { parseZernioInbound, type ZernioIdentity, type ZernioInboundMessage } from "./webhook";

export interface ZernioIngestResult {
  status: "ingested" | "duplicate" | "ignored" | "unknown_account";
  conversationId?: string;
  messageId?: string;
  /** Por que foi ignorado — vai para o log, e é o que se lê quando "sumiu". */
  reason?: string;
  /**
   * A mensagem ENTROU, mas a pausa da IA não foi garantida (#534). O motivo é
   * nomeado (`corrida_persistente`, `conversa_inexistente`, …) e a falha também
   * vai para o audit: a ingestão não pode dizer "ingested" limpo quando a IA
   * pode falar por cima do atendente.
   */
  pausaDaIANaoGravada?: MotivoDaPausaNaoGravada;
}

/** Aviso durável de que a pausa da IA não foi garantida (#534). */
function avisarPausaNaoGravada(falha: FalhaDaPausa): Promise<void> {
  return audit({
    action: "ai.pausa_nao_gravada",
    organizationId: falha.organizationId,
    resourceType: "conversation",
    resourceId: falha.conversationId,
    metadata: {
      canal: falha.canal ?? "zernio",
      motivo: falha.motivo,
      detalhe: falha.detalhe ?? null,
    },
  });
}

/**
 * Identidade → `wa_identity`, o mesmo vocabulário que o canal por QR já usa
 * (`phone:+E164` | `lid:<digits>`).
 *
 * O BSUID vira `lid:` de propósito: é a mesma NATUREZA de identificador — um id
 * opaco da plataforma que não é telefone — e reusar o prefixo faz o contato ser
 * o mesmo contato quando a pessoa aparece pelos dois canais. Inventar um
 * terceiro prefixo criaria dois contatos para uma pessoa só.
 */
export function waIdentityFrom(identity: ZernioIdentity): string | null {
  if (!identity.anchor) return null;
  return identity.anchor.kind === "phone"
    ? `phone:${identity.anchor.value}`
    : `lid:${identity.anchor.value}`;
}

/**
 * Grava um evento já lido e autenticado.
 *
 * Recebe o `channel_session_id` resolvido pela rota (que é quem conhece o
 * token): este módulo não descobre de quem é o webhook, só escreve o que já se
 * sabe de quem é.
 */
export async function ingestZernioInbound(
  admin: SupabaseClient,
  input: {
    organizationId: string;
    channelSessionId: string;
    payload: unknown;
    socialMessage?: SocialMessage;
    /**
     * Quem chama é o reprocessamento de uma entrada aceita: a linha do arquivo
     * não terminou, então um `duplicate` pode ser ingestão interrompida depois
     * do INSERT. A entrada ao vivo não passa isto: lá o `duplicate` é reentrega,
     * e a primeira entrega pode estar com os efeitos em curso.
     */
    retomada?: boolean;
  },
): Promise<ZernioIngestResult> {
  const lida = input.socialMessage ?? parseZernioInbound(input.payload);
  if (!lida) return { status: "ignored", reason: "evento_sem_interesse" };
  // O pino do WhatsApp traz as coordenadas no próprio evento (`metadata.location`);
  // aqui ele só ganha o endereço aproximado. Sem elas, a busca pela API é da nova
  // busca do pino, fora desta requisição. Rede social não manda pino.
  const msg = input.socialMessage ? lida : await completarLocalizacao(admin, input.organizationId, lida);

  // Evento de DESFECHO: a mensagem já existe (ou nem é nossa). Só atualiza o
  // status — inserir aqui criaria uma segunda linha para a mesma mensagem, uma
  // por transição de estado.
  if (msg.kind === "status") {
    const { data } = await admin
      .from("messages")
      .update({
        status: msg.status,
        ...(msg.errorReason ? { error_message: msg.errorReason, error_code: "zernio_error" } : {}),
      })
      .eq("organization_id", input.organizationId)
      .eq("channel_session_id", input.channelSessionId)
      .eq("external_id", msg.externalId)
      // Não rebaixa: `read` chegando depois de `delivered` é progresso, mas um
      // `delivered` atrasado depois de `read` voltaria o tique para trás. A
      // ordem de entrega do webhook não é garantida.
      .not("status", "in", "(read)")
      .select("id");
    const afetadas = (data ?? []).length;
    await carimbarHoraDoDesfecho(admin, input, msg);
    return afetadas > 0
      ? { status: "ingested", reason: `status_${msg.status}` }
      : { status: "ignored", reason: "mensagem_desconhecida" };
  }

  // ── O NÚMERO INTERNO DE AVISOS NÃO VIRA ATENDIMENTO ─────────────────────
  //
  // Antes da resolução pela thread E do upsert do contato — os DOIS caminhos
  // criam conversa, e é o nascimento dela que dispara o pedido de rodízio pelo
  // banco. Só o ramo do telefone é alcançável aqui: a âncora opaca deste canal
  // é do provedor, não o identificador de privacidade do WhatsApp, e casar por
  // ela exigiria um segundo campo na configuração sem consumidor nenhum hoje.
  if (
    await consumirRespostaDoSuporteNoWhatsApp(admin, {
      organizationId: input.organizationId,
      channelSessionId: input.channelSessionId,
      identidade: {
        kind: msg.identity.phone ? "phone" : "unknown",
        phone: msg.identity.phone,
        lid: null,
      },
      direction: msg.direction,
      texto: msg.text,
      externalId: msg.externalId,
    })
  ) {
    return { status: "ignored", reason: "numero_interno_de_aviso" };
  }

  // ─── A THREAD é a prova de identidade, e vem ANTES da âncora ─────────────
  //
  // O provider dá um id próprio à conversa. Se já existe uma com esse id, é a
  // MESMA pessoa — ele acabou de dizer isso. Perguntar de novo "quem é?" pela
  // âncora abre espaço para divergência, e ela apareceu em produção:
  //
  //   contato A: phone:+595982857447        ← criado pelo evento de SAÍDA
  //   contato B: lid:PY.1383674957068636    ← criado pelo evento de ENTRADA
  //
  // Mesma pessoa, dois contatos, duas conversas — porque a entrada traz o BSUID
  // (âncora preferida) e a saída só traz o telefone do participante. Resolver
  // pela thread primeiro fecha isso na origem, e de quebra deixa a ingestão
  // imune a qualquer identidade nova que o provider invente depois.
  const existente = await conversaPelaThread(
    admin,
    input.organizationId,
    msg.conversationId,
    input.channelSessionId,
  );
  if (existente) {
    /** Preenchido quando a pausa da IA falhou — sobe no resultado (#534). */
    let pausaDaIANaoGravada: MotivoDaPausaNaoGravada | undefined;
    if (input.socialMessage) {
      // A plataforma vai CRUA para uma coluna com CHECK. Num clone cujo banco
      // ainda não conhece esta rede, o INSERT volta 23514 — e como o provedor
      // REENTREGA o webhook, isso vira 500 eterno, com a tela mostrando a conta
      // ligada e a conversa nunca aparecendo. Recusar aqui troca o laço infinito
      // por uma linha de log que diz o nome da rede e o que falta.
      if (!ehCanalDeConversa(input.socialMessage.platform)) {
        logger.error("zernio: rede sem canal correspondente no banco — conversa não atualizada", {
          organization_id: input.organizationId,
          platform: input.socialMessage.platform,
          detalhe: "falta o valor no CHECK de conversations.channel (migration)",
        });
        return { status: "ignored", reason: "canal_desconhecido" };
      }
      const { error } = await admin
        .from("conversations")
        .update({ channel: input.socialMessage.platform })
        .eq("organization_id", input.organizationId)
        .eq("id", existente.id);
      if (error) throw new Error("social_conversation_update_failed");
    }
    const ecoNaExistente = await ecoDeTransmissao(admin, input.organizationId, msg, existente.contact_id);
    const inseridaNaExistente = await insertMessage(admin, {
      organizationId: input.organizationId,
      conversationId: existente.id,
      contactId: existente.contact_id,
      channelSessionId: input.channelSessionId,
      msg,
      eco: ecoNaExistente,
    });
    // O telefone pode chegar agora e faltar no contato — vale gravar.
    //
    // (#3315) E vale AVISAR: a ficha que estava sem número passou a ter um, que é
    // exatamente a transição que `contact.phone_added` nomeia. O `select("id")`
    // não é decoração — é ele que prova que ESTA gravação foi a que preencheu
    // (o filtro `.is("phone_number", null)` recusa a ficha que já tinha número, e
    // o dublê que só conta a chamada diria "emitiu" para os dois casos).
    if (msg.identity.phone) {
      const telefone = canonicalPhoneBR(msg.identity.phone);
      const { data: afetadas } = await admin
        .from("contacts")
        .update({ phone_number: telefone })
        .eq("id", existente.contact_id)
        .is("phone_number", null)
        .select("id");
      if (((afetadas as { id: string }[] | null) ?? []).length > 0) {
        await emitirTelefoneGanho(admin, {
          organizationId: input.organizationId,
          contactId: existente.contact_id,
          telefone,
          origem: "zernio.ingest.thread_existente",
        });
      }
    }
    if (inseridaNaExistente !== "duplicate") {
      await marcarConversa(admin, input.organizationId, existente.id, msg);
      if (msg.attachments[0]?.url) {
        await pedirPersistenciaDaMidia(
          admin,
          input.organizationId,
          existente.id,
          inseridaNaExistente,
        );
      }
      if (!input.socialMessage && pinoFicouSemCoordenadas(msg)) {
        await pedirNovaBuscaDoPino(admin, input.organizationId, inseridaNaExistente, msg);
      }
      await efeitosDaEntrada(
        admin,
        input,
        msg,
        existente.contact_id,
        existente.id,
        inseridaNaExistente,
      );
      if (ecoNaExistente) {
        await ligarEcoAoDestinatario(admin, ecoNaExistente, {
          id: inseridaNaExistente,
          conversationId: existente.id,
          enviadaEm: msg.sentAt,
        });
      }
      if (input.socialMessage && msg.direction === "outbound" && !ehDeTransmissao(msg)) {
        // #534: sem `avisar`, a falha da pausa viraria um `false` que ninguém lê.
        await pausarIaPorAtendimentoManual(admin, {
          organizationId: input.organizationId,
          conversationId: existente.id,
          canal: "zernio",
          avisar: (falha) => {
            pausaDaIANaoGravada = falha.motivo;
            return avisarPausaNaoGravada(falha);
          },
        });
      }
    }
    if (inseridaNaExistente === "duplicate") {
      return desfechoDaDuplicata(admin, input, msg, existente.id);
    }
    return {
      status: "ingested",
      conversationId: existente.id,
      messageId: inseridaNaExistente,
      ...(pausaDaIANaoGravada ? { pausaDaIANaoGravada } : {}),
    };
  }

  const identity = input.socialMessage
    ? `${input.socialMessage.platform}:${msg.accountId}:${input.socialMessage.participantId}`
    : waIdentityFrom(msg.identity);
  if (!identity) {
    // Evento sem âncora utilizável. Recusar é o certo: criar contato anônimo
    // faria a próxima mensagem da MESMA pessoa virar um segundo contato.
    return { status: "ignored", reason: "sem_identidade_utilizavel" };
  }

  const contactId = input.socialMessage
    ? await upsertSocialContact(admin, input.organizationId, identity, msg.identity.displayName)
    : await upsertContact(admin, input.organizationId, msg, identity);
  if (!contactId) return { status: "ignored", reason: "contato_nao_resolvido" };

  const conversationId = await upsertConversation(admin, {
    organizationId: input.organizationId,
    contactId,
    channelSessionId: input.channelSessionId,
    providerConversationId: msg.conversationId,
  });
  if (!conversationId) return { status: "ignored", reason: "conversa_nao_resolvida" };
  if (input.socialMessage) {
    // Mesma guarda do ramo acima: sem ela, rede nova = 23514 reentregue para
    // sempre. Ver `lib/channels/canais-de-conversa.ts`.
    if (!ehCanalDeConversa(input.socialMessage.platform)) {
      logger.error("zernio: rede sem canal correspondente no banco — conversa não atualizada", {
        organization_id: input.organizationId,
        platform: input.socialMessage.platform,
        detalhe: "falta o valor no CHECK de conversations.channel (migration)",
      });
      return { status: "ignored", reason: "canal_desconhecido" };
    }
    const { error } = await admin
      .from("conversations")
      .update({ channel: input.socialMessage.platform })
      .eq("organization_id", input.organizationId)
      .eq("id", conversationId);
    if (error) throw new Error("social_conversation_update_failed");
  }

  const eco = await ecoDeTransmissao(admin, input.organizationId, msg, contactId);
  const inserted = await insertMessage(admin, {
    organizationId: input.organizationId,
    conversationId,
    contactId,
    channelSessionId: input.channelSessionId,
    msg,
    eco,
  });

  if (inserted === "duplicate") return desfechoDaDuplicata(admin, input, msg, conversationId);

  await marcarConversa(admin, input.organizationId, conversationId, msg);
  if (msg.attachments[0]?.url) {
    await pedirPersistenciaDaMidia(admin, input.organizationId, conversationId, inserted);
  }
  if (!input.socialMessage && pinoFicouSemCoordenadas(msg)) {
    await pedirNovaBuscaDoPino(admin, input.organizationId, inserted, msg);
  }
  await efeitosDaEntrada(admin, input, msg, contactId, conversationId, inserted);
  if (eco) {
    await ligarEcoAoDestinatario(admin, eco, { id: inserted, conversationId, enviadaEm: msg.sentAt });
  }

  // SAÍDA feita por fora do CRM = uma pessoa respondeu o cliente à mão (celular,
  // outra plataforma na mesma conta). A IA para nesta conversa. O eco do nosso
  // próprio envio já saiu como `"duplicate"` acima. NÃO mexe na origem do lead.
  //
  // A TRANSMISSÃO do provedor não é pessoa nenhuma: é a campanha mandando a
  // lista de uma vez (migration 0712). Medido em 16/10/2026: sem esta exceção,
  // cada número novo da lista ficava com a IA calada pelo prazo do atendimento
  // manual — justamente quando as respostas chegam juntas.
  let pausaDaIANaoGravada: MotivoDaPausaNaoGravada | undefined;
  if (msg.direction === "outbound" && !ehDeTransmissao(msg)) {
    // #534: sem `avisar`, a falha da pausa viraria um `false` que ninguém lê.
    await pausarIaPorAtendimentoManual(admin, {
      organizationId: input.organizationId,
      conversationId,
      canal: "zernio",
      avisar: (falha) => {
        pausaDaIANaoGravada = falha.motivo;
        return avisarPausaNaoGravada(falha);
      },
    });
  }

  return {
    status: "ingested",
    conversationId,
    messageId: inserted,
    ...(pausaDaIANaoGravada ? { pausaDaIANaoGravada } : {}),
  };
}

async function desfechoDaDuplicata(
  admin: SupabaseClient,
  input: { organizationId: string; channelSessionId: string; retomada?: boolean },
  msg: ZernioInboundMessage,
  conversationId: string,
): Promise<ZernioIngestResult> {
  const retomou = input.retomada && (await retomarEfeitosDaEntrada(admin, input, msg));
  return retomou
    ? { status: "duplicate", conversationId, reason: "efeitos_retomados" }
    : { status: "duplicate", conversationId };
}

/**
 * A mensagem gravada cuja ingestão parou no meio: o processo morreu depois do
 * INSERT, e o `23505` da repetição não diz se os efeitos rodaram.
 *
 * ─── O marco é o pedido de despacho ────────────────────────────────────────
 *
 * `ai_agent.dispatch_requested` é o último efeito com consequência na ordem de
 * `pos-entrada.ts`, então a presença dele com esta mensagem diz que o resto
 * veio antes, e rodar de novo daria um segundo turno. Sem ele, os efeitos rodam
 * outra vez: contato pessoal, número interno e canal desativado nunca despacham
 * e chegam aqui sempre, mas neles a repetição só refaz o que é idempotente
 * (opt-out, primeiro toque, negócio por contato).
 *
 * O carimbo tem régua própria porque soma a não lida: só roda se a conversa
 * ainda não chegou à hora desta mensagem. Uma mensagem mais nova que já carimbou
 * deixa a não lida um a menos, que é o lado certo do erro.
 *
 * Só ENTRADA: saída duplicada é o eco do nosso envio, e não há como separá-la
 * da saída pelo celular interrompida. O que falhar aqui lança e a linha segue
 * pendente. `false` = não havia o que retomar.
 */
async function retomarEfeitosDaEntrada(
  admin: SupabaseClient,
  input: { organizationId: string; channelSessionId: string },
  msg: ZernioInboundMessage,
): Promise<boolean> {
  const { data, error } = await admin
    .from("messages")
    .select("id, conversation_id, contact_id, direction, sent_at")
    .eq("organization_id", input.organizationId)
    .eq("external_id", msg.externalId)
    .maybeSingle();
  if (error) throw new Error(`zernio_retomada_failed: ${error.message}`);
  const gravada = data as {
    id: string;
    conversation_id: string;
    contact_id: string;
    direction: string;
    sent_at: string;
  } | null;
  if (gravada?.direction !== "inbound") return false;

  if (await despachoJaPedido(admin, input.organizationId, gravada.id)) return false;

  const { data: conversa, error: erroDaConversa } = await admin
    .from("conversations")
    .select("last_inbound_at")
    .eq("organization_id", input.organizationId)
    .eq("id", gravada.conversation_id)
    .maybeSingle();
  if (erroDaConversa) throw new Error(`zernio_retomada_failed: ${erroDaConversa.message}`);
  const carimbo = (conversa as { last_inbound_at: string | null } | null)?.last_inbound_at ?? null;
  if (carimbo === null || Date.parse(carimbo) < Date.parse(gravada.sent_at)) {
    await marcarConversa(admin, input.organizationId, gravada.conversation_id, { ...msg, sentAt: gravada.sent_at });
  }

  if (msg.attachments[0]?.url) {
    await pedirPersistenciaDaMidia(admin, input.organizationId, gravada.conversation_id, gravada.id);
  }
  if (pinoFicouSemCoordenadas(msg)) {
    await pedirNovaBuscaDoPino(admin, input.organizationId, gravada.id, msg);
  }
  await efeitosDaEntrada(admin, input, msg, gravada.contact_id, gravada.conversation_id, gravada.id);
  return true;
}

/**
 * O marco de `retomarEfeitosDaEntrada`: esta mensagem já pediu o despacho do
 * agente? O reprocessamento usa a mesma régua na linha que passou da janela.
 * Lança se a leitura falhar.
 */
export async function despachoJaPedido(
  admin: SupabaseClient,
  organizationId: string,
  messageId: string,
): Promise<boolean> {
  const { data, error } = await admin
    .from("event_log")
    .select("id")
    .eq("organization_id", organizationId)
    .eq("entity_kind", "message")
    .eq("entity_id", messageId)
    .eq("event_type", "ai_agent.dispatch_requested")
    .limit(1)
    .maybeSingle();
  if (error) throw new Error(`zernio_retomada_failed: ${error.message}`);
  return data !== null;
}

/**
 * O contato já cadastrado desta mensagem, sem criar nada: pela thread, depois
 * pelo telefone. `null` = ninguém no CRM para esta pessoa. Lança se a leitura da
 * thread falhar.
 *
 * ponytail: sem busca por `wa_lid`. A âncora opaca nasce neste canal, então quem
 * a tem já tem a thread; contato só com âncora em OUTRA thread e sem telefone fica de fora.
 */
export async function contatoConhecidoDaEntrada(
  admin: SupabaseClient,
  organizationId: string,
  channelSessionId: string | null,
  msg: ZernioInboundMessage,
): Promise<{ contactId: string; conversationId: string | null } | null> {
  const conversa = channelSessionId
    ? await conversaPelaThread(admin, organizationId, msg.conversationId, channelSessionId)
    : null;
  if (conversa) return { contactId: conversa.contact_id, conversationId: conversa.id };
  const contato = msg.identity.phone
    ? await encontrarContatoPorTelefone(admin, organizationId, msg.identity.phone)
    : null;
  return contato ? { contactId: contato.id, conversationId: null } : null;
}

/**
 * Os efeitos de negócio da mensagem que acabou de entrar.
 *
 * Delega no passo COMPARTILHADO (`lib/channels/pos-entrada.ts`) em vez de
 * reimplementar: opt-out, nascimento do lead e despacho do agente são regra do
 * produto, não característica deste transporte. Foi exatamente a cópia privada
 * dentro do outro ingest que fez este canal ficar sem os três.
 *
 * Chamada nos DOIS caminhos de inserção — thread conhecida e âncora. Chamar só
 * num deles deixaria a maioria das mensagens sem efeito, que é a forma mais
 * cara de "consertar" isto pela metade.
 *
 * Só para ENTRADA: o eco de um envio nosso (ou do celular do operador) não pede
 * para sair, não abre demanda e não acorda o agente.
 */
async function efeitosDaEntrada(
  admin: SupabaseClient,
  input: {
    organizationId: string;
    channelSessionId: string;
    requestId?: string;
    socialMessage?: Pick<SocialMessage, "platform">;
  },
  msg: ZernioInboundMessage,
  contactId: string,
  conversationId: string,
  messageId: string,
): Promise<void> {
  if (msg.direction !== "inbound") return;

  // O `referral` do webhook oficial é o caminho CONFIÁVEL de atribuição —
  // documentado pela plataforma, ao contrário do best-effort do WAHA. Mesma
  // regra de primeiro-toque: `estamparAtribuicaoDoContato` só grava se o
  // contato ainda não tem `ad_platform`.
  const atribuicao = extrairAtribuicaoMeta(msg.referral);
  if (atribuicao) await estamparAtribuicaoDoContato(admin, input.organizationId, contactId, atribuicao);

  // Irmão do bloco acima, para o Google: o dado não vem no `referral` (que é
  // exclusivo da Meta), vem no PRÓPRIO texto da mensagem — ver o cabeçalho de
  // `lib/plataformas-de-anuncio/google/atribuicao.ts`. Best-effort, mesma postura.
  await extrairEEstamparAtribuicaoGoogle(admin, input.organizationId, contactId, msg.text);

  await aplicarEfeitosPosEntrada(admin, {
    organizationId: input.organizationId,
    contactId,
    conversationId,
    messageId,
    channelSessionId: input.channelSessionId,
    texto: msg.text,
    nomeDoContato: msg.identity.displayName,
    requestId: input.requestId,
    origem: "zernio_webhook",
    // A rede é o canal que os dois ramos acima já gravaram em
    // `conversations.channel`; sem ela o negócio nasce como WhatsApp.
    canal: input.socialMessage?.platform,
  });
}

/**
 * Carimba a conversa com o que acabou de chegar.
 *
 * ─── O que faltava, e o que isso quebrava ──────────────────────────────────
 *
 * Esta chamada não existia neste canal. Os outros dois a fazem; eu escrevi este
 * ingest e a omiti. Medido em produção: conversa com CINCO mensagens do cliente
 * e `last_inbound_at` NULL.
 *
 * O estrago não é cosmético, porque `last_inbound_at` é a fonte da janela de
 * 24h:
 *
 *   - o selo dizia "o cliente nunca escreveu" numa conversa em que ele acabara
 *     de escrever, e o composer barrava o envio por um motivo falso;
 *   - o guardrail do agente calcula a janela do MESMO campo, então o assistente
 *     tratava toda conversa deste canal como fechada e nunca respondia texto
 *     livre — inclusive dentro da janela;
 *   - `unread_count_for_assignee` nunca subia, então o contador de não lidas da
 *     lista ficava em zero mesmo com mensagem nova.
 *
 * Três sintomas sem relação aparente, uma linha ausente.
 *
 * Não carimba no `duplicate`: a reentrega é a MESMA mensagem, e somar de novo
 * inflaria o contador de não lidas a cada reenvio do provider. A exceção é o
 * reprocessamento, com a régua de `retomarEfeitosDaEntrada`.
 *
 * ⚠️ A FALHA DEIXOU DE SER SÓ `logger.warn`, que some no próximo restart do
 * contêiner. O destino agora é o mesmo dos outros canais — uma linha em
 * `event_log` —, e quem decide isso é `lib/channels/marcar-conversa.ts`.
 */
async function marcarConversa(
  admin: SupabaseClient,
  organizationId: string,
  conversationId: string,
  msg: ZernioInboundMessage,
): Promise<void> {
  await marcarConversaComMensagem(admin, {
    organizationId,
    conversationId,
    direction: msg.direction,
    preview: (msg.text ?? "").slice(0, 200),
    // `sentAt` do provider quando existe: a ordem da lista e o cálculo da janela
    // têm que usar a hora em que o cliente ESCREVEU, não a hora em que o webhook
    // chegou — numa reentrega atrasada as duas diferem por horas.
    at: msg.sentAt ?? new Date().toISOString(),
    canal: "zernio",
  });

}

/**
 * Pede a persistência dos bytes do anexo.
 *
 * Mesmo evento e mesmo payload que o canal por QR emite — o consumidor é o
 * único (`workers/media-persist-worker.ts`), e um payload diferente por canal
 * faria o worker adivinhar de quem veio.
 *
 * Best-effort: a mensagem já está gravada e visível. Derrubar a ingestão aqui
 * devolveria 500 ao provider, que reenviaria tudo — trocaria uma mídia
 * faltando por uma tempestade de reentregas.
 */
async function pedirPersistenciaDaMidia(
  admin: SupabaseClient,
  organizationId: string,
  conversationId: string,
  messageId: string,
): Promise<void> {
  const { error } = await admin.rpc(
    "emit_event" as never,
    {
      p_event_type: "media.persist_requested",
      p_entity_kind: "message",
      p_entity_id: messageId,
      p_payload: { message_id: messageId, conversation_id: conversationId },
      p_metadata: { source: "zernio_webhook" },
      p_organization_id: organizationId,
    } as never,
  );
  if (error) {
    logger.warn("[zernio] emit media.persist_requested falhou", {
      messageId,
      detail: error.message,
    });
  }
}

/** A conversa que o provider já associou a esta thread, se houver. */
async function conversaPelaThread(
  admin: SupabaseClient,
  organizationId: string,
  providerConversationId: string,
  channelSessionId: string,
): Promise<{ id: string; contact_id: string } | null> {
  const { data, error } = await admin
    .from("conversations")
    .select("id, contact_id")
    .eq("organization_id", organizationId)
    .eq("channel_session_id", channelSessionId)
    .eq("provider_conversation_id", providerConversationId)
    .maybeSingle();
  // PGRST116 = mais de uma conversa nesta thread (o índice não é único): não é
  // falha de banco, e a resolução pela âncora segue valendo. Qualquer outro erro
  // lança: engolido, cairia nas RPCs abaixo e criaria um segundo contato.
  if (error && error.code !== "PGRST116") throw new Error(`zernio_thread_lookup_failed: ${error.message}`);
  const row = data as { id: string; contact_id: string | null } | null;
  return row?.contact_id ? { id: row.id, contact_id: row.contact_id } : null;
}

async function upsertContact(
  admin: SupabaseClient,
  organizationId: string,
  msg: ZernioInboundMessage,
  identity: string,
): Promise<string | null> {
  const kind = identity.startsWith("phone:") ? "phone" : "lid";
  const valor = identity.slice(identity.indexOf(":") + 1);
  const phoneBruto = kind === "phone" ? valor : msg.identity.phone;
  const existente = phoneBruto
    ? await encontrarContatoPorTelefone(admin, organizationId, phoneBruto)
    : null;
  const phone = existente?.phone_number
    ? canonicalPhoneBR(existente.phone_number)
    : phoneBruto
      ? canonicalPhoneBR(phoneBruto)
      : null;

  // Reusa a RPC do canal por QR: ela já resolve a corrida de dois webhooks
  // simultâneos numa transação, e escrever um segundo upsert seria criar um
  // segundo lugar onde a mesma corrida pode voltar.
  // p_phone também no kind lid: senão a captação (contato só com número) vira
  // um segundo cadastro quando o WhatsApp chega com BSUID.
  const { data, error } = await admin.rpc("fn_upsert_wa_contact", {
    p_org: organizationId,
    p_kind: kind,
    p_phone: phone,
    p_lid: kind === "lid" ? valor : null,
    p_chat_id: msg.conversationId,
    p_notify: msg.identity.displayName ?? msg.identity.username ?? null,
  });
  // Erro de banco LANÇA, como em `insertMessage`: devolver `null` faria a ingestão
  // responder `ignored` e a mensagem aceita fechar como processada sem ter entrado.
  if (error) throw new Error(`zernio_contact_upsert_failed: ${error.message}`);
  const contactId = (data as string) ?? null;
  if (!contactId) return null;

  // O telefone entra MESMO quando a âncora é o id opaco.
  //
  // A âncora certa é o BSUID (sobrevive a trocar de número), mas o payload
  // traz os DOIS — e a RPC só grava `phone_number` quando o `kind` é phone.
  // Descartá-lo custou caro: o contato ficava sem número, o envio parava em
  // `missing_phone_number`, e a tela não tinha o que mostrar ao atendente que
  // precisa saber com quem está falando.
  //
  // `is null` no filtro: só preenche o que está vazio. Sobrescrever apagaria
  // uma correção feita à mão na tela por um valor que a plataforma pode mandar
  // diferente entre eventos.
  //
  // (#3315) O mesmo filtro é o CRITÉRIO da emissão de `contact.phone_added`: o
  // `select("id")` diz se a gravação casou linha, isto é, se o contato estava
  // MESMO sem número até agora. Sem validar isso, emitir aqui seria disparar a
  // regra numa troca de número — que não é "ganhar" nenhum.
  //
  // E não há disparo em dobro com o `emit_event` que a própria RPC
  // (`fn_upsert_wa_contact`, #3315) faz: quando ela preenche a ficha vazia, este
  // `update` já não casa linha nenhuma e não emite; quando ela não preenche (ou
  // `kind` é lid), quem preenche é este update, e só ele emite.
  // Achada pelo EXTRA (`contact_phones`): o número não é o principal da ficha, e
  // gravá-lo aqui o promoveria a principal — o banco recusaria com 23505 e a
  // ingestão cairia. Ficha dona de extra sem principal é caso raro; não grava.
  if (msg.identity.phone && !existente?.achado_pelo_extra) {
    const telefone = canonicalPhoneBR(msg.identity.phone);
    const { data: afetadas } = await admin
      .from("contacts")
      .update({ phone_number: telefone })
      .eq("id", contactId)
      .is("phone_number", null)
      .select("id");
    if (((afetadas as { id: string }[] | null) ?? []).length > 0) {
      await emitirTelefoneGanho(admin, {
        organizationId,
        contactId,
        telefone,
        origem: "zernio.ingest.upsert_contato",
      });
    }
  }

  return contactId;
}

async function upsertConversation(
  admin: SupabaseClient,
  input: {
    organizationId: string;
    contactId: string;
    channelSessionId: string;
    providerConversationId: string;
  },
): Promise<string | null> {
  const { data, error } = await admin.rpc("fn_upsert_wa_conversation", {
    p_org: input.organizationId,
    p_contact: input.contactId,
    p_session: input.channelSessionId,
  });
  if (error) throw new Error(`zernio_conversation_upsert_failed: ${error.message}`);
  if (!data) return null;
  const conversationId = data as string;

  // A thread do provider, que é o motivo deste módulo existir.
  //
  // Gravada SEMPRE, sem condição. A primeira versão tinha um
  // `.neq("provider_conversation_id", ...)` para "só escrever se mudou" — e
  // isso QUEBROU exatamente o que este módulo existe para fazer:
  //
  //   em SQL, `NULL <> 'valor'` é NULL, não TRUE.
  //
  // Conversa recém-criada tem a coluna NULL, então o `neq` nunca a alcançava e
  // o update não pegava linha nenhuma. Medido em produção: o webhook respondia
  // `{"status":"ingested"}` e a coluna ficava `null` — a mensagem aparecia no
  // inbox e responder era impossível.
  //
  // O teste com dublê não pegou porque o fake tratava `.neq()` como no-op:
  // afirmava que o `update` foi CHAMADO com o payload certo, que era verdade, e
  // não que ele tivesse casado alguma linha. Escrever sempre é uma escrita a
  // mais por mensagem e zero condições sutis para errar.
  await admin
    .from("conversations")
    .update({ provider_conversation_id: input.providerConversationId })
    .eq("id", conversationId);

  return conversationId;
}

async function insertMessage(
  admin: SupabaseClient,
  input: {
    organizationId: string;
    conversationId: string;
    contactId: string;
    channelSessionId: string;
    msg: ZernioInboundMessage;
    /** O eco de uma transmissão de campanha: a linha nasce como a da campanha. */
    eco?: EcoDaTransmissao | null;
  },
): Promise<string | "duplicate"> {
  const { msg, eco } = input;
  const temAnexo = msg.attachments.length > 0;
  const primeiro = msg.attachments[0];
  // O anúncio DESTA mensagem, quando ela nasceu de um clique. O contato guarda
  // só o primeiro (primeiro toque, em `efeitosDaEntrada`); sem este rastro, quem
  // volta por OUTRO anúncio aparece na conversa como se viesse do antigo.
  // Só na entrada: eco de saída não vem de clique de ninguém.
  const anuncio =
    msg.direction === "inbound" ? metadataDoAnuncio(extrairAtribuicaoMeta(msg.referral)) : {};

  const { data, error } = await admin
    .from("messages")
    .insert({
      organization_id: input.organizationId,
      conversation_id: input.conversationId,
      contact_id: input.contactId,
      channel_session_id: input.channelSessionId,
      external_id: msg.externalId,
      // ─── Por que a SAÍDA também entra ──────────────────────────────────
      //
      // A primeira versão descartava tudo que era `outgoing`, para não duplicar
      // os envios do próprio CRM. O efeito colateral custou caro: mensagem
      // mandada do celular do operador, ou por outra plataforma ligada à mesma
      // conta, NUNCA aparecia — e o histórico do cliente ficava pela metade sem
      // nada avisando.
      //
      // A duplicação que se temia já estava resolvida por outro lado: o
      // `unique (organization_id, external_id)` devolve 23505 no eco do nosso
      // próprio envio, e o chamador lê isso como "duplicate", não como erro.
      direction: msg.direction,
      // ─── Toda linha nascida do webhook veio de FORA do CRM ──────────────
      //
      // O default da coluna é `'crm'`, e ele mente aqui: quem passou por este
      // arquivo chegou pelo webhook do provider, não pelo composer. O canal por
      // QR já carimba `external_device` no mesmo lugar.
      //
      // Não é cosmético. As três funções de fricção contam SÓ
      // `external_device`, então sem este campo o painel lia "zero atendimento
      // por fora" neste número — mesmo com o operador respondendo o dia inteiro
      // pelo celular. E `removerEcoDoProprioEnvio` filtra por este valor: sem
      // ele, o eco do nosso próprio envio escapa da rede e vira linha
      // duplicada na tela.
      sent_via: eco ? "automation" : "external_device",
      status: msg.direction === "outbound" ? (msg.status ?? "sent") : "delivered",
      type: eco ? "template" : temAnexo ? tipoDoAnexo(primeiro?.type) : msg.location ? "location" : "text",
      // O pino vira link do mapa no corpo: é o que o agente lê e o que a
      // prévia da conversa mostra. As coordenadas ficam no metadata para a tela.
      //
      // O eco da transmissão chega com "[Template: nome]" no lugar do texto: o
      // que a pessoa leu é o modelo congelado no destinatário, e é ele que o
      // agente precisa ver quando a resposta chegar.
      body: eco?.corpo ?? (msg.location ? corpoDaLocalizacao(msg.location) : msg.text),
      ...(eco ? { template_name: eco.modelo, template_language: eco.idioma } : {}),
      // A URL do anexo NÃO é pública: é endpoint autenticado do provider, e a
      // plataforma descarta a mídia depois de um tempo. Ela é PONTEIRO, não
      // conteúdo — e é por isso que grava aqui e o worker baixa os bytes já.
      //
      // Antes esta linha guardava os anexos só no `metadata`, e `media_url`
      // ficava nulo: o worker de persistência saía com "no media_url" e a
      // mídia do cliente virava linha sem bytes. O atendente via "imagem" sem
      // imagem — e como ele responde à mão, era o pior defeito do canal.
      ...(temAnexo && primeiro?.url
        ? { media_url: primeiro.url, media_mime: mimeDoAnexo(primeiro.type) }
        : {}),
      metadata: {
        ...(temAnexo
          ? { provider_attachments: msg.attachments }
          : msg.location
            ? { location: msg.location }
            : {}),
        ...anuncio,
        ...(eco ? metadataDoEco(eco) : {}),
        ...(msg.direction === "inbound" ? ancoraCtwaDaMensagem(extrairAtribuicaoMeta(msg.referral)) : {}),
      },
      ...(msg.sentAt ? { sent_at: msg.sentAt } : {}),
    })
    .select("id")
    .maybeSingle();

  // 23505 = unique violation em (organization_id, external_id). É o desfecho
  // ESPERADO de uma reentrega, não um erro: o provider reenvia quando não
  // recebe 200, e tratar isto como falha faria a rota devolver 500 e ele
  // reenviar de novo, para sempre.
  if (error?.code === "23505") return "duplicate";
  if (error || !data) throw new Error(`zernio_ingest_insert_failed: ${error?.message ?? "sem id"}`);

  return (data as { id: string }).id;
}

/** A saída veio de uma transmissão do provedor — de campanha ou do painel dele. */
function ehDeTransmissao(msg: ZernioInboundMessage): boolean {
  return msg.direction === "outbound" && (msg.sentVia === "broadcast" || !!msg.broadcastId);
}

/** A campanha e o destinatário do eco, quando ele é de uma transmissão de campanha. */
async function ecoDeTransmissao(
  admin: SupabaseClient,
  organizationId: string,
  msg: ZernioInboundMessage,
  contactId: string,
): Promise<EcoDaTransmissao | null> {
  if (msg.direction !== "outbound" || msg.kind !== "message" || !msg.broadcastId) return null;
  return ecoDaTransmissao(admin, {
    organizationId,
    transmissaoId: msg.broadcastId,
    contactId,
    telefone: msg.identity.phone,
  });
}

/**
 * Dica de mime a partir do tipo declarado no webhook.
 *
 * DICA, não verdade: quem manda é o `content-type` da resposta ao baixar. Serve
 * para a tela ter o que mostrar antes de os bytes chegarem, e para o caso em
 * que o provider não devolve o cabeçalho.
 */
function mimeDoAnexo(tipo: string | undefined): string | null {
  switch (tipo) {
    case "image":
      return "image/jpeg";
    case "video":
      return "video/mp4";
    case "audio":
      return "audio/ogg";
    default:
      return null;
  }
}

/** Tipo do anexo do provider → vocabulário de `messages.type`. */
function tipoDoAnexo(tipo: string | undefined): string {
  switch (tipo) {
    case "image":
      return "image";
    case "video":
      return "video";
    case "audio":
      return "audio";
    case "sticker":
      return "sticker";
    default:
      return "document";
  }
}

/**
 * Aplica a edição ou o apagamento numa mensagem que JÁ existe.
 *
 * Não cria linha: uma edição de mensagem que nunca chegou não deve inventar uma
 * conversa do nada, com um texto sem nada antes dele. Devolve `"sem_alvo"`, que
 * é informação — não erro —, e a rota segue respondendo 200.
 *
 * O corpo é sobrescrito e o original não é guardado: o que o CRM mostra tem que
 * ser o que o cliente vê agora. E a linha apagada continua existindo, com
 * `revoked_at`: removê-la levaria junto o contexto das vizinhas e o histórico
 * de quem atendeu.
 */
export async function aplicarEdicaoZernio(
  admin: SupabaseClient,
  organizationId: string,
  edicao: { externalId: string; tipo: "edited" | "deleted"; body: string | null },
): Promise<"aplicado" | "sem_alvo"> {
  const agora = new Date().toISOString();
  const patch =
    edicao.tipo === "deleted"
      ? { revoked_at: agora }
      : // Edição sem corpo novo não zera o texto: seria trocar a versão velha
        // (útil) por um vazio (inútil), e o evento sem corpo é justamente o
        // caso em que não sabemos o texto novo.
        edicao.body !== null
        ? { body: edicao.body, edited_at: agora }
        : { edited_at: agora };

  const { data } = await admin
    .from("messages")
    .update(patch)
    .eq("organization_id", organizationId)
    .eq("external_id", edicao.externalId)
    .select("id");

  return (data ?? []).length > 0 ? "aplicado" : "sem_alvo";
}

/** Concurrent first messages share a database uniqueness constraint. */
async function upsertSocialContact(
  admin: SupabaseClient,
  org: string,
  identity: string,
  name: string | null,
): Promise<string> {
  // Contato FUNDIDO não é alvo: ele aponta para o vencedor da fusão, e
  // escrever nele é escrever num cadastro que ninguém mais lê — o mesmo motivo
  // de `contato-por-telefone`. Aqui a guarda ainda evita um segundo defeito: o
  // índice único é PARCIAL (`where ... and is_merged_into is null`), então duas
  // linhas com a mesma identidade — uma mesclada, uma viva — são estado
  // legítimo, e sem o filtro o `.maybeSingle()` estoura.
  const { data: existing, error: readError } = await admin
    .from("contacts")
    .select("id")
    .eq("organization_id", org)
    .eq("social_identity", identity)
    .is("is_merged_into", null)
    .maybeSingle();
  if (readError) throw new Error("social_contact_lookup_failed");
  if (existing) return existing.id as string;
  const { data, error } = await admin
    .from("contacts")
    .insert({
      organization_id: org,
      social_identity: identity,
      name,
      display_name: name,
      source: "social",
    })
    .select("id")
    .single();
  if (error?.code === "23505") {
    const { data: winner, error: retryError } = await admin
      .from("contacts")
      .select("id")
      .eq("organization_id", org)
      .eq("social_identity", identity)
      // Mesma guarda da busca acima: quem perdeu a corrida procura o VIVO.
      // O `.single()` reclama de zero e de duas — sem o filtro, uma ficha
      // mesclada com a mesma identidade tornaria "duas" alcançável.
      .is("is_merged_into", null)
      .single();
    if (retryError || !winner) throw new Error("social_contact_race_failed");
    return winner.id as string;
  }
  if (error || !data) throw new Error("social_contact_create_failed");
  return data.id as string;
}

/**
 * A HORA do desfecho — `delivered_at` e `read_at` — e não só o estado.
 *
 * O canal oficial direto já carimbava as duas (`lib/channels/meta/status-update.ts`);
 * por aqui só o `status` mudava, e as colunas ficavam nulas para sempre: a
 * conversa mostrava o tique certo, mas "quanto o cliente demorou para ler" não
 * tinha como ser medido. Medido numa instalação real (24/09/2026): 41 mensagens
 * entregues no dia, nenhuma com `delivered_at`.
 *
 * Cada coluna é gravada UMA vez (`is null`): o primeiro evento que a alcança é o
 * que vale. Por isso é um update à parte do de `status` — aquele recusa rebaixar
 * `read` para `delivered`, e um `delivered` atrasado ainda precisa carimbar a
 * entrega. Um `read` sem `delivered` antes (a ordem do webhook não é garantida)
 * carimba as duas com a mesma hora: quem leu, recebeu.
 *
 * Best-effort: o carimbo é métrica; falhar aqui não pode derrubar a ingestão.
 */
async function carimbarHoraDoDesfecho(
  admin: SupabaseClient,
  input: { organizationId: string; channelSessionId: string },
  msg: ZernioInboundMessage,
): Promise<void> {
  const colunas =
    msg.status === "read" ? (["delivered_at", "read_at"] as const)
    : msg.status === "delivered" ? (["delivered_at"] as const)
    : [];
  const quando = msg.statusAt ?? new Date().toISOString();
  for (const coluna of colunas) {
    const { error } = await admin
      .from("messages")
      .update({ [coluna]: quando })
      .eq("organization_id", input.organizationId)
      .eq("channel_session_id", input.channelSessionId)
      .eq("external_id", msg.externalId)
      .is(coluna, null);
    if (error) {
      logger.warn("[zernio] hora do desfecho não gravada", {
        organization_id: input.organizationId,
        coluna,
        erro: error.message,
      });
    }
  }
}