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

#![allow(clippy::disallowed_macros)]
#![allow(deprecated)] // TODO: Update querybuilder usage to DQL
extern crate safer_ffi;
extern crate serde;
extern crate serde_cbor;

use chrono::{DateTime, Duration, NaiveDateTime, Utc};
use dittolive_ditto::prelude::*;
use serde::{Deserialize, Serialize};
use serde_cbor::Value;
#[allow(deprecated)]
use tracing::{debug, error, info, trace};

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

const COLLECTION_NAME: &str = "latency_probes";

#[allow(deprecated)] // TODO: Update querybuilder usage to DQL
fn main() -> Result<(), Box<dyn std::error::Error>> {
    let app_id = AppId::from_env("BOUNCE_LATENCY_APP_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| app_id.default_auth_url());
            let token = std::env::var("AUTH_TOKEN").unwrap();
            OnlinePlayground::new(ditto_root, app_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| app_id.default_sync_url());
            config.connect.websocket_urls.insert(sync_url);
            config
        })?
        .build()?;
    {
        #![allow(deprecated)]
        debug!("Ditto receiver app started!");
    }

    let store = ditto.store();
    let collection = store.collection(COLLECTION_NAME);

    ditto.sync().start()?;

    let (tx, rx) = std::sync::mpsc::sync_channel(120);
    // This handler is called every time docs from local or remote sources are
    // committed to the local store which match the associated query.
    // `documents` is a vec of ALL documents matching the query after application of
    // the transaction.
    // `event` can be used to dissect out which of these are insertions.
    let event_handler = move |documents: Vec<ffi_sdk::BoxedDocument>, event| {
        {
            #![allow(deprecated)]
            trace!(
                ?event,
                updated_documents = %documents.len(),
                "latency receiver observed event",
            );
        }
        match event {
            LiveQueryEvent::Initial { .. } => {
                // On an initial sync, we can calculate the latency for arrival of the first
                // document
                if let Some(first_doc) = documents.first() {
                    let latency = calc_latency(first_doc);
                    {
                        #![allow(deprecated)]
                        info!(%latency, "initial sync observed");
                    }
                }
            }
            LiveQueryEvent::Update { insertions, .. } => {
                // We only want to send the newest event
                for idx in insertions.iter() {
                    if let Some(doc) = documents.get(*idx) {
                        let latency = calc_latency(doc);
                        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!("timestamp >= {}", start_time);
    // this registers and starts the live query
    let collection = collection.unwrap();
    let pending_op = collection.find(&query);
    let _lq = pending_op.observe_local(event_handler);
    let _sub = pending_op.subscribe();
    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();
        {
            #![allow(deprecated)]
            info!("latency is {} +/- {} ms", msl, stddev);
        }
    }

    Ok(())
}

/// Parse the Document and get the latency Duration
#[allow(deprecated)] // TODO: Update querybuilder usage to DQL
fn calc_latency(doc: &ffi_sdk::Document) -> Duration {
    let received = Utc::now();
    let mut seconds = 0;
    let mut nanos = 0;
    let ts_val = doc
        .get("timestamp")
        .expect("failed to get value for timestamp");
    if let Value::Integer(ts) = ts_val {
        seconds = ts;
    } else {
        {
            #![allow(deprecated)]
            error!("timestamp is not encoded as an U64");
        }
    }
    let ns_val = doc
        .get("subsec_nanos")
        .expect("Failed to get value for subsec_nanos");
    if let Value::Integer(ns) = ns_val {
        nanos = ns;
    } else {
        {
            #![allow(deprecated)]
            error!("subsec_nanos is not encoded as an U64");
        }
    }
    #[allow(deprecated)]
    let sent = DateTime::<Utc>::from_utc(
        NaiveDateTime::from_timestamp_opt(seconds as i64, nanos as u32).unwrap(),
        Utc,
    );
    let latency: Duration = received - sent;
    {
        #![allow(deprecated)]
        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"
///    }
///  }
/// }
/// ```
struct MyHandler {
    token: String,
    provider: String,
}

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

    fn authentication_expiring_soon(
        &self,
        _auth: dittolive_ditto::auth::DittoAuthenticator,
        seconds_remaining: std::time::Duration,
    ) {
        {
            #![allow(deprecated)]
            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"),
        }
    }
}
