//! SocketManager — persistent Rust-native Socket.IO connection via WebSocket.
//!
//! Implements Engine.IO v4 and Socket.IO v4 protocols directly over WebSocket
//! using `tokio-tungstenite` with `rustls` TLS.
//!
//! Responsibilities:
//! - MCP `listTools` / `toolCall` handled directly via the WorkflowRegistry
//! - Non-MCP server events forwarded to running skills and to the frontend
//! - Connection state logging for observability
//! - Automatic reconnection with exponential backoff

use std::collections::HashMap;
use std::sync::{
    atomic::{AtomicU64, Ordering},
    Arc, OnceLock,
};

use parking_lot::{Mutex, RwLock};
use serde_json::json;
use tokio::sync::{mpsc, oneshot, watch};
use tokio::time::Duration;

use crate::api::models::socket::{ConnectionStatus, SocketState};
use crate::openhuman::webhooks::WebhookRouter;

use super::token_provider::{static_token_provider, TokenProvider};
use super::ws_loop::ws_loop;

// ---------------------------------------------------------------------------
// Global accessor
// ---------------------------------------------------------------------------

static GLOBAL_SOCKET_MANAGER: OnceLock<Arc<SocketManager>> = OnceLock::new();

/// Register the global `SocketManager` instance (called once during bootstrap).
pub fn set_global_socket_manager(mgr: Arc<SocketManager>) {
    if GLOBAL_SOCKET_MANAGER.set(mgr).is_err() {
        log::warn!("[socket] global SocketManager already set — ignoring duplicate");
    }
}

/// Retrieve the global `SocketManager`, if initialized.
pub fn global_socket_manager() -> Option<&'static Arc<SocketManager>> {
    GLOBAL_SOCKET_MANAGER.get()
}

// ---------------------------------------------------------------------------
// Shared state (visible to sibling modules)
// ---------------------------------------------------------------------------

/// State shared between the `SocketManager` handle and the background loop.
pub(super) struct SharedState {
    /// Router for delivering incoming webhooks to skills.
    pub(super) webhook_router: RwLock<Option<Arc<WebhookRouter>>>,
    /// Pending Socket.IO ACK callbacks keyed by outbound ack id.
    pub(super) ack_registry: AckRegistry,
    /// Current connection status.
    pub(super) status: RwLock<ConnectionStatus>,
    /// Socket ID assigned by the server.
    pub(super) socket_id: RwLock<Option<String>>,
    /// Last user-visible connection warning surfaced through `SocketState.error`
    /// (e.g. "backend redirected ws→wss; update BACKEND_URL"). Cleared on every
    /// successful handshake and on disconnect.
    pub(super) error: RwLock<Option<String>>,
}

pub(super) struct AckRegistry {
    next_id: AtomicU64,
    pending: Mutex<HashMap<u64, oneshot::Sender<serde_json::Value>>>,
}

impl Default for AckRegistry {
    fn default() -> Self {
        Self {
            next_id: AtomicU64::new(1),
            pending: Mutex::new(HashMap::new()),
        }
    }
}

impl AckRegistry {
    pub(super) fn register(&self) -> (u64, oneshot::Receiver<serde_json::Value>) {
        let ack_id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let (tx, rx) = oneshot::channel();
        self.pending.lock().insert(ack_id, tx);
        (ack_id, rx)
    }

    pub(super) fn resolve(&self, ack_id: u64, data: serde_json::Value) -> bool {
        if let Some(tx) = self.pending.lock().remove(&ack_id) {
            let _ = tx.send(data);
            true
        } else {
            false
        }
    }

    pub(super) fn remove(&self, ack_id: u64) {
        self.pending.lock().remove(&ack_id);
    }

    pub(super) fn cancel_all(&self) {
        self.pending.lock().clear();
    }
}

// ---------------------------------------------------------------------------
// SocketManager
// ---------------------------------------------------------------------------

/// Manages a persistent Socket.IO connection to the backend.
///
/// Handles protocol-level handshakes (Engine.IO / Socket.IO), heartbeats, and
/// automatic reconnection while providing a high-level API for emitting events
/// and syncing tool state.
pub struct SocketManager {
    /// Shared state accessible from both the manager and the background loop.
    pub(super) shared: Arc<SharedState>,
    /// Channel for sending outgoing messages to the background loop.
    emit_tx: tokio::sync::Mutex<Option<mpsc::UnboundedSender<String>>>,
    /// Channel for signaling the background loop to shut down.
    shutdown_tx: tokio::sync::Mutex<Option<watch::Sender<bool>>>,
    /// Join handle for the background connection loop.
    loop_handle: tokio::sync::Mutex<Option<tokio::task::JoinHandle<()>>>,
}

impl SocketManager {
    /// Create a new, disconnected SocketManager.
    pub fn new() -> Self {
        log::debug!("[socket] SocketManager created (disconnected)");
        Self {
            shared: Arc::new(SharedState {
                webhook_router: RwLock::new(None),
                ack_registry: AckRegistry::default(),
                status: RwLock::new(ConnectionStatus::Disconnected),
                socket_id: RwLock::new(None),
                error: RwLock::new(None),
            }),
            emit_tx: tokio::sync::Mutex::new(None),
            shutdown_tx: tokio::sync::Mutex::new(None),
            loop_handle: tokio::sync::Mutex::new(None),
        }
    }

    /// Set the webhook router for skill-targeted webhook delivery.
    pub fn set_webhook_router(&self, router: Arc<WebhookRouter>) {
        log::debug!("[socket] WebhookRouter attached");
        *self.shared.webhook_router.write() = Some(router);
    }

    /// Get the webhook router, if one has been set.
    pub fn webhook_router(&self) -> Option<Arc<WebhookRouter>> {
        self.shared.webhook_router.read().clone()
    }

    /// Get the current socket state (status, ID, error).
    pub fn get_state(&self) -> SocketState {
        SocketState {
            status: *self.shared.status.read(),
            socket_id: self.shared.socket_id.read().clone(),
            error: self.shared.error.read().clone(),
        }
    }

    /// Check if the socket is currently connected.
    #[allow(dead_code)]
    pub fn is_connected(&self) -> bool {
        *self.shared.status.read() == ConnectionStatus::Connected
    }

    // -----------------------------------------------------------------------
    // Connection lifecycle
    // -----------------------------------------------------------------------

    /// Connect to the specified URL using the provided authentication token.
    ///
    /// Spawns a background `ws_loop` that manages the connection with automatic
    /// reconnection and exponential backoff.
    ///
    /// Returns `Err` immediately if `token` is empty — every reconnect attempt
    /// would either 401 at the SIO CONNECT step or fail upstream at the gateway,
    /// producing exactly the kind of retry-storm noise this module is designed to
    /// suppress. Callers receive an actionable error and the RPC response reflects
    /// the actual outcome rather than optimistically reporting `{"status":"Connecting"}`.
    pub async fn connect(&self, url: &str, token: &str) -> Result<(), String> {
        if token.trim().is_empty() {
            log::error!("[socket] connect: refusing to start — empty session token");
            return Err("empty session token — authenticate first".to_string());
        }
        // Wrap the static token in a provider closure. Existing callers that
        // pass a concrete token value continue to work unchanged; the provider
        // returns that same token on every call (static semantics). For
        // live-session refresh, callers should use `connect_with_session` which
        // builds a provider via `token_provider_from_config`.
        let provider = static_token_provider(token.to_string());
        self.spawn_loop(url, provider).await
    }

    /// Connect using a **live-refresh token provider**.
    ///
    /// Unlike [`connect`] which wraps a single static token, this method
    /// accepts a [`TokenProvider`] closure that is called before every
    /// reconnect attempt. Use this when the token may change between retries
    /// (e.g. after a session refresh or re-login) so the loop always sends the
    /// freshest available credential.
    ///
    /// The provider is called immediately to validate that a token is available
    /// before the background task is spawned — callers receive an actionable
    /// `Err` if no token is stored rather than spawning a doomed retry loop.
    pub async fn connect_with_provider(
        &self,
        url: &str,
        token_provider: TokenProvider,
    ) -> Result<(), String> {
        // Validate that a token is available right now before spawning. This
        // mirrors the empty-token guard in `connect()` and ensures callers
        // see an immediate error if the session store is empty.
        match token_provider() {
            Ok(t) if !t.trim().is_empty() => {}
            Ok(_) => {
                log::error!(
                    "[socket] connect_with_provider: refusing to start — provider returned empty token"
                );
                return Err("empty session token — authenticate first".to_string());
            }
            Err(e) => {
                log::error!(
                    "[socket] connect_with_provider: refusing to start — provider error: {e}"
                );
                return Err(e);
            }
        }
        self.spawn_loop(url, token_provider).await
    }

    /// Shared spawn path used by both [`connect`] and [`connect_with_provider`].
    ///
    /// Installs the rustls crypto provider, tears down any existing connection,
    /// constructs the channel pair, and spawns the background `ws_loop` task.
    /// Entry-point-specific validation (empty-token guard, provider pre-check)
    /// is done by the callers before this is called.
    async fn spawn_loop(&self, url: &str, provider: TokenProvider) -> Result<(), String> {
        // Ensure the rustls crypto provider is installed (needed for wss:// TLS).
        // This is a no-op if already installed.
        let _ = rustls::crypto::ring::default_provider().install_default();

        self.disconnect().await?;

        log::info!("[socket] Connecting to {}", url);

        *self.shared.status.write() = ConnectionStatus::Connecting;
        *self.shared.error.write() = None;
        emit_state_change(&self.shared);

        let (emit_tx, emit_rx) = mpsc::unbounded_channel::<String>();
        let internal_tx = emit_tx.clone();
        let (shutdown_tx, shutdown_rx) = watch::channel(false);

        *self.emit_tx.lock().await = Some(emit_tx);
        *self.shutdown_tx.lock().await = Some(shutdown_tx);

        let url = url.to_string();
        let shared = Arc::clone(&self.shared);

        let handle = tokio::spawn(async move {
            ws_loop(url, provider, shared, emit_rx, shutdown_rx, internal_tx).await;
        });

        *self.loop_handle.lock().await = Some(handle);
        Ok(())
    }

    /// Disconnect from the server and shut down the background loop.
    pub async fn disconnect(&self) -> Result<(), String> {
        if let Some(tx) = self.shutdown_tx.lock().await.take() {
            let _ = tx.send(true);
        }
        self.shared.ack_registry.cancel_all();
        self.emit_tx.lock().await.take();
        if let Some(handle) = self.loop_handle.lock().await.take() {
            let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
        }
        *self.shared.status.write() = ConnectionStatus::Disconnected;
        *self.shared.socket_id.write() = None;
        *self.shared.error.write() = None;
        emit_state_change(&self.shared);
        log::debug!("[socket] Disconnected");
        Ok(())
    }

    /// Emit a Socket.IO event to the server.
    pub async fn emit(&self, event: &str, data: serde_json::Value) -> Result<(), String> {
        if let Some(ref tx) = *self.emit_tx.lock().await {
            let msg = encode_sio_event(event, data, None)?;
            tx.send(msg).map_err(|_| "Socket not connected".to_string())
        } else {
            Err("Not connected".to_string())
        }
    }

    /// Emit a Socket.IO event and wait for the backend ACK callback.
    pub async fn emit_with_ack(
        &self,
        event: &str,
        data: serde_json::Value,
        timeout: Duration,
    ) -> Result<serde_json::Value, String> {
        let tx = self
            .emit_tx
            .lock()
            .await
            .clone()
            .ok_or_else(|| "Not connected".to_string())?;
        let (ack_id, ack_rx) = self.shared.ack_registry.register();
        let msg = encode_sio_event(event, data, Some(ack_id))?;
        if let Err(e) = tx.send(msg) {
            self.shared.ack_registry.remove(ack_id);
            return Err(format!("Socket not connected: {e}"));
        }

        log::debug!("[socket] emit_with_ack sent event={event} ack_id={ack_id}");
        match tokio::time::timeout(timeout, ack_rx).await {
            Ok(Ok(data)) => {
                log::debug!("[socket] emit_with_ack resolved event={event} ack_id={ack_id}");
                Ok(data)
            }
            Ok(Err(_)) => Err(format!(
                "Socket ack channel dropped for event {event} ack_id={ack_id}"
            )),
            Err(_) => {
                self.shared.ack_registry.remove(ack_id);
                Err(format!(
                    "Socket ack timeout for event {event} ack_id={ack_id}"
                ))
            }
        }
    }
}

fn encode_sio_event(
    event: &str,
    data: serde_json::Value,
    ack_id: Option<u64>,
) -> Result<String, String> {
    let payload = serde_json::to_string(&json!([event, data])).map_err(|e| format!("{e}"))?;
    let ack = ack_id.map(|id| id.to_string()).unwrap_or_default();
    Ok(format!("42{ack}{payload}"))
}

impl Default for SocketManager {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// State-change helpers (used by sibling modules)
// ---------------------------------------------------------------------------

/// Log a state change for observability.
pub(super) fn emit_state_change(shared: &SharedState) {
    let status = *shared.status.read();
    let socket_id = shared.socket_id.read().clone();
    log::debug!("[socket] State changed: {:?}, sid={:?}", status, socket_id);
}

/// Log a server event for observability.
pub(super) fn emit_server_event(_shared: &SharedState, event_name: &str, _data: serde_json::Value) {
    log::debug!("[socket] Server event: {}", event_name);
}

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

    #[test]
    fn new_manager_is_disconnected_with_no_sid() {
        let mgr = SocketManager::new();
        let state = mgr.get_state();
        assert_eq!(state.status, ConnectionStatus::Disconnected);
        assert!(state.socket_id.is_none());
        assert!(state.error.is_none());
        assert!(!mgr.is_connected());
    }

    #[test]
    fn default_impl_matches_new() {
        let a = SocketManager::new();
        let b = SocketManager::default();
        assert_eq!(a.get_state().status, b.get_state().status);
    }

    #[test]
    fn is_connected_tracks_status_transitions() {
        let mgr = SocketManager::new();
        assert!(!mgr.is_connected());
        *mgr.shared.status.write() = ConnectionStatus::Connected;
        assert!(mgr.is_connected());
        *mgr.shared.status.write() = ConnectionStatus::Error;
        assert!(!mgr.is_connected());
    }

    #[test]
    fn get_state_reflects_stored_sid_and_status() {
        let mgr = SocketManager::new();
        *mgr.shared.status.write() = ConnectionStatus::Connected;
        *mgr.shared.socket_id.write() = Some("sid-abc".to_string());
        let state = mgr.get_state();
        assert_eq!(state.status, ConnectionStatus::Connected);
        assert_eq!(state.socket_id.as_deref(), Some("sid-abc"));
    }

    #[test]
    fn get_state_surfaces_stored_error_to_callers() {
        let mgr = SocketManager::new();
        *mgr.shared.error.write() =
            Some("backend redirected ws→wss; update BACKEND_URL".to_string());
        let state = mgr.get_state();
        assert_eq!(
            state.error.as_deref(),
            Some("backend redirected ws→wss; update BACKEND_URL")
        );
    }

    #[tokio::test]
    async fn emit_without_connection_errors_without_panic() {
        let mgr = SocketManager::new();
        let err = mgr.emit("test.event", json!({"k":"v"})).await.unwrap_err();
        assert_eq!(err, "Not connected");
    }

    #[tokio::test]
    async fn emit_with_ack_without_connection_errors_without_waiting() {
        let mgr = SocketManager::new();
        let err = mgr
            .emit_with_ack("test.event", json!({"k":"v"}), Duration::from_secs(30))
            .await
            .unwrap_err();
        assert_eq!(err, "Not connected");
    }

    #[tokio::test]
    async fn emit_with_ack_uses_emit_queue_while_connecting() {
        let mgr = SocketManager::new();
        let (tx, mut rx) = mpsc::unbounded_channel::<String>();
        *mgr.emit_tx.lock().await = Some(tx);
        *mgr.shared.status.write() = ConnectionStatus::Connecting;

        let result = mgr
            .emit_with_ack("test.event", json!({"k": "v"}), Duration::from_millis(10))
            .await;

        let queued = rx
            .try_recv()
            .unwrap_or_else(|_| panic!("expected queued ACK emit, got result={result:?}"));
        assert_eq!(queued, r#"421["test.event",{"k":"v"}]"#);
        let err = result.unwrap_err();
        assert!(
            err.starts_with("Socket ack timeout for event test.event ack_id=1"),
            "unexpected error: {err}"
        );
    }

    #[tokio::test]
    async fn disconnect_on_fresh_manager_is_idempotent() {
        let mgr = SocketManager::new();
        assert!(mgr.disconnect().await.is_ok());
        // Calling again must still succeed.
        assert!(mgr.disconnect().await.is_ok());
        assert_eq!(mgr.get_state().status, ConnectionStatus::Disconnected);
    }

    #[test]
    fn emit_state_change_is_safe_to_call_on_empty_shared() {
        let shared = SharedState {
            webhook_router: RwLock::new(None),
            ack_registry: AckRegistry::default(),
            status: RwLock::new(ConnectionStatus::Connecting),
            socket_id: RwLock::new(None),
            error: RwLock::new(None),
        };
        // Must not panic even with all default state.
        emit_state_change(&shared);
    }

    #[test]
    fn emit_server_event_is_safe_without_subscribers() {
        let shared = SharedState {
            webhook_router: RwLock::new(None),
            ack_registry: AckRegistry::default(),
            status: RwLock::new(ConnectionStatus::Connected),
            socket_id: RwLock::new(Some("x".into())),
            error: RwLock::new(None),
        };
        // Pure logging — must not touch state or panic.
        emit_server_event(&shared, "any.event", json!({}));
        assert_eq!(*shared.status.read(), ConnectionStatus::Connected);
    }

    #[test]
    fn set_webhook_router_populates_the_shared_slot() {
        let mgr = SocketManager::new();
        assert!(mgr.shared.webhook_router.read().is_none());
        let router = Arc::new(WebhookRouter::new(None));
        mgr.set_webhook_router(router);
        assert!(mgr.shared.webhook_router.read().is_some());
    }

    #[test]
    fn set_webhook_router_overwrites_previous_router() {
        // Replacing the router is allowed so callers can hot-swap during
        // reconfiguration — this test nails that observable behaviour down.
        let mgr = SocketManager::new();
        mgr.set_webhook_router(Arc::new(WebhookRouter::new(None)));
        let second = Arc::new(WebhookRouter::new(None));
        let second_ptr = Arc::as_ptr(&second);
        mgr.set_webhook_router(Arc::clone(&second));
        let stored = mgr.shared.webhook_router.read().clone().unwrap();
        assert!(std::ptr::eq(Arc::as_ptr(&stored), second_ptr));
    }

    #[tokio::test]
    async fn emit_after_disconnect_errors_not_connected() {
        // Even without ever calling connect(), the disconnect() call path
        // leaves the emit channel torn down — and emit() must reject.
        let mgr = SocketManager::new();
        mgr.disconnect().await.unwrap();
        let err = mgr.emit("x", json!({})).await.unwrap_err();
        assert_eq!(err, "Not connected");
    }

    /// Empty-token guard at the `SocketManager::connect` boundary:
    /// the RPC caller must receive an `Err` immediately — not
    /// `{"status":"Connecting"}` — so the UI can surface an actionable error.
    #[tokio::test]
    async fn connect_rejects_empty_token_and_returns_err() {
        let mgr = SocketManager::new();

        // Bare empty string.
        let err = mgr.connect("http://localhost:1", "").await.unwrap_err();
        assert!(
            err.contains("empty session token"),
            "expected 'empty session token' in error, got: {err}"
        );
        assert_eq!(mgr.get_state().status, ConnectionStatus::Disconnected);

        // Whitespace-only string (trim check).
        let err = mgr.connect("http://localhost:1", "   ").await.unwrap_err();
        assert!(err.contains("empty session token"), "{err}");
        assert_eq!(mgr.get_state().status, ConnectionStatus::Disconnected);
    }
}
