amethystate 0.10.0

use crate::SubscriptionKind;
#[cfg(any(feature = "redb", feature = "sqlite"))]
use crate::store::util::debouncer::Debouncer;
use crate::store::{StoreEvent, SubscriptionEntry};
#[cfg(any(feature = "redb", feature = "sqlite"))]
use crate::{StorageResult, StoreOp};
#[cfg(any(feature = "redb", feature = "sqlite"))]
use parking_lot::Mutex;
use parking_lot::RwLock;
#[cfg(any(feature = "redb", feature = "sqlite"))]
use std::sync::Arc;

pub fn emit_events(subs_lock: &RwLock<Vec<SubscriptionEntry>>, event: StoreEvent) {
    let callbacks = {
        let guard = subs_lock.read();
        guard
            .iter()
            .filter(|s| matches_kind(&s.kind, &event.path))
            .map(|s| s.callback.clone())
            .collect::<Vec<_>>()
    };
    for cb in callbacks {
        cb(&event);
    }
}

fn matches_kind(kind: &SubscriptionKind, path: &str) -> bool {
    match kind {
        SubscriptionKind::Any => true,
        SubscriptionKind::ExactPath(p) => **p == *path,
        SubscriptionKind::Prefix(prefix) => {
            *path == **prefix
                || path
                    .strip_prefix(&**prefix)
                    .is_some_and(|t| t.starts_with('.'))
        }
    }
}

#[cfg(any(feature = "sqlite", feature = "redb"))]
pub type Pending = std::collections::HashMap<std::sync::Arc<str>, Option<Vec<u8>>>;

/// Everything buffered under `prefix`, left in place.
///
/// The buffer is only cleared once the write has actually landed, by
/// [`clear_committed`]. Taking entries out first meant any error below lost
/// them: not on disk, not in memory, and nothing left to retry.
#[cfg(any(feature = "sqlite", feature = "redb"))]
pub fn pending_prefix(pending: &Pending, prefix: &str) -> Pending {
    if pending.is_empty() {
        return Pending::new();
    }

    if prefix.is_empty() {
        return pending.clone();
    }

    let prefix_dot = format!("{}.", prefix);
    pending
        .iter()
        .filter(|(k, _)| k.starts_with(&prefix_dot) || &***k == prefix)
        .map(|(k, v)| (k.clone(), v.clone()))
        .collect()
}

/// Drops from the buffer exactly what was committed.
///
/// A key whose buffered value has changed since is a write that landed while
/// the commit was in flight; it is not on disk, so it stays for the next one.
#[cfg(any(feature = "sqlite", feature = "redb"))]
pub fn clear_committed(pending: &mut Pending, committed: &Pending) {
    for (key, value) in committed {
        if pending.get(key) == Some(value) {
            pending.remove(key);
        }
    }
}

#[cfg(any(feature = "redb", feature = "sqlite"))]
pub fn set_raw_pending(
    pending: &Mutex<std::collections::HashMap<Arc<str>, Option<Vec<u8>>>>,
    subscriptions: &RwLock<Vec<SubscriptionEntry>>,
    debouncer: &Debouncer,
    key: &str,
    value: &[u8],
) -> StorageResult<()> {
    let key_arc: Arc<str> = Arc::from(key);
    let old_bytes = {
        let lock = pending.lock();
        lock.get(&*key_arc).cloned().flatten()
    };
    {
        let mut lock = pending.lock();
        lock.insert(key_arc.clone(), Some(value.to_vec()));
    }
    emit_events(
        subscriptions,
        StoreEvent {
            path: key_arc,
            op: StoreOp::Set,
            old: old_bytes,
            new: Some(value.to_vec()),
            source: None,
        },
    );
    debouncer.schedule();
    Ok(())
}

#[cfg(all(test, any(feature = "redb", feature = "sqlite")))]
mod tests {
    use super::*;
    use std::sync::Arc;

    fn buffer(entries: &[(&str, Option<&[u8]>)]) -> Pending {
        entries
            .iter()
            .map(|(k, v)| (Arc::from(*k), v.map(|b| b.to_vec())))
            .collect()
    }

    #[test]
    fn collecting_leaves_the_buffer_alone() {
        let pending = buffer(&[("a.x", Some(b"1")), ("a.y", Some(b"2"))]);

        let taken = pending_prefix(&pending, "a");

        assert_eq!(taken.len(), 2);
        assert_eq!(
            pending.len(),
            2,
            "entries must survive until the write lands, or a failure below \
             loses them from memory and disk both"
        );
    }

    #[test]
    fn an_empty_prefix_means_everything() {
        let pending = buffer(&[("a.x", Some(b"1")), ("b.y", Some(b"2"))]);
        assert_eq!(pending_prefix(&pending, "").len(), 2);
    }

    #[test]
    fn a_prefix_matches_its_own_key_and_its_children() {
        let pending = buffer(&[
            ("a", Some(b"root")),
            ("a.x", Some(b"child")),
            ("ab", Some(b"sibling")),
        ]);

        let taken = pending_prefix(&pending, "a");

        assert!(taken.contains_key("a"));
        assert!(taken.contains_key("a.x"));
        assert!(!taken.contains_key("ab"), "a prefix is not a substring");
    }

    #[test]
    fn committed_entries_are_dropped() {
        let mut pending = buffer(&[("a.x", Some(b"1")), ("a.y", Some(b"2"))]);
        let committed = pending.clone();

        clear_committed(&mut pending, &committed);

        assert!(pending.is_empty());
    }

    #[test]
    fn a_value_that_changed_during_the_commit_survives() {
        let committed = buffer(&[("a.x", Some(b"old"))]);
        let mut pending = buffer(&[("a.x", Some(b"new"))]);

        clear_committed(&mut pending, &committed);

        assert_eq!(
            pending.get("a.x"),
            Some(&Some(b"new".to_vec())),
            "the newer write is not on disk, so dropping it would lose it"
        );
    }

    #[test]
    fn a_key_written_after_the_commit_survives() {
        let committed = buffer(&[("a.x", Some(b"1"))]);
        let mut pending = buffer(&[("a.x", Some(b"1")), ("a.z", Some(b"9"))]);

        clear_committed(&mut pending, &committed);

        assert!(!pending.contains_key("a.x"));
        assert!(pending.contains_key("a.z"), "it was never committed");
    }

    #[test]
    fn a_pending_delete_is_committed_like_any_other_entry() {
        let mut pending = buffer(&[("a.x", None)]);
        let committed = pending.clone();

        clear_committed(&mut pending, &committed);

        assert!(pending.is_empty());
    }
}