//! 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, sync::Arc, time::Instant};

use bytesize::ByteSize;
use dittolive_ditto::{
    dql::QueryResult,
    fs::TempRoot,
    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();
}

/// Default offline app ID + embedded offline license. Runs P2P-only with a
/// self-contained dev fixture; override via `DITTO_APP_ID` / `DITTO_LICENSE`.
const CARSAPP_DATABASE_ID: &str = "d0e9a40e-9beb-4c90-bb38-769269cbb6fb";
const DEFAULT_LICENSE: &str = "o2d1c2VyX2lkcmludGVybmFsQGRpdHRvLmNvbWZleHBpcnl0MjAyOS0wOS0wNFQxNjowMDowMFppc2lnbmF0dXJleFhadm1IcFRPL3l2VE9VeURxd25zOUQ5QlFtcG54WHBCazAyTVRvOW5QeFlQSnFwNmFoKzNJa29PbzBSd1RFVFl5emxiYS9yd1JlU0ZGZE1FV0xFMU5nZz09";

fn main() -> Result<(), Box<dyn std::error::Error>> {
    dotenv().ok();

    let app_id = std::env::var("DITTO_APP_ID").unwrap_or_else(|_| CARSAPP_DATABASE_ID.to_string());

    let root = TempRoot::new();
    let ditto = Ditto::open_sync(
        DittoConfig::new(
            app_id,
            DittoConfigConnect::SmallPeersOnly { private_key: None },
        )
        .with_persistence_directory(root.root_path()),
    )?;
    let license = std::env::var("DITTO_LICENSE").unwrap_or_else(|_| DEFAULT_LICENSE.to_string());
    ditto.set_offline_only_license_token(&license)?;

    DittoLogger::set_minimum_log_level(LogLevel::Info);

    ditto.update_transport_config(|tc| {
        tc.listen.tcp.enabled = true;
        tc.listen.tcp.interface_ip = "0.0.0.0".to_string();
        tc.listen.tcp.port = 4040;
        tc.peer_to_peer.bluetooth_le.enabled = true;
        tc.peer_to_peer.lan.enabled = true;
    });
    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(())
}
