//! Example Ditto Cars App "Relay"
//!
//! Listens for incoming connections from the Ditto Car's App
//! on BLE, LAN/Multicast, and TCP transports
//! syncing data with connected peers.
//!
//! Example Invocation
//! ``` shell
//! cargo run -p dittolive-ditto --example carsapp
//! ```

extern crate dittolive_ditto;
extern crate safer_ffi;

use std::{fs::File, str::FromStr, sync::Arc, time::Instant};

use bytesize::ByteSize;
use dittolive_ditto::{
    dql::QueryResult,
    info,
    prelude::*,
    store::attachment::{
        DittoAttachmentFetchEvent, DittoAttachmentToken, DittoAttachmentTokenLike as _,
    },
    trace,
};
use dotenvy::dotenv;

// Fetch attachment - called from update and insertion events in main closure
// Arguments - the store, the attachment token
fn fetch_attachment(ditto: &Ditto, attachment_token: DittoAttachmentToken) {
    let start = Instant::now();
    ditto
        .store()
        .fetch_attachment(attachment_token, move |event| {
            if let DittoAttachmentFetchEvent::Completed { attachment: att } = event {
                let duration = start.elapsed();
                let file = File::open(att.path()).unwrap();
                let len = file.metadata().unwrap().len();
                let speed = (len / 1024) as f32 / duration.as_secs_f32();
                let speed_str = format!("{} KB/s", speed);
                info!(
                    elapsed_time = %humantime::Duration::from(duration),
                    speed = %speed_str,
                    size = %ByteSize::b(len),
                    "attachment received"
                );
            }
        })
        .unwrap();
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    dotenv().ok();
    let ditto = Ditto::builder()
        .with_temp_dir()
        .with_identity(|ditto_root| {
            // expected by other apps
            let database_id = DatabaseId::from_str("d0e9a40e-9beb-4c90-bb38-769269cbb6fb")?;
            // We don't want a fully random OfflinePlayground Identity as we
            // need to make sure we have a specific DatabaseId shared by all peers
            identity::OfflinePlayground::new(ditto_root, database_id)
        })?
        .with_minimum_log_level(LogLevel::Info)
        .with_transport_config(|_identity| -> TransportConfig {
            let mut transport_config = TransportConfig::new();
            transport_config.listen.tcp.enabled = true;
            transport_config.listen.tcp.interface_ip = "0.0.0.0".to_string();
            transport_config.listen.tcp.port = 4040;
            transport_config.peer_to_peer.bluetooth_le.enabled = true;
            transport_config.peer_to_peer.lan.enabled = true;
            transport_config
        })?
        .build()?;
    ditto.set_license_from_env("DITTO_LICENSE")?;
    ditto.sync().start()?;
    let ditto = Arc::new(ditto);

    info!("Ditto cars app started!");

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

    // Set up the subscription and observer to watch for car documents using DQL
    let query = "SELECT * FROM cars";
    let _subscription = ditto.sync().register_subscription(query).unwrap();
    let _observer = ditto
        .store()
        .register_observer(query, {
            let ditto = ditto.clone();
            move |query_result: QueryResult| {
                trace!(
                    updated_documents = %query_result.item_count(),
                    "live query observed event",
                );

                // Process all documents in the query result
                for item in query_result.iter() {
                    let value = item.value();

                    // Convert to CBOR for sending through channel
                    let car_cbor = value.clone();
                    let _ = tx.send(car_cbor);

                    // Check for attachments
                    let Some(att_token_value) = value.get("att") else {
                        continue;
                    };

                    let Some(att_token_map) = att_token_value.as_object() else {
                        continue;
                    };

                    let Ok(att_token) = att_token_map.parse_attachment_token() else {
                        continue;
                    };

                    fetch_attachment(&ditto, att_token);
                }
            }
        })
        .unwrap();

    // downgrade our logging output before running the query
    DittoLogger::set_minimum_log_level(LogLevel::Debug);

    // Process documents as they come through the channel
    for car_cbor in rx.iter() {
        let car_str = serde_json::to_string(&*car_cbor).unwrap();
        println!("{}", &car_str);
    }
    Ok(())
}
