mod dashboard;
pub mod sql;
pub use dashboard::DbPlugin;
use larust_core::AppError;
use redb::{Database, ReadableTable, TableDefinition};
use serde::de::DeserializeOwned;
use serde::Serialize;
use std::path::Path;
use std::sync::{Arc, OnceLock};
const TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("kv");
static DATABASE: OnceLock<Arc<Database>> = OnceLock::new();
pub async fn connect(path: impl AsRef<Path>) -> Result<(), AppError> {
let path = path.as_ref().to_path_buf();
let db = tokio::task::spawn_blocking(move || -> Result<Database, AppError> {
if let Some(parent) = path.parent() {
if !parent.as_os_str().is_empty() {
std::fs::create_dir_all(parent).map_err(internal)?;
}
}
let db = Database::create(&path).map_err(internal)?;
let txn = db.begin_write().map_err(internal)?;
{
txn.open_table(TABLE).map_err(internal)?;
}
txn.commit().map_err(internal)?;
Ok(db)
})
.await
.map_err(internal)??;
DATABASE.set(Arc::new(db)).map_err(|_| {
AppError::Internal(Box::new(std::io::Error::other(
"connect() called more than once",
)))
})
}
fn store() -> Result<Arc<Database>, AppError> {
DATABASE.get().cloned().ok_or_else(|| {
AppError::Internal(Box::new(std::io::Error::other(
"embedded db not connected; call larust_db::connect() (via \
larust_support::db::connect) at startup before using it",
)))
})
}
pub async fn put<T: Serialize>(key: &str, value: &T) -> Result<(), AppError> {
let value = serde_json::to_value(value).map_err(internal)?;
put_raw(key, value).await
}
pub async fn get<T: DeserializeOwned>(key: &str) -> Result<Option<T>, AppError> {
match get_raw(key).await? {
Some(value) => Ok(Some(serde_json::from_value(value).map_err(internal)?)),
None => Ok(None),
}
}
pub async fn forget(key: &str) -> Result<(), AppError> {
let db = store()?;
let key = key.to_string();
tokio::task::spawn_blocking(move || {
let txn = db.begin_write().map_err(internal)?;
{
let mut table = txn.open_table(TABLE).map_err(internal)?;
table.remove(key.as_str()).map_err(internal)?;
}
txn.commit().map_err(internal)?;
Ok(())
})
.await
.map_err(internal)?
}
pub async fn keys() -> Result<Vec<String>, AppError> {
let db = store()?;
tokio::task::spawn_blocking(move || {
let txn = db.begin_read().map_err(internal)?;
let table = txn.open_table(TABLE).map_err(internal)?;
let mut out = Vec::new();
for entry in table.iter().map_err(internal)? {
let (k, _v) = entry.map_err(internal)?;
out.push(k.value().to_string());
}
Ok(out)
})
.await
.map_err(internal)?
}
pub async fn get_raw(key: &str) -> Result<Option<serde_json::Value>, AppError> {
let db = store()?;
let key = key.to_string();
tokio::task::spawn_blocking(move || {
let txn = db.begin_read().map_err(internal)?;
let table = txn.open_table(TABLE).map_err(internal)?;
match table.get(key.as_str()).map_err(internal)? {
Some(value) => Ok(Some(
serde_json::from_slice(value.value()).map_err(internal)?,
)),
None => Ok(None),
}
})
.await
.map_err(internal)?
}
pub async fn put_raw(key: &str, value: serde_json::Value) -> Result<(), AppError> {
let db = store()?;
let key = key.to_string();
tokio::task::spawn_blocking(move || {
let bytes = serde_json::to_vec(&value).map_err(internal)?;
let txn = db.begin_write().map_err(internal)?;
{
let mut table = txn.open_table(TABLE).map_err(internal)?;
table
.insert(key.as_str(), bytes.as_slice())
.map_err(internal)?;
}
txn.commit().map_err(internal)?;
Ok(())
})
.await
.map_err(internal)?
}
pub fn parse_cli_value(raw: &str) -> serde_json::Value {
serde_json::from_str(raw).unwrap_or_else(|_| serde_json::Value::String(raw.to_string()))
}
fn internal<E: std::error::Error + Send + Sync + 'static>(error: E) -> AppError {
AppError::Internal(Box::new(error))
}
#[cfg(test)]
mod tests {
use super::*;
use serde::Deserialize;
#[derive(Debug, Serialize, Deserialize, PartialEq)]
struct Item {
name: String,
count: u32,
}
#[tokio::test]
async fn store_behaves_correctly_across_every_scenario() {
let dir = tempfile::tempdir().unwrap();
connect(dir.path().join("test.redb")).await.unwrap();
assert_eq!(get::<Item>("missing").await.unwrap(), None);
assert_eq!(keys().await.unwrap(), Vec::<String>::new());
let item = Item {
name: "widget".to_string(),
count: 3,
};
put("item", &item).await.unwrap();
assert_eq!(get::<Item>("item").await.unwrap(), Some(item));
assert_eq!(keys().await.unwrap(), vec!["item".to_string()]);
forget("item").await.unwrap();
assert_eq!(get::<Item>("item").await.unwrap(), None);
assert_eq!(keys().await.unwrap(), Vec::<String>::new());
forget("nope").await.unwrap();
put("count", &1).await.unwrap();
put("count", &2).await.unwrap();
assert_eq!(get::<i64>("count").await.unwrap(), Some(2));
put_raw("raw", serde_json::json!({"a": 1, "b": [true, null]}))
.await
.unwrap();
assert_eq!(
get_raw("raw").await.unwrap(),
Some(serde_json::json!({"a": 1, "b": [true, null]}))
);
let second = tempfile::tempdir().unwrap();
let err = connect(second.path().join("other.redb")).await.unwrap_err();
assert!(err.to_string().contains("more than once"));
}
#[test]
fn parse_cli_value_parses_real_json() {
assert_eq!(parse_cli_value("42"), serde_json::json!(42));
assert_eq!(parse_cli_value("true"), serde_json::json!(true));
assert_eq!(parse_cli_value("\"Alice\""), serde_json::json!("Alice"));
}
#[test]
fn parse_cli_value_falls_back_to_a_plain_string() {
assert_eq!(parse_cli_value("Alice"), serde_json::json!("Alice"));
}
}