//! Bounce latency receiver
//!
//! Subscribes to documents inserted by the Bounce Latency Sender
//! and computes the total travel latency.

extern crate safer_ffi;
extern crate serde;
extern crate serde_cbor;

use chrono::{Duration, TimeZone, Utc};
use dittolive_ditto::{
    debug,
    dql::{QueryResult, QueryResultItem},
    error, info,
    prelude::*,
    trace,
};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
#[expect(dead_code)]
struct LatencyProbe {
    timestamp: i64,
    subsec_nanos: u32,
}

const COLLECTION_NAME: &str = "latency_probes";

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let database_id = DatabaseId::from_env("BOUNCE_LATENCY_DATABASE_ID")?; // MUST match that of the Receiver App
    let ditto = Ditto::builder()
        .with_temp_dir()
        .with_minimum_log_level(LogLevel::Debug)
        .with_identity(|ditto_root| {
            // Override this auth url outside of prod
            let auth_url =
                std::env::var("DITTO_AUTH_URL").unwrap_or_else(|_x| database_id.default_auth_url());
            let token = std::env::var("AUTH_TOKEN").unwrap();
            OnlinePlayground::new(
                ditto_root,
                database_id.to_owned(),
                token,
                true,
                Some(&auth_url),
            )
        })?
        .with_transport_config(|_identity| {
            let mut config = TransportConfig::new();
            let sync_url =
                std::env::var("DITTO_SYNC_URL").unwrap_or_else(|_x| database_id.default_sync_url());
            config.connect.websocket_urls.insert(sync_url);
            config
        })?
        .build()?;
    debug!("Ditto receiver app started!");

    ditto.sync().start()?;

    let (tx, rx) = std::sync::mpsc::sync_channel(120);

    // New DQL-based observer that processes QueryResult
    let observer_handler = move |query_result: QueryResult| {
        #[allow(deprecated)] // Workaround for patched tracing
        {
            trace!(
                updated_documents = %query_result.item_count(),
                "latency receiver observed event",
            );
        }

        // Process all documents in the query result
        for item in query_result.iter() {
            let latency = calc_latency(&item);
            let _ = tx.send(latency);
        }
    };
    let start_time = Utc::now().timestamp(); // The time of this session
                                             // this will filter for only events in the current test session
    let query: String = format!(
        "SELECT * FROM {} WHERE timestamp >= {}",
        COLLECTION_NAME, start_time
    );
    // this registers and starts the live query using new DQL API
    let _subscription = ditto.sync().register_subscription(&query).unwrap();
    let _observer = ditto
        .store()
        .register_observer(&query, observer_handler)
        .unwrap();
    let mut running_stats: Accumulator = Accumulator::default();
    for measurement in rx.iter() {
        // We round down to nearest millisecond precision
        // This can be adjusted if needed
        running_stats.push(measurement.num_milliseconds() as f64);
        let msl = running_stats.mean();
        let stddev = running_stats.std_dev();
        info!("latency is {} +/- {} ms", msl, stddev);
    }

    Ok(())
}

/// Parse the QueryResultItem and get the latency Duration
fn calc_latency(item: &QueryResultItem) -> Duration {
    let received = Utc::now();
    let value = item.value();

    let timestamp = value
        .get("timestamp")
        .and_then(|v| v.as_i64())
        .expect("failed to get timestamp value");

    let subsec_nanos = value
        .get("subsec_nanos")
        .and_then(|v| v.as_u64())
        .expect("failed to get subsec_nanos value") as u32;

    let sent = Utc.timestamp_opt(timestamp, subsec_nanos).unwrap();
    let latency: Duration = received - sent;
    info!(
        "message received at {}, sent at {}, with latency {}",
        &received, &sent, &latency
    );
    latency
}

/// A univariate statistical accumulator (rolling average)
struct Accumulator {
    /// total number of elements observed
    count: usize,
    moment_1: f64,
    moment_2: f64,
    moment_3: f64,
    moment_4: f64,
    min: f64,
    max: f64,
    // Median, quartiles, and Mode are more awkward to compute
    // without windowing or discretization
}

impl Default for Accumulator {
    fn default() -> Self {
        Accumulator {
            count: 0,
            moment_1: 0.0,
            moment_2: 0.0,
            moment_3: 0.0,
            moment_4: 0.0,
            min: 0.0,
            max: 0.0,
        }
    }
}

impl Accumulator {
    /// Push a new observation
    fn push(&mut self, x: f64) {
        let old_count = self.count as f64;
        self.count += 1;
        let n = self.count as f64; // current count as a float
        let delta = x - self.moment_1;
        let delta_n = delta / (self.count as f64);
        let delta_n2 = delta_n * delta_n;
        let term_1 = delta * delta_n * old_count;
        // update our mean
        self.moment_1 += delta_n;

        // Update our kurtosis
        self.moment_4 += term_1 * delta_n2 * (n * n - 3.0 * n + 3.0)
            + 6.0 * delta_n2 * self.moment_2
            - 4.0 * delta_n * self.moment_3;

        // update our skew
        self.moment_3 += term_1 * delta_n * (n - 2.0) - 3.0 * delta_n * self.moment_2;

        // update our variance
        self.moment_2 += term_1;

        if x > self.max {
            self.max = x;
        }
        if x < self.min {
            self.min = x;
        }
    }

    fn mean(&self) -> f64 {
        self.moment_1
    }

    fn variance(&self) -> f64 {
        self.moment_2 / ((self.count - 1) as f64)
    }

    fn std_dev(&self) -> f64 {
        self.variance().sqrt()
    }
}
/// Each App will need to implement a version of this depending on its
/// authentication system architecture
/// It is not required for the OnlinePlayground identity
///
/// This example assumes the following provider has been registered with Ditto
/// Cloud: ```json
/// {
///  "authenticationEnabled": true,
///  "providers": {
///    "test-provider": {
///      "type": "tokenWebhook",
///      "webhookUrl": "https://dummy-auth-webhook/validate_token"
///    }
///  }
/// }
/// ```
#[expect(dead_code)]
struct MyHandler {
    token: String,
    provider: String,
}

impl DittoAuthenticationEventHandler for MyHandler {
    fn authentication_required(&self, auth: dittolive_ditto::identity::DittoAuthenticator) {
        if let Err(error) = auth.login(&self.token, &self.provider) {
            error!(%error, "login failed");
        }
    }

    fn authentication_expiring_soon(
        &self,
        _auth: dittolive_ditto::identity::DittoAuthenticator,
        seconds_remaining: std::time::Duration,
    ) {
        info!(
            remaining = %humantime::Duration::from(seconds_remaining),
            "auth token expiring soon",
        );
    }
}

impl Default for MyHandler {
    fn default() -> Self {
        MyHandler {
            token: String::from("letmein"),
            provider: String::from("test-provider"),
        }
    }
}
