dahua-camera-server 0.3.1

//! Shared state for every handler.
//!
//! Wraps the camera service and, when the `alarms` feature is on, the alarm
//! registry. [`Deref`](std::ops::Deref) to the service keeps every existing
//! handler that says `state.get(id)` working unchanged.

use dahua_camera_rtsp::CameraService;
use std::sync::Arc;

/// What handlers are given.
pub type AppState = Arc<AppStateInner>;

/// The state behind [`AppState`].
pub struct AppStateInner {
    /// The cameras.
    pub service: Arc<CameraService>,

    /// Latched alarms, fed by a single aggregator task.
    ///
    /// A [`std::sync::Mutex`], not tokio's: every critical section is a few
    /// map operations with no `.await` inside, so an async mutex would add a
    /// scheduling hop for nothing.
    #[cfg(feature = "alarms")]
    pub alarms: Arc<std::sync::Mutex<dahua_camera_alarms::AlarmRegistry>>,
}

impl AppStateInner {
    /// State for a service, with alarm tracking started if the feature is on.
    ///
    /// # Task budget
    ///
    /// The alarm aggregator is **one task for the whole service**, not one per
    /// camera: it reads the already-merged event bus. The workspace budget
    /// allows no feature more than one task per camera, and `events` already
    /// spends that one.
    pub fn new(service: Arc<CameraService>) -> AppState {
        #[cfg(feature = "alarms")]
        let alarms = {
            let registry = Arc::new(std::sync::Mutex::new(
                dahua_camera_alarms::AlarmRegistry::new(Default::default()),
            ));
            spawn_alarm_aggregator(service.clone(), registry.clone());
            registry
        };

        Arc::new(Self {
            service,
            #[cfg(feature = "alarms")]
            alarms,
        })
    }
}

/// Fold the merged event stream into the alarm registry.
///
/// One task total. Reads only; never touches the video path.
#[cfg(feature = "alarms")]
fn spawn_alarm_aggregator(
    service: Arc<CameraService>,
    registry: Arc<std::sync::Mutex<dahua_camera_alarms::AlarmRegistry>>,
) {
    let mut events = service.subscribe_events();

    tokio::spawn(async move {
        loop {
            match events.recv().await {
                Ok(event) => {
                    let now = now_unix_ms();
                    if let Ok(mut registry) = registry.lock() {
                        registry.observe(&event, now);
                    }
                }
                // Alarms are edge-triggered, so a dropped burst can leave a
                // condition latched with no STOP coming. The auto-clear
                // timeout is what recovers from exactly this.
                Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
                    tracing::warn!(
                        skipped,
                        "alarm aggregator lagged; any alarm whose STOP was dropped \
                         will auto-clear on its timeout"
                    );
                }
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            }
        }
    });
}

/// Host wall-clock time in Unix milliseconds.
#[cfg(feature = "alarms")]
pub(crate) fn now_unix_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

/// So a handler written against the service keeps working unchanged.
impl std::ops::Deref for AppStateInner {
    type Target = CameraService;
    fn deref(&self) -> &CameraService {
        &self.service
    }
}