cyclonedds 3.0.0-alpha.3

use crate::entity::{OwnedEntity, OwnedHandle};
use crate::error::check_entity;
use crate::{DdsEntity, DdsResult, DdsType};
use cyclonedds_rust_sys::*;
use std::collections::HashMap;
use std::ffi::c_void;
use std::sync::{Arc, Mutex, OnceLock};

// ---------------------------------------------------------------------------
// Global registry for QueryCondition closures
// ---------------------------------------------------------------------------
// The C API `dds_create_querycondition` accepts a filter function with
// signature `fn(*const c_void) -> bool` – no context/arg parameter.  To
// bridge Rust closures we store them in a global map keyed by the condition
// entity handle.  The trampoline dispatches through this map.

/// `Arc`, not `Box`: the trampoline must be able to keep the closure alive for
/// the duration of the call *after* releasing the registry lock. Holding a raw
/// pointer past the `MutexGuard` (as this used to) is a use-after-free — a
/// `QueryCondition` is `Send` and its `Drop` removes the entry, so another
/// thread dropping the condition between the lookup and the call would free the
/// closure while it is about to be invoked.
type QcClosure = std::sync::Arc<dyn Fn(*const c_void) -> bool + Send + Sync>;

fn qc_registry() -> &'static Mutex<HashMap<dds_entity_t, QcClosure>> {
    static REGISTRY: OnceLock<Mutex<HashMap<dds_entity_t, QcClosure>>> = OnceLock::new();
    REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Trampoline used by all closure-backed QueryConditions.
///
/// The C callback receives no context argument, so the active condition is
/// published through a thread-local set by [`QueryCondition::activate`]
/// (RAII) around the DDS operation that evaluates the filter (read/take/
/// waitset_wait run the filter synchronously on the calling thread).
///
/// FFI-QC-010 hardening:
/// - poisoned registry mutex is recovered instead of `unwrap`-panicking;
/// - a panicking closure is contained by `catch_unwind` (a panic must never
///   cross the `extern "C"` boundary) — the sample is EXCLUDED and the
///   panic is logged;
/// - with no active condition the filter FAILS CLOSED (excludes the sample)
///   and logs: silently passing everything through would make the
///   QueryCondition return unfiltered data (the original bug class).
unsafe extern "C" fn trampoline_qc_filter(sample: *const c_void) -> bool {
    let handle = QC_ACTIVE.with(|active| active.get());
    if handle <= 0 {
        eprintln!("QueryCondition filter invoked with no active condition; sample excluded");
        return false;
    }
    // Clone the Arc *inside* the lock so the closure cannot be dropped out from
    // under us once the guard is released.
    let closure = {
        let registry = qc_registry().lock().unwrap_or_else(|e| e.into_inner());
        registry.get(&handle).cloned()
    };
    match closure {
        Some(c) => match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| c(sample))) {
            Ok(keep) => keep,
            Err(_) => {
                eprintln!("QueryCondition filter closure panicked; sample excluded");
                false
            }
        },
        None => {
            eprintln!("QueryCondition {handle} not found in registry; sample excluded");
            false
        }
    }
}

thread_local! {
    static QC_ACTIVE: std::cell::Cell<dds_entity_t> = const { std::cell::Cell::new(0) };
}

/// Set the active QueryCondition entity for the current thread.
///
/// Low-level escape hatch — prefer [`QueryCondition::activate`], which
/// restores the previous value on drop. If you call this directly, reset to
/// 0 after the DDS operation.
pub(crate) fn set_active_qc(entity: dds_entity_t) {
    QC_ACTIVE.with(|active| active.set(entity));
}

fn active_qc() -> dds_entity_t {
    QC_ACTIVE.with(|active| active.get())
}

/// RAII guard que ativa uma [`QueryCondition`] na thread atual e restaura o
/// valor anterior no drop. Filtros de QueryCondition são avaliados
/// sincronamente na thread que chama a operação DDS (read/take/wait), então
/// basta envolver a operação com o guard:
///
/// ```ignore
/// let guard = qc.activate();
/// reader.take()?;           // filtro da qc aplicado nesta chamada
/// drop(guard);
/// ```
pub struct QcGuard(dds_entity_t);

impl Drop for QcGuard {
    fn drop(&mut self) {
        set_active_qc(self.0);
    }
}

// ---------------------------------------------------------------------------
// WaitSet
// ---------------------------------------------------------------------------

pub struct WaitSet {
    inner: Arc<OwnedEntity>,
}

impl WaitSet {
    /// Create a waitset on `participant`, holding it alive.
    pub fn new(participant: &crate::DomainParticipant) -> DdsResult<Self> {
        let entity = unsafe { dds_create_waitset(participant.entity()) };
        check_entity(entity)?;
        Ok(WaitSet {
            inner: OwnedEntity::new(entity, "WaitSet", None, vec![participant.owned().clone()]),
        })
    }

    /// Create a waitset on `participant`, keeping `parents` alive with it.
    ///
    /// Used by the async streams, which look the participant handle up from a
    /// reader and want the *reader* held up for the duration of the wait.
    pub(crate) fn for_parents(
        participant: dds_entity_t,
        parents: Vec<Arc<OwnedEntity>>,
    ) -> DdsResult<Self> {
        let entity = unsafe { dds_create_waitset(participant) };
        check_entity(entity)?;
        Ok(WaitSet {
            inner: OwnedEntity::new(entity, "WaitSet", None, parents),
        })
    }

    /// Create a waitset from a raw participant handle.
    ///
    /// # Safety
    ///
    /// Does not hold the participant alive. Prefer [`WaitSet::new`].
    pub unsafe fn from_entity(participant: dds_entity_t) -> DdsResult<Self> {
        let entity = unsafe { dds_create_waitset(participant) };
        check_entity(entity)?;
        Ok(WaitSet {
            inner: OwnedEntity::unowned(entity, "WaitSet"),
        })
    }

    pub fn attach(&self, entity: dds_entity_t, cookie: i64) -> DdsResult<()> {
        let ret = unsafe { dds_waitset_attach(self.entity(), entity, cookie as dds_attach_t) };
        crate::error::check(ret)
    }

    pub fn detach(&self, entity: dds_entity_t) -> DdsResult<()> {
        let ret = unsafe { dds_waitset_detach(self.entity(), entity) };
        crate::error::check(ret)
    }

    pub fn set_trigger(&self, trigger: bool) -> DdsResult<()> {
        let ret = unsafe { dds_waitset_set_trigger(self.entity(), trigger) };
        crate::error::check(ret)
    }

    pub fn wait(&self, timeout_ns: i64) -> DdsResult<Vec<i64>> {
        let max_results: usize = 64;
        let mut xs: Vec<dds_attach_t> = vec![0; max_results];
        let n =
            unsafe { dds_waitset_wait(self.entity(), xs.as_mut_ptr(), max_results, timeout_ns) };
        if n < 0 {
            return Err(crate::DdsError::from(n));
        }
        let n = n as usize;
        xs.truncate(n);
        Ok(xs.into_iter().map(|x| x as i64).collect())
    }

    pub fn get_entities(&self) -> DdsResult<Vec<dds_entity_t>> {
        unsafe {
            let count = dds_waitset_get_entities(self.entity(), std::ptr::null_mut(), 0);
            if count < 0 {
                return Err(crate::DdsError::from(count));
            }
            let count = count as usize;
            if count == 0 {
                return Ok(Vec::new());
            }

            let mut entities = vec![0; count];
            let actual =
                dds_waitset_get_entities(self.entity(), entities.as_mut_ptr(), entities.len());
            if actual < 0 {
                return Err(crate::DdsError::from(actual));
            }
            entities.truncate(actual as usize);
            Ok(entities)
        }
    }
}

impl DdsEntity for WaitSet {
    fn entity(&self) -> dds_entity_t {
        self.inner.handle()
    }
}

impl OwnedHandle for WaitSet {
    fn owned(&self) -> &Arc<OwnedEntity> {
        &self.inner
    }
}

// ---------------------------------------------------------------------------
// ReadCondition
// ---------------------------------------------------------------------------

pub struct ReadCondition {
    inner: Arc<OwnedEntity>,
}

const ANY_STATE: u32 = cyclonedds_rust_sys::DDS_ANY_SAMPLE_STATE
    | cyclonedds_rust_sys::DDS_ANY_INSTANCE_STATE
    | cyclonedds_rust_sys::DDS_ANY_VIEW_STATE;

const NOT_READ_STATE: u32 = cyclonedds_rust_sys::DDS_NOT_READ_SAMPLE_STATE
    | cyclonedds_rust_sys::DDS_ANY_INSTANCE_STATE
    | cyclonedds_rust_sys::DDS_ANY_VIEW_STATE;

impl ReadCondition {
    /// Create a read condition on `reader`, holding it alive.
    pub fn new<T: DdsType>(reader: &crate::DataReader<T>, mask: u32) -> DdsResult<Self> {
        let entity = unsafe { dds_create_readcondition(reader.entity(), mask) };
        check_entity(entity)?;
        Ok(ReadCondition {
            inner: OwnedEntity::new(entity, "ReadCondition", None, vec![reader.owned().clone()]),
        })
    }

    pub fn any<T: DdsType>(reader: &crate::DataReader<T>) -> DdsResult<Self> {
        Self::new(reader, ANY_STATE)
    }

    pub fn not_read<T: DdsType>(reader: &crate::DataReader<T>) -> DdsResult<Self> {
        Self::new(reader, NOT_READ_STATE)
    }

    /// Create a read condition from a raw reader handle.
    ///
    /// # Safety
    ///
    /// Does not hold the reader alive. Prefer [`ReadCondition::new`].
    pub unsafe fn from_entity(reader: dds_entity_t, mask: u32) -> DdsResult<Self> {
        let entity = unsafe { dds_create_readcondition(reader, mask) };
        check_entity(entity)?;
        Ok(ReadCondition {
            inner: OwnedEntity::unowned(entity, "ReadCondition"),
        })
    }
}

impl DdsEntity for ReadCondition {
    fn entity(&self) -> dds_entity_t {
        self.inner.handle()
    }
}

impl OwnedHandle for ReadCondition {
    fn owned(&self) -> &Arc<OwnedEntity> {
        &self.inner
    }
}

// ---------------------------------------------------------------------------
// QueryCondition
// ---------------------------------------------------------------------------

pub struct QueryCondition {
    entity: dds_entity_t,
    /// Whether this QueryCondition owns a closure in the global registry.
    owns_closure: bool,
    /// The reader this condition sits on. Declared last so it is released only
    /// after `Drop` has deregistered the closure and deleted the entity.
    ///
    /// `QueryCondition` keeps its own `Drop` rather than moving into
    /// `OwnedEntity`, because deleting the entity is not all it has to do: the
    /// closure has to leave the global registry at the same moment, and that
    /// ordering is easier to see here than threaded through a generic owner.
    _parents: Vec<Arc<OwnedEntity>>,
}

impl QueryCondition {
    /// Create a QueryCondition with a raw C function pointer filter.
    ///
    /// This is the low-level constructor matching the C API directly.
    /// For a Rust-closure-based API, see [`QueryCondition::with_filter`].
    pub fn new<T: DdsType>(
        reader: &crate::DataReader<T>,
        mask: u32,
        filter: unsafe extern "C" fn(*const std::ffi::c_void) -> bool,
    ) -> DdsResult<Self> {
        let entity = unsafe { dds_create_querycondition(reader.entity(), mask, Some(filter)) };
        check_entity(entity)?;
        Ok(QueryCondition {
            entity,
            owns_closure: false,
            _parents: vec![reader.owned().clone()],
        })
    }

    /// [`QueryCondition::new`] from a raw reader handle.
    ///
    /// # Safety
    ///
    /// Does not hold the reader alive.
    pub unsafe fn from_entity(
        reader: dds_entity_t,
        mask: u32,
        filter: unsafe extern "C" fn(*const std::ffi::c_void) -> bool,
    ) -> DdsResult<Self> {
        let entity = unsafe { dds_create_querycondition(reader, mask, Some(filter)) };
        check_entity(entity)?;
        Ok(QueryCondition {
            entity,
            owns_closure: false,
            _parents: Vec::new(),
        })
    }

    /// Create a QueryCondition with a Rust closure as the filter.
    ///
    /// The closure receives a raw `*const c_void` pointer to the sample data
    /// (matching the C API contract).  Return `true` to include the sample,
    /// `false` to exclude it.
    ///
    /// # Note on thread-local activation
    ///
    /// Because the CycloneDDS C API does not pass a user-data argument to
    /// the filter callback, the closure is dispatched through a thread-local
    /// "active handle" mechanism. Wrap any DDS operation that may trigger this
    /// filter (`dds_read`, `dds_take`, `dds_waitset_wait`) in the RAII guard
    /// returned by [`QueryCondition::activate`], which sets and restores the
    /// thread-local for you.
    pub fn with_filter<T: DdsType, F>(
        reader: &crate::DataReader<T>,
        mask: u32,
        filter: F,
    ) -> DdsResult<Self>
    where
        F: Fn(*const c_void) -> bool + Send + Sync + 'static,
    {
        let entity =
            unsafe { dds_create_querycondition(reader.entity(), mask, Some(trampoline_qc_filter)) };
        check_entity(entity)?;

        // Register the closure.
        {
            // Recover from a poisoned registry instead of panicking — the
            // trampoline already does, and a panic here would be raised from a
            // constructor on the user's thread for an unrelated failure.
            let mut registry = qc_registry().lock().unwrap_or_else(|e| e.into_inner());
            registry.insert(entity, std::sync::Arc::new(filter));
        }

        Ok(QueryCondition {
            entity,
            owns_closure: true,
            _parents: vec![reader.owned().clone()],
        })
    }

    /// Ativa esta condition na thread atual (RAII — restaura o valor anterior
    /// no drop). FFI-QC-010: substitui a ativação TLS manual; envolva a
    /// operação DDS que deve avaliar o filtro (read/take/wait) com o guard.
    pub fn activate(&self) -> QcGuard {
        let previous = active_qc();
        set_active_qc(self.entity);
        QcGuard(previous)
    }
}

impl DdsEntity for QueryCondition {
    fn entity(&self) -> dds_entity_t {
        self.entity
    }
}

impl Drop for QueryCondition {
    fn drop(&mut self) {
        if self.owns_closure {
            // Never panic in a Drop: if this runs during an unwind, a second
            // panic aborts the process. A poisoned registry is recoverable.
            let mut registry = qc_registry().lock().unwrap_or_else(|e| e.into_inner());
            registry.remove(&self.entity);
        }
        crate::entity::delete_entity(self.entity, "QueryCondition");
    }
}

// ---------------------------------------------------------------------------
// GuardCondition
// ---------------------------------------------------------------------------

pub struct GuardCondition {
    inner: Arc<OwnedEntity>,
}

impl GuardCondition {
    /// Create a guard condition on `participant`, holding it alive.
    pub fn new(participant: &crate::DomainParticipant) -> DdsResult<Self> {
        let entity = unsafe { dds_create_guardcondition(participant.entity()) };
        check_entity(entity)?;
        Ok(GuardCondition {
            inner: OwnedEntity::new(
                entity,
                "GuardCondition",
                None,
                vec![participant.owned().clone()],
            ),
        })
    }

    /// Create a guard condition from a raw participant handle.
    ///
    /// # Safety
    ///
    /// Does not hold the participant alive. Prefer [`GuardCondition::new`].
    pub unsafe fn from_entity(participant: dds_entity_t) -> DdsResult<Self> {
        let entity = unsafe { dds_create_guardcondition(participant) };
        check_entity(entity)?;
        Ok(GuardCondition {
            inner: OwnedEntity::unowned(entity, "GuardCondition"),
        })
    }

    pub fn set_triggered(&self, triggered: bool) -> DdsResult<()> {
        let ret = unsafe { dds_set_guardcondition(self.entity(), triggered) };
        crate::error::check(ret)
    }

    /// Read the current trigger state without consuming it.
    pub fn read(&self) -> DdsResult<bool> {
        let mut triggered = false;
        let ret = unsafe { dds_read_guardcondition(self.entity(), &mut triggered) };
        crate::error::check(ret)?;
        Ok(triggered)
    }

    /// Take (read and reset) the current trigger state.
    pub fn take(&self) -> DdsResult<bool> {
        let mut triggered = false;
        let ret = unsafe { dds_take_guardcondition(self.entity(), &mut triggered) };
        crate::error::check(ret)?;
        Ok(triggered)
    }
}

impl DdsEntity for GuardCondition {
    fn entity(&self) -> dds_entity_t {
        self.inner.handle()
    }
}

impl OwnedHandle for GuardCondition {
    fn owned(&self) -> &Arc<OwnedEntity> {
        &self.inner
    }
}

// ---------------------------------------------------------------------------
// FFI-QC-010: testes do dispatch de filtros (sem entidades DDS reais — o
// trampoline consulta apenas o registry + TLS, então handles sintéticos
// bastam para exercitar a lógica endurecida).
// ---------------------------------------------------------------------------
#[cfg(all(test, feature = "std"))]
mod qc_tests {
    use super::*;

    fn register_fake(
        handle: dds_entity_t,
        f: impl Fn(*const c_void) -> bool + Send + Sync + 'static,
    ) {
        qc_registry()
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert(handle, std::sync::Arc::new(f));
    }

    fn unregister_fake(handle: dds_entity_t) {
        qc_registry()
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(&handle);
    }

    /// O handle não é passado adiante: `trampoline_qc_filter` resolve a condition
    /// pelo TLS `QC_ACTIVE`. O parâmetro fica só para deixar as chamadas legíveis.
    fn call_trampoline(_handle: dds_entity_t) -> bool {
        let sample = 42i32;
        unsafe { trampoline_qc_filter(&sample as *const i32 as *const c_void) }
    }

    #[test]
    fn filtro_aplicado_com_guard_sem_tls_manual() {
        let handle = 9_000_001;
        register_fake(handle, |s| unsafe { *(s as *const i32) } == 42);
        let guard = QcGuard(0);
        set_active_qc(handle);
        std::mem::forget(guard); // simula activate(): ativo = handle
        assert!(call_trampoline(handle));
        set_active_qc(0);
        unregister_fake(handle);
    }

    #[test]
    fn fail_closed_sem_condition_ativa() {
        assert!(!call_trampoline(9_000_002)); // QC_ACTIVE = 0 → exclui
    }

    #[test]
    fn fail_closed_condition_desconhecida() {
        set_active_qc(9_000_003);
        assert!(!call_trampoline(9_000_003)); // não está no registry → exclui
        set_active_qc(0);
    }

    #[test]
    fn panic_na_closure_nao_atravessa_ffi() {
        let handle = 9_000_004;
        register_fake(handle, |_| panic!("panic injetado"));
        set_active_qc(handle);
        assert!(!call_trampoline(handle)); // panic contido → amostra excluída
        set_active_qc(0);
        unregister_fake(handle);
    }

    #[test]
    fn registry_envenenado_recupera_sem_panic() {
        let handle = 9_000_005;
        register_fake(handle, |_| true);
        // Envenena o mutex do registry segurando-o durante um panic.
        let reg = qc_registry();
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _guard = reg.lock().unwrap();
            panic!("envenenando o registry");
        }));
        // O panic ocorreu com o guard vivo, então o envenenamento é determinístico.
        // (Antes: `reg.is_poisoned() || true`, uma tautologia que não asseverava nada
        // e ainda reprovava no gate `clippy -D warnings` do CI.)
        assert!(reg.is_poisoned(), "mutex deveria estar envenenado");
        set_active_qc(handle);
        assert!(call_trampoline(handle)); // into_inner → closure ainda roda
        set_active_qc(0);
        unregister_fake(handle);
    }

    #[test]
    fn guard_restaura_valor_anterior() {
        let (h1, h2) = (9_000_006, 9_000_007);
        register_fake(h1, |_| true);
        register_fake(h2, |_| false);
        set_active_qc(h1);
        {
            let _g = QcGuard(h1); // simula activate(h2): guarda o anterior
            set_active_qc(h2);
            assert!(!call_trampoline(h2)); // h2 ativo (filtro false)
        } // _g dropado aqui: restaura h1
        assert!(call_trampoline(h1)); // h1 de volta (filtro true)
        set_active_qc(0);
        unregister_fake(h1);
        unregister_fake(h2);
    }
}