//! Bounded receive-rate aggregation for the browser's MAVLink monitor.

use std::{
    collections::BTreeMap,
    time::{Duration, Instant},
};

use serde::Serialize;

const RATE_WINDOW: Duration = Duration::from_secs(1);
const MAX_PACKET_KINDS: usize = 128;

/// Identity and decoded value of one CRC-valid frame received from the flight controller.
#[derive(Clone, Debug)]
pub struct PacketObservation {
    pub system_id: u8,
    pub component_id: u8,
    pub message_id: u32,
    pub name: &'static str,
    pub values: serde_json::Value,
}

/// Read-only browser projection of the bounded packet aggregate.
#[derive(Clone, Debug, Serialize)]
pub struct PacketSnapshot {
    pub packets: Vec<PacketView>,
}

/// One source/message tuple observed on the receive link.
#[derive(Clone, Debug, Serialize)]
pub struct PacketView {
    pub system_id: u8,
    pub component_id: u8,
    pub message_id: u32,
    pub name: &'static str,
    pub count: u64,
    pub frequency_hz: f64,
    /// Decoded fields from the most recently received frame of this identity.
    pub values: serde_json::Value,
}

#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
struct PacketKey {
    system_id: u8,
    component_id: u8,
    message_id: u32,
}

struct PacketAggregate {
    name: &'static str,
    count: u64,
    window_count: u64,
    frequency_hz: f64,
    last_seen: Instant,
    values: serde_json::Value,
}

#[derive(Default)]
pub struct PacketMonitor {
    entries: BTreeMap<PacketKey, PacketAggregate>,
    window_started: Option<Instant>,
}

impl PacketMonitor {
    /// Records a serial receive batch without retaining individual frames.
    pub fn observe(
        &mut self,
        observations: impl IntoIterator<Item = PacketObservation>,
        now: Instant,
    ) {
        self.roll_window(now);
        for observation in observations {
            let key = PacketKey {
                system_id: observation.system_id,
                component_id: observation.component_id,
                message_id: observation.message_id,
            };
            if !self.entries.contains_key(&key) && self.entries.len() == MAX_PACKET_KINDS {
                self.evict_oldest();
            }
            let entry = self.entries.entry(key).or_insert(PacketAggregate {
                name: observation.name,
                count: 0,
                window_count: 0,
                frequency_hz: 0.0,
                last_seen: now,
                values: serde_json::Value::Null,
            });
            entry.name = observation.name;
            entry.count = entry.count.saturating_add(1);
            entry.window_count = entry.window_count.saturating_add(1);
            entry.last_seen = now;
            entry.values = observation.values;
        }
    }

    /// Returns the current aggregate and closes any elapsed measurement window.
    pub fn snapshot(&mut self, now: Instant) -> PacketSnapshot {
        self.roll_window(now);
        PacketSnapshot {
            packets: self
                .entries
                .iter()
                .map(|(key, entry)| PacketView {
                    system_id: key.system_id,
                    component_id: key.component_id,
                    message_id: key.message_id,
                    name: entry.name,
                    count: entry.count,
                    frequency_hz: entry.frequency_hz,
                    values: entry.values.clone(),
                })
                .collect(),
        }
    }

    pub fn clear(&mut self) {
        self.entries.clear();
        self.window_started = None;
    }

    fn roll_window(&mut self, now: Instant) {
        let Some(started) = self.window_started else {
            self.window_started = Some(now);
            return;
        };
        let elapsed = now.saturating_duration_since(started);
        if elapsed < RATE_WINDOW {
            return;
        }
        let seconds = elapsed.as_secs_f64();
        for entry in self.entries.values_mut() {
            entry.frequency_hz = entry.window_count as f64 / seconds;
            entry.window_count = 0;
        }
        self.window_started = Some(now);
    }

    fn evict_oldest(&mut self) {
        let oldest = self
            .entries
            .iter()
            .min_by_key(|(_, entry)| entry.last_seen)
            .map(|(key, _)| *key);
        if let Some(key) = oldest {
            self.entries.remove(&key);
        }
    }
}

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

    fn packet(message_id: u32, name: &'static str) -> PacketObservation {
        PacketObservation {
            system_id: 42,
            component_id: 7,
            message_id,
            name,
            values: serde_json::json!({"sequence": message_id}),
        }
    }

    #[test]
    fn aggregates_counts_and_rates_without_packet_history() {
        let start = Instant::now();
        let mut monitor = PacketMonitor::default();
        monitor.observe([packet(0, "HEARTBEAT"), packet(0, "HEARTBEAT")], start);
        monitor.observe(
            [
                packet(0, "HEARTBEAT"),
                packet(0, "HEARTBEAT"),
                packet(0, "HEARTBEAT"),
            ],
            start + Duration::from_millis(500),
        );

        let snapshot = monitor.snapshot(start + RATE_WINDOW);
        assert_eq!(snapshot.packets.len(), 1);
        assert_eq!(snapshot.packets[0].count, 5);
        assert_eq!(snapshot.packets[0].frequency_hz, 5.0);
        assert_eq!(snapshot.packets[0].values["sequence"], 0);
    }

    #[test]
    fn reports_zero_after_a_window_with_no_packets() {
        let start = Instant::now();
        let mut monitor = PacketMonitor::default();
        monitor.observe([packet(0, "HEARTBEAT")], start);
        assert_eq!(
            monitor.snapshot(start + RATE_WINDOW).packets[0].frequency_hz,
            1.0
        );
        assert_eq!(
            monitor.snapshot(start + RATE_WINDOW + RATE_WINDOW).packets[0].frequency_hz,
            0.0
        );
    }

    #[test]
    fn retains_only_the_latest_decoded_values_for_each_identity() {
        let start = Instant::now();
        let mut first = packet(51021, "KAPTAN_ERROR");
        first.values = serde_json::json!({"path": "first"});
        let mut second = packet(51021, "KAPTAN_ERROR");
        second.values = serde_json::json!({"path": "second"});
        let mut monitor = PacketMonitor::default();

        monitor.observe([first], start);
        monitor.observe([second], start + Duration::from_millis(1));

        let snapshot = monitor.snapshot(start + Duration::from_millis(1));
        assert_eq!(snapshot.packets.len(), 1);
        assert_eq!(snapshot.packets[0].count, 2);
        assert_eq!(snapshot.packets[0].values["path"], "second");
    }

    #[test]
    fn evicts_the_oldest_identity_at_the_fixed_capacity() {
        let start = Instant::now();
        let mut monitor = PacketMonitor::default();
        for message_id in 0..u32::try_from(MAX_PACKET_KINDS).unwrap() {
            monitor.observe([packet(message_id, "KNOWN")], start);
        }
        monitor.observe([packet(99_999, "NEW")], start + Duration::from_millis(1));

        let snapshot = monitor.snapshot(start + Duration::from_millis(1));
        assert_eq!(snapshot.packets.len(), MAX_PACKET_KINDS);
        assert!(!snapshot.packets.iter().any(|packet| packet.message_id == 0));
        assert!(
            snapshot
                .packets
                .iter()
                .any(|packet| packet.message_id == 99_999)
        );
    }

    #[test]
    fn clear_starts_a_new_link_session() {
        let start = Instant::now();
        let mut monitor = PacketMonitor::default();
        monitor.observe([packet(0, "HEARTBEAT")], start);
        monitor.clear();
        assert!(monitor.snapshot(start + RATE_WINDOW).packets.is_empty());
    }
}