pub mod backend;
pub mod builder;
pub mod config;
pub mod util;
#[cfg(feature = "json")]
pub use backend::json::JsonStore;
use std::collections::{HashMap, HashSet};
use std::fmt::Display;
use std::hash::Hash;
use std::str::FromStr;
#[cfg(feature = "redb")]
pub use backend::redb::RedbStore;
use crate::migration::set::MigrationSet;
use crate::reactive::{MapChange, ReactiveMap, SubscriberAny, SubscriberKey};
use crate::{AccessMode, Field, MigrationReport, Result, Signal, StoreSubscription, WritableMode};
use bytes::Bytes;
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::sync::atomic::AtomicUsize;
use std::sync::{Arc, Mutex};
pub type SubscriptionId = u64;
pub type StoreCallback = Arc<dyn Fn(&StoreEvent) + Send + Sync + 'static>;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StoreOp {
Set,
Delete,
Patch,
}
#[derive(Debug, Clone)]
pub struct StoreEvent {
pub path: Arc<str>,
pub op: StoreOp,
pub old: Option<Bytes>,
pub new: Option<Bytes>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SubscriptionKind {
Any,
ExactPath(Arc<str>),
Prefix(Arc<str>),
}
pub trait Store: Send + Sync + 'static {
fn get<T: DeserializeOwned>(&self, path: &str) -> Result<Option<T>>;
fn set<T: Serialize>(&self, path: &str, value: &T) -> Result<()>;
fn set_owned<T: Serialize>(&self, path: Arc<str>, value: &T) -> Result<()> {
self.set(&path, value)
}
fn scan_prefix(&self, prefix: &str) -> Result<Vec<(String, Bytes)>>;
fn delete(&self, path: &str) -> Result<()>;
fn subscribe(&self, kind: SubscriptionKind, callback: StoreCallback) -> SubscriptionId;
fn unsubscribe(&self, id: SubscriptionId);
fn decode<T: DeserializeOwned + Default>(&self, bytes: &[u8]) -> Result<T>;
fn flush_prefix(&self, prefix: &str) -> Result<()>;
}
pub trait SchemaAwareStore: Store {
fn run_migrations(&self, mset: MigrationSet) -> Result<MigrationReport>;
}
pub trait StateScope {
const PREFIX: &'static str;
}
pub fn scoped_path<T: StateScope>(key: &str) -> String {
if key.trim().is_empty() {
return T::PREFIX.to_string();
}
format!(
"{}.{}",
T::PREFIX.trim_end_matches('.'),
key.trim_start_matches('.')
)
}
pub fn field<TScope, TValue, S>(
store: &Arc<S>,
key: &str,
default: TValue,
) -> Result<Field<TValue, S, WritableMode>>
where
TScope: StateScope,
S: Store,
TValue: Serialize + DeserializeOwned + Default + Clone + Send + Sync + 'static,
{
let path: Arc<str> = scoped_path::<TScope>(key).into();
field_with_path(store, path, default)
}
pub fn field_with_path<TValue, S, M>(
store: &Arc<S>,
path: Arc<str>,
default: TValue,
) -> Result<Field<TValue, S, M>>
where
S: Store,
TValue: Serialize + DeserializeOwned + Default + Clone + Send + Sync + 'static,
M: AccessMode,
{
if store.get::<TValue>(&path)?.is_none() {
store.set(&path, &default)?;
}
let current = store
.get::<TValue>(&path)?
.unwrap_or_else(|| default.clone());
let signal = Signal::new(current);
let sig_clone = signal.clone();
let store_clone = Arc::clone(store);
let path_log = Arc::clone(&path);
let id = store.subscribe(
SubscriptionKind::ExactPath(path.clone()),
Arc::new(move |event| {
if let Some(raw) = &event.new {
match store_clone.decode::<TValue>(raw) {
Ok(parsed) => sig_clone.set(parsed),
Err(e) => tracing::error!(path = %path_log, error = %e, "decode failed"),
}
}
}),
);
Ok(Field {
signal,
path,
store_sub: Some(Arc::new(StoreSubscription {
store: Arc::clone(store),
id,
})),
_mode: std::marker::PhantomData,
interceptors: Arc::new(Mutex::new(Vec::new())),
next_interceptor_id: Arc::new(AtomicUsize::new(0)),
intercept_depth: Arc::new(AtomicUsize::new(0)),
})
}
pub fn reactive_map<TScope, K, V, S>(
store: &Arc<S>,
key: &str,
default: HashMap<K, V>,
) -> Result<ReactiveMap<K, V, S, WritableMode>>
where
TScope: StateScope,
S: Store,
K: FromStr + Display + Clone + Hash + Eq + Send + Sync + 'static,
V: Serialize + DeserializeOwned + Default + Clone + Send + Sync + 'static,
{
let path: Arc<str> = scoped_path::<TScope>(key).into();
reactive_map_with_path(store, path, default)
}
pub fn reactive_map_with_path<K, V, S, M>(
store: &Arc<S>,
path: Arc<str>,
default: HashMap<K, V>,
) -> Result<ReactiveMap<K, V, S, M>>
where
S: Store,
K: FromStr + Display + Clone + Hash + Eq + Send + Sync + 'static,
V: Serialize + Default + DeserializeOwned + Clone + Send + Sync + 'static,
M: AccessMode,
{
let mut known_keys = HashSet::new();
let prefix = format!("{}.", path);
let existing = store.scan_prefix(&prefix)?;
for (fpath, _) in existing {
if let Some(k_str) = fpath.strip_prefix(&prefix)
&& let Ok(k) = K::from_str(k_str)
{
known_keys.insert(k);
}
}
for (k, v) in default {
let full_path = format!("{}.{}", path, k);
if store.get::<V>(&full_path)?.is_none() {
store.set(&full_path, &v)?;
}
}
let known_keys = Arc::new(std::sync::Mutex::new(known_keys));
let subscribers_any = Arc::new(Mutex::new(Vec::<(u64, SubscriberAny<K, V>)>::new()));
let subscribers_key = Arc::new(Mutex::new(
HashMap::<K, Vec<(u64, SubscriberKey<K, V>)>>::new(),
));
let subs_any = Arc::clone(&subscribers_any);
let subs_key = Arc::clone(&subscribers_key);
let keys_tracker = Arc::clone(&known_keys);
let path_for_sub = path.clone();
let prefix_for_strip = format!("{}.", path);
let store_clone = Arc::clone(store);
let id = store.subscribe(
SubscriptionKind::Prefix(path_for_sub),
Arc::new(move |event| {
if let Some(key_str) = event.path.strip_prefix(&prefix_for_strip)
&& let Ok(k) = K::from_str(key_str)
{
let mut keys = keys_tracker.lock().unwrap();
let new_val = event
.new
.as_ref()
.and_then(|b| store_clone.decode::<V>(b).ok());
let old_val = event
.old
.as_ref()
.and_then(|b| store_clone.decode::<V>(b).ok());
let change = match event.op {
StoreOp::Set | StoreOp::Patch => {
if keys.contains(&k) {
MapChange::Update {
key: k.clone(),
old_value: old_val.unwrap_or_default(),
new_value: new_val.unwrap_or_default(),
}
} else {
keys.insert(k.clone());
MapChange::Insert {
key: k.clone(),
value: new_val.unwrap_or_default(),
}
}
}
StoreOp::Delete => {
keys.remove(&k);
MapChange::Remove {
key: k.clone(),
old_value: old_val.unwrap_or_default(),
}
}
};
if let Ok(lock) = subs_key.lock()
&& let Some(cbs) = lock.get(&k)
{
for (_, cb) in cbs {
cb(&change);
}
}
if let Ok(lock) = subs_any.lock() {
for (_, cb) in lock.iter() {
cb(&change);
}
}
}
}),
);
Ok(ReactiveMap {
path,
store: Arc::clone(store),
_mode: std::marker::PhantomData,
_key: std::marker::PhantomData,
_value: std::marker::PhantomData,
interceptors_any: Arc::new(std::sync::Mutex::new(Vec::new())),
interceptors_key: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
subscribers_any,
subscribers_key,
next_id: Arc::new(std::sync::atomic::AtomicU64::new(0)),
intercept_depth: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
store_sub: Arc::new(StoreSubscription {
store: Arc::clone(store),
id,
}),
known_keys,
})
}
#[derive(Clone)]
pub struct SubscriptionEntry {
pub id: SubscriptionId,
pub kind: SubscriptionKind,
pub callback: StoreCallback,
}
pub 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('.'))
}
}
}