onvif_server/service/

events.rs

use async_trait::async_trait;
use bytes::Bytes;
use chrono::{DateTime, Duration, Utc};
use quick_xml::events::Event;
use quick_xml::NsReader;
use soap_server::{escape_text, SoapFault, SoapHandler};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use uuid::Uuid;

use crate::error::OnvifError;
use crate::service::xml_util::extract_text_ns;
use crate::traits::EventService;

/// ONVIF namespaces accepted for event request elements.
const ONVIF_EVENTS_NS: &[u8] = b"http://www.onvif.org/ver10/events/wsdl";
const ONVIF_SCHEMA_NS: &[u8] = b"http://www.onvif.org/ver10/schema";
/// WS-BaseNotification namespace (used for InitialTerminationTime in CreatePullPointSubscription).
const WSNT_NS: &[u8] = b"http://docs.oasis-open.org/wsn/b-2";

/// Default subscription lifetime in seconds when no (or unparseable)
/// `InitialTerminationTime` is provided by the client.
const DEFAULT_SUBSCRIPTION_SECS: i64 = 60;

struct SubscriptionInfo {
    termination_time: DateTime<Utc>,
}

#[allow(dead_code)]
pub struct EventServiceHandler {
    pub(crate) svc: Arc<dyn EventService>,
    pub(crate) xaddr: String,
    subscriptions: Arc<Mutex<HashMap<String, SubscriptionInfo>>>,
}

impl EventServiceHandler {
    pub fn new(svc: Arc<dyn EventService>, xaddr: impl Into<String>) -> Self {
        Self {
            svc,
            xaddr: xaddr.into(),
            subscriptions: Arc::new(Mutex::new(HashMap::new())),
        }
    }
}

#[async_trait]
impl SoapHandler for EventServiceHandler {
    async fn handle(&self, body: Bytes) -> Result<Bytes, SoapFault> {
        // Delegate through handle_with_headers with an empty header slice.
        self.handle_with_headers(body, &[]).await
    }

    /// Override to receive SOAP header fragments so WS-Addressing reference
    /// parameters (echoed SubscriptionId) can be used for routing.
    async fn handle_with_headers(
        &self,
        body: Bytes,
        headers: &[Bytes],
    ) -> Result<Bytes, SoapFault> {
        let op = extract_local_name(&body)?;
        match op.as_str() {
            "GetEventProperties" => self.handle_get_event_properties().await,
            "CreatePullPointSubscription" => {
                self.handle_create_pull_point_subscription(&body).await
            }
            "PullMessages" => self.handle_pull_messages(&body, headers).await,
            "Unsubscribe" => self.handle_unsubscribe(&body, headers).await,
            _ => Err(OnvifError::ActionNotSupported.into_soap_fault()),
        }
    }
}

fn extract_local_name(body: &Bytes) -> Result<String, SoapFault> {
    let mut reader = NsReader::from_reader(body.as_ref());
    reader.config_mut().trim_text(true);
    loop {
        match reader
            .read_resolved_event()
            .map_err(|e| SoapFault::sender(format!("{e}")))?
        {
            (_, Event::Start(e)) | (_, Event::Empty(e)) => {
                let local = std::str::from_utf8(e.local_name().as_ref())
                    .map_err(|e| SoapFault::sender(format!("{e}")))?
                    .to_string();
                return Ok(local);
            }
            (_, Event::Eof) => return Err(SoapFault::sender("Empty body".to_string())),
            _ => {}
        }
    }
}

fn extract_text_element(body: &Bytes, element_name: &str) -> Result<String, SoapFault> {
    extract_text_ns(body, element_name, &[ONVIF_EVENTS_NS, ONVIF_SCHEMA_NS])
}

/// Resolve the SubscriptionId for PullMessages / Unsubscribe.
///
/// WS-Addressing reference parameters: the SubscriptionId the server placed inside
/// `ReferenceParameters` at CreatePullPointSubscription time is echoed back by
/// conforming clients as a direct child of `<SOAP:Header>`.  Parse it from the
/// first header element whose local name is `SubscriptionId` (accepted in the
/// ONVIF events namespace or unbound).  Fall back to the body element for
/// backwards-compat with clients that embed it in the operation body.
fn resolve_subscription_id(body: &Bytes, headers: &[Bytes]) -> Result<String, SoapFault> {
    // 1. Try headers first (WS-Addressing reference parameter echo).
    for header in headers {
        if let Ok(id) = extract_text_ns(
            header,
            "SubscriptionId",
            &[ONVIF_EVENTS_NS, ONVIF_SCHEMA_NS],
        ) {
            if !id.is_empty() {
                return Ok(id);
            }
        }
    }

    // 2. Fall back to body element (backward compat).
    extract_text_element(body, "SubscriptionId")
}

/// Parse an `InitialTerminationTime` value into a `Duration`.
///
/// Accepts:
/// - ISO 8601 duration strings: `PT<seconds>S` (e.g. `PT60S`, `PT300S`).
///   Only seconds-level resolution is required for ONVIF pull-point subscriptions.
/// - Absolute RFC 3339 / ISO 8601 datetimes: interpreted as an absolute expiry.
/// - Anything unparseable: returns `None` (caller should use the default).
fn parse_termination_time(value: &str, now: DateTime<Utc>) -> Option<Duration> {
    let trimmed = value.trim();

    // ISO 8601 duration: must start with 'P'.
    if trimmed.starts_with('P') {
        return parse_iso8601_duration(trimmed);
    }

    // Try absolute datetime (RFC 3339).
    if let Ok(absolute) = trimmed.parse::<DateTime<Utc>>() {
        let delta = absolute.signed_duration_since(now);
        if delta > Duration::zero() {
            return Some(delta);
        }
        // Already-past time — use default.
        return None;
    }

    None
}

/// Parse a bounded subset of ISO 8601 durations: `P[nY][nM][nDT[nH][nM][nS]]`.
/// Only handles whole-unit fields; fractional seconds are truncated.
/// Returns `None` if the string does not conform.
fn parse_iso8601_duration(s: &str) -> Option<Duration> {
    // Strip leading 'P'.
    let s = s.strip_prefix('P')?;

    let mut total_secs: i64 = 0;
    let (date_part, time_part) = if let Some(t) = s.split_once('T') {
        (t.0, Some(t.1))
    } else {
        (s, None)
    };

    // Parse date fields: nY, nM, nD.
    let mut remaining = date_part;
    total_secs += consume_duration_field(&mut remaining, 'Y')? * 365 * 86400;
    total_secs += consume_duration_field(&mut remaining, 'M')? * 30 * 86400;
    total_secs += consume_duration_field(&mut remaining, 'D')? * 86400;

    // Parse time fields: nH, nM, nS.
    if let Some(time) = time_part {
        let mut rem = time;
        total_secs += consume_duration_field(&mut rem, 'H')? * 3600;
        total_secs += consume_duration_field(&mut rem, 'M')? * 60;
        // Seconds: allow fractional (truncate).
        total_secs += consume_duration_seconds(&mut rem)?;
        if !rem.is_empty() {
            return None;
        }
    } else if !remaining.is_empty() {
        return None;
    }

    Some(Duration::seconds(total_secs))
}

/// Consume an optional `<integer><designator>` prefix from `*s`, returning the
/// integer (or 0 if this designator is absent).  Returns `None` on parse error.
fn consume_duration_field(s: &mut &str, designator: char) -> Option<i64> {
    if let Some(pos) = s.find(designator) {
        // Everything before the designator must be a non-empty integer.
        let num_str = &s[..pos];
        if num_str.is_empty() {
            return None;
        }
        let n: i64 = num_str.parse().ok()?;
        *s = &s[pos + designator.len_utf8()..];
        Some(n)
    } else {
        Some(0)
    }
}

/// Consume optional `<integer-or-float>S` from the front of `*s`, returning the
/// truncated integer seconds.  Returns 0 (not None) if 'S' is absent.
fn consume_duration_seconds(s: &mut &str) -> Option<i64> {
    if let Some(pos) = s.find('S') {
        let num_str = &s[..pos];
        if num_str.is_empty() {
            return None;
        }
        // Accept integer or decimal (e.g. "30.5S" → 30).
        let n: i64 = if let Some(dot) = num_str.find('.') {
            num_str[..dot].parse().ok()?
        } else {
            num_str.parse().ok()?
        };
        *s = &s[pos + 1..];
        Some(n)
    } else {
        Some(0)
    }
}

impl EventServiceHandler {
    async fn handle_get_event_properties(&self) -> Result<Bytes, SoapFault> {
        let xml = r#"<tev:GetEventPropertiesResponse xmlns:tev="http://www.onvif.org/ver10/events/wsdl" xmlns:wsnt="http://docs.oasis-open.org/wsn/b-2" xmlns:wstop="http://docs.oasis-open.org/wsn/t-1"><tev:TopicNamespaceLocation>http://www.onvif.org/onvif/ver10/topics/topicns.xml</tev:TopicNamespaceLocation><wsnt:FixedTopicSet>true</wsnt:FixedTopicSet><wstop:TopicSet/><wsnt:TopicExpressionDialect>http://docs.oasis-open.org/wsn/t-1/TopicExpression/Concrete</wsnt:TopicExpressionDialect><wsnt:TopicExpressionDialect>http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet</wsnt:TopicExpressionDialect><tev:MessageContentFilterDialect>http://www.onvif.org/ver10/tev/messageContentFilter/ItemFilter</tev:MessageContentFilterDialect><tev:MessageContentSchemaLocation>http://www.onvif.org/ver10/schema/onvif.xsd</tev:MessageContentSchemaLocation></tev:GetEventPropertiesResponse>"#;
        Ok(Bytes::from(xml))
    }

    async fn handle_create_pull_point_subscription(
        &self,
        body: &Bytes,
    ) -> Result<Bytes, SoapFault> {
        let sub_id = Uuid::new_v4().to_string();
        let now = Utc::now();

        // Parse InitialTerminationTime from the request body.
        // The element may appear in the ONVIF events namespace, the WS-Notification
        // namespace (wsnt), or unbound — accept all three.
        // Accept either an ISO 8601 duration (e.g. PT60S) or an absolute datetime.
        // Fall back to DEFAULT_SUBSCRIPTION_SECS if absent or unparseable.
        let lifetime = extract_text_ns(
            body,
            "InitialTerminationTime",
            &[ONVIF_EVENTS_NS, ONVIF_SCHEMA_NS, WSNT_NS],
        )
        .ok()
        .and_then(|v| parse_termination_time(&v, now))
        .unwrap_or_else(|| Duration::seconds(DEFAULT_SUBSCRIPTION_SECS));

        let termination = now + lifetime;

        {
            let mut subs = self
                .subscriptions
                .lock()
                .map_err(|e| SoapFault::sender(format!("lock poisoned: {e}")))?;
            subs.insert(
                sub_id.clone(),
                SubscriptionInfo {
                    termination_time: termination,
                },
            );
        }

        let xml = format!(
            r#"<tev:CreatePullPointSubscriptionResponse xmlns:tev="http://www.onvif.org/ver10/events/wsdl" xmlns:wsa5="http://www.w3.org/2005/08/addressing" xmlns:wsnt="http://docs.oasis-open.org/wsn/b-2"><tev:SubscriptionReference><wsa5:Address>{xaddr}</wsa5:Address><wsa5:ReferenceParameters><tev:SubscriptionId>{sub_id}</tev:SubscriptionId></wsa5:ReferenceParameters></tev:SubscriptionReference><wsnt:CurrentTime>{current}</wsnt:CurrentTime><wsnt:TerminationTime>{termination}</wsnt:TerminationTime></tev:CreatePullPointSubscriptionResponse>"#,
            xaddr = escape_text(&self.xaddr),
            sub_id = escape_text(&sub_id),
            current = now.to_rfc3339(),
            termination = termination.to_rfc3339(),
        );
        Ok(Bytes::from(xml))
    }

    async fn handle_pull_messages(
        &self,
        body: &Bytes,
        headers: &[Bytes],
    ) -> Result<Bytes, SoapFault> {
        // Resolve SubscriptionId: header (WS-Addressing ref param) first, body fallback.
        let sub_id = resolve_subscription_id(body, headers)?;

        let now = Utc::now();
        let termination_time = {
            let subs = self
                .subscriptions
                .lock()
                .map_err(|e| SoapFault::sender(format!("lock poisoned: {e}")))?;
            match subs.get(&sub_id) {
                Some(info) if info.termination_time > now => info.termination_time,
                Some(_) => {
                    // Subscription exists but has expired.
                    // Log detail server-side; return generic fault reason to client (#12).
                    eprintln!("[onvif-events] PullMessages: subscription expired (id not echoed to client)");
                    return Err(SoapFault::sender("Subscription not found".to_string()));
                }
                None => {
                    // Log detail server-side; return generic fault reason to client (#12).
                    eprintln!("[onvif-events] PullMessages: unknown subscription (id not echoed to client)");
                    return Err(SoapFault::sender("Subscription not found".to_string()));
                }
            }
        };

        let xml = format!(
            r#"<tev:PullMessagesResponse xmlns:tev="http://www.onvif.org/ver10/events/wsdl" xmlns:wsnt="http://docs.oasis-open.org/wsn/b-2"><tev:CurrentTime>{current}</tev:CurrentTime><tev:TerminationTime>{termination}</tev:TerminationTime></tev:PullMessagesResponse>"#,
            current = now.to_rfc3339(),
            termination = termination_time.to_rfc3339(),
        );
        Ok(Bytes::from(xml))
    }

    async fn handle_unsubscribe(
        &self,
        body: &Bytes,
        headers: &[Bytes],
    ) -> Result<Bytes, SoapFault> {
        // Resolve SubscriptionId: header (WS-Addressing ref param) first, body fallback.
        let sub_id = resolve_subscription_id(body, headers)?;

        {
            let mut subs = self
                .subscriptions
                .lock()
                .map_err(|e| SoapFault::sender(format!("lock poisoned: {e}")))?;
            if subs.remove(&sub_id).is_none() {
                // Unknown or already-expired subscription — log server-side, fault to client (#12).
                eprintln!("[onvif-events] Unsubscribe: unknown or expired subscription (id not echoed to client)");
                return Err(SoapFault::sender("Subscription not found".to_string()));
            }
        }

        let xml =
            r#"<tev:UnsubscribeResponse xmlns:tev="http://www.onvif.org/ver10/events/wsdl"/>"#;
        Ok(Bytes::from(xml))
    }
}

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

    // ── parse_iso8601_duration unit tests ─────────────────────────────────────

    #[test]
    fn parse_pt60s_returns_60_seconds() {
        let d = parse_iso8601_duration("PT60S").unwrap();
        assert_eq!(d.num_seconds(), 60);
    }

    #[test]
    fn parse_pt300s_returns_300_seconds() {
        let d = parse_iso8601_duration("PT300S").unwrap();
        assert_eq!(d.num_seconds(), 300);
    }

    #[test]
    fn parse_pt1h_returns_3600_seconds() {
        let d = parse_iso8601_duration("PT1H").unwrap();
        assert_eq!(d.num_seconds(), 3600);
    }

    #[test]
    fn parse_pt1h30m_returns_5400_seconds() {
        let d = parse_iso8601_duration("PT1H30M").unwrap();
        assert_eq!(d.num_seconds(), 5400);
    }

    #[test]
    fn parse_p1d_returns_86400_seconds() {
        let d = parse_iso8601_duration("P1D").unwrap();
        assert_eq!(d.num_seconds(), 86400);
    }

    #[test]
    fn parse_empty_string_returns_none() {
        assert!(parse_iso8601_duration("").is_none());
    }

    #[test]
    fn parse_garbage_returns_none() {
        assert!(parse_iso8601_duration("garbage").is_none());
        assert!(parse_iso8601_duration("60S").is_none()); // missing 'P'
    }

    // ── parse_termination_time absolute datetime tests ────────────────────────

    #[test]
    fn parse_future_absolute_datetime_returns_positive_duration() {
        let now = Utc::now();
        let future = now + Duration::seconds(120);
        let s = future.to_rfc3339();
        let d = parse_termination_time(&s, now).unwrap();
        // Should be approximately 120 seconds (allow ±2s for rounding).
        assert!(d.num_seconds() >= 118 && d.num_seconds() <= 122);
    }

    #[test]
    fn parse_past_absolute_datetime_returns_none() {
        let now = Utc::now();
        let past = now - Duration::seconds(10);
        let s = past.to_rfc3339();
        assert!(parse_termination_time(&s, now).is_none());
    }

    // ── resolve_subscription_id: header-first, body-fallback ─────────────────

    #[test]
    fn resolve_sub_id_from_header_takes_priority_over_body() {
        let body = Bytes::from(
            r#"<tev:PullMessages xmlns:tev="http://www.onvif.org/ver10/events/wsdl"><tev:SubscriptionId>body-id</tev:SubscriptionId></tev:PullMessages>"#,
        );
        let header = Bytes::from(
            r#"<tev:SubscriptionId xmlns:tev="http://www.onvif.org/ver10/events/wsdl">header-id</tev:SubscriptionId>"#,
        );
        let id = resolve_subscription_id(&body, &[header]).unwrap();
        assert_eq!(id, "header-id");
    }

    #[test]
    fn resolve_sub_id_falls_back_to_body_when_no_header() {
        let body = Bytes::from(
            r#"<tev:PullMessages xmlns:tev="http://www.onvif.org/ver10/events/wsdl"><tev:SubscriptionId>body-id</tev:SubscriptionId></tev:PullMessages>"#,
        );
        let id = resolve_subscription_id(&body, &[]).unwrap();
        assert_eq!(id, "body-id");
    }

    #[test]
    fn resolve_sub_id_unbound_namespace_in_header_accepted() {
        // Header element with no namespace binding — must be accepted.
        let body = Bytes::from(
            r#"<tev:PullMessages xmlns:tev="http://www.onvif.org/ver10/events/wsdl"><tev:SubscriptionId>body-id</tev:SubscriptionId></tev:PullMessages>"#,
        );
        let header = Bytes::from(r#"<SubscriptionId>bare-id</SubscriptionId>"#);
        let id = resolve_subscription_id(&body, &[header]).unwrap();
        assert_eq!(id, "bare-id");
    }
}