grafton_ndi/

receiver.rs

//! NDI receiving functionality for video, audio, and metadata.
//!
//! # Monitoring Connection Status
//!
//! The receiver can monitor connection status changes:
//!
//! ```no_run
//! # use grafton_ndi::{NDI, Receiver, ReceiverOptions, ReceiverBandwidth, Source};
//! # use std::time::Duration;
//! # fn main() -> Result<(), grafton_ndi::Error> {
//! # let ndi = NDI::new()?;
//! # let source = Source::default();
//! // In real usage, obtain the source from Finder::find_sources().
//! let options = ReceiverOptions::builder(source)
//!     .bandwidth(ReceiverBandwidth::MetadataOnly)
//!     .build();
//! let receiver = Receiver::new(&ndi, &options)?;
//!
//! // Poll for status changes
//! if let Some(status) = receiver.poll_status_change(Duration::from_millis(1000))? {
//!     println!("Active connections: {}", status.connections);
//! }
//! # Ok(())
//! # }
//! ```

use std::{
    ffi::CString,
    marker::PhantomData,
    ptr,
    sync::atomic::{AtomicBool, Ordering},
    time::{Duration, Instant},
};

use parking_lot::{RwLock, RwLockReadGuard};

use crate::{
    capture::{capture_raw, AudioKind, CaptureKind, CaptureResult, MetadataKind, VideoKind},
    finder::{RawSource, Source},
    ndi_lib::*,
    to_ms_checked, Error, MetadataFrame, Result, NDI,
};

/// Retry policy configuration for frame capture operations.
///
/// This struct encapsulates the timing parameters for the retry loop used by
/// the reliable [`Capture::capture`] verb across all frame kinds.
struct RetryPolicy {
    /// Timeout per individual capture attempt.
    poll_interval: Duration,
    /// Sleep duration between retry attempts to avoid busy-waiting.
    sleep_between: Duration,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            poll_interval: Duration::from_millis(100),
            sleep_between: Duration::from_millis(10),
        }
    }
}

/// Generic retry helper for frame capture operations.
///
/// This function encapsulates the retry logic that handles NDI SDK synchronization
/// behavior during initial connection. The first few capture calls may return
/// immediately while the stream synchronizes.
///
/// # Parameters
///
/// - `timeout`: Total time allowed for the operation to succeed.
/// - `policy`: Retry timing configuration.
/// - `capture_fn`: A closure that attempts to capture a frame with a given timeout.
///
/// # Returns
///
/// - `Ok(T)`: The captured frame on success.
/// - `Err(Error::FrameTimeout)`: If no frame is captured within the total timeout.
fn retry_capture<T, F>(timeout: Duration, policy: &RetryPolicy, capture_fn: F) -> Result<T>
where
    F: FnMut(Duration) -> Result<Option<T>>,
{
    let start_time = Instant::now();
    retry_capture_with_clock(
        timeout,
        policy,
        capture_fn,
        || start_time.elapsed(),
        std::thread::sleep,
    )
}

fn retry_capture_with_clock<T, F, E, S>(
    timeout: Duration,
    policy: &RetryPolicy,
    mut capture_fn: F,
    mut elapsed: E,
    mut sleep: S,
) -> Result<T>
where
    F: FnMut(Duration) -> Result<Option<T>>,
    E: FnMut() -> Duration,
    S: FnMut(Duration),
{
    to_ms_checked(timeout)?;

    let mut attempts = 0;

    if timeout.is_zero() {
        attempts += 1;
        return match capture_fn(Duration::ZERO)? {
            Some(frame) => Ok(frame),
            None => Err(Error::FrameTimeout {
                attempts,
                elapsed: elapsed(),
            }),
        };
    }

    loop {
        let elapsed_before_attempt = elapsed();
        let Some(remaining) = timeout.checked_sub(elapsed_before_attempt) else {
            return Err(Error::FrameTimeout {
                attempts,
                elapsed: elapsed_before_attempt,
            });
        };

        if remaining.is_zero() {
            return Err(Error::FrameTimeout {
                attempts,
                elapsed: elapsed_before_attempt,
            });
        }

        let poll_timeout = policy.poll_interval.min(remaining);
        attempts += 1;

        match capture_fn(poll_timeout)? {
            Some(frame) => return Ok(frame),
            None => {
                let elapsed_after_attempt = elapsed();
                let Some(remaining) = timeout.checked_sub(elapsed_after_attempt) else {
                    return Err(Error::FrameTimeout {
                        attempts,
                        elapsed: elapsed_after_attempt,
                    });
                };

                if remaining.is_zero() {
                    return Err(Error::FrameTimeout {
                        attempts,
                        elapsed: elapsed_after_attempt,
                    });
                }

                let sleep_duration = policy.sleep_between.min(remaining);
                if !sleep_duration.is_zero() {
                    sleep(sleep_duration);
                }
            }
        }
    }
}

fn validate_ptz_value(name: &str, value: f32, min: f32, max: f32) -> Result<()> {
    if value.is_finite() && (min..=max).contains(&value) {
        Ok(())
    } else {
        Err(Error::InvalidConfiguration(format!(
            "{name} must be finite and within {min}..={max}"
        )))
    }
}

fn validate_ptz_preset(preset: i64) -> Result<i32> {
    if (0..=99).contains(&preset) {
        Ok(preset as i32)
    } else {
        Err(Error::InvalidConfiguration(
            "PTZ preset must be within 0..=99".into(),
        ))
    }
}

macro_rules! ptz_command {
    ($self:expr, $func:ident, $err_msg:expr) => {
        if unsafe { $func($self.core.instance) } {
            Ok(())
        } else {
            Err(Error::PtzCommandFailed($err_msg.into()))
        }
    };
    ($self:expr, $func:ident, $param:expr, $err_msg:expr) => {
        if unsafe { $func($self.core.instance, $param) } {
            Ok(())
        } else {
            Err(Error::PtzCommandFailed($err_msg))
        }
    };
    ($self:expr, $func:ident, $param1:expr, $param2:expr, $err_msg:expr) => {
        if unsafe { $func($self.core.instance, $param1, $param2) } {
            Ok(())
        } else {
            Err(Error::PtzCommandFailed($err_msg))
        }
    };
    ($self:expr, $func:ident, $param1:expr, $param2:expr, $param3:expr, $err_msg:expr) => {
        if unsafe { $func($self.core.instance, $param1, $param2, $param3) } {
            Ok(())
        } else {
            Err(Error::PtzCommandFailed($err_msg))
        }
    };
}

/// Operational color formats for received video frames.
///
/// These values are receiver configuration modes that the NDI SDK can use for
/// normal receive operation. SDK `Max` sentinel values are intentionally not
/// exposed in safe Rust because they are placeholders, not valid receiver
/// modes.
///
/// This enum is marked `#[non_exhaustive]` so future SDK receiver modes can be
/// added without another avoidable public enum break. Downstream `match`
/// expressions should include a wildcard arm.
///
/// Choose an explicit operational value such as [`ReceiverColorFormat::Fastest`],
/// [`ReceiverColorFormat::Best`], [`ReceiverColorFormat::RGBX_RGBA`], or
/// [`ReceiverColorFormat::BGRX_BGRA`] for receiver configuration.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ReceiverColorFormat {
    #[default]
    BGRX_BGRA,
    UYVY_BGRA,
    RGBX_RGBA,
    UYVY_RGBA,
    Fastest,
    Best,
}

impl From<ReceiverColorFormat> for NDIlib_recv_color_format_e {
    fn from(format: ReceiverColorFormat) -> Self {
        match format {
            ReceiverColorFormat::BGRX_BGRA => {
                NDIlib_recv_color_format_e_NDIlib_recv_color_format_BGRX_BGRA
            }
            ReceiverColorFormat::UYVY_BGRA => {
                NDIlib_recv_color_format_e_NDIlib_recv_color_format_UYVY_BGRA
            }
            ReceiverColorFormat::RGBX_RGBA => {
                NDIlib_recv_color_format_e_NDIlib_recv_color_format_RGBX_RGBA
            }
            ReceiverColorFormat::UYVY_RGBA => {
                NDIlib_recv_color_format_e_NDIlib_recv_color_format_UYVY_RGBA
            }
            ReceiverColorFormat::Fastest => {
                NDIlib_recv_color_format_e_NDIlib_recv_color_format_fastest
            }
            ReceiverColorFormat::Best => NDIlib_recv_color_format_e_NDIlib_recv_color_format_best,
        }
    }
}

/// Operational bandwidth modes for receivers.
///
/// These values are receiver configuration modes that the NDI SDK can use for
/// normal receive operation. SDK `Max` sentinel values are intentionally not
/// exposed in safe Rust because they are placeholders, not valid receiver
/// modes.
///
/// This enum is marked `#[non_exhaustive]` so future SDK receiver modes can be
/// added without another avoidable public enum break. Downstream `match`
/// expressions should include a wildcard arm.
///
/// Choose an explicit operational value such as [`ReceiverBandwidth::Highest`],
/// [`ReceiverBandwidth::Lowest`], [`ReceiverBandwidth::AudioOnly`], or
/// [`ReceiverBandwidth::MetadataOnly`] for receiver configuration.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ReceiverBandwidth {
    MetadataOnly,
    AudioOnly,
    Lowest,
    #[default]
    Highest,
}

impl From<ReceiverBandwidth> for NDIlib_recv_bandwidth_e {
    fn from(bandwidth: ReceiverBandwidth) -> Self {
        match bandwidth {
            ReceiverBandwidth::MetadataOnly => {
                NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_metadata_only
            }
            ReceiverBandwidth::AudioOnly => {
                NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_audio_only
            }
            ReceiverBandwidth::Lowest => NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_lowest,
            ReceiverBandwidth::Highest => NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_highest,
        }
    }
}

#[derive(Debug, Default, Clone)]
pub struct ReceiverOptions {
    pub source_to_connect_to: Source,
    pub color_format: ReceiverColorFormat,
    pub bandwidth: ReceiverBandwidth,
    pub allow_video_fields: bool,
    pub ndi_recv_name: Option<String>,
}

#[repr(C)]
pub(crate) struct RawRecvCreateV3 {
    _source: RawSource,
    _name: Option<CString>,
    pub raw: NDIlib_recv_create_v3_t,
}

impl ReceiverOptions {
    /// Convert to raw format for FFI use
    ///
    /// # Safety
    ///
    /// The returned RawRecvCreateV3 struct uses #[repr(C)] to guarantee C-compatible layout
    /// for safe FFI interop with the NDI SDK.
    pub(crate) fn to_raw(&self) -> Result<RawRecvCreateV3> {
        let source = self.source_to_connect_to.to_raw()?;
        let name = self
            .ndi_recv_name
            .as_ref()
            .map(|n| CString::new(n.clone()))
            .transpose()
            .map_err(Error::InvalidCString)?;

        let p_ndi_recv_name = name.as_ref().map_or(ptr::null(), |n| n.as_ptr());
        let source_raw = source.raw;

        Ok(RawRecvCreateV3 {
            raw: NDIlib_recv_create_v3_t {
                source_to_connect_to: source_raw,
                color_format: self.color_format.into(),
                bandwidth: self.bandwidth.into(),
                allow_video_fields: self.allow_video_fields,
                p_ndi_recv_name,
            },
            _source: source,
            _name: name,
        })
    }

    /// Create a builder for configuring a receiver
    pub fn builder(source: Source) -> ReceiverOptionsBuilder {
        ReceiverOptionsBuilder::new(source)
    }
}

/// Builder for configuring a ReceiverOptions with ergonomic method chaining
#[derive(Debug, Clone)]
pub struct ReceiverOptionsBuilder {
    source_to_connect_to: Source,
    color_format: Option<ReceiverColorFormat>,
    bandwidth: Option<ReceiverBandwidth>,
    allow_video_fields: Option<bool>,
    ndi_recv_name: Option<String>,
}

impl ReceiverOptionsBuilder {
    /// Create a new builder with the specified source
    pub fn new(source: Source) -> Self {
        Self {
            source_to_connect_to: source,
            color_format: None,
            bandwidth: None,
            allow_video_fields: None,
            ndi_recv_name: None,
        }
    }

    /// Preset for capturing snapshots (low resolution, RGBA, lowest bandwidth).
    ///
    /// This preset is optimized for:
    /// - Image export and snapshot capture
    /// - AI/ML processing pipelines
    /// - Thumbnail generation
    /// - Low bandwidth environments
    ///
    /// Configuration:
    /// - Color format: `RGBX_RGBA` (compatible with image encoding)
    /// - Bandwidth: `Lowest` (reduces resolution and bitrate)
    /// - Video fields: Disabled (progressive frames only)
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, Finder, FinderOptions, ReceiverOptionsBuilder, Receiver};
    /// # use std::time::Duration;
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let finder = Finder::new(&ndi, &FinderOptions::default())?;
    /// # finder.wait_for_sources(Duration::from_secs(1))?;
    /// # let sources = finder.current_sources()?;
    /// let options = ReceiverOptionsBuilder::snapshot_preset(sources[0].clone())
    ///     .name("Snapshot Receiver")
    ///     .build();
    /// let receiver = Receiver::new(&ndi, &options)?;
    ///
    /// // Capture and encode in one line (requires image-encoding feature)
    /// #[cfg(feature = "image-encoding")]
    /// {
    ///     let frame = receiver.video().capture(Duration::from_secs(5))?;
    ///     let png_bytes = frame.encode_png()?;
    ///     std::fs::write("snapshot.png", &png_bytes)?;
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn snapshot_preset(source: Source) -> Self {
        Self::new(source)
            .color(ReceiverColorFormat::RGBX_RGBA)
            .bandwidth(ReceiverBandwidth::Lowest)
            .allow_video_fields(false)
    }

    /// Preset for high-quality video processing (full resolution, highest bandwidth).
    ///
    /// This preset is optimized for:
    /// - Professional video processing workflows
    /// - Recording and archival
    /// - Real-time video analysis requiring full quality
    /// - Broadcasting and production
    ///
    /// Configuration:
    /// - Color format: `RGBX_RGBA` (uncompressed, compatible with most tools)
    /// - Bandwidth: `Highest` (full resolution and bitrate)
    /// - Video fields: Enabled (supports interlaced sources)
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, Finder, FinderOptions, ReceiverOptionsBuilder, Receiver};
    /// # use std::time::Duration;
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let finder = Finder::new(&ndi, &FinderOptions::default())?;
    /// # finder.wait_for_sources(Duration::from_secs(1))?;
    /// # let sources = finder.current_sources()?;
    /// let options = ReceiverOptionsBuilder::high_quality_preset(sources[0].clone())
    ///     .name("High Quality Receiver")
    ///     .build();
    /// let receiver = Receiver::new(&ndi, &options)?;
    ///
    /// // Capture full quality frames
    /// let frame = receiver.video().capture(Duration::from_secs(5))?;
    /// println!("Captured {width}x{height} frame", width = frame.width(), height = frame.height());
    /// # Ok(())
    /// # }
    /// ```
    pub fn high_quality_preset(source: Source) -> Self {
        Self::new(source)
            .color(ReceiverColorFormat::RGBX_RGBA)
            .bandwidth(ReceiverBandwidth::Highest)
            .allow_video_fields(true)
    }

    /// Preset for metadata and connection monitoring only (no video/audio).
    ///
    /// This preset is optimized for:
    /// - Connection status tracking
    /// - PTZ control applications
    /// - Minimal bandwidth overhead
    ///
    /// Configuration:
    /// - Bandwidth: `MetadataOnly` (no video or audio data)
    /// - Color format: Default (not used for metadata-only)
    /// - Video fields: Disabled (not applicable)
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, Finder, FinderOptions, ReceiverOptionsBuilder, Receiver};
    /// # use std::time::Duration;
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let finder = Finder::new(&ndi, &FinderOptions::default())?;
    /// # finder.wait_for_sources(Duration::from_secs(1))?;
    /// # let sources = finder.current_sources()?;
    /// let options = ReceiverOptionsBuilder::monitoring_preset(sources[0].clone())
    ///     .name("Status Monitor")
    ///     .build();
    /// let receiver = Receiver::new(&ndi, &options)?;
    ///
    /// // Poll for status changes
    /// if let Some(status) = receiver.poll_status_change(Duration::from_millis(1000))? {
    ///     println!("Active connections: {}", status.connections);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn monitoring_preset(source: Source) -> Self {
        Self::new(source)
            .bandwidth(ReceiverBandwidth::MetadataOnly)
            .allow_video_fields(false)
    }

    /// Set the color format for received video
    #[must_use]
    pub fn color(mut self, fmt: ReceiverColorFormat) -> Self {
        self.color_format = Some(fmt);
        self
    }

    /// Set the bandwidth mode for the receiver
    #[must_use]
    pub fn bandwidth(mut self, bw: ReceiverBandwidth) -> Self {
        self.bandwidth = Some(bw);
        self
    }

    /// Configure whether to allow video fields
    #[must_use]
    pub fn allow_video_fields(mut self, allow: bool) -> Self {
        self.allow_video_fields = Some(allow);
        self
    }

    /// Set the name for this receiver
    #[must_use]
    pub fn name<S: Into<String>>(mut self, name: S) -> Self {
        self.ndi_recv_name = Some(name.into());
        self
    }

    /// Build the receiver options
    ///
    /// This method is infallible and simply applies defaults for any unset options.
    /// To create a `Receiver`, pass the resulting `ReceiverOptions` to `Receiver::new()`.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, ReceiverOptions, Receiver, Source};
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let source = Source::default();
    /// let options = ReceiverOptions::builder(source).build();
    /// let receiver = Receiver::new(&ndi, &options)?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn build(self) -> ReceiverOptions {
        ReceiverOptions {
            source_to_connect_to: self.source_to_connect_to,
            color_format: self.color_format.unwrap_or(ReceiverColorFormat::BGRX_BGRA),
            bandwidth: self.bandwidth.unwrap_or(ReceiverBandwidth::Highest),
            allow_video_fields: self.allow_video_fields.unwrap_or(true),
            ndi_recv_name: self.ndi_recv_name,
        }
    }
}

mod mode_sealed {
    pub trait Sealed {}
}

/// Sealed receiver capability mode. Common controls, audio, and metadata work in both modes.
pub trait ReceiverMode: mode_sealed::Sealed + Send + Sync {}

/// Ordinary receiver mode, permitting raw video capture.
#[derive(Debug)]
pub enum RawCapture {}
/// Receiver mode held exclusively by [`crate::FrameSync`], without raw video access.
#[derive(Debug)]
pub enum FrameSyncAttached {}
impl mode_sealed::Sealed for RawCapture {}
impl mode_sealed::Sealed for FrameSyncAttached {}
impl ReceiverMode for RawCapture {}
impl ReceiverMode for FrameSyncAttached {}

/// An owned NDI receiver. The default mode permits all raw capture kinds.
/// FrameSync privately changes the mode while retaining the same resource core.
pub struct Receiver<M: ReceiverMode = RawCapture> {
    pub(crate) core: ReceiverCore,
    _mode: PhantomData<M>,
}

pub(crate) struct ReceiverCore {
    pub(crate) instance: NDIlib_recv_instance_t,
    pub(crate) _ndi: NDI,
    source: Source,
    /// Serializes connection changes against in-flight captures.
    ///
    /// Capture calls take this lock *shared* (so video, audio, and metadata
    /// captures still run concurrently), while [`Receiver::reconnect`] takes it
    /// *exclusively*. This guarantees `NDIlib_recv_connect` never overlaps a
    /// `NDIlib_recv_capture_v3` on the same instance — the one combination the
    /// SDK does not make safe — without otherwise serializing capture.
    capture_guard: RwLock<()>,
    pending_status: AtomicBool,
}

impl Receiver {
    pub fn new(ndi: &NDI, create: &ReceiverOptions) -> Result<Self> {
        let create_raw = create.to_raw()?;
        let instance = unsafe { NDIlib_recv_create_v3(&create_raw.raw) };
        if instance.is_null() {
            Err(Error::InitializationFailed(
                "Failed to create NDI recv instance".into(),
            ))
        } else {
            Ok(Self {
                core: ReceiverCore {
                    instance,
                    _ndi: ndi.clone(),
                    source: create.source_to_connect_to.clone(),
                    capture_guard: RwLock::new(()),
                    pending_status: AtomicBool::new(false),
                },
                _mode: PhantomData,
            })
        }
    }

    /// Capture **video** frames from this receiver.
    ///
    /// Returns a [`Capture`] view whose verbs cover the three capture styles:
    ///
    /// - [`capture`](Capture::capture) — reliable owned capture that retries
    ///   across the NDI SDK's initial-sync warm-up.
    /// - [`try_capture`](Capture::try_capture) — a single owned poll, `Ok(None)`
    ///   when no frame is ready.
    /// - [`try_capture_ref`](Capture::try_capture_ref) — a single zero-copy poll
    ///   borrowing the SDK buffer in place.
    ///
    /// Video frames may carry per-row padding (e.g. RGBX/BGRX); see
    /// [`VideoFrame`](crate::VideoFrame) for how to read them correctly.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, Receiver, ReceiverOptions, Source, SourceAddress};
    /// # use std::time::Duration;
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let source = Source { name: "Test".into(), address: SourceAddress::None };
    /// # let options = ReceiverOptions::builder(source).build();
    /// # let receiver = Receiver::new(&ndi, &options)?;
    /// // Reliable owned capture
    /// let frame = receiver.video().capture(Duration::from_secs(5))?;
    /// println!("{}x{}", frame.width(), frame.height());
    ///
    /// // Zero-copy borrow
    /// if let Some(frame) = receiver.video().try_capture_ref(Duration::from_secs(1))? {
    ///     for plane in frame.planes() {
    ///         for row in plane.rows() {
    ///             println!("{} active bytes in place", row.len());
    ///         }
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[must_use = "the Capture view does nothing until a capture verb is called"]
    pub fn video(&self) -> Capture<'_, VideoKind> {
        Capture::new(&self.core)
    }
}

impl<M: ReceiverMode> Receiver<M> {
    pub(crate) fn into_mode<T: ReceiverMode>(self) -> Receiver<T> {
        Receiver {
            core: self.core,
            _mode: PhantomData,
        }
    }

    /// Publish this receiver's program/preview tally upstream.
    ///
    /// Returns the SDK's status flag unchanged. SDK implementations may report
    /// success while disconnected; use [`Self::is_connected`] to check the
    /// connection. The SDK retains tally state for future connections.
    pub fn set_tally(&self, tally: &Tally) -> bool {
        unsafe { NDIlib_recv_set_tally(self.core.instance, &tally.to_raw()) }
    }

    /// Send metadata upstream and return the SDK's status flag unchanged.
    ///
    /// This flag is not a connection query; use [`Self::is_connected`] for that.
    pub fn send_metadata(&self, metadata: &MetadataFrame) -> bool {
        let raw = metadata.to_raw();
        // The borrowed C string remains live through this synchronous SDK call.
        unsafe { NDIlib_recv_send_metadata(self.core.instance, &raw) }
    }

    /// Return whether the connected source supports PTZ commands.
    pub fn ptz_is_supported(&self) -> bool {
        unsafe { NDIlib_recv_ptz_is_supported(self.core.instance) }
    }

    /// Recall preset `0..=99` at a finite speed within `0.0..=1.0`.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_recall_preset(&self, preset: u32, speed: f32) -> Result<()> {
        let preset = validate_ptz_preset(i64::from(preset))?;
        validate_ptz_value("speed", speed, 0.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_recall_preset,
            preset,
            speed,
            format!("Failed to recall PTZ preset {preset} with speed {speed}")
        )
    }

    /// Set absolute zoom to a finite value within `0.0..=1.0`.
    /// Zero is zoomed in; one is zoomed out.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_zoom(&self, zoom_value: f32) -> Result<()> {
        validate_ptz_value("zoom_value", zoom_value, 0.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_zoom,
            zoom_value,
            format!("Failed to set PTZ zoom to {zoom_value}")
        )
    }

    /// Set zoom speed within `-1.0..=1.0` (finite).
    /// Negative moves outward; positive moves inward.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_zoom_speed(&self, zoom_speed: f32) -> Result<()> {
        validate_ptz_value("zoom_speed", zoom_speed, -1.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_zoom_speed,
            zoom_speed,
            format!("Failed to set PTZ zoom speed to {zoom_speed}")
        )
    }

    /// Set finite pan and tilt positions within `-1.0..=1.0`.
    /// Pan increases from left to right; tilt increases from bottom to top.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_pan_tilt(&self, pan_value: f32, tilt_value: f32) -> Result<()> {
        validate_ptz_value("pan_value", pan_value, -1.0, 1.0)?;
        validate_ptz_value("tilt_value", tilt_value, -1.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_pan_tilt,
            pan_value,
            tilt_value,
            format!("Failed to set PTZ pan/tilt to ({pan_value}, {tilt_value})")
        )
    }

    /// Set finite pan and tilt speeds within `-1.0..=1.0`.
    /// Positive pan moves left; positive tilt moves up. Zero stops motion.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_pan_tilt_speed(&self, pan_speed: f32, tilt_speed: f32) -> Result<()> {
        validate_ptz_value("pan_speed", pan_speed, -1.0, 1.0)?;
        validate_ptz_value("tilt_speed", tilt_speed, -1.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_pan_tilt_speed,
            pan_speed,
            tilt_speed,
            format!("Failed to set PTZ pan/tilt speed to ({pan_speed}, {tilt_speed})")
        )
    }

    /// Store the current camera settings as preset `0..=99`.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_store_preset(&self, preset_no: i32) -> Result<()> {
        let preset_no = validate_ptz_preset(i64::from(preset_no))?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_store_preset,
            preset_no,
            format!("Failed to store PTZ preset {preset_no}")
        )
    }

    /// Enable automatic focus.
    ///
    /// # Errors
    ///
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_auto_focus(&self) -> Result<()> {
        ptz_command!(
            self,
            NDIlib_recv_ptz_auto_focus,
            "Failed to enable PTZ auto focus"
        )
    }

    /// Set finite absolute focus within `0.0..=1.0`.
    /// Zero focuses at infinity; one focuses as close as possible.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_focus(&self, focus_value: f32) -> Result<()> {
        validate_ptz_value("focus_value", focus_value, 0.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_focus,
            focus_value,
            format!("Failed to set PTZ focus to {focus_value}")
        )
    }

    /// Set finite focus speed within `-1.0..=1.0`.
    /// Negative focuses outward; positive focuses inward.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_focus_speed(&self, focus_speed: f32) -> Result<()> {
        validate_ptz_value("focus_speed", focus_speed, -1.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_focus_speed,
            focus_speed,
            format!("Failed to set PTZ focus speed to {focus_speed}")
        )
    }

    /// Enable automatic white balance.
    ///
    /// # Errors
    ///
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_white_balance_auto(&self) -> Result<()> {
        ptz_command!(
            self,
            NDIlib_recv_ptz_white_balance_auto,
            "Failed to set PTZ auto white balance"
        )
    }

    /// Select the indoor white-balance preset.
    ///
    /// # Errors
    ///
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_white_balance_indoor(&self) -> Result<()> {
        ptz_command!(
            self,
            NDIlib_recv_ptz_white_balance_indoor,
            "Failed to set PTZ indoor white balance"
        )
    }

    /// Select the outdoor white-balance preset.
    ///
    /// # Errors
    ///
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_white_balance_outdoor(&self) -> Result<()> {
        ptz_command!(
            self,
            NDIlib_recv_ptz_white_balance_outdoor,
            "Failed to set PTZ outdoor white balance"
        )
    }

    /// Set white balance from the current image.
    ///
    /// # Errors
    ///
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_white_balance_oneshot(&self) -> Result<()> {
        ptz_command!(
            self,
            NDIlib_recv_ptz_white_balance_oneshot,
            "Failed to set PTZ oneshot white balance"
        )
    }

    /// Set finite red and blue white-balance levels within `0.0..=1.0`.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_white_balance_manual(&self, red: f32, blue: f32) -> Result<()> {
        validate_ptz_value("red", red, 0.0, 1.0)?;
        validate_ptz_value("blue", blue, 0.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_white_balance_manual,
            red,
            blue,
            format!("Failed to set PTZ manual white balance (red: {red}, blue: {blue})")
        )
    }

    /// Enable automatic exposure.
    ///
    /// # Errors
    ///
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_exposure_auto(&self) -> Result<()> {
        ptz_command!(
            self,
            NDIlib_recv_ptz_exposure_auto,
            "Failed to set PTZ auto exposure"
        )
    }

    /// Set finite exposure within `0.0..=1.0`, from dark to light.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_exposure_manual(&self, exposure_level: f32) -> Result<()> {
        validate_ptz_value("exposure_level", exposure_level, 0.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_exposure_manual,
            exposure_level,
            format!("Failed to set PTZ manual exposure to {exposure_level}")
        )
    }

    /// Set finite iris, gain, and shutter-speed levels within `0.0..=1.0`.
    /// Iris and gain increase from dark to light; shutter speed increases from slow to fast.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for out-of-range or non-finite
    /// parameters, before issuing an SDK command.
    /// Returns [`Error::PtzCommandFailed`] if the SDK rejects the command.
    pub fn ptz_exposure_manual_v2(&self, iris: f32, gain: f32, shutter_speed: f32) -> Result<()> {
        validate_ptz_value("iris", iris, 0.0, 1.0)?;
        validate_ptz_value("gain", gain, 0.0, 1.0)?;
        validate_ptz_value("shutter_speed", shutter_speed, 0.0, 1.0)?;
        ptz_command!(
            self,
            NDIlib_recv_ptz_exposure_manual_v2,
            iris,
            gain,
            shutter_speed,
            format!("Failed to set PTZ manual exposure v2 (iris: {iris}, gain: {gain}, shutter: {shutter_speed})")
        )
    }

    /// Capture **audio** frames from this receiver.
    ///
    /// Returns a [`Capture`] view; see [`Receiver::video`] for the three
    /// capture verbs it exposes ([`capture`](Capture::capture),
    /// [`try_capture`](Capture::try_capture),
    /// [`try_capture_ref`](Capture::try_capture_ref)).
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, Receiver, ReceiverOptions, Source, SourceAddress};
    /// # use std::time::Duration;
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let source = Source { name: "Test".into(), address: SourceAddress::None };
    /// # let options = ReceiverOptions::builder(source).build();
    /// # let receiver = Receiver::new(&ndi, &options)?;
    /// if let Some(audio) = receiver.audio().try_capture_ref(Duration::from_secs(1))? {
    ///     println!("{} channels, {} samples", audio.num_channels(), audio.num_samples());
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[must_use = "the Capture view does nothing until a capture verb is called"]
    pub fn audio(&self) -> Capture<'_, AudioKind> {
        Capture::new(&self.core)
    }

    /// Capture **metadata** frames from this receiver.
    ///
    /// Returns a [`Capture`] view; see [`Receiver::video`] for the three
    /// capture verbs it exposes ([`capture`](Capture::capture),
    /// [`try_capture`](Capture::try_capture),
    /// [`try_capture_ref`](Capture::try_capture_ref)).
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, Receiver, ReceiverOptions, Source, SourceAddress};
    /// # use std::time::Duration;
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let source = Source { name: "Test".into(), address: SourceAddress::None };
    /// # let options = ReceiverOptions::builder(source).build();
    /// # let receiver = Receiver::new(&ndi, &options)?;
    /// if let Some(meta) = receiver.metadata().try_capture(Duration::from_millis(100))? {
    ///     println!("metadata: {}", meta.data());
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[must_use = "the Capture view does nothing until a capture verb is called"]
    pub fn metadata(&self) -> Capture<'_, MetadataKind> {
        Capture::new(&self.core)
    }

    /// Check if the receiver is still connected to its source.
    ///
    /// Returns `true` if there is at least one active connection to the source,
    /// `false` otherwise. This can be used to detect when a source goes offline
    /// or becomes unavailable.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, Receiver, ReceiverOptions, Source, SourceAddress};
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let source = Source { name: "Test".into(), address: SourceAddress::None };
    /// # let options = ReceiverOptions::builder(source).build();
    /// # let receiver = Receiver::new(&ndi, &options)?;
    /// if receiver.is_connected() {
    ///     println!("Still connected to source");
    /// } else {
    ///     println!("Lost connection to source");
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn is_connected(&self) -> bool {
        unsafe { NDIlib_recv_get_no_connections(self.core.instance) > 0 }
    }

    /// Re-establish this receiver's connection to its original source,
    /// in place.
    ///
    /// Re-points the existing NDI receiver instance at the [`Source`] it
    /// was created for **without** destroying and recreating the
    /// receiver. This is intended for mid-session recovery: when a
    /// source's feed has gone silent (for example an encoder-bound
    /// NDI|HX camera that dropped its proxy stream under load), a
    /// reconnect forces a fresh connection attempt while avoiding the
    /// source-rediscovery round trip — and the discovery race — that
    /// tearing down and rebuilding the receiver would incur.
    ///
    /// Liveness can be observed via [`Self::is_connected`] and
    /// [`Self::connection_stats`] (`video_frames_received` resuming its
    /// climb indicates frames are flowing again).
    ///
    /// # Thread safety
    ///
    /// `NDIlib_recv_connect` is the one receive call the SDK does *not* make
    /// safe to run concurrently with `NDIlib_recv_capture_v3` on the same
    /// instance. This method therefore takes the receiver's capture guard
    /// exclusively: it **blocks until every in-flight capture on this receiver
    /// returns**, and holds off any capture that starts while it runs. Captures
    /// on the same receiver still run concurrently with each other; only a
    /// reconnect is exclusive. Because it can block for as long as a capture's
    /// timeout, prefer short capture timeouts on receivers you intend to
    /// recover, and call it off the async runtime — `AsyncReceiver::reconnect`
    /// does this for you.
    ///
    /// # Errors
    ///
    /// Returns an error only if the stored [`Source`] cannot be
    /// re-marshalled to its FFI representation; `NDIlib_recv_connect`
    /// itself reports no status, so a returned `Ok` means the reconnect was
    /// *issued*, not that the feed has recovered — confirm recovery via
    /// [`Self::connection_stats`].
    pub fn reconnect(&self) -> Result<()> {
        // Marshal before locking; this touches only `self.core.source`, not the
        // instance, so it need not hold off captures.
        let raw = self.core.source.to_raw()?;
        // Exclusive guard: waits for in-flight captures to drain and blocks new
        // ones, so `NDIlib_recv_connect` never overlaps `NDIlib_recv_capture_v3`.
        let _exclusive = self.core.capture_guard.write();
        // SAFETY: `self.core.instance` is a valid receiver instance for the
        // lifetime of `self`, and `raw` (which owns the backing
        // CStrings) outlives the call, so the pointers inside `raw.raw`
        // stay valid while the SDK copies the source descriptor.
        unsafe { NDIlib_recv_connect(self.core.instance, &raw.raw) };
        Ok(())
    }

    /// Get the source this receiver is connected to.
    ///
    /// Returns a reference to the [`Source`] that was specified when creating
    /// this receiver. This is useful for identifying which source a receiver
    /// is associated with when managing multiple receivers.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, Receiver, ReceiverOptions, Source, SourceAddress};
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let source = Source { name: "Test".into(), address: SourceAddress::None };
    /// # let options = ReceiverOptions::builder(source).build();
    /// # let receiver = Receiver::new(&ndi, &options)?;
    /// let source = receiver.source();
    /// println!("Connected to: {name}", name = source.name);
    /// # Ok(())
    /// # }
    /// ```
    pub fn source(&self) -> &Source {
        &self.core.source
    }

    /// Get connection and performance statistics for this receiver.
    ///
    /// Provides detailed statistics including:
    /// - Number of active connections
    /// - Total frames received (video, audio, metadata)
    /// - Dropped frames (video, audio, metadata)
    /// - Queued frames waiting to be processed
    ///
    /// This is useful for monitoring receiver health and diagnosing
    /// performance issues in production applications.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use grafton_ndi::{NDI, Receiver, ReceiverOptions, Source, SourceAddress};
    /// # fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let source = Source { name: "Test".into(), address: SourceAddress::None };
    /// # let options = ReceiverOptions::builder(source).build();
    /// # let receiver = Receiver::new(&ndi, &options)?;
    /// let stats = receiver.connection_stats();
    /// println!("Connections: {connections}", connections = stats.connections);
    /// println!("Video frames: {received} (dropped: {dropped})",
    ///          received = stats.video_frames_received,
    ///          dropped = stats.video_frames_dropped);
    /// println!("Frame drop rate: {rate:.2}%",
    ///          rate = stats.video_drop_percentage());
    /// # Ok(())
    /// # }
    /// ```
    pub fn connection_stats(&self) -> ConnectionStats {
        let connections = unsafe { NDIlib_recv_get_no_connections(self.core.instance) };

        let mut total = NDIlib_recv_performance_t::default();
        let mut dropped = NDIlib_recv_performance_t::default();
        unsafe {
            NDIlib_recv_get_performance(self.core.instance, &mut total, &mut dropped);
        }

        let mut queue = NDIlib_recv_queue_t::default();
        unsafe {
            NDIlib_recv_get_queue(self.core.instance, &mut queue);
        }

        ConnectionStats {
            connections: connections.max(0) as u32,
            video_frames_received: total.video_frames.max(0) as u64,
            audio_frames_received: total.audio_frames.max(0) as u64,
            metadata_frames_received: total.metadata_frames.max(0) as u64,
            video_frames_dropped: dropped.video_frames.max(0) as u64,
            audio_frames_dropped: dropped.audio_frames.max(0) as u64,
            metadata_frames_dropped: dropped.metadata_frames.max(0) as u64,
            video_frames_queued: queue.video_frames.max(0) as u32,
            audio_frames_queued: queue.audio_frames.max(0) as u32,
            metadata_frames_queued: queue.metadata_frames.max(0) as u32,
        }
    }

    /// Poll for SDK receiver status changes, including notifications consumed by media capture.
    ///
    /// Multiple unobserved notifications can coalesce. Concurrent status pollers
    /// compete to consume them; this is not a broadcast or event history. A pending
    /// notification avoids waiting, but publication cannot interrupt an SDK wait
    /// already in progress. Notifications are rechecked after lock/SDK waits.
    ///
    /// # Arguments
    ///
    /// * `timeout` - Maximum time to wait for status change.
    ///   Must not exceed [`crate::MAX_TIMEOUT`] (~49.7 days).
    ///
    /// # Returns
    ///
    /// * `Some(ReceiverStatus)` - Status has changed
    /// * `None` - Timeout occurred with no status change, including when a
    ///   reconnect holds the capture lock for the remaining budget.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] if `timeout` exceeds [`crate::MAX_TIMEOUT`],
    /// or [`Error::CaptureFailed`] if the SDK reports a lost connection.
    /// The budget includes lock contention; SDK calls may overshoot it.
    pub fn poll_status_change(&self, timeout: Duration) -> Result<Option<ReceiverStatus>> {
        self.core.poll_status_change(timeout)
    }
}

impl ReceiverCore {
    /// Acquire only within the caller's existing budget, including contention.
    fn capture_lock_until(&self, deadline: Instant) -> Option<RwLockReadGuard<'_, ()>> {
        self.capture_guard.try_read_until(deadline)
    }

    /// Single zero-copy poll backing every [`Capture::try_capture_ref`].
    ///
    /// Holds the shared capture guard for the duration of one
    /// `NDIlib_recv_capture_v3` call, so it runs concurrently with other
    /// captures but never overlaps a [`reconnect`](Receiver::reconnect).
    fn capture_ref_kind<K: CaptureKind>(&self, timeout: Duration) -> Result<Option<K::Ref<'_>>> {
        to_ms_checked(timeout)?;
        self.capture_ref_kind_until::<K>(Instant::now() + timeout)
    }

    fn capture_ref_kind_until<K: CaptureKind>(
        &self,
        deadline: Instant,
    ) -> Result<Option<K::Ref<'_>>> {
        let Some(_capture) = self.capture_lock_until(deadline) else {
            return Ok(None);
        };
        let timeout_ms = to_ms_checked(deadline.saturating_duration_since(Instant::now()))?;
        // SAFETY: self.instance is a valid NDI receiver instance.
        self.capture_result::<K>(unsafe { capture_raw::<K>(self.instance, timeout_ms) })
    }

    fn capture_result<'rx, K: CaptureKind>(
        &'rx self,
        result: CaptureResult<'rx, K>,
    ) -> Result<Option<K::Ref<'rx>>> {
        match result {
            CaptureResult::Frame(guard) => {
                // Validation (e.g. FourCC) happens during ref construction.
                let frame_ref = unsafe { K::make_ref(guard)? };
                Ok(Some(frame_ref))
            }
            CaptureResult::StatusChanged => {
                self.pending_status.store(true, Ordering::Release);
                Ok(None)
            }
            CaptureResult::None => Ok(None),
            CaptureResult::Error => Err(Error::CaptureFailed("Received an error frame".into())),
        }
    }

    /// Single owned poll backing every [`Capture::try_capture`].
    pub(crate) fn try_capture_kind<K: CaptureKind>(
        &self,
        timeout: Duration,
    ) -> Result<Option<K::Owned>> {
        match self.capture_ref_kind::<K>(timeout)? {
            Some(frame_ref) => Ok(Some(K::ref_to_owned(&frame_ref)?)),
            None => Ok(None),
        }
    }

    pub(crate) fn try_capture_kind_until<K: CaptureKind>(
        &self,
        deadline: Instant,
    ) -> Result<Option<K::Owned>> {
        match self.capture_ref_kind_until::<K>(deadline)? {
            Some(frame_ref) => Ok(Some(K::ref_to_owned(&frame_ref)?)),
            None => Ok(None),
        }
    }

    /// Retried owned capture backing every [`Capture::capture`].
    ///
    /// Polls [`try_capture_kind`](Self::try_capture_kind) under the default
    /// [`RetryPolicy`], absorbing the SDK's initial-sync misses where it returns
    /// `none` immediately instead of blocking for the full timeout.
    pub(crate) fn capture_kind<K: CaptureKind>(&self, timeout: Duration) -> Result<K::Owned> {
        to_ms_checked(timeout)?;
        self.capture_kind_until::<K>(Instant::now() + timeout)
    }

    pub(crate) fn capture_kind_until<K: CaptureKind>(&self, deadline: Instant) -> Result<K::Owned> {
        retry_capture(
            deadline.saturating_duration_since(Instant::now()),
            &RetryPolicy::default(),
            |poll| self.try_capture_kind_until::<K>(deadline.min(Instant::now() + poll)),
        )
    }

    pub fn poll_status_change(&self, timeout: Duration) -> Result<Option<ReceiverStatus>> {
        self.poll_status_with(timeout, |deadline| {
            let _capture = self.capture_lock_until(deadline)?;
            let timeout_ms = to_ms_checked(deadline.saturating_duration_since(Instant::now()))
                .expect("remaining validated timeout");
            // SAFETY: The live receiver and shared guard exclude reconnect;
            // null media slots request only notifications (Processing.NDI.Recv.h).
            Some(unsafe {
                NDIlib_recv_capture_v3(
                    self.instance,
                    ptr::null_mut(),
                    ptr::null_mut(),
                    ptr::null_mut(),
                    timeout_ms,
                )
            })
        })
    }

    fn take_pending_status(&self) -> Option<ReceiverStatus> {
        // Atomic swap linearizes competing pollers with capture publication.
        // A store after this take remains pending; no load/store pair can erase it.
        self.pending_status
            .swap(false, Ordering::AcqRel)
            .then(|| self.current_status())
    }

    fn current_status(&self) -> ReceiverStatus {
        ReceiverStatus {
            connections: unsafe { NDIlib_recv_get_no_connections(self.instance).max(0) as u32 },
        }
    }

    fn poll_status_with(
        &self,
        timeout: Duration,
        capture: impl FnOnce(Instant) -> Option<NDIlib_frame_type_e>,
    ) -> Result<Option<ReceiverStatus>> {
        to_ms_checked(timeout)?;
        let deadline = Instant::now() + timeout;
        if let Some(status) = self.take_pending_status() {
            return Ok(Some(status));
        }
        match capture(deadline) {
            Some(NDIlib_frame_type_e_NDIlib_frame_type_error) => {
                // Errors must not consume notifications published during the wait.
                Err(Error::CaptureFailed("Received an error frame".into()))
            }
            Some(NDIlib_frame_type_e_NDIlib_frame_type_status_change) => {
                Ok(Some(self.current_status()))
            }
            _ => Ok(self.take_pending_status()),
        }
    }
}

/// A typed view over a [`Receiver`] for capturing frames of one kind.
///
/// Created by [`Receiver::video`], [`Receiver::audio`], and
/// [`Receiver::metadata`]. The view is a cheap borrow of the receiver and does
/// nothing on its own — call one of its verbs to capture:
///
/// - [`capture`](Self::capture) — reliable owned capture with built-in retry.
///   **The primary method.** It absorbs the NDI SDK's initial-sync warm-up (the
///   first few polls after connecting may return `none` immediately instead of
///   blocking). It can be used in a continuous capture loop.
/// - [`try_capture`](Self::try_capture) — a single owned poll; `Ok(None)` when
///   no frame is ready. For manual polling where you handle timing yourself.
/// - [`try_capture_ref`](Self::try_capture_ref) — a single zero-copy poll that
///   borrows the SDK's buffer in place (no allocation, no memcpy). The
///   recommended API for performance-critical, in-place processing: for a
///   1920×1080 BGRA frame it avoids ~8.3 MB of copying per frame (~475 MB/s at
///   60 fps). Convert to an owned frame with `to_owned` when you need to keep or
///   send it.
///
/// Media capture retains status notifications for [`Receiver::poll_status_change`],
/// even when retries later time out.
///
/// Every verb holds a shared capture guard for the duration of a single SDK
/// call, so captures of different kinds run concurrently with each other but
/// never overlap a [`Receiver::reconnect`]. Each timeout budget starts when the
/// verb is called and includes waiting for that guard. A zero timeout never
/// waits for the guard; if it is unavailable, no SDK capture is attempted.
/// SDK calls and owned-frame copying may finish after the budget expires, so
/// the timeout is not a strict wall-clock completion guarantee.
///
/// Only the ordinary receiver mode exposes a video view. FrameSync owns video
/// capture until [`crate::FrameSync::into_receiver`] restores the ordinary mode.
pub struct Capture<'rx, K: CaptureKind> {
    rx: &'rx ReceiverCore,
    _kind: PhantomData<K>,
}

impl<'rx, K: CaptureKind> Capture<'rx, K> {
    fn new(rx: &'rx ReceiverCore) -> Self {
        Self {
            rx,
            _kind: PhantomData,
        }
    }

    /// Reliable owned capture: blocks up to `timeout`, retrying across the NDI
    /// SDK's initial-sync warm-up.
    ///
    /// `timeout` is a total retry budget; each individual SDK poll is capped to
    /// the remaining budget, and [`Duration::ZERO`] performs one non-blocking
    /// attempt, skipping the SDK call if the capture guard is unavailable.
    /// Waiting for a reconnect lock consumes the same budget. The SDK itself may return after its requested timeout, so this
    /// budget is not a strict wall-clock completion guarantee.
    ///
    /// # Arguments
    ///
    /// * `timeout` - Total time to wait for a frame. Must not exceed
    ///   [`crate::MAX_TIMEOUT`] (~49.7 days).
    ///
    /// # Errors
    ///
    /// Returns [`Error::FrameTimeout`] if no frame arrives within `timeout`;
    /// returns [`Error::InvalidConfiguration`] for an excessive timeout. SDK and frame-validation errors
    /// propagate from the capture itself.
    pub fn capture(&self, timeout: Duration) -> Result<K::Owned> {
        self.rx.capture_kind::<K>(timeout)
    }

    /// Single owned poll: returns `Ok(None)` if no frame is available within
    /// `timeout`.
    ///
    /// Prefer [`capture`](Self::capture) for reliable capture; this variant does
    /// not retry the SDK's warm-up misses. The timeout includes waiting for a
    /// reconnect lock; see [`Capture`] for the shared timeout contract.
    ///
    /// # Arguments
    ///
    /// * `timeout` - Maximum time to wait for a frame. Must not exceed
    ///   [`crate::MAX_TIMEOUT`] (~49.7 days).
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for an excessive timeout. SDK connection errors return
    /// [`Error::CaptureFailed`]; malformed frame descriptors also return errors.
    pub fn try_capture(&self, timeout: Duration) -> Result<Option<K::Owned>> {
        self.rx.try_capture_kind::<K>(timeout)
    }

    /// Single zero-copy poll: returns a borrowed view of the SDK's buffer, or
    /// `Ok(None)` if no frame is available within `timeout`.
    ///
    /// The returned reference borrows the [`Receiver`] (not this temporary
    /// view), so it may outlive the `Capture` while keeping the receiver
    /// borrowed. The timeout includes waiting for a reconnect lock; see
    /// [`Capture`] for the shared timeout contract.
    ///
    /// # Arguments
    ///
    /// * `timeout` - Maximum time to wait for a frame. Must not exceed
    ///   [`crate::MAX_TIMEOUT`] (~49.7 days).
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidConfiguration`] for an excessive timeout. SDK connection errors return
    /// [`Error::CaptureFailed`]; malformed frame descriptors also return errors.
    pub fn try_capture_ref(&self, timeout: Duration) -> Result<Option<K::Ref<'rx>>> {
        // Bind through `'rx` so the returned borrow is tied to the receiver, not
        // to this short-lived view.
        let rx: &'rx ReceiverCore = self.rx;
        rx.capture_ref_kind::<K>(timeout)
    }
}

impl Drop for ReceiverCore {
    fn drop(&mut self) {
        unsafe {
            NDIlib_recv_destroy(self.instance);
        }
        #[cfg(test)]
        lifecycle_test_events::record("receiver");
    }
}

/// # Safety
///
/// The NDI 6 SDK documentation explicitly states that recv operations are thread-safe.
/// `NDIlib_recv_capture_v3` and related functions use internal synchronization.
/// The Receiver struct only holds an opaque pointer returned by the SDK, and the SDK
/// guarantees that this pointer can be safely moved between threads.
unsafe impl Send for ReceiverCore {}

/// # Safety
///
/// The NDI 6 SDK documentation guarantees that `NDIlib_recv_capture_v3` is internally
/// synchronized and can be called concurrently from multiple threads. This is explicitly
/// mentioned in the SDK manual's thread safety section. The capture verbs (via
/// [`Receiver::video`], [`Receiver::audio`], and [`Receiver::metadata`]) can be
/// safely called from multiple threads simultaneously. The wrapper's capture
/// guard excludes reconnect calls from receiver capture calls. Receiver modes
/// prevent raw video capture while a FrameSync owns this core.
/// Borrowed frame guards retain the receiver borrow until they free their buffers.
unsafe impl Sync for ReceiverCore {}

/// An SDK status-change notification and the current connection count.
///
/// The event may also indicate changed source capabilities or web control;
/// the SDK does not identify a specific changed field. Tally is published
/// upstream through [`Receiver::set_tally`], rather than received here.
#[derive(Debug, Clone)]
pub struct ReceiverStatus {
    /// Current number of connected sources (normally zero or one).
    pub connections: u32,
}

#[derive(Debug, Clone)]
pub struct Tally {
    pub on_program: bool,
    pub on_preview: bool,
}

impl Tally {
    pub fn new(on_program: bool, on_preview: bool) -> Self {
        Tally {
            on_program,
            on_preview,
        }
    }

    pub(crate) fn to_raw(&self) -> NDIlib_tally_t {
        NDIlib_tally_t {
            on_program: self.on_program,
            on_preview: self.on_preview,
        }
    }
}

/// Connection and performance statistics for a receiver.
///
/// Provides detailed metrics about receiver health including connection count,
/// frame counts, and drop rates. Useful for monitoring and diagnostics.
#[derive(Debug, Clone)]
pub struct ConnectionStats {
    /// Number of active connections to this receiver
    pub connections: u32,

    /// Total number of video frames received
    pub video_frames_received: u64,

    /// Total number of audio frames received
    pub audio_frames_received: u64,

    /// Total number of metadata frames received
    pub metadata_frames_received: u64,

    /// Number of video frames dropped due to buffer overflow or processing delays
    pub video_frames_dropped: u64,

    /// Number of audio frames dropped
    pub audio_frames_dropped: u64,

    /// Number of metadata frames dropped
    pub metadata_frames_dropped: u64,

    /// Number of video frames currently queued for processing
    pub video_frames_queued: u32,

    /// Number of audio frames currently queued
    pub audio_frames_queued: u32,

    /// Number of metadata frames currently queued
    pub metadata_frames_queued: u32,
}

impl ConnectionStats {
    /// Calculate video frame drop percentage.
    ///
    /// Returns the percentage of video frames that were dropped out of the total
    /// received + dropped. Returns 0.0 if both counters are zero.
    ///
    /// # Examples
    ///
    /// ```
    /// # use grafton_ndi::ConnectionStats;
    /// let stats = ConnectionStats {
    ///     connections: 1,
    ///     video_frames_received: 900,
    ///     video_frames_dropped: 100,
    ///     audio_frames_received: 0,
    ///     audio_frames_dropped: 0,
    ///     metadata_frames_received: 0,
    ///     metadata_frames_dropped: 0,
    ///     video_frames_queued: 5,
    ///     audio_frames_queued: 0,
    ///     metadata_frames_queued: 0,
    /// };
    /// assert_eq!(stats.video_drop_percentage(), 10.0);
    /// ```
    pub fn video_drop_percentage(&self) -> f64 {
        drop_percentage(self.video_frames_received, self.video_frames_dropped)
    }

    /// Calculate audio frame drop percentage.
    ///
    /// Returns the percentage of audio frames that were dropped out of the total
    /// received + dropped. Returns 0.0 if both counters are zero.
    pub fn audio_drop_percentage(&self) -> f64 {
        drop_percentage(self.audio_frames_received, self.audio_frames_dropped)
    }

    /// Calculate metadata frame drop percentage.
    ///
    /// Returns the percentage of metadata frames that were dropped out of the total
    /// received + dropped. Returns 0.0 if both counters are zero.
    pub fn metadata_drop_percentage(&self) -> f64 {
        drop_percentage(self.metadata_frames_received, self.metadata_frames_dropped)
    }

    /// Check if the receiver is currently connected.
    ///
    /// Returns `true` if there is at least one active connection.
    pub fn is_connected(&self) -> bool {
        self.connections > 0
    }
}

fn drop_percentage(received: u64, dropped: u64) -> f64 {
    let total = u128::from(received) + u128::from(dropped);
    if total == 0 {
        0.0
    } else {
        (dropped as f64 / total as f64) * 100.0
    }
}

#[cfg(test)]
pub(crate) mod lifecycle_test_events {
    use std::cell::RefCell;
    std::thread_local! {
        static EVENTS: RefCell<Option<Vec<&'static str>>> = const { RefCell::new(None) };
    }
    pub(crate) fn record(event: &'static str) {
        EVENTS.with(|events| {
            if let Some(events) = &mut *events.borrow_mut() {
                events.push(event);
            }
        });
    }
    pub(crate) fn track(action: impl FnOnce()) -> Vec<&'static str> {
        EVENTS.with(|events| *events.borrow_mut() = Some(Vec::new()));
        action();
        EVENTS.with(|events| events.borrow_mut().take().unwrap())
    }
}

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

    use std::cell::Cell;

    fn test_source() -> Source {
        Source::default()
    }

    #[derive(Default)]
    struct FakeClock {
        elapsed: Cell<Duration>,
    }

    impl FakeClock {
        fn elapsed(&self) -> Duration {
            self.elapsed.get()
        }

        fn advance(&self, duration: Duration) {
            self.elapsed.set(self.elapsed.get() + duration);
        }
    }

    fn assert_raw_receiver_modes(
        raw: &RawRecvCreateV3,
        expected_color_format: NDIlib_recv_color_format_e,
        expected_bandwidth: NDIlib_recv_bandwidth_e,
    ) {
        assert_eq!(raw.raw.color_format, expected_color_format);
        assert_eq!(raw.raw.bandwidth, expected_bandwidth);
        assert_ne!(
            raw.raw.color_format,
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_max
        );
        assert_ne!(
            raw.raw.bandwidth,
            NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_max
        );
    }

    #[test]
    fn media_status_is_retained_coalesced_and_consumed_once() {
        let ndi = NDI::new().unwrap();
        let receiver = Receiver::new(&ndi, &ReceiverOptions::default()).unwrap();
        fn check<K: CaptureKind>(receiver: &Receiver) {
            let (_, allocations) = crate::test_alloc::allocation_count(|| {
                for _ in 0..3 {
                    assert!(receiver
                        .core
                        .capture_result::<K>(CaptureResult::StatusChanged)
                        .unwrap()
                        .is_none());
                }
                assert!(receiver
                    .core
                    .capture_result::<K>(CaptureResult::None)
                    .unwrap()
                    .is_none());
                assert!(receiver
                    .core
                    .poll_status_with(Duration::ZERO, |_| panic!("pending skips SDK"))
                    .unwrap()
                    .is_some());
                assert!(receiver
                    .core
                    .poll_status_with(Duration::ZERO, |_| Some(
                        NDIlib_frame_type_e_NDIlib_frame_type_none
                    ))
                    .unwrap()
                    .is_none());
            });
            assert_eq!(allocations, 0);
        }
        check::<VideoKind>(&receiver);
        check::<AudioKind>(&receiver);
        check::<MetadataKind>(&receiver);
    }

    #[test]
    fn status_wait_rechecks_publication_and_preserves_it_on_errors() {
        let ndi = NDI::new().unwrap();
        let receiver = Receiver::new(&ndi, &ReceiverOptions::default()).unwrap();
        for outcome in [None, Some(NDIlib_frame_type_e_NDIlib_frame_type_none)] {
            assert!(receiver
                .core
                .poll_status_with(Duration::ZERO, |_| {
                    receiver.core.pending_status.store(true, Ordering::Release);
                    outcome
                })
                .unwrap()
                .is_some());
            assert!(!receiver.core.pending_status.load(Ordering::Acquire));
        }
        assert!(matches!(
            receiver.core.poll_status_with(Duration::ZERO, |_| {
                receiver.core.pending_status.store(true, Ordering::Release);
                Some(NDIlib_frame_type_e_NDIlib_frame_type_error)
            }),
            Err(Error::CaptureFailed(_))
        ));
        assert!(receiver.poll_status_change(Duration::MAX).is_err());
        assert!(receiver.core.pending_status.load(Ordering::Acquire));
        assert!(receiver
            .poll_status_change(Duration::ZERO)
            .unwrap()
            .is_some());
        assert!(receiver
            .core
            .poll_status_with(Duration::ZERO, |_| Some(
                NDIlib_frame_type_e_NDIlib_frame_type_status_change
            ))
            .unwrap()
            .is_some());
    }

    #[test]
    fn concurrent_status_pollers_take_once_and_later_publications_survive() {
        let ndi = NDI::new().unwrap();
        let receiver = Receiver::new(&ndi, &ReceiverOptions::default()).unwrap();
        receiver.core.pending_status.store(true, Ordering::Release);
        let barrier = std::sync::Barrier::new(3);
        std::thread::scope(|scope| {
            let handles: Vec<_> = (0..2)
                .map(|_| {
                    let receiver = &receiver;
                    let barrier = &barrier;
                    scope.spawn(move || {
                        barrier.wait();
                        receiver.core.take_pending_status().is_some()
                    })
                })
                .collect();
            barrier.wait();
            assert_eq!(
                handles
                    .into_iter()
                    .map(|h| usize::from(h.join().unwrap()))
                    .sum::<usize>(),
                1
            );
        });
        receiver.core.pending_status.store(true, Ordering::Release);
        assert!(receiver.core.take_pending_status().is_some());
    }

    #[test]
    fn status_publication_during_waits_is_reported_and_newer_publication_survives() {
        let ndi = NDI::new().unwrap();
        let receiver = Receiver::new(&ndi, &ReceiverOptions::default()).unwrap();
        let (entered_tx, entered_rx) = std::sync::mpsc::channel();
        let (resume_tx, resume_rx) = std::sync::mpsc::channel();
        std::thread::scope(|scope| {
            let receiver = &receiver;
            let worker = scope.spawn(move || {
                receiver.core.poll_status_with(Duration::from_secs(1), |_| {
                    entered_tx.send(()).unwrap();
                    resume_rx.recv().unwrap();
                    Some(NDIlib_frame_type_e_NDIlib_frame_type_none)
                })
            });
            entered_rx.recv_timeout(Duration::from_secs(2)).unwrap();
            receiver
                .core
                .capture_result::<MetadataKind>(CaptureResult::StatusChanged)
                .unwrap();
            resume_tx.send(()).unwrap();
            assert!(worker.join().unwrap().unwrap().is_some());
        });
        assert!(receiver.core.take_pending_status().is_none());
        receiver
            .core
            .capture_result::<AudioKind>(CaptureResult::StatusChanged)
            .unwrap();
        assert!(receiver.core.take_pending_status().is_some());

        // A pending notification avoids even a contended reconnect lock.
        let _exclusive = receiver.core.capture_guard.write();
        receiver
            .core
            .capture_result::<VideoKind>(CaptureResult::StatusChanged)
            .unwrap();
        assert!(receiver
            .poll_status_change(Duration::ZERO)
            .unwrap()
            .is_some());
        // Publication during a failed lock wait is also rechecked, without a
        // second SDK call or a fresh deadline.
        assert!(receiver
            .core
            .poll_status_with(Duration::ZERO, |deadline| {
                receiver
                    .core
                    .capture_result::<VideoKind>(CaptureResult::StatusChanged)
                    .unwrap();
                assert!(receiver.core.capture_lock_until(deadline).is_none());
                None
            })
            .unwrap()
            .is_some());
    }

    #[test]
    fn receiver_mode_transitions_preserve_pending_status_and_core_size() {
        let ndi = NDI::new().unwrap();
        let receiver = Receiver::new(&ndi, &ReceiverOptions::default()).unwrap();
        receiver.core.pending_status.store(true, Ordering::Release);
        let fs = crate::FrameSync::new(receiver).unwrap();
        assert!(fs
            .receiver()
            .poll_status_change(Duration::ZERO)
            .unwrap()
            .is_some());
        fs.receiver()
            .core
            .pending_status
            .store(true, Ordering::Release);
        let receiver = fs.into_receiver();
        assert!(receiver
            .poll_status_change(Duration::ZERO)
            .unwrap()
            .is_some());
        assert!(!receiver.core.pending_status.load(Ordering::Acquire));
        assert_eq!(
            std::mem::size_of::<Receiver>(),
            std::mem::size_of::<ReceiverCore>()
        );
        assert_eq!(
            std::mem::size_of::<Receiver<FrameSyncAttached>>(),
            std::mem::size_of::<ReceiverCore>()
        );
    }

    #[test]
    fn drop_rates_handle_extreme_public_counters() {
        assert_eq!(drop_percentage(0, 0), 0.0);
        assert_eq!(drop_percentage(900, 100), 10.0);
        assert_eq!(drop_percentage(u64::MAX, u64::MAX), 50.0);
        assert_eq!(drop_percentage(0, u64::MAX), 100.0);
    }

    #[test]
    fn ptz_rejects_invalid_commands_before_calling_sdk() {
        let ndi = NDI::new().unwrap();
        let receiver = Receiver::new(&ndi, &ReceiverOptions::default()).unwrap();
        for result in [
            receiver.ptz_recall_preset(u32::MAX, 0.5),
            receiver.ptz_recall_preset(100, 0.5),
            receiver.ptz_recall_preset(0, f32::NAN),
            receiver.ptz_store_preset(-1),
            receiver.ptz_zoom(f32::INFINITY),
            receiver.ptz_zoom_speed(-1.1),
            receiver.ptz_pan_tilt(0.0, 1.1),
            receiver.ptz_pan_tilt_speed(f32::NAN, 0.0),
            receiver.ptz_focus(-0.1),
            receiver.ptz_focus_speed(1.1),
            receiver.ptz_white_balance_manual(0.0, 1.1),
            receiver.ptz_exposure_manual(f32::NEG_INFINITY),
            receiver.ptz_exposure_manual_v2(0.5, f32::NAN, 0.5),
        ] {
            assert!(matches!(result, Err(Error::InvalidConfiguration(_))));
        }
        assert_eq!(validate_ptz_preset(0).unwrap(), 0);
        assert_eq!(validate_ptz_preset(99).unwrap(), 99);
        assert!(validate_ptz_value("speed", -1.0, -1.0, 1.0).is_ok());
        assert!(validate_ptz_value("speed", 1.0, -1.0, 1.0).is_ok());
    }

    #[test]
    fn outbound_controls_accept_invocation_without_a_source() {
        let ndi = NDI::new().unwrap();
        let receiver = Receiver::new(&ndi, &ReceiverOptions::default()).unwrap();
        // Runtime implementations differ in their status flags when no source
        // is connected. These controls must remain callable in that state.
        let _sdk_status = receiver.set_tally(&Tally::new(true, false));
        let frame = MetadataFrame::with_data("<ndi_text>é</ndi_text>", 42).unwrap();
        let pointer = frame.to_raw().p_data;
        for _ in 0..3 {
            let _sdk_status = receiver.send_metadata(&frame);
            assert_eq!(frame.to_raw().p_data, pointer);
        }
    }

    #[test]
    fn capture_contention_obeys_zero_and_bounded_budgets() {
        let ndi = NDI::new().unwrap();
        let receiver = Receiver::new(&ndi, &ReceiverOptions::default()).unwrap();
        let exclusive = receiver.core.capture_guard.write();
        std::thread::scope(|scope| {
            let (done, completed) = std::sync::mpsc::channel();
            let receiver = &receiver;
            let worker = scope.spawn(move || {
                assert!(receiver
                    .video()
                    .try_capture(Duration::ZERO)
                    .unwrap()
                    .is_none());
                assert!(receiver
                    .poll_status_change(Duration::ZERO)
                    .unwrap()
                    .is_none());
                assert!(receiver
                    .audio()
                    .try_capture(Duration::from_millis(5))
                    .unwrap()
                    .is_none());
                assert!(matches!(
                    receiver.video().capture(Duration::from_millis(5)),
                    Err(Error::FrameTimeout { .. })
                ));
                done.send(()).unwrap();
            });
            // The writer remains held until the worker has returned: a stale
            // unbounded read would deadlock instead of finishing within budget.
            let result = completed.recv_timeout(Duration::from_secs(2));
            drop(exclusive);
            worker.join().unwrap();
            result.unwrap();
        });
    }

    #[test]
    fn receiver_color_format_maps_to_exact_sdk_values() {
        assert_eq!(
            NDIlib_recv_color_format_e::from(ReceiverColorFormat::BGRX_BGRA),
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_BGRX_BGRA
        );
        assert_eq!(
            NDIlib_recv_color_format_e::from(ReceiverColorFormat::UYVY_BGRA),
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_UYVY_BGRA
        );
        assert_eq!(
            NDIlib_recv_color_format_e::from(ReceiverColorFormat::RGBX_RGBA),
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_RGBX_RGBA
        );
        assert_eq!(
            NDIlib_recv_color_format_e::from(ReceiverColorFormat::UYVY_RGBA),
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_UYVY_RGBA
        );
        assert_eq!(
            NDIlib_recv_color_format_e::from(ReceiverColorFormat::Fastest),
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_fastest
        );
        assert_eq!(
            NDIlib_recv_color_format_e::from(ReceiverColorFormat::Best),
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_best
        );
    }

    #[test]
    fn receiver_bandwidth_maps_to_exact_sdk_values() {
        assert_eq!(
            NDIlib_recv_bandwidth_e::from(ReceiverBandwidth::MetadataOnly),
            NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_metadata_only
        );
        assert_eq!(
            NDIlib_recv_bandwidth_e::from(ReceiverBandwidth::AudioOnly),
            NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_audio_only
        );
        assert_eq!(
            NDIlib_recv_bandwidth_e::from(ReceiverBandwidth::Lowest),
            NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_lowest
        );
        assert_eq!(
            NDIlib_recv_bandwidth_e::from(ReceiverBandwidth::Highest),
            NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_highest
        );
    }

    #[test]
    fn receiver_options_defaults_emit_operational_raw_modes() {
        let raw = ReceiverOptions::builder(test_source())
            .build()
            .to_raw()
            .unwrap();

        assert_raw_receiver_modes(
            &raw,
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_BGRX_BGRA,
            NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_highest,
        );
        assert!(raw.raw.allow_video_fields);
    }

    #[test]
    fn snapshot_preset_emits_operational_raw_modes() {
        let raw = ReceiverOptionsBuilder::snapshot_preset(test_source())
            .build()
            .to_raw()
            .unwrap();

        assert_raw_receiver_modes(
            &raw,
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_RGBX_RGBA,
            NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_lowest,
        );
        assert!(!raw.raw.allow_video_fields);
    }

    #[test]
    fn high_quality_preset_emits_operational_raw_modes() {
        let raw = ReceiverOptionsBuilder::high_quality_preset(test_source())
            .build()
            .to_raw()
            .unwrap();

        assert_raw_receiver_modes(
            &raw,
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_RGBX_RGBA,
            NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_highest,
        );
        assert!(raw.raw.allow_video_fields);
    }

    #[test]
    fn monitoring_preset_emits_operational_raw_modes() {
        let raw = ReceiverOptionsBuilder::monitoring_preset(test_source())
            .build()
            .to_raw()
            .unwrap();

        assert_raw_receiver_modes(
            &raw,
            NDIlib_recv_color_format_e_NDIlib_recv_color_format_BGRX_BGRA,
            NDIlib_recv_bandwidth_e_NDIlib_recv_bandwidth_metadata_only,
        );
        assert!(!raw.raw.allow_video_fields);
    }

    #[test]
    fn retry_caps_sub_poll_timeout_to_total_budget() {
        let clock = FakeClock::default();
        let policy = RetryPolicy {
            poll_interval: Duration::from_millis(100),
            sleep_between: Duration::from_millis(10),
        };
        let mut polls = Vec::new();

        let result = retry_capture_with_clock(
            Duration::from_millis(25),
            &policy,
            |poll| {
                polls.push(poll);
                Ok(Some(42))
            },
            || clock.elapsed(),
            |_| {},
        );

        assert_eq!(result.unwrap(), 42);
        assert_eq!(polls, [Duration::from_millis(25)]);
    }

    #[test]
    fn retry_shrinks_later_poll_to_remaining_budget() {
        let clock = FakeClock::default();
        let policy = RetryPolicy {
            poll_interval: Duration::from_millis(40),
            sleep_between: Duration::from_millis(10),
        };
        let mut polls = Vec::new();
        let mut sleeps = Vec::new();

        let result: Result<i32> = retry_capture_with_clock(
            Duration::from_millis(50),
            &policy,
            |poll| {
                polls.push(poll);
                if polls.len() == 1 {
                    clock.advance(Duration::from_millis(5));
                } else {
                    clock.advance(poll);
                }
                Ok(None)
            },
            || clock.elapsed(),
            |sleep| {
                sleeps.push(sleep);
                clock.advance(sleep);
            },
        );

        assert!(matches!(
            result,
            Err(Error::FrameTimeout { attempts: 2, .. })
        ));
        assert_eq!(
            polls,
            [Duration::from_millis(40), Duration::from_millis(35)]
        );
        assert_eq!(sleeps, [Duration::from_millis(10)]);
    }

    #[test]
    fn retry_zero_timeout_success_makes_one_nonblocking_attempt() {
        let clock = FakeClock::default();
        let mut polls = Vec::new();

        let result = retry_capture_with_clock(
            Duration::ZERO,
            &RetryPolicy::default(),
            |poll| {
                polls.push(poll);
                Ok(Some(42))
            },
            || clock.elapsed(),
            |_| panic!("zero-timeout success should not sleep"),
        );

        assert_eq!(result.unwrap(), 42);
        assert_eq!(polls, [Duration::ZERO]);
    }

    #[test]
    fn retry_zero_timeout_miss_times_out_after_one_attempt() {
        let clock = FakeClock::default();
        let mut polls = Vec::new();

        let result: Result<i32> = retry_capture_with_clock(
            Duration::ZERO,
            &RetryPolicy::default(),
            |poll| {
                polls.push(poll);
                Ok(None)
            },
            || clock.elapsed(),
            |_| panic!("zero-timeout miss should not sleep"),
        );

        match result {
            Err(Error::FrameTimeout { attempts, elapsed }) => {
                assert_eq!(attempts, 1);
                assert_eq!(elapsed, Duration::ZERO);
            }
            _ => panic!("Expected FrameTimeout error"),
        }
        assert_eq!(polls, [Duration::ZERO]);
    }

    #[test]
    fn retry_timeout_counts_only_actual_capture_attempts() {
        let clock = FakeClock::default();
        let policy = RetryPolicy {
            poll_interval: Duration::from_millis(8),
            sleep_between: Duration::from_millis(5),
        };
        let mut polls = Vec::new();

        let result: Result<i32> = retry_capture_with_clock(
            Duration::from_millis(10),
            &policy,
            |poll| {
                polls.push(poll);
                clock.advance(Duration::from_millis(10));
                Ok(None)
            },
            || clock.elapsed(),
            |_| panic!("expired timeout should not sleep"),
        );

        match result {
            Err(Error::FrameTimeout { attempts, elapsed }) => {
                assert_eq!(attempts, 1);
                assert_eq!(elapsed, Duration::from_millis(10));
            }
            _ => panic!("Expected FrameTimeout error"),
        }
        assert_eq!(polls, [Duration::from_millis(8)]);
    }

    #[test]
    fn retry_caps_sleep_to_remaining_budget() {
        let clock = FakeClock::default();
        let policy = RetryPolicy {
            poll_interval: Duration::from_millis(8),
            sleep_between: Duration::from_millis(5),
        };
        let mut sleeps = Vec::new();

        let result: Result<i32> = retry_capture_with_clock(
            Duration::from_millis(10),
            &policy,
            |poll| {
                assert_eq!(poll, Duration::from_millis(8));
                clock.advance(Duration::from_millis(8));
                Ok(None)
            },
            || clock.elapsed(),
            |sleep| {
                sleeps.push(sleep);
                clock.advance(sleep);
            },
        );

        assert!(matches!(
            result,
            Err(Error::FrameTimeout { attempts: 1, .. })
        ));
        assert_eq!(sleeps, [Duration::from_millis(2)]);
    }

    #[test]
    fn retry_validates_total_timeout_before_attempts() {
        let attempts = Cell::new(0);

        let result: Result<i32> = retry_capture(
            crate::MAX_TIMEOUT + Duration::from_nanos(1),
            &RetryPolicy::default(),
            |_| {
                attempts.set(attempts.get() + 1);
                Ok(Some(42))
            },
        );

        assert!(matches!(result, Err(Error::InvalidConfiguration(_))));
        assert_eq!(attempts.get(), 0);
    }

    #[test]
    fn retry_succeeds_after_retry_before_timeout() {
        let clock = FakeClock::default();
        let attempts = Cell::new(0);

        let result = retry_capture_with_clock(
            Duration::from_millis(50),
            &RetryPolicy::default(),
            |_| {
                attempts.set(attempts.get() + 1);
                if attempts.get() == 1 {
                    Ok(None)
                } else {
                    Ok(Some(42))
                }
            },
            || clock.elapsed(),
            |sleep| clock.advance(sleep),
        );

        assert_eq!(result.unwrap(), 42);
        assert_eq!(attempts.get(), 2);
    }

    #[test]
    fn retry_propagates_error() {
        let clock = FakeClock::default();
        let attempts = Cell::new(0);
        let mut sleeps = Vec::new();

        let result: Result<i32> = retry_capture_with_clock(
            Duration::from_secs(1),
            &RetryPolicy::default(),
            |_| {
                attempts.set(attempts.get() + 1);
                Err(Error::CaptureFailed("test error".into()))
            },
            || clock.elapsed(),
            |sleep| sleeps.push(sleep),
        );

        assert!(
            matches!(result, Err(Error::CaptureFailed(_))),
            "Should propagate CaptureFailed error"
        );
        assert_eq!(attempts.get(), 1);
        assert!(sleeps.is_empty());
    }
}