rusty-telemetry/src/feed_state.rs

// rusty-telemetry — Sim racing telemetry analysis platform
// Copyright (C) 2026  Jan-Peter v. Hunnius
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
use crate::telemetry_frame::TelemetryFrame;
use std::time::Instant;

const RING_CAPACITY: usize = 50;
const STALE_THRESHOLD_SECS: f64 = 5.0;
const MIN_TIME_SPAN_SECS: f64 = 0.001;

#[derive(Clone, Copy, Debug, PartialEq)]
pub enum FeedHealth {
    Waiting,
    Alive,
    Stale,
    Error,
}

#[derive(Clone, Copy, Debug, PartialEq)]
pub enum HandshakePhase {
    NotStarted,
    WaitingForResponse,
    Subscribing,
    Subscribed,
}

#[derive(Clone, Copy, Debug, PartialEq)]
pub enum FeedMode {
    BindListen,
    Handshake {
        target_port: u16,
        recv_port: u16,
    },
    Disabled,
}

#[derive(Clone)]
pub struct FeedState {
    pub health: FeedHealth,
    pub handshake_phase: HandshakePhase,
    pub status_text: String,
    pub total_packets: u64,
    pub total_bytes: u64,
    pub frequency_hz: f64,
    pub data_rate_kb_s: f64,
    pub last_packet_ago_secs: Option<f64>,
    pub start_time: Option<Instant>,
    pub last_packet_time: Option<Instant>,
    pub packet_timestamps: [Instant; RING_CAPACITY],
    pub ring_index: usize,
    pub ring_count: usize,
    pub last_frame: Option<TelemetryFrame>,
}

impl FeedState {
    pub fn new() -> Self {
        Self {
            health: FeedHealth::Waiting,
            handshake_phase: HandshakePhase::NotStarted,
            status_text: String::from("Bound, waiting for data"),
            total_packets: 0,
            total_bytes: 0,
            frequency_hz: 0.0,
            data_rate_kb_s: 0.0,
            last_packet_ago_secs: None,
            start_time: None,
            last_packet_time: None,
            packet_timestamps: [Instant::now(); RING_CAPACITY],
            ring_index: 0,
            ring_count: 0,
            last_frame: None,
        }
    }

    pub fn update_frame(&mut self, frame: TelemetryFrame) {
        self.last_frame = Some(frame);
    }

    pub fn record_packet(&mut self, len: usize) {
        let now = Instant::now();
        if self.start_time.is_none() {
            self.start_time = Some(now);
        }

        self.total_packets += 1;
        self.total_bytes += len as u64;
        self.last_packet_time = Some(now);
        self.last_packet_ago_secs = Some(0.0);

        if self.ring_count < RING_CAPACITY {
            self.packet_timestamps[self.ring_count] = now;
            self.ring_count += 1;
            self.ring_index = self.ring_count % RING_CAPACITY;
        } else {
            self.packet_timestamps[self.ring_index] = now;
            self.ring_index = (self.ring_index + 1) % RING_CAPACITY;
        }

        self.frequency_hz = self.calc_frequency();
        self.data_rate_kb_s = self.calc_data_rate();
        self.health = FeedHealth::Alive;
        if self.status_text != "Receiving" {
            self.status_text = String::from("Receiving");
        }
    }

    fn calc_frequency(&self) -> f64 {
        if self.ring_count < 2 {
            return 0.0;
        }

        let count = self.ring_count.min(RING_CAPACITY);
        let oldest_idx = if self.ring_count >= RING_CAPACITY {
            self.ring_index
        } else {
            0
        };
        let newest_idx = if self.ring_count >= RING_CAPACITY {
            if self.ring_index == 0 {
                RING_CAPACITY - 1
            } else {
                self.ring_index - 1
            }
        } else {
            self.ring_count - 1
        };

        let oldest = self.packet_timestamps[oldest_idx];
        let newest = self.packet_timestamps[newest_idx];
        let span = newest.duration_since(oldest).as_secs_f64();

        if span < MIN_TIME_SPAN_SECS {
            return 0.0;
        }

        count as f64 / span
    }

    fn calc_data_rate(&self) -> f64 {
        match self.start_time {
            Some(start) => {
                let elapsed = start.elapsed().as_secs_f64();
                if elapsed > 0.0 {
                    (self.total_bytes as f64 / 1024.0) / elapsed
                } else {
                    0.0
                }
            }
            None => 0.0,
        }
    }

    pub fn update_health(&mut self) {
        match self.last_packet_ago_secs {
            None => {
                if self.health != FeedHealth::Error {
                    self.health = FeedHealth::Waiting;
                    self.status_text = String::from("Waiting for data");
                }
            }
            Some(ago) => {
                if ago < STALE_THRESHOLD_SECS {
                    self.health = FeedHealth::Alive;
                    self.status_text = String::from("Receiving");
                } else {
                    self.health = FeedHealth::Stale;
                    self.status_text = format!("Stale ({:.1}s since last packet)", ago);
                }
            }
        }
    }

    pub fn tick(&mut self) {
        if let Some(last) = self.last_packet_time {
            self.last_packet_ago_secs = Some(last.elapsed().as_secs_f64());
        }
        self.frequency_hz = self.calc_frequency();
        self.data_rate_kb_s = self.calc_data_rate();
        self.update_health();
    }

    pub fn set_handshake_phase(&mut self, phase: HandshakePhase) {
        self.handshake_phase = phase;
    }

    pub fn set_error(&mut self, msg: &str) {
        self.health = FeedHealth::Error;
        self.status_text = format!("Error: {}", msg);
    }
}

impl Default for FeedState {
    fn default() -> Self {
        Self::new()
    }
}

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

    #[test]
    fn test_new_state_is_waiting() {
        let state = FeedState::new();
        assert_eq!(state.health, FeedHealth::Waiting);
        assert_eq!(state.handshake_phase, HandshakePhase::NotStarted);
        assert_eq!(state.total_packets, 0);
        assert_eq!(state.total_bytes, 0);
        assert_eq!(state.frequency_hz, 0.0);
        assert!(state.last_packet_ago_secs.is_none());
        assert!(state.last_frame.is_none());
    }

    #[test]
    fn test_record_packet_increments_counters() {
        let mut state = FeedState::new();
        state.record_packet(100);
        assert_eq!(state.total_packets, 1);
        assert_eq!(state.total_bytes, 100);
        assert_eq!(state.health, FeedHealth::Alive);
        assert_eq!(state.status_text, "Receiving");

        state.record_packet(200);
        assert_eq!(state.total_packets, 2);
        assert_eq!(state.total_bytes, 300);
    }

    #[test]
    fn test_frequency_zero_with_fewer_than_two_packets() {
        let mut state = FeedState::new();
        state.record_packet(100);
        assert_eq!(state.frequency_hz, 0.0);

        // With 2 packets at the same instant, span is < MIN_TIME_SPAN_SECS, so still 0
        state.record_packet(100);
        // frequency could be 0 if the two packets are within MIN_TIME_SPAN_SECS
        // This is expected behavior
    }

    #[test]
    fn test_frequency_nonzero_after_spaced_packets() {
        let mut state = FeedState::new();
        for _ in 0..10 {
            state.record_packet(100);
            thread::sleep(std::time::Duration::from_millis(10));
        }
        // 10 packets over ~90ms → frequency should be > 0
        assert!(state.frequency_hz > 0.0);
        // Rough sanity: 10 packets / 0.09s ≈ 111 Hz, but with overhead, just check > 0
        assert!(state.frequency_hz < 10_000.0); // sanity upper bound
    }

    #[test]
    fn test_ring_buffer_fills_without_wraparound() {
        let mut state = FeedState::new();
        for _ in 0..RING_CAPACITY {
            state.record_packet(50);
        }
        assert_eq!(state.ring_count, RING_CAPACITY);
        assert_eq!(state.ring_index, 0); // wraps to 0 after filling
    }

    #[test]
    fn test_ring_buffer_wraps_around() {
        let mut state = FeedState::new();
        for _ in 0..RING_CAPACITY {
            state.record_packet(50);
        }
        assert_eq!(state.ring_index, 0);

        // Add more packets to trigger wraparound
        for _ in 0..10 {
            state.record_packet(50);
        }
        assert_eq!(state.ring_count, RING_CAPACITY);
        assert_eq!(state.ring_index, 10);
        assert_eq!(state.total_packets, (RING_CAPACITY + 10) as u64);
    }

    #[test]
    fn test_health_waiting_to_alive() {
        let mut state = FeedState::new();
        assert_eq!(state.health, FeedHealth::Waiting);

        state.record_packet(100);
        assert_eq!(state.health, FeedHealth::Alive);
    }

    #[test]
    fn test_health_alive_to_stale() {
        let mut state = FeedState::new();
        state.record_packet(100);
        assert_eq!(state.health, FeedHealth::Alive);

        // Simulate time passing by manually setting last_packet_ago_secs
        state.last_packet_ago_secs = Some(6.0);
        state.update_health();
        assert_eq!(state.health, FeedHealth::Stale);
        assert!(state.status_text.starts_with("Stale"));
    }

    #[test]
    fn test_health_stale_to_alive() {
        let mut state = FeedState::new();
        state.record_packet(100);
        state.last_packet_ago_secs = Some(6.0);
        state.update_health();
        assert_eq!(state.health, FeedHealth::Stale);

        // New packet brings it back to alive
        state.record_packet(100);
        assert_eq!(state.health, FeedHealth::Alive);
    }

    #[test]
    fn test_health_error_is_sticky_via_update() {
        let mut state = FeedState::new();
        state.set_error("test error");
        assert_eq!(state.health, FeedHealth::Error);

        // update_health should not override Error when ago is None
        state.last_packet_ago_secs = None;
        state.update_health();
        assert_eq!(state.health, FeedHealth::Error);
    }

    #[test]
    fn test_tick_updates_ago_secs() {
        let mut state = FeedState::new();

        // Before any packet, tick should not set ago
        state.tick();
        assert!(state.last_packet_ago_secs.is_none());

        state.record_packet(100);
        // Immediately after, ago should be ~0
        state.tick();
        assert!(state.last_packet_ago_secs.is_some());
        assert!(state.last_packet_ago_secs.unwrap() < 1.0);

        // After sleeping, ago should be > 0
        thread::sleep(std::time::Duration::from_millis(100));
        state.tick();
        let ago = state.last_packet_ago_secs.unwrap();
        assert!(ago >= 0.09, "expected ago >= 0.09, got {}", ago);
    }

    #[test]
    fn test_data_rate_calculation() {
        let mut state = FeedState::new();
        state.record_packet(1024);
        thread::sleep(std::time::Duration::from_millis(50));
        state.record_packet(1024);

        // 2048 bytes over ~50ms → roughly 40 KB/s
        state.data_rate_kb_s = state.calc_data_rate();
        assert!(state.data_rate_kb_s > 0.0);
    }

    #[test]
    fn test_update_frame() {
        let mut state = FeedState::new();
        assert!(state.last_frame.is_none());

        let frame = TelemetryFrame::default();
        state.update_frame(frame);
        assert!(state.last_frame.is_some());
    }

    #[test]
    fn test_set_handshake_phase() {
        let mut state = FeedState::new();
        assert_eq!(state.handshake_phase, HandshakePhase::NotStarted);

        state.set_handshake_phase(HandshakePhase::WaitingForResponse);
        assert_eq!(state.handshake_phase, HandshakePhase::WaitingForResponse);

        state.set_handshake_phase(HandshakePhase::Subscribed);
        assert_eq!(state.handshake_phase, HandshakePhase::Subscribed);
    }

    #[test]
    fn test_set_error() {
        let mut state = FeedState::new();
        state.set_error("bind failed");
        assert_eq!(state.health, FeedHealth::Error);
        assert_eq!(state.status_text, "Error: bind failed");
    }
}