//! 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

extern crate safer_ffi;

use std::sync::Arc;

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

#[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;

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()?;
    ditto.sync().start()?;
    let ditto = Arc::new(ditto);

    // This block can be used to set up a subscription and live query for local and
    // remote documents using the new DQL API.
    let (tx, rx) = std::sync::mpsc::sync_channel(120);

    // 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({
        let ditto = ditto.clone();
        move || loop {
            let now = Utc::now();
            let msg = LatencyProbe {
                timestamp: now.timestamp(),
                subsec_nanos: now.timestamp_subsec_nanos(),
                id: DocumentId::new(&now.to_rfc3339()).unwrap(),
            };
            let _result = tokio::runtime::Handle::current()
                .block_on(ditto.store().execute((
                    "INSERT INTO latency_probes DOCUMENTS (:doc) ON ID CONFLICT DO UPDATE",
                    serde_json::json!({ "doc": msg }),
                )))
                .unwrap();
            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!(
        "SELECT * FROM {} WHERE timestamp >= {}",
        COLLECTION_NAME, start_time
    );

    let observer_handler = move |query_result: QueryResult| {
        // Process all documents in the query result
        #[allow(deprecated)] // Workaround for patched tracing
        {
            trace!(
                updated_documents = %query_result.item_count(),
                "latency sender received event",
            );
        }
        for item in query_result.iter() {
            let value = item.value();
            let _ = tx.send(value.clone());
        }
    };
    let _subscription = ditto.sync().register_subscription(&query).unwrap();
    let _observer = ditto
        .store()
        .register_observer(&query, observer_handler)
        .unwrap();

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

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