harn-vm 0.10.161

//! Process-wide registry of observers for committed session-metadata changes.
//!
//! Split out of `session_store` rather than living beside the store opener: it
//! is a notification concern, not a storage one, and the two have no shared
//! state beyond the hook the opener attaches.
//!
//! Writes in this process still publish immediately through the store hook.
//! Writes from another process reach the same SQLite file; [`super::session_wal_watch`]
//! notices those via WAL + `PRAGMA data_version` and publishes through the
//! same fanout.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock, RwLockWriteGuard};

use harn_session_store::{SessionMeta, SharedSessionChangeObserver};

use super::session_wal_watch;

/// Process-wide observers notified when session metadata is committed.
///
/// Deliberately process-scoped, not thread-local like the redaction policy
/// beside it: a store handle is opened on whatever thread needs it, and an
/// update commits on whatever executor thread the caller happens to be on. A
/// thread-local sink would be installed on one thread and silently miss every
/// write made from another, which reads as "notifications do not work" rather
/// than as a wiring mistake.
///
/// A list rather than one slot, because more than one surface can care about
/// the same store and a single slot makes the second registration silently
/// evict the first.
static OBSERVERS: RwLock<Vec<(u64, SharedSessionChangeObserver)>> = RwLock::new(Vec::new());
static NEXT_SUBSCRIPTION: AtomicU64 = AtomicU64::new(1);
static SESSION_MEMORY: RwLock<Vec<(PathBuf, SessionMeta)>> = RwLock::new(Vec::new());

fn observers() -> RwLockWriteGuard<'static, Vec<(u64, SharedSessionChangeObserver)>> {
    OBSERVERS
        .write()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

fn session_memory() -> RwLockWriteGuard<'static, Vec<(PathBuf, SessionMeta)>> {
    SESSION_MEMORY
        .write()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

/// Live registration for [`subscribe`]. Unregisters on drop so
/// a surface that goes away cannot keep receiving, and so a test cannot leak a
/// sink into the next one sharing the process.
#[must_use = "dropping the subscription immediately unregisters the observer"]
pub struct SessionChangeSubscription {
    id: u64,
}

impl Drop for SessionChangeSubscription {
    fn drop(&mut self) {
        observers().retain(|(id, _)| *id != self.id);
        let live = subscriber_count() > 0;
        if !live {
            session_memory().clear();
        }
        session_wal_watch::sync_watchers(live);
    }
}

/// Register an observer notified after any session metadata update commits
/// through a store this VM opens, or after a WAL watcher sees another
/// process rename a session in the same file.
///
/// Every canonical store opened here carries the registered observers, so a
/// surface does not have to be handed the specific handle a writer happened to
/// use.
pub fn subscribe(observer: SharedSessionChangeObserver) -> SessionChangeSubscription {
    let id = NEXT_SUBSCRIPTION.fetch_add(1, Ordering::Relaxed);
    observers().push((id, observer));
    session_wal_watch::sync_watchers(true);
    SessionChangeSubscription { id }
}

/// Claim a canonical store file so a live subscription can watch it.
///
/// The returned registration belongs to the store handle that opened the file.
/// While it lives the path is watchable; when the last handle for that path
/// drops, so does the watcher.
#[must_use = "dropping the registration stops the path being watched"]
pub(crate) fn watch_store(path: &Path) -> session_wal_watch::StoreWatchRegistration {
    let registration = session_wal_watch::register_store_path(path);
    if subscriber_count() > 0 {
        session_wal_watch::sync_watchers(true);
    }
    registration
}

fn subscriber_count() -> usize {
    OBSERVERS
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
        .len()
}

#[derive(Clone, Copy, PartialEq, Eq)]
pub(super) enum SnapshotChange {
    New,
    Unchanged,
    Changed,
}

/// One comparison owner for local commits, foreign commits, and watch baselines.
pub(super) fn remember_snapshot(path: &Path, meta: &SessionMeta) -> SnapshotChange {
    let mut sessions = session_memory();
    if let Some((_, previous)) = sessions
        .iter_mut()
        .find(|(database, seen)| database == path && seen.id == meta.id)
    {
        if previous == meta {
            return SnapshotChange::Unchanged;
        }
        *previous = meta.clone();
        return SnapshotChange::Changed;
    }
    sessions.push((path.to_owned(), meta.clone()));
    SnapshotChange::New
}

/// Fans one committed change out to every live subscriber.
pub(super) fn dispatch(path: &Path, meta: &SessionMeta) {
    dispatch_snapshot(path, meta, true);
}

/// A watcher's first encounter establishes its baseline, not an update.
pub(super) fn dispatch_foreign(path: &Path, meta: &SessionMeta) {
    dispatch_snapshot(path, meta, false);
}

fn dispatch_snapshot(path: &Path, meta: &SessionMeta, publish_new: bool) {
    // Either the local post-commit hook or the WAL reader can arrive first.
    // Claim the whole committed snapshot once, not just its title: model,
    // usage and other metadata changes still notify when the title is stable.
    let change = remember_snapshot(path, meta);
    if change == SnapshotChange::Unchanged || (change == SnapshotChange::New && !publish_new) {
        return;
    }
    publish(meta);
}

fn publish(meta: &SessionMeta) {
    let observers: Vec<SharedSessionChangeObserver> = OBSERVERS
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
        .iter()
        .map(|(_, observer)| Arc::clone(observer))
        .collect();
    // Copy out before dispatching: an observer is allowed to subscribe or
    // unsubscribe in response, which would deadlock against a held guard.
    for observer in observers {
        observer.session_updated(meta);
    }
}

struct SessionChangeFanout {
    database: Option<PathBuf>,
}

impl harn_session_store::SessionChangeObserver for SessionChangeFanout {
    fn session_updated(&self, meta: &SessionMeta) {
        if let Some(path) = self.database.as_deref() {
            dispatch(path, meta);
        } else {
            // In-memory imports have no file watcher and no shared database
            // identity, so they do not participate in file deduplication.
            publish(meta);
        }
    }
}

pub(crate) fn current_observer(database: Option<&Path>) -> Option<SharedSessionChangeObserver> {
    if subscriber_count() == 0 {
        return None;
    }
    Some(Arc::new(SessionChangeFanout {
        database: database.map(Path::to_owned),
    }))
}

#[cfg(test)]
pub(crate) mod test_support {
    use std::sync::OnceLock;

    use tokio::sync::{Mutex, MutexGuard};

    /// Exclusive access to the process-wide change bus for one test.
    ///
    /// The observer list, the remembered titles and the running watchers are
    /// one process singleton, the way the process environment is. Two cases
    /// that subscribe at the same time each receive the other's committed
    /// titles, because a subscriber is registered against the process and not
    /// against a store: the double-publish case then read a sibling's rename
    /// as its own republish (harn#7960). One lock, held for the life of the
    /// subscription, is what makes each case see only its own traffic.
    ///
    /// An async mutex rather than a `std` one because every case that needs it
    /// holds it across an await, which is exactly what a blocking guard must
    /// not do. It also has no poisoning to recover: the lock guards no
    /// invariant of its own, so a panicking holder leaves nothing behind.
    #[must_use = "the bus is shared for as long as the guard lives"]
    pub(crate) async fn exclusive_bus() -> MutexGuard<'static, ()> {
        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
        LOCK.get_or_init(|| Mutex::new(())).lock().await
    }
}