//! 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::config::TorClientConfigBuilder;
use arti_client::{DormantMode, TorClient, TorClientConfig};
use futures::StreamExt;
use futures::io::{AsyncReadExt, AsyncWriteExt};
use safelog::DisplayRedacted;
use tor_cell::relaycell::msg::Connected;
use tor_hsservice::status::State;
use tor_hsservice::{HsNickname, OnionServiceConfig, RunningOnionService, handle_rend_requests};
use tor_rtcompat::PreferredRuntime;

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

type Client = Arc<TorClient<PreferredRuntime>>;

fn client_config(state_dir: &std::path::Path, cache_dir: &std::path::Path) -> TorClientConfig {
    std::fs::create_dir_all(state_dir).expect("state dir");
    std::fs::create_dir_all(cache_dir).expect("cache dir");
    let mut builder = TorClientConfigBuilder::from_directories(state_dir, cache_dir);
    // Temp dirs sit under $TMP; skip fs-mistrust on the parent tree.
    builder.storage().permissions().dangerously_trust_everyone();
    builder.build().expect("TorClientConfig")
}

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

fn hs_config(nickname: &str) -> OnionServiceConfig {
    let nickname = HsNickname::new(nickname.to_string()).expect("HsNickname");
    OnionServiceConfig::builder()
        .nickname(nickname)
        .build()
        .expect("OnionServiceConfig")
}

fn onion_string(svc: &RunningOnionService) -> String {
    let id = svc
        .onion_address()
        .expect("onion identity missing from keystore");
    id.display_unredacted().to_string()
}

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_config(nickname))
        .expect("launch_onion_service")
        .expect("onion service disabled in config — fail closed");
    let (svc, rend) = launched;
    let onion = onion_string(&svc);
    let echo = spawn_echo(rend);
    (svc, echo, onion)
}

async fn wait_until_published(svc: &RunningOnionService, label: &str) {
    let deadline = Instant::now() + Duration::from_secs(180);
    let mut events = svc.status_events();
    loop {
        let st = svc.status();
        eprintln!("{label} hs status: {:?}", st.state());
        match st.state() {
            State::Running | State::DegradedReachable => return,
            State::Broken => {
                panic!("{label}: onion service broken: {:?}", st.current_problem())
            }
            _ => {}
        }
        if Instant::now() >= deadline {
            panic!(
                "{label}: onion service did not publish within 180s: {:?}",
                st.state()
            );
        }
        tokio::select! {
            _ = events.next() => {}
            _ = tokio::time::sleep(Duration::from_secs(2)) => {}
        }
    }
}

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");

    wait_until_published(&alice_svc, "alice").await;
    wait_until_published(&bob_svc, "bob").await;

    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);
}