//! Independent count and time windows must match a filtered replay through
//! StatsCollector for ordinary metrics, while directional loss has an independent
//! interval model and cumulative accounting retains the complete history.
//! Includes delayed timeout discovery and replies with backdated timestamps.

use std::time::{Duration, Instant, UNIX_EPOCH};

use irtt_client::{
    ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, ReceivedStatsSample, RttSample,
    ServerTiming, SignedDuration,
};
use irtt_stats::{LateReplyMode, SampleMode, Snapshot, StatsCollector, StatsConfig};
use proptest::prelude::*;

/// One generated operation. `Advance` moves the shared monotonic clock
/// forward without producing an event; other variants produce a normalized
/// event, with delayed events timestamped before the clock's current value.
#[derive(Debug, Clone)]
enum Op {
    /// Advance the shared clock by this many milliseconds.
    Advance(u16),
    /// A probe send.
    Send,
    /// An on-time unique reply with the given client-observed RTT.
    Reply { raw_ms: u16 },
    /// A reply processed after later events, but timestamped at receipt.
    DelayedReply { raw_ms: u16, age_ms: u16 },
    /// A late reply matched to retained send state (measurable).
    LateReplyMatched { raw_ms: u16 },
    /// A late reply that could not be matched to retained state.
    LateReplyUnmatched,
    /// A duplicate reply for an already-completed sequence.
    Duplicate,
    /// A probe timeout / loss.
    Loss,
    /// A timeout discovered this many milliseconds after its deadline.
    DelayedLoss { age_ms: u16 },
    /// A diagnostic warning event.
    Warning,
}

fn op_strategy() -> impl Strategy<Value = Op> {
    prop_oneof![
        3 => (0u16..40).prop_map(Op::Advance),
        4 => Just(Op::Send),
        4 => (0u16..60).prop_map(|raw_ms| Op::Reply { raw_ms }),
        2 => (0u16..60, 0u16..100).prop_map(|(raw_ms, age_ms)| Op::DelayedReply { raw_ms, age_ms }),
        2 => (0u16..60).prop_map(|raw_ms| Op::LateReplyMatched { raw_ms }),
        1 => Just(Op::LateReplyUnmatched),
        1 => Just(Op::Duplicate),
        2 => Just(Op::Loss),
        2 => (0u16..100).prop_map(|age_ms| Op::DelayedLoss { age_ms }),
        1 => Just(Op::Warning),
    ]
}

const ADDR: &str = "127.0.0.1:2112";

/// Builds a `ClientTimestamp` purely by adding a millisecond offset to a
/// fixed `base` `Instant` (and to `UNIX_EPOCH` for the wall-clock half). No
/// operation below ever reads the real clock more than once (to establish
/// `base`), so the resulting timestamps are fully deterministic functions of
/// the generated `Op`s, with no wall-clock jitter to make a proptest shrink
/// or CI run flaky.
fn ts_at(base: Instant, ms: u64) -> ClientTimestamp {
    ClientTimestamp {
        mono: base + Duration::from_millis(ms),
        wall: UNIX_EPOCH + Duration::from_millis(ms),
    }
}

fn server_timing(seq: u32) -> ServerTiming {
    let base_ns = i64::from(seq) * 10_000_000;
    ServerTiming {
        receive_mono_ns: Some(base_ns + 1_000_000),
        send_mono_ns: Some(base_ns + 2_000_000),
        receive_wall_ns: Some(base_ns + 1_000_000),
        send_wall_ns: Some(base_ns + 2_000_000),
        midpoint_mono_ns: None,
        midpoint_wall_ns: None,
        processing: Some(Duration::from_micros(100)),
    }
}

fn one_way(seq: u32) -> OneWayDelaySample {
    OneWayDelaySample {
        client_to_server: Some(SignedDuration::from_nanos(1_000_000 + i128::from(seq))),
        server_to_client: Some(SignedDuration::from_nanos(2_000_000 + i128::from(seq))),
    }
}

/// Builds the `ClientEvent` for one non-`Advance` op. `clock_ms` is the
/// shared clock's current value; delayed events use their earlier deadline or
/// receipt time as the event's windowing timestamp.
fn build_event(op: &Op, seq: u32, clock_ms: u64, base: Instant) -> ClientEvent {
    let clock_ms = match *op {
        Op::DelayedReply { age_ms, .. } => clock_ms.saturating_sub(u64::from(age_ms)),
        _ => clock_ms,
    };
    match *op {
        Op::Advance(_) => unreachable!("Advance does not produce an event"),
        Op::Send => ClientEvent::EchoSent {
            seq,
            remote: ADDR.parse().unwrap(),
            scheduled_at: Some(ts_at(base, clock_ms).mono),
            sent_at: ts_at(base, clock_ms),
            bytes: 32,
            send_call: Duration::from_micros(10),
            timer_error: Some(Duration::from_micros(2)),
        },
        Op::Reply { raw_ms } | Op::DelayedReply { raw_ms, .. } => {
            let sent_at = ts_at(base, clock_ms.saturating_sub(u64::from(raw_ms)));
            let received_at = ts_at(base, clock_ms);
            ClientEvent::EchoReply {
                seq,
                remote: ADDR.parse().unwrap(),
                sent_at,
                received_at,
                rtt: RttSample {
                    raw: Duration::from_millis(u64::from(raw_ms)),
                    adjusted: None,
                    effective: SignedDuration::from_duration(Duration::from_millis(u64::from(
                        raw_ms,
                    ))),
                },
                server_timing: Some(server_timing(seq)),
                one_way: Some(one_way(seq)),
                received_stats: Some(ReceivedStatsSample {
                    count: Some(seq + 1),
                    window: Some(0xff),
                }),
                bytes: 64,
                packet_meta: PacketMeta::default(),
            }
        }
        Op::LateReplyMatched { raw_ms } => {
            let sent_at = ts_at(base, clock_ms.saturating_sub(u64::from(raw_ms)));
            let received_at = ts_at(base, clock_ms);
            ClientEvent::LateReply {
                seq,
                highest_seen: seq + 1,
                remote: ADDR.parse().unwrap(),
                sent_at: Some(sent_at),
                received_at,
                rtt: Some(RttSample {
                    raw: Duration::from_millis(u64::from(raw_ms)),
                    adjusted: None,
                    effective: SignedDuration::from_duration(Duration::from_millis(u64::from(
                        raw_ms,
                    ))),
                }),
                server_timing: Some(server_timing(seq)),
                one_way: Some(one_way(seq)),
                received_stats: Some(ReceivedStatsSample {
                    count: Some(seq + 1),
                    window: Some(0xff),
                }),
                bytes: 64,
                packet_meta: PacketMeta::default(),
            }
        }
        Op::LateReplyUnmatched => ClientEvent::LateReply {
            seq,
            highest_seen: seq + 1,
            remote: ADDR.parse().unwrap(),
            sent_at: None,
            received_at: ts_at(base, clock_ms),
            rtt: None,
            server_timing: None,
            one_way: None,
            received_stats: None,
            bytes: 64,
            packet_meta: PacketMeta::default(),
        },
        Op::Duplicate => ClientEvent::DuplicateReply {
            seq,
            remote: ADDR.parse().unwrap(),
            received_at: ts_at(base, clock_ms),
            bytes: 64,
        },
        Op::Loss => ClientEvent::EchoLoss {
            seq,
            sent_at: ts_at(base, clock_ms),
            timeout_at: ts_at(base, clock_ms).mono - Duration::from_millis(10),
        },
        Op::DelayedLoss { age_ms } => {
            let deadline_ms = clock_ms.saturating_sub(u64::from(age_ms));
            ClientEvent::EchoLoss {
                seq,
                sent_at: ts_at(base, deadline_ms.saturating_sub(3)),
                timeout_at: ts_at(base, deadline_ms).mono,
            }
        }
        Op::Warning => ClientEvent::Warning {
            kind: irtt_client::WarningKind::UntrackedReply,
            message: "generated".to_owned(),
            at: ts_at(base, clock_ms),
        },
    }
}

/// Public reference: collect exactly the events retained by one window.
fn replay(events: &[ClientEvent], late_replies: LateReplyMode) -> Snapshot {
    let mut collector = StatsCollector::new(StatsConfig {
        samples: SampleMode::RunningOnly,
        rolling_count: None,
        rolling_time: None,
        late_replies,
    });
    for event in events {
        collector.process(event);
    }
    collector.snapshot()
}

/// Retain ordinary metrics through replay, but derive directional loss directly
/// from the full public event history, including the discarded prefix.
fn rolling_reference(
    history: &[(ClientEvent, u64)],
    retain: impl Fn(usize, u64) -> bool,
    late_replies: LateReplyMode,
) -> Snapshot {
    let retained: Vec<ClientEvent> = history
        .iter()
        .enumerate()
        .filter(|(index, (_, at))| retain(*index, *at))
        .map(|(_, (event, _))| event.clone())
        .collect();
    let mut snapshot = replay(&retained, late_replies);
    let packets: Vec<_> = history
        .iter()
        .enumerate()
        .filter(|(_, (event, _))| {
            matches!(
                event,
                ClientEvent::EchoSent { .. }
                    | ClientEvent::EchoReply { .. }
                    | ClientEvent::LateReply { .. }
                    | ClientEvent::DuplicateReply { .. }
            )
        })
        .collect();
    let selected: Vec<_> = packets
        .iter()
        .enumerate()
        .filter(|(_, (index, (_, at)))| retain(*index, *at))
        .collect();
    let count = |event: &ClientEvent| match event {
        ClientEvent::EchoReply { received_stats, .. }
        | ClientEvent::LateReply {
            sent_at: Some(_),
            rtt: Some(_),
            received_stats,
            ..
        } => received_stats.and_then(|stats| stats.count).map(u64::from),
        _ => None,
    };
    let delta = selected
        .first()
        .zip(selected.last())
        .and_then(|((first, _), (last, _))| {
            // Timestamp filtering must not omit a packet between the endpoints.
            if last - first + 1 != selected.len() {
                return None;
            }
            // Partition the full observed history at ambiguous discontinuities.
            // This independent oracle keeps observation lists for each segment
            // rather than relying on the collector's retained metadata.
            let mut observations = Vec::new();
            let mut segment_start = 0;
            for (index, (_, (event, _))) in packets[..=*last].iter().enumerate() {
                if let Some(count) = count(event) {
                    if observations
                        .iter()
                        .max()
                        .is_some_and(|highest: &u64| highest.abs_diff(count) >= (1_u64 << 31))
                    {
                        segment_start = index + 1;
                        observations.clear();
                    }
                    observations.push(count);
                }
            }
            if segment_start > *first {
                return None;
            }
            let baseline = if *first == 0 {
                Some(0)
            } else {
                // The immediately preceding packet must itself report the
                // highest count. Earlier observations cannot bound this window.
                let (_, (previous_event, _)) = packets[*first - 1];
                let previous = count(previous_event)?;
                if packets[segment_start.saturating_sub(1)..*first - 1]
                    .iter()
                    .filter_map(|(_, (event, _))| count(event))
                    .max()
                    .is_some_and(|highest| highest > previous)
                {
                    return None;
                }
                Some(previous)
            }?;
            let endpoint = observations.into_iter().max()?;
            Some(endpoint - baseline)
        });
    snapshot.loss.upstream_loss_packets =
        delta.map(|delta| i128::from(snapshot.packets.packets_sent) - i128::from(delta));
    snapshot.loss.downstream_loss_packets =
        delta.map(|delta| i128::from(delta) - i128::from(snapshot.packets.packets_received));
    snapshot.loss.upstream_loss_percent = if snapshot.packets.packets_sent == 0 {
        0.0
    } else {
        snapshot.loss.upstream_loss_packets.map_or(0.0, |loss| {
            100.0 * loss as f64 / snapshot.packets.packets_sent as f64
        })
    };
    snapshot.loss.downstream_loss_percent = match delta {
        Some(delta) if delta != 0 => {
            100.0 * snapshot.loss.downstream_loss_packets.unwrap() as f64 / delta as f64
        }
        _ => 0.0,
    };
    snapshot
}

#[test]
fn collector_equality_ignores_expired_time_window_bookkeeping() {
    let base = Instant::now();
    let config = StatsConfig {
        rolling_time: Some(Duration::from_millis(3)),
        ..StatsConfig::continuous()
    };
    let mut left = StatsCollector::new(config);
    let mut right = StatsCollector::new(config);
    for at_ms in [1, 2, 4, 5, 6] {
        left.process(&build_event(&Op::Warning, 0, at_ms, base));
    }
    for at_ms in [1, 4, 2, 5, 6] {
        right.process(&build_event(&Op::Warning, 0, at_ms, base));
    }
    // Different expired histories leave the same live events in arrival order.
    assert_eq!(left.snapshot(), right.snapshot());
    assert_eq!(left.rolling_time(), right.rolling_time());
    assert_eq!(left, right);
}

#[test]
fn time_expiry_preserves_arrival_order_for_server_receive_window() {
    let base = Instant::now();
    let mut collector = StatsCollector::new(StatsConfig {
        rolling_time: Some(Duration::from_millis(3)),
        ..StatsConfig::continuous()
    });
    for (seq, at_ms, window) in [(1, 5, 0x5), (0, 4, 0x4)] {
        let mut event = build_event(&Op::Reply { raw_ms: 1 }, seq, at_ms, base);
        let ClientEvent::EchoReply { received_stats, .. } = &mut event else {
            unreachable!();
        };
        received_stats.as_mut().unwrap().window = Some(window);
        collector.process(&event);
    }
    // The last arrival owns the observation even with an older receive time.
    assert_eq!(
        collector
            .rolling_time()
            .unwrap()
            .packets
            .server_received_window,
        Some(0x4)
    );
    collector.process(&build_event(&Op::Warning, 2, 8, base));
    let rolling = collector.rolling_time().unwrap();
    assert_eq!(rolling.rtt.primary.count, 1);
    assert_eq!(rolling.packets.server_received_window, Some(0x5));
    // The hidden, expired observation remains in cumulative accounting.
    assert_eq!(
        collector.snapshot().packets.server_received_window,
        Some(0x4)
    );
    collector.process(&build_event(&Op::Warning, 3, 9, base));
    let rolling = collector.rolling_time().unwrap();
    assert_eq!(rolling.rtt.primary.count, 0);
    assert_eq!(rolling.packets.server_received_window, None);
}

proptest! {
    #![proptest_config(ProptestConfig { cases: 128, .. ProptestConfig::default() })]

    /// With `rolling_count` and `rolling_time` configured simultaneously,
    /// each window must match a reference model that applies *only its own*
    /// bound to the full event history, re-filtered from scratch after every
    /// operation. Any accidental AND/OR coupling between the two bounds (for
    /// example the count window also dropping events outside the time
    /// window, or vice versa) would show up here as a mismatch.
    ///
    /// The cumulative snapshot is checked too: it must equal a reference
    /// that replays the *entire* history, proving rolling eviction never
    /// reaches into the cumulative accounting.
    #[test]
    fn rolling_count_and_rolling_time_windows_are_independent(
        ops in prop::collection::vec(op_strategy(), 1..40),
        count_limit in 1usize..6,
        time_limit_ms in 0u64..80,
        count_only in prop::bool::ANY,
    ) {
        let late_replies = if count_only {
            LateReplyMode::CountOnly
        } else {
            LateReplyMode::Measure
        };
        let config = StatsConfig {
            samples: SampleMode::RunningOnly,
            rolling_count: Some(count_limit),
            rolling_time: Some(Duration::from_millis(time_limit_ms)),
            late_replies,
        };
        let mut collector = StatsCollector::new(config);
        let base = Instant::now();
        let mut clock_ms: u64 = 0;
        // Full, unfiltered event history alongside each event's windowing
        // timestamp, so the reference windows below can be recomputed from
        // scratch after every step.
        let mut history: Vec<(ClientEvent, u64)> = Vec::new();

        // Force a loss whose clamped time evicts the initial send, while its
        // raw pre-send deadline would retain it. Every generated case must
        // therefore distinguish the two normalization choices.
        let ops = [Op::Send, Op::Advance(u16::try_from(time_limit_ms + 5).unwrap()), Op::Loss, Op::Advance(10), Op::Warning]
            .into_iter().chain(ops).collect::<Vec<_>>();
        for (idx, op) in ops.iter().enumerate() {
            if let Op::Advance(delta) = op {
                clock_ms += u64::from(*delta);
            } else {
                let seq = u32::try_from(idx).unwrap();
                let event = build_event(op, seq, clock_ms, base);
                collector.process(&event);
                let at_ms = match op {
                    Op::DelayedLoss { age_ms } | Op::DelayedReply { age_ms, .. } => clock_ms.saturating_sub(u64::from(*age_ms)),
                    _ => clock_ms,
                };
                history.push((event, at_ms));
            }

            // Cumulative snapshot: unaffected by rolling eviction, so it must
            // equal a fresh replay of the entire history so far.
            let full: Vec<ClientEvent> = history.iter().map(|(event, _)| event.clone()).collect();
            prop_assert_eq!(
                collector.snapshot(),
                replay(&full, late_replies),
                "cumulative snapshot diverged from a full replay after op {}", idx
            );

            // Count window reference: the last `count_limit` retained
            // events, ignoring the time bound entirely.
            let count_start = history.len().saturating_sub(count_limit);
            prop_assert_eq!(
                collector.rolling_count(),
                Some(rolling_reference(&history, |index, _| index >= count_start, late_replies)),
                "count window diverged from the count-only reference after op {}", idx
            );

            // Time window reference: every retained event whose timestamp is
            // within `time_limit_ms` of the maximum observed event timestamp,
            // ignoring the count bound entirely. Delayed losses cannot move
            // this anchor backwards.
            if let Some(latest_at_ms) = history.iter().map(|(_, at_ms)| at_ms).max() {
                let cutoff = latest_at_ms.checked_sub(time_limit_ms);
                prop_assert_eq!(
                    collector.rolling_time(),
                    Some(rolling_reference(&history, |_, at_ms| cutoff.is_none_or(|cutoff| at_ms >= cutoff), late_replies)),
                    "time window diverged from the time-only reference after op {}", idx
                );
            } else {
                prop_assert_eq!(
                    collector.rolling_time(),
                    Some(replay(&[], late_replies)),
                    "time window should be empty before any event after op {}", idx
                );
            }
        }
    }
}