adaptite 0.3.0

use alloc::boxed::Box;
use alloc::rc::Rc;
use alloc::vec::Vec;
use core::cell::{Cell, RefCell};

// A `BTreeMap` rather than a hash map: subscriber ids come from a monotonic counter, so
// ordered iteration *is* registration order. `Event::on` feeds its queue from an ordinary
// immediate subscriber, so with hash iteration a re-entrant `emit` could enqueue values out of
// emission order — contradicting the documented guarantee — and the order varied per process.
use alloc::collections::BTreeMap;

use crate::scope::{OwnedDisposable, adopt_into_current};
use crate::{NodeId, NodeKind, Reactor, current, trace_targets};

type SubscriberFn<T> = dyn Fn(&T) + 'static;

/// Creates an event in the current thread's default reactor.
#[track_caller]
pub fn event<T: 'static>() -> Event<T> {
    current().event()
}

/// Creates an event associated with `reactor`.
#[track_caller]
pub fn event_in<T: 'static>(reactor: &Reactor) -> Event<T> {
    reactor.event()
}

/// Creates a reactive draining subscription for `event` in the current reactor.
///
/// Unlike [`Event::subscribe`], the handler does not run inline with `emit`: values queue up
/// and are drained by an effect on the microtask queue, in emission order.
///
/// # Examples
///
/// ```rust
/// use std::cell::RefCell;
/// use std::rc::Rc;
///
/// use adaptite::{event, on};
/// use runite::{queue_macrotask, run};
///
/// let seen = Rc::new(RefCell::new(Vec::new()));
///
/// queue_macrotask({
///     let seen = Rc::clone(&seen);
///     move || {
///         let messages = event::<u32>();
///         on(&messages, {
///             let seen = Rc::clone(&seen);
///             move |message| seen.borrow_mut().push(*message)
///         })
///         .leak();
///
///         messages.emit(1);
///         messages.emit(2);
///         // Delivery is deferred to the microtask flush.
///         assert!(seen.borrow().is_empty());
///     }
/// });
/// run();
///
/// assert_eq!(*seen.borrow(), [1, 2]);
/// ```
#[must_use = "subscriptions created with `on` must be used to stay active"]
#[track_caller]
pub fn on<T: Clone + 'static>(event: &Event<T>, handler: impl Fn(&T) + 'static) -> Subscription {
    current().on(event, handler)
}

/// Creates a reactive draining subscription for `event` associated with `reactor`.
#[must_use = "subscriptions created with `on_in` must be used to stay active"]
#[track_caller]
pub fn on_in<T: Clone + 'static>(
    reactor: &Reactor,
    event: &Event<T>,
    handler: impl Fn(&T) + 'static,
) -> Subscription {
    reactor.on(event, handler)
}

/// Push-style event source.
///
/// Events carry values that happen *at a moment* rather than state that *is*. Immediate
/// subscribers added with [`subscribe`](Event::subscribe) are called synchronously during
/// [`emit`](Event::emit); reactive draining subscriptions created with [`on`] queue values and
/// deliver them through an effect on the microtask queue.
///
/// # Examples
///
/// ```rust
/// use std::cell::RefCell;
/// use std::rc::Rc;
///
/// use adaptite::event;
///
/// let clicks = event::<u32>();
/// let seen = Rc::new(RefCell::new(Vec::new()));
///
/// let subscription = clicks.subscribe({
///     let seen = Rc::clone(&seen);
///     move |button| seen.borrow_mut().push(*button)
/// });
///
/// clicks.emit(1);
/// clicks.emit(2);
/// assert_eq!(*seen.borrow(), [1, 2]);
///
/// drop(subscription); // unsubscribes
/// clicks.emit(3);
/// assert_eq!(*seen.borrow(), [1, 2]);
/// ```
pub struct Event<T> {
    inner: Rc<EventInner<T>>,
}

// Manual impl: cloning the handle shares the node and must not require `T: Clone`.
impl<T> Clone for Event<T> {
    fn clone(&self) -> Self {
        Self {
            inner: Rc::clone(&self.inner),
        }
    }
}

/// Disposable subscription handle.
///
/// A subscription created outside any owner is cancelled when the last clone of its handle is
/// dropped; call [`leak`](Subscription::leak) to keep the subscriber active for the remainder
/// of the program without retaining the handle. A subscription created inside an owner (an
/// effect's run, or a [`crate::scope`]) is kept alive by that owner instead and is cancelled
/// with it — the same ownership rule as [`crate::EffectHandle`].
///
/// # Examples
///
/// ```rust
/// use std::cell::RefCell;
/// use std::rc::Rc;
///
/// use adaptite::{event, on, scope};
/// use runite::{queue_macrotask, run};
///
/// let seen = Rc::new(RefCell::new(Vec::new()));
///
/// queue_macrotask({
///     let seen = Rc::clone(&seen);
///     move || {
///         let messages = event::<u32>();
///         let (handle, subscription) = scope({
///             let messages = messages.clone();
///             let seen = Rc::clone(&seen);
///             move || on(&messages, move |message| seen.borrow_mut().push(*message))
///         });
///
///         messages.emit(1);
///         runite::queue_macrotask(move || {
///             handle.dispose();
///             assert!(!subscription.is_active(), "cancelled by its owner");
///             messages.emit(2); // no longer delivered
///         });
///     }
/// });
/// run();
///
/// assert_eq!(*seen.borrow(), [1]);
/// ```
#[derive(Clone)]
#[must_use = "an unowned subscription is cancelled when its handle is dropped; hold the handle, leak it, or subscribe inside a scope"]
pub struct Subscription {
    inner: Rc<SubscriptionInner>,
}

impl Reactor {
    /// Creates an event source associated with this reactor.
    #[track_caller]
    pub fn event<T: 'static>(&self) -> Event<T> {
        Event::new(self.clone())
    }

    /// Creates a reactive draining subscription for `event`.
    #[must_use = "subscriptions created with `on` must be used to stay active"]
    #[track_caller]
    pub fn on<T: Clone + 'static>(
        &self,
        event: &Event<T>,
        handler: impl Fn(&T) + 'static,
    ) -> Subscription {
        let queue = Rc::new(RefCell::new(Vec::new()));
        let direct = event.subscribe({
            let queue = Rc::clone(&queue);
            move |value| queue.borrow_mut().push(value.clone())
        });
        let effect = self.effect({
            let event = event.clone();
            let queue = Rc::clone(&queue);
            move || {
                event.observe();
                let drained = {
                    let mut queued = queue.borrow_mut();
                    queued.drain(..).collect::<Vec<_>>()
                };
                #[cfg(debug_assertions)]
                tracing::trace!(
                    target: trace_targets::EVENT,
                    event = "drain_event_queue",
                    event_id = event.inner.id.0,
                    drained = drained.len(),
                    "draining queued event values reactively"
                );
                // The handler is not a reactive observer: suspend tracking so its reads never
                // become dependencies of the draining effect (matching `watch` handlers and
                // immediate `emit` subscribers).
                crate::untrack(|| {
                    for value in &drained {
                        handler(value);
                    }
                });
            }
        });

        Subscription::new(move || {
            direct.unsubscribe();
            effect.dispose();
        })
    }
}

impl<T: 'static> Event<T> {
    /// Returns this event's node id, for use with the graph queries on
    /// [`Reactor`] — [`observer_count`](Reactor::observer_count),
    /// [`dependencies_of`](Reactor::dependencies_of), [`node_origin`](Reactor::node_origin) and
    /// friends. Ids are unique within one reactor, so aggregate on
    /// `(`[`Reactor::id`]`, id)`.
    pub fn id(&self) -> NodeId {
        self.inner.id
    }

    #[track_caller]
    fn new(reactor: Reactor) -> Self {
        let id = reactor.allocate_node(NodeKind::Event);
        tracing::debug!(
            target: trace_targets::EVENT,
            event = "create_event",
            node_id = id.0,
            "created reactive event"
        );
        Self {
            inner: Rc::new(EventInner {
                reactor,
                id,
                next_subscriber: Cell::new(1),
                subscribers: RefCell::new(Default::default()),
            }),
        }
    }

    fn observe(&self) {
        #[cfg(debug_assertions)]
        tracing::trace!(
            target: trace_targets::EVENT,
            event = "observe_event",
            event_id = self.inner.id.0,
            "observing event reactively"
        );
        self.inner.reactor.observe(self.inner.id);
    }

    /// Emits a value to immediate subscribers, then notifies reactive dependents.
    ///
    /// Subscribers run untracked: their reads never become dependencies of whatever observer
    /// is emitting. The subscriber list is snapshotted when `emit` begins, so a handler that
    /// unsubscribes (or a subscriber added) during delivery takes effect from the next emit —
    /// the current emit still reaches the original set.
    #[track_caller]
    pub fn emit(&self, value: T) {
        let subscribers = self
            .inner
            .subscribers
            .borrow()
            .values()
            .cloned()
            .collect::<Vec<_>>();
        tracing::debug!(
            target: trace_targets::EVENT,
            event = "emit_event",
            event_id = self.inner.id.0,
            subscriber_count = subscribers.len(),
            "emitting event value"
        );
        // Immediate subscribers are not reactive observers: suspend tracking so that a
        // subscriber's reads never become dependencies of whatever observer is emitting.
        crate::untrack(|| {
            for subscriber in subscribers {
                subscriber(&value);
            }
        });
        self.inner.reactor.trigger(self.inner.id);
    }

    /// Adds an immediate subscriber to this event.
    pub fn subscribe(&self, handler: impl Fn(&T) + 'static) -> Subscription {
        let id = self.inner.next_subscriber.get();
        self.inner.next_subscriber.set(id.wrapping_add(1));
        self.inner
            .subscribers
            .borrow_mut()
            .insert(id, Rc::new(handler) as Rc<SubscriberFn<T>>);
        tracing::debug!(
            target: trace_targets::EVENT,
            event = "subscribe_event",
            event_id = self.inner.id.0,
            subscription_id = id,
            subscriber_count = self.inner.subscribers.borrow().len(),
            "added event subscriber"
        );

        let inner = Rc::clone(&self.inner);
        Subscription::new(move || {
            #[cfg(debug_assertions)]
            tracing::trace!(
                target: trace_targets::EVENT,
                event = "unsubscribe_event",
                event_id = inner.id.0,
                subscription_id = id,
                "removing event subscriber"
            );
            inner.subscribers.borrow_mut().remove(&id);
        })
    }
}

impl<T> Drop for EventInner<T> {
    fn drop(&mut self) {
        self.reactor.dispose(self.id);
    }
}

impl Subscription {
    #[track_caller]
    fn new(cancel: impl FnOnce() + 'static) -> Self {
        let inner = Rc::new(SubscriptionInner {
            active: Cell::new(true),
            cancel: RefCell::new(Some(Box::new(cancel))),
        });
        // If an owner (an enclosing effect run or scope) is active, it keeps this subscription
        // alive and cancels it. Crucially, for `on` this ties the queue-feeding direct
        // subscriber and the draining effect together: the owner tears both down through the
        // subscription's cancel closure, rather than disposing the effect and leaving the
        // queue to grow unread.
        let owned: Rc<dyn OwnedDisposable> = inner.clone();
        let _ = adopt_into_current(owned);
        Self { inner }
    }

    /// Cancels the subscription immediately.
    pub fn unsubscribe(&self) {
        self.inner.unsubscribe();
    }

    /// Returns `true` if the subscription is still active.
    pub fn is_active(&self) -> bool {
        self.inner.active.get()
    }

    /// Consumes the handle without cancelling the subscription, letting an unowned subscriber
    /// stay active for the remainder of the program.
    ///
    /// This forfeits only the handle's lifetime management: a subscription created inside an
    /// owner (an effect's run, or a [`crate::scope`]) is still cancelled with that owner — `leak`
    /// is not a way to detach from it, and calling it there costs an allocation without buying
    /// any lifetime. The same carve-out as [`crate::EffectHandle::leak`]. You CANNOT recover a
    /// `Subscription` after calling this method, so be sure to call it on a subscription you will
    /// never need to cancel.
    pub fn leak(self) {
        core::mem::forget(self);
    }
}

impl<T> core::fmt::Debug for Event<T> {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Event")
            .field("id", &self.inner.id)
            .finish_non_exhaustive()
    }
}

impl core::fmt::Debug for Subscription {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Subscription")
            .field("active", &self.inner.active.get())
            .finish()
    }
}

struct EventInner<T> {
    reactor: Reactor,
    id: NodeId,
    next_subscriber: Cell<usize>,
    subscribers: RefCell<BTreeMap<usize, Rc<SubscriberFn<T>>>>,
}

struct SubscriptionInner {
    active: Cell<bool>,
    /// Runs once, and is released as it runs.
    ///
    /// `FnOnce` in an `Option` rather than a `Fn`, because what this closure captures is not
    /// incidental: for [`Event::subscribe`] it is an `Rc<EventInner>`, and for [`Reactor::on`] it
    /// is the draining [`crate::EffectHandle`] and the consumer's handler as well. Keeping it
    /// after cancellation means a `Subscription` the consumer has explicitly cancelled — and
    /// which reports `is_active() == false` — still pins its event's graph node and the whole
    /// handler environment, for as long as any handle is held. `DiagnosticSubscription` in
    /// `crate::diagnostics` has always had this shape; this is the same reasoning.
    cancel: RefCell<Option<Box<dyn FnOnce() + 'static>>>,
}

impl OwnedDisposable for SubscriptionInner {
    fn dispose_owned(&self) {
        self.unsubscribe();
    }

    fn is_disposed_owned(&self) -> bool {
        !self.active.get()
    }
}

impl SubscriptionInner {
    fn unsubscribe(&self) {
        if !self.active.replace(false) {
            return;
        }
        #[cfg(debug_assertions)]
        tracing::trace!(
            target: trace_targets::EVENT,
            event = "unsubscribe",
            "cancelling subscription"
        );
        // Taken before it is called, not after: the borrow must not be held while consumer code
        // runs (cancelling `on` disposes an effect, which runs cleanups), and a closure that
        // panics must still be released rather than left to pin its event forever.
        let cancel = self.cancel.borrow_mut().take();
        if let Some(cancel) = cancel {
            cancel();
        }
    }
}

impl Drop for SubscriptionInner {
    fn drop(&mut self) {
        self.unsubscribe();
    }
}

#[cfg(test)]
mod ordering_tests {
    use alloc::rc::Rc;
    use core::cell::RefCell;

    use crate::{Reactor, event_in};

    #[test]
    fn immediate_subscribers_run_in_registration_order() {
        // Hash iteration made this arbitrary and different on every process run, so a consumer
        // registering a logger before a handler had no guarantee about which ran first.
        let reactor = Reactor::new();
        let stream = event_in::<u32>(&reactor);
        let seen = Rc::new(RefCell::new(Vec::new()));

        let _subs = (0..8)
            .map(|i| {
                let seen = Rc::clone(&seen);
                stream.subscribe(move |_| seen.borrow_mut().push(i))
            })
            .collect::<Vec<_>>();
        stream.emit(1);

        assert_eq!(*seen.borrow(), [0, 1, 2, 3, 4, 5, 6, 7]);
    }

    #[test]
    fn a_re_entrant_emit_still_drains_in_emission_order() {
        // `on` documents delivery "in emission order". Its queue is fed by an ordinary immediate
        // subscriber, so when another subscriber re-emits, the two enqueues raced in hash order:
        // the nested value could be queued first and delivered before the value that caused it.
        let reactor = Reactor::new();
        let stream = event_in::<u32>(&reactor);
        let drained = Rc::new(RefCell::new(Vec::new()));

        let _drain = reactor.on(&stream, {
            let drained = Rc::clone(&drained);
            move |value| drained.borrow_mut().push(*value)
        });
        let _echo = stream.subscribe({
            let stream = stream.clone();
            move |value| {
                if *value < 100 {
                    stream.emit(value + 100);
                }
            }
        });

        stream.emit(1);
        reactor.flush_now();
        assert_eq!(
            *drained.borrow(),
            [1, 101],
            "the value that triggered the nested emit must be delivered first"
        );
    }
}

#[cfg(test)]
mod tests {
    use std::cell::RefCell;
    use std::rc::Rc;

    use runite::{queue_macrotask, run};

    use crate::{Reactor, event_in};

    use super::{Subscription, on_in};

    #[test]
    fn emit_delivers_immediately_and_on_drains_reactively() {
        let immediate = Rc::new(RefCell::new(Vec::new()));
        let reactive = Rc::new(RefCell::new(Vec::new()));
        let keep_alive = Rc::new(RefCell::new(Vec::<Subscription>::new()));

        queue_macrotask({
            let immediate = Rc::clone(&immediate);
            let reactive = Rc::clone(&reactive);
            let keep_alive = Rc::clone(&keep_alive);
            move || {
                let reactor = Reactor::new();
                let event = event_in::<usize>(&reactor);

                let direct = event.subscribe({
                    let immediate = Rc::clone(&immediate);
                    move |value| immediate.borrow_mut().push(*value)
                });
                let draining = on_in(&reactor, &event, {
                    let reactive = Rc::clone(&reactive);
                    move |value| reactive.borrow_mut().push(*value)
                });

                event.emit(1);
                event.emit(2);

                assert_eq!(&*immediate.borrow(), &[1, 2]);
                assert!(
                    reactive.borrow().is_empty(),
                    "reactive drain should be deferred"
                );

                keep_alive.borrow_mut().extend([direct, draining]);
            }
        });

        run();
        assert_eq!(&*reactive.borrow(), &[1, 2]);
    }

    #[test]
    fn on_handlers_run_untracked() {
        let drains = Rc::new(RefCell::new(0usize));

        queue_macrotask({
            let drains = Rc::clone(&drains);
            move || {
                let reactor = Reactor::new();
                let event = event_in::<usize>(&reactor);
                let unrelated = crate::signal_in(&reactor, 0usize);

                let subscription = on_in(&reactor, &event, {
                    let drains = Rc::clone(&drains);
                    let unrelated = unrelated.clone();
                    move |_value| {
                        // The handler reads a signal it must not subscribe to.
                        let _ = unrelated.get();
                        *drains.borrow_mut() += 1;
                    }
                });

                event.emit(1);
                reactor.flush_now();
                assert_eq!(*drains.borrow(), 1);

                // If the handler's read had been tracked, this write would re-run the draining
                // effect (delivering nothing, but observable via graph churn on later emits).
                unrelated.set(99);
                reactor.flush_now();
                assert_eq!(*drains.borrow(), 1, "handler reads must not be tracked");

                event.emit(2);
                reactor.flush_now();
                assert_eq!(*drains.borrow(), 2);

                subscription.leak();
            }
        });

        run();
    }

    #[test]
    fn subscriptions_created_inside_a_scope_are_cancelled_with_it() {
        let seen = Rc::new(RefCell::new(Vec::new()));

        queue_macrotask({
            let seen = Rc::clone(&seen);
            move || {
                let reactor = Reactor::new();
                let event = event_in::<usize>(&reactor);

                // The scope owns the draining subscription: BOTH halves of `on` (the
                // queue-feeding direct subscriber and the draining effect) must be torn down
                // together when the scope is disposed.
                let (scope_handle, subscription) = crate::scope({
                    let reactor = reactor.clone();
                    let event = event.clone();
                    let seen = Rc::clone(&seen);
                    move || {
                        on_in(&reactor, &event, move |value| {
                            seen.borrow_mut().push(*value);
                        })
                    }
                });

                event.emit(1);
                reactor.flush_now();
                assert!(subscription.is_active());

                scope_handle.dispose();
                assert!(
                    !subscription.is_active(),
                    "the owner must cancel the subscription"
                );

                event.emit(2);
                reactor.flush_now();
            }
        });

        run();
        assert_eq!(
            &*seen.borrow(),
            &[1],
            "events after scope disposal must not be delivered"
        );
    }

    #[test]
    fn self_unsubscribe_during_emit_applies_to_the_next_emit() {
        let seen = Rc::new(RefCell::new(Vec::new()));
        let subscription_slot = Rc::new(RefCell::new(None::<Subscription>));

        let reactor = Reactor::new();
        let event = event_in::<usize>(&reactor);

        let subscription = event.subscribe({
            let seen = Rc::clone(&seen);
            let subscription_slot = Rc::clone(&subscription_slot);
            move |value| {
                seen.borrow_mut().push(*value);
                if let Some(subscription) = subscription_slot.borrow().as_ref() {
                    subscription.unsubscribe();
                }
            }
        });
        *subscription_slot.borrow_mut() = Some(subscription);

        // The emit that triggers the unsubscribe still delivers (subscribers are snapshotted),
        // but the next emit must not.
        event.emit(1);
        event.emit(2);
        assert_eq!(&*seen.borrow(), &[1]);
    }

    #[test]
    fn dropping_a_subscription_cancels_it() {
        let seen = Rc::new(RefCell::new(Vec::new()));

        let reactor = Reactor::new();
        let event = event_in::<usize>(&reactor);

        let subscription = event.subscribe({
            let seen = Rc::clone(&seen);
            move |value| seen.borrow_mut().push(*value)
        });
        event.emit(1);
        assert!(subscription.is_active());

        drop(subscription);
        event.emit(2);
        assert_eq!(&*seen.borrow(), &[1], "dropped subscription must not fire");
    }

    #[test]
    fn leaked_subscriptions_stay_active() {
        let seen = Rc::new(RefCell::new(Vec::new()));

        let reactor = Reactor::new();
        let event = event_in::<usize>(&reactor);

        event
            .subscribe({
                let seen = Rc::clone(&seen);
                move |value| seen.borrow_mut().push(*value)
            })
            .leak();
        event.emit(1);
        event.emit(2);
        assert_eq!(&*seen.borrow(), &[1, 2]);
    }
}