bcms 0.1.12

use crate::data::IoTData;
use std::collections::HashMap;
use std::time::Duration;

/// Tracks per-device submission state for fault-tolerant data submission.
///
/// Each device maintains its own last-submitted timestamp, preventing:
/// - Clock skew on one device from affecting others
/// - Unregistered devices from blocking registered device progress
/// - Duplicate submissions caused by global timestamp rollback
#[derive(Debug, Default)]
pub struct SubmissionTracker {
    /// Per-device last successfully submitted timestamp
    timestamps: HashMap<String, i64>,
    /// Per-device consecutive failure count for backoff
    failures: HashMap<String, u32>,
}

impl SubmissionTracker {
    pub fn new() -> Self {
        Self::default()
    }

    /// Get the last submitted timestamp for a device (0 if never submitted)
    pub fn get_device_timestamp(&self, address: &str) -> i64 {
        self.timestamps.get(address).copied().unwrap_or(0)
    }

    /// Get all device timestamps
    pub fn get_all_timestamps(&self) -> &HashMap<String, i64> {
        &self.timestamps
    }

    /// Filter data to only include new data per device.
    /// Each device is filtered against its own last-submitted timestamp.
    pub fn filter_new_data(&self, data: Vec<IoTData>) -> Vec<IoTData> {
        data.into_iter()
            .filter(|d| {
                let last_ts = self.get_device_timestamp(d.address());
                d.timestamp() > last_ts
            })
            .collect()
    }

    /// Update timestamps after successful submission.
    /// Only updates timestamps for devices that were actually submitted.
    pub fn mark_submitted(&mut self, submitted: &[&IoTData]) {
        // Group by device and find max timestamp per device
        let mut max_timestamps: HashMap<&str, i64> = HashMap::new();

        for data in submitted {
            let addr = data.address();
            let ts = data.timestamp();
            max_timestamps
                .entry(addr)
                .and_modify(|existing| *existing = (*existing).max(ts))
                .or_insert(ts);
        }

        // Update stored timestamps
        for (addr, ts) in max_timestamps {
            self.timestamps
                .entry(addr.to_string())
                .and_modify(|existing| *existing = (*existing).max(ts))
                .or_insert(ts);
        }
    }

    /// Mark a device submission as failed (increments failure counter)
    pub fn mark_failed(&mut self, address: &str) {
        *self.failures.entry(address.to_string()).or_insert(0) += 1;
    }

    /// Mark all devices in a batch as failed
    pub fn mark_batch_failed(&mut self, addresses: &[&str]) {
        for addr in addresses {
            self.mark_failed(addr);
        }
    }

    /// Reset failure count for a device (call on successful submission)
    pub fn mark_recovered(&mut self, address: &str) {
        self.failures.remove(address);
    }

    /// Reset failure count for all devices in a batch
    pub fn mark_batch_recovered(&mut self, addresses: &[&str]) {
        for addr in addresses {
            self.mark_recovered(addr);
        }
    }

    /// Get consecutive failure count for a device
    pub fn get_failure_count(&self, address: &str) -> u32 {
        self.failures.get(address).copied().unwrap_or(0)
    }

    /// Get total consecutive failures across all devices (for global backoff)
    pub fn get_total_failures(&self) -> u32 {
        self.failures.values().sum()
    }

    /// Check if any device has failures
    pub fn has_failures(&self) -> bool {
        !self.failures.is_empty()
    }

    /// Calculate backoff duration for a device using exponential backoff
    pub fn get_device_backoff(&self, address: &str, base_interval_secs: f64, max_backoff_secs: f64) -> Duration {
        let failures = self.get_failure_count(address);
        if failures == 0 {
            return Duration::from_secs(0);
        }
        let backoff_secs = (base_interval_secs * 2f64.powi(failures as i32)).min(max_backoff_secs);
        Duration::from_secs_f64(backoff_secs)
    }

    /// Calculate global backoff based on max failures across all devices
    pub fn get_global_backoff(&self, base_interval_secs: f64, max_backoff_secs: f64) -> Duration {
        let max_failures = self.failures.values().copied().max().unwrap_or(0);
        if max_failures == 0 {
            return Duration::from_secs(0);
        }
        let backoff_secs = (base_interval_secs * 2f64.powi(max_failures as i32)).min(max_backoff_secs);
        Duration::from_secs_f64(backoff_secs)
    }

    /// Clear all state (for testing or reset)
    pub fn clear(&mut self) {
        self.timestamps.clear();
        self.failures.clear();
    }

    /// Get statistics for monitoring
    pub fn stats(&self) -> SubmissionStats {
        SubmissionStats {
            tracked_devices: self.timestamps.len(),
            devices_with_failures: self.failures.len(),
            total_failures: self.get_total_failures(),
            max_failures: self.failures.values().copied().max().unwrap_or(0),
        }
    }
}

/// Statistics about submission tracking state
#[derive(Debug, Clone)]
pub struct SubmissionStats {
    pub tracked_devices: usize,
    pub devices_with_failures: usize,
    pub total_failures: u32,
    pub max_failures: u32,
}

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

    fn create_test_data(address: &str, timestamp: i64) -> IoTData {
        helpers::create_blood_pressure_data(address.to_string(), 120, 80, timestamp)
    }

    #[test]
    fn test_new_tracker_has_zero_timestamps() {
        let tracker = SubmissionTracker::new();
        assert_eq!(tracker.get_device_timestamp("device1"), 0);
        assert_eq!(tracker.get_device_timestamp("unknown"), 0);
    }

    #[test]
    fn test_filter_new_data_all_new() {
        let tracker = SubmissionTracker::new();
        let data = vec![
            create_test_data("device1", 100),
            create_test_data("device2", 200),
        ];

        let filtered = tracker.filter_new_data(data);
        assert_eq!(filtered.len(), 2);
    }

    #[test]
    fn test_filter_new_data_excludes_already_submitted() {
        let mut tracker = SubmissionTracker::new();

        // Mark device1 as submitted up to timestamp 150
        let submitted = vec![create_test_data("device1", 150)];
        let refs: Vec<&IoTData> = submitted.iter().collect();
        tracker.mark_submitted(&refs);

        let data = vec![
            create_test_data("device1", 100),  // Should be filtered (100 <= 150)
            create_test_data("device1", 150),  // Should be filtered (150 <= 150, strict >)
            create_test_data("device1", 200),  // Should pass (200 > 150)
            create_test_data("device2", 50),   // Should pass (device2 has no history)
        ];

        let filtered = tracker.filter_new_data(data);
        assert_eq!(filtered.len(), 2);
        assert!(filtered.iter().any(|d| d.address() == "device1" && d.timestamp() == 200));
        assert!(filtered.iter().any(|d| d.address() == "device2" && d.timestamp() == 50));
    }

    #[test]
    fn test_mark_submitted_updates_max_timestamp() {
        let mut tracker = SubmissionTracker::new();

        let data = vec![
            create_test_data("device1", 100),
            create_test_data("device1", 200),
            create_test_data("device1", 150),  // Out of order
        ];
        let refs: Vec<&IoTData> = data.iter().collect();
        tracker.mark_submitted(&refs);

        // Should have max timestamp (200)
        assert_eq!(tracker.get_device_timestamp("device1"), 200);
    }

    #[test]
    fn test_mark_submitted_independent_per_device() {
        let mut tracker = SubmissionTracker::new();

        let data = vec![
            create_test_data("device1", 100),
            create_test_data("device2", 500),
        ];
        let refs: Vec<&IoTData> = data.iter().collect();
        tracker.mark_submitted(&refs);

        assert_eq!(tracker.get_device_timestamp("device1"), 100);
        assert_eq!(tracker.get_device_timestamp("device2"), 500);
    }

    #[test]
    fn test_clock_skew_doesnt_lose_data() {
        let mut tracker = SubmissionTracker::new();

        // Device A submits at t=1000
        let data_a = vec![create_test_data("deviceA", 1000)];
        let refs: Vec<&IoTData> = data_a.iter().collect();
        tracker.mark_submitted(&refs);

        // Device B has clock skew, collecting at t=500
        let new_data = vec![
            create_test_data("deviceA", 1100),  // New for A
            create_test_data("deviceB", 500),   // Device B with "old" timestamp
        ];

        let filtered = tracker.filter_new_data(new_data);

        // Both should pass - device B has no submission history
        assert_eq!(filtered.len(), 2);
        assert!(filtered.iter().any(|d| d.address() == "deviceA"));
        assert!(filtered.iter().any(|d| d.address() == "deviceB"));
    }

    #[test]
    fn test_partial_submission_only_updates_submitted() {
        let mut tracker = SubmissionTracker::new();

        // Only device1 data was submitted (device2 might be unregistered)
        let submitted = vec![create_test_data("device1", 100)];
        let refs: Vec<&IoTData> = submitted.iter().collect();
        tracker.mark_submitted(&refs);

        // Device1 is tracked, device2 is not
        assert_eq!(tracker.get_device_timestamp("device1"), 100);
        assert_eq!(tracker.get_device_timestamp("device2"), 0);

        // Next cycle: device2 data should still be included
        let new_data = vec![
            create_test_data("device1", 50),   // Filtered (50 < 100)
            create_test_data("device2", 50),   // Passes (no history)
        ];

        let filtered = tracker.filter_new_data(new_data);
        assert_eq!(filtered.len(), 1);
        assert_eq!(filtered[0].address(), "device2");
    }

    #[test]
    fn test_failure_tracking() {
        let mut tracker = SubmissionTracker::new();

        assert_eq!(tracker.get_failure_count("device1"), 0);
        assert!(!tracker.has_failures());

        tracker.mark_failed("device1");
        assert_eq!(tracker.get_failure_count("device1"), 1);
        assert!(tracker.has_failures());

        tracker.mark_failed("device1");
        assert_eq!(tracker.get_failure_count("device1"), 2);

        tracker.mark_recovered("device1");
        assert_eq!(tracker.get_failure_count("device1"), 0);
        assert!(!tracker.has_failures());
    }

    #[test]
    fn test_batch_failure_tracking() {
        let mut tracker = SubmissionTracker::new();

        tracker.mark_batch_failed(&["device1", "device2"]);
        assert_eq!(tracker.get_failure_count("device1"), 1);
        assert_eq!(tracker.get_failure_count("device2"), 1);
        assert_eq!(tracker.get_total_failures(), 2);

        tracker.mark_batch_recovered(&["device1", "device2"]);
        assert_eq!(tracker.get_total_failures(), 0);
    }

    #[test]
    fn test_exponential_backoff() {
        let mut tracker = SubmissionTracker::new();
        let base = 30.0;
        let max = 300.0;

        // No failures = no backoff
        assert_eq!(tracker.get_device_backoff("device1", base, max), Duration::from_secs(0));

        // 1 failure = 60s
        tracker.mark_failed("device1");
        assert_eq!(tracker.get_device_backoff("device1", base, max), Duration::from_secs(60));

        // 2 failures = 120s
        tracker.mark_failed("device1");
        assert_eq!(tracker.get_device_backoff("device1", base, max), Duration::from_secs(120));

        // 3 failures = 240s
        tracker.mark_failed("device1");
        assert_eq!(tracker.get_device_backoff("device1", base, max), Duration::from_secs(240));

        // 4 failures = 480s, but capped at 300s
        tracker.mark_failed("device1");
        assert_eq!(tracker.get_device_backoff("device1", base, max), Duration::from_secs(300));
    }

    #[test]
    fn test_global_backoff_uses_max_failures() {
        let mut tracker = SubmissionTracker::new();

        tracker.mark_failed("device1");
        tracker.mark_failed("device1");  // 2 failures
        tracker.mark_failed("device2");  // 1 failure

        // Global backoff should use max (2 failures) = 120s
        let backoff = tracker.get_global_backoff(30.0, 300.0);
        assert_eq!(backoff, Duration::from_secs(120));
    }

    #[test]
    fn test_stats() {
        let mut tracker = SubmissionTracker::new();

        let data = vec![
            create_test_data("device1", 100),
            create_test_data("device2", 200),
        ];
        let refs: Vec<&IoTData> = data.iter().collect();
        tracker.mark_submitted(&refs);

        tracker.mark_failed("device1");
        tracker.mark_failed("device1");
        tracker.mark_failed("device3");

        let stats = tracker.stats();
        assert_eq!(stats.tracked_devices, 2);
        assert_eq!(stats.devices_with_failures, 2);
        assert_eq!(stats.total_failures, 3);
        assert_eq!(stats.max_failures, 2);
    }

    #[test]
    fn test_incremental_timestamp_updates() {
        let mut tracker = SubmissionTracker::new();

        // First submission
        let data1 = vec![create_test_data("device1", 100)];
        let refs1: Vec<&IoTData> = data1.iter().collect();
        tracker.mark_submitted(&refs1);
        assert_eq!(tracker.get_device_timestamp("device1"), 100);

        // Second submission with higher timestamp
        let data2 = vec![create_test_data("device1", 200)];
        let refs2: Vec<&IoTData> = data2.iter().collect();
        tracker.mark_submitted(&refs2);
        assert_eq!(tracker.get_device_timestamp("device1"), 200);

        // Submission with lower timestamp shouldn't regress
        let data3 = vec![create_test_data("device1", 150)];
        let refs3: Vec<&IoTData> = data3.iter().collect();
        tracker.mark_submitted(&refs3);
        assert_eq!(tracker.get_device_timestamp("device1"), 200);  // Still 200
    }

    #[test]
    fn test_empty_submission() {
        let mut tracker = SubmissionTracker::new();
        let empty: Vec<&IoTData> = vec![];
        tracker.mark_submitted(&empty);

        // Should not crash, no timestamps updated
        assert!(tracker.get_all_timestamps().is_empty());
    }

    #[test]
    fn test_filter_empty_data() {
        let tracker = SubmissionTracker::new();
        let empty: Vec<IoTData> = vec![];
        let filtered = tracker.filter_new_data(empty);
        assert!(filtered.is_empty());
    }
}