rust-elm 0.6.0

use std::collections::VecDeque;
use std::hash::Hash;
use std::marker::PhantomData;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use crossbeam_channel::{Receiver, Sender};
use key_paths_core::{FieldDiff, hash_value};
use parking_lot::Mutex;

use crate::bus::BusSender;
use crate::effect::EffectId;
use crate::optics::Casepath;
use crate::runtime::binding::StateBinding;
use key_paths_core::RefKpTrait;
use super::interpreter::InterpreterState;

/// Calls [`StoreWork::end_store_work`] on drop unless [`Self::disarm`]d.
///
/// Mirrors stack-unwind / `finally` semantics so [`StoreTask`] waiters are not left
/// hanging when a dispatch aborts (e.g. reducer panic).
pub(crate) struct StoreWorkUnwindGuard<'a, S, M, B: StoreWork<S, M> + ?Sized>
where
    S: Send + 'static,
    M: Send + 'static,
{
    backend: &'a B,
    active: bool,
    _marker: PhantomData<(S, M)>,
}

impl<'a, S, M, B> StoreWorkUnwindGuard<'a, S, M, B>
where
    S: Send + 'static,
    M: Send + 'static,
    B: StoreWork<S, M> + ?Sized,
{
    pub(crate) fn new(backend: &'a B) -> Self {
        Self {
            backend,
            active: true,
            _marker: PhantomData,
        }
    }

    pub(crate) fn disarm(&mut self) {
        self.active = false;
    }
}

impl<S, M, B> Drop for StoreWorkUnwindGuard<'_, S, M, B>
where
    S: Send + 'static,
    M: Send + 'static,
    B: StoreWork<S, M> + ?Sized,
{
    fn drop(&mut self) {
        if self.active {
            self.backend.end_store_work();
        }
    }
}

/// Cloneable dispatch handle for a running [`Runtime`](crate::Runtime).
///
/// `update` runs on the runtime thread — actions are never applied synchronously inside
/// `send`, so reducers cannot re-enter themselves from the caller's stack (UDF parity).
pub struct Store<S, M> {
    pub(crate) backend: StoreBackend<S, M>,
}

impl<S, M> Clone for Store<S, M>
where
    S: Send + 'static,
    M: Send + 'static,
{
    fn clone(&self) -> Self {
        Self {
            backend: self.backend.clone(),
        }
    }
}

pub(crate) struct StoreHub<S, M> {
    pub sender: BusSender<M>,
    pub state_listeners: Arc<Mutex<Vec<Sender<()>>>>,
    pub effect_done: Arc<Mutex<VecDeque<Sender<()>>>>,
    pub in_flight: Arc<AtomicUsize>,
    pub interpreter: Arc<InterpreterState<M>>,
    _marker: PhantomData<S>,
}

/// Shared store-work accounting (in-flight dispatches, listener notify, effect-done signals).
pub(crate) trait StoreWork<S, M>
where
    S: Send + 'static,
    M: Send + 'static,
{
    fn hub(&self) -> &Arc<StoreHub<S, M>>;

    fn begin_store_work(&self) {
        self.hub().in_flight.fetch_add(1, Ordering::SeqCst);
    }

    fn end_store_work(&self) {
        let prev = self.hub().in_flight.fetch_sub(1, Ordering::SeqCst);
        if prev == 1 {
            self.hub().notify_state();
            self.hub().signal_effect_done();
        }
    }

    fn notify_state(&self) {
        self.hub().notify_state();
    }
}

impl<S, M> StoreHub<S, M>
where
    S: Send + 'static,
    M: Send + 'static,
{
    pub fn new(sender: BusSender<M>, interpreter: Arc<InterpreterState<M>>) -> Arc<Self> {
        Arc::new(Self {
            sender,
            state_listeners: Arc::new(Mutex::new(Vec::new())),
            effect_done: Arc::new(Mutex::new(VecDeque::new())),
            in_flight: Arc::new(AtomicUsize::new(0)),
            interpreter,
            _marker: PhantomData,
        })
    }

    pub fn notify_state(&self) {
        self.state_listeners
            .lock()
            .retain(|tx| tx.send(()).is_ok());
    }

    pub fn pop_effect_done(&self) -> Option<Sender<()>> {
        self.effect_done.lock().pop_front()
    }

    pub fn signal_effect_done(&self) {
        if let Some(tx) = self.pop_effect_done() {
            let _ = tx.send(());
        }
    }
}

pub(crate) struct StoreBackend<S, M> {
    pub state: Arc<Mutex<S>>,
    pub hub: Arc<StoreHub<S, M>>,
}

impl<S, M> Clone for StoreBackend<S, M>
where
    S: Send + 'static,
    M: Send + 'static,
{
    fn clone(&self) -> Self {
        Self {
            state: self.state.clone(),
            hub: self.hub.clone(),
        }
    }
}

impl<S, M> StoreWork<S, M> for StoreBackend<S, M>
where
    S: Send + 'static,
    M: Send + 'static,
{
    fn hub(&self) -> &Arc<StoreHub<S, M>> {
        &self.hub
    }
}

impl<S, M> StoreBackend<S, M>
where
    S: Send + 'static,
    M: Send + 'static,
{
    pub fn new(
        state: Arc<Mutex<S>>,
        sender: BusSender<M>,
        interpreter: Arc<InterpreterState<M>>,
    ) -> Self {
        Self {
            state,
            hub: StoreHub::new(sender, interpreter),
        }
    }

    pub fn store(&self) -> Store<S, M> {
        Store {
            backend: self.clone(),
        }
    }
}

impl<S, M> Store<S, M>
where
    S: Send + Sync + Clone + 'static,
    M: Send + 'static,
{
    /// Dispatch an action and return a handle that completes when its effects finish.
    pub fn send(&self, action: M) -> StoreTask {
        let (tx, rx) = crossbeam_channel::bounded(1);
        self.backend.hub.effect_done.lock().push_back(tx);
        self.backend.begin_store_work();
        let _ = self.backend.hub.sender.send_blocking(action);
        StoreTask::from_rx(rx)
    }

    /// Fire-and-forget dispatch.
    pub fn dispatch(&self, action: M) {
        let _ = self.send(action);
    }

    pub fn state(&self) -> S {
        self.backend.state.lock().clone()
    }

    /// Zero-copy read/write handle to store state (no clone of `S`).
    pub fn binding(&self) -> StateBinding<S> {
        StateBinding::new(Arc::clone(&self.backend.state))
    }

    pub fn cancel(&self, id: EffectId) {
        if let Some(handle) = self.backend.hub.interpreter.cancel_tokens.lock().remove(&id) {
            handle.abort();
        }
    }

    /// Subscribe to deduplicated `Arc` state snapshots (skips consecutive equal states).
    pub fn subscribe_state(&self) -> StateSubscriber<S>
    where
        S: Clone + PartialEq,
    {
        let (tx, rx) = crossbeam_channel::unbounded();
        self.backend.hub.state_listeners.lock().push(tx);
        StateSubscriber {
            rx,
            state: self.backend.state.clone(),
            last: Some(Arc::new(self.backend.state.lock().clone())),
        }
    }

    /// Subscribe to field-level change signals without cloning whole state.
    ///
    /// Dedup uses root + per-field hashes ([`FieldDiff`]); subscribers read changed
    /// fields on demand via keypaths / [`StateBinding`].
    pub fn subscribe_changes(&self) -> ChangeSubscriber<S>
    where
        S: FieldDiff + Hash,
    {
        let (tx, rx) = crossbeam_channel::unbounded();
        self.backend.hub.state_listeners.lock().push(tx);
        let (last_hash, last_fields) = {
            let guard = self.backend.state.lock();
            let root = hash_value(&*guard);
            let mut fields = Vec::new();
            guard.field_hashes(&mut fields);
            (Some(root), fields)
        };
        ChangeSubscriber {
            rx,
            state: self.backend.state.clone(),
            last_hash,
            last_fields,
            _marker: PhantomData,
        }
    }

    /// Focus a child store via state and action casepaths (see [`crate::optics`]).
    pub fn scope<CS, CM, AK, SK>(
        &self,
        state_kp: SK,
        action_kp: AK,
    ) -> ScopedStore<S, M, CS, CM, AK, SK>
    where
        CS: Clone + PartialEq + Send + Sync + 'static,
        CM: Clone + Send + 'static,
        S: 'static,
        M: 'static,
        AK: Casepath<M, CM> + Clone + Send + Sync + 'static,
        SK: RefKpTrait<S, CS> + Clone,
    {
        ScopedStore {
            store: self.clone(),
            state_kp,
            action_kp,
            _marker: PhantomData,
        }
    }
}

/// Awaitable handle for in-flight effects from a single [`Store::send`].
pub struct StoreTask {
    rx: Receiver<()>,
}

impl StoreTask {
    pub(crate) fn from_rx(rx: Receiver<()>) -> Self {
        Self { rx }
    }

    pub fn finish(self) -> Result<(), StoreTaskError> {
        self.finish_with_timeout(Duration::from_secs(5))
    }

    pub fn finish_with_timeout(self, timeout: Duration) -> Result<(), StoreTaskError> {
        self.rx
            .recv_timeout(timeout)
            .map_err(|_| StoreTaskError::Timeout)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StoreTaskError {
    Timeout,
}

impl std::fmt::Display for StoreTaskError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Timeout => write!(f, "timed out waiting for store effects to finish"),
        }
    }
}

impl std::error::Error for StoreTaskError {}

/// Which top-level (or derived) fields changed since the last signal.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChangeSet<P> {
    pub paths: Vec<P>,
}

/// Iterator-like subscription to field change signals (no full-state clone).
pub struct ChangeSubscriber<S>
where
    S: FieldDiff + Hash,
{
    rx: Receiver<()>,
    state: Arc<Mutex<S>>,
    last_hash: Option<u64>,
    last_fields: Vec<(S::Path, u64)>,
    _marker: PhantomData<S>,
}

impl<S> ChangeSubscriber<S>
where
    S: FieldDiff + Hash,
{
    fn snapshot_fields(&self) -> (u64, Vec<(S::Path, u64)>) {
        let guard = self.state.lock();
        let root = hash_value(&*guard);
        let mut fields = Vec::new();
        guard.field_hashes(&mut fields);
        (root, fields)
    }

    fn diff_fields(&self, fields: &[(S::Path, u64)]) -> Vec<S::Path> {
        fields
            .iter()
            .filter(|(path, hash)| {
                self.last_fields
                    .iter()
                    .find(|(prev_path, _)| prev_path == path)
                    .is_none_or(|(_, prev_hash)| prev_hash != hash)
            })
            .map(|(path, _)| *path)
            .collect()
    }

    /// Zero-copy binding to read changed fields after a signal.
    pub fn binding(&self) -> StateBinding<S> {
        StateBinding::new(Arc::clone(&self.state))
    }

    #[allow(clippy::should_implement_trait)]
    pub fn next(&mut self) -> Option<ChangeSet<S::Path>> {
        loop {
            match self.rx.try_recv() {
                Ok(()) => {
                    while self.rx.try_recv().is_ok() {}
                    let (root_hash, fields) = self.snapshot_fields();
                    if Some(root_hash) == self.last_hash {
                        continue;
                    }
                    let paths = self.diff_fields(&fields);
                    self.last_hash = Some(root_hash);
                    self.last_fields = fields;
                    if paths.is_empty() {
                        continue;
                    }
                    return Some(ChangeSet { paths });
                }
                Err(crossbeam_channel::TryRecvError::Empty) => return None,
                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
            }
        }
    }

    pub fn wait_next(&mut self, timeout: Duration) -> Option<ChangeSet<S::Path>> {
        let deadline = std::time::Instant::now() + timeout;
        while std::time::Instant::now() < deadline {
            if let Some(change) = self.next() {
                return Some(change);
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        None
    }
}

/// Iterator-like subscription to state snapshots.
pub struct StateSubscriber<S> {
    rx: Receiver<()>,
    state: Arc<Mutex<S>>,
    last: Option<Arc<S>>,
}

impl<S: PartialEq + Clone> StateSubscriber<S> {
    #[allow(clippy::should_implement_trait)]
    pub fn next(&mut self) -> Option<Arc<S>> {
        loop {
            match self.rx.try_recv() {
                Ok(()) => {
                    while self.rx.try_recv().is_ok() {}
                    let snapshot = Arc::new(self.state.lock().clone());
                    if self
                        .last
                        .as_ref()
                        .is_some_and(|prev| prev.as_ref() == snapshot.as_ref())
                    {
                        continue;
                    }
                    self.last = Some(snapshot.clone());
                    return Some(snapshot);
                }
                Err(crossbeam_channel::TryRecvError::Empty) => return None,
                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
            }
        }
    }

    pub fn wait_next(&mut self, timeout: Duration) -> Option<Arc<S>> {
        let deadline = std::time::Instant::now() + timeout;
        while std::time::Instant::now() < deadline {
            if let Some(s) = self.next() {
                return Some(s);
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        None
    }

    pub fn latest(&self) -> Option<Arc<S>> {
        self.last.clone()
    }
}

/// Child store routing actions through a parent action casepath.
pub struct ScopedStore<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK>
where
    SK: RefKpTrait<S, CS> + Clone,
{
    store: Store<S, M>,
    state_kp: SK,
    action_kp: AK,
    _marker: PhantomData<(M, CM, S, CS)>,
}

impl<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK> Clone
    for ScopedStore<S, M, CS, CM, AK, SK>
where
    S: Send,
    M: Send,
    AK: Clone,
    SK: RefKpTrait<S, CS> + Clone,
{
    fn clone(&self) -> Self {
        Self {
            store: self.store.clone(),
            state_kp: self.state_kp.clone(),
            action_kp: self.action_kp.clone(),
            _marker: PhantomData,
        }
    }
}

impl<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK>
    ScopedStore<S, M, CS, CM, AK, SK>
where
    S: Send + Sync + Clone,
    M: Send,
    CS: Clone + PartialEq + Send + Sync,
    CM: Clone + Send,
    AK: Casepath<M, CM> + Clone + Send + Sync + 'static,
    SK: RefKpTrait<S, CS> + Clone,
{
    pub fn send(&self, action: CM) -> StoreTask {
        self.store.send(self.action_kp.wrap(action))
    }

    pub fn dispatch(&self, action: CM) {
        self.store.dispatch(self.action_kp.wrap(action));
    }

    pub fn child_state(&self) -> Option<CS> {
        self.state_kp.focus(&self.store.state()).cloned()
    }

    /// Zero-copy binding projected to the scoped child via keypath.
    pub fn binding(&self) -> crate::runtime::binding::ProjectedBinding<S, CS, SK> {
        self.store.binding().project(self.state_kp.clone())
    }

    pub fn subscribe_state(&self) -> ScopedStateSubscriber<S, CS, SK>
    where
        S: Clone + PartialEq,
        SK: Clone,
    {
        ScopedStateSubscriber {
            inner: self.store.subscribe_state(),
            state_kp: self.state_kp.clone(),
            _marker: PhantomData,
        }
    }

    /// Field-level changes for the scoped child (hashes computed under lock, no clone of `S`).
    pub fn subscribe_changes(&self) -> ScopedChangeSubscriber<S, CS, SK>
    where
        CS: FieldDiff + Hash,
    {
        let (tx, rx) = crossbeam_channel::unbounded();
        self.store.backend.hub.state_listeners.lock().push(tx);
        let (last_hash, last_fields) = {
            let guard = self.store.backend.state.lock();
            if let Some(child) = self.state_kp.focus(&*guard) {
                let root = hash_value(child);
                let mut fields = Vec::new();
                child.field_hashes(&mut fields);
                (Some(root), fields)
            } else {
                (None, Vec::new())
            }
        };
        ScopedChangeSubscriber {
            rx,
            state: self.store.backend.state.clone(),
            state_kp: self.state_kp.clone(),
            last_hash,
            last_fields,
            _marker: PhantomData,
        }
    }
}

pub struct ScopedStateSubscriber<S: 'static, CS: 'static, SK>
where
    SK: RefKpTrait<S, CS> + Clone,
{
    inner: StateSubscriber<S>,
    state_kp: SK,
    _marker: PhantomData<(S, CS)>,
}

impl<S, CS, SK> ScopedStateSubscriber<S, CS, SK>
where
    S: PartialEq + Clone,
    CS: Clone + PartialEq,
    SK: RefKpTrait<S, CS> + Clone,
{
    #[allow(clippy::should_implement_trait)]
    pub fn next(&mut self) -> Option<CS> {
        loop {
            let parent = self.inner.next()?;
            let Some(child) = self.state_kp.focus(parent.as_ref()).cloned() else {
                continue;
            };
            return Some(child);
        }
    }
}

/// Field change subscription focused to a scoped child store.
pub struct ScopedChangeSubscriber<S: 'static, CS: 'static, SK>
where
    CS: FieldDiff,
    SK: RefKpTrait<S, CS> + Clone,
{
    rx: Receiver<()>,
    state: Arc<Mutex<S>>,
    state_kp: SK,
    last_hash: Option<u64>,
    last_fields: Vec<(CS::Path, u64)>,
    _marker: PhantomData<(S, CS)>,
}

impl<S, CS, SK> ScopedChangeSubscriber<S, CS, SK>
where
    S: Send + Sync,
    CS: FieldDiff + Hash + Send + Sync,
    SK: RefKpTrait<S, CS> + Clone,
{
    fn snapshot_fields(&self) -> Option<(u64, Vec<(CS::Path, u64)>)> {
        let guard = self.state.lock();
        let child = self.state_kp.focus(&*guard)?;
        let root = hash_value(child);
        let mut fields = Vec::new();
        child.field_hashes(&mut fields);
        Some((root, fields))
    }

    fn diff_fields(&self, fields: &[(CS::Path, u64)]) -> Vec<CS::Path> {
        fields
            .iter()
            .filter(|(path, hash)| {
                self.last_fields
                    .iter()
                    .find(|(prev_path, _)| prev_path == path)
                    .is_none_or(|(_, prev_hash)| prev_hash != hash)
            })
            .map(|(path, _)| *path)
            .collect()
    }

    pub fn binding(&self) -> crate::runtime::binding::ProjectedBinding<S, CS, SK> {
        crate::runtime::binding::StateBinding::new(Arc::clone(&self.state))
            .project(self.state_kp.clone())
    }

    #[allow(clippy::should_implement_trait)]
    pub fn next(&mut self) -> Option<ChangeSet<CS::Path>> {
        loop {
            match self.rx.try_recv() {
                Ok(()) => {
                    while self.rx.try_recv().is_ok() {}
                    let Some((root_hash, fields)) = self.snapshot_fields() else {
                        continue;
                    };
                    if Some(root_hash) == self.last_hash {
                        continue;
                    }
                    let paths = self.diff_fields(&fields);
                    self.last_hash = Some(root_hash);
                    self.last_fields = fields;
                    if paths.is_empty() {
                        continue;
                    }
                    return Some(ChangeSet { paths });
                }
                Err(crossbeam_channel::TryRecvError::Empty) => return None,
                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
            }
        }
    }

    pub fn wait_next(&mut self, timeout: Duration) -> Option<ChangeSet<CS::Path>> {
        let deadline = std::time::Instant::now() + timeout;
        while std::time::Instant::now() < deadline {
            if let Some(change) = self.next() {
                return Some(change);
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        None
    }
}