//! A basic example showing how to build a simple app using the
//! Rust DittoSDK.
//!
//! This exmaple will create a basic peer and register a single subscription
//! with a central provider (tcp+tls://hydra.dev.ditto.io:34468/).
//!
//! It can then be used to generate a stream of events for purposes of measuring
//! bounce latency

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

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

#[derive(Serialize, Deserialize, Debug)]
struct LatencyProbe {
    timestamp: i64,
    subsec_nanos: u32,
    #[serde(rename = "_id")]
    id: DocumentId,
}

const COLLECTION_NAME: &str = "latency_probes";
const PROBE_INTERVAL_MS: u64 = 1000;

#[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()?;
    ditto.sync().start()?;

    // This block can be used to set up a subscription and live query for local and
    // remote documents. The Documents and LiveQueryEvents processed by an
    // EventHandler callback, which can use a channel to push them back to the
    // main thread for printing.
    let store = ditto.store();
    let collection = store.collection(COLLECTION_NAME);
    let (tx, rx) = std::sync::mpsc::sync_channel(120);
    let event_handler = move |documents: Vec<ffi_sdk::BoxedDocument>, event| {
        // Remember these are generally LOCAL document updates
        {
            #![allow(deprecated)]
            trace!(
                ?event,
                modified_documents = %documents.len(),
                "latency sender received event",
            );
        }
        for doc in documents.into_iter() {
            let _ = tx.send(doc);
        }
    };

    // This loop inserts new docs into the LOCAL collection
    // It uses a background thread so the main thread is free to respond
    // LiveQueryEvents
    std::thread::spawn(move || loop {
        let store = ditto.store();
        let collection = store.collection(COLLECTION_NAME).unwrap();
        let now = Utc::now();
        let msg = LatencyProbe {
            timestamp: now.timestamp(),
            subsec_nanos: now.timestamp_subsec_nanos(),
            id: DocumentId::new(&now.to_rfc3339()).unwrap(),
        };
        let _doc_id = collection.upsert(msg);
        std::thread::sleep(std::time::Duration::from_millis(PROBE_INTERVAL_MS));
    });

    // this registers and starts the live query
    // this returns IMMEDIATELY and will run until the LiveQuery (Handle) is dropped
    // Note that Find ALL is not presently supported by the Subscription Server
    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);
    let collection = collection.unwrap();
    let pending_op = collection.find(&query);
    let _sub = pending_op.subscribe();
    let _lq = pending_op.observe_local(event_handler);

    for doc in rx.iter() {
        {
            #![allow(deprecated)]
            trace!(document = ?doc, "latency sender inserted document");
        }
    }
    Ok(())
}

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"),
        }
    }
}
