import { assertAgentOperationPg, type AgentOperationContext } from '@/lib/ai/agents/operation';
import { assertApprovedReplyPg, type ApprovedReplyContext } from '@/lib/ai/replies/delivery';
import { assertMeetingDeliveryPg, type MeetingDeliveryContext } from '@/lib/agenda/meet-delivery';
import type { JobClaim } from '../../queue/claim';
import { assertAgendaEffectPg } from '@/lib/agenda/efeito';
import { AgendaPuladoError } from '@/lib/agenda/escolha-no-envio';
import { escolhaDoFluxoPg } from '@/lib/agenda/escolha-no-envio-leitura';
import { AgendaDeferredError } from '@/lib/agenda/protecao-followup';
import { StaleServiceBoundaryError } from '@/lib/atendimento/fronteira';
import { requireCurrentAutonomousTurn, requireCurrentServiceBoundary } from '@/lib/atendimento/fronteira-server';
import { parseServiceBoundary } from '@/lib/atendimento/fronteira';
/**
 * Borda de saída pós-fusão: envio de mensagem SEMPRE via `sendMessageHandler` do
 * próprio app (app/api/v1/messages/_handler.ts) — o handler insere a linha
 * outbound, envia pelo WAHA, atualiza conversa, audita e emite evento; e dá de
 * graça o guard is_blocked (ApiError 403). A tool `send_message` do agente chama
 * ESTA função depois da cadeia de guardrails; nenhum output de modelo vira
 * mensagem sem passar por aqui.
 *
 * Idempotência (o handler NÃO tem idempotency key própria — o ledger cobre):
 *   1. transação lógica: insert em `send_ledger` (unique (job_id, seq)); o
 *      `send_ledger.id` É a idempotency_key, enviada em `metadata.idempotency_key`
 *      da mensagem;
 *   2. chamada ao handler; 'sent' → accepted; 'queued'/'failed' → registrados;
 *   3. retry pós-crash: 'accepted' pula; 'requested' PRIMEIRO procura em
 *      `messages` uma linha com essa idempotency_key (o crash pode ter sido
 *      DEPOIS do envio) — achou, reconcilia o ledger sem reenviar; 'failed'
 *      rotaciona o id (tentativa lógica nova).
 */
import { sendWithLedger, pgSendLedger, type SendOutcome } from './send-ledger';
import { registrarCustoDeVoz, resolverCaminhoDeVoz, VozLedgerTerminalError } from './send-voice';
import type { FormatoDaVoz } from '@/lib/ai/voz/formato';
import { logger } from '@/lib/logger';
export type { SendOutcome, SendLedgerStatus } from './send-ledger';

import { ApiError } from '@/lib/api/types';
import { sendMessageHandler } from '@/app/api/v1/messages/_handler';
import type { Message } from '@/lib/types/messaging';

import type { Queryable } from '../../queue/queue';
import { cancelJob, rescheduleJob, type JobRow } from '../../queue/queue';
import { cancelPendingCronsForLead } from '../../cron/scheduler';
import { CrmTransportError, type CrmEdgeConfig } from './mcp-client';
import type { TipoDeMidiaComLegenda } from '@/lib/messaging/media/midia-com-legenda';

/** Erro de negócio não classificado do handler (ex.: conversa inexistente) — ledger fica 'requested'. */
export class SendToolError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'SendToolError';
  }
}

export interface SendMessageInput {
  agentOperation?: AgentOperationContext;
  jobClaim?: JobClaim;
  tenantId: string;
  leadId: string | null;
  jobId: string;
  /** Posição da mensagem no turno (1..n) — com jobId forma a identidade da intenção. */
  seq: number;
  conversationId: string;
  body: string;
  /**
   * Presente = envio de TEMPLATE. O `body` continua sendo o texto RENDERIZADO — é
   * ele que entra no hash de idempotência e é ele que os gates de conteúdo avaliaram.
   * Trocar a chave por "nome do template" faria dois envios com valores diferentes
   * colidirem no ledger e o segundo virar `already_sent` sem ter saído.
   */
  template?: { name: string; language: string; values: Record<string, string> };
  /**
   * Presente = esta mensagem sai como NOTA DE VOZ (Fase 6a do plano de voz,
   * `docs/superpowers/plans/2026-10-29-voz-elevenlabs.md`), não texto.
   *
   * `body` SEGUE sendo o texto FALADO (não o original escrito pelo agente) —
   * é ele que entra no hash de idempotência do ledger, pela mesma razão do
   * `template` acima: o retry precisa achar a MESMA linha, e a linha é
   * identificada pelo que de fato foi sintetizado. `textoOriginal` é só para
   * a thread não ficar cega (`media_derived_text` — plano §4.4); não participa
   * de hash nem de guardrail.
   *
   * ⚠️ Ausente = comportamento desta função IDÊNTICO a antes desta Fase —
   * nada aqui muda um envio de texto existente.
   */
  voz?: {
    textoOriginal: string;
    apiKey: string;
    voiceId: string;
    modelId?: string;
    /** mp3 ou ogg/opus — o que o canal da conversa entrega como nota de voz (`lib/ai/voz/formato.ts`). */
    formato?: FormatoDaVoz;
    /**
     * Grava os bytes no Storage e devolve o path — ver `send-voice.ts`.
     *
     * ⚠️ O path PRECISA começar com `${organizationId}/${conversationId}/`:
     * `isMediaPathOwnedBy` (`lib/messaging/media/upload-validation.ts`) recusa
     * qualquer outro prefixo com `invalid_media_path`, e `sendMessageHandler`
     * aplica essa checagem para TODO `media_storage_path`, agente incluso.
     */
    upload: (audio: Buffer, mimeType: string) => Promise<string>;
  };
  /**
   * Presente = o envio sai como MÍDIA e o `body` é a LEGENDA.
   *
   * Duas procedências, um caminho só (o handler de envio): `storagePath` é a imagem ou o áudio
   * que já mora na pasta da conversa (`whatsapp-media/...`, foto de produto — assinada
   * curta por quem conhece o provider), e `url` é a mídia de um passo automático
   * (#2734, ampliada pelo #2972 para áudio, vídeo e documento), que o operador
   * aponta direto e o handler confere antes de buscar.
   */
  media?:
    | { kind?: 'audio' | 'image'; storagePath: string; mime: string }
    | { url: string; mime: string; type: TipoDeMidiaComLegenda };
}

/**
 * O corpo que o handler de mensagens recebe: texto, template ou mídia da
 * conversa. Exportado para o teste — o tipo decide o caminho no handler, e um
 * `type` errado degrada áudio ou imagem sem ninguém ver.
 */
export function corpoDoEnvio(
  input: SendMessageInput,
  idempotencyKey: string,
  /** A nota de voz já sintetizada (`input.voz` resolvido); ausente = texto/template/mídia. */
  voz?: { mediaStoragePath: string; mimeType: string },
): Parameters<typeof sendMessageHandler>[2] {
  return {
    conversation_id: input.conversationId,
    ...(voz
      ? { type: 'audio' as const, media_storage_path: voz.mediaStoragePath, media_mime: voz.mimeType }
      : input.template
      ? {
          type: 'template' as const,
          template_name: input.template.name,
          template_language: input.template.language,
          template_values: input.template.values,
        }
      : input.media
        ? 'url' in input.media
          ? {
              // #2734 — mídia por URL de um passo automático: o `type` vem decidido
              // (imagem × vídeo × áudio × documento) e o handler é quem busca e
              // confere o arquivo.
              type: input.media.type,
              media_url: input.media.url,
              media_mime: input.media.mime,
            }
          : {
              type: input.media.kind ?? 'image',
              media_storage_path: input.media.storagePath,
              media_mime: input.media.mime,
            }
        : { type: 'text' as const }),
    // Foto sem legenda vai sem `body`: o schema do envio pede corpo não vazio.
    ...(input.body !== '' || !input.media ? { body: input.body } : {}),
    metadata: { idempotency_key: idempotencyKey },
  };
}

/** Fallback do ator ai_agent quando não há agente publicado (cfg.agentActorId). */
export const AGENT_ACTOR_ID = 'agent-engine';

/**
 * Envia UMA mensagem do turno pelo handler do app. Intenção exactly-once,
 * entrega at-least-once: throws (transporte) deixam o ledger em 'requested' —
 * o retry reconcilia pela PK (`messages.id` = chave) antes de reenviar, com
 * `messages.metadata.idempotency_key` só como reserva para linha antiga.
 */
export async function sendTurnMessage(
  db: Queryable,
  cfg: CrmEdgeConfig,
  input: SendMessageInput,
): Promise<SendOutcome> {
  // Antes do ledger, não só no handler: recusada lá dentro, a bolha deixa uma
  // linha 'requested' que o disjuntor de saúde conta como envio (total_sends).
  if (input.agentOperation) await assertAgentOperationPg(db, input.agentOperation);
  const { rows: sourceJobs } = await db.query<Pick<JobRow, 'kind' | 'payload'>>(
    'select payload,kind from job_queue where id=$1 and organization_id=$2 and contact_id=$3',
    [input.jobId, input.tenantId, input.leadId],
  );
  await requireCurrentServiceBoundary(
    db,
    parseServiceBoundary(sourceJobs[0]?.payload.service_boundary),
  );
  if (sourceJobs[0]) await requireCurrentAutonomousTurn(db, {
    id: input.jobId,
    organization_id: input.tenantId,
    contact_id: input.leadId,
    kind: sourceJobs[0].kind,
    locked_by: input.jobClaim?.worker_id ?? null,
    claim_acquired_at: input.jobClaim?.acquired_at,
  });
  // #3813 — a escolha do FLUXO ("Quando o contato tiver horário marcado") vale
  // também aqui: este é o único ponto de envio do worker `pg`, e sem a escolha
  // no contexto a guarda abaixo leria o padrão (`adiar`) e desfaria o que o
  // fluxo publicou — um fluxo em `enviar` seguiria sendo segurado, e um em
  // `nao_enviar` seguiria adiando em vez de registrar o pulo.
  const enrollmentDoJob =
    typeof sourceJobs[0]?.payload.followup_enrollment_id === 'string'
      ? sourceJobs[0].payload.followup_enrollment_id
      : undefined;
  const proactiveContext =
    sourceJobs[0]?.kind === 'followup_turn' && input.leadId
      ? {
          organizationId: input.tenantId,
          contactId: input.leadId,
          jobId: input.jobId,
          jobClaim: input.jobClaim,
          enrollmentId: enrollmentDoJob,
          nodeId:
            typeof sourceJobs[0].payload.node_id === 'string'
              ? sourceJobs[0].payload.node_id
              : undefined,
          ...(enrollmentDoJob
            ? { escolhaNoEnvio: await escolhaDoFluxoPg(db, input.tenantId, enrollmentDoJob) }
            : {}),
        }
      : undefined;
  if (proactiveContext) await assertAgendaEffectPg(db, proactiveContext);
  let meetingDelivery: MeetingDeliveryContext | undefined;
  if (sourceJobs[0]?.kind === 'transactional_delivery') {
    if (!input.jobClaim) throw new StaleServiceBoundaryError();
    meetingDelivery = {
      organizationId: input.tenantId,
      jobId: input.jobId,
      jobClaim: input.jobClaim,
    };
    await assertMeetingDeliveryPg(db, meetingDelivery);
  }
  let approvedReply: ApprovedReplyContext | undefined;
  if (sourceJobs[0]?.kind === 'approved_reply') {
    if (!input.jobClaim) throw new StaleServiceBoundaryError();
    approvedReply = {
      organizationId: input.tenantId,
      jobId: input.jobId,
      jobClaim: input.jobClaim,
    };
    const policy = await assertApprovedReplyPg(db, approvedReply);
    if (policy.body !== input.body || policy.conversation_id !== input.conversationId)
      throw new StaleServiceBoundaryError();
  }
  // ─── Voz (Fase 6a): resolve o ÁUDIO antes do ledger decidir o que fazer ───
  //
  // Precisa acontecer ANTES de `sendWithLedger` porque a síntese é cara e
  // idempotente pela MESMA identidade (job_id, seq) que o ledger usa — ver o
  // cabeçalho de `send-voice.ts`. `undefined` quando `input.voz` não existe:
  // nenhum envio de texto passa por aqui, e o resto desta função continua
  // byte a byte igual ao de antes desta Fase.
  let vozResolvida: { mediaStoragePath: string; mimeType: string } | undefined;
  if (input.voz) {
    try {
      const r = await resolverCaminhoDeVoz({
        store: pgSendLedger(db),
        identidade: { tenantId: input.tenantId, leadId: input.leadId, jobId: input.jobId, seq: input.seq },
        textoFalado: input.body,
        voz: {
          apiKey: input.voz.apiKey,
          voiceId: input.voz.voiceId,
          modelId: input.voz.modelId,
          ...(input.voz.formato ? { formato: input.voz.formato } : {}),
        },
        upload: input.voz.upload,
      });
      vozResolvida = { mediaStoragePath: r.mediaStoragePath, mimeType: r.mimeType };
      // Custo entra no orçamento SÓ quando a síntese de fato ACONTECEU agora
      // — um retry que reusou o path de uma tentativa anterior não pagou o
      // ElevenLabs de novo, e registrar de novo contaria um gasto que não
      // existiu, inflando o orçamento da organização por retry.
      if (r.sintetizouAgora) {
        registrarCustoDeVoz({
          supabase: cfg.supabase,
          organizationId: input.tenantId,
          jobId: input.jobId,
          caracteres: r.caracteres,
          modelId: input.voz.modelId ?? 'eleven_multilingual_v2',
        });
      }
    } catch (err) {
      // A linha já tinha desfecho TERMINAL antes de chegarmos aqui — não há
      // síntese a fazer nem envio a tentar; o resultado É esse desfecho.
      if (err instanceof VozLedgerTerminalError) {
        return err.kind === 'already_sent'
          ? { kind: 'already_sent', idempotencyKey: err.key, crmMessageId: err.crmMessageId }
          : { kind: 'blocked', idempotencyKey: err.key };
      }
      // ⚠️ ACHADO NA REVISÃO, ANTES DA FASE 7: isto já foi `throw err` — e
      // derrubava o TURNO INTEIRO (o cliente não recebia resposta nenhuma)
      // quando a falha era do ElevenLabs/Storage (rede, 503, bucket fora),
      // não do ledger. Contradizia a própria doutrina deste módulo ("voz é
      // um EXTRA sobre um caminho que já funciona" — ver o cabeçalho de
      // `voz-do-turno.ts`), que só valia na resolução de CREDENCIAL, antes
      // de chegar aqui. `vozResolvida` fica `undefined` e o fluxo abaixo
      // segue como se `input.voz` nunca tivesse existido — `input.body` (o
      // texto FALADO, já limpo) sai como mensagem de TEXTO normal. O
      // cliente recebe uma resposta; só não é a voz que seria.
      logger.warn('[send-message] síntese de voz falhou — respondendo em texto', {
        organization_id: input.tenantId,
        job_id: input.jobId,
        error: err instanceof Error ? err.message : String(err),
      });
    }
  }

  return sendWithLedger(pgSendLedger(db), input, async (idempotencyKey, messageId) => {
    let message: Message;
    try {
      message = await sendMessageHandler(
        cfg.supabase,
        {
          organization_id: input.tenantId,
          actor: { type: 'ai_agent', id: cfg.agentActorId ?? AGENT_ACTOR_ID, role: 'manager' },
          requestId: idempotencyKey,
          serviceBoundary: parseServiceBoundary(sourceJobs[0]?.payload.service_boundary),
          proactiveContext,
          meetingDelivery,
          approvedReply,
          agentOperation: input.agentOperation,
          internalMessageId: messageId,
          // Nunca vem do corpo de uma requisição pública (ver o comentário de
          // `aiVoiceDerivedText` em `HandlerCtx`) — só este seam a preenche.
          ...(input.voz && vozResolvida ? { aiVoiceDerivedText: input.voz.textoOriginal } : {}),
        },
        corpoDoEnvio(input, idempotencyKey, vozResolvida),
      );
    } catch (err) {
      // #3813 — o PULO por compromisso não é falha de transporte: quem chama
      // registra `skipped` com o motivo. Convertê-lo em `CrmTransportError`
      // ("handler indisponível") mentiria sobre a causa e mandaria o turno
      // re-tentar um envio que o fluxo decidiu não fazer.
      if (
        err instanceof AgendaPuladoError ||
        err instanceof AgendaDeferredError ||
        err instanceof StaleServiceBoundaryError
      )
        throw err;
      if (err instanceof ApiError && err.status === 403) {
        throw err;
      }
      if (err instanceof ApiError && err.status === 404) {
        await touchLedgerError(db, input.tenantId, idempotencyKey, 'conversa não encontrada');
        throw new SendToolError('envio recusado: conversa não encontrada');
      }
      // Qualquer outra falha (Supabase fora, erro interno do handler): transiente —
      // o ledger fica 'requested' e o replay reconcilia pela key.
      const msg = err instanceof Error ? err.message : String(err);
      await touchLedgerError(db, input.tenantId, idempotencyKey, msg);
      throw new CrmTransportError(`handler de envio indisponível: ${msg.slice(0, 120)}`);
    }

    return message;
  });
}

async function touchLedgerError(
  db: Queryable,
  org: string,
  id: string,
  errorText: string,
): Promise<void> {
  await db.query(
    `update send_ledger set last_error = $2, updated_at = now() where id = $1 and organization_id = $3`,
    [id, errorText.slice(0, 300), org],
  );
}

export type SendDisposition =
  /** Job cancelado em definitivo (veto is_blocked) — não re-tenta. */
  | { action: 'canceled'; job: JobRow | null }
  /** Job devolvido a 'pending' com run_after adiado, sem consumir attempts. */
  | { action: 'requeued'; job: JobRow | null }
  /** Nada a fazer com o job aqui: 'sent'/'already_sent' seguem para complete; 'failed' segue para failJob. */
  | { action: 'none' };

/**
 * Disposição do JOB conforme o outcome do envio:
 * - blocked → cancela o job (terminal — opt-out não é incidente) e cancela TODOS
 *   os follow-ups agendados do contato (irrevogável, regra dura nº 2). A fonte
 *   do bloqueio JÁ é contacts.is_blocked — não existe mais cache a atualizar;
 * - queued → reagenda com `delayMs` (knob SEND_QUEUED_RETRY_MS) SEM consumir
 *   attempts — sessão fora não pode matar mensagem de lead saudável;
 * - demais → responsabilidade do worker (complete/failJob pelos caminhos normais).
 */
export async function applySendOutcome(
  db: Queryable,
  outcome: SendOutcome,
  job: {
    jobId: string;
    workerId: string;
    tenantId: string;
    leadId: string | null;
    jobClaim?: JobClaim;
  },
  knobs: { queuedRetryDelayMs: number },
): Promise<SendDisposition> {
  switch (outcome.kind) {
    case 'blocked': {
      const canceled = await cancelJob(
        db,
        job.jobId,
        job.workerId,
        'envio vetado pelo sink: contato bloqueado (is_blocked) — opt-out irrevogável',
        job.jobClaim?.acquired_at,
      );
      if (job.leadId) {
        await cancelPendingCronsForLead(db, job.tenantId, job.leadId);
      }
      return { action: 'canceled', job: canceled };
    }
    case 'queued': {
      const requeued = await rescheduleJob(db, job.jobId, job.workerId, {
        delayMs: knobs.queuedRetryDelayMs,
        acquiredAt: job.jobClaim?.acquired_at,
        reason: 'sessão do canal fora (resposta queued) — reagendado sem consumir attempts',
      });
      return { action: 'requeued', job: requeued };
    }
    default:
      return { action: 'none' };
  }
}