//! End-to-end UAT for peat-mesh#175 — delivery-ratio thresholds for
//! sustained writes over a bandwidth-constrained peer link.
//!
//! peat-mesh#175 reported that Automerge sync delivery degrades far
//! below 100% at sustained write rates well below raw link bandwidth
//! (90.8% at 10 Hz × 400 B on a 256 kbps / 100 ms link in the original
//! lab repro). The root cause was traced to the `Arc<SyncChannelManager>`
//! dropped at the end of `AutomergeBackend::new` — the persistent
//! multiplexed sync stream path was structurally bypassed, and every
//! consumer fell back to the per-message-stream legacy path. ADR-063
//! and the rc.25 fixes (peat-mesh#176, #178, #180) wired up the
//! persistent path.
//!
//! `tests/two_backend_persistent_sync_e2e.rs` is the *architectural*
//! regression pin (asserts the strong-`Arc` survives construction).
//! That file's own header notes:
//!
//! > the central wire-up was a no-op, observable only by a careful
//! > code-read … or by lab measurement (UAT thresholds failed without
//! > it).
//!
//! This file is the *behavioural* UAT. It codifies the delivery-ratio
//! thresholds from peat-mesh#175 as cargo-runnable, CI-runnable
//! assertions — no sim, no external harness, no consumer-specific
//! terminology. Two `AutomergeBackend::with_iroh` instances run in
//! the same process; their iroh QUIC traffic is routed through a
//! userspace UDP relay that applies a token-bucket bandwidth limit
//! plus a fixed one-way delay (the same shape the original netem
//! setup imposed). The writer (peer A) emits documents at a fixed
//! rate for a steady-state window; after a drain window, the reader
//! (peer B) is polled for delivered documents and the ratio is
//! asserted against the threshold.
//!
//! ## Scope — what this UAT validates and what it does NOT
//!
//! **In scope (this UAT):** the *symmetric two-peer / single-relay*
//! case — one writer per direction, one shaped path per direction,
//! no contention from additional emitters. Passing this UAT proves
//! the persistent multiplexed sync stream (ADR-063) is wired through
//! `AutomergeBackend::with_iroh` and that the writer-task + bounded
//! mpsc + multi-frame loop architecture sustains the issue-defined
//! delivery thresholds at 1 Hz, 10 Hz, and 25 Hz under
//! 256 kbps / 100 ms shaping with ~MTU bucket capacity.
//!
//! **NOT in scope (handled in `peat-sim/experiments/7n-dual-c2`):**
//! the *multi-emitter / oversubscribed-link* case — two or more
//! concurrent writers sharing a constrained downlink, with the
//! offered load exceeding link capacity (e.g. two C2 nodes each
//! emitting at 25 Hz × 400 B onto a 256 kbps platform link
//! concurrent with attachment download). That workload is
//! bandwidth-bound, not sync-coordinator-bound, and a 100% pass
//! here will coexist with sub-threshold delivery there — see
//! peat-mesh#175 closing characterisation
//! (<https://github.com/defenseunicorns/peat-mesh/issues/175>) for
//! the rc.26 lab measurements (1/5/10 Hz at 100%/100%/99.9%, 25 Hz
//! c2-alpha 93.4% under multi-emitter contention).
//!
//! In practice: a regression here means the persistent-stream
//! architecture broke. A regression in the peat-sim 7n-dual-c2
//! sweep without a regression here means something changed in the
//! workload or the shaping, not in the sync coordinator. Keep the
//! distinction explicit when interpreting CI signal.
//!
//! ## Test design notes
//!
//! - **Single relay socket.** Both peers are told the other lives at
//!   the relay address (via `MemoryLookup`). The relay forwards on
//!   source-port. QUIC sees a stable peer address on each side, so
//!   no path-migration validation is triggered.
//! - **Bucket paces sends; delay is independent.** The token bucket
//!   controls when a packet enters the simulated wire; the one-way
//!   delay simulates in-flight time. Bucket capacity is ~MTU rather
//!   than one second of bandwidth, so each packet pays its own
//!   refill wait at the link rate (peat-mesh#184 QA: a multi-second
//!   capacity gives so much burst credit that the shaping is
//!   effectively unconstrained for the workload sizes this UAT
//!   uses).
//! - **Per-direction buckets.** Uplink and downlink are independent,
//!   matching the netem per-interface shaping in the original UAT.
//! - **Bidirectional emission.** peat-mesh#175's repro had two
//!   emitters competing on the same constrained link
//!   ("c2-alpha + c2-bravo each emitting"). The dropped-`Arc`
//!   regression only collapses delivery under that multi-emitter
//!   contention — a single-emitter test would also pass the legacy
//!   fallback path, so the test asserts on the *worse* of the two
//!   directions.
//! - **CI budget compromise.** peat-mesh#175's UAT calls for 60 s
//!   steady-state and 60 s drain; this test runs 30 s steady + 10 s
//!   drain per case to keep total CI wall-clock under a minute when
//!   the three tests run in parallel. The thresholds are unchanged;
//!   a shorter window just narrows the statistical headroom.
//! - **No consumer-specific naming.** "Peer A" / "peer B" rather
//!   than "C2" / "platform". A `peat-mesh` test is part of the
//!   protocol substrate; it should not name what's downstream of it.

#![cfg(feature = "automerge-backend")]

use std::collections::{HashMap, HashSet};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::{Duration, Instant};

use iroh::{EndpointAddr, TransportAddr};
use peat_mesh::security::FormationKey;
use peat_mesh::sync::{
    AutomergeBackend, AutomergeBackendConfig, BackendConfig, DataSyncBackend, Document, Query,
    SyncEngine,
};
use tempfile::TempDir;
use tokio::net::UdpSocket;
use tokio::sync::{mpsc, Mutex};

// peat-mesh#175 UAT thresholds. These are the issue's pass criteria
// verbatim — bumping these should be done by amending the issue first.
const THRESHOLD_1HZ: f64 = 0.995;
const THRESHOLD_10HZ: f64 = 0.990;
const THRESHOLD_25HZ: f64 = 0.950;

// peat-mesh#175 lab parameters. The shaping numbers reproduce the
// netem setup recorded in the issue (`rate 256kbit` in netem's
// conventional bits-per-second).
const PAYLOAD_BYTES: usize = 400;
const LINK_RATE_BYTES_PER_SEC: u64 = 256_000 / 8; // 256 kbps = 32 000 B/s
const ONE_WAY_DELAY: Duration = Duration::from_millis(100);

// Token-bucket capacity. Sized at ~MTU rather than one second of
// bandwidth so the bucket actually paces individual packets at link
// rate (peat-mesh#184 QA review): a multi-second capacity gives so
// much burst credit that a regression dropping delivery from 100% to
// ~96% would still pass the issue's thresholds. With capacity ≈ MTU,
// every packet exceeding one MTU has to wait for the bucket to refill,
// which is the netem `rate` shape the issue documents.
const BUCKET_CAPACITY_BYTES: u64 = 1500;

// CI-tuned windows (see header note). Steady-state long enough that
// each rate produces a statistically meaningful sample; drain long
// enough to absorb in-flight packets at the constrained rate.
const STEADY_SECS: u64 = 30;
const DRAIN_SECS: u64 = 10;

/// Token bucket. Always blocks until tokens are available — models
/// netem `rate` (which delays rather than drops at the link layer).
///
/// **Assumes a single producer per bucket.** The wait-time computation
/// in `acquire` releases the lock before sleeping; if a second producer
/// drained tokens between unlock and wake, the sleeper would wake to
/// insufficient tokens and add another full wait, biasing the rate
/// downwards under contention. The relay design instantiates one
/// bucket per direction with one sender task per bucket, so this is
/// dormant — but a third producer would break it (peat-mesh#184 QA).
struct TokenBucket {
    rate_bytes_per_sec: u64,
    capacity: u64,
    inner: Mutex<TokenBucketState>,
}

struct TokenBucketState {
    last_refill: Instant,
    tokens: u64,
}

impl TokenBucket {
    fn new(rate_bytes_per_sec: u64, capacity: u64) -> Self {
        Self {
            rate_bytes_per_sec,
            capacity,
            inner: Mutex::new(TokenBucketState {
                last_refill: Instant::now(),
                tokens: capacity,
            }),
        }
    }

    async fn acquire(&self, n: u64) {
        loop {
            let wait = {
                let mut state = self.inner.lock().await;
                let now = Instant::now();
                let elapsed = now.duration_since(state.last_refill).as_secs_f64();
                let refill = (elapsed * self.rate_bytes_per_sec as f64) as u64;
                state.tokens = state.tokens.saturating_add(refill).min(self.capacity);
                state.last_refill = now;
                if state.tokens >= n {
                    state.tokens -= n;
                    return;
                }
                let needed = n - state.tokens;
                Duration::from_secs_f64(needed as f64 / self.rate_bytes_per_sec as f64)
            };
            tokio::time::sleep(wait).await;
        }
    }
}

/// Userspace UDP relay applying per-direction token-bucket rate-limit
/// plus a fixed one-way delay. Returns the relay's bound socket
/// address; callers send their peer-lookup pointers at this address.
struct Relay {
    addr: SocketAddr,
    _abort: tokio::task::AbortHandle,
}

impl Relay {
    async fn spawn(peer_a: SocketAddr, peer_b: SocketAddr) -> Self {
        let sock = UdpSocket::bind(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0))
            .await
            .expect("relay bind");
        let addr = sock.local_addr().expect("relay local_addr");
        let sock = Arc::new(sock);

        let (tx_ab, mut rx_ab) = mpsc::channel::<Vec<u8>>(4096);
        let (tx_ba, mut rx_ba) = mpsc::channel::<Vec<u8>>(4096);

        // A → B sender task: paces by the A→B bucket, then schedules
        // each forwarded packet with the one-way delay applied
        // independently (so bucket pacing models link rate, delay
        // models RTT).
        //
        // Per-packet `tokio::spawn(sleep + send)` is the simplest way
        // to apply a fixed forward delay without serialising the sender
        // task on its own sleeps. At the rates this UAT runs (≤ 25 Hz
        // × 2 directions × ~30 s) total spawned tasks are O(10⁴), which
        // tokio handles trivially. A `DelayQueue`-based forwarder is
        // strictly lighter and would also serialise sends in arrival
        // order; that optimisation belongs in peat-sim's broader
        // network-shape harness, not in this UAT, per the per-repo
        // UAT vs peat-sim split (peat-mesh#184 QA review).
        let sock_ab = sock.clone();
        tokio::spawn(async move {
            let bucket = Arc::new(TokenBucket::new(
                LINK_RATE_BYTES_PER_SEC,
                BUCKET_CAPACITY_BYTES,
            ));
            while let Some(data) = rx_ab.recv().await {
                bucket.acquire(data.len() as u64).await;
                let sock_clone = sock_ab.clone();
                tokio::spawn(async move {
                    tokio::time::sleep(ONE_WAY_DELAY).await;
                    let _ = sock_clone.send_to(&data, peer_b).await;
                });
            }
        });

        let sock_ba = sock.clone();
        tokio::spawn(async move {
            let bucket = Arc::new(TokenBucket::new(
                LINK_RATE_BYTES_PER_SEC,
                BUCKET_CAPACITY_BYTES,
            ));
            while let Some(data) = rx_ba.recv().await {
                bucket.acquire(data.len() as u64).await;
                let sock_clone = sock_ba.clone();
                tokio::spawn(async move {
                    tokio::time::sleep(ONE_WAY_DELAY).await;
                    let _ = sock_clone.send_to(&data, peer_a).await;
                });
            }
        });

        // Dispatch loop. Routes by source port — packets from peer A's
        // bound port go to the A→B channel; symmetric for B.
        let dispatcher = tokio::spawn(async move {
            let mut buf = vec![0u8; 65536];
            loop {
                let (n, src) = match sock.recv_from(&mut buf).await {
                    Ok(v) => v,
                    Err(_) => break,
                };
                if src == peer_a {
                    let _ = tx_ab.send(buf[..n].to_vec()).await;
                } else if src == peer_b {
                    let _ = tx_ba.send(buf[..n].to_vec()).await;
                }
                // stray packets from other sources: silently drop.
            }
        });

        Relay {
            addr,
            _abort: dispatcher.abort_handle(),
        }
    }
}

async fn build_backend(formation_id: &str, shared_key: &str) -> (Arc<AutomergeBackend>, TempDir) {
    let dir = TempDir::new().expect("tempdir");
    let mut cfg = AutomergeBackendConfig::default();
    cfg.data_dir = dir.path().to_path_buf();
    cfg.formation_id = formation_id.to_string();
    cfg.base64_shared_key = shared_key.to_string();
    cfg.iroh_bind_addr = Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
    let backend = AutomergeBackend::with_iroh(cfg).await.expect("with_iroh");
    backend
        .initialize(BackendConfig {
            app_id: "uat".to_string(),
            persistence_dir: dir.path().to_path_buf(),
            shared_key: None,
            transport: peat_mesh::sync::TransportConfig::default(),
            extra: std::collections::HashMap::new(),
        })
        .await
        .expect("initialize");
    (backend, dir)
}

/// Spawn an emitter task: writes documents at `target_rate_hz` for
/// `STEADY_SECS`, each tagged with `id_prefix` so the receiving side
/// can distinguish this emitter's docs. Returns the set of emitted
/// document ids when the steady-state window closes.
async fn emit_loop(
    backend: Arc<AutomergeBackend>,
    id_prefix: &'static str,
    target_rate_hz: f64,
) -> HashSet<String> {
    let collection = "uat";
    let interval = Duration::from_secs_f64(1.0 / target_rate_hz);
    let steady_until = Instant::now() + Duration::from_secs(STEADY_SECS);
    let mut emitted_ids: HashSet<String> = HashSet::new();
    let mut next_send = Instant::now();
    let mut seq: u64 = 0;
    while Instant::now() < steady_until {
        let now = Instant::now();
        if next_send > now {
            tokio::time::sleep(next_send - now).await;
        }
        next_send += interval;

        let doc_id = format!("{}-{:08}", id_prefix, seq);
        seq += 1;
        let mut payload = HashMap::new();
        // PAYLOAD_BYTES of filler. On-wire size is larger
        // (Automerge framing + QUIC overhead) but matches the
        // issue's per-doc accounting.
        payload.insert(
            "filler".to_string(),
            serde_json::Value::String("x".repeat(PAYLOAD_BYTES)),
        );
        let doc = Document::with_id(&doc_id, payload);
        backend
            .document_store()
            .upsert(collection, doc)
            .await
            .expect("upsert");
        emitted_ids.insert(doc_id);
    }
    emitted_ids
}

/// Result of one direction (one emitter → its receiver) of the trial.
#[derive(Debug, Clone, Copy)]
struct DirectionResult {
    emitted: usize,
    delivered: usize,
}

impl DirectionResult {
    fn ratio(&self) -> f64 {
        if self.emitted == 0 {
            0.0
        } else {
            self.delivered as f64 / self.emitted as f64
        }
    }
}

/// Run one bidirectional delivery-ratio trial: both peers emit at
/// `target_rate_hz` for `STEADY_SECS`, then drain for `DRAIN_SECS`,
/// then both delivery directions are counted independently. Returns
/// (A→B, B→A) results.
async fn run_delivery_trial(target_rate_hz: f64) -> (DirectionResult, DirectionResult) {
    let shared_key = FormationKey::generate_secret();
    let (backend_a, _adir) = build_backend("uat-formation", &shared_key).await;
    let (backend_b, _bdir) = build_backend("uat-formation", &shared_key).await;

    let addr_a = backend_a.blob_store().endpoint().addr();
    let addr_b = backend_b.blob_store().endpoint().addr();
    let sock_a = first_loopback_addr(&addr_a);
    let sock_b = first_loopback_addr(&addr_b);
    let id_a = backend_a.blob_store().endpoint().id();
    let id_b = backend_b.blob_store().endpoint().id();

    // Spawn the constrained-bandwidth relay between the two real
    // sockets. Returns the relay's listening address — that's what
    // each side will believe is the other peer's address.
    let relay = Relay::spawn(sock_a, sock_b).await;

    // Lie to both sides: each peer believes the other lives at the
    // relay. QUIC packets traverse A → relay → B (and symmetrically),
    // shaped at LINK_RATE_BYTES_PER_SEC + ONE_WAY_DELAY in each direction.
    let fake_b = EndpointAddr {
        id: id_b,
        addrs: std::collections::BTreeSet::from([TransportAddr::Ip(relay.addr)]),
    };
    let fake_a = EndpointAddr {
        id: id_a,
        addrs: std::collections::BTreeSet::from([TransportAddr::Ip(relay.addr)]),
    };
    backend_a
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(fake_b);
    backend_b
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(fake_a);

    backend_a.start_sync().await.expect("start_sync A");
    backend_b.start_sync().await.expect("start_sync B");

    // A initiates the formation handshake against B. The `SyncEngine`
    // trait's `connect_to_peer` does *only* the auth handshake +
    // connection-registration on the initiator side — it does NOT
    // wire the per-connection sync streams the way the acceptor's
    // `SyncProtocolHandler::accept` does. Without that wire-up on the
    // initiator, sync is single-direction: A → B works (B's accept
    // handler set up the streams) but B → A silently never flows
    // (A's side has the connection but no per-connection sync task).
    //
    // peat-node mirrors this pattern explicitly — see
    // `peat-node/src/node.rs` `connect_peer`: connect_and_authenticate
    // → start_sync_connection. This test replicates that contract so
    // the bidirectional persistent-stream behaviour is what's under
    // test, not the trait-method asymmetry. (If `connect_to_peer`'s
    // contract is ever revised to do the wire-up implicitly, this
    // block collapses to a single call.)
    let conn_ab = backend_a
        .transport()
        .connect_and_authenticate(id_b)
        .await
        .expect("A connect_and_authenticate to B");
    backend_a
        .transport()
        .start_sync_connection(conn_ab, Arc::clone(backend_a.coordinator()));

    // Run both emitters concurrently for the steady-state window.
    // Disjoint doc-id prefixes ("a-" and "b-") so each side's
    // delivered-set check counts only the other side's emissions.
    let a_emitter = tokio::spawn(emit_loop(backend_a.clone(), "a", target_rate_hz));
    let b_emitter = tokio::spawn(emit_loop(backend_b.clone(), "b", target_rate_hz));
    let emitted_a = a_emitter.await.expect("A emitter");
    let emitted_b = b_emitter.await.expect("B emitter");

    // Drain window — let the persistent sync stream catch up.
    tokio::time::sleep(Duration::from_secs(DRAIN_SECS)).await;

    // Count delivery in each direction. A → B = B's view of A's
    // emissions; B → A = A's view of B's emissions.
    let collection = "uat";
    let a_to_b_delivered = count_delivered(&backend_b, collection, &emitted_a).await;
    let b_to_a_delivered = count_delivered(&backend_a, collection, &emitted_b).await;

    backend_a.shutdown().await.expect("shutdown A");
    backend_b.shutdown().await.expect("shutdown B");
    drop(relay);

    (
        DirectionResult {
            emitted: emitted_a.len(),
            delivered: a_to_b_delivered,
        },
        DirectionResult {
            emitted: emitted_b.len(),
            delivered: b_to_a_delivered,
        },
    )
}

async fn count_delivered(
    receiver: &Arc<AutomergeBackend>,
    collection: &str,
    expected_ids: &HashSet<String>,
) -> usize {
    // The trait-default `DocumentStore::get(collection, id)` resolves
    // to `query(collection, Eq{id, value})`, and `AutomergeBackend::query`
    // does a `scan_prefix` + Automerge deserialize over every entry in
    // the collection, filtering in Rust. Per-id `get` in a loop is
    // therefore O(N²); at 25 Hz × 30 s × 2 directions that's ~2.25 M
    // Automerge deserialize ops, which dominated the original test's
    // wall-clock (~17 min for the 25 Hz case alone). Single bulk
    // `Query::All` collapses that to O(N).
    let received: HashSet<String> = receiver
        .document_store()
        .query(collection, &Query::All)
        .await
        .expect("receiver query")
        .into_iter()
        .filter_map(|d| d.id)
        .collect();
    expected_ids.intersection(&received).count()
}

fn first_loopback_addr(ep: &EndpointAddr) -> SocketAddr {
    ep.addrs
        .iter()
        .find_map(|ta| match ta {
            TransportAddr::Ip(s) if s.ip().is_loopback() => Some(*s),
            _ => None,
        })
        .or_else(|| {
            ep.addrs.iter().find_map(|ta| match ta {
                TransportAddr::Ip(s) => Some(*s),
                _ => None,
            })
        })
        .expect("endpoint has at least one ip transport addr")
}

fn assert_both_directions_meet_threshold(
    rate_hz: f64,
    threshold: f64,
    a_to_b: DirectionResult,
    b_to_a: DirectionResult,
) {
    let r_ab = a_to_b.ratio();
    let r_ba = b_to_a.ratio();
    eprintln!(
        "peat-mesh#175 UAT: {rate_hz} Hz × {steady}s bidirectional — \
         A→B {ad}/{ae}={rab:.4}, B→A {bd}/{be}={rba:.4} (threshold ≥ {threshold:.3})",
        steady = STEADY_SECS,
        ad = a_to_b.delivered,
        ae = a_to_b.emitted,
        bd = b_to_a.delivered,
        be = b_to_a.emitted,
        rab = r_ab,
        rba = r_ba,
    );
    // Worst direction must meet the threshold. #175's "target =
    // platform-3" gate was directional (one specific receiver); in
    // a symmetric two-peer test, "worst direction" is the strictest
    // equivalent.
    let worst = r_ab.min(r_ba);
    assert!(
        worst >= threshold,
        "worst-direction delivery ratio {worst:.4} at {rate_hz} Hz is below the \
         peat-mesh#175 threshold {threshold:.3} — the persistent multiplexed sync stream \
         path is not sustaining the bidirectional workload (A→B {r_ab:.4}, B→A {r_ba:.4})"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn delivery_at_1hz_meets_99_5_percent_threshold() {
    let (ab, ba) = run_delivery_trial(1.0).await;
    assert_both_directions_meet_threshold(1.0, THRESHOLD_1HZ, ab, ba);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn delivery_at_10hz_meets_99_0_percent_threshold() {
    let (ab, ba) = run_delivery_trial(10.0).await;
    assert_both_directions_meet_threshold(10.0, THRESHOLD_10HZ, ab, ba);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn delivery_at_25hz_meets_95_0_percent_threshold() {
    let (ab, ba) = run_delivery_trial(25.0).await;
    assert_both_directions_meet_threshold(25.0, THRESHOLD_25HZ, ab, ba);
}
