//! Node-wide figures for /admin/stats that no other part keeps. Recorded where things
//! happen anyway; the one exception is the document's saved size (see
//! `measure_saved_size`).

use std::sync::{Mutex, MutexGuard};
use std::time::Duration;

use tokio::time::Instant;
use tracing::{debug, warn};

use crate::state::AppState;

/// Saving takes time proportional to the document; we spend at most about 1% of it on
/// measuring, and measure at least this long apart.
const SAVED_SIZE_MIN_INTERVAL: Duration = Duration::from_secs(60);
const SAVED_SIZE_DUTY: u32 = 100;

#[derive(Debug)]
pub(crate) struct NodeStats {
    started: Instant,
    inner: Mutex<Inner>,
}

#[derive(Debug, Default)]
struct Inner {
    caught_up: Option<CaughtUp>,
    catch_up_sources: u32,
    max_heads: usize,
    stale_hello_replies: u64,
    saved_size: Option<SavedSize>,
    measuring: bool,
}

/// The document's size as saved (compressed), when it was measured and how long that
/// took (copy and save).
#[derive(Debug, Clone, Copy)]
pub(crate) struct SavedSize {
    pub(crate) bytes: u64,
    pub(crate) at: Instant,
    pub(crate) took: Duration,
}

/// When the initial sync finished, and what it took.
#[derive(Debug, Clone, Copy)]
pub(crate) struct CaughtUp {
    /// Since the process started (handshakes included).
    pub(crate) after: Duration,
    /// Sync payload received until then, from all peers.
    pub(crate) sync_bytes: u64,
}

/// A copy of the figures.
#[derive(Debug, Clone, Copy)]
pub(crate) struct Snapshot {
    pub(crate) uptime: Duration,
    pub(crate) caught_up: Option<CaughtUp>,
    pub(crate) catch_up_sources: u32,
    pub(crate) max_heads: usize,
    pub(crate) stale_hello_replies: u64,
}

impl Default for NodeStats {
    fn default() -> Self {
        Self {
            started: Instant::now(),
            inner: Mutex::default(),
        }
    }
}

impl NodeStats {
    fn lock(&self) -> MutexGuard<'_, Inner> {
        self.inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    pub(crate) fn caught_up(&self, sync_bytes: u64) {
        let after = self.started.elapsed();
        self.lock()
            .caught_up
            .get_or_insert(CaughtUp { after, sync_bytes });
    }

    /// A catch-up source was chosen (the first, or a replacement).
    pub(crate) fn catch_up_source_chosen(&self) {
        self.lock().catch_up_sources += 1;
    }

    /// Our document has `count` heads (after applying a sync message).
    pub(crate) fn heads_seen(&self, count: usize) {
        let mut inner = self.lock();
        inner.max_heads = inner.max_heads.max(count);
    }

    pub(crate) fn stale_hello_reply(&self) {
        self.lock().stale_hello_replies += 1;
    }

    /// The last measurement of the saved size, if any.
    pub(crate) fn saved_size(&self) -> Option<SavedSize> {
        self.lock().saved_size
    }

    /// Whether a measurement is due; if so, it counts as started.
    fn start_measuring(&self, now: Instant) -> bool {
        let mut inner = self.lock();
        let due = !inner.measuring
            && inner.saved_size.is_none_or(|last| {
                now >= last.at + SAVED_SIZE_MIN_INTERVAL.max(last.took * SAVED_SIZE_DUTY)
            });
        if due {
            inner.measuring = true;
        }
        due
    }

    fn finish_measuring(&self, measured: Option<SavedSize>) {
        let mut inner = self.lock();
        inner.measuring = false;
        if let Some(measured) = measured {
            inner.saved_size = Some(measured);
        }
    }

    pub(crate) fn snapshot(&self) -> Snapshot {
        let inner = self.lock();
        Snapshot {
            uptime: self.started.elapsed(),
            caught_up: inner.caught_up,
            catch_up_sources: inner.catch_up_sources,
            max_heads: inner.max_heads,
            stale_hello_replies: inner.stale_hello_replies,
        }
    }
}

/// Measures the document's saved size in the background when due: the first time, then
/// at least a minute apart and spending no more than about 1% of the time on it (a
/// document that takes 2 s to copy and save is measured every 200 s). Called by the
/// peer worker every few seconds.
///
/// The copy holds the read lock (writes wait for it); the save runs without any lock.
pub(crate) fn measure_saved_size(state: &AppState) {
    if !state.stats.start_measuring(Instant::now()) {
        return;
    }
    let state = state.clone();
    tokio::spawn(async move {
        let start = Instant::now();
        let doc = std::sync::Arc::clone(&state.doc);
        // Both steps take time proportional to the document: off the async threads.
        let measured = match tokio::task::spawn_blocking(move || {
            let mut copy = doc.blocking_read().doc.clone();
            copy.save().len() as u64
        })
        .await
        {
            Ok(bytes) => {
                let took = start.elapsed();
                debug!(
                    bytes,
                    took_ms = took.as_millis() as u64,
                    "Measured the saved document size"
                );
                Some(SavedSize {
                    bytes,
                    at: Instant::now(),
                    took,
                })
            }
            Err(error) => {
                warn!(error = %error, "Measuring the saved document size failed");
                None
            }
        };
        state.stats.finish_measuring(measured);
    });
}