haematite 0.7.0

//! Census row W: the wasm (`browser_local`) parity of the root-advance event seam
//! (`docs/design/ROOT-ADVANCE-SEAM.md` §3, §2). Same surface as the native seam,
//! SINGLE-THREADED: the subscription callback drops the `Send + Sync` bound and
//! emission runs on the one browser thread. The advance invariant, the monotone
//! delivery filter (latest-wins per shard), and the R3 write wall are identical.
//!
//! This subsystem never shares an address space with the native `db` seam (the two
//! are mutually-exclusive cfg builds), so the event type is mirrored here rather
//! than imported from `crate::db::root_advance` (which is native-only).

use std::cell::{Cell, RefCell};
use std::collections::BTreeMap;
use std::rc::Rc;

use crate::tree::Hash;

/// A durable committed-root advance for one shard, wasm rung (design §2).
///
/// Mirrors `crate::db::root_advance::RootAdvance` field-for-field. `advance_gen`
/// counts root ADVANCES of this shard only (`+1` iff `prior != new`); assigned gens
/// are gap-free, but DELIVERED tells strictly increase and may skip superseded
/// values (latest-wins). First-commit `prior_root` is the empty-root constant.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RootAdvance {
    /// The shard whose committed root advanced.
    pub shard_id: usize,
    /// The root this commit replaced (the empty-root constant for the first).
    #[serde(with = "hash_serde")]
    pub prior_root: Hash,
    /// The durably committed root.
    #[serde(with = "hash_serde")]
    pub new_root: Hash,
    /// `+1` per root ADVANCE of this shard (iff `prior != new`); process-lifetime
    /// monotonic. DELIVERED tells strictly increase but may skip superseded values.
    pub advance_gen: u64,
}

/// Per-shard emission state: the `advance_gen` slot plus the `last_told_gen`
/// high-water mark of the monotone delivery filter. `RefCell`/`Cell`, not a mutex —
/// the browser rung is single-threaded.
#[derive(Debug, Default)]
struct ShardEmit {
    assigned_gen: u64,
    last_told_gen: u64,
}

/// One registered subscriber: cancellation id + single-threaded callback.
struct Subscriber {
    id: u64,
    callback: Rc<dyn Fn(RootAdvance)>,
}

#[derive(Default)]
struct SeamInner {
    next_id: u64,
    subscribers: Vec<Subscriber>,
    shards: BTreeMap<usize, ShardEmit>,
}

/// The wasm root-advance seam: a process-local subscriber registry plus per-shard
/// emission state, shared by cheap `Rc` clone. Inert with zero subscribers.
#[derive(Clone, Default)]
pub(super) struct BrowserRootAdvanceSeam {
    inner: Rc<RefCell<SeamInner>>,
}

thread_local! {
    /// True while a row-W callback runs. The R3 write wall reads it; a write issued
    /// from inside a callback is refused (single-threaded self-recursion / borrow
    /// re-entry). Single browser thread, so a plain `Cell` suffices.
    static IN_EMISSION: Cell<bool> = const { Cell::new(false) };
}

/// Whether the current (single) browser thread is inside a row-W callback.
pub(super) fn in_emission() -> bool {
    IN_EMISSION.with(Cell::get)
}

/// RAII guard raising the R3 in-emission flag for a callback batch; restores the
/// prior value on drop (including panic unwind).
struct InEmissionGuard {
    previous: bool,
}

impl InEmissionGuard {
    fn enter() -> Self {
        Self {
            previous: IN_EMISSION.with(|flag| flag.replace(true)),
        }
    }
}

impl Drop for InEmissionGuard {
    fn drop(&mut self) {
        IN_EMISSION.with(|flag| flag.set(self.previous));
    }
}

impl BrowserRootAdvanceSeam {
    pub(super) fn new() -> Self {
        Self::default()
    }

    fn subscribe(&self, callback: Rc<dyn Fn(RootAdvance)>) -> u64 {
        let mut inner = self.inner.borrow_mut();
        let id = inner.next_id;
        inner.next_id = inner.next_id.wrapping_add(1);
        inner.subscribers.push(Subscriber { id, callback });
        id
    }

    fn cancel(&self, id: u64) {
        self.inner
            .borrow_mut()
            .subscribers
            .retain(|subscriber| subscriber.id != id);
    }

    /// Assign this shard's next `advance_gen` (called once per advance, in commit
    /// order) and emit the tell through the monotone filter. `prior != new` is the
    /// caller's obligation (the advance invariant, §2): this is only called for a
    /// shard whose root actually moved.
    ///
    /// The subscriber snapshot is cloned out of the registry BEFORE any callback
    /// runs, so a subscribe/cancel from within a callback cannot disturb this
    /// delivery, and the `RefCell` is never borrowed across a callback (mirrors the
    /// native seam holding only its leaf emission mutex — R1).
    pub(super) fn emit_advance(&self, shard_id: usize, prior_root: Hash, new_root: Hash) {
        let (advance_gen, subscribers) = {
            let mut inner = self.inner.borrow_mut();
            let shard = inner.shards.entry(shard_id).or_default();
            shard.assigned_gen = shard.assigned_gen.wrapping_add(1);
            let advance_gen = shard.assigned_gen;
            // Monotone delivery filter: this freshly-assigned gen is always the
            // highest, so it is never superseded; advance last_told BEFORE invoking.
            shard.last_told_gen = advance_gen;
            let subscribers: Vec<Rc<dyn Fn(RootAdvance)>> = inner
                .subscribers
                .iter()
                .map(|subscriber| Rc::clone(&subscriber.callback))
                .collect();
            (advance_gen, subscribers)
        };
        if subscribers.is_empty() {
            return;
        }
        let event = RootAdvance {
            shard_id,
            prior_root,
            new_root,
            advance_gen,
        };
        let _wall = InEmissionGuard::enter();
        for callback in &subscribers {
            callback(event);
        }
    }
}

/// An RAII subscription to a browser-local store's root-advance seam.
///
/// Unsubscribes on drop and by explicit [`BrowserRootAdvanceSubscription::cancel`].
/// Single-threaded: the callback needs no `Send + Sync`.
#[must_use = "dropping the subscription immediately unsubscribes"]
pub struct BrowserRootAdvanceSubscription {
    seam: BrowserRootAdvanceSeam,
    id: u64,
    active: bool,
}

impl BrowserRootAdvanceSubscription {
    pub(super) fn new(seam: &BrowserRootAdvanceSeam, callback: Rc<dyn Fn(RootAdvance)>) -> Self {
        let id = seam.subscribe(callback);
        Self {
            seam: seam.clone(),
            id,
            active: true,
        }
    }

    /// Explicitly unsubscribe now. Idempotent; a later drop is a no-op.
    pub fn cancel(mut self) {
        self.deactivate();
    }

    fn deactivate(&mut self) {
        if self.active {
            self.active = false;
            self.seam.cancel(self.id);
        }
    }
}

impl Drop for BrowserRootAdvanceSubscription {
    fn drop(&mut self) {
        self.deactivate();
    }
}

mod hash_serde {
    use crate::tree::Hash;
    use serde::{Deserialize, Deserializer, Serialize, Serializer};

    pub(super) fn serialize<S: Serializer>(hash: &Hash, serializer: S) -> Result<S::Ok, S::Error> {
        hash.as_bytes().serialize(serializer)
    }

    pub(super) fn deserialize<'de, D: Deserializer<'de>>(
        deserializer: D,
    ) -> Result<Hash, D::Error> {
        let bytes = <[u8; crate::tree::node::HASH_SIZE]>::deserialize(deserializer)?;
        Ok(Hash::from_bytes(bytes))
    }
}

// §9.10 wasm parity note: `browser_local` is a `wasm32`-only subsystem and the
// crate's dev-dependency graph (tokio/mio) does not build for
// wasm32-unknown-unknown, so there is NO test harness that can compile or run a
// browser_local unit test in this repo (the wasm-clippy CI job builds the lib
// only, not test targets). Row W's delivery logic — the monotone `advance_gen`
// assignment, latest-wins per shard, zero-subscriber inertness, and the R3 wall —
// is byte-for-byte the same shape as the native seam, which the §9.1–§9.9 tests
// exercise directly; row W itself is compile-gated under BOTH wasm clippy
// compositions (`--features wasm` and `--features wasm-runtime`).