//! Inbound bridge — translates a wa-agent `AgentCtx` into a broker event
//! and blocks on the LLM reply so `run_agent_with` can render the
//! response with its native typing heartbeat.
//!
//! We publish on `plugin.inbound.whatsapp` with `Event.session_id` set
//! to a deterministic UUIDv5 derived from the remote JID. Core's agent
//! runtime already debounces by `session_id` and replies on
//! `plugin.outbound.whatsapp` carrying the same `session_id`, which
//! the outbound dispatcher routes back through the [`PendingMap`].

use std::sync::Arc;
use std::time::Duration;

use crate::config::WhatsappPluginConfig;
use anyhow::Result;
use nexo_broker::{AnyBroker, BrokerHandle, Event};
use tokio::sync::oneshot;
use uuid::Uuid;
use whatsapp_rs::agent::{AgentCtx, Response};

use crate::events::InboundEvent;
use crate::plugin::AskReplyIndex;
use crate::plugin::PendingMap;
use crate::session_id::session_id_for_jid;
use whatsapp_rs::messages::MessageContent;

pub const TOPIC_INBOUND: &str = "plugin.inbound.whatsapp";
pub const SOURCE: &str = "plugin.whatsapp";

/// Build the inbound topic for this account. When `instance` is set,
/// publishes land on `plugin.inbound.whatsapp.<instance>` so an agent
/// binding (`inbound_bindings: [{plugin: whatsapp, instance: X}]`) can
/// pin itself to one account without cross-talk.
pub fn inbound_topic_for(instance: Option<&str>) -> String {
    match instance {
        Some(inst) if !inst.is_empty() => format!("{}.{}", TOPIC_INBOUND, inst),
        _ => TOPIC_INBOUND.to_string(),
    }
}

/// Build the outbound topic for this account. Mirrors `inbound_topic_for`
/// — the dispatcher subscribes to its own
/// `plugin.outbound.whatsapp.<instance>` so agents can route replies to
/// a specific account.
pub fn outbound_topic_for(instance: Option<&str>) -> String {
    match instance {
        Some(inst) if !inst.is_empty() => {
            format!("{}.{}", crate::dispatch::TOPIC_OUTBOUND, inst)
        }
        _ => crate::dispatch::TOPIC_OUTBOUND.to_string(),
    }
}

/// Encapsulates one reactive inbound → broker → oneshot cycle. Kept as
/// its own function so unit tests can drive it with a `LocalBroker`
/// without spinning up a real `wa-agent` session.
///
/// Returns the reply text the outbound dispatcher delivered, or `None`
/// on timeout (caller decides what to render to the user).
pub async fn bridge_step(
    broker: &AnyBroker,
    pending: &PendingMap,
    cfg: &WhatsappPluginConfig,
    session_id: Uuid,
    event_payload: InboundEvent,
) -> Option<String> {
    let inbound_topic = inbound_topic_for(cfg.instance.as_deref());
    tracing::info!(
        target: "whatsapp.bridge",
        %session_id,
        topic = %inbound_topic,
        timeout_ms = cfg.bridge.response_timeout_ms,
        "bridge_step entered — about to publish inbound + await reply"
    );
    let (tx, rx) = oneshot::channel::<String>();
    // Last inbound wins: if a previous message on this session is still
    // awaiting, its handler sees a dropped sender and falls through to
    // the timeout path. This matches the core runtime's debounce model.
    pending.insert(session_id, tx);

    let mut event = Event::new(&inbound_topic, SOURCE, event_payload.to_payload());
    event.session_id = Some(session_id);
    if let Err(e) = broker.publish(&inbound_topic, event).await {
        tracing::warn!(%session_id, error = %e, "inbound publish failed");
        pending.remove(&session_id);
        return None;
    }
    tracing::info!(
        target: "whatsapp.bridge",
        %session_id,
        topic = %inbound_topic,
        "broker.publish ok — awaiting agent reply"
    );

    match tokio::time::timeout(Duration::from_millis(cfg.bridge.response_timeout_ms), rx).await {
        Ok(Ok(text)) => Some(text),
        Ok(Err(_cancelled)) => {
            // Sender was dropped — a newer inbound on the same session
            // won the slot. Silent Noop is the right call here.
            tracing::debug!(%session_id, "bridge sender superseded");
            None
        }
        Err(_elapsed) => {
            pending.remove(&session_id);
            let ev = InboundEvent::BridgeTimeout { session_id };
            let mut out = Event::new(&inbound_topic, SOURCE, ev.to_payload());
            out.session_id = Some(session_id);
            let _ = broker.publish(&inbound_topic, out).await;
            None
        }
    }
}

/// Build the closure passed to `Session::run_agent_with`. Captures
/// cheap clones of broker / pending / cfg / session so each invocation
/// stays self-contained.
///
/// When the incoming `AgentCtx` carries media, a sibling task downloads
/// the file to `cfg.media_dir` and publishes an
/// `InboundEvent::MediaReceived` — the handler itself never blocks on
/// the download, so the typing heartbeat stays responsive and other
/// chats are not held up by slow media fetches.
pub fn build_handler(
    broker: AnyBroker,
    pending: PendingMap,
    ask_reply_index: AskReplyIndex,
    cfg: Arc<WhatsappPluginConfig>,
    session: Arc<whatsapp_rs::Session>,
) -> impl Fn(AgentCtx) -> futures::future::BoxFuture<'static, Response> + Send + Sync + 'static {
    move |ctx: AgentCtx| {
        let broker = broker.clone();
        let pending = pending.clone();
        let ask_reply_index = ask_reply_index.clone();
        let cfg = cfg.clone();
        let session = session.clone();
        Box::pin(async move {
            tracing::info!(
                target: "whatsapp.bridge",
                msg_id = %ctx.msg.key.id,
                remote_jid = %ctx.msg.key.remote_jid,
                from_me = ctx.msg.key.from_me,
                text_len = ctx.text.as_deref().map(|t| t.len()).unwrap_or(0),
                "handler ENTRY — wa-agent dispatched message to bridge handler"
            );
            if cfg.behavior.ignore_groups && ctx.msg.key.remote_jid.ends_with("@g.us") {
                tracing::info!(
                    target: "whatsapp.bridge",
                    msg_id = %ctx.msg.key.id,
                    "skip: ignore_groups + @g.us"
                );
                return Response::Noop;
            }
            // Drop offline backlog older than the configured
            // threshold. WhatsApp Multi-Device re-delivers
            // buffered messages on every reconnect even after we ACK
            // them; without this gate the agent replies to the same
            // stale backlog after every daemon restart. The check is
            // a no-op when `skip_backlog_age_secs == 0`.
            tracing::info!(
                target: "whatsapp.bridge",
                msg_id = %ctx.msg.key.id,
                skip_backlog_age_secs = cfg.behavior.skip_backlog_age_secs,
                msg_timestamp = ctx.msg.message_timestamp,
                "pre-age check"
            );
            if cfg.behavior.skip_backlog_age_secs > 0 {
                let now_secs = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_secs())
                    .unwrap_or(ctx.msg.message_timestamp);
                let age_secs = now_secs.saturating_sub(ctx.msg.message_timestamp);
                if age_secs > cfg.behavior.skip_backlog_age_secs {
                    tracing::info!(
                        target: "whatsapp.bridge",
                        msg_id = %ctx.msg.key.id,
                        age_secs,
                        threshold = cfg.behavior.skip_backlog_age_secs,
                        "skipping offline backlog message older than threshold"
                    );
                    return Response::Noop;
                }
            }
            tracing::info!(
                target: "whatsapp.bridge",
                msg_id = %ctx.msg.key.id,
                "post-age check, before audio"
            );

            // Audio inbound: download synchronously so the
            // `InboundEvent::Message` payload includes media_path /
            // mime up-front. An auto-discovered
            // `*_inbound_transform` tool (e.g. STT) needs the file
            // available BEFORE the LLM turn runs, otherwise the
            // agent sees an empty inbound and replies "no entiendo".
            // Non-audio media (image / video / document) keeps the
            // legacy async-spawn path — those don't gate the agent.
            let mut audio_attachment: Option<crate::media::DownloadedInbound> = None;
            if let Some(content) = ctx.msg.message.as_ref() {
                if let Some((variant, _, _)) = crate::media::variant_of_content(content) {
                    let is_audio = matches!(
                        variant,
                        crate::media::MediaVariant::Audio | crate::media::MediaVariant::VoiceNote
                    );
                    if is_audio {
                        match crate::media::download_inbound_to_disk(&session, &cfg, &ctx.msg).await
                        {
                            Ok(downloaded) => audio_attachment = downloaded,
                            Err(e) => tracing::warn!(
                                error = %e,
                                "inbound audio download failed; agent will see empty text"
                            ),
                        }
                    } else {
                        let broker_m = broker.clone();
                        let cfg_m = cfg.clone();
                        let session_m = session.clone();
                        let msg_m = ctx.msg.clone();
                        tokio::spawn(async move {
                            if let Err(e) = crate::media::download_inbound(
                                &session_m, &broker_m, &cfg_m, &msg_m,
                            )
                            .await
                            {
                                tracing::warn!(error = %e, "inbound media download failed");
                            }
                        });
                    }
                }
            }

            let session_id = session_id_for_jid(ctx.jid());
            let reply_to_id = extract_reply_to_id(ctx.msg.message.as_ref());
            let ask_qid = reply_to_id
                .as_deref()
                .and_then(|rid| ask_reply_index.get(rid).map(|v| v.value().clone()))
                .or_else(|| ctx.text.as_deref().and_then(extract_question_id_marker));
            let (media_kind, media_path, media) = match audio_attachment.as_ref() {
                Some(a) => (
                    Some(match a.variant {
                        crate::media::MediaVariant::VoiceNote => "audio_voice".to_string(),
                        crate::media::MediaVariant::Audio => "audio".to_string(),
                        _ => "audio".to_string(),
                    }),
                    Some(a.local_path.to_string_lossy().into_owned()),
                    Some(crate::events::InboundMessageMedia {
                        mime_type: a.mime.clone(),
                    }),
                ),
                None => (None, None, None),
            };
            tracing::info!(
                target: "whatsapp.bridge",
                msg_id = %ctx.msg.key.id,
                session_id = %session_id,
                "about to call bridge_step"
            );
            let payload = InboundEvent::Message {
                from: ctx.sender().to_string(),
                chat: ctx.jid().to_string(),
                text: ctx.text.clone(),
                reply_to: reply_to_id,
                reply_to_question_id: ask_qid.clone(),
                ask_question_id: ask_qid,
                is_group: ctx.msg.key.remote_jid.ends_with("@g.us"),
                timestamp: chrono::Utc::now().timestamp(),
                msg_id: ctx.msg.key.id.clone(),
                media_kind,
                media_path,
                media,
            };

            match bridge_step(&broker, &pending, &cfg, session_id, payload).await {
                Some(text) => Response::Text(text),
                None => match cfg.bridge.on_timeout.as_str() {
                    "apology_text" => Response::Text(cfg.bridge.apology_text.clone()),
                    _ => Response::Noop,
                },
            }
        })
    }
}

/// Thin helper used by `start()` to convert plugin boot errors into
/// `anyhow` without pulling wa-agent types into the plugin.rs module.
pub fn forward_err<T: std::fmt::Display>(label: &str, e: T) -> anyhow::Error {
    anyhow::anyhow!("{label}: {e}")
}

/// Build the `whatsapp_rs::agent::Acl` from plugin config + env var.
pub fn build_acl(cfg: &WhatsappPluginConfig) -> whatsapp_rs::agent::Acl {
    let mut acl = if cfg.acl.from_env.is_empty() {
        whatsapp_rs::agent::Acl::open()
    } else {
        whatsapp_rs::agent::Acl::from_env(&cfg.acl.from_env)
    };
    for jid in &cfg.acl.allow_list {
        acl = acl.allow(jid);
    }
    acl
}

fn extract_question_id_marker(text: &str) -> Option<String> {
    for line in text.lines() {
        let t = line.trim();
        if let Some(rest) = t.strip_prefix("[question_id]") {
            let id = rest.trim();
            if !id.is_empty() {
                return Some(id.to_string());
            }
        }
    }
    None
}

fn extract_reply_to_id(content: Option<&MessageContent>) -> Option<String> {
    match content {
        Some(MessageContent::Reply { reply_to_id, .. }) if !reply_to_id.is_empty() => {
            Some(reply_to_id.clone())
        }
        _ => None,
    }
}

#[doc(hidden)]
pub fn _unused_result<T>() -> Result<T>
where
    T: Default,
{
    Ok(T::default())
}

#[cfg(test)]
mod tests {
    use super::{extract_question_id_marker, extract_reply_to_id};
    use whatsapp_rs::messages::MessageContent;

    #[test]
    fn extract_question_id_marker_ok() {
        let text = "[question] x\n[question_id] q-123\n";
        assert_eq!(extract_question_id_marker(text).as_deref(), Some("q-123"));
    }

    #[test]
    fn extract_reply_to_id_from_reply_variant() {
        let c = MessageContent::Reply {
            reply_to_id: "wamid-1".to_string(),
            text: "answer".to_string(),
        };
        assert_eq!(extract_reply_to_id(Some(&c)).as_deref(), Some("wamid-1"));
    }
}
