//! Lazy-launch / idle-teardown Chromium manager for the daemon.
//!
//! The first worker that calls [`BrowserManager::acquire`] triggers a real
//! Chromium launch (and the `on_launch` hook — used to re-inject the
//! PHPSESSID cookie on every fresh process). Each acquire bumps an active
//! counter; the returned [`BrowserLease`] decrements it on drop.
//!
//! When the active counter hits zero, a background reaper task waits
//! `idle_timeout`. If still zero on wake, it closes Chromium and clears the
//! cached handle. The next acquire re-launches.
//!
//! `idle_timeout = Duration::ZERO` disables the reaper — Chromium stays alive
//! until [`BrowserManager::shutdown`].

use std::ops::Deref;
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use anyhow::Context;
use chromiumoxide::browser::Browser;
use futures_util::future::BoxFuture;
use tokio::sync::{Mutex, Notify};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;

use crate::crawler::browser::{self, LaunchOptions};

/// Hook invoked on every fresh launch with the new browser. Typically used
/// to re-inject PHPSESSID + run the session probe. Errors abort the
/// `acquire` that triggered the launch — the next acquire will re-launch.
pub type OnLaunch =
    Arc<dyn Fn(Arc<Browser>) -> BoxFuture<'static, anyhow::Result<()>> + Send + Sync>;

/// Returns an `OnLaunch` that does nothing — useful when no session is
/// configured (e.g. CLI metadata-only runs).
pub fn noop_on_launch() -> OnLaunch {
    Arc::new(|_| Box::pin(async { Ok(()) }))
}

/// Decoupled active-lease tracker. Owns the atomic counter and the idle
/// notifier so the wiring is unit-testable without standing up a real
/// `BrowserManager` (which would require launching Chromium).
#[derive(Default)]
pub(crate) struct ActiveTracker {
    counter: AtomicUsize,
    idle_signal: Notify,
}

impl ActiveTracker {
    pub(crate) fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    pub(crate) fn acquire(self: &Arc<Self>) {
        self.counter.fetch_add(1, Ordering::AcqRel);
    }

    pub(crate) fn release(self: &Arc<Self>) {
        if self.counter.fetch_sub(1, Ordering::AcqRel) == 1 {
            self.idle_signal.notify_one();
        }
    }

    pub(crate) fn current(&self) -> usize {
        self.counter.load(Ordering::Acquire)
    }

    pub(crate) fn idle_signal(&self) -> &Notify {
        &self.idle_signal
    }
}

/// Lifecycle gate for a coordinated browser restart. `acquire()` parks
/// while not [`RestartPhase::Healthy`] so no new navigation starts mid-
/// restart; long-lived lease holders (the metadata pass) cooperate by
/// checking [`BrowserManager::is_restart_pending`] at safe boundaries.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum RestartPhase {
    /// Normal operation — acquires proceed.
    Healthy,
    /// Restart requested; new acquires park, waiting for in-flight leases
    /// to drain.
    Draining,
    /// Chromium is being closed + relaunched.
    Restarting,
}

const PHASE_HEALTHY: u8 = 0;
const PHASE_DRAINING: u8 = 1;
const PHASE_RESTARTING: u8 = 2;

pub struct BrowserManager {
    inner: Mutex<Inner>,
    active: Arc<ActiveTracker>,
    launch_opts: LaunchOptions,
    idle_timeout: Duration,
    on_launch: OnLaunch,
    /// Coarse lifecycle phase (one of the `PHASE_*` constants).
    phase: AtomicU8,
    /// Woken when the phase returns to `Healthy` so parked acquires resume.
    resume: Notify,
    /// Serialises coordinated restarts so concurrent requests collapse into
    /// a single relaunch.
    restart_lock: Mutex<()>,
    /// Result of the most recent relaunch, so a caller that coalesced into
    /// an in-progress restart reports that restart's real outcome instead
    /// of a blind success.
    last_restart_ok: AtomicBool,
}

struct Inner {
    handle: Option<browser::Handle>,
    shared: Option<Arc<Browser>>,
}

impl BrowserManager {
    pub fn new(
        launch_opts: LaunchOptions,
        idle_timeout: Duration,
        on_launch: OnLaunch,
    ) -> Arc<Self> {
        Arc::new(Self {
            inner: Mutex::new(Inner {
                handle: None,
                shared: None,
            }),
            active: ActiveTracker::new(),
            launch_opts,
            idle_timeout,
            on_launch,
            phase: AtomicU8::new(PHASE_HEALTHY),
            resume: Notify::new(),
            restart_lock: Mutex::new(()),
            last_restart_ok: AtomicBool::new(true),
        })
    }

    /// Current restart phase.
    pub fn phase(&self) -> RestartPhase {
        match self.phase.load(Ordering::Acquire) {
            PHASE_DRAINING => RestartPhase::Draining,
            PHASE_RESTARTING => RestartPhase::Restarting,
            _ => RestartPhase::Healthy,
        }
    }

    fn set_phase(&self, phase: RestartPhase) {
        let v = match phase {
            RestartPhase::Healthy => PHASE_HEALTHY,
            RestartPhase::Draining => PHASE_DRAINING,
            RestartPhase::Restarting => PHASE_RESTARTING,
        };
        self.phase.store(v, Ordering::Release);
    }

    /// Whether a coordinated restart is in progress. Long-lived lease
    /// holders poll this at safe boundaries and yield their lease so the
    /// drain can complete promptly.
    pub fn is_restart_pending(&self) -> bool {
        self.phase() != RestartPhase::Healthy
    }

    /// Launch Chromium into `guard`, running the `on_launch` hook before
    /// publishing the handle so a probe failure doesn't leave a half-
    /// initialised browser behind.
    async fn launch_into(&self, guard: &mut Inner) -> anyhow::Result<()> {
        let handle = browser::launch(self.launch_opts.clone())
            .await
            .context("BrowserManager: launch chromium")?;
        let shared = handle.shared();
        if let Err(e) = (self.on_launch)(Arc::clone(&shared)).await {
            let _ = handle.close().await;
            return Err(e.context("BrowserManager: on_launch hook failed"));
        }
        guard.handle = Some(handle);
        guard.shared = Some(shared);
        Ok(())
    }

    /// Acquire a shared browser lease. The first acquire after a teardown
    /// launches a fresh Chromium (and runs `on_launch`); subsequent acquires
    /// while a process is alive just bump the counter and clone the `Arc`.
    pub async fn acquire(&self) -> anyhow::Result<BrowserLease> {
        // Park while a coordinated restart is draining/relaunching so no new
        // navigation starts against a browser that's about to be torn down.
        // The short sleep fallback guarantees liveness even if a `resume`
        // notification is missed (classic Notify lost-wakeup).
        while self.phase() != RestartPhase::Healthy {
            tokio::select! {
                _ = self.resume.notified() => {}
                _ = tokio::time::sleep(Duration::from_millis(100)) => {}
            }
        }
        let mut guard = self.inner.lock().await;
        if guard.handle.is_none() {
            self.launch_into(&mut guard).await?;
        }
        let browser = guard
            .shared
            .as_ref()
            .expect("shared set above")
            .clone();
        self.active.acquire();
        Ok(BrowserLease {
            browser,
            active: Arc::clone(&self.active),
        })
    }

    /// Coordinated restart: block new acquires, wait for in-flight leases
    /// to drain (up to `drain_deadline`, then force), close + relaunch
    /// Chromium (re-running `on_launch` → re-inject session + probe), then
    /// resume parked acquirers. Concurrent calls collapse into one
    /// relaunch. The phase is always returned to `Healthy` — even if the
    /// relaunch errors — so a failed restart never permanently wedges
    /// acquisition (the next acquire retries the launch lazily).
    pub async fn coordinated_restart(&self, drain_deadline: Duration) -> anyhow::Result<()> {
        // Dedup: if a restart is already running, wait for it and report
        // that restart's real outcome (not a blind success).
        let _restart_guard = match self.restart_lock.try_lock() {
            Ok(g) => g,
            Err(_) => {
                let _ = self.restart_lock.lock().await;
                return if self.last_restart_ok.load(Ordering::Acquire) {
                    Ok(())
                } else {
                    Err(anyhow::anyhow!("a concurrent coordinated browser restart failed"))
                };
            }
        };

        self.set_phase(RestartPhase::Draining);
        await_drain(&self.active, drain_deadline).await;

        self.set_phase(RestartPhase::Restarting);
        // Take the dead handle out under the lock, release the lock, THEN run
        // the (slow) Chromium teardown — so a worker that raced past the drain
        // can't block on `acquire()` behind one dead browser's close(). Mirrors
        // the idle reaper's take-drop-close ordering. Re-acquire to relaunch.
        let dead = {
            let mut guard = self.inner.lock().await;
            guard.shared = None;
            guard.handle.take()
        };
        if let Some(handle) = dead {
            let _ = handle.close().await;
        }
        let relaunch = {
            let mut guard = self.inner.lock().await;
            self.launch_into(&mut guard).await
        };

        self.last_restart_ok.store(relaunch.is_ok(), Ordering::Release);
        self.set_phase(RestartPhase::Healthy);
        self.resume.notify_waiters();
        match &relaunch {
            Ok(()) => tracing::info!("BrowserManager: coordinated restart complete"),
            Err(e) => tracing::error!(error = ?e, "BrowserManager: coordinated restart relaunch failed"),
        }
        relaunch.context("coordinated_restart: relaunch")
    }

    /// Forcefully close the cached browser regardless of active count.
    /// Used on daemon shutdown. After this returns the next acquire will
    /// re-launch from scratch.
    pub async fn shutdown(&self) {
        // Take-then-drop-then-close: don't hold the lock across Chromium
        // teardown (see `invalidate` / the idle reaper).
        let handle = {
            let mut guard = self.inner.lock().await;
            guard.shared = None;
            guard.handle.take()
        };
        if let Some(handle) = handle {
            let _ = handle.close().await;
        }
    }

    /// Mark the cached browser handle as unhealthy. The next `acquire`
    /// will re-launch Chromium from scratch.
    ///
    /// Same semantics as `shutdown` — the difference is intent:
    /// `shutdown` runs once at daemon teardown, while `invalidate` is a
    /// recovery hook callers fire after a CDP / connection / navigation
    /// failure that suggests the underlying process has died. Calling
    /// this while other workers still hold leases is safe — their
    /// outstanding CDP operations will return channel-closed errors
    /// and those workers will then re-acquire (re-launching Chromium).
    ///
    /// Idempotent: calling on an already-invalidated manager is a
    /// no-op.
    pub async fn invalidate(&self) {
        // Take the handle out under the lock, then release the lock BEFORE the
        // slow Chromium close() — otherwise every other worker's `acquire()`
        // serializes behind one dead browser's teardown. Matches the idle
        // reaper's ordering at the bottom of this file.
        let handle = {
            let mut guard = self.inner.lock().await;
            guard.shared = None;
            guard.handle.take()
        };
        if let Some(handle) = handle {
            let _ = handle.close().await;
            tracing::warn!("BrowserManager: handle invalidated — next acquire will relaunch");
        }
    }

    fn idle_timeout(&self) -> Duration {
        self.idle_timeout
    }

    fn active(&self) -> Arc<ActiveTracker> {
        Arc::clone(&self.active)
    }
}

/// Wait for the active-lease count to reach zero, up to `deadline`. Wakes
/// on the tracker's idle signal and re-checks on a short poll so a missed
/// signal can't strand the drain. Returns when drained or when the
/// deadline elapses (the caller then force-restarts). Extracted as a free
/// fn so the timing logic is unit-testable without launching Chromium.
async fn await_drain(active: &Arc<ActiveTracker>, deadline: Duration) {
    let start = tokio::time::Instant::now();
    while active.current() > 0 {
        let Some(remaining) = deadline.checked_sub(start.elapsed()) else {
            tracing::warn!(
                active = active.current(),
                "coordinated_restart: drain deadline exceeded — forcing relaunch"
            );
            return;
        };
        let nap = remaining.min(Duration::from_millis(250));
        tokio::select! {
            _ = active.idle_signal().notified() => {}
            _ = tokio::time::sleep(nap) => {}
        }
    }
}

/// Background reaper. Returns immediately when `idle_timeout == 0`.
/// Otherwise spawns a task that:
///   1. Waits on `idle_signal` (woken when active hits zero).
///   2. Sleeps `idle_timeout`.
///   3. Re-checks the counter under the mutex — if still zero, takes the
///      handle and closes it.
///
/// Repeats forever until `cancel` fires.
pub fn spawn_idle_reaper(mgr: Arc<BrowserManager>, cancel: CancellationToken) -> JoinHandle<()> {
    tokio::spawn(async move {
        if mgr.idle_timeout().is_zero() {
            // Block until cancellation, then exit.
            cancel.cancelled().await;
            return;
        }
        let active = mgr.active();
        loop {
            tokio::select! {
                _ = cancel.cancelled() => return,
                _ = active.idle_signal().notified() => {}
            }
            if active.current() > 0 {
                continue;
            }
            tokio::select! {
                _ = cancel.cancelled() => return,
                _ = tokio::time::sleep(mgr.idle_timeout()) => {}
            }
            let mut guard = mgr.inner.lock().await;
            if active.current() > 0 {
                // A worker grabbed a lease during the sleep — abort teardown.
                continue;
            }
            let handle = guard.handle.take();
            guard.shared = None;
            drop(guard);
            if let Some(h) = handle {
                let _ = h.close().await;
                tracing::info!("BrowserManager: idle teardown — Chromium closed");
            }
        }
    })
}

/// A worker-side handle that keeps the browser alive while in scope.
/// `Deref<Target = Browser>` so callers can pass `&*lease` to APIs that
/// expect `&Browser`.
pub struct BrowserLease {
    browser: Arc<Browser>,
    active: Arc<ActiveTracker>,
}

impl Deref for BrowserLease {
    type Target = Browser;
    fn deref(&self) -> &Browser {
        &self.browser
    }
}

impl Drop for BrowserLease {
    fn drop(&mut self) {
        self.active.release();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::AtomicBool;

    #[test]
    fn noop_on_launch_is_send_sync() {
        fn assert_send_sync<T: Send + Sync>(_: &T) {}
        let h = noop_on_launch();
        assert_send_sync(&h);
    }

    /// Invalidate is the only `BrowserManager` method that's safe to
    /// exercise in a unit test without launching Chromium — it's a
    /// no-op when no handle has been cached, and that path is exactly
    /// the one we want to verify is idempotent.
    #[tokio::test]
    async fn invalidate_is_a_noop_when_no_handle_cached() {
        let mgr = BrowserManager::new(
            crate::crawler::browser::LaunchOptions::default(),
            Duration::ZERO,
            noop_on_launch(),
        );
        // Two back-to-back invalidates must both complete; the second
        // would hang or panic if the first had left torn state.
        mgr.invalidate().await;
        mgr.invalidate().await;
    }

    #[tokio::test]
    async fn await_drain_returns_immediately_when_already_idle() {
        let active = ActiveTracker::new();
        let start = tokio::time::Instant::now();
        await_drain(&active, Duration::from_secs(5)).await;
        assert!(start.elapsed() < Duration::from_millis(200), "no wait when idle");
    }

    #[tokio::test]
    async fn await_drain_completes_when_lease_released() {
        let active = ActiveTracker::new();
        active.acquire();
        let bg = {
            let a = Arc::clone(&active);
            tokio::spawn(async move {
                tokio::time::sleep(Duration::from_millis(100)).await;
                a.release();
            })
        };
        // Generous deadline; should return shortly after the release, not
        // at the deadline.
        let start = tokio::time::Instant::now();
        await_drain(&active, Duration::from_secs(5)).await;
        assert!(start.elapsed() < Duration::from_secs(2), "drained on release");
        assert_eq!(active.current(), 0);
        bg.await.unwrap();
    }

    #[tokio::test]
    async fn await_drain_force_returns_after_deadline_when_stuck() {
        let active = ActiveTracker::new();
        active.acquire(); // never released
        let start = tokio::time::Instant::now();
        await_drain(&active, Duration::from_millis(300)).await;
        let elapsed = start.elapsed();
        assert!(elapsed >= Duration::from_millis(250), "waited ~deadline: {elapsed:?}");
        assert!(elapsed < Duration::from_secs(2), "but not forever: {elapsed:?}");
        assert_eq!(active.current(), 1, "still held — caller force-restarts");
    }

    #[test]
    fn phase_transitions_reflect_is_restart_pending() {
        let mgr = BrowserManager::new(
            crate::crawler::browser::LaunchOptions::default(),
            Duration::ZERO,
            noop_on_launch(),
        );
        assert_eq!(mgr.phase(), RestartPhase::Healthy);
        assert!(!mgr.is_restart_pending());
        mgr.set_phase(RestartPhase::Draining);
        assert!(mgr.is_restart_pending());
        mgr.set_phase(RestartPhase::Restarting);
        assert!(mgr.is_restart_pending());
        mgr.set_phase(RestartPhase::Healthy);
        assert!(!mgr.is_restart_pending());
    }

    #[tokio::test]
    async fn active_tracker_signals_idle_only_on_zero_transition() {
        let tracker = ActiveTracker::new();
        let signaled = Arc::new(AtomicBool::new(false));
        {
            let s = Arc::clone(&signaled);
            let t = Arc::clone(&tracker);
            tokio::spawn(async move {
                t.idle_signal().notified().await;
                s.store(true, Ordering::Release);
            });
        }

        tracker.acquire();
        tracker.acquire();
        assert_eq!(tracker.current(), 2);
        tracker.release();
        assert_eq!(tracker.current(), 1);
        tokio::time::sleep(Duration::from_millis(20)).await;
        assert!(!signaled.load(Ordering::Acquire), "no idle signal at count 1");
        tracker.release();
        tokio::time::sleep(Duration::from_millis(20)).await;
        assert_eq!(tracker.current(), 0);
        assert!(
            signaled.load(Ordering::Acquire),
            "idle signal fires on 1 -> 0 transition"
        );
    }
}
