//! Distributed network: we join as a leaf child. The server sends parent
//! candidates via NetInfo; we connect to one (type "D"), complete the
//! branch handshake, and receive relayed `DistributedSearchRequest`
//! messages, which we answer from our shares.
//!
//! We do not accept children (leaf only) — that keeps the implementation
//! tractable while still making us a search-answering participant.

use std::net::SocketAddrV4;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;

use soulseek_proto::{dist, peer, server};

use crate::{conn, responder, Inner};

const CONNECT_TIMEOUT: Duration = Duration::from_secs(8);
const READ_TIMEOUT: Duration = Duration::from_secs(120);

/// Try each parent candidate until one connects; on success, run the
/// distributed read loop until it drops, then allow re-solicitation.
pub(crate) fn on_net_info(inner: Arc<Inner>, parents: Vec<(String, SocketAddrV4)>) {
    if inner.has_parent.load(Ordering::Relaxed) || parents.is_empty() {
        return;
    }
    tokio::spawn(async move {
        for (username, addr) in parents {
            if inner.has_parent.load(Ordering::Relaxed) {
                return;
            }
            match connect_parent(&inner, &username, addr).await {
                Ok(stream) => {
                    inner.has_parent.store(true, Ordering::Relaxed);
                    log::info!("distributed parent: {username} ({addr})");
                    let _ = run_parent(&inner, stream).await;
                    inner.has_parent.store(false, Ordering::Relaxed);
                    // Ask the server for fresh parents next time.
                    let _ = inner.send_server(server::have_no_parents(true));
                    return;
                }
                Err(e) => log::debug!("parent {username} ({addr}) failed: {e}"),
            }
        }
    });
}

async fn connect_parent(
    inner: &Arc<Inner>,
    _username: &str,
    addr: SocketAddrV4,
) -> std::io::Result<TcpStream> {
    let mut stream = tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect(addr))
        .await
        .map_err(|_| std::io::Error::other("connect timeout"))??;
    stream.set_nodelay(true).ok();

    // Identify as a distributed peer.
    let token = inner.next_token();
    let init = peer::peer_init(&inner.own_username(), peer::CONN_TYPE_DISTRIBUTED, token);
    stream.write_all(&init).await?;

    // Announce ourselves as a leaf: branch level 0, root = us.
    stream.write_all(&dist::branch_level(0)).await?;
    stream
        .write_all(&dist::branch_root(&inner.own_username()))
        .await?;
    Ok(stream)
}

async fn run_parent(inner: &Arc<Inner>, mut stream: TcpStream) -> std::io::Result<()> {
    loop {
        let frame = match tokio::time::timeout(READ_TIMEOUT, conn::read_frame(&mut stream)).await {
            Ok(f) => f?,
            Err(_) => {
                // Keepalive ping; if it fails the connection is dead.
                if stream.write_all(&dist::ping()).await.is_err() {
                    return Ok(());
                }
                continue;
            }
        };
        let Some(frame) = frame else { return Ok(()) };

        match dist::decode(frame) {
            Ok(dist::DistributedEvent::SearchRequest {
                username,
                token,
                query,
            }) => {
                responder::respond_to_search(inner.clone(), username, token, query);
            }
            Ok(dist::DistributedEvent::BranchLevel(level)) => {
                // Report our level (parent + 1) to the server.
                let _ = inner.send_server(server::branch_level(level + 1));
            }
            Ok(dist::DistributedEvent::BranchRoot(root)) => {
                let _ = inner.send_server(server::branch_root(&root));
            }
            Ok(_) => {}
            Err(e) => log::trace!("bad distributed message: {e}"),
        }
    }
}