lingshu-gateway 0.10.0

//! # WhatsApp adapter via the Hermes Baileys bridge
//!
//! Hermes already ships a production-tested Node bridge that handles the
//! difficult parts of WhatsApp connectivity. Lingshu reuses that bridge
//! rather than reimplementing the protocol stack in Rust.

use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::Mutex;
use std::time::{Duration, Instant};

use anyhow::Context;
use async_trait::async_trait;
use lingshu_core::{config::WhatsAppGatewayConfig, lingshu_home};
use lingshu_types::Platform;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tracing::{debug, info, warn};

use crate::platform::{
    IncomingMessage, MessageAttachment, MessageAttachmentKind, MessageMetadata, OutgoingMessage,
    PlatformAdapter,
};

const MAX_MESSAGE_LENGTH: usize = 65_536;
const DEFAULT_BRIDGE_PORT: u16 = 3000;
const DEFAULT_POLL_INTERVAL_MS: u64 = 1_000;
const DEFAULT_STARTUP_TIMEOUT_SECS: u64 = 20;
const PERSONAL_JID_SUFFIX: &str = "@s.whatsapp.net";

#[derive(Debug, Clone)]
pub struct BridgeAssets {
    pub bridge_dir: PathBuf,
    pub bridge_script: PathBuf,
}

#[derive(Debug, Clone)]
pub struct WhatsappAdapterConfig {
    pub bridge_port: u16,
    pub bridge_url: Option<String>,
    pub bridge_dir: Option<PathBuf>,
    pub bridge_script: Option<PathBuf>,
    pub session_path: PathBuf,
    pub mode: String,
    pub allowed_users: Vec<String>,
    pub reply_prefix: Option<String>,
    pub install_dependencies: bool,
    pub startup_timeout: Duration,
    pub poll_interval: Duration,
    pub log_path: PathBuf,
}

impl Default for WhatsappAdapterConfig {
    fn default() -> Self {
        let home = lingshu_home();
        Self {
            bridge_port: DEFAULT_BRIDGE_PORT,
            bridge_url: None,
            bridge_dir: None,
            bridge_script: None,
            session_path: home.join("whatsapp").join("session"),
            mode: "self-chat".into(),
            allowed_users: Vec::new(),
            reply_prefix: Some("\u{2695} *Lingshu Agent*\n------------\n".into()),
            install_dependencies: true,
            startup_timeout: Duration::from_secs(DEFAULT_STARTUP_TIMEOUT_SECS),
            poll_interval: Duration::from_millis(DEFAULT_POLL_INTERVAL_MS),
            log_path: home.join("logs").join("whatsapp-bridge.log"),
        }
    }
}

impl From<&WhatsAppGatewayConfig> for WhatsappAdapterConfig {
    fn from(value: &WhatsAppGatewayConfig) -> Self {
        Self {
            bridge_port: value.bridge_port,
            bridge_url: value.bridge_url.clone(),
            bridge_dir: value.bridge_dir.clone(),
            bridge_script: value.bridge_script.clone(),
            session_path: value
                .session_path
                .clone()
                .unwrap_or_else(|| lingshu_home().join("whatsapp").join("session")),
            mode: value.mode.clone(),
            // Normalize: strip '+' prefix so numbers match WhatsApp JIDs
            allowed_users: value
                .allowed_users
                .iter()
                .map(|u| u.trim_start_matches('+').to_string())
                .collect(),
            reply_prefix: value.reply_prefix.clone(),
            install_dependencies: value.install_dependencies,
            ..Self::default()
        }
    }
}

impl WhatsappAdapterConfig {
    pub fn bridge_url(&self) -> String {
        self.bridge_url
            .clone()
            .unwrap_or_else(|| format!("http://127.0.0.1:{}", self.bridge_port))
    }

    pub fn health_url(&self) -> String {
        format!("{}/health", self.bridge_url())
    }

    pub fn messages_url(&self) -> String {
        format!("{}/messages", self.bridge_url())
    }

    pub fn send_url(&self) -> String {
        format!("{}/send", self.bridge_url())
    }

    pub fn send_media_url(&self) -> String {
        format!("{}/send-media", self.bridge_url())
    }

    pub fn send_clarify_url(&self) -> String {
        format!("{}/send-clarify", self.bridge_url())
    }
}

#[derive(Debug, Deserialize)]
struct WhatsAppHealth {
    status: String,
}

#[derive(Debug, Deserialize)]
struct WhatsAppInboundEvent {
    #[serde(rename = "messageId")]
    message_id: Option<String>,
    #[serde(rename = "chatId")]
    chat_id: String,
    #[serde(rename = "senderId")]
    sender_id: String,
    #[serde(rename = "senderName")]
    sender_name: Option<String>,
    /// Whether the message was sent by the account owner.
    /// Defence-in-depth: bridge.js already filters contact messages in
    /// self-chat mode, but the Rust adapter enforces the same rule so that
    /// a stale/old bridge binary can never leak contacts to the agent.
    /// Defaults to `false` (conservative — treat unknown origin as contact).
    #[serde(rename = "fromMe", default)]
    from_me: bool,
    body: String,
    #[serde(rename = "mediaUrls", default)]
    media_urls: Vec<String>,
}

#[derive(Debug, Serialize)]
struct SendRequest<'a> {
    #[serde(rename = "chatId")]
    chat_id: &'a str,
    message: &'a str,
}

#[derive(Debug, Serialize)]
struct SendMediaRequest<'a> {
    #[serde(rename = "chatId")]
    chat_id: &'a str,
    #[serde(rename = "filePath")]
    file_path: &'a str,
    #[serde(rename = "mediaType")]
    media_type: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    caption: Option<&'a str>,
    #[serde(rename = "fileName", skip_serializing_if = "Option::is_none")]
    file_name: Option<&'a str>,
}

#[derive(Debug, Serialize)]
struct SendClarifyRequest<'a> {
    #[serde(rename = "chatId")]
    chat_id: &'a str,
    question: &'a str,
    #[serde(rename = "interactionId")]
    interaction_id: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    choices: Option<&'a [String]>,
}

#[derive(Debug, Deserialize)]
struct SendClarifyResponse {
    #[serde(default)]
    success: bool,
    #[serde(default)]
    used_buttons: bool,
    #[serde(default)]
    used_list: bool,
    #[serde(default)]
    error: Option<String>,
}

#[derive(Debug, Deserialize)]
struct SendResponse {
    #[serde(default)]
    success: bool,
    #[serde(default)]
    error: Option<String>,
}

#[derive(Debug)]
struct BridgeSendFailure {
    detail: String,
}

impl std::fmt::Display for BridgeSendFailure {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.detail)
    }
}

impl std::error::Error for BridgeSendFailure {}

pub struct WhatsAppAdapter {
    config: WhatsappAdapterConfig,
    client: reqwest::Client,
    managed_bridge: Mutex<Option<Child>>,
}

impl WhatsAppAdapter {
    pub fn new(config: WhatsappAdapterConfig) -> anyhow::Result<Self> {
        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(10))
            .build()?;

        // Propagate WHATSAPP_MODE to the parent process so that the auth module
        // can read it via std::env::var("WHATSAPP_MODE") when checking
        // authorization. Without this, only the bridge subprocess would see the
        // value (it is also passed via cmd.env later) and check_authorization()
        // would always fall through to NoAllowlistDeny in self-chat mode.
        //
        // Safety: this runs at gateway boot before any concurrent access to
        // WHATSAPP_MODE, and the value is stable for the lifetime of the adapter.
        #[allow(unused_unsafe)]
        unsafe {
            std::env::set_var("WHATSAPP_MODE", &config.mode);
        }

        Ok(Self {
            config,
            client,
            managed_bridge: Mutex::new(None),
        })
    }

    pub fn is_available(config: &WhatsappAdapterConfig) -> bool {
        if config.bridge_url.is_some() {
            true
        } else {
            has_node() && resolve_bridge_assets(config).is_ok()
        }
    }

    pub fn default_session_path() -> PathBuf {
        lingshu_home().join("whatsapp").join("session")
    }

    pub fn default_log_path() -> PathBuf {
        lingshu_home().join("logs").join("whatsapp-bridge.log")
    }

    pub fn resolve_bridge_assets(config: &WhatsappAdapterConfig) -> anyhow::Result<BridgeAssets> {
        resolve_bridge_assets(config)
    }

    pub fn pair(config: &WhatsappAdapterConfig) -> anyhow::Result<()> {
        ensure_node_available()?;
        let assets = resolve_bridge_assets(config)?;
        ensure_dependencies(config, &assets)?;
        std::fs::create_dir_all(&config.session_path)?;

        let status = Command::new("node")
            .arg(&assets.bridge_script)
            .arg("--pair-only")
            .arg("--session")
            .arg(&config.session_path)
            .arg("--mode")
            .arg(&config.mode)
            .current_dir(&assets.bridge_dir)
            .status()
            .context("failed to start WhatsApp pairing bridge")?;

        if !status.success() {
            anyhow::bail!("WhatsApp pairing exited with status {}", status);
        }
        Ok(())
    }

    pub async fn health(config: &WhatsappAdapterConfig) -> anyhow::Result<String> {
        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(3))
            .build()?;
        let health = client
            .get(config.health_url())
            .send()
            .await?
            .error_for_status()?
            .json::<WhatsAppHealth>()
            .await?;
        Ok(health.status)
    }

    async fn ensure_bridge_ready(&self) -> anyhow::Result<()> {
        if let Ok(status) = Self::health(&self.config).await
            && status == "connected"
        {
            return Ok(());
        }

        self.start_managed_bridge()?;

        let deadline = Instant::now() + self.config.startup_timeout;
        while Instant::now() < deadline {
            match Self::health(&self.config).await {
                Ok(status) if status == "connected" => return Ok(()),
                Ok(status) => {
                    debug!(status, "waiting for WhatsApp bridge");
                }
                Err(error) => {
                    debug!(%error, "waiting for WhatsApp bridge health endpoint");
                }
            }
            tokio::time::sleep(Duration::from_millis(500)).await;
        }

        anyhow::bail!(
            "WhatsApp bridge did not become ready; check {}",
            self.config.log_path.display()
        );
    }

    fn start_managed_bridge(&self) -> anyhow::Result<()> {
        if self.config.bridge_url.is_some() {
            anyhow::bail!("remote WhatsApp bridge is unavailable");
        }

        if self
            .managed_bridge
            .lock()
            .expect("managed_bridge mutex poisoned")
            .is_some()
        {
            return Ok(());
        }

        ensure_node_available()?;
        let assets = resolve_bridge_assets(&self.config)?;
        ensure_dependencies(&self.config, &assets)?;
        std::fs::create_dir_all(
            self.config
                .log_path
                .parent()
                .unwrap_or_else(|| Path::new(".")),
        )?;
        std::fs::create_dir_all(&self.config.session_path)?;

        let log_file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.config.log_path)
            .with_context(|| format!("failed to open {}", self.config.log_path.display()))?;
        let stderr_file = log_file.try_clone()?;

        let mut cmd = Command::new("node");
        cmd.arg(&assets.bridge_script)
            .arg("--port")
            .arg(self.config.bridge_port.to_string())
            .arg("--session")
            .arg(&self.config.session_path)
            .arg("--mode")
            .arg(&self.config.mode)
            .current_dir(&assets.bridge_dir)
            .stdout(Stdio::from(log_file))
            .stderr(Stdio::from(stderr_file))
            .stdin(Stdio::null());
        if !self.config.allowed_users.is_empty() {
            // Normalize: strip leading '+' from phone numbers so the bridge can
            // compare against WhatsApp JIDs which never include '+'.
            let normalized: Vec<String> = self
                .config
                .allowed_users
                .iter()
                .map(|u| u.trim_start_matches('+').to_string())
                .collect();
            cmd.env("WHATSAPP_ALLOWED_USERS", normalized.join(","));
        }
        if let Some(prefix) = &self.config.reply_prefix {
            cmd.env("WHATSAPP_REPLY_PREFIX", prefix);
        }
        cmd.env("WHATSAPP_MODE", &self.config.mode);

        let child = cmd.spawn().context("failed to spawn WhatsApp bridge")?;
        info!(
            pid = child.id(),
            log = %self.config.log_path.display(),
            "started WhatsApp bridge"
        );
        *self
            .managed_bridge
            .lock()
            .expect("managed_bridge mutex poisoned") = Some(child);
        Ok(())
    }

    async fn poll_messages(&self) -> anyhow::Result<Vec<WhatsAppInboundEvent>> {
        let response = self
            .client
            .get(self.config.messages_url())
            .send()
            .await?
            .error_for_status()?;
        Ok(response.json::<Vec<WhatsAppInboundEvent>>().await?)
    }

    fn event_to_message(event: WhatsAppInboundEvent) -> IncomingMessage {
        let attachments = event
            .media_urls
            .iter()
            .map(|path| whatsapp_attachment(path))
            .collect::<Vec<_>>();
        let text = render_incoming_text(&event.body, &attachments);

        IncomingMessage {
            platform: Platform::Whatsapp,
            user_id: event.sender_id,
            channel_id: Some(event.chat_id.clone()),
            chat_type: crate::platform::ChatType::Dm,
            text,
            thread_id: None,
            metadata: MessageMetadata {
                message_id: event.message_id,
                channel_id: Some(event.chat_id),
                user_display_name: event.sender_name,
                attachments,
                ..Default::default()
            },
        }
    }

    async fn send_bridge_clarify(
        &self,
        chat_id: &str,
        question: &str,
        interaction_id: u64,
        choices: Option<&[String]>,
    ) -> Result<(bool, bool), BridgeSendFailure> {
        let body = SendClarifyRequest {
            chat_id,
            question,
            interaction_id,
            choices,
        };
        let response = self
            .client
            .post(self.config.send_clarify_url())
            .json(&body)
            .send()
            .await
            .map_err(|error| BridgeSendFailure {
                detail: format!(
                    "failed to contact the local clarify relay ({}): {error}",
                    self.config.send_clarify_url()
                ),
            })?;

        let status = response.status();
        let raw_body = response.text().await.map_err(|error| BridgeSendFailure {
            detail: format!(
                "failed to read the local clarify relay response ({}): {error}",
                self.config.send_clarify_url()
            ),
        })?;

        if !status.is_success() {
            return Err(BridgeSendFailure {
                detail: parse_bridge_error(&raw_body).unwrap_or_else(|| {
                    format!("clarify relay returned HTTP {status}: {raw_body}")
                }),
            });
        }

        let payload: SendClarifyResponse = serde_json::from_str(&raw_body).map_err(|error| {
            BridgeSendFailure {
                detail: format!("invalid clarify relay response: {error}; body={raw_body}"),
            }
        })?;
        if !payload.success {
            return Err(BridgeSendFailure {
                detail: payload
                    .error
                    .unwrap_or_else(|| "clarify relay reported failure".into()),
            });
        }
        Ok((payload.used_buttons, payload.used_list))
    }

    async fn handle_clarify_tap(&self, event: &WhatsAppInboundEvent) {
        let Some(rest) = event.body.strip_prefix("cl:") else {
            return;
        };
        let mut parts = rest.split(':');
        let Some(id_str) = parts.next() else {
            return;
        };
        let Ok(interaction_id) = id_str.parse::<u64>() else {
            return;
        };
        let Some(choice_key) = parts.next() else {
            return;
        };

        let chat_id = normalize_outbound_chat_id(&event.chat_id);
        if choice_key == "other" {
            let _ = self
                .send_bridge_message(&chat_id, "✏️ Type your answer:")
                .await;
            return;
        }

        let Ok(idx) = choice_key.parse::<usize>() else {
            return;
        };
        let answer = crate::clarify_wiring::peek_clarify(interaction_id)
            .await
            .and_then(|(_, choices)| choices)
            .and_then(|choices| choices.get(idx).cloned())
            .unwrap_or_else(|| choice_key.to_string());

        if crate::clarify_wiring::resolve_clarify_button(interaction_id, answer).await {
            debug!(interaction_id, "WhatsApp clarify button resolved");
        }
    }

    async fn send_bridge_message(
        &self,
        chat_id: &str,
        message: &str,
    ) -> Result<(), BridgeSendFailure> {
        let body = SendRequest { chat_id, message };
        let response = self
            .client
            .post(self.config.send_url())
            .json(&body)
            .send()
            .await
            .map_err(|error| BridgeSendFailure {
                detail: format!(
                    "failed to contact the local message relay ({}): {error}",
                    self.config.send_url()
                ),
            })?;

        let status = response.status();
        let raw_body = response.text().await.map_err(|error| BridgeSendFailure {
            detail: format!(
                "failed to read the local message relay response ({}): {error}",
                self.config.send_url()
            ),
        })?;

        if !status.is_success() {
            let detail = parse_bridge_error(&raw_body).unwrap_or_else(|| {
                status
                    .canonical_reason()
                    .unwrap_or("unknown error")
                    .to_string()
            });
            return Err(BridgeSendFailure { detail });
        }

        let payload =
            serde_json::from_str::<SendResponse>(&raw_body).map_err(|error| BridgeSendFailure {
                detail: format!(
                    "invalid response from the local message relay ({}): {error}",
                    self.config.send_url()
                ),
            })?;
        if !payload.success {
            return Err(BridgeSendFailure {
                detail: payload.error.unwrap_or_else(|| "unknown error".into()),
            });
        }
        Ok(())
    }

    async fn send_bridge_media(
        &self,
        chat_id: &str,
        file_path: &str,
        media_type: &str,
        caption: Option<&str>,
        file_name: Option<&str>,
    ) -> Result<(), BridgeSendFailure> {
        let body = SendMediaRequest {
            chat_id,
            file_path,
            media_type,
            caption,
            file_name,
        };
        let response = self
            .client
            .post(self.config.send_media_url())
            .json(&body)
            .send()
            .await
            .map_err(|error| BridgeSendFailure {
                detail: format!(
                    "failed to contact the local media relay ({}): {error}",
                    self.config.send_media_url()
                ),
            })?;

        let status = response.status();
        let raw_body = response.text().await.map_err(|error| BridgeSendFailure {
            detail: format!(
                "failed to read the local media relay response ({}): {error}",
                self.config.send_media_url()
            ),
        })?;

        if !status.is_success() {
            let detail = parse_bridge_error(&raw_body).unwrap_or_else(|| {
                status
                    .canonical_reason()
                    .unwrap_or("unknown error")
                    .to_string()
            });
            return Err(BridgeSendFailure { detail });
        }

        let payload =
            serde_json::from_str::<SendResponse>(&raw_body).map_err(|error| BridgeSendFailure {
                detail: format!(
                    "invalid response from the local media relay ({}): {error}",
                    self.config.send_media_url()
                ),
            })?;
        if !payload.success {
            return Err(BridgeSendFailure {
                detail: payload.error.unwrap_or_else(|| "unknown error".into()),
            });
        }
        Ok(())
    }
}

fn normalize_outbound_chat_id(chat_id: &str) -> String {
    let trimmed = chat_id.trim();
    if trimmed.is_empty() {
        return String::new();
    }

    if let Some((local_part, suffix)) = split_supported_jid(trimmed) {
        let normalized_local = normalize_jid_local_part(local_part, suffix);
        if !normalized_local.is_empty() {
            return format!("{normalized_local}@{suffix}");
        }
    }

    let compact = compact_phone_candidate(trimmed);
    if !compact.is_empty() && compact.chars().all(|c| c.is_ascii_digit()) {
        return format!("{compact}{PERSONAL_JID_SUFFIX}");
    }

    trimmed.to_string()
}

fn split_supported_jid(value: &str) -> Option<(&str, &str)> {
    if let Some((local_part, suffix)) = value.split_once('@') {
        let suffix = suffix.trim();
        if is_supported_jid_suffix(suffix) {
            return Some((local_part, suffix));
        }
    }

    if let Some((local_part, suffix)) = value.split_once(':') {
        let suffix = suffix.trim();
        if is_supported_jid_suffix(suffix) {
            return Some((local_part, suffix));
        }
    }

    None
}

fn is_supported_jid_suffix(suffix: &str) -> bool {
    matches!(suffix, "s.whatsapp.net" | "g.us" | "lid")
}

fn normalize_jid_local_part(local_part: &str, suffix: &str) -> String {
    let device_stripped = local_part.split(':').next().unwrap_or(local_part).trim();
    if suffix == "g.us" {
        return device_stripped
            .chars()
            .filter(|c| !c.is_ascii_whitespace())
            .collect();
    }

    compact_phone_candidate(device_stripped)
}

fn compact_phone_candidate(value: &str) -> String {
    let compact: String = value
        .chars()
        .filter(|c| !c.is_ascii_whitespace() && !matches!(c, '-' | '(' | ')'))
        .collect();
    compact.trim_start_matches('+').to_string()
}

fn parse_bridge_error(raw_body: &str) -> Option<String> {
    let trimmed = raw_body.trim();
    if trimmed.is_empty() {
        return None;
    }

    serde_json::from_str::<SendResponse>(trimmed)
        .ok()
        .and_then(|payload| payload.error)
        .or_else(|| Some(trimmed.to_string()))
}

fn whatsapp_attachment(path: &str) -> MessageAttachment {
    let file_name = Path::new(path)
        .file_name()
        .and_then(|name| name.to_str())
        .map(ToString::to_string);
    MessageAttachment {
        kind: classify_whatsapp_attachment(path),
        file_name,
        local_path: Some(path.to_string()),
        ..Default::default()
    }
}

fn classify_whatsapp_attachment(path: &str) -> MessageAttachmentKind {
    let extension = Path::new(path)
        .extension()
        .and_then(|ext| ext.to_str())
        .map(|ext| ext.to_ascii_lowercase());

    match extension.as_deref() {
        Some("jpg" | "jpeg" | "png" | "gif" | "webp" | "bmp" | "heic") => {
            MessageAttachmentKind::Image
        }
        Some("mp4" | "mov" | "avi" | "mkv" | "webm") => MessageAttachmentKind::Video,
        Some("ogg" | "opus" | "mp3" | "wav" | "m4a" | "aac") => MessageAttachmentKind::Audio,
        Some("pdf" | "doc" | "docx" | "xls" | "xlsx" | "ppt" | "pptx" | "txt" | "csv" | "zip") => {
            MessageAttachmentKind::Document
        }
        _ => MessageAttachmentKind::Other,
    }
}

fn render_incoming_text(body: &str, attachments: &[MessageAttachment]) -> String {
    let body = body.trim();
    if attachments.is_empty() {
        return body.to_string();
    }

    let mut lines = Vec::new();
    if !body.is_empty() {
        lines.push(body.to_string());
        lines.push(String::new());
    }

    let noun = if attachments.len() == 1 {
        "attachment"
    } else {
        "attachments"
    };
    lines.push(format!("Shared {} {}:", attachments.len(), noun));

    for attachment in attachments {
        let label = attachment.kind.label();
        let file_name = attachment.file_name.as_deref().unwrap_or(label);
        if let Some(path) = attachment.local_path.as_deref() {
            lines.push(format!("- {}: {} ({})", label, file_name, path));
        } else {
            lines.push(format!("- {}: {}", label, file_name));
        }
    }

    lines.join("\n")
}

#[async_trait]
impl PlatformAdapter for WhatsAppAdapter {
    fn platform(&self) -> Platform {
        Platform::Whatsapp
    }

    async fn start(&self, tx: mpsc::Sender<IncomingMessage>) -> anyhow::Result<()> {
        self.ensure_bridge_ready().await?;
        info!("WhatsApp adapter polling bridge");

        let mode = self.config.mode.clone();
        let allowed_users = self.config.allowed_users.clone();

        loop {
            match self.poll_messages().await {
                Ok(events) => {
                    for event in events {
                        // ── Rust-side safety filter ──────────────────────────
                        // Defence-in-depth: enforce the same rules as bridge.js
                        // so a stale bridge binary can never leak contact messages
                        // to the agent.
                        //
                        // self-chat mode  → only fromMe=true events; drop contacts
                        // bot mode        → pass through (allowlist enforced by bridge)
                        if mode != "bot" && !event.from_me {
                            debug!(
                                chat_id = %event.chat_id,
                                sender_id = %event.sender_id,
                                "dropping contact message in self-chat mode (Rust guard)"
                            );
                            continue;
                        }
                        // In bot mode, enforce the allowlist as a second guard if configured.
                        if mode == "bot" && !allowed_users.is_empty() {
                            let sender_num = event
                                .sender_id
                                .split('@')
                                .next()
                                .unwrap_or(&event.sender_id);
                            if !allowed_users
                                .iter()
                                .any(|u| u == sender_num || u == &event.sender_id)
                            {
                                debug!(
                                    sender_id = %event.sender_id,
                                    "dropping non-allowlisted contact (Rust guard)"
                                );
                                continue;
                            }
                        }

                        if event.body.starts_with("cl:") {
                            self.handle_clarify_tap(&event).await;
                            continue;
                        }

                        let message = Self::event_to_message(event);
                        tx.send(message)
                            .await
                            .context("failed to forward WhatsApp message")?;
                    }
                }
                Err(error) => {
                    warn!(%error, "WhatsApp bridge poll failed");
                }
            }
            tokio::time::sleep(self.config.poll_interval).await;
        }
    }

    async fn send(&self, msg: OutgoingMessage) -> anyhow::Result<()> {
        let original_chat_id = msg
            .metadata
            .channel_id
            .as_deref()
            .ok_or_else(|| anyhow::anyhow!("WhatsApp delivery requires channel_id"))?;
        let normalized_chat_id = normalize_outbound_chat_id(original_chat_id);
        if normalized_chat_id.is_empty() {
            anyhow::bail!("WhatsApp delivery requires a non-empty channel_id");
        }

        self.send_bridge_message(&normalized_chat_id, &msg.text)
            .await
            .map_err(|error| {
                anyhow::anyhow!(
                    "WhatsApp bridge send failed for '{}' (normalized to '{}'): {}",
                    original_chat_id,
                    normalized_chat_id,
                    error
                )
            })
    }

    fn format_response(&self, text: &str, _metadata: &MessageMetadata) -> String {
        text.to_string()
    }

    fn max_message_length(&self) -> usize {
        MAX_MESSAGE_LENGTH
    }

    fn supports_markdown(&self) -> bool {
        true
    }

    fn supports_images(&self) -> bool {
        true
    }

    fn supports_files(&self) -> bool {
        true
    }

    async fn send_clarify(
        &self,
        prompt: &crate::platform::ClarifyPrompt,
        metadata: &MessageMetadata,
    ) -> anyhow::Result<bool> {
        let original_chat_id = metadata
            .channel_id
            .as_deref()
            .ok_or_else(|| anyhow::anyhow!("WhatsApp clarify requires channel_id"))?;
        let normalized_chat_id = normalize_outbound_chat_id(original_chat_id);
        if normalized_chat_id.is_empty() {
            anyhow::bail!("WhatsApp clarify requires a non-empty channel_id");
        }

        let choices = prompt.choices.as_ref().filter(|c| !c.is_empty());
        if choices.is_some_and(|c| c.len() > 10) {
            return Ok(false);
        }

        match self
            .send_bridge_clarify(
                &normalized_chat_id,
                &prompt.question,
                prompt.interaction_id,
                choices.map(|c| c.as_slice()),
            )
            .await
        {
            Ok((used_buttons, used_list)) => Ok(used_buttons || used_list),
            Err(error) => {
                warn!(%error, "WhatsApp clarify buttons failed; falling back to text");
                Ok(false)
            }
        }
    }

    async fn send_photo(
        &self,
        path: &str,
        caption: Option<&str>,
        metadata: &MessageMetadata,
    ) -> anyhow::Result<()> {
        if !std::fs::metadata(path).is_ok_and(|metadata| metadata.is_file()) {
            anyhow::bail!("WhatsApp photo path does not exist: {path}");
        }

        let original_chat_id = metadata
            .channel_id
            .as_deref()
            .ok_or_else(|| anyhow::anyhow!("WhatsApp photo delivery requires channel_id"))?;
        let normalized_chat_id = normalize_outbound_chat_id(original_chat_id);
        if normalized_chat_id.is_empty() {
            anyhow::bail!("WhatsApp photo delivery requires a non-empty channel_id");
        }

        self.send_bridge_media(&normalized_chat_id, path, "image", caption, None)
            .await
            .map_err(|error| {
                anyhow::anyhow!(
                    "WhatsApp bridge media send failed for '{}' (normalized to '{}'): {}",
                    original_chat_id,
                    normalized_chat_id,
                    error
                )
            })
    }

    async fn send_document(
        &self,
        path: &str,
        caption: Option<&str>,
        metadata: &MessageMetadata,
    ) -> anyhow::Result<()> {
        if !std::fs::metadata(path).is_ok_and(|metadata| metadata.is_file()) {
            anyhow::bail!("WhatsApp document path does not exist: {path}");
        }

        let original_chat_id = metadata
            .channel_id
            .as_deref()
            .ok_or_else(|| anyhow::anyhow!("WhatsApp document delivery requires channel_id"))?;
        let normalized_chat_id = normalize_outbound_chat_id(original_chat_id);
        if normalized_chat_id.is_empty() {
            anyhow::bail!("WhatsApp document delivery requires a non-empty channel_id");
        }

        let file_name = std::path::Path::new(path)
            .file_name()
            .and_then(|value| value.to_str());
        self.send_bridge_media(&normalized_chat_id, path, "document", caption, file_name)
            .await
            .map_err(|error| {
                anyhow::anyhow!(
                    "WhatsApp bridge media send failed for '{}' (normalized to '{}'): {}",
                    original_chat_id,
                    normalized_chat_id,
                    error
                )
            })
    }

    async fn send_voice(
        &self,
        path: &str,
        caption: Option<&str>,
        metadata: &MessageMetadata,
    ) -> anyhow::Result<()> {
        if !std::fs::metadata(path).is_ok_and(|metadata| metadata.is_file()) {
            anyhow::bail!("WhatsApp audio path does not exist: {path}");
        }

        let original_chat_id = metadata
            .channel_id
            .as_deref()
            .ok_or_else(|| anyhow::anyhow!("WhatsApp audio delivery requires channel_id"))?;
        let normalized_chat_id = normalize_outbound_chat_id(original_chat_id);
        if normalized_chat_id.is_empty() {
            anyhow::bail!("WhatsApp audio delivery requires a non-empty channel_id");
        }

        let file_name = std::path::Path::new(path)
            .file_name()
            .and_then(|value| value.to_str());
        self.send_bridge_media(&normalized_chat_id, path, "audio", caption, file_name)
            .await
            .map_err(|error| {
                anyhow::anyhow!(
                    "WhatsApp bridge media send failed for '{}' (normalized to '{}'): {}",
                    original_chat_id,
                    normalized_chat_id,
                    error
                )
            })
    }
}

fn has_node() -> bool {
    Command::new("node")
        .arg("--version")
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .map(|status| status.success())
        .unwrap_or(false)
}

fn ensure_node_available() -> anyhow::Result<()> {
    if has_node() {
        Ok(())
    } else {
        anyhow::bail!("Node.js is required for WhatsApp support")
    }
}

fn resolve_bridge_assets(config: &WhatsappAdapterConfig) -> anyhow::Result<BridgeAssets> {
    if let Some(script) = &config.bridge_script {
        let bridge_dir = config.bridge_dir.clone().unwrap_or_else(|| {
            script
                .parent()
                .unwrap_or_else(|| Path::new("."))
                .to_path_buf()
        });
        if script.exists() {
            return Ok(BridgeAssets {
                bridge_dir,
                bridge_script: script.clone(),
            });
        }
    }

    if let Some(dir) = &config.bridge_dir {
        let script = dir.join("bridge.js");
        if script.exists() {
            return Ok(BridgeAssets {
                bridge_dir: dir.clone(),
                bridge_script: script,
            });
        }
    }

    let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
    // Look relative to the gateway crate: crates/lingshu-gateway -> lingshu root
    let crate_root = manifest_dir
        .join("../..")
        .canonicalize()
        .unwrap_or(manifest_dir.clone());
    let repo_root = manifest_dir
        .join("../../..")
        .canonicalize()
        .unwrap_or(manifest_dir.clone());
    let candidates = [
        // Prefer lingshu's own bridge first (sibling to crates/)
        crate_root.join("scripts").join("whatsapp-bridge"),
        // Then try mono-repo layout
        repo_root
            .join("lingshu")
            .join("scripts")
            .join("whatsapp-bridge"),
        // Fallback to hermes-agent bridge
        repo_root
            .join("hermes-agent")
            .join("scripts")
            .join("whatsapp-bridge"),
    ];

    for bridge_dir in candidates {
        let bridge_script = bridge_dir.join("bridge.js");
        if bridge_script.exists() {
            return Ok(BridgeAssets {
                bridge_dir,
                bridge_script,
            });
        }
    }

    anyhow::bail!("could not find a WhatsApp bridge directory")
}

fn ensure_dependencies(
    config: &WhatsappAdapterConfig,
    assets: &BridgeAssets,
) -> anyhow::Result<()> {
    if !config.install_dependencies || assets.bridge_dir.join("node_modules").exists() {
        return Ok(());
    }

    info!(dir = %assets.bridge_dir.display(), "Installing WhatsApp bridge dependencies");

    // Step 1: Install with --ignore-scripts to avoid sharp's native build
    // which commonly fails when platform-specific prebuilt binaries aren't
    // resolved automatically (node-gyp + node-addon-api issues).
    let status = Command::new("npm")
        .args(["install", "--ignore-scripts"])
        .current_dir(&assets.bridge_dir)
        .status()
        .context("failed to run npm install for WhatsApp bridge")?;
    if !status.success() {
        anyhow::bail!("npm install failed in {}", assets.bridge_dir.display());
    }

    // Step 2: Run the postinstall script which installs the correct
    // platform-specific prebuilt sharp binary (e.g. @img/sharp-darwin-arm64).
    let postinstall = assets.bridge_dir.join("install-sharp-prebuilt.js");
    if postinstall.exists() {
        let post_status = Command::new("node")
            .arg(&postinstall)
            .current_dir(&assets.bridge_dir)
            .status();
        match post_status {
            Ok(s) if s.success() => {
                info!("Sharp prebuilt installed successfully");
            }
            Ok(s) => {
                // Non-fatal: sharp is only needed for media thumbnails
                warn!(
                    code = s.code(),
                    "sharp prebuilt install exited with error (non-fatal)"
                );
            }
            Err(e) => {
                warn!(%e, "could not run sharp prebuilt installer (non-fatal)");
            }
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::{Json, Router, routing::post};
    use serde_json::{Value, json};
    use std::sync::{Arc, Mutex};

    #[test]
    fn config_from_gateway_config() {
        let source = WhatsAppGatewayConfig {
            enabled: true,
            bridge_port: 3111,
            mode: "bot".into(),
            allowed_users: vec!["12345".into()],
            ..Default::default()
        };
        let config = WhatsappAdapterConfig::from(&source);
        assert_eq!(config.bridge_port, 3111);
        assert_eq!(config.mode, "bot");
        assert_eq!(config.allowed_users, vec!["12345"]);
    }

    #[test]
    fn incoming_media_paths_are_appended() {
        let incoming = WhatsAppInboundEvent {
            message_id: Some("msg-1".into()),
            chat_id: "123@s.whatsapp.net".into(),
            sender_id: "123@s.whatsapp.net".into(),
            sender_name: Some("Raphael".into()),
            from_me: true,
            body: "look".into(),
            media_urls: vec!["/tmp/test.png".into(), "/tmp/test.pdf".into()],
        };
        let message = WhatsAppAdapter::event_to_message(incoming);
        assert!(message.text.contains("Shared 2 attachments:"));
        assert!(message.text.contains("image: test.png"));
        assert!(message.text.contains("document: test.pdf"));
        assert_eq!(
            message.metadata.channel_id.as_deref(),
            Some("123@s.whatsapp.net")
        );
        assert_eq!(message.metadata.attachments.len(), 2);
    }

    #[test]
    fn bridge_assets_use_explicit_script() {
        let temp = tempfile::tempdir().expect("tempdir");
        let script = temp.path().join("bridge.js");
        std::fs::write(&script, "console.log('ok');").expect("write bridge");

        let config = WhatsappAdapterConfig {
            bridge_script: Some(script.clone()),
            ..Default::default()
        };
        let assets = resolve_bridge_assets(&config).expect("assets");
        assert_eq!(assets.bridge_script, script);
        assert_eq!(assets.bridge_dir, temp.path());
    }

    // ─── attachment classification edge cases ──────────────────────────────

    #[test]
    fn heic_classified_as_image() {
        // iPhone photos sent via WhatsApp arrive as .heic
        let att = whatsapp_attachment("/tmp/photo.heic");
        assert_eq!(att.kind, MessageAttachmentKind::Image);
    }

    #[test]
    fn jpg_and_png_classified_as_image() {
        for ext in &["jpg", "jpeg", "png", "gif", "webp", "bmp"] {
            let path = format!("/tmp/file.{ext}");
            let att = whatsapp_attachment(&path);
            assert_eq!(
                att.kind,
                MessageAttachmentKind::Image,
                ".{ext} must be classified as Image"
            );
        }
    }

    #[test]
    fn mp4_classified_as_video() {
        let att = whatsapp_attachment("/cache/video.mp4");
        assert_eq!(att.kind, MessageAttachmentKind::Video);
    }

    #[test]
    fn ogg_classified_as_audio() {
        let att = whatsapp_attachment("/cache/voice.ogg");
        assert_eq!(att.kind, MessageAttachmentKind::Audio);
    }

    #[test]
    fn unknown_extension_classified_as_other() {
        let att = whatsapp_attachment("/cache/mystery.xyz");
        assert_eq!(att.kind, MessageAttachmentKind::Other);
    }

    #[test]
    fn local_path_is_stored_in_attachment() {
        let path = "/home/user/.lingshu/image_cache/img_abc123.jpg";
        let att = whatsapp_attachment(path);
        assert_eq!(att.local_path.as_deref(), Some(path));
    }

    #[test]
    fn file_name_extracted_from_path() {
        let att = whatsapp_attachment("/cache/img_abc123.jpg");
        assert_eq!(att.file_name.as_deref(), Some("img_abc123.jpg"));
    }

    // ─── render_incoming_text edge cases ──────────────────────────────────

    #[test]
    fn empty_body_with_image_still_shows_attachment_summary() {
        // Edge case: user sends ONLY an image with no caption
        let incoming = WhatsAppInboundEvent {
            message_id: Some("msg-2".into()),
            chat_id: "42@s.whatsapp.net".into(),
            sender_id: "42@s.whatsapp.net".into(),
            sender_name: None,
            from_me: true,
            body: "".into(), // no caption
            media_urls: vec!["/home/.lingshu/image_cache/img_aabbcc.jpg".into()],
        };
        let message = WhatsAppAdapter::event_to_message(incoming);
        // Text must still include the attachment summary for the agent to see the image
        assert!(
            message.text.contains("Shared 1 attachment:"),
            "empty-body image message must produce attachment summary, got: {:?}",
            message.text
        );
        assert!(message.text.contains("image_cache/img_aabbcc.jpg"));
        assert_eq!(message.metadata.attachments.len(), 1);
    }

    #[test]
    fn no_media_returns_plain_body() {
        let incoming = WhatsAppInboundEvent {
            message_id: None,
            chat_id: "99@s.whatsapp.net".into(),
            sender_id: "99@s.whatsapp.net".into(),
            sender_name: None,
            from_me: true,
            body: "Hello, world!".into(),
            media_urls: vec![],
        };
        let message = WhatsAppAdapter::event_to_message(incoming);
        assert_eq!(message.text, "Hello, world!");
        assert!(message.metadata.attachments.is_empty());
    }

    #[test]
    fn allowed_users_plus_prefix_is_stripped() {
        // '+' prefix in phone numbers must be removed so they match WhatsApp JIDs
        let source = WhatsAppGatewayConfig {
            enabled: true,
            allowed_users: vec!["+33612345678".into(), "0044123456".into()],
            ..Default::default()
        };
        let config = WhatsappAdapterConfig::from(&source);
        assert_eq!(config.allowed_users, vec!["33612345678", "0044123456"]);
    }

    #[test]
    fn outbound_phone_number_is_normalized_to_personal_jid() {
        assert_eq!(
            normalize_outbound_chat_id("+33614251689"),
            "33614251689@s.whatsapp.net"
        );
        assert_eq!(
            normalize_outbound_chat_id("336 14 25 16 89"),
            "33614251689@s.whatsapp.net"
        );
    }

    #[test]
    fn outbound_existing_jids_are_preserved() {
        assert_eq!(
            normalize_outbound_chat_id("120363048792887850@g.us"),
            "120363048792887850@g.us"
        );
        assert_eq!(
            normalize_outbound_chat_id("131211435503789@lid"),
            "131211435503789@lid"
        );
    }

    #[test]
    fn outbound_device_scoped_jids_are_reduced_to_chat_jids() {
        assert_eq!(
            normalize_outbound_chat_id("33614251689:10@s.whatsapp.net"),
            "33614251689@s.whatsapp.net"
        );
        assert_eq!(
            normalize_outbound_chat_id("131211435503789:10@lid"),
            "131211435503789@lid"
        );
    }

    #[test]
    fn outbound_suffix_shorthand_is_supported() {
        assert_eq!(
            normalize_outbound_chat_id("131211435503789:lid"),
            "131211435503789@lid"
        );
        assert_eq!(
            normalize_outbound_chat_id("33614251689:s.whatsapp.net"),
            "33614251689@s.whatsapp.net"
        );
    }

    #[tokio::test]
    async fn send_document_uses_bridge_media_endpoint() {
        let captured = Arc::new(Mutex::new(Vec::<Value>::new()));
        let captured_state = Arc::clone(&captured);
        let app = Router::new().route(
            "/send-media",
            post(move |Json(body): Json<Value>| {
                let captured = Arc::clone(&captured_state);
                async move {
                    captured.lock().expect("lock").push(body);
                    Json(json!({ "success": true }))
                }
            }),
        );

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("listener");
        let addr = listener.local_addr().expect("addr");
        let server = tokio::spawn(async move {
            axum::serve(listener, app).await.expect("server");
        });

        let adapter = WhatsAppAdapter::new(WhatsappAdapterConfig {
            bridge_url: Some(format!("http://{addr}")),
            ..Default::default()
        })
        .expect("adapter");
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("report.pdf");
        std::fs::write(&path, b"lingshu").expect("write");
        let metadata = MessageMetadata {
            channel_id: Some("+33614251689".into()),
            ..Default::default()
        };

        adapter
            .send_document(path.to_string_lossy().as_ref(), Some("report"), &metadata)
            .await
            .expect("send document");

        let requests = captured.lock().expect("lock");
        assert_eq!(requests.len(), 1);
        assert_eq!(requests[0]["chatId"], "33614251689@s.whatsapp.net");
        assert_eq!(requests[0]["mediaType"], "document");
        assert_eq!(requests[0]["caption"], "report");
        assert_eq!(requests[0]["filePath"], path.to_string_lossy().as_ref());

        server.abort();
    }

    // ─── self-chat mode: from_me field & filtering ────────────────────────

    #[test]
    fn from_me_defaults_to_false_when_absent() {
        let json = serde_json::json!({
            "chatId": "123@s.whatsapp.net",
            "senderId": "123@s.whatsapp.net",
            "body": "hello"
        });
        let event: WhatsAppInboundEvent = serde_json::from_value(json).expect("deser");
        assert!(
            !event.from_me,
            "from_me must default to false (conservative)"
        );
    }

    #[test]
    fn from_me_true_when_present() {
        let json = serde_json::json!({
            "chatId": "123@s.whatsapp.net",
            "senderId": "123@s.whatsapp.net",
            "body": "hello",
            "fromMe": true
        });
        let event: WhatsAppInboundEvent = serde_json::from_value(json).expect("deser");
        assert!(event.from_me);
    }

    /// Simulates the Rust-side filter predicate used in the polling loop.
    /// `mode != "bot" && !event.from_me` → true means "drop this event".
    fn should_drop(mode: &str, from_me: bool) -> bool {
        mode != "bot" && !from_me
    }

    #[test]
    fn self_chat_mode_drops_contact_messages() {
        // Contact message in self-chat → MUST be dropped
        assert!(should_drop("self-chat", false));
    }

    #[test]
    fn self_chat_mode_passes_own_messages() {
        // Own message in self-chat → MUST pass through
        assert!(!should_drop("self-chat", true));
    }

    #[test]
    fn bot_mode_passes_all_messages() {
        // Bot mode does not use from_me filter (allowlist is enforced separately)
        assert!(!should_drop("bot", false));
        assert!(!should_drop("bot", true));
    }

    /// Simulates the bot-mode allowlist predicate.
    fn bot_allowlist_allows(sender_id: &str, allowed_users: &[&str]) -> bool {
        if allowed_users.is_empty() {
            return true;
        }
        let sender_num = sender_id.split('@').next().unwrap_or(sender_id);
        allowed_users
            .iter()
            .any(|u| *u == sender_num || *u == sender_id)
    }

    #[test]
    fn bot_mode_empty_allowlist_allows_everyone() {
        assert!(bot_allowlist_allows("33612345678@s.whatsapp.net", &[]));
    }

    #[test]
    fn bot_mode_allowlist_accepts_listed_user() {
        assert!(bot_allowlist_allows(
            "33612345678@s.whatsapp.net",
            &["33612345678"]
        ));
    }

    #[test]
    fn bot_mode_allowlist_rejects_unlisted_user() {
        assert!(!bot_allowlist_allows(
            "44700000000@s.whatsapp.net",
            &["33612345678"]
        ));
    }

    #[test]
    fn bot_mode_allowlist_accepts_full_jid() {
        assert!(bot_allowlist_allows(
            "33612345678@s.whatsapp.net",
            &["33612345678@s.whatsapp.net"]
        ));
    }
}