use std::sync::{
    Arc,
    atomic::{AtomicU64, AtomicUsize, Ordering},
};

use kithara_events::{AbrMode, AbrReason, VariantIndex};
use kithara_platform::{
    sync::Mutex,
    time::{Duration, Instant},
};
use num_traits::ToPrimitive;

use super::{decision::AbrDecision, view::AbrView};

/// Per-peer ABR state owned by a peer and shared with the controller.
///
/// Variants live in the peer (single source of truth) and reach the
/// state's `decide()` via [`AbrView::variants`]. The state itself only
/// tracks runtime control: current index, mode, switch timing, locks,
/// pending boundary commit.
pub struct AbrState {
    current_variant: Arc<AtomicUsize>,
    last_switch_at_nanos: AtomicU64,
    max_bandwidth_bps: AtomicU64,
    lock_count: AtomicUsize,
    mode: AtomicUsize,
    reference_instant: Instant,
    /// Phase 2 boundary-commit slot. `Some` means a switch has been
    /// requested via [`request_target`](AbrState::request_target) and
    /// the scheduler has not yet observed it on a segment boundary.
    /// Replace-pending semantics: a fresh `request_target` overwrites
    /// any prior unobserved entry (latest-wins, matching the
    /// "switch-only-on-boundaries" contract from the two-cursor plan).
    pending: Mutex<Option<PendingApply>>,
}

/// Captured intent of a pending switch: the target variant index plus
/// the reason the requestor (controller, manual UI, scheduler) wants
/// recorded once the boundary commit lands.
#[derive(Clone, Copy, Debug)]
struct PendingApply {
    reason: AbrReason,
    target: VariantIndex,
}

/// `true` for `AbrReason` variants that originate from a throughput
/// estimate — the ones that go stale across a position jump and must
/// not survive a seek into the post-unlock boundary commit. Manual or
/// first-pick reasons stay valid through a seek and are preserved.
fn is_throughput_driven(reason: AbrReason) -> bool {
    matches!(
        reason,
        AbrReason::UpSwitch | AbrReason::DownSwitch | AbrReason::UrgentDownSwitch
    )
}

/// Rebuild the typed [`AbrDecision`] a pending boundary-commit publishes,
/// using the live `from` and the reason captured at `request_target`. The
/// only reasons that reach a pending slot are switch-class (a `Stay` never
/// requests a target), so a non-down, non-manual reason classifies as an
/// up-switch.
fn pending_decision(from: VariantIndex, to: VariantIndex, reason: AbrReason) -> AbrDecision {
    match reason {
        AbrReason::ManualOverride => AbrDecision::Manual { from, to },
        AbrReason::DownSwitch | AbrReason::UrgentDownSwitch => {
            AbrDecision::DownSwitch { from, to, reason }
        }
        _ => AbrDecision::UpSwitch { from, to, reason },
    }
}

impl AbrState {
    const NO_BANDWIDTH_CAP: u64 = 0;
    const NO_SWITCH: u64 = 0;

    /// Build an `AbrState` with the initial variant set from `mode`.
    #[must_use]
    pub fn new(mode: AbrMode) -> Self {
        let initial_variant = match mode {
            AbrMode::Auto(Some(idx)) | AbrMode::Manual(idx) => idx.get(),
            AbrMode::Auto(None) => 0,
        };
        Self {
            current_variant: Arc::new(AtomicUsize::new(initial_variant)),
            last_switch_at_nanos: AtomicU64::new(Self::NO_SWITCH),
            max_bandwidth_bps: AtomicU64::new(Self::NO_BANDWIDTH_CAP),
            lock_count: AtomicUsize::new(0),
            mode: AtomicUsize::new(mode.into()),
            reference_instant: Instant::now(),
            pending: Mutex::default(),
        }
    }

    /// Publish the switch: a [`AbrDecision::Stay`] is a no-op; otherwise
    /// atomically clear the pending slot **iff** it still references the
    /// same target as `decision`, then store `current_variant :=
    /// decision.target()` and record the switch timestamp.
    ///
    /// The "iff" rule preserves the replace-pending semantic: if an
    /// external `request_target` overwrote the slot with a different
    /// target between [`peek_pending_decision`](Self::peek_pending_decision)
    /// and this call, the new pending stays untouched and the next
    /// boundary commits it. The captured `decision` still publishes —
    /// the caller has already prepared `v_new` for that target.
    pub fn apply_decision(&self, decision: &AbrDecision, now: Instant) {
        if !decision.changed() {
            return;
        }
        let target = decision.target();
        let mut slot = self.pending.lock();
        if slot.as_ref().is_some_and(|p| p.target == target) {
            *slot = None;
        }
        drop(slot);
        self.current_variant.store(target.get(), Ordering::Release);
        self.record_switch(now);
    }

    pub(super) fn can_switch_now(&self, now: Instant, min_interval: Duration) -> bool {
        let nanos = self.last_switch_at_nanos.load(Ordering::Acquire);
        if nanos == Self::NO_SWITCH {
            return true;
        }
        let last = self.reference_instant + Duration::from_nanos(nanos);
        now.duration_since(last) >= min_interval
    }

    #[must_use]
    pub fn current_variant_index(&self) -> VariantIndex {
        VariantIndex::new(self.current_variant.load(Ordering::Acquire))
    }

    /// Produce a decision without mutating state.
    #[must_use]
    pub fn decide(&self, view: &AbrView<'_>, now: Instant) -> AbrDecision {
        super::decision::evaluate(self, view, now)
    }

    fn instant_to_nanos(&self, instant: Instant) -> u64 {
        let nanos = instant
            .saturating_duration_since(self.reference_instant)
            .as_nanos()
            .to_u64()
            .unwrap_or(u64::MAX);
        nanos.max(1)
    }

    /// Drop a throughput-driven pending boundary-commit. Called when a
    /// new seek epoch arrives (`HlsPeer::apply_seek_change`): a pending
    /// up-switch chosen against pre-seek throughput becomes stale once
    /// the reader jumps, and would otherwise commit on the first
    /// boundary cross after the seek lands — forcing a decoder recreate
    /// while the new-variant cache is still empty (the prod `app.log`
    /// `HangDetector` signature). After the seek, `decide()`
    /// re-evaluates against post-seek throughput before any new auto
    /// switch is requested.
    ///
    /// `AbrReason::ManualOverride` and `AbrReason::Initial` are
    /// preserved — they encode user-driven or first-pick intent that
    /// is not invalidated by a position jump.
    ///
    /// Distinct from [`Self::lock`]: locking gates *publish* but
    /// preserves the pending intent so it can resume on unlock (the
    /// blender-fence contract codified in
    /// `peek_pending_decision_returns_none_during_seek`). Invalidating
    /// is destructive and happens only at semantic seek boundaries.
    pub fn invalidate_pending(&self) {
        let mut slot = self.pending.lock();
        if slot
            .as_ref()
            .is_some_and(|p| is_throughput_driven(p.reason))
        {
            *slot = None;
        }
    }

    #[must_use]
    pub fn is_locked(&self) -> bool {
        self.lock_count.load(Ordering::Acquire) > 0
    }

    /// Gate publication of pending decisions: while `lock_count > 0`,
    /// [`peek_pending_decision`](Self::peek_pending_decision) returns
    /// `None` so the boundary commit defers until unlock. The pending
    /// intent itself is preserved across lock/unlock — destructive
    /// invalidation of stale pre-seek intents is the job of
    /// [`invalidate_pending`](Self::invalidate_pending), called from
    /// `coord::reset_for_seek` on a semantic seek boundary.
    pub fn lock(&self) {
        self.lock_count.fetch_add(1, Ordering::AcqRel);
    }

    #[must_use]
    pub fn lock_count(&self) -> usize {
        self.lock_count.load(Ordering::Acquire)
    }

    #[must_use]
    pub fn max_bandwidth_bps(&self) -> Option<u64> {
        let v = self.max_bandwidth_bps.load(Ordering::Acquire);
        if v == Self::NO_BANDWIDTH_CAP {
            None
        } else {
            Some(v)
        }
    }

    #[must_use]
    pub fn mode(&self) -> AbrMode {
        AbrMode::from(self.mode.load(Ordering::Acquire))
    }

    /// Read-only peek at the pending decision. Returns the
    /// [`AbrDecision`] that [`apply_decision`](Self::apply_decision)
    /// would publish, or `None` when:
    /// - pending slot is empty;
    /// - state is locked (the seek-no-switch / blender invariant);
    /// - pending target equals `current` (no-op switch).
    ///
    /// Does not mutate. `current` is supplied by the caller to avoid a
    /// race with concurrent reads of [`current_variant_index`]; pass
    /// `self.current_variant_index()` if you do not need an externally
    /// pinned snapshot.
    #[must_use]
    pub fn peek_pending_decision(&self, current: VariantIndex) -> Option<AbrDecision> {
        if self.is_locked() {
            return None;
        }
        let pending = *self.pending.lock().as_ref()?;
        if pending.target == current {
            return None;
        }
        Some(pending_decision(current, pending.target, pending.reason))
    }

    /// Phase 2 read-only view of the unobserved pending switch (if any).
    /// Used by the Phase 3 scheduler boundary check and by tests.
    #[must_use]
    pub fn pending_target(&self) -> Option<VariantIndex> {
        self.pending.lock().as_ref().map(|p| p.target)
    }

    fn record_switch(&self, now: Instant) {
        self.last_switch_at_nanos
            .store(self.instant_to_nanos(now), Ordering::Release);
    }

    /// Phase 2 of the two-cursor refactor: record the intent to switch
    /// to `target` without committing the variant change. The boundary
    /// commit is driven by [`commit_pending`](Self::commit_pending) at
    /// segment boundaries (Phase 3 wires the scheduler to call it).
    ///
    /// Replace-pending semantics: a fresh `request_target` overwrites
    /// any prior unobserved entry. This honours the user-stated
    /// contract that a switch only commits at the next segment boundary
    /// — between two boundaries we keep the latest intent and discard
    /// stale ones.
    ///
    /// The pending intent is written regardless of `is_locked` — the
    /// publish gate lives in
    /// [`peek_pending_decision`](Self::peek_pending_decision), which
    /// returns `None` while locked so the boundary commit defers until
    /// post-unlock. Destructive drop of stale pre-seek intents is the
    /// job of [`invalidate_pending`](Self::invalidate_pending), called
    /// from `coord::reset_for_seek` on a semantic seek boundary.
    ///
    /// A manual pin admits exactly one pending target — its own index.
    /// A request for any other target under `Manual` mode is a decision
    /// derived from a superseded mode (a tick racing
    /// [`set_mode`](Self::set_mode)) and is refused: nothing cleans the
    /// slot later (`Stay` ticks never touch it), so letting the write
    /// through would queue a switch the live mode never decided. The
    /// mode is read under the slot lock, pairing with the
    /// store-mode-then-clear-slot order in `set_mode`.
    pub fn request_target(&self, target: VariantIndex, reason: AbrReason) {
        let mut slot = self.pending.lock();
        if let AbrMode::Manual(idx) = self.mode()
            && idx != target
        {
            return;
        }
        *slot = Some(PendingApply { reason, target });
    }

    pub fn set_max_bandwidth_bps(&self, cap: Option<u64>) {
        self.max_bandwidth_bps
            .store(cap.unwrap_or(Self::NO_BANDWIDTH_CAP), Ordering::Release);
    }

    /// Apply a new mode. Caller is responsible for validating that
    /// `Manual(idx)` references a known variant — variants live on the
    /// peer, not the state, so validation must happen against
    /// [`Abr::variants()`](crate::Abr::variants) at the call site.
    ///
    /// An explicit mode change supersedes any queued boundary switch:
    /// the pending slot is cleared, and the new mode's intent re-derives
    /// on the next tick (the controller ticks synchronously from
    /// `on_mode_changed`). The mode is stored before the slot clears so
    /// a [`request_target`](Self::request_target) that acquires the slot
    /// lock afterwards observes the new mode in its consistency check.
    pub fn set_mode(&self, mode: AbrMode) {
        self.mode.store(mode.into(), Ordering::Release);
        *self.pending.lock() = None;
    }

    pub fn unlock(&self) {
        let prev = self.lock_count.fetch_sub(1, Ordering::AcqRel);
        debug_assert!(prev > 0, "unlock called without matching lock");
    }
}
