//! Feeds whether the swarm-queue client is connected into an OTEL gauge on a
//! timer. The client reconnects in the background and keeps running while
//! detached, so the gauge is the only place a detachment shows.

use std::time::Duration;

use async_nats::connection::State;

/// Half the exporter's one-minute interval, so every export carries a
/// value at most one tick old.
const TICK: Duration = Duration::from_secs(30);

/// Background loop: record the client's connection state every [`TICK`].
/// Detached task — runs for the harness's lifetime.
///
/// Records `0` until the first connect has produced a client, so an agent that
/// never reaches the queue reports disconnected instead of nothing.
pub async fn run() {
    crate::otel_turn_metrics::record_swarm_queue_connected(false);
    let client = crate::swarm_queue::client().await;
    watch(
        || client.connection_state(),
        crate::otel_turn_metrics::record_swarm_queue_connected,
        TICK,
    )
    .await;
}

/// Whether `state` counts as connected. `Pending` (never connected) and
/// `Disconnected` both do not.
fn is_connected(state: &State) -> bool {
    *state == State::Connected
}

/// Sample `state` every `tick` and hand `record` whether it is connected.
async fn watch(state: impl Fn() -> State, mut record: impl FnMut(bool), tick: Duration) {
    loop {
        record(is_connected(&state()));
        tokio::time::sleep(tick).await;
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};
    use std::time::Duration;

    use async_nats::connection::State;

    use super::{is_connected, watch};

    #[test]
    fn only_a_connected_client_counts_as_connected() {
        assert!(is_connected(&State::Connected));
        assert!(!is_connected(&State::Disconnected));
        assert!(!is_connected(&State::Pending));
    }

    #[tokio::test(start_paused = true)]
    async fn the_gauge_follows_the_connection_state_as_it_changes() {
        let state = Arc::new(Mutex::new(State::Connected));
        let seen = Arc::new(Mutex::new(Vec::new()));
        let task = tokio::spawn(watch(
            {
                let state = Arc::clone(&state);
                move || state.lock().unwrap().clone()
            },
            {
                let seen = Arc::clone(&seen);
                move |connected| seen.lock().unwrap().push(connected)
            },
            Duration::from_secs(10),
        ));
        tokio::time::sleep(Duration::from_secs(15)).await;
        *state.lock().unwrap() = State::Disconnected;
        tokio::time::sleep(Duration::from_secs(10)).await;
        *state.lock().unwrap() = State::Connected;
        tokio::time::sleep(Duration::from_secs(10)).await;
        task.abort();
        assert_eq!(*seen.lock().unwrap(), [true, true, false, true]);
    }
}