//! Invariants for `ChannelManager` routing under adversarial op
//! sequences (ADR-039/ADR-040). The harness mirrors the manager's
//! documented behavior exactly — parked-chunk and drop counters must
//! match to the unit after every op, parked bytes must respect the
//! per-channel cap bound, and every byte routed into a known channel
//! must be readable until an EOF sentinel (lossless bounded-buffer
//! routing, ADR-040 REQ-CH-05). The §6.2-1 probe: distinct
//! never-adopted channel ids each park up to 64 chunks — the fuzzer
//! explores whether the total can exceed that documented bound.

use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use bytes::Bytes;

use alkcall::channels::manager::{ChannelManager, ChannelSide, ManagerError};
use alkcall::channels::mux::MuxRunner;
use alkcall::channels::reassembly::{MpscRecvStream, DEFAULT_BUFFER_CAP};

/// The documented per-channel early-arrival cap (`EARLY_ARRIVAL_CAP`,
/// private in `manager.rs`; review 006 E-04 treats 64 parked chunks as
/// the observable bound).
pub const EARLY_ARRIVAL_CAP: usize = 64;

/// The maximum chunk payload length (16 MiB, ADR-034).
pub const MAX_CHUNK_LEN: u32 = 16 * 1024 * 1024;

#[derive(Debug, arbitrary::Arbitrary, Clone)]
pub enum ManagerOp {
    Route {
        channel_id: u32,
        len: u16,
        byte: u8,
    },
    Open {
        alpn_idx: u8,
        opener_idx: u8,
    },
    /// Adopt a channel whose id the REMOTE side allocated. The
    /// harness manager is `ChannelSide::Accept`, so the peer is the
    /// Connect side and its allocated ids are odd (ADR-047 §5) — the
    /// op masks the fuzzer's arbitrary id into the legal range.
    Adopt {
        channel_id: u32,
    },
    /// Teardown an arbitrary channel id (any live id — teardown is
    /// local, not range-constrained).
    Teardown {
        channel_id: u32,
    },
    /// Transport EOF — clear everything.
    ClearAll,
}

#[derive(Debug, arbitrary::Arbitrary)]
pub struct ManagerSequence {
    pub ops: Vec<ManagerOp>,
}

/// Exact model of the manager's early-arrival bookkeeping.
#[derive(Default)]
struct ParkModel {
    /// Parked chunk payloads per channel id, FIFO (0-length marks the
    /// EOF sentinel a drained queue will deliver).
    parked: HashMap<u32, VecDeque<u16>>,
    /// Model of the monotonic `early_arrival_count`.
    parked_total: u64,
    /// Model of the monotonic `dropped_unknown_chunks`.
    dropped_total: u64,
}

impl ParkModel {
    fn route_to_unknown(&mut self, channel_id: u32, len: u16) {
        let queue = self.parked.entry(channel_id).or_default();
        if queue.len() >= EARLY_ARRIVAL_CAP {
            self.dropped_total += 1;
        } else {
            queue.push_back(len);
            self.parked_total += 1;
        }
    }

    fn parked_bytes(&self) -> u64 {
        self.parked
            .values()
            .map(|q| q.iter().map(|l| u64::from(*l)).sum::<u64>())
            .sum()
    }

    /// Bytes of a drained queue that are readable (everything before
    /// the first zero-length sentinel — the reassembled stream latches
    /// EOF there and drops the rest, reassembly.rs `poll_read`).
    fn readable_bytes_until_sentinel(queue: &VecDeque<u16>) -> u64 {
        let mut total = 0;
        for len in queue {
            if *len == 0 {
                break;
            }
            total += u64::from(*len);
        }
        total
    }
}

struct ManagerHarness {
    manager: ChannelManager,
    model: ParkModel,
    /// Readable-byte model per drained channel id (what its drainer
    /// will have read once the channel ends). Drainers accumulate:
    /// `open_channel` then `adopt_channel` on the same id each install
    /// a fresh stream; earlier drainers end at the sender drop but the
    /// id's routed bytes must reconcile exactly once, so each new
    /// drainer supersedes the id's model entry.
    readable: HashMap<u32, u64>,
    /// Byte totals for drainers that ended mid-sequence (sender drop
    /// on adopt-over-adopt) — reconciled into the post-sequence total.
    retired_drainer_bytes: u64,
    /// Whether the CURRENT receiver for each id latched an EOF
    /// sentinel — per-receiver state, not per-id: a fresh adopt
    /// installs a fresh reassembled stream with no latched EOF, while
    /// a sentinel inside a drained early-arrival queue latches the
    /// NEW receiver too (`drain_early_arrivals` delivers it into it).
    sentinel_seen: HashMap<u32, bool>,
    /// (channel_id, bytes-read counter, drainer task).
    drainers: Vec<(u32, Arc<AtomicU64>, tokio::task::JoinHandle<()>)>,
    _mux_task: tokio::task::JoinHandle<()>,
}

impl ManagerHarness {
    fn new() -> Self {
        let (_client, server) = tokio::io::duplex(4096);
        let (_reader, writer) = tokio::io::split(server);
        let (mux_handle, runner) = MuxRunner::new(Box::new(writer));
        let mux_task = tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = ChannelManager::new(
            mux_handle,
            64,
            DEFAULT_BUFFER_CAP,
            None,
            ChannelSide::Accept,
        );
        Self {
            manager,
            model: ParkModel::default(),
            readable: HashMap::new(),
            retired_drainer_bytes: 0,
            sentinel_seen: HashMap::new(),
            drainers: Vec::new(),
            _mux_task: mux_task,
        }
    }

    async fn step(&mut self, op: &ManagerOp, max_payload: usize) -> Result<(), String> {
        match op {
            ManagerOp::Route {
                channel_id,
                len,
                byte,
            } => {
                let payload_len = (*len as usize) % (max_payload + 1);
                let len = payload_len as u16;
                let known = self.manager.has_channel(*channel_id);
                self.manager
                    .route_payload(*channel_id, Bytes::from(vec![*byte; payload_len]))
                    .await;
                if known {
                    if !self.sentinel_seen(*channel_id) {
                        *self.readable.entry(*channel_id).or_insert(0) += u64::from(len);
                        if len == 0 {
                            self.sentinel_seen.insert(*channel_id, true);
                        }
                    }
                } else {
                    self.model.route_to_unknown(*channel_id, len);
                }
            }
            ManagerOp::Open {
                alpn_idx,
                opener_idx,
            } => {
                let alpn = pick(alpn_idx, ["alk/tty", "alk/sub", "alk/x"]);
                let opener = pick(opener_idx, ["alice", "bob", "carol"]);
                match self.manager.open_channel(alpn, opener, None).await {
                    Ok((_id, _send, recv)) => self.spawn_drainer(_id, recv),
                    // The per-connection cap is a documented bound.
                    Err(ManagerError::TooManyChannels { .. }) => {}
                    Err(other) => return Err(format!("open_channel: unexpected {other:?}")),
                }
            }
            ManagerOp::Adopt { channel_id } => {
                // The harness manager is the Accept side; the peer
                // (Connect side) allocates odd ids, so only odd ids
                // arrive for adoption (ADR-047 §5 odd/even split).
                let channel_id = (*channel_id | 1).max(1);
                match self
                    .manager
                    .adopt_channel(channel_id, "alk/tty", None)
                    .await
                {
                    Ok((_send, recv)) => {
                        // Fresh receiver (no latched EOF), then
                        // supersede the id's prior drainer (retire its
                        // model, reset its sentinel latch), THEN fold
                        // the drained parked queue into the new
                        // receiver's model — `drain_early_arrivals`
                        // delivers it into the new receiver, sentinel
                        // included.
                        self.spawn_drainer(channel_id, recv);
                        if let Some(queue) = self.model.parked.remove(&channel_id) {
                            let drained_readable = ParkModel::readable_bytes_until_sentinel(&queue);
                            *self.readable.entry(channel_id).or_insert(0) += drained_readable;
                            if queue.contains(&0) {
                                self.sentinel_seen.insert(channel_id, true);
                            }
                        }
                    }
                    Err(ManagerError::ChannelExists(_)) => {}
                    Err(ManagerError::TooManyChannels { .. }) => {}
                    Err(other) => return Err(format!("adopt_channel: unexpected {other:?}")),
                }
            }
            ManagerOp::Teardown { channel_id } => {
                match self.manager.teardown_channel(*channel_id) {
                    Ok(_task) => {}
                    Err(ManagerError::UnknownChannel(_)) => {}
                    Err(other) => return Err(format!("teardown_channel: unexpected {other:?}")),
                }
            }
            ManagerOp::ClearAll => {
                let dropped_before = self.manager.dropped_unknown_chunks();
                let _ledger_entries = self.manager.clear_all();
                // clear_all returns the opener-ledger entries (ADR-047
                // §7 — the per-identity decrement source), not the
                // channel map: adopted channels carry no ledger entry
                // (the remote side owns their count) and torn-down
                // channels keep theirs.
                self.model.parked.clear();
                assert_eq!(
                    self.manager.dropped_unknown_chunks(),
                    dropped_before,
                    "clear_all does not reset the monotonic drop counter"
                );
                assert_eq!(
                    self.manager.open_count(),
                    0,
                    "clear_all empties the channel map"
                );
            }
        }
        self.assert_counters()
    }

    fn sentinel_seen(&self, channel_id: u32) -> bool {
        self.sentinel_seen
            .get(&channel_id)
            .copied()
            .unwrap_or(false)
    }

    fn spawn_drainer(&mut self, channel_id: u32, recv: MpscRecvStream) {
        // A fresh drainer for an id supersedes any earlier one: the
        // earlier stream's sender was dropped (teardown) or routed
        // past an EOF sentinel, so its readable-byte model is retired
        // into the reconciled total and the id's model restarts. The
        // id's sentinel latch also resets — the new receiver is a
        // fresh reassembled stream (any sentinel in the drained
        // early-arrival queue is re-applied by the adopt arm AFTER
        // this, since `drain_early_arrivals` delivers it into the new
        // receiver too).
        if let Some(bytes) = self.readable.remove(&channel_id) {
            self.retired_drainer_bytes += bytes;
        }
        self.sentinel_seen.remove(&channel_id);
        let counter = Arc::new(AtomicU64::new(0));
        let task = tokio::spawn(drain_task(recv, Arc::clone(&counter)));
        self.drainers.push((channel_id, counter, task));
    }

    fn assert_counters(&self) -> Result<(), String> {
        let actual_parked = self.manager.early_arrival_count();
        let actual_dropped = self.manager.dropped_unknown_chunks();
        if self.model.parked_total != actual_parked {
            return Err(format!(
                "parked-chunk model {} != manager counter {actual_parked}",
                self.model.parked_total
            ));
        }
        if self.model.dropped_total != actual_dropped {
            return Err(format!(
                "dropped-chunk model {} != manager counter {actual_dropped}",
                self.model.dropped_total
            ));
        }
        Ok(())
    }
}

/// Drain a channel's reassembled stream to EOF, counting bytes read.
/// Runs until Ok(0) — sender drop (teardown / clear_all) or EOF
/// sentinel both end the stream after queued bytes drain.
async fn drain_task(mut recv: MpscRecvStream, counter: Arc<AtomicU64>) {
    use tokio::io::AsyncReadExt;
    let mut buf = [0u8; 512];
    loop {
        match recv.read(&mut buf).await {
            Ok(0) => break,
            Ok(n) => {
                counter.fetch_add(n as u64, Ordering::Relaxed);
            }
            Err(_) => break,
        }
    }
}

fn pick<const N: usize>(idx: &u8, options: [&'static str; N]) -> &'static str {
    options[(*idx as usize) % N]
}

/// The whole-sequence driver. Bounds: 256 ops and 4 KiB payloads keep
/// exec/s high while covering the DoS class (distinct-id spray: a
/// 256-op sequence at 4 KiB parks up to ~1 MiB, and the exact counter
/// plus cap-bound invariants are what the fuzzer must not break).
pub fn fuzz_manager_routing(seq: &ManagerSequence) {
    const MAX_OPS: usize = 256;
    const MAX_PAYLOAD: usize = 4096;

    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("current-thread runtime builds")
        .block_on(async {
            let mut harness = ManagerHarness::new();
            for (i, op) in seq.ops.iter().take(MAX_OPS).enumerate() {
                if let Err(msg) = harness.step(op, MAX_PAYLOAD).await {
                    panic!("invariant violated at op {i} ({op:?}): {msg}");
                }
                // The parked-bytes total must respect the documented
                // per-channel bound summed over distinct parked ids —
                // the §6.2-1 property.
                let bound = harness.model.parked.len() as u64
                    * EARLY_ARRIVAL_CAP as u64
                    * u64::from(MAX_CHUNK_LEN);
                assert!(
                    harness.model.parked_bytes() <= bound,
                    "parked bytes {} exceed the documented bound {bound} at op {i}",
                    harness.model.parked_bytes()
                );
            }
            // Post-sequence: clear_all (the transport-EOF path) drops
            // every sender — all drainers run to EOF and their byte
            // totals must reconcile with the readable-byte model
            // (lossless bounded-buffer routing, sentinel semantics
            // included).
            harness
                .step(&ManagerOp::ClearAll, MAX_PAYLOAD)
                .await
                .expect("clear_all");
            let mut expected_total = harness.retired_drainer_bytes;
            for bytes in harness.readable.values() {
                expected_total += bytes;
            }
            let mut actual_total = 0u64;
            for (_id, counter, task) in harness.drainers.drain(..) {
                let _ = task.await;
                actual_total += counter.load(Ordering::Relaxed);
            }
            assert_eq!(
                actual_total, expected_total,
                "routed-to-known bytes must reconcile with drained bytes (lossless reassembly)"
            );
        });
}

#[cfg(test)]
mod corpus_replay {
    use super::*;

    /// Replay the committed op-sequence corpus through the driver.
    /// Note `fuzz_manager_routing` spawns a current-thread runtime per
    /// call; each sequence runs to completion inside it.
    #[test]
    fn committed_corpus_replays_through_the_invariants() {
        let corpus =
            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../corpus/manager_routing");
        let mut count = 0usize;
        for entry in std::fs::read_dir(&corpus).expect("corpus directory is committed") {
            let path = entry.expect("corpus entry readable").path();
            let data = std::fs::read(&path).expect("corpus entry readable");
            let u = arbitrary::Unstructured::new(&data);
            let seq: ManagerSequence =
                arbitrary::Arbitrary::arbitrary_take_rest(u).expect("seed decodes");
            fuzz_manager_routing(&seq);
            count += 1;
        }
        assert!(
            count >= 2,
            "committed seed corpus is present, found {count}"
        );
    }

    /// Semantic fixtures: the op sequences that pinned the adopt/open
    /// insert-replace bug, hand-encoded through `Unstructured` bytes
    /// the same way libFuzzer would feed them.
    #[test]
    fn duplicate_adopt_sequence_survives_the_fix() {
        // Encoded via the mr-decode tool's format (arbitrary 1.4.x):
        // 3 adopts of one id, a route, an EOF sentinel route.
        let seq = ManagerSequence {
            ops: vec![
                ManagerOp::Adopt { channel_id: 42 },
                ManagerOp::Adopt { channel_id: 42 },
                ManagerOp::Adopt { channel_id: 42 },
                ManagerOp::Route {
                    channel_id: 42,
                    len: 2418,
                    byte: 121,
                },
                ManagerOp::Route {
                    channel_id: 0,
                    len: 0,
                    byte: 0,
                },
            ],
        };
        fuzz_manager_routing(&seq);
    }
}