//! Client for talking to the mux daemon.

use super::endpoint::{SocketEndpoint, socket_endpoint};
use super::ipc;
use super::protocol::{MuxRequest, MuxResponse};
use anyhow::{Context, Result};
use interprocess::local_socket::Stream;
use interprocess::local_socket::traits::Stream as StreamTrait;
use parking_lot::Mutex;
#[cfg(test)]
use std::collections::HashMap;
use std::path::PathBuf;
use std::process::Command;
use std::sync::OnceLock;
use std::time::Duration;

#[cfg(not(test))]
static CLIENT: OnceLock<Mutex<MuxClient>> = OnceLock::new();
#[cfg(not(test))]
static ENDPOINT: OnceLock<SocketEndpoint> = OnceLock::new();
#[cfg(test)]
static TEST_CLIENTS: OnceLock<Mutex<HashMap<String, MuxClient>>> = OnceLock::new();
const DAEMON_CONNECT_RETRY_INTERVAL: Duration = Duration::from_millis(25);
#[cfg(test)]
const DAEMON_CONNECT_RETRY_ATTEMPTS: usize = 800;
#[cfg(not(test))]
const DAEMON_CONNECT_RETRY_ATTEMPTS: usize = 200;

/// Send a request to the mux daemon.
///
/// This will start the daemon if needed.
///
/// # Errors
///
/// Returns an error if the daemon cannot be reached or the request fails.
pub fn request(req: &MuxRequest) -> Result<MuxResponse> {
    let endpoint = endpoint()?;

    #[cfg(test)]
    {
        let scope = test_scope_key();
        {
            let clients = TEST_CLIENTS.get_or_init(|| Mutex::new(HashMap::new()));
            let mut clients = clients.lock();
            let client = clients
                .entry(scope)
                .or_insert_with(|| MuxClient::new(endpoint.clone()));
            if client.endpoint.display != endpoint.display {
                *client = MuxClient::new(endpoint);
            }
            let response = client.request(req);
            drop(clients);
            response
        }
    }

    #[cfg(not(test))]
    {
        let client = CLIENT.get_or_init(|| Mutex::new(MuxClient::new(endpoint)));
        let mut client = client.lock();
        client.request(req)
    }
}

pub(super) fn endpoint() -> Result<SocketEndpoint> {
    #[cfg(test)]
    {
        socket_endpoint()
    }

    #[cfg(not(test))]
    {
        if let Some(endpoint) = ENDPOINT.get() {
            return Ok(endpoint.clone());
        }

        let endpoint = socket_endpoint()?;
        let _ = ENDPOINT.set(endpoint.clone());
        Ok(endpoint)
    }
}

/// A synchronous request/response mux client.
#[derive(Debug)]
pub struct MuxClient {
    /// IPC endpoint.
    pub endpoint: SocketEndpoint,
    /// Active connection when available.
    pub stream: Option<Stream>,
}

impl MuxClient {
    /// Create a new client for an endpoint.
    #[must_use]
    pub const fn new(endpoint: SocketEndpoint) -> Self {
        Self {
            endpoint,
            stream: None,
        }
    }

    /// Send a request.
    ///
    /// # Errors
    ///
    /// Returns an error if the daemon cannot be reached or the request fails.
    pub fn request(&mut self, req: &MuxRequest) -> Result<MuxResponse> {
        self.ensure_connected()?;

        if let Some(stream) = self.stream.as_mut()
            && let Ok(response) = send_request(stream, req)
        {
            return Ok(response);
        }

        self.stream = None;
        self.ensure_connected()?;
        let stream = self
            .stream
            .as_mut()
            .context("Mux stream missing after reconnect")?;
        send_request(stream, req)
    }

    fn ensure_connected(&mut self) -> Result<()> {
        if self.stream.is_some() {
            return Ok(());
        }

        if let Ok(stream) = Stream::connect(self.endpoint.name.clone()) {
            self.stream = Some(stream);
            return Ok(());
        }

        start_daemon(&self.endpoint)?;

        for _ in 0..DAEMON_CONNECT_RETRY_ATTEMPTS {
            match Stream::connect(self.endpoint.name.clone()) {
                Ok(stream) => {
                    self.stream = Some(stream);
                    return Ok(());
                }
                Err(_) => std::thread::sleep(DAEMON_CONNECT_RETRY_INTERVAL),
            }
        }

        Err(anyhow::anyhow!(
            "Failed to connect to mux daemon at {}",
            self.endpoint.display
        ))
    }
}

fn send_request(stream: &mut Stream, req: &MuxRequest) -> Result<MuxResponse> {
    ipc::write_json(stream, req)?;
    ipc::read_json(stream)
}

fn start_daemon(endpoint: &SocketEndpoint) -> Result<()> {
    let exe = resolve_tenex_executable()?;
    #[cfg(test)]
    let mut cmd = {
        let current_exe =
            std::env::current_exe().context("Failed to resolve current test executable")?;
        let is_test_harness_fallback = exe == current_exe
            && exe
                .file_stem()
                .and_then(|name| name.to_str())
                .is_none_or(|name| name != env!("CARGO_PKG_NAME"));

        if is_test_harness_fallback {
            let mut cmd = Command::new(current_exe);
            cmd.args(["--exact", "__tenex_muxd_test_entry", "--nocapture"])
                .env("TENEX_RUN_MUXD_TEST_ENTRY", "1");
            cmd
        } else {
            let mut cmd = Command::new(exe);
            cmd.arg("muxd");
            cmd
        }
    };

    #[cfg(not(test))]
    let mut cmd = {
        let mut cmd = Command::new(exe);
        cmd.arg("muxd");
        cmd
    };

    cmd.env("TENEX_MUX_SOCKET", &endpoint.display);

    cmd.stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null());

    // Put the mux daemon in a separate process group so Ctrl+C in the Tenex TUI
    // doesn't tear down the daemon (and thus all agent sessions).
    #[cfg(target_os = "linux")]
    {
        use std::os::unix::process::CommandExt as _;
        cmd.process_group(0);
    }

    cmd.spawn().context("Failed to spawn mux daemon")?;
    Ok(())
}

fn resolve_tenex_executable() -> Result<PathBuf> {
    let exe = std::env::current_exe().context("Failed to resolve current executable")?;

    let is_tenex = exe
        .file_stem()
        .and_then(|name| name.to_str())
        .is_some_and(|name| name == env!("CARGO_PKG_NAME"));
    if is_tenex {
        return Ok(exe);
    }

    let candidate_dir = exe
        .parent()
        .and_then(std::path::Path::parent)
        .context("Failed to resolve build output directory")?;

    let base_name = env!("CARGO_PKG_NAME");
    let candidates = [
        candidate_dir.join(format!("{base_name}{}", std::env::consts::EXE_SUFFIX)),
        candidate_dir.join(base_name),
    ];

    for candidate in candidates {
        if candidate.exists() {
            return Ok(candidate);
        }
    }

    Ok(exe)
}

#[cfg(test)]
fn test_scope_key() -> String {
    std::thread::current().name().map_or_else(
        || format!("{:?}", std::thread::current().id()),
        std::borrow::ToOwned::to_owned,
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use anyhow::anyhow;

    #[test]
    fn __tenex_muxd_test_entry() -> Result<()> {
        if std::env::var_os("TENEX_RUN_MUXD_TEST_ENTRY").is_some() {
            crate::mux::run_mux_daemon()?;
        }
        Ok(())
    }

    #[test]
    fn test_resolve_tenex_executable_returns_existing_path() -> Result<()> {
        let exe = resolve_tenex_executable()?;
        assert!(exe.exists());
        Ok(())
    }

    #[test]
    fn test_test_scope_key_uses_current_thread_name() {
        let thread = std::thread::current();
        let current = thread.name().unwrap_or("unknown");
        assert_eq!(test_scope_key(), current);
    }

    #[test]
    fn test_test_scope_key_falls_back_for_unnamed_thread() -> Result<(), Box<dyn std::error::Error>>
    {
        let handle = std::thread::spawn(test_scope_key);
        let scope = handle
            .join()
            .map_err(|_| anyhow!("Unnamed client thread panicked"))?;
        assert!(scope.starts_with("ThreadId("));
        Ok(())
    }

    #[test]
    fn test_endpoint_is_not_cached_across_named_threads() -> Result<()> {
        let first = std::thread::Builder::new()
            .name("client-endpoint-one".to_string())
            .spawn(|| -> Result<String> {
                crate::mux::set_socket_override("tenex-mux-client-one")?;
                Ok(endpoint()?.display)
            })?
            .join()
            .map_err(|_| anyhow!("First client endpoint thread panicked"))??;
        let second = std::thread::Builder::new()
            .name("client-endpoint-two".to_string())
            .spawn(|| -> Result<String> {
                crate::mux::set_socket_override("tenex-mux-client-two")?;
                Ok(endpoint()?.display)
            })?
            .join()
            .map_err(|_| anyhow!("Second client endpoint thread panicked"))??;

        assert_eq!(first, "tenex-mux-client-one");
        assert_eq!(second, "tenex-mux-client-two");
        Ok(())
    }

    #[test]
    fn test_mux_client_reconnects_after_failed_request() -> Result<()> {
        use interprocess::local_socket::traits::ListenerExt;
        use interprocess::local_socket::{ListenerOptions, prelude::*};

        #[cfg(windows)]
        let endpoint = {
            use interprocess::local_socket::GenericNamespaced;

            let display = format!("tenex-mux-client-{}", uuid::Uuid::new_v4());
            let name = display
                .clone()
                .to_ns_name::<GenericNamespaced>()?
                .into_owned();
            SocketEndpoint {
                name,
                cleanup_path: None,
                display,
            }
        };

        #[cfg(not(windows))]
        let endpoint = {
            use interprocess::local_socket::GenericFilePath;

            let socket_name = format!("tx-mux-client-{}.sock", uuid::Uuid::new_v4().simple());
            let socket_path = if std::path::Path::new("/tmp").is_dir() {
                std::path::Path::new("/tmp").join(&socket_name)
            } else {
                std::env::temp_dir().join(socket_name)
            };
            let name = socket_path
                .as_path()
                .to_fs_name::<GenericFilePath>()?
                .into_owned();
            SocketEndpoint {
                name,
                cleanup_path: Some(socket_path.clone()),
                display: socket_path.to_string_lossy().into_owned(),
            }
        };

        let listener = ListenerOptions::new()
            .name(endpoint.name.clone())
            .create_sync()?;
        let handle = std::thread::spawn(move || -> Result<()> {
            let mut incoming = listener.incoming();

            let mut stream = incoming
                .next()
                .ok_or_else(|| anyhow::anyhow!("Missing initial connection"))??;
            let _: MuxRequest = ipc::read_json(&mut stream)?;
            drop(stream);

            let mut stream = incoming
                .next()
                .ok_or_else(|| anyhow::anyhow!("Missing reconnect connection"))??;
            let _: MuxRequest = ipc::read_json(&mut stream)?;
            ipc::write_json(
                &mut stream,
                &MuxResponse::Pong {
                    version: "test".to_string(),
                },
            )?;
            Ok(())
        });

        let mut client = MuxClient::new(endpoint);
        let response = client.request(&MuxRequest::Ping)?;
        assert!(matches!(response, MuxResponse::Pong { .. }));

        match handle.join() {
            Ok(Ok(())) => Ok(()),
            Ok(Err(err)) => Err(err),
            Err(_) => Err(anyhow::anyhow!("Server thread panicked")),
        }
    }
}
