#![allow(clippy::disallowed_macros)]
#![allow(deprecated)]

use std::{str::FromStr, sync::Arc};

use anyhow::Result;
use dittolive_ditto::{
    experimental::{bus::WriteOnly, peer_pubkey::PeerPubkey},
    prelude::*,
};

#[tokio::main]
async fn main() -> Result<()> {
    let app_id = AppId::from_str("bus_example_app").unwrap();

    let ditto = Ditto::builder()
        .with_temp_dir()
        .with_identity(|r| identity::OfflinePlayground::new(r, app_id))?
        .with_minimum_log_level(LogLevel::Debug)
        .with_transport_config(|_| {
            let mut tc = TransportConfig::new();
            tc.enable_all_peer_to_peer();
            tc
        })?
        .build()?;
    ditto.set_license_from_env("DITTO_LICENSE")?;

    ditto.sync().start()?;
    let ditto = Arc::new(ditto);

    let _acceptor = ditto
        .bus()
        .bind_topic("example")
        .on_receive_factory(tokio::sync::mpsc::unbounded_channel)
        .finish_with(|mut stream| {
            tokio::task::spawn(async move {
                let pk = stream.peer_pubkey();
                while let Some(payload) = stream.recv().await {
                    let msg = String::from_utf8_lossy(&payload);

                    match &*msg {
                        "Hello, world!" => {
                            println!("Received hello from peer: {}", pk);
                            stream.message("Hello back!").send();
                        }
                        _ => {
                            println!("Received message: {}", msg);
                        }
                    }
                }
            });
        });

    // wait for another peer to join and grab its PeerPubkey
    let remote_peer = loop {
        let peers = ditto.presence().graph().remote_peers;

        if let Some(peer) = peers.iter().next() {
            break PeerPubkey::from_str(&peer.peer_key_string).unwrap();
        }

        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    };

    let stream = ditto
        .bus()
        .connect(remote_peer.clone(), "example")
        .on_receive_factory(WriteOnly)
        .finish_async()
        .await
        .unwrap();
    loop {
        println!("Sending hello to: {}", &remote_peer);

        stream.message("Hello, world!").send();

        tokio::time::sleep(std::time::Duration::from_secs(10)).await;
    }
}
