//! M0: two in-process Arti clients, each hosting a v3 onion, echo ping.
//!
//! Needs a live Tor network. `cargo test --test tor_hs` skips it;
//! run with `--ignored` on a networked machine.

use std::sync::Arc;
use std::time::{Duration, Instant};

use arti_client::DormantMode;
use futures::StreamExt;
use futures::io::{AsyncReadExt, AsyncWriteExt};
use onionwire::hs::{self, Client, HS_PORT};
use tor_cell::relaycell::msg::Connected;
use tor_hsservice::{RunningOnionService, handle_rend_requests};

const PING: &[u8] = b"ping";
const BOOTSTRAP_LOG: &str = "info";

async fn bootstrapped(state_dir: &std::path::Path, cache_dir: &std::path::Path) -> Client {
    hs::bootstrapped(state_dir, cache_dir)
        .await
        .expect("Arti bootstrap failed — fail closed, no C-tor fallback")
}

fn spawn_echo(
    rend: impl futures::Stream<Item = tor_hsservice::RendRequest> + Send + 'static,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut requests = std::pin::pin!(handle_rend_requests(rend));
        while let Some(req) = requests.next().await {
            tokio::spawn(async move {
                let Ok(mut stream) = req.accept(Connected::new_empty()).await else {
                    return;
                };
                let mut buf = [0u8; 64];
                let Ok(n) = stream.read(&mut buf).await else {
                    return;
                };
                let _ = stream.write_all(&buf[..n]).await;
                let _ = stream.flush().await;
            });
        }
    })
}

async fn launch_echo(
    client: &Client,
    nickname: &str,
) -> (Arc<RunningOnionService>, tokio::task::JoinHandle<()>, String) {
    let launched = client
        .launch_onion_service(hs::hs_config(nickname).expect("hs_config"))
        .expect("launch_onion_service")
        .expect("onion service disabled in config — fail closed");
    let (svc, rend) = launched;
    let onion = hs::onion_string(&svc).expect("onion_string");
    let echo = spawn_echo(rend);
    (svc, echo, onion)
}

async fn echo_ping(client: &Client, onion: &str) -> Result<(), String> {
    let mut stream = client
        .connect((onion, HS_PORT))
        .await
        .map_err(|e| format!("connect {onion}:{HS_PORT}: {e}"))?;
    stream
        .write_all(PING)
        .await
        .map_err(|e| format!("write ping: {e}"))?;
    stream.flush().await.map_err(|e| format!("flush ping: {e}"))?;
    let mut buf = vec![0u8; PING.len()];
    stream
        .read_exact(&mut buf)
        .await
        .map_err(|e| format!("read echo: {e}"))?;
    if buf != PING {
        return Err(format!("echo payload {buf:?}"));
    }
    Ok(())
}

async fn echo_ping_retry(client: &Client, onion: &str, timeout: Duration) {
    let deadline = Instant::now() + timeout;
    let mut last = None::<String>;
    while Instant::now() < deadline {
        match tokio::time::timeout(Duration::from_secs(45), echo_ping(client, onion)).await {
            Ok(Ok(())) => return,
            Ok(Err(e)) => last = Some(e),
            Err(_) => last = Some("attempt timeout".to_string()),
        }
        tokio::time::sleep(Duration::from_secs(3)).await;
    }
    panic!(
        "echo ping to {onion} failed within {:?}: {:?}",
        timeout, last
    );
}

#[tokio::test(flavor = "multi_thread")]
#[ignore = "needs live Tor network"]
async fn two_node_byte_pipe_restart_and_dormant() {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| BOOTSTRAP_LOG.into()),
        )
        .with_test_writer()
        .try_init();

    let root = tempfile::tempdir().expect("tempdir");
    let alice_state = root.path().join("alice/state");
    let alice_cache = root.path().join("alice/cache");
    let bob_state = root.path().join("bob/state");
    let bob_cache = root.path().join("bob/cache");

    eprintln!("bootstrapping alice + bob TorClients…");
    let (alice, bob) = tokio::join!(
        bootstrapped(&alice_state, &alice_cache),
        bootstrapped(&bob_state, &bob_cache),
    );

    let (alice_svc, alice_echo, alice_onion) = launch_echo(&alice, "onionwire-a").await;
    let (bob_svc, _bob_echo, bob_onion) = launch_echo(&bob, "onionwire-b").await;
    eprintln!("alice onion={alice_onion}");
    eprintln!("bob onion={bob_onion}");
    assert_ne!(alice_onion, bob_onion, "separate HS identities");

    hs::wait_until_published(&alice, &alice_svc, &alice_onion)
        .await
        .expect("alice publish");
    hs::wait_until_published(&bob, &bob_svc, &bob_onion)
        .await
        .expect("bob publish");

    eprintln!("echo ping alice → bob");
    echo_ping_retry(&alice, &bob_onion, Duration::from_secs(180)).await;

    // Restart alice: same state dir, same nickname → same onion locator.
    // Abort the echo task and drop the client so Arti releases the state lock.
    alice_echo.abort();
    drop(alice_svc);
    drop(alice);
    tokio::time::sleep(Duration::from_secs(3)).await;
    eprintln!("restarting alice from persisted state…");
    let alice = bootstrapped(&alice_state, &alice_cache).await;
    let (alice_svc, _alice_echo, alice_onion_after) = launch_echo(&alice, "onionwire-a").await;
    assert_eq!(
        alice_onion, alice_onion_after,
        "onion locator must be stable across restart"
    );
    // Alice as client → Bob does not need Alice's HS republished; the locator
    // assert above is the restart invariant.
    echo_ping_retry(&alice, &bob_onion, Duration::from_secs(180)).await;

    // DormantMode::Soft then wake: still accepts within 120s.
    eprintln!("bob DormantMode::Soft");
    bob.set_dormant(DormantMode::Soft);
    tokio::time::sleep(Duration::from_secs(3)).await;
    eprintln!("bob DormantMode::Normal (wake)");
    bob.set_dormant(DormantMode::Normal);
    let t0 = Instant::now();
    echo_ping_retry(&alice, &bob_onion, Duration::from_secs(120)).await;
    let woke = t0.elapsed();
    eprintln!("post-dormant echo succeeded in {woke:?}");
    assert!(
        woke <= Duration::from_secs(120),
        "wake accept took {woke:?} (>120s)"
    );

    let _ = (alice_svc, bob_svc);
}