//! # `chanora_core`
//!
//! Top-level Rust API and orchestration for the Chanora client.
//!
//! ## Role per SAD §7.2
//!
//! `chanora_core` is the integration point. It owns no protocol,
//! audio, or storage logic directly; instead it composes the
//! subsystem crates (`chanora_protocol`, `chanora_state`,
//! `chanora_audio`, `chanora_storage`, `chanora_diagnostics`)
//! behind a stable, typed API consumed by `chanora_bridge`.
//!
//! ## Invariants
//!
//! * Single active server connection at runtime (DEC-006 / SAD-064).
//! * Protocol-specific types from `tsclientlib` do not cross out of
//!   `chanora_protocol` (SAD-067).
//! * Secret material never lands in `chanora_storage`'s non-secret
//!   side (DEC-013.2 / SS-AUD-001/002).
//!
//! ## A.6 supervisor and reconnect
//!
//! `ChanoraSession` spawns a per-connection supervisor task on a
//! successful [`Self::connect`]. The supervisor:
//!
//! * awaits the `ProtocolClient`'s loss notifier;
//! * on user-driven disconnect, exits silently;
//! * on stream-end or error, re-dials with exponential backoff
//!   (1s, 2s, 5s, 15s, 30s, 60s capped); cancellation-safe via the
//!   per-session `cancel_tx` oneshot;
//! * re-attaches the audio engine to the fresh protocol client if
//!   audio was running prior to the loss.
//!
//! Lifecycle changes are broadcast to subscribers via
//! [`ChanoraSession::subscribe_events`].

#![forbid(unsafe_code)]
#![warn(missing_docs)]

use std::sync::Arc;
use std::time::Duration;

use thiserror::Error;
use tokio::sync::{broadcast, oneshot, watch, Mutex};
use tokio::task::JoinHandle;
use tracing::{info, warn};

pub mod ptt;

pub use chanora_audio::{
    AudioEngine, AudioEngineConfig, PttBackendDescriptor, PttCapabilityLevel,
};
pub use chanora_audio::{PttBinding, PttInputClass};
pub use chanora_diagnostics::{
    DiagnosticExport, InMemoryLogSink, KnownSecretRegistry, RedactingLogLayer, Redactor,
};
pub use chanora_protocol::{
    ChannelInfo, ClientInfo, ConnectConfig, DisconnectReason, ProtocolError, ServerSnapshot,
};
pub use chanora_storage::{Bookmark, BookmarkRepository, IdentityFileStore};

/// Errors that can arise during top-level orchestration.
#[derive(Debug, Error)]
pub enum CoreError {
    /// Protocol-layer error.
    #[error("protocol: {0}")]
    Protocol(#[from] chanora_protocol::ProtocolError),
    /// State-synchronisation error.
    #[error("state: {0}")]
    State(#[from] chanora_state::StateError),
    /// Audio-subsystem error.
    #[error("audio: {0}")]
    Audio(#[from] chanora_audio::AudioError),
    /// Storage error.
    #[error("storage: {0}")]
    Storage(#[from] chanora_storage::StorageError),
    /// Diagnostics error.
    #[error("diagnostics: {0}")]
    Diagnostics(#[from] chanora_diagnostics::DiagnosticsError),
    /// A precondition was violated (typically caller bug or
    /// concurrent misuse).
    #[error("invariant violated: {0}")]
    Invariant(&'static str),
    /// No active connection.
    #[error("not connected")]
    NotConnected,
    /// An attempt was made to start a second connection while one
    /// was already active (forbidden by DEC-006).
    #[error("already connected")]
    AlreadyConnected,
    /// Audio engine is not running.
    #[error("audio not started")]
    AudioNotStarted,
    /// PTT controller error.
    #[error("ptt: {0}")]
    Ptt(#[from] ptt::PttControllerError),
}

/// High-level lifecycle event surfaced to subscribers.
///
/// This is the minimal set needed for A.6 (reconnect banner). The
/// full event catalogue lands in A.4.
#[derive(Debug, Clone)]
pub enum SessionEvent {
    /// Initial connect succeeded, or reconnect attempt succeeded.
    Connected {
        /// Server name reported in the snapshot.
        server_name: String,
    },
    /// Connection lost; the supervisor will retry.
    Lost {
        /// Reason classification from the protocol layer.
        reason: String,
    },
    /// Supervisor is sleeping before its next reconnect attempt.
    Reconnecting {
        /// 1-based attempt counter for the current outage.
        attempt: u32,
        /// Seconds the supervisor will sleep before this attempt.
        delay_secs: u32,
    },
    /// Supervisor gave up after `attempt` failed retries (or the
    /// user explicitly disconnected mid-outage).
    Disconnected {
        /// Reason classification from the protocol layer.
        reason: String,
    },
    /// Audio engine started (e.g. after a successful reconnect with
    /// reattachment).
    AudioStarted,
    /// Audio engine stopped (e.g. before a reconnect cycle, or by
    /// explicit user action).
    AudioStopped,
    /// A snapshot probe observed a change in channel or client
    /// counts. Useful for UI auto-refresh without active polling
    /// from the Dart side. Carries the counts so subscribers can
    /// decide whether to re-fetch.
    SnapshotChanged {
        /// Number of channels in the latest probe snapshot.
        channels: u32,
        /// Number of clients in the latest probe snapshot.
        clients: u32,
    },
    /// Detected desktop Push-to-Talk capability (gen2 v0.9.3 /
    /// DEC-023..028). Published when the audio engine starts or
    /// when the active backend transitions (for example macOS
    /// permission state change). Carries only the privacy-safe
    /// descriptor — capability level, backend identifier, bound
    /// input class — per SRS-202 / DEC-027.
    PttCapability {
        /// Stable level name from `PttCapabilityLevel::as_str()`.
        level: String,
        /// Stable backend identifier (e.g. `"focused"`).
        backend_id: String,
        /// Coarse bound input class (e.g. `"keyboard"`); empty when
        /// no binding is active.
        bound_input_class: String,
    },
}

/// Coarse OS-reported network state. Populated by the Flutter side
/// via `connectivity_plus`; on platforms where no signal is wired
/// we stay at `Unknown` forever and the supervisor falls back to
/// pure watchdog/backoff behaviour.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NetworkState {
    /// No signal seen yet — treat as ambiguous; don't change behaviour.
    Unknown,
    /// OS reports at least one network with internet capability.
    Online,
    /// OS reports no networks available.
    Offline,
}

/// Channel capacity for the broadcast events. Generous because
/// reconnect cycles emit several events per attempt; if subscribers
/// fall behind we'd rather skip than block the supervisor.
const EVENT_CHANNEL_CAPACITY: usize = 64;

struct SupervisorInner {
    /// Optional cached AudioEngineConfig — set when start_audio is
    /// first called, used to re-create the engine after a reconnect.
    audio_cfg: Option<AudioEngineConfig>,
    /// True when audio is currently desired (start_audio called, no
    /// stop yet). Drives whether the supervisor reattaches audio on
    /// a successful reconnect.
    audio_running: bool,
}

struct ConnectedState {
    protocol: chanora_protocol::ProtocolClient,
    audio: Option<chanora_audio::AudioEngine>,
    /// Active PTT controller (SDD-088). Owns the platform input
    /// backend, the active binding, and the capability watch
    /// channel that drives `BridgeEvent::PttCapability`. Wired in
    /// `start_audio` after the engine is up; torn down by
    /// `stop_audio` / disconnect.
    ptt_controller: Option<Arc<ptt::PttController>>,
    /// Cancellation signal for the supervisor task. Dropped on
    /// explicit disconnect to break the supervisor out of its
    /// backoff sleep.
    cancel_tx: Option<oneshot::Sender<()>>,
    /// Supervisor task handle. Awaited on disconnect for clean
    /// teardown.
    supervisor: Option<JoinHandle<()>>,
    /// Connection config used to dial this connection; retained for
    /// future diagnostics. The supervisor task holds its own clone.
    #[allow(dead_code)]
    cfg: ConnectConfig,
    /// Audio supervision state. Wrapped in Arc<Mutex<_>> so the
    /// supervisor and the public API both see updates.
    sup_inner: Arc<Mutex<SupervisorInner>>,
}

/// The top-level Chanora session. Owns at most one active server
/// connection (DEC-006).
#[derive(Clone)]
pub struct ChanoraSession {
    inner: Arc<Mutex<Option<ConnectedState>>>,
    events_tx: broadcast::Sender<SessionEvent>,
    /// OS-reported network state. Updated by the bridge from
    /// `connectivity_plus` callbacks. Supervisor observes via
    /// [`watch::Receiver`].
    network_tx: watch::Sender<NetworkState>,
    /// Beta identity persistence. Optional — when `None`, the
    /// per-connect identity is whatever `ConnectConfig::identity`
    /// carries (or a fresh ephemeral one if that is also `None`).
    /// Wired by [`Self::init_storage`].
    identity_store: Arc<Mutex<Option<IdentityFileStore>>>,
    /// Bookmark store backed by SQLite (A.2 + External Beta
    /// extension). Lives alongside the identity file. Wired by
    /// [`Self::init_storage`].
    bookmark_store: Arc<Mutex<Option<BookmarkRepository>>>,
}

impl ChanoraSession {
    /// Construct an empty session. Performs no I/O.
    pub fn new() -> Self {
        let (events_tx, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
        let (network_tx, _) = watch::channel(NetworkState::Unknown);
        Self {
            inner: Arc::new(Mutex::new(None)),
            events_tx,
            network_tx,
            identity_store: Arc::new(Mutex::new(None)),
            bookmark_store: Arc::new(Mutex::new(None)),
        }
    }

    /// Wire a directory-backed identity store. Called by the bridge
    /// during `bridge_init` once Flutter has resolved the platform
    /// app-private storage directory. Subsequent [`Self::connect`]
    /// calls will reuse the stored identity, or generate-and-store
    /// one if none exists yet.
    ///
    /// Beta caveat: the file is *not* encrypted at rest — see
    /// `chanora_storage::IdentityFileStore` for the full gap notice.
    pub async fn init_storage(&self, dir: impl AsRef<std::path::Path>) -> Result<(), CoreError> {
        let dir = dir.as_ref();
        let store = IdentityFileStore::new(dir)?;
        info!(target: "chanora_core", path = ?store.path(), "identity store initialised");

        // Wire the bookmark repository under the same DEK so server
        // passwords are protected at rest (MVP hardening). On any
        // crypto setup failure we fall back to a plain bookmark
        // store — better to retain bookmark functionality than to
        // hard-fail.
        let bookmarks = match store.crypto() {
            Ok(c) => BookmarkRepository::with_crypto(dir, c)?,
            Err(e) => {
                warn!(
                    target: "chanora_core",
                    error = %e,
                    "could not derive DEK for bookmark store; falling back to plaintext"
                );
                BookmarkRepository::new(dir)?
            }
        };
        let encrypts = bookmarks.encrypts_passwords();
        *self.identity_store.lock().await = Some(store);
        *self.bookmark_store.lock().await = Some(bookmarks);
        info!(
            target: "chanora_core",
            encrypts_passwords = encrypts,
            "bookmark store initialised"
        );
        Ok(())
    }

    /// List persisted bookmarks. Returns an empty list if the store
    /// has not been wired or has no entries.
    pub async fn list_bookmarks(&self) -> Result<Vec<Bookmark>, CoreError> {
        let guard = self.bookmark_store.lock().await;
        match guard.as_ref() {
            Some(repo) => Ok(repo.list()?),
            None => Ok(Vec::new()),
        }
    }

    /// Insert a bookmark. Returns the assigned id.
    pub async fn add_bookmark(&self, b: Bookmark) -> Result<i64, CoreError> {
        let guard = self.bookmark_store.lock().await;
        let repo = guard
            .as_ref()
            .ok_or(CoreError::Invariant("bookmark store not initialised"))?;
        Ok(repo.add(&b)?)
    }

    /// Update an existing bookmark.
    pub async fn update_bookmark(&self, b: Bookmark) -> Result<(), CoreError> {
        let guard = self.bookmark_store.lock().await;
        let repo = guard
            .as_ref()
            .ok_or(CoreError::Invariant("bookmark store not initialised"))?;
        Ok(repo.update(&b)?)
    }

    /// Delete a bookmark by id.
    pub async fn delete_bookmark(&self, id: i64) -> Result<(), CoreError> {
        let guard = self.bookmark_store.lock().await;
        let repo = guard
            .as_ref()
            .ok_or(CoreError::Invariant("bookmark store not initialised"))?;
        Ok(repo.delete(id)?)
    }

    /// Push an OS connectivity update. Called by the bridge when
    /// `connectivity_plus` fires. Safe to call from any thread.
    pub fn set_network_state(&self, state: NetworkState) {
        // `send_if_modified` would suppress duplicate sends, but
        // `watch::Sender::send` already drops sends with no
        // subscribers gracefully. Using `send_replace` so the value
        // is updated even before any subscriber attaches.
        let _ = self.network_tx.send_replace(state);
    }

    /// Read the latest OS connectivity state. Useful for diagnostics.
    pub fn network_state(&self) -> NetworkState {
        *self.network_tx.borrow()
    }

    /// Subscribe to lifecycle events. The returned receiver fires
    /// on connect / lost / reconnecting / disconnected /
    /// audio-started / audio-stopped transitions. Multiple
    /// subscribers are allowed; each gets its own slow-consumer
    /// behaviour (events drop with [`broadcast::error::RecvError::Lagged`]).
    pub fn subscribe_events(&self) -> broadcast::Receiver<SessionEvent> {
        self.events_tx.subscribe()
    }

    /// Connect to a server. Fails with [`CoreError::AlreadyConnected`]
    /// if a connection is already active (DEC-006). Audio is not
    /// started automatically; call [`Self::start_audio`] after.
    ///
    /// Identity policy (A.2): if `cfg.identity` is `Some(_)`, use it
    /// verbatim. Otherwise, if [`Self::init_storage`] has been
    /// called and the store already holds an identity, load and
    /// reuse it. Otherwise generate a fresh identity and — if a
    /// store is wired — persist it before dialling so the *same*
    /// UID is presented on every subsequent connect from this
    /// install.
    pub async fn connect(&self, mut cfg: ConnectConfig) -> Result<ServerSnapshot, CoreError> {
        let mut guard = self.inner.lock().await;
        if guard.is_some() {
            return Err(CoreError::AlreadyConnected);
        }

        // Resolve identity from the store before dialling.
        if cfg.identity.is_none() {
            let store_guard = self.identity_store.lock().await;
            if let Some(store) = store_guard.as_ref() {
                match store.load()? {
                    Some(saved) => {
                        info!(target: "chanora_core", "reusing persisted identity");
                        cfg.identity = Some(saved);
                    }
                    None => {
                        let fresh = chanora_protocol::ProtocolClient::generate_identity();
                        store.save(&fresh)?;
                        info!(target: "chanora_core", "generated + persisted fresh identity");
                        cfg.identity = Some(fresh);
                    }
                }
            }
        }

        let client = chanora_protocol::ProtocolClient::connect(cfg.clone()).await?;
        let snap = client.snapshot().await?;

        // Set up the supervisor.
        let (cancel_tx, cancel_rx) = oneshot::channel::<()>();
        let lost_rx = client
            .take_loss_notifier()
            .ok_or(CoreError::Invariant("lost notifier already taken"))?;
        let probe = client.snapshot_probe();

        let sup_inner = Arc::new(Mutex::new(SupervisorInner {
            audio_cfg: None,
            audio_running: false,
        }));

        let supervisor = tokio::spawn(supervisor_loop(
            self.inner.clone(),
            self.events_tx.clone(),
            cfg.clone(),
            lost_rx,
            probe,
            cancel_rx,
            sup_inner.clone(),
            self.network_tx.subscribe(),
        ));

        let _ = self.events_tx.send(SessionEvent::Connected {
            server_name: snap.server_name.clone(),
        });

        *guard = Some(ConnectedState {
            protocol: client,
            audio: None,
            ptt_controller: None,
            cancel_tx: Some(cancel_tx),
            supervisor: Some(supervisor),
            cfg,
            sup_inner,
        });
        Ok(snap)
    }

    /// Return a fresh snapshot of the current server state.
    pub async fn snapshot(&self) -> Result<ServerSnapshot, CoreError> {
        let guard = self.inner.lock().await;
        let state = guard.as_ref().ok_or(CoreError::NotConnected)?;
        Ok(state.protocol.snapshot().await?)
    }

    /// True if a connection is currently active.
    pub async fn is_connected(&self) -> bool {
        self.inner.lock().await.is_some()
    }

    /// Start the audio engine attached to the current connection.
    /// Fails if not connected. Idempotent — calling twice replaces
    /// the engine. Stores the config so the supervisor can restart
    /// audio after a reconnect.
    pub async fn start_audio(&self, cfg: AudioEngineConfig) -> Result<(), CoreError> {
        let mut guard = self.inner.lock().await;
        let state = guard.as_mut().ok_or(CoreError::NotConnected)?;

        // Tear down any prior engine + controller. The controller
        // must be torn down before the engine because its forwarder
        // task references the gate that lives in the engine.
        if let Some(prev) = state.ptt_controller.take() {
            prev.stop().await;
        }
        if let Some(mut prev) = state.audio.take() {
            prev.stop();
        }

        let voice_out = state.protocol.voice_out();
        let voice_in = state
            .protocol
            .take_voice_in()
            .ok_or(CoreError::Invariant("voice_in already taken"))?;
        let engine = chanora_audio::AudioEngine::start(cfg.clone(), voice_out, voice_in)?;
        let gate = engine.transmit_gate().clone();
        state.audio = Some(engine);

        // Wire the PTT controller (SDD-088). It owns the platform
        // backend, the active binding, and the capability watch
        // channel. Constructed here because the controller needs
        // the engine's gate clone, and we want the engine to
        // exist before any backend tries to drive the gate.
        let controller = ptt::PttController::new(gate);
        state.ptt_controller = Some(controller.clone());

        // Record desired state so the supervisor will re-start audio
        // after a reconnect.
        {
            let mut sup = state.sup_inner.lock().await;
            sup.audio_cfg = Some(cfg);
            sup.audio_running = true;
        }
        let _ = self.events_tx.send(SessionEvent::AudioStarted);

        // Publish the current PTT capability so the UI badge can
        // render an honest value (SRS-196 / SDD-091). Backends
        // that resolve asynchronously (notably the Linux portal
        // backend after the user accepts the BindShortcuts
        // dialog) re-publish through the controller's
        // descriptor-watch.
        let initial_desc = controller.descriptor().await;
        let _ = self.events_tx.send(SessionEvent::PttCapability {
            level: initial_desc.level.as_str().to_string(),
            backend_id: initial_desc.backend_id.to_string(),
            bound_input_class: initial_desc.bound_input_class.unwrap_or("").to_string(),
        });
        let mut watch_rx = controller.descriptor_watch();
        let events_tx = self.events_tx.clone();
        tokio::spawn(async move {
            while watch_rx.changed().await.is_ok() {
                let d = watch_rx.borrow_and_update().clone();
                let _ = events_tx.send(SessionEvent::PttCapability {
                    level: d.level.as_str().to_string(),
                    backend_id: d.backend_id.to_string(),
                    bound_input_class: d.bound_input_class.unwrap_or("").to_string(),
                });
            }
        });
        Ok(())
    }

    /// Set push-to-talk state. No-op if audio not started.
    pub async fn set_ptt(&self, active: bool) -> Result<(), CoreError> {
        let guard = self.inner.lock().await;
        let state = guard.as_ref().ok_or(CoreError::NotConnected)?;
        let audio = state.audio.as_ref().ok_or(CoreError::AudioNotStarted)?;
        audio.set_ptt(active);
        Ok(())
    }

    /// Update the active PTT binding (gen2 v0.9.3 / DEC-026). The
    /// binding flows into the platform backend's `rebind` hook via
    /// the [`ptt::PttController`] (SDD-088) and the freshly-published
    /// `PttBackendDescriptor` is broadcast as
    /// `SessionEvent::PttCapability` so the UI badge updates
    /// immediately. The audio engine must be running.
    pub async fn set_ptt_binding(&self, binding: PttBinding) -> Result<(), CoreError> {
        let guard = self.inner.lock().await;
        let state = guard.as_ref().ok_or(CoreError::NotConnected)?;
        let controller = state
            .ptt_controller
            .as_ref()
            .ok_or(CoreError::AudioNotStarted)?;
        let desc = controller.set_binding(binding).await?;
        let _ = self.events_tx.send(SessionEvent::PttCapability {
            level: desc.level.as_str().to_string(),
            backend_id: desc.backend_id.to_string(),
            bound_input_class: desc.bound_input_class.unwrap_or("").to_string(),
        });
        Ok(())
    }

    /// Returns the active PTT capability descriptor as a triple
    /// `(level, backend_id, bound_input_class)`. Useful for the
    /// initial UI render before the first `PttCapability` event
    /// arrives. Returns the universal Focused descriptor when no
    /// audio engine is running.
    pub async fn ptt_descriptor(&self) -> (String, String, String) {
        let guard = self.inner.lock().await;
        let desc = match guard.as_ref().and_then(|s| s.ptt_controller.as_ref()) {
            Some(controller) => controller.descriptor().await,
            None => PttBackendDescriptor::focused(),
        };
        (
            desc.level.as_str().to_string(),
            desc.backend_id.to_string(),
            desc.bound_input_class.unwrap_or("").to_string(),
        )
    }

    /// Move our own client to `channel_id`. Optional channel
    /// `password` for password-protected channels (empty string
    /// counts as no password).
    pub async fn move_to_channel(
        &self,
        channel_id: u64,
        password: Option<String>,
    ) -> Result<(), CoreError> {
        let guard = self.inner.lock().await;
        let state = guard.as_ref().ok_or(CoreError::NotConnected)?;
        state.protocol.move_to_channel(channel_id, password).await?;
        Ok(())
    }

    /// Update self-mute state. `input` mutes the microphone, `output`
    /// mutes the local speaker for remote clients. Pass `None` to
    /// leave a field unchanged. Adjusting the local output mute also
    /// updates the audio engine's master output gain so playback
    /// silences immediately, independent of the server's broadcast.
    pub async fn set_self_muted(
        &self,
        input: Option<bool>,
        output: Option<bool>,
    ) -> Result<(), CoreError> {
        let guard = self.inner.lock().await;
        let state = guard.as_ref().ok_or(CoreError::NotConnected)?;
        state.protocol.set_muted(input, output).await?;
        if let Some(muted) = output {
            if let Some(audio) = state.audio.as_ref() {
                audio.set_output_muted(muted);
            }
        }
        Ok(())
    }

    /// Set master output gain (0.0 = silent, 1.0 = unity). Errors
    /// only when audio is not started.
    pub async fn set_output_gain(&self, gain: f32) -> Result<(), CoreError> {
        let guard = self.inner.lock().await;
        let state = guard.as_ref().ok_or(CoreError::NotConnected)?;
        let audio = state.audio.as_ref().ok_or(CoreError::AudioNotStarted)?;
        audio.set_output_gain(gain);
        Ok(())
    }

    /// Read audio engine statistics: (frames_sent, frames_received, ptt_active).
    pub async fn audio_stats(&self) -> Result<(u32, u32, bool), CoreError> {
        let guard = self.inner.lock().await;
        let state = guard.as_ref().ok_or(CoreError::NotConnected)?;
        let audio = state.audio.as_ref().ok_or(CoreError::AudioNotStarted)?;
        Ok((audio.frames_sent(), audio.frames_received(), audio.ptt()))
    }

    /// Disconnect from the server. No-op if not connected.
    pub async fn disconnect(&self) -> Result<(), CoreError> {
        let mut guard = self.inner.lock().await;
        if let Some(mut state) = guard.take() {
            // Signal the supervisor to exit (cancels any backoff sleep).
            if let Some(tx) = state.cancel_tx.take() {
                let _ = tx.send(());
            }
            if let Some(controller) = state.ptt_controller.take() {
                controller.stop().await;
            }
            if let Some(mut audio) = state.audio.take() {
                audio.stop();
                let _ = self.events_tx.send(SessionEvent::AudioStopped);
            }
            state.protocol.disconnect().await;
            // Wait for the supervisor to wind down so we don't race
            // a redial against the explicit disconnect.
            if let Some(handle) = state.supervisor.take() {
                let _ = handle.await;
            }
            let _ = self.events_tx.send(SessionEvent::Disconnected {
                reason: "user requested".to_string(),
            });
        }
        Ok(())
    }
}

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

/// Exponential backoff schedule for reconnect attempts (seconds).
/// After exhausting the schedule the supervisor stays at the last
/// entry. We cap at 60 s so a long outage still has a chance to
/// recover without consuming the device's battery polling tighter.
const BACKOFF_SCHEDULE: &[u32] = &[1, 2, 5, 15, 30, 60];

/// Watchdog probe interval. Every tick the supervisor issues a
/// `snapshot()` RPC against the live protocol client; if the server
/// has silently stopped responding (e.g. evicted us after a long
/// network outage) the probe fails. After
/// [`WATCHDOG_MAX_MISSES`] consecutive failures the supervisor
/// treats the connection as lost and triggers a redial. This is the
/// belt to the `lost_rx`-from-`connection_task` braces: tsclientlib
/// does not always surface a UDP idle-timeout as a stream error, so
/// without this watchdog the client can stay "ghost connected" —
/// UI shows Connected, server has long since removed us.
const WATCHDOG_INTERVAL: Duration = Duration::from_secs(5);
/// Per-probe timeout. Must be shorter than [`WATCHDOG_INTERVAL`] so
/// a stuck probe cannot wedge the supervisor.
const WATCHDOG_PROBE_TIMEOUT: Duration = Duration::from_secs(4);
/// Number of consecutive watchdog failures before the supervisor
/// declares the connection lost.
const WATCHDOG_MAX_MISSES: u32 = 3;

async fn supervisor_loop(
    state_arc: Arc<Mutex<Option<ConnectedState>>>,
    events_tx: broadcast::Sender<SessionEvent>,
    initial_cfg: ConnectConfig,
    initial_lost_rx: oneshot::Receiver<chanora_protocol::DisconnectReason>,
    initial_probe: chanora_protocol::SnapshotProbe,
    mut cancel_rx: oneshot::Receiver<()>,
    sup_inner: Arc<Mutex<SupervisorInner>>,
    mut network_rx: watch::Receiver<NetworkState>,
) {
    let mut lost_rx = initial_lost_rx;
    let mut probe = initial_probe;
    let mut cfg = initial_cfg;
    // Stable content hash of the last snapshot observed by the
    // watchdog. Used to emit `SessionEvent::SnapshotChanged`
    // whenever the channel or client list mutates in any way —
    // count, ordering, names, or per-client channel membership.
    // Stored as a u64 so the comparison is cheap and the field
    // doesn't grow with the snapshot.
    let mut last_signature: Option<u64> = None;

    loop {
        // Watch the current connection: race the protocol task's
        // own loss notifier against our app-level watchdog. Either
        // signal yields a `DisconnectReason` we then act on.
        let mut watchdog = tokio::time::interval(WATCHDOG_INTERVAL);
        // Skip the immediate tick so the first probe runs one
        // interval after connect, not instantly.
        watchdog.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        watchdog.tick().await; // consume the immediate tick
        let mut misses: u32 = 0;

        let reason: chanora_protocol::DisconnectReason = loop {
            tokio::select! {
                biased;
                _ = &mut cancel_rx => {
                    info!(target: "chanora_core", "supervisor cancelled by user");
                    return;
                }
                r = &mut lost_rx => {
                    match r {
                        Ok(reason) => break reason,
                        Err(_) => {
                            info!(
                                target: "chanora_core",
                                "supervisor: loss notifier dropped without firing"
                            );
                            return;
                        }
                    }
                }
                changed = network_rx.changed() => {
                    if changed.is_err() {
                        // The session was dropped; nothing more to do.
                        return;
                    }
                    let new_state = *network_rx.borrow_and_update();
                    if new_state == NetworkState::Offline {
                        // OS says we're offline. Don't fabricate a
                        // loss outright — captive portals and
                        // transient flicker can produce false
                        // negatives — but pre-charge the watchdog
                        // so the next probe failure trips it
                        // immediately. This shrinks UI latency from
                        // ~15s to ~5s in the common case.
                        misses = misses.saturating_add(1).min(WATCHDOG_MAX_MISSES - 1);
                        info!(
                            target: "chanora_core",
                            misses,
                            "connectivity: OS reports offline; pre-charging watchdog"
                        );
                    } else if new_state == NetworkState::Online {
                        // Reset on confirmed online so we don't
                        // carry stale Offline charges into the
                        // healthy state.
                        if misses > 0 {
                            info!(
                                target: "chanora_core",
                                "connectivity: OS reports online; clearing watchdog misses"
                            );
                        }
                        misses = 0;
                    }
                }
                _ = watchdog.tick() => {
                    match tokio::time::timeout(WATCHDOG_PROBE_TIMEOUT, probe.probe()).await {
                        Ok(Ok(snap)) => {
                            if misses > 0 {
                                info!(
                                    target: "chanora_core",
                                    misses,
                                    "watchdog: probe recovered"
                                );
                            }
                            misses = 0;
                            // A.4 / MVP: emit on any tree change.
                            // Hash channels (id, parent, order, name)
                            // and clients (id, channel, name) so
                            // in-channel moves and renames surface
                            // alongside count changes.
                            let sig = snapshot_signature(&snap);
                            if last_signature != Some(sig) {
                                last_signature = Some(sig);
                                let _ = events_tx.send(SessionEvent::SnapshotChanged {
                                    channels: snap.channels.len() as u32,
                                    clients: snap.clients.len() as u32,
                                });
                            }
                        }
                        Ok(Err(e)) => {
                            misses = misses.saturating_add(1);
                            warn!(
                                target: "chanora_core",
                                misses,
                                error = %e,
                                "watchdog: probe failed"
                            );
                        }
                        Err(_) => {
                            misses = misses.saturating_add(1);
                            warn!(
                                target: "chanora_core",
                                misses,
                                "watchdog: probe timed out"
                            );
                        }
                    }
                    if misses >= WATCHDOG_MAX_MISSES {
                        warn!(
                            target: "chanora_core",
                            misses,
                            "watchdog: declaring connection lost"
                        );
                        break chanora_protocol::DisconnectReason::Error(
                            "watchdog: server stopped responding".to_string(),
                        );
                    }
                }
            }
        };

        match reason {
            chanora_protocol::DisconnectReason::UserRequested => {
                info!(target: "chanora_core", "supervisor: user-requested disconnect; exiting");
                return;
            }
            chanora_protocol::DisconnectReason::StreamEnded
            | chanora_protocol::DisconnectReason::Error(_) => {
                let reason_str = format!("{reason:?}");
                warn!(target: "chanora_core", reason = %reason_str, "connection lost; will reconnect");
                let _ = events_tx.send(SessionEvent::Lost {
                    reason: reason_str.clone(),
                });

                // Stop the audio engine before reconnect — its
                // voice_out_tx points at the dead protocol client.
                // Also drop the dead protocol client itself so
                // tsclientlib closes its socket promptly; the
                // supervisor will install a new one on success.
                {
                    let mut guard = state_arc.lock().await;
                    if let Some(state) = guard.as_mut() {
                        if let Some(controller) = state.ptt_controller.take() {
                            controller.stop().await;
                        }
                        if let Some(mut audio) = state.audio.take() {
                            audio.stop();
                            let _ = events_tx.send(SessionEvent::AudioStopped);
                        }
                    }
                }

                // Reconnect loop.
                let mut attempt: u32 = 0;
                loop {
                    attempt = attempt.saturating_add(1);
                    let delay_secs = BACKOFF_SCHEDULE
                        .get(attempt as usize - 1)
                        .copied()
                        .unwrap_or(*BACKOFF_SCHEDULE.last().unwrap());

                    let _ = events_tx.send(SessionEvent::Reconnecting {
                        attempt,
                        delay_secs,
                    });
                    info!(
                        target: "chanora_core",
                        attempt,
                        delay_secs,
                        "reconnect: sleeping before next attempt"
                    );

                    // Sleep with cancellation support. An OS
                    // "online" notification short-circuits the
                    // sleep and resets the attempt counter so the
                    // next outage starts with the smallest backoff
                    // window again.
                    enum SleepOutcome { Elapsed, NetworkUp, Cancelled }
                    let outcome = tokio::select! {
                        biased;
                        _ = &mut cancel_rx => SleepOutcome::Cancelled,
                        changed = network_rx.changed() => {
                            if changed.is_err() {
                                SleepOutcome::Cancelled
                            } else if *network_rx.borrow_and_update() == NetworkState::Online {
                                SleepOutcome::NetworkUp
                            } else {
                                // Offline / Unknown transition — keep waiting the rest of the slot.
                                tokio::select! {
                                    _ = &mut cancel_rx => SleepOutcome::Cancelled,
                                    _ = tokio::time::sleep(Duration::from_secs(delay_secs as u64)) => SleepOutcome::Elapsed,
                                }
                            }
                        }
                        _ = tokio::time::sleep(Duration::from_secs(delay_secs as u64)) => SleepOutcome::Elapsed,
                    };
                    match outcome {
                        SleepOutcome::Cancelled => {
                            info!(target: "chanora_core", "supervisor cancelled during backoff");
                            return;
                        }
                        SleepOutcome::NetworkUp => {
                            info!(
                                target: "chanora_core",
                                attempt,
                                "connectivity: OS reports online; redialling immediately"
                            );
                            // Reset attempt counter so future losses
                            // start the backoff schedule fresh.
                            attempt = 0;
                        }
                        SleepOutcome::Elapsed => {}
                    }

                    info!(target: "chanora_core", attempt, "reconnect: dialling");
                    match chanora_protocol::ProtocolClient::connect(cfg.clone()).await {
                        Ok(new_client) => {
                            // Successful reconnect. Snapshot for the event.
                            let snap_name = match new_client.snapshot().await {
                                Ok(s) => s.server_name,
                                Err(_) => String::new(),
                            };
                            let new_lost_rx = match new_client.take_loss_notifier() {
                                Some(rx) => rx,
                                None => {
                                    warn!(
                                        target: "chanora_core",
                                        "reconnect: new client missing loss notifier"
                                    );
                                    return;
                                }
                            };
                            let new_probe = new_client.snapshot_probe();

                            // Reattach into the session state.
                            let restart_audio = {
                                let mut guard = state_arc.lock().await;
                                let state = match guard.as_mut() {
                                    Some(s) => s,
                                    None => {
                                        // Session was disposed mid-reconnect.
                                        return;
                                    }
                                };
                                // Replace the dead protocol client with the new one.
                                // The old client's background task either already
                                // exited (loss notifier fired) or will exit when
                                // its request channel drops (watchdog path).
                                let old = std::mem::replace(&mut state.protocol, new_client);
                                drop(old);

                                let sup = sup_inner.lock().await;
                                sup.audio_running && sup.audio_cfg.is_some()
                            };

                            let _ = events_tx.send(SessionEvent::Connected {
                                server_name: snap_name,
                            });

                            // Optionally restart audio.
                            if restart_audio {
                                let audio_cfg = {
                                    let sup = sup_inner.lock().await;
                                    sup.audio_cfg.clone().expect("audio_running ⇒ audio_cfg")
                                };
                                let mut guard = state_arc.lock().await;
                                if let Some(state) = guard.as_mut() {
                                    let voice_out = state.protocol.voice_out();
                                    if let Some(voice_in) = state.protocol.take_voice_in() {
                                        match chanora_audio::AudioEngine::start(
                                            audio_cfg, voice_out, voice_in,
                                        ) {
                                            Ok(engine) => {
                                                let gate = engine.transmit_gate().clone();
                                                state.audio = Some(engine);
                                                // Re-arm the PTT controller against
                                                // the new engine's gate (SDD-088).
                                                let controller = ptt::PttController::new(gate);
                                                state.ptt_controller = Some(controller.clone());
                                                let _ = events_tx
                                                    .send(SessionEvent::AudioStarted);
                                                // Re-publish the post-reconnect
                                                // capability (SRS-196 / SDD-091).
                                                let d = controller.descriptor().await;
                                                let _ = events_tx.send(
                                                    SessionEvent::PttCapability {
                                                        level: d.level.as_str().to_string(),
                                                        backend_id: d.backend_id.to_string(),
                                                        bound_input_class: d
                                                            .bound_input_class
                                                            .unwrap_or("")
                                                            .to_string(),
                                                    },
                                                );
                                            }
                                            Err(e) => {
                                                warn!(
                                                    target: "chanora_core",
                                                    error = %e,
                                                    "audio engine failed to restart after reconnect"
                                                );
                                            }
                                        }
                                    }
                                }
                            }

                            // Loop back to waiting for the next loss.
                            lost_rx = new_lost_rx;
                            probe = new_probe;
                            // Force re-emission of SnapshotChanged
                            // for the freshly reconnected session.
                            last_signature = None;
                            break;
                        }
                        Err(e) => {
                            warn!(
                                target: "chanora_core",
                                attempt,
                                error = %e,
                                "reconnect attempt failed"
                            );
                            let _ = e;
                            cfg = cfg.clone();
                        }
                    }
                }
            }
        }
    }
}

/// Build a stable 64-bit signature of `snap` covering everything
/// the UI would render. Two snapshots with identical channel
/// memberships, names, and orderings produce the same signature;
/// any in-channel move, rename, or reorder produces a different one.
fn snapshot_signature(snap: &ServerSnapshot) -> u64 {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};
    let mut h = DefaultHasher::new();
    // Server-level fields the UI shows.
    snap.server_name.hash(&mut h);
    // Channels — sorted by id so the hash is order-independent of
    // the input vector's iteration order.
    let mut channels: Vec<_> = snap.channels.iter().collect();
    channels.sort_by_key(|c| c.id.0);
    for c in &channels {
        c.id.0.hash(&mut h);
        c.parent.0.hash(&mut h);
        c.order.hash(&mut h);
        c.name.hash(&mut h);
    }
    // Clients — sorted by id.
    let mut clients: Vec<_> = snap.clients.iter().collect();
    clients.sort_by_key(|c| c.id.0);
    for c in &clients {
        c.id.0.hash(&mut h);
        c.channel.0.hash(&mut h);
        c.name.hash(&mut h);
    }
    h.finish()
}

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

    #[test]
    fn session_can_be_constructed() {
        let _ = ChanoraSession::new();
    }

    #[tokio::test]
    async fn disconnect_when_not_connected_is_noop() {
        let s = ChanoraSession::new();
        assert!(!s.is_connected().await);
        s.disconnect().await.unwrap();
    }

    #[test]
    fn signature_detects_in_channel_move() {
        use chanora_protocol::{ChannelInfo, ClientInfo};
        let a = ServerSnapshot {
            server_name: "s".into(),
            welcome_message: "".into(),
            platform: "".into(),
            version: "".into(),
            channels: vec![
                ChannelInfo {
                    id: chanora_protocol::ChannelId(1),
                    parent: chanora_protocol::ChannelId(0),
                    name: "a".into(),
                    order: 0,
                },
                ChannelInfo {
                    id: chanora_protocol::ChannelId(2),
                    parent: chanora_protocol::ChannelId(0),
                    name: "b".into(),
                    order: 1,
                },
            ],
            clients: vec![ClientInfo {
                id: chanora_protocol::ClientId(10),
                channel: chanora_protocol::ChannelId(1),
                name: "u".into(),
            }],
        };
        let mut b = a.clone();
        // User moves from channel 1 → 2. Counts unchanged.
        b.clients[0].channel = chanora_protocol::ChannelId(2);
        assert_ne!(
            super::snapshot_signature(&a),
            super::snapshot_signature(&b)
        );
    }

    #[test]
    fn signature_is_stable_under_input_reorder() {
        use chanora_protocol::{ChannelInfo, ClientInfo};
        let a = ServerSnapshot {
            server_name: "s".into(),
            welcome_message: "".into(),
            platform: "".into(),
            version: "".into(),
            channels: vec![
                ChannelInfo {
                    id: chanora_protocol::ChannelId(2),
                    parent: chanora_protocol::ChannelId(0),
                    name: "b".into(),
                    order: 1,
                },
                ChannelInfo {
                    id: chanora_protocol::ChannelId(1),
                    parent: chanora_protocol::ChannelId(0),
                    name: "a".into(),
                    order: 0,
                },
            ],
            clients: vec![],
        };
        let mut b = a.clone();
        b.channels.reverse();
        assert_eq!(
            super::snapshot_signature(&a),
            super::snapshot_signature(&b)
        );
    }

    #[tokio::test]
    async fn empty_address_is_rejected() {
        let s = ChanoraSession::new();
        let r = s.connect(ConnectConfig::default()).await;
        assert!(matches!(r, Err(CoreError::Protocol(ProtocolError::Invalid(_)))));
    }

    #[tokio::test]
    async fn ptt_without_audio_errors() {
        let s = ChanoraSession::new();
        let r = s.set_ptt(true).await;
        assert!(matches!(r, Err(CoreError::NotConnected)));
    }

    #[tokio::test]
    async fn subscribe_before_connect_works() {
        // Ensures the event broadcast doesn't need a prior connect
        // to be subscribable.
        let s = ChanoraSession::new();
        let _rx = s.subscribe_events();
        assert!(!s.is_connected().await);
    }
}
