asyn-rs 0.13.6

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::SystemTime;

use tokio::sync::broadcast;

use crate::param::ParamValue;

/// Filter for selecting which interrupts to receive.
#[derive(Debug, Clone, Default)]
pub struct InterruptFilter {
    /// If set, only receive interrupts with this reason (parameter index).
    pub reason: Option<usize>,
    /// If set, only receive interrupts with this addr.
    pub addr: Option<i32>,
    /// For UInt32Digital: bitmask of bits this subscriber is interested in.
    /// If set, only interrupts where changed bits overlap this mask are forwarded.
    /// C parity: pInterrupt->mask in asynUInt32DigitalInterrupt.
    pub uint32_mask: Option<u32>,
}

/// Value delivered through the interrupt system.
#[derive(Debug, Clone)]
pub struct InterruptValue {
    pub reason: usize,
    pub addr: i32,
    pub value: ParamValue,
    pub timestamp: SystemTime,
    /// For UInt32Digital: bitmask of which bits changed (for per-callback filtering).
    pub uint32_changed_mask: u32,
}

// ---------------------------------------------------------------------------
// Mailbox-based subscription (replaces broadcast filter+forward tasks)
// ---------------------------------------------------------------------------

/// Per-subscriber mailbox: stores the latest matching interrupt value.
/// Intermediate updates are coalesced — consumer always sees the most recent state.
struct SubscriptionMailbox {
    filter: InterruptFilter,
    /// Latest matching value (overwritten on each notify, taken on recv).
    latest: parking_lot::Mutex<Option<InterruptValue>>,
    /// Wakeup signal for the consumer.
    wakeup: tokio::sync::Notify,
    /// Set to false when the subscription is dropped.
    active: AtomicBool,
}

impl SubscriptionMailbox {
    fn matches(&self, iv: &InterruptValue) -> bool {
        if let Some(r) = self.filter.reason {
            if iv.reason != r {
                return false;
            }
        }
        if let Some(a) = self.filter.addr {
            if iv.addr != a {
                return false;
            }
        }
        if let Some(m) = self.filter.uint32_mask {
            if iv.uint32_changed_mask & m == 0 {
                return false;
            }
        }
        true
    }
}

/// Receiver for a filtered interrupt subscription.
///
/// Uses a per-subscriber mailbox instead of broadcast+filter task.
/// Intermediate updates are coalesced: if the consumer is slow, only the latest
/// value is preserved. This eliminates broadcast Lagged errors entirely.
pub struct InterruptReceiver {
    mailbox: Arc<SubscriptionMailbox>,
}

impl InterruptReceiver {
    /// Wait for the next interrupt value matching this subscription's filter.
    /// Returns `None` when the subscription is cancelled (dropped).
    pub async fn recv(&mut self) -> Option<InterruptValue> {
        loop {
            // Register wakeup interest BEFORE checking the slot.
            // This avoids the race where notify_one fires between our check and await.
            let notified = self.mailbox.wakeup.notified();

            // Check if a value is already waiting.
            if let Some(value) = self.mailbox.latest.lock().take() {
                return Some(value);
            }
            // Check if subscription was cancelled.
            if !self.mailbox.active.load(Ordering::Acquire) {
                return None;
            }

            // No value yet — wait for wakeup.
            notified.await;
        }
    }
}

/// RAII subscription handle. Dropping this cancels the subscription.
pub struct InterruptSubscription {
    mailbox: Arc<SubscriptionMailbox>,
    state: Arc<InterruptSharedState>,
}

impl Drop for InterruptSubscription {
    fn drop(&mut self) {
        self.mailbox.active.store(false, Ordering::Release);
        // Wake consumer so it sees active=false and returns None.
        self.mailbox.wakeup.notify_one();
        // Remove from subscription list.
        self.state
            .mailboxes
            .lock()
            .retain(|s| s.active.load(Ordering::Relaxed));
    }
}

// ---------------------------------------------------------------------------
// Shared state between InterruptManager instances (driver ↔ PortHandle)
// ---------------------------------------------------------------------------

/// Shared interrupt infrastructure. Both the driver's InterruptManager and the
/// PortHandle's InterruptManager reference the same `InterruptSharedState` so
/// that subscribers registered on either side receive notifications.
pub struct InterruptSharedState {
    /// Broadcast channel for unfiltered subscribers (subscribe_async).
    /// Kept for backward compatibility with transport layer and tests.
    async_tx: broadcast::Sender<InterruptValue>,
    /// Mailbox-based subscriptions for filtered subscribers (I/O Intr records).
    mailboxes: parking_lot::Mutex<Vec<Arc<SubscriptionMailbox>>>,
    /// Total number of notify() calls.
    notify_count: AtomicU64,
    /// Number of times a mailbox value was overwritten before the consumer read it.
    coalesce_count: AtomicU64,
}

// ---------------------------------------------------------------------------
// InterruptManager
// ---------------------------------------------------------------------------

/// Manages interrupt/callback delivery.
///
/// Two delivery paths:
/// - **Filtered subscriptions** (I/O Intr records): mailbox-based, no data loss.
///   `register_interrupt_user()` creates a per-subscriber mailbox that stores
///   the latest matching value. Intermediate updates are coalesced.
/// - **Unfiltered subscriptions** (transport, tests): broadcast-based.
///   `subscribe_async()` returns a broadcast receiver for backward compatibility.
pub struct InterruptManager {
    state: Arc<InterruptSharedState>,
}

impl InterruptManager {
    pub fn new(async_capacity: usize) -> Self {
        let (async_tx, _) = broadcast::channel(async_capacity);
        Self {
            state: Arc::new(InterruptSharedState {
                async_tx,
                mailboxes: parking_lot::Mutex::new(Vec::new()),
                notify_count: AtomicU64::new(0),
                coalesce_count: AtomicU64::new(0),
            }),
        }
    }

    /// Create an InterruptManager sharing the same state as another.
    /// Used by `create_port_runtime` so the PortHandle and the driver share
    /// the same subscription list and broadcast channel.
    pub fn from_shared_state(state: Arc<InterruptSharedState>) -> Self {
        Self { state }
    }

    /// Get the shared state for cross-manager sharing.
    pub fn shared_state(&self) -> Arc<InterruptSharedState> {
        self.state.clone()
    }

    /// Create an InterruptManager sharing an existing broadcast sender.
    /// **Deprecated**: prefer `from_shared_state` which also shares mailbox subscriptions.
    /// Kept for backward compatibility.
    pub fn from_broadcast_sender(sender: broadcast::Sender<InterruptValue>) -> Self {
        Self {
            state: Arc::new(InterruptSharedState {
                async_tx: sender,
                mailboxes: parking_lot::Mutex::new(Vec::new()),
                notify_count: AtomicU64::new(0),
                coalesce_count: AtomicU64::new(0),
            }),
        }
    }

    /// Subscribe for async interrupt delivery (unfiltered, broadcast-based).
    /// Multiple subscribers OK. Used by transport layer and tests.
    pub fn subscribe_async(&self) -> broadcast::Receiver<InterruptValue> {
        self.state.async_tx.subscribe()
    }

    /// Clone the broadcast sender for sharing.
    pub fn broadcast_sender(&self) -> broadcast::Sender<InterruptValue> {
        self.state.async_tx.clone()
    }

    /// Send an interrupt to all subscribers (both broadcast and mailbox).
    pub fn notify(&self, value: InterruptValue) {
        self.state.notify_count.fetch_add(1, Ordering::Relaxed);

        // Deliver to mailbox subscribers (filtered, coalescing).
        let subs = self.state.mailboxes.lock();
        for sub in subs.iter() {
            if !sub.active.load(Ordering::Relaxed) {
                continue;
            }
            if !sub.matches(&value) {
                continue;
            }
            let mut slot = sub.latest.lock();
            if slot.is_some() {
                self.state.coalesce_count.fetch_add(1, Ordering::Relaxed);
            }
            *slot = Some(value.clone());
            drop(slot);
            sub.wakeup.notify_one();
        }
        drop(subs);

        // Deliver to broadcast subscribers (unfiltered, legacy).
        let _ = self.state.async_tx.send(value);
    }

    /// Register a filtered interrupt subscription using the mailbox model.
    ///
    /// Returns an RAII `InterruptSubscription` (dropping it unsubscribes) and an
    /// `InterruptReceiver` for receiving matching interrupts.
    ///
    /// Unlike the broadcast-based approach, this **never drops values** due to
    /// channel pressure. If the consumer is slow, intermediate updates are
    /// coalesced (latest value preserved, coalesce_count incremented).
    pub fn register_interrupt_user(
        &self,
        filter: InterruptFilter,
    ) -> (InterruptSubscription, InterruptReceiver) {
        let mailbox = Arc::new(SubscriptionMailbox {
            filter,
            latest: parking_lot::Mutex::new(None),
            wakeup: tokio::sync::Notify::new(),
            active: AtomicBool::new(true),
        });
        self.state.mailboxes.lock().push(mailbox.clone());
        (
            InterruptSubscription {
                mailbox: mailbox.clone(),
                state: self.state.clone(),
            },
            InterruptReceiver { mailbox },
        )
    }

    // --- Metrics ---

    /// Total number of notify() calls since creation.
    pub fn notify_count(&self) -> u64 {
        self.state.notify_count.load(Ordering::Relaxed)
    }

    /// Number of times a mailbox value was overwritten before the consumer read it.
    /// High coalesce count at moderate frame rates indicates consumer backpressure.
    pub fn coalesce_count(&self) -> u64 {
        self.state.coalesce_count.load(Ordering::Relaxed)
    }
}

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

    #[tokio::test]
    async fn test_async_subscribe_receive() {
        let im = InterruptManager::new(16);
        let mut rx = im.subscribe_async();
        im.notify(InterruptValue {
            reason: 1,
            addr: 0,
            value: ParamValue::Float64(3.14),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });
        let v = rx.recv().await.unwrap();
        assert_eq!(v.reason, 1);
    }

    #[tokio::test]
    async fn test_async_multiple_subscribers() {
        let im = InterruptManager::new(16);
        let mut rx1 = im.subscribe_async();
        let mut rx2 = im.subscribe_async();
        im.notify(InterruptValue {
            reason: 0,
            addr: 0,
            value: ParamValue::Int32(99),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });
        let v1 = rx1.recv().await.unwrap();
        let v2 = rx2.recv().await.unwrap();
        assert_eq!(v1.reason, 0);
        assert_eq!(v2.reason, 0);
    }

    #[tokio::test]
    async fn test_register_interrupt_user_filter_by_reason() {
        let im = InterruptManager::new(16);
        let (_sub, mut rx) = im.register_interrupt_user(InterruptFilter {
            reason: Some(1),
            addr: None,
            ..Default::default()
        });

        // Send reason 0 — should NOT be received
        im.notify(InterruptValue {
            reason: 0,
            addr: 0,
            value: ParamValue::Int32(10),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });

        // Send reason 1 — should be received
        im.notify(InterruptValue {
            reason: 1,
            addr: 0,
            value: ParamValue::Int32(20),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });

        let v = tokio::time::timeout(std::time::Duration::from_millis(100), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(v.reason, 1);
        if let ParamValue::Int32(n) = v.value {
            assert_eq!(n, 20);
        } else {
            panic!("expected Int32");
        }
    }

    #[tokio::test]
    async fn test_register_interrupt_user_filter_by_addr() {
        let im = InterruptManager::new(16);
        let (_sub, mut rx) = im.register_interrupt_user(InterruptFilter {
            reason: None,
            addr: Some(3),
            ..Default::default()
        });

        im.notify(InterruptValue {
            reason: 0,
            addr: 0,
            value: ParamValue::Int32(1),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });
        im.notify(InterruptValue {
            reason: 0,
            addr: 3,
            value: ParamValue::Int32(2),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });

        let v = tokio::time::timeout(std::time::Duration::from_millis(100), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(v.addr, 3);
    }

    #[tokio::test]
    async fn test_register_interrupt_user_no_filter() {
        let im = InterruptManager::new(16);
        let (_sub, mut rx) = im.register_interrupt_user(InterruptFilter::default());

        im.notify(InterruptValue {
            reason: 5,
            addr: 2,
            value: ParamValue::Float64(1.5),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });

        let v = tokio::time::timeout(std::time::Duration::from_millis(100), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(v.reason, 5);
        assert_eq!(v.addr, 2);
    }

    #[tokio::test]
    async fn test_register_interrupt_user_drop_unsubscribes() {
        let im = InterruptManager::new(16);
        let (sub, mut rx) = im.register_interrupt_user(InterruptFilter::default());

        // Drop subscription
        drop(sub);

        // Consumer should see None (subscription cancelled)
        let result = tokio::time::timeout(std::time::Duration::from_millis(50), rx.recv()).await;
        match result {
            Ok(None) => {} // cancelled — correct
            Err(_) => {}   // timed out — also acceptable
            Ok(Some(_)) => panic!("should not receive after unsubscribe"),
        }
    }

    #[tokio::test]
    async fn test_register_interrupt_user_multiple_subscribers() {
        let im = InterruptManager::new(16);
        let (_sub1, mut rx1) = im.register_interrupt_user(InterruptFilter {
            reason: Some(0),
            addr: None,
            ..Default::default()
        });
        let (_sub2, mut rx2) = im.register_interrupt_user(InterruptFilter {
            reason: Some(1),
            addr: None,
            ..Default::default()
        });

        im.notify(InterruptValue {
            reason: 0,
            addr: 0,
            value: ParamValue::Int32(10),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });
        im.notify(InterruptValue {
            reason: 1,
            addr: 0,
            value: ParamValue::Int32(20),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });

        let v1 = tokio::time::timeout(std::time::Duration::from_millis(100), rx1.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(v1.reason, 0);

        let v2 = tokio::time::timeout(std::time::Duration::from_millis(100), rx2.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(v2.reason, 1);
    }

    #[test]
    fn test_notify_no_subscribers_no_panic() {
        let im = InterruptManager::new(16);
        im.notify(InterruptValue {
            reason: 0,
            addr: 0,
            value: ParamValue::Int32(1),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });
    }

    #[tokio::test]
    async fn test_coalescing() {
        let im = InterruptManager::new(16);
        let (_sub, mut rx) = im.register_interrupt_user(InterruptFilter {
            reason: Some(0),
            ..Default::default()
        });

        // Send 3 values without consumer reading — should coalesce to latest.
        im.notify(InterruptValue {
            reason: 0,
            addr: 0,
            value: ParamValue::Int32(1),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });
        im.notify(InterruptValue {
            reason: 0,
            addr: 0,
            value: ParamValue::Int32(2),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });
        im.notify(InterruptValue {
            reason: 0,
            addr: 0,
            value: ParamValue::Int32(3),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });

        // Consumer should see only the latest value (3).
        let v = tokio::time::timeout(std::time::Duration::from_millis(100), rx.recv())
            .await
            .unwrap()
            .unwrap();
        if let ParamValue::Int32(n) = v.value {
            assert_eq!(n, 3);
        } else {
            panic!("expected Int32");
        }

        // Coalesce count should be 2 (first write creates, next two overwrite).
        assert_eq!(im.coalesce_count(), 2);
    }

    #[tokio::test]
    async fn test_shared_state_between_managers() {
        let im1 = InterruptManager::new(16);
        let shared = im1.shared_state();
        let im2 = InterruptManager::from_shared_state(shared);

        // Subscribe via im2
        let (_sub, mut rx) = im2.register_interrupt_user(InterruptFilter {
            reason: Some(0),
            ..Default::default()
        });

        // Notify via im1 — subscriber should receive because state is shared
        im1.notify(InterruptValue {
            reason: 0,
            addr: 0,
            value: ParamValue::Int32(42),
            timestamp: SystemTime::now(),
            uint32_changed_mask: 0,
        });

        let v = tokio::time::timeout(std::time::Duration::from_millis(100), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(v.reason, 0);
        if let ParamValue::Int32(n) = v.value {
            assert_eq!(n, 42);
        } else {
            panic!("expected Int32");
        }
    }
}