//! `SessionDriver` impl for tail-fin daemon sessions.
//!
//! Each session is tied to one **site** (one browser tab) and one **Chrome host**.
//! `TailFinSession` owns a `Box<dyn SiteHandler>` created at launch time, so
//! stateful site adapters can keep per-session client state across multiple
//! commands without rebuilding from scratch.

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

use anyhow::{anyhow, Context};
use night_fury_core::{BrowserSession, SessionBuilder};
use night_fury_daemon_core::{protocol::Response, LaunchError, SessionDriver};
use serde_json::Value;
use tail_fin_cli_core::{browser_session, launch_stealth_session_blocking_cf};
use tail_fin_core::Site;

use crate::handlers::SiteHandler;

/// How a registered site's session is built. Drives [`TailFinSession::launch`]:
/// `Browser` builds a Chrome session (attach or stealth); `Http` builds no
/// browser at all (pure-HTTP descriptor sites — arkham, alphora, gemini).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SiteTransport {
    Browser,
    Http,
}

/// How `launch` obtains a browser for this site. `AttachOnly` sites must
/// attach to an existing user tab (fresh CDP attach is blocked upstream —
/// e.g. upstream anti-bot session binding) and reject `--host auto`.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LaunchPolicy {
    Standard,
    AttachOnly { url_fragment: &'static str },
}

enum HandlerFactory {
    Browser(Arc<dyn Fn(BrowserSession) -> Box<dyn SiteHandler> + Send + Sync>),
    Http(Arc<dyn Fn() -> Box<dyn SiteHandler> + Send + Sync>),
}

impl Clone for HandlerFactory {
    fn clone(&self) -> Self {
        match self {
            Self::Browser(make) => Self::Browser(Arc::clone(make)),
            Self::Http(make) => Self::Http(Arc::clone(make)),
        }
    }
}

/// Pool-key host token for HTTP entries — an HTTP launch ignores the requested
/// host, so the pool keys all HTTP sessions under one fixed token.
pub const HTTP_POOL_HOST: &str = "http";

/// Bundles a `Site` (lifecycle) with a handler factory (command execution).
/// Register one per site via [`register_sites`].
///
/// `Clone` is cheap — every field is an `Arc` or a `'static`
/// ref — so the registry can be built once (see `registry::all_site_entries`)
/// and cloned into the server-side map without rebuilding.
#[derive(Clone)]
pub struct SiteEntry {
    pub site: Arc<dyn Site>,
    /// Descriptor table exported by the backing `CommandModule`.
    pub specs: &'static [tail_fin_core::CommandSpec],
    launch_policy: LaunchPolicy,
    factory: HandlerFactory,
}

impl SiteEntry {
    pub fn launch_policy(&self) -> LaunchPolicy {
        self.launch_policy
    }

    /// Whether `launch` builds a browser or skips it (pure HTTP).
    pub fn transport(&self) -> SiteTransport {
        match &self.factory {
            HandlerFactory::Browser(_) => SiteTransport::Browser,
            HandlerFactory::Http(_) => SiteTransport::Http,
        }
    }

    /// Register a site backed by a descriptor [`CommandModule`]. Builds an
    /// [`ExecutorHandler`] per session from `M::build_executor`, and records
    /// `M::specs()` on the entry. Use for browser-served migrated sites.
    pub fn from_module<M>(site: impl Site + 'static) -> Self
    where
        M: tail_fin_core::CommandModule + 'static,
    {
        let site_id = site.id();
        Self {
            site: Arc::new(site),
            specs: M::specs(),
            launch_policy: LaunchPolicy::Standard,
            factory: HandlerFactory::Browser(Arc::new(move |session| {
                Box::new(crate::handlers::executor::ExecutorHandler::new(
                    M::build_executor(tail_fin_core::SessionCtx::Browser {
                        session,
                        cookies_path: Some(tail_fin_cli_core::default_cookies_path(site_id)),
                    }),
                ))
            })),
        }
    }

    /// Register a pure-HTTP descriptor site. Builds the per-session
    /// [`ExecutorHandler`] from `M::build_executor(SessionCtx::Http(..))` with no
    /// browser. The cookie file defaults to `~/.tail-fin/<site>-cookies.txt`
    /// (anonymous sites ignore it; cookie-auth sites like gemini read it). Tags
    /// the entry `SiteTransport::Http` so `launch` skips Chrome.
    pub fn from_http_module<M>(site: impl Site + 'static) -> Self
    where
        M: tail_fin_core::CommandModule + 'static,
    {
        let site_id = site.id();
        Self {
            site: Arc::new(site),
            specs: M::specs(),
            launch_policy: LaunchPolicy::Standard,
            factory: HandlerFactory::Http(Arc::new(move || {
                let cookies_path = Some(tail_fin_cli_core::default_cookies_path(site_id));
                Box::new(crate::handlers::executor::ExecutorHandler::new(
                    M::build_executor(tail_fin_core::SessionCtx::Http(tail_fin_core::HttpCtx {
                        cookies_path,
                    })),
                ))
            })),
        }
    }

    pub fn with_launch_policy(mut self, policy: LaunchPolicy) -> Self {
        self.launch_policy = policy;
        self
    }
}

/// Global registry of known sites, keyed by `Site::id()`.
static SITE_REGISTRY: OnceLock<HashMap<&'static str, SiteEntry>> = OnceLock::new();

/// Register all sites this daemon instance can serve. Call once at startup
/// before `serve()`. Subsequent calls are no-ops.
pub fn register_sites(entries: Vec<SiteEntry>) {
    let map: HashMap<&'static str, SiteEntry> =
        entries.into_iter().map(|e| (e.site.id(), e)).collect();
    let _ = SITE_REGISTRY.set(map);
}

/// The registered transport for a site, if known. The daemon pool uses this to
/// key HTTP sites host-agnostically — an HTTP launch ignores the requested host,
/// so keying the pool by host would needlessly fragment session reuse.
pub fn site_transport(id: &str) -> Option<SiteTransport> {
    SITE_REGISTRY.get()?.get(id).map(|e| e.transport())
}

/// Return all registered site ids, for CLI --help / error messages.
pub fn registered_site_ids() -> Vec<&'static str> {
    SITE_REGISTRY
        .get()
        .map(|m| m.keys().copied().collect())
        .unwrap_or_default()
}

/// One browser tab bound to a single site + host.
pub struct TailFinSession {
    pub site: Arc<dyn Site>,
    pub host: String,
    handler: Box<dyn SiteHandler>,
    /// Profile dir for stealth-launched Chromium when host="auto". `None`
    /// when attaching to user Chrome via --connect. Drops AFTER `handler`
    /// so Chromium tears down before the profile dir is unlinked.
    _profile: Option<tempfile::TempDir>,
}

impl SessionDriver for TailFinSession {
    async fn launch(params: Value) -> Result<Self, LaunchError> {
        let site_id = params
            .get("site")
            .and_then(|v| v.as_str())
            .ok_or_else(|| anyhow!("launch: missing 'site'"))?;

        // Warn in debug builds if register_sites() was never called — this is
        // almost always a startup wiring bug, not a missing site.
        debug_assert!(
            SITE_REGISTRY.get().is_some(),
            "SITE_REGISTRY is uninitialised: call register_sites() before serve()"
        );
        // Treat an uninitialized registry the same as an empty one: the site
        // is simply not registered, so return the canonical "unknown site" error.
        let empty_map = HashMap::new();
        let registry = SITE_REGISTRY.get().unwrap_or(&empty_map);

        let entry = registry.get(site_id).ok_or_else(|| {
            let known = registered_site_ids().join(", ");
            if known.is_empty() {
                anyhow!("unknown site '{site_id}' (no sites registered)")
            } else {
                anyhow!("unknown site '{site_id}' (registered: {known})")
            }
        })?;

        let site = entry.site.clone();
        let make_browser_handler = match &entry.factory {
            // Pure-HTTP sites build no browser: no Chrome, no TempDir, no CDP.
            // The handler is created from a `SessionCtx::Http` executor (cookie
            // path resolved by `from_http_module`). `host` is irrelevant here.
            HandlerFactory::Http(make) => {
                let make = Arc::clone(make);
                return Ok(Self {
                    site,
                    host: HTTP_POOL_HOST.to_string(),
                    handler: make(),
                    _profile: None,
                });
            }
            HandlerFactory::Browser(make) => Arc::clone(make),
        };

        let host = params
            .get("host")
            .and_then(|v| v.as_str())
            .unwrap_or(crate::cli::DEFAULT_CDP_HOST)
            .to_string();

        let (profile, session) = match entry.launch_policy {
            LaunchPolicy::AttachOnly { url_fragment } => {
                if host == "auto" {
                    return Err(LaunchError::new(format!(
                        "site='{site_id}' does not support --host auto; pass --host <ip:port> with an already-open tab (attach-only site)"
                    )));
                }

                let session = SessionBuilder::default()
                    .connect_to(format!("ws://{host}"))
                    .attach_to_existing_url_containing(url_fragment)
                    .headed(false)
                    .build()
                    .await
                    .with_context(|| format!("connect+attach to site='{site_id}' tab at {host}"))?;
                (None, session)
            }
            LaunchPolicy::Standard if host == "auto" => {
                // Daemon-managed stealth Chromium. Each pool member is its own process;
                // the TempDir is held alongside the session in TailFinSession so it
                // drops when the session does. Blocking CF wait (30s) ensures the
                // session has cleared the interstitial before any request fires.
                let url: &'static str = site.refresh_url();
                let (temp_dir, browser_session) = launch_stealth_session_blocking_cf(
                    url,
                    false,
                    std::time::Duration::from_secs(30),
                )
                .await
                .with_context(|| format!("launch stealth Chromium for site={site_id}"))?;
                (Some(temp_dir), browser_session)
            }
            LaunchPolicy::Standard => {
                // Standard CDP attach to a user-managed Chrome at <ip:port>.
                let session = browser_session(&host, false)
                    .await
                    .with_context(|| format!("connect to chrome at {host}"))?;
                (None, session)
            }
        };

        let handler = make_browser_handler(session);

        Ok(Self {
            site,
            host,
            handler,
            _profile: profile,
        })
    }

    async fn handle(&self, id: &str, cmd: &str, params: &Value) -> (Response, bool) {
        let resp = self.handler.handle(id, cmd, params).await;
        (resp, false)
    }
}

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

    struct FakeSite;

    #[async_trait::async_trait]
    impl Site for FakeSite {
        fn id(&self) -> &'static str {
            "fake"
        }
        fn display_name(&self) -> &'static str {
            "Fake"
        }
        fn cookie_domain_patterns(&self) -> &'static [&'static str] {
            &["*.fake.test"]
        }
        fn refresh_url(&self) -> &'static str {
            "https://fake.test/"
        }
    }

    #[test]
    fn from_module_records_specs() {
        use tail_fin_core::{
            ArgSpec, CommandModule, CommandSpec, Executor, OutputShape, SessionCtx, Transport,
        };

        const SPECS: &[CommandSpec] = &[CommandSpec {
            name: "ping",
            wire: "fake.ping",
            about: "ping",
            args: &[] as &[ArgSpec],
            transport: Transport::Browser,
            output: OutputShape::Json,
        }];

        struct M;
        impl CommandModule for M {
            fn site_id() -> &'static str {
                "fake"
            }
            fn specs() -> &'static [CommandSpec] {
                SPECS
            }
            fn build_executor(_ctx: SessionCtx) -> Box<dyn Executor> {
                // Never invoked in this test (no session is created).
                unreachable!("build_executor not called in specs-marker test")
            }
        }

        let entry = SiteEntry::from_module::<M>(FakeSite);
        // CommandSpec does not derive PartialEq, so compare the recorded specs
        // by their wire set (the parity property — same approach as Task 4's
        // reddit_parity guard), proving the module's specs were recorded.
        let recorded: Vec<&str> = entry.specs.iter().map(|s| s.wire).collect();
        let expected: Vec<&str> = SPECS.iter().map(|s| s.wire).collect();
        assert_eq!(recorded, expected);
        assert_eq!(entry.site.id(), FakeSite.id());
        assert_eq!(entry.transport(), SiteTransport::Browser);
    }

    #[test]
    fn site_entries_default_to_standard_launch_policy() {
        use tail_fin_core::{
            ArgSpec, CommandModule, CommandSpec, Executor, OutputShape, SessionCtx, Transport,
        };

        const SPECS: &[CommandSpec] = &[CommandSpec {
            name: "ping",
            wire: "fake.ping",
            about: "ping",
            args: &[] as &[ArgSpec],
            transport: Transport::Browser,
            output: OutputShape::Json,
        }];

        struct M;
        impl CommandModule for M {
            fn site_id() -> &'static str {
                "fake"
            }
            fn specs() -> &'static [CommandSpec] {
                SPECS
            }
            fn build_executor(_ctx: SessionCtx) -> Box<dyn Executor> {
                unreachable!("build_executor not called in launch-policy test")
            }
        }

        let browser_entry = SiteEntry::from_module::<M>(FakeSite);
        let http_entry = SiteEntry::from_http_module::<M>(FakeSite);

        assert_eq!(browser_entry.launch_policy(), LaunchPolicy::Standard);
        assert_eq!(http_entry.launch_policy(), LaunchPolicy::Standard);
    }

    #[test]
    fn with_launch_policy_records_attach_only_policy() {
        use tail_fin_core::{
            ArgSpec, CommandModule, CommandSpec, Executor, OutputShape, SessionCtx, Transport,
        };

        const SPECS: &[CommandSpec] = &[CommandSpec {
            name: "ping",
            wire: "fake.ping",
            about: "ping",
            args: &[] as &[ArgSpec],
            transport: Transport::Browser,
            output: OutputShape::Json,
        }];

        struct M;
        impl CommandModule for M {
            fn site_id() -> &'static str {
                "fake"
            }
            fn specs() -> &'static [CommandSpec] {
                SPECS
            }
            fn build_executor(_ctx: SessionCtx) -> Box<dyn Executor> {
                unreachable!("build_executor not called in launch-policy test")
            }
        }

        let entry =
            SiteEntry::from_module::<M>(FakeSite).with_launch_policy(LaunchPolicy::AttachOnly {
                url_fragment: "fake.test",
            });

        assert_eq!(
            entry.launch_policy(),
            LaunchPolicy::AttachOnly {
                url_fragment: "fake.test"
            }
        );
    }

    #[tokio::test]
    async fn attach_only_site_rejects_host_auto() {
        register_sites(crate::registry::all_site_entries().to_vec());
        let site_id = crate::registry::all_site_entries()
            .iter()
            .find(|entry| matches!(entry.launch_policy(), LaunchPolicy::AttachOnly { .. }))
            .map(|entry| entry.site.id())
            .expect("at least one attach-only site is registered");

        let err = match TailFinSession::launch(serde_json::json!({
            "site": site_id,
            "host": "auto"
        }))
        .await
        {
            Ok(_) => panic!("attach-only site must reject --host auto before launching a browser"),
            Err(err) => err,
        };
        let msg = err.to_string();

        assert!(
            msg.contains("does not support --host auto"),
            "unexpected error: {msg}"
        );
        assert!(msg.contains("attach-only site"), "unexpected error: {msg}");
    }

    #[test]
    fn from_http_module_records_specs_and_http_transport() {
        use tail_fin_core::{
            ArgSpec, CommandModule, CommandSpec, Executor, OutputShape, SessionCtx, Transport,
        };

        const SPECS: &[CommandSpec] = &[CommandSpec {
            name: "ping",
            wire: "fake.ping",
            about: "ping",
            args: &[] as &[ArgSpec],
            transport: Transport::Http,
            output: OutputShape::Json,
        }];

        struct HExec {
            saw_http: bool,
        }
        #[async_trait::async_trait]
        impl Executor for HExec {
            async fn execute(
                &self,
                wire: &str,
                _p: serde_json::Value,
            ) -> Result<serde_json::Value, tail_fin_common::TailFinError> {
                Ok(serde_json::json!({"wire": wire, "saw_http": self.saw_http}))
            }
        }

        struct M;
        impl CommandModule for M {
            fn site_id() -> &'static str {
                "fake"
            }
            fn specs() -> &'static [CommandSpec] {
                SPECS
            }
            fn build_executor(ctx: SessionCtx) -> Box<dyn Executor> {
                // HTTP entry must be built with an Http ctx, never Browser.
                let saw_http = matches!(ctx, SessionCtx::Http(_));
                Box::new(HExec { saw_http })
            }
        }

        let entry = SiteEntry::from_http_module::<M>(FakeSite);
        assert_eq!(HTTP_POOL_HOST, "http");
        assert_eq!(entry.transport(), SiteTransport::Http);
        let recorded: Vec<&str> = entry.specs.iter().map(|s| s.wire).collect();
        assert_eq!(recorded, vec!["fake.ping"]);
    }
}
