// RUNTIME verification of the money-path narrowing (two-client split).
//
// Builds the SAME two nostr-sdk clients that `run_service` now builds:
//   * MONEY client  -> grim::nym::NymWebSocketTransport (mixnet; scoped exit for
//                      relay.floonet.dev) -> kind-1059 gift-wraps ONLY.
//   * GENERAL client -> stock nostr-sdk transport (CLEARNET-direct) -> identity
//                      (0/10002/10050) + profile/DM-relay lookups.
//
// The transport routing under test (NymWebSocketTransport -> pool.exit_for ->
// streamexit) is 100% real grim code; only the per-op client assignment (the
// split itself) is mirrored here so we can control connection lifecycle and read
// the scratch exit's per-stream byte log cleanly.
//
// The pool is pointed at a SCRATCH scoped exit (env SCRATCH_EXIT). That exit is a
// plain floonet-mixexit whose stdout logs `stream closed (X B in, Y B out)` — our
// byte counter. Because the GENERAL client is clearnet, the scratch exit never
// even opens a stream for identity/lookup: that is the proof they are OFF the exit.
//
//   HOME=/tmp/e2e-home SCRATCH_EXIT="<addr>" cargo run --example nostr_split_measure

use std::time::Duration;

use nostr_sdk::{
	Client, EventBuilder, Filter, Keys, Kind, Metadata, RelayUrl, SubscriptionId, Tag, TagKind,
	Timestamp,
};

const RELAY: &str = "wss://relay.floonet.dev";

async fn connect(client: &Client, url: &str) {
	let _ = client.add_relay(url).await;
	client.connect().await;
	// Give the handshake time (clearnet ~instant, exit ~2-6s over the mixnet).
	for _ in 0..40 {
		if client
			.relays()
			.await
			.values()
			.any(|r| r.status() == nostr_sdk::RelayStatus::Connected)
		{
			return;
		}
		tokio::time::sleep(Duration::from_millis(500)).await;
	}
}

#[tokio::main]
async fn main() {
	let _ = rustls::crypto::ring::default_provider().install_default();
	let exit =
		std::env::var("SCRATCH_EXIT").expect("set SCRATCH_EXIT to the scratch exit nym addr");
	println!("[split] HOME={:?}", std::env::var("HOME").ok());
	println!("[split] scratch exit = {exit}");

	// Bring up the wallet's REAL nym stack (tunnel + scoped-exit streamexit client).
	println!("[split] warm_up(): starting nym…");
	grim::nym::warm_up();
	let mut ready = false;
	for _ in 0..360 {
		if grim::nym::is_ready() {
			ready = true;
			break;
		}
		tokio::time::sleep(Duration::from_millis(500)).await;
	}
	println!("[split] nym is_ready = {ready}");
	// Sanity: the pool we wrote must resolve relay.floonet.dev -> the scratch exit.
	match grim::nostr::pool::load().exit_for(RELAY) {
		Some(e) if e == exit => println!("[split] pool.exit_for({RELAY}) -> scratch exit OK"),
		other => println!("[split] WARN pool.exit_for -> {other:?} (expected scratch exit)"),
	}

	let keys = Keys::generate();
	let me = keys.public_key();
	println!("[split] ephemeral npub pubkey = {}", me.to_hex());

	// ================= PHASE 1: GENERAL (clearnet) =================
	// identity publish (kinds 10050 + 10002 + 0) + a profile lookup. None of this
	// must touch the scratch exit.
	println!("\n[split] ===== PHASE 1: general/clearnet (identity + lookup) =====");
	let general = Client::builder().signer(keys.clone()).build();
	connect(&general, RELAY).await;

	let dm_tags = vec![Tag::custom(TagKind::custom("relay"), [RELAY.to_string()])];
	let inbox = EventBuilder::new(Kind::InboxRelays, "").tags(dm_tags);
	let relay_list = EventBuilder::relay_list(RelayUrl::parse(RELAY).ok().map(|u| (u, None)));
	let meta = EventBuilder::metadata(&Metadata::new().name("split-e2e"));
	for (label, b) in [
		("kind10050", inbox),
		("kind10002", relay_list),
		("kind0", meta),
	] {
		match general.sign_event_builder(b).await {
			Ok(event) => match general.send_event_to([RELAY], &event).await {
				Ok(o) => println!("[split] general publish {label} -> id {}", o.val.to_hex()),
				Err(e) => println!("[split] general publish {label} FAILED: {e}"),
			},
			Err(e) => println!("[split] general sign {label} FAILED: {e}"),
		}
	}
	// A profile (kind-0) lookup of a random pubkey — the general-client read path.
	let stranger = Keys::generate().public_key();
	let f = Filter::new().kind(Kind::Metadata).author(stranger).limit(1);
	let _ = general
		.fetch_events_from([RELAY], f, Duration::from_secs(8))
		.await;
	println!("[split] general profile lookup done (clearnet)");
	general.disconnect().await;
	tokio::time::sleep(Duration::from_secs(2)).await;
	println!("[split] PHASE 1 complete — scratch exit should show ZERO streams so far.");

	// ================= PHASE 2: MONEY (scoped exit) =================
	println!("\n[split] ===== PHASE 2: money/exit (subscribe + catch-up + 1059) =====");
	let money = Client::builder()
		.signer(keys.clone())
		.websocket_transport(grim::nym::NymWebSocketTransport)
		.build();
	connect(&money, RELAY).await;
	let connected = money
		.relays()
		.await
		.values()
		.any(|r| r.status() == nostr_sdk::RelayStatus::Connected);
	println!("[split] money client connected over scratch exit = {connected}");

	// Gift-wrap inbox subscription + 3-day catch-up (the receive money path).
	let since = Timestamp::from_secs(Timestamp::now().as_u64().saturating_sub(3 * 86_400));
	let giftwrap = Filter::new().kind(Kind::GiftWrap).pubkey(me).since(since);
	let _ = money
		.subscribe_with_id_to(
			[RELAY],
			SubscriptionId::new("split-giftwrap"),
			giftwrap.clone(),
			None,
		)
		.await;
	let _ = money
		.fetch_events_from([RELAY], giftwrap, Duration::from_secs(20))
		.await;
	println!("[split] money subscribe + catch-up done (exit)");

	// The kind-1059 PUBLISH — a NIP-59 gift-wrap, exactly dispatch_dm's v2 path.
	// Synthetic ~30 KB slatepack so the 1059 is a clear, large stream on the exit.
	let receiver = Keys::generate().public_key();
	let slatepack = format!("BEGINSLATEPACK.{}.ENDSLATEPACK", "A".repeat(30_000));
	let content = format!("goblin:pay\n{slatepack}");
	let sent = money
		.send_private_msg_to([RELAY], receiver, content, Vec::<Tag>::new())
		.await;
	let wrap_id = match sent {
		Ok(o) => {
			println!(
				"[split] money 1059 publish -> gift-wrap id {}",
				o.val.to_hex()
			);
			Some(o.val)
		}
		Err(e) => {
			println!("[split] money 1059 publish FAILED: {e}");
			None
		}
	};

	// Confirm read-back over the SAME exit (dispatch_dm's silent-loss guard).
	let mut confirmed_via_exit = false;
	if let Some(id) = wrap_id {
		let cf = Filter::new().id(id).limit(1);
		if let Ok(evs) = money
			.fetch_events_from([RELAY], cf, Duration::from_secs(15))
			.await
		{
			confirmed_via_exit = !evs.is_empty();
		}
	}
	println!("[split] 1059 confirmed via exit read-back = {confirmed_via_exit}");
	money.disconnect().await; // forces the scratch exit to log `stream closed (X in, Y out)`
	tokio::time::sleep(Duration::from_secs(2)).await;

	// ================= PHASE 3: CLEARNET ORACLE =================
	// Independently confirm the 1059 actually LANDED on relay.floonet.dev.
	println!("\n[split] ===== PHASE 3: clearnet oracle (did the 1059 land?) =====");
	let mut landed = false;
	if let Some(id) = wrap_id {
		let oracle = Client::builder().signer(keys.clone()).build();
		connect(&oracle, RELAY).await;
		let of = Filter::new().id(id).limit(1);
		if let Ok(evs) = oracle
			.fetch_events_from([RELAY], of, Duration::from_secs(15))
			.await
		{
			landed = !evs.is_empty();
		}
		oracle.disconnect().await;
	}
	println!("[split] 1059 present on relay.floonet.dev via CLEARNET oracle = {landed}");

	println!("\n[split] ===== DONE. Read the scratch exit log for the byte breakdown. =====");
	println!("[split] expectation: 0 streams during phase 1; 1 money stream in phase 2 whose");
	println!("[split]  `X B in` is dominated by the ~30 KB 1059 publish (identity+lookup absent).");
}