modio-logger-db 0.11.1

// Author: D.S. Ljungmark <spider@skuggor.se>, Modio AB
// SPDX-License-Identifier: AGPL-3.0-or-later
use crate::types::Metric;
use async_std::sync::Mutex;
use std::sync::Arc;
use tracing::instrument;

type BufMetric = (Metric, TimeStatus);
type BufMetrics = Vec<BufMetric>;

#[derive(Debug, Clone, Copy)]
pub(crate) enum TimeStatus {
    TimeFail,
    None,
}
impl TimeStatus {
    pub const fn as_str(&self) -> &str {
        match self {
            Self::TimeFail => "TIMEFAIL",
            Self::None => "NONE",
        }
    }
    pub const fn from_bool(timefail: bool) -> Self {
        if timefail {
            Self::TimeFail
        } else {
            Self::None
        }
    }
}

#[derive(Clone)]
pub(crate) struct VecBuffer {
    data: Arc<Mutex<BufMetrics>>,
}
// To avoid printing out _all_ the values in a buffer, or the spammy Arc<Mutex> state, we implement
// a simplified Debug.
impl std::fmt::Debug for VecBuffer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut d = f.debug_struct("VecBuffer");
        match self.data.try_lock() {
            Some(inner) => {
                d.field("len", &inner.len());
            }
            _ => {
                d.field("len", &"<Locked>");
            }
        }
        d.finish()
    }
}

fn metric_cmp(a: &BufMetric, b: &BufMetric) -> std::cmp::Ordering {
    a.0.time.partial_cmp(&b.0.time).expect("NaN or Inf time")
}

impl VecBuffer {
    pub fn new() -> Self {
        let inner = Vec::with_capacity(100);
        let data = Arc::new(Mutex::new(inner));
        Self { data }
    }

    #[instrument(skip_all)]
    pub async fn add_metric(&self, metric: Metric, status: TimeStatus) {
        let pair = (metric, status);
        {
            let inner = &mut self.data.lock().await;
            inner.push(pair);
        }
    }

    #[instrument(skip_all)]
    pub async fn get_metric(&self, name: &str) -> Option<Metric> {
        let res = {
            let inner = &self.data.lock().await;
            inner
                .iter()
                .filter(|m| m.0.name == name) // Only matching keys
                .max_by(|a, b| metric_cmp(a, b)) // Order by timestamp
                .map(|m| m.0.clone()) // Copy the data out
        };
        res
    }

    #[instrument(skip_all)]
    pub fn count(&self) -> Option<usize> {
        // async_std::{Mutex,RwLock}::try_(lock|read|write)() returns an Option, bubble "None" out as if nothing
        // was found.
        let res = {
            let inner = &self.data.try_lock()?;
            inner.len()
        };
        Some(res)
    }

    // Oldest does not wait, if it is blocked we return None
    #[instrument(skip_all)]
    pub fn oldest(&self) -> Option<f64> {
        let res = {
            let inner = &self.data.try_lock()?;
            inner
                .iter()
                .map(|m| m.0.time)
                .reduce(f64::min)
                .map_or(0.0, |ts| ts)
        };
        Some(res)
    }

    #[instrument(skip_all)]
    pub async fn consume_metrics(&self) -> BufMetrics {
        {
            let inner = &mut self.data.lock().await;
            inner.drain(..).collect()
        }
    }

    #[instrument(skip_all)]
    pub async fn add_metrics(&self, mut data: BufMetrics) {
        data.sort_unstable_by(metric_cmp);
        {
            let inner = &mut self.data.lock().await;
            inner.append(&mut data);
            // We want the buffer to be sorted by time, with freshest at the end.
            inner.sort_unstable_by(metric_cmp);
        }
    }
}

#[must_use]
pub fn inixtime() -> i64 {
    use std::time::SystemTime;
    // If time wraps we have a bigger problem...
    #[allow(clippy::cast_possible_wrap)]
    SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map_or(0, |n| n.as_secs() as i64)
}

#[must_use]
pub fn unixtime() -> u64 {
    use std::time::SystemTime;
    SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map_or(0, |n| n.as_secs())
}

#[must_use]
pub fn fxtime() -> f64 {
    use std::time::SystemTime;
    // If time wraps we have a bigger problem than clock bein 0.0
    SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map_or(0.0, |n| n.as_secs_f64())
}

#[must_use]
/// round time away from too many digits of precision
pub fn fxtime_ms() -> f64 {
    use std::time::SystemTime;
    #[allow(clippy::cast_precision_loss)]
    SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map_or(0.0, |n| {
            (n.as_secs() as f64) + (f64::from(n.subsec_millis()) / 1_000.0)
        })
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::error::Error;
    use test_log::test;
    use timeout_macro::timeouttest;

    fn mtrc(name: &str, value: &str) -> Metric {
        Metric {
            name: name.to_string(),
            value: value.to_string(),
            time: fxtime(),
        }
    }

    #[test(timeouttest)]
    async fn test_insert_remove() -> Result<(), Box<dyn Error>> {
        let ds = VecBuffer::new();
        do_insert_remove(ds).await?;

        Ok(())
    }

    async fn do_insert_remove(ds: VecBuffer) -> Result<(), Box<dyn Error>> {
        let m = mtrc("test.case", "test.value");
        ds.add_metric(m, TimeStatus::None).await;
        let m = mtrc("test.case", "test.value");
        ds.add_metric(m, TimeStatus::TimeFail).await;
        // Should have two metrcis
        assert_eq!(2, ds.count().expect("Locked?"));

        // Drain should have two metrics too
        let drain = ds.consume_metrics().await;
        assert_eq!(drain.len(), 2);

        // Should have zero metrics
        assert_eq!(0, ds.count().expect("Locked?"));

        // Should be zero when draining too
        let drain = ds.consume_metrics().await;
        assert_eq!(drain.len(), 0);
        Ok(())
    }

    #[test(timeouttest)]
    async fn test_consume_append() -> Result<(), Box<dyn Error>> {
        let ds = VecBuffer::new();
        let m = mtrc("test.case", "test.value");
        ds.add_metric(m, TimeStatus::None).await;
        let m = mtrc("test.case", "test.value");
        ds.add_metric(m, TimeStatus::TimeFail).await;
        // Should have two metrcis
        assert_eq!(2, ds.count().expect("Locked?"));

        // Drain should have two metrics too
        let drain = ds.consume_metrics().await;
        assert_eq!(drain.len(), 2);

        // Should have zero metrics
        assert_eq!(0, ds.count().expect("Locked?"));

        // Re-add metrics
        ds.add_metrics(drain).await;

        // Should have two metrics again
        assert_eq!(2, ds.count().expect("Locked?"));
        Ok(())
    }

    #[test(timeouttest)]
    async fn test_keys() -> Result<(), Box<dyn Error>> {
        let ds = VecBuffer::new();
        do_test_keys(ds).await?;
        Ok(())
    }

    async fn do_test_keys(ds: VecBuffer) -> Result<(), Box<dyn Error>> {
        let m = Metric {
            name: "test.case.one".to_string(),
            value: "null".to_string(),
            time: 1234.0,
        };
        ds.add_metric(m, TimeStatus::None).await;
        let m = mtrc("test.case.two", "test.value");
        ds.add_metric(m, TimeStatus::TimeFail).await;

        // Should exist
        assert!(
            ds.get_metric("test.case.one").await.is_some(),
            "Test case one should exist"
        );
        assert!(
            ds.get_metric("test.case.two").await.is_some(),
            "Test case two should exist"
        );
        // Should not exist
        assert!(
            ds.get_metric("test.case.three").await.is_none(),
            "Three should not exist"
        );
        assert_eq!(
            1234.0,
            ds.oldest().expect("Locked?"),
            "1234 should be the oldest value"
        );
        Ok(())
    }
}