use anyhow::Result;
use dittolive_ditto::prelude::*;
#[cfg(feature = "preview-datastreams")]
use dittolive_ditto::preview::{datastreams::ConnectionError, peer_pubkey::PeerPubkey};
use rand::Rng;
use safer_ffi::bytes::Bytes;

#[cfg(feature = "preview-datastreams")]
pub(crate) fn pubkey(ditto: &Ditto) -> PeerPubkey {
    use std::str::FromStr;
    PeerPubkey::from_str(&ditto.presence().graph().local_peer.peer_key).unwrap()
}

#[derive(Debug, Clone, Copy, Default)]
pub(crate) enum Transport {
    Tcp,
    #[default]
    Udp,
    Websocket,
}

impl std::str::FromStr for Transport {
    type Err = ();

    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
        match s {
            "tcp" => Ok(Transport::Tcp),
            "udp" => Ok(Transport::Udp),
            "ws" | "wss" | "websocket" => Ok(Transport::Websocket),
            _ => Err(()),
        }
    }
}

impl Transport {
    pub fn scheme(&self) -> &'static str {
        match self {
            Transport::Tcp => "tcp",
            Transport::Udp => "udp",
            Transport::Websocket => "ws",
        }
    }
}

fn get_ditto(database_id: Option<DatabaseId>) -> Result<Ditto> {
    let database_id = database_id.unwrap_or_else(DatabaseId::generate);

    let ditto = Ditto::builder()
        .with_temp_dir()
        // If you need to benchmark against persistent storage, uncomment this and set a path
        // .with_root(Arc::new(PersistentRoot::new(format!(
        //     "/Volumes/Untitled/ditto/{}",
        //     rand::rng().gen::<u64>()
        // ))?))
        .with_identity(|r| identity::OfflinePlayground::new(r, database_id))?
        // Note we can't turn off logs completely as the API doesn't support it
        .with_minimum_log_level(LogLevel::Error)
        .build()?;

    ditto.set_license_from_env("DITTO_LICENSE")?;
    Ok(ditto)
}

#[allow(dead_code)]
pub(crate) async fn setup(n: usize, transport: Transport) -> Result<Vec<Ditto>> {
    // SAFETY: This is safe because we are in a single-threaded context here.
    unsafe { std::env::set_var("DITTO_DISABLE_ALWAYS_DEBUG_ON_DISK_LOGS", "1") };

    let mut output = vec![];

    let database_id = DatabaseId::generate();

    let mut tc = TransportConfig::default();
    tc.peer_to_peer.lan.enabled = false;
    tc.peer_to_peer.bluetooth_le.enabled = false;
    tc.listen.tcp.enabled = false;
    tc.listen.http.enabled = false;

    let interface_ip = match transport {
        Transport::Tcp => {
            tc.listen.tcp.enabled = true;
            &mut tc.listen.tcp.interface_ip
        }
        // UDP config is currently a mess and relies on the TCP listen config, this needs fixing
        Transport::Udp => {
            // SAFETY: This is safe because we are in a single-threaded context here.
            unsafe {
                std::env::set_var("DITTO_NETWORK_ENABLE_NGN", "true");
                std::env::set_var("DITTO_REPLICATION_OVER_NGN", "true");
            }

            // note: it inherits the interface_ip from tcp listen config
            // todo: fix this to have a completely separate config
            tc.listen.tcp.enabled = true;
            &mut tc.listen.tcp.interface_ip
        }
        Transport::Websocket => {
            tc.listen.http.enabled = true;
            tc.listen.http.websocket_sync = true;
            &mut tc.listen.http.interface_ip
        }
    };
    *interface_ip = "127.0.0.1".into();

    let port = rand::rng().random_range(10000..(65535 - n as u16));

    for i in 0..(n as u16) {
        let ditto = get_ditto(Some(database_id.clone()))?;

        let mut tc = tc.clone();

        match transport {
            Transport::Tcp => tc.listen.tcp.port = port + i,
            Transport::Udp => {
                use serde_json::json;
                ditto
                    .store()
                    .execute((
                        "ALTER SYSTEM SET transports_udp_port = :port",
                        json!({
                            "port": port + i
                        }),
                    ))
                    .await
                    .unwrap();
            }
            Transport::Websocket => tc.listen.http.port = port + i,
        }

        ditto.set_transport_config(tc.clone());
        ditto.sync().start()?;
        output.push(ditto);
    }
    tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
    for i in 0..(n as u16) {
        let ditto = &output[i as usize];

        if i != 0 {
            let remote_port = port + i - 1;

            use serde_json::json;
            ditto
                .store()
                .execute((
                    "ALTER SYSTEM SET transports_discovered_peers = :peers",
                    json!({
                        "peers": [{
                            "address": format!("{}://127.0.0.1:{remote_port}", transport.scheme()),
                            "type": "force"
                        }]
                    }),
                ))
                .await
                .unwrap();
        }
    }

    establish_full_reachability(&output).await;

    Ok(output)
}

#[cfg(feature = "preview-datastreams")]
async fn establish_full_reachability(output: &[Ditto]) {
    loop {
        if tokio::time::timeout(
            std::time::Duration::from_secs(15),
            establish_full_reachability_via_presence(output),
        )
        .await
        .is_ok()
        {
            // Let's wait for the router to populate the routing table. OSPF takes about 500ms
            // to do so because of timers and debouncing.
            tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
            return;
        } else {
            eprintln!("Presence-based reachability timed out, attempting via Data Streams...");

            return establish_full_reachability_via_datastreams(output).await;
        }
    }
}

#[cfg(not(feature = "preview-datastreams"))]
async fn establish_full_reachability(output: &[Ditto]) {
    establish_full_reachability_via_presence(output).await
}

#[cfg(feature = "preview-datastreams")]
async fn establish_full_reachability_via_datastreams(output: &[Ditto]) {
    let _acceptors = output
        .iter()
        .map(|ditto| {
            ditto
                .datastreams()
                .bind_topic("detect")
                .finish_with(|candidate| {
                    _ = candidate.open_write_only();
                })
        })
        .collect::<Vec<_>>();

    let mut full_mesh = false;
    let n = output.len();

    while !full_mesh {
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        full_mesh = true;
        eprintln!("Checking connectivity...");
        for i in 0..n {
            for j in 0..i {
                match output[i]
                    .datastreams()
                    .connect(pubkey(&output[j]), "detect")
                    .finish_async()
                    .await
                {
                    Err(e @ ConnectionError::InvalidTarget) => {
                        full_mesh = false;
                        eprintln!("{i} ❌ {j} yet: {e:?}");
                    }
                    other => {
                        eprintln!("{i} ✅ {j}: {other:?}");
                    }
                }
            }
        }
    }
}

async fn establish_full_reachability_via_presence(output: &[Ditto]) {
    eprintln!("Establishing full reachability via presence...");
    let n = output.len();
    for ditto in output {
        while ditto.presence().graph().remote_peers.len() != n - 1 {
            // wait for all peers to connect
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;

            eprintln!(
                "Waiting for peers to connect, currently have {}/{}",
                ditto.presence().graph().remote_peers.len(),
                n - 1
            );
        }
    }
}

/// Generate a random payload of `len` bytes. It's important to use random data
/// so that the compression middleware in the transports doesn't skew the results.
pub(crate) fn generate_payload(len: usize) -> Bytes<'static> {
    let mut rng = rand::rng();
    let mut payload = vec![0u8; len];
    rng.fill(&mut payload[..]);

    payload.into()
}
