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

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

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

use bytesize::ByteSize;
use dittolive_ditto::{
    prelude::*,
    store::{
        ditto_attachment_fetch_event::DittoAttachmentFetchEvent,
        ditto_attachment_token::DittoAttachmentToken,
    },
};
use dotenvy::dotenv;
#[allow(deprecated)]
use tracing::{info, trace};

// Fetch attachment - called from update and insertion events in main closure
// Arguments - the collection (collection), the attachment token
// (attachment_token).
fn fetch_attachment(store: &Store, attachment_token: DittoAttachmentToken) {
    let start = Instant::now();
    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);
                {
                    #![allow(deprecated)]
                    info!(
                        elapsed_time = %humantime::Duration::from(duration),
                        speed = %speed_str,
                        size = %ByteSize::b(len),
                        "attachment received"
                    );
                }
            }
        })
        .unwrap();
}

#[allow(deprecated)] // TODO: Update querybuilder usage to DQL
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 app_id = AppId::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 AppId shared by all peers
            identity::OfflinePlayground::new(ditto_root, app_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()?;

    {
        #![allow(deprecated)]
        info!("Ditto cars app started!");
    }

    let store = ditto.store().clone();
    let collection = store.collection("cars").unwrap();

    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(),
                "live query observed event",
            );
        }
        match event {
            LiveQueryEvent::Initial { .. } => {
                // On an initial sync, we can calculate the latency for arrival of the first
                // document
                {
                    #![allow(deprecated)]
                    info!("initial data loaded");
                }
            }
            LiveQueryEvent::Update {
                insertions,
                updates,
                ..
            } => {
                // We only want to send the newest event
                for idx in insertions.iter().chain(updates.iter()) {
                    if let Some(car) = documents.get(*idx) {
                        let car_cbor = car.to_cbor().unwrap();
                        let _ = tx.send(car_cbor);
                        if let Ok(att_token) = car.get::<DittoAttachmentToken>("att") {
                            fetch_attachment(&store, att_token);
                        }
                    }
                }
            }
        }
    };
    // downgrade our logging output before running the query
    DittoLogger::set_minimum_log_level(LogLevel::Debug);
    // Find and report on all cars
    let pending_op = collection.find_all();
    let _sub = pending_op.subscribe();
    let _lq = pending_op.observe_local(event_handler);
    for car_cbor in rx.iter() {
        let car_str = serde_json::to_string(&car_cbor).unwrap();
        println!("{}", &car_str);
    }
    Ok(())
}
