use crate::session::*;
use libremetaverse_types::compat::CancellationToken;
use std::collections::VecDeque;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

#[derive(Clone)]
enum LoginPlan {
    Failure {
        after: Duration,
        failure: SessionFailure,
    },
    Success {
        after: Duration,
        signals: VecDeque<(Duration, SessionSignal)>,
    },
}

#[derive(Default)]
struct FakeStats {
    logins: AtomicUsize,
    logouts: AtomicUsize,
    flushes: AtomicUsize,
    active_sessions: AtomicUsize,
    active_callbacks: AtomicUsize,
    active_workers: AtomicUsize,
    active_sockets: AtomicUsize,
    max_sessions: AtomicUsize,
    max_callbacks: AtomicUsize,
    max_workers: AtomicUsize,
    max_sockets: AtomicUsize,
}

impl FakeStats {
    fn acquire(&self) {
        update_max(&self.active_sessions, &self.max_sessions, 1);
        update_max(&self.active_callbacks, &self.max_callbacks, 4);
        update_max(&self.active_workers, &self.max_workers, 2);
        update_max(&self.active_sockets, &self.max_sockets, 1);
    }

    fn release(&self) {
        self.active_sessions.fetch_sub(1, Ordering::AcqRel);
        self.active_callbacks.fetch_sub(4, Ordering::AcqRel);
        self.active_workers.fetch_sub(2, Ordering::AcqRel);
        self.active_sockets.fetch_sub(1, Ordering::AcqRel);
    }
}

fn update_max(active: &AtomicUsize, maximum: &AtomicUsize, amount: usize) {
    let next = active.fetch_add(amount, Ordering::AcqRel) + amount;
    maximum.fetch_max(next, Ordering::AcqRel);
}

struct FakeBackend {
    plans: Mutex<VecDeque<LoginPlan>>,
    stats: Arc<FakeStats>,
}

impl FakeBackend {
    fn new(plans: impl IntoIterator<Item = LoginPlan>) -> (Arc<Self>, Arc<FakeStats>) {
        let stats = Arc::new(FakeStats::default());
        (
            Arc::new(Self {
                plans: Mutex::new(plans.into_iter().collect()),
                stats: Arc::clone(&stats),
            }),
            stats,
        )
    }
}

impl GridSessionBackend for FakeBackend {
    fn login(
        &self,
        generation: u64,
        cancellation: CancellationToken,
    ) -> SessionFuture<'_, Result<Box<dyn GridSession>, SessionFailure>> {
        self.stats.logins.fetch_add(1, Ordering::AcqRel);
        let plan = self
            .plans
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .pop_front()
            .unwrap_or(LoginPlan::Success {
                after: Duration::ZERO,
                signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]),
            });
        let stats = Arc::clone(&self.stats);
        Box::pin(async move {
            let after = match &plan {
                LoginPlan::Failure { after, .. } | LoginPlan::Success { after, .. } => *after,
            };
            tokio::select! {
                () = tokio::time::sleep(after) => {}
                () = cancellation.cancelled() => {
                    return Err(SessionFailure::new(SessionFailureKind::TransientTransport));
                }
            }
            match plan {
                LoginPlan::Failure { failure, .. } => Err(failure),
                LoginPlan::Success { signals, .. } => {
                    stats.acquire();
                    let session: Box<dyn GridSession> = Box::new(FakeSession {
                        generation,
                        signals,
                        stats,
                        released: false,
                    });
                    Ok(session)
                }
            }
        })
    }

    fn flush_audit(
        &self,
        _cancellation: CancellationToken,
    ) -> SessionFuture<'_, Result<(), SessionFailure>> {
        self.stats.flushes.fetch_add(1, Ordering::AcqRel);
        Box::pin(async { Ok(()) })
    }
}

struct FakeSession {
    generation: u64,
    signals: VecDeque<(Duration, SessionSignal)>,
    stats: Arc<FakeStats>,
    released: bool,
}

impl FakeSession {
    fn release(&mut self) {
        if !self.released {
            self.released = true;
            self.stats.release();
        }
    }
}

impl GridSession for FakeSession {
    fn generation(&self) -> u64 {
        self.generation
    }

    fn next_signal(&mut self, cancellation: CancellationToken) -> SessionFuture<'_, SessionSignal> {
        let next = self.signals.pop_front();
        Box::pin(async move {
            if let Some((after, signal)) = next {
                tokio::select! {
                    () = tokio::time::sleep(after) => signal,
                    () = cancellation.cancelled() => SessionSignal::Disconnected(
                        SessionFailure::new(SessionFailureKind::TransientTransport)
                    ),
                }
            } else {
                cancellation.cancelled().await;
                SessionSignal::Disconnected(SessionFailure::new(
                    SessionFailureKind::TransientTransport,
                ))
            }
        })
    }

    fn logout(
        mut self: Box<Self>,
        _cancellation: CancellationToken,
    ) -> SessionFuture<'static, Result<(), SessionFailure>> {
        self.stats.logouts.fetch_add(1, Ordering::AcqRel);
        self.release();
        Box::pin(async { Ok(()) })
    }
}

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

fn ready_then(after: Duration, failure: SessionFailureKind) -> LoginPlan {
    LoginPlan::Success {
        after: Duration::ZERO,
        signals: VecDeque::from([
            (Duration::ZERO, SessionSignal::Ready),
            (
                after,
                SessionSignal::Disconnected(SessionFailure::new(failure)),
            ),
        ]),
    }
}

fn test_policy() -> ReconnectPolicy {
    ReconnectPolicy {
        initial_delay: Duration::from_secs(1),
        maximum_delay: Duration::from_secs(8),
        readiness_timeout: Duration::from_secs(30),
        stable_reset_after: Duration::from_secs(10),
        shutdown_deadline: Duration::from_secs(2),
        jitter_basis_points: 0,
        instance_seed: 7,
        offline_work_capacity: 2,
    }
}

#[test]
fn retry_hints_are_minimums_and_all_delays_remain_capped() {
    let mut policy = test_policy();
    policy.jitter_basis_points = 5_000;
    for seed in 0..128 {
        policy.instance_seed = seed;
        let hinted = Duration::from_secs(6);
        let delay = policy.retry_delay(1, Some(hinted));
        assert!(delay >= hinted);
        assert!(delay <= policy.maximum_delay);
        assert!(policy.retry_delay(64, None) <= policy.maximum_delay);
    }
}

fn start(plans: impl IntoIterator<Item = LoginPlan>) -> (SessionSupervisorHandle, Arc<FakeStats>) {
    let (backend, stats) = FakeBackend::new(plans);
    let erased: Arc<dyn GridSessionBackend> = backend;
    let handle = SessionSupervisor::new(erased, test_policy(), 32, 256)
        .expect("supervisor")
        .start();
    (handle, stats)
}

async fn settle() {
    for _ in 0..20 {
        tokio::task::yield_now().await;
    }
}

async fn wait_state(handle: &SessionSupervisorHandle, expected: SessionState) {
    for _ in 0..100 {
        if handle.state() == expected {
            return;
        }
        tokio::task::yield_now().await;
    }
    panic!("expected {expected:?}, got {:?}", handle.status());
}

#[tokio::test(start_paused = true)]
async fn paused_time_classifies_failures_backoff_flapping_and_stable_reset() {
    let plans = [
        LoginPlan::Failure {
            after: Duration::ZERO,
            failure: SessionFailure::new(SessionFailureKind::TransientTransport),
        },
        LoginPlan::Failure {
            after: Duration::ZERO,
            failure: SessionFailure::with_retry_after(
                SessionFailureKind::Maintenance,
                Duration::from_secs(5),
            ),
        },
        ready_then(Duration::from_secs(1), SessionFailureKind::Kicked),
        ready_then(
            Duration::from_secs(11),
            SessionFailureKind::SimulatorDisconnected,
        ),
        LoginPlan::Success {
            after: Duration::ZERO,
            signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]),
        },
    ];
    let (mut handle, stats) = start(plans);
    wait_state(&handle, SessionState::Backoff).await;
    assert_eq!(handle.status().consecutive_failures, 1);
    tokio::time::advance(Duration::from_secs(1)).await;
    settle().await;
    wait_state(&handle, SessionState::Backoff).await;
    assert_eq!(handle.status().consecutive_failures, 2);
    tokio::time::advance(Duration::from_secs(5)).await;
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    let first_online = handle.status().generation;
    tokio::time::advance(Duration::from_secs(1)).await;
    settle().await;
    wait_state(&handle, SessionState::Backoff).await;
    assert!(!handle.accepts_result(first_online));
    tokio::time::advance(Duration::from_secs(4)).await;
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    tokio::time::advance(Duration::from_secs(11)).await;
    settle().await;
    wait_state(&handle, SessionState::Backoff).await;
    assert_eq!(handle.status().consecutive_failures, 1);
    tokio::time::advance(Duration::from_secs(1)).await;
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    handle.shutdown().await.expect("shutdown");
    assert_eq!(stats.flushes.load(Ordering::Acquire), 1);
}

#[tokio::test(start_paused = true)]
async fn invalid_credentials_and_configuration_block_until_operator_reconnect() {
    for failure in [
        SessionFailureKind::InvalidCredentials,
        SessionFailureKind::InvalidConfiguration,
    ] {
        let plans = [
            LoginPlan::Failure {
                after: Duration::ZERO,
                failure: SessionFailure::new(failure),
            },
            LoginPlan::Success {
                after: Duration::ZERO,
                signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]),
            },
        ];
        let (mut handle, stats) = start(plans);
        wait_state(&handle, SessionState::AuthenticationBlocked).await;
        tokio::time::advance(Duration::from_hours(1)).await;
        settle().await;
        assert_eq!(stats.logins.load(Ordering::Acquire), 1);
        handle
            .control(SessionControl::ForceReconnect)
            .await
            .expect("force reconnect");
        settle().await;
        wait_state(&handle, SessionState::Online).await;
        handle.shutdown().await.expect("shutdown");
    }
}

#[tokio::test(start_paused = true)]
async fn manual_pause_logout_reconnect_and_offline_work_are_predictable() {
    let (mut handle, stats) = start([LoginPlan::Success {
        after: Duration::ZERO,
        signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]),
    }]);
    wait_state(&handle, SessionState::Online).await;
    let generation = handle.status().generation;
    let token = handle
        .generation_token(generation)
        .expect("generation token");
    handle.control(SessionControl::Pause).await.expect("pause");
    settle().await;
    wait_state(&handle, SessionState::Paused).await;
    assert!(token.is_cancellation_requested());
    assert!(!handle.accepts_result(generation));
    assert_eq!(
        handle
            .submit(SessionWork::new("mutate", WorkKind::IdempotentMutation).expect("work"))
            .await
            .expect("decision"),
        WorkDisposition::RejectedOfflineMutation
    );
    assert_eq!(
        handle
            .submit(SessionWork::new("read", WorkKind::ReadOnly).expect("work"))
            .await
            .expect("decision"),
        WorkDisposition::Queued
    );
    handle
        .control(SessionControl::Resume)
        .await
        .expect("resume");
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    assert_ne!(handle.status().generation, generation);
    assert_eq!(
        handle
            .submit(SessionWork::new("read", WorkKind::ReadOnly).expect("work"))
            .await
            .expect("dedupe decision"),
        WorkDisposition::RejectedDuplicate
    );
    let before_force = handle.status().generation;
    handle
        .control(SessionControl::ForceReconnect)
        .await
        .expect("active reconnect");
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    assert_ne!(handle.status().generation, before_force);
    handle
        .control(SessionControl::Logout)
        .await
        .expect("logout");
    settle().await;
    wait_state(&handle, SessionState::Stopped).await;
    handle
        .control(SessionControl::ForceReconnect)
        .await
        .expect("reconnect");
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    handle.shutdown().await.expect("shutdown");
    assert_eq!(stats.active_sessions.load(Ordering::Acquire), 0);
}

#[tokio::test(start_paused = true)]
async fn transport_connection_is_distinct_from_full_agent_readiness() {
    let (mut handle, _) = start([LoginPlan::Success {
        after: Duration::ZERO,
        signals: VecDeque::from([(Duration::from_secs(10), SessionSignal::Ready)]),
    }]);
    wait_state(&handle, SessionState::Degraded).await;
    let degraded = handle.status();
    assert!(degraded.transport_connected);
    assert!(!degraded.agent_ready);
    assert!(!handle.accepts_result(degraded.generation));
    tokio::time::advance(Duration::from_secs(10)).await;
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    assert!(handle.status().agent_ready);
    handle.shutdown().await.expect("shutdown");
}

#[tokio::test(start_paused = true)]
async fn readiness_timeout_logs_out_and_retries_instead_of_staying_degraded() {
    let plans = [
        LoginPlan::Success {
            after: Duration::ZERO,
            signals: VecDeque::new(),
        },
        LoginPlan::Success {
            after: Duration::ZERO,
            signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]),
        },
    ];
    let (backend, stats) = FakeBackend::new(plans);
    let mut policy = test_policy();
    policy.readiness_timeout = Duration::from_secs(5);
    let erased: Arc<dyn GridSessionBackend> = backend;
    let mut handle = SessionSupervisor::new(erased, policy, 32, 256)
        .expect("supervisor")
        .start();

    wait_state(&handle, SessionState::Degraded).await;
    tokio::time::advance(Duration::from_secs(5)).await;
    settle().await;
    wait_state(&handle, SessionState::Backoff).await;
    assert_eq!(stats.logouts.load(Ordering::Acquire), 1);
    tokio::time::advance(Duration::from_secs(1)).await;
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    handle.shutdown().await.expect("shutdown");
    assert_eq!(stats.active_sessions.load(Ordering::Acquire), 0);
}

#[tokio::test(start_paused = true)]
async fn shutdown_covers_degraded_authentication_blocked_stopped_and_online_states() {
    let cases = [
        (
            LoginPlan::Success {
                after: Duration::ZERO,
                signals: VecDeque::from([(Duration::from_hours(1), SessionSignal::Ready)]),
            },
            SessionState::Degraded,
        ),
        (
            LoginPlan::Failure {
                after: Duration::ZERO,
                failure: SessionFailure::new(SessionFailureKind::InvalidCredentials),
            },
            SessionState::AuthenticationBlocked,
        ),
        (
            LoginPlan::Success {
                after: Duration::ZERO,
                signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]),
            },
            SessionState::Online,
        ),
    ];
    for (plan, expected) in cases {
        let (mut handle, stats) = start([plan]);
        wait_state(&handle, expected).await;
        if expected == SessionState::Online {
            handle
                .control(SessionControl::Logout)
                .await
                .expect("logout");
            settle().await;
            wait_state(&handle, SessionState::Stopped).await;
        }
        handle.shutdown().await.expect("shutdown");
        assert_eq!(stats.active_sessions.load(Ordering::Acquire), 0);
        assert_eq!(stats.flushes.load(Ordering::Acquire), 1);
    }
}

#[tokio::test(start_paused = true)]
async fn fake_grid_reconnects_keep_exactly_one_resource_set_and_no_stale_execution() {
    let plans = [
        ready_then(Duration::from_secs(1), SessionFailureKind::Maintenance),
        ready_then(Duration::from_secs(1), SessionFailureKind::Kicked),
        LoginPlan::Success {
            after: Duration::ZERO,
            signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]),
        },
    ];
    let (mut handle, stats) = start(plans);
    wait_state(&handle, SessionState::Online).await;
    let stale = handle.status().generation;
    tokio::time::advance(Duration::from_secs(1)).await;
    settle().await;
    tokio::time::advance(Duration::from_secs(1)).await;
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    tokio::time::advance(Duration::from_secs(1)).await;
    settle().await;
    tokio::time::advance(Duration::from_secs(2)).await;
    settle().await;
    wait_state(&handle, SessionState::Online).await;
    assert!(!handle.accepts_result(stale));
    assert!(handle.accepts_result(handle.status().generation));
    assert_eq!(stats.max_sessions.load(Ordering::Acquire), 1);
    assert_eq!(stats.max_callbacks.load(Ordering::Acquire), 4);
    assert_eq!(stats.max_workers.load(Ordering::Acquire), 2);
    assert_eq!(stats.max_sockets.load(Ordering::Acquire), 1);
    handle.shutdown().await.expect("shutdown");
    assert_eq!(stats.active_sessions.load(Ordering::Acquire), 0);
    assert_eq!(stats.active_callbacks.load(Ordering::Acquire), 0);
    assert_eq!(stats.active_workers.load(Ordering::Acquire), 0);
    assert_eq!(stats.active_sockets.load(Ordering::Acquire), 0);
    assert_eq!(stats.logouts.load(Ordering::Acquire), 3);
    assert_eq!(stats.flushes.load(Ordering::Acquire), 1);
}

#[tokio::test(start_paused = true)]
async fn shutdown_is_cancellation_driven_during_connect_backoff_paused_and_online() {
    let scenarios = [
        LoginPlan::Success {
            after: Duration::from_hours(1),
            signals: VecDeque::new(),
        },
        LoginPlan::Failure {
            after: Duration::ZERO,
            failure: SessionFailure::new(SessionFailureKind::TransientTransport),
        },
        LoginPlan::Success {
            after: Duration::ZERO,
            signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]),
        },
    ];
    for (index, plan) in scenarios.into_iter().enumerate() {
        let (mut handle, stats) = start([plan]);
        settle().await;
        if index == 2 {
            wait_state(&handle, SessionState::Online).await;
            handle.control(SessionControl::Pause).await.expect("pause");
            settle().await;
            wait_state(&handle, SessionState::Paused).await;
        }
        handle.shutdown().await.expect("bounded shutdown");
        assert_eq!(handle.state(), SessionState::Stopped);
        assert_eq!(stats.active_sessions.load(Ordering::Acquire), 0);
        assert_eq!(stats.flushes.load(Ordering::Acquire), 1);
    }
}