cvkg-core 0.2.15

use crate::agents;
use crate::*;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;

/// Thread-safe reactive state with multiple storage mechanism.
///
/// State<T> supports both lightweight single-reader access (via arc_swap)
/// and compound atomic transactions (via stm::TVar for non-WASM targets).
///
/// # Storage Trade-offs
/// - `swap` (ArcSwap): Lock-free reads, atomic pointer swap. Best for read-heavy workloads.
/// - `tvar` (TVar): Blocking compound transactions. Best for coordinated multi-state operations.
/// - `subscribers`: Async notification list for reactive updates.
///
/// # Example
/// ```no_run
/// use cvkg_core::State;
/// let state = State::new(42u32);
/// state.set(100);
/// assert_eq!(state.get(), 100);
/// ```
#[derive(Clone)]
pub struct State<T: Clone + Send + Sync + 'static> {
    swap: Arc<arc_swap::ArcSwap<T>>,
    metadata_swap: Arc<arc_swap::ArcSwap<Option<agents::MutationMetadata>>>,
    #[cfg(not(target_arch = "wasm32"))]
    tvar: Arc<stm::TVar<T>>,
    #[cfg(not(target_arch = "wasm32"))]
    metadata_tvar: Arc<stm::TVar<Option<agents::MutationMetadata>>>,
    subscribers: SubscriberList<T>,
    version: Arc<AtomicU64>,
    resolution: agents::ConflictResolution,
}

impl<T: Clone + Send + Sync + 'static> State<T> {
    /// Create a new reactive state with the given initial value.
    pub fn new(value: T) -> Self {
        // Initialize metadata (None for fresh state, no mutation history)
        let metadata: Option<agents::MutationMetadata> = None;
        Self {
            #[cfg(not(target_arch = "wasm32"))]
            tvar: Arc::new(stm::TVar::new(value.clone())),
            #[cfg(not(target_arch = "wasm32"))]
            metadata_tvar: Arc::new(stm::TVar::new(metadata)),
            swap: Arc::new(arc_swap::ArcSwap::new(Arc::new(value))),
            metadata_swap: Arc::new(arc_swap::ArcSwap::new(Arc::new(metadata))),
            subscribers: Arc::new(std::sync::Mutex::new(Vec::new())),
            version: Arc::new(AtomicU64::new(0)),
            resolution: agents::ConflictResolution::LastWriterWins,
        }
    }

    /// Get the current value.
    pub fn get(&self) -> T {
        (**self.swap.load()).clone()
    }

    /// Set the conflict resolution strategy.
    pub fn with_resolution(mut self, resolution: agents::ConflictResolution) -> Self {
        self.resolution = resolution;
        self
    }

    /// Set the value via arc_swap (fast path). Notifies subscribers.
    /// Respects conflict resolution strategy (PriorityWins skips lower-priority writes).
    pub fn set(&self, value: T) {
        #[cfg(not(target_arch = "wasm32"))]
        {
            let result = stm::atomically(|tx| {
                let new_meta = agents::get_current_mutation_metadata();
                let existing_meta = self.metadata_tvar.read(tx)?;
                let mut skip = false;
                if self.resolution == agents::ConflictResolution::PriorityWins
                    && let (Some(new_m), Some(old_m)) = (new_meta, existing_meta)
                    && new_m.priority < old_m.priority
                {
                    skip = true;
                }
                if !skip {
                    self.tvar.write(tx, value.clone())?;
                    self.metadata_tvar.write(tx, new_meta)?;
                    Ok::<_, stm::StmError>(false)
                } else {
                    Ok::<_, stm::StmError>(true)
                }
            });
            if result {
                return;
            }
        }
        self.swap.store(Arc::new(value.clone()));
        let _version = self.version.fetch_add(1, Ordering::Release) + 1;
        self.notify(&value);
    }

    /// Subscribe to state changes.
    pub fn subscribe<F: Fn(&T) + Send + Sync + 'static>(&self, callback: F) {
        self.subscribers
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .push(Box::new(callback));
    }

    /// Notify all subscribers of a new value.
    fn notify(&self, value: &T) {
        invoke_subscribers_safely(&self.subscribers, value);
    }

    /// Get the current version counter.
    pub fn version(&self) -> u64 {
        self.version.load(Ordering::Acquire)
    }

    /// Set the value directly via arc_swap without TVar transaction (fastest path).
    /// Suitable for high-frequency updates where transactional consistency is not required.
    /// Notifies subscribers just like `set()`.
    pub fn set_direct(&self, value: T) {
        self.swap.store(Arc::new(value.clone()));
        let _version = self.version.fetch_add(1, Ordering::Release) + 1;
        // Bypass TVar: direct mode does not update the transactional storage.
        // This means compound transactions won't see this value.
        self.notify(&value);
    }

    /// Get metadata for the last mutation.
    pub fn get_metadata(&self) -> Option<agents::MutationMetadata> {
        **self.metadata_swap.load()
    }

    /// Set value with mutation metadata (actor + reason).
    pub fn set_with_metadata(&self, value: T, meta: agents::MutationMetadata) {
        self.swap.store(Arc::new(value.clone()));
        self.metadata_swap.store(Arc::new(Some(meta)));
        let _version = self.version.fetch_add(1, Ordering::Release) + 1;
        #[cfg(not(target_arch = "wasm32"))]
        {
            let tvar = &*self.tvar;
            stm::atomically(|tx| {
                tvar.write(tx, value.clone())?;
                Ok::<_, stm::StmError>(())
            });
        }
        self.notify(&value);
    }
}

/// A read/write projection into a `State<T>` owned elsewhere.
#[derive(Clone)]
pub struct Binding<T: Clone + Send + Sync + 'static> {
    swap: Arc<arc_swap::ArcSwap<T>>,
    version: Arc<AtomicU64>,
}

impl<T: Clone + Send + Sync + 'static> Binding<T> {
    #[allow(dead_code)]
    pub(crate) fn new(swap: Arc<arc_swap::ArcSwap<T>>, version: Arc<AtomicU64>) -> Self {
        Self { swap, version }
    }

    /// Read the current value.
    pub fn get(&self) -> T {
        (**self.swap.load()).clone()
    }

    /// Set the value (updates the underlying State).
    pub fn set(&self, value: T) {
        self.swap.store(Arc::new(value));
        self.version.fetch_add(1, Ordering::Release);
    }
}