feat: add storage

Signed-off-by: kjuulh <contact@kjuulh.io>
This commit is contained in:
Kasper Juul Hermansen 2024-04-29 23:34:04 +02:00
parent cce97f62f8
commit 8e4b46d2d7
Signed by: kjuulh
GPG Key ID: 57B6E1465221F912
10 changed files with 530 additions and 47 deletions

56
Cargo.lock generated
View File

@ -405,6 +405,27 @@ dependencies = [
"subtle", "subtle",
] ]
[[package]]
name = "dirs"
version = "5.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "44c45a9d03d6676652bcb5e724c7e988de1acad23a711b5217ab9cbecbec2225"
dependencies = [
"dirs-sys",
]
[[package]]
name = "dirs-sys"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "520f05a5cbd335fae5a99ff7a6ab8627577660ee5cfd6a94a6a929b52ff0321c"
dependencies = [
"libc",
"option-ext",
"redox_users",
"windows-sys 0.48.0",
]
[[package]] [[package]]
name = "dotenv" name = "dotenv"
version = "0.15.0" version = "0.15.0"
@ -780,11 +801,13 @@ dependencies = [
"axum", "axum",
"bus", "bus",
"clap", "clap",
"dirs",
"dotenv", "dotenv",
"serde", "serde",
"serde_json", "serde_json",
"similar-asserts", "similar-asserts",
"sqlx", "sqlx",
"tempfile",
"tokio", "tokio",
"tower-http", "tower-http",
"tracing", "tracing",
@ -871,6 +894,16 @@ version = "0.2.8"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4ec2a862134d2a7d32d7983ddcdd1c4923530833c9f2ea1a44fc5fa473989058" checksum = "4ec2a862134d2a7d32d7983ddcdd1c4923530833c9f2ea1a44fc5fa473989058"
[[package]]
name = "libredox"
version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c0ff37bd590ca25063e35af745c343cb7a0271906fb7b37e4813e8f79f00268d"
dependencies = [
"bitflags 2.4.2",
"libc",
]
[[package]] [[package]]
name = "libsqlite3-sys" name = "libsqlite3-sys"
version = "0.27.0" version = "0.27.0"
@ -1050,6 +1083,12 @@ version = "1.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dd8b5dd2ae5ed71462c540258bedcb51965123ad7e7ccf4b9a8cafaa4a63576d" checksum = "dd8b5dd2ae5ed71462c540258bedcb51965123ad7e7ccf4b9a8cafaa4a63576d"
[[package]]
name = "option-ext"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"
[[package]] [[package]]
name = "overload" name = "overload"
version = "0.1.1" version = "0.1.1"
@ -1074,7 +1113,7 @@ checksum = "93f00c865fe7cabf650081affecd3871070f26767e7b2070a3ffae14c654b447"
dependencies = [ dependencies = [
"cfg-if", "cfg-if",
"libc", "libc",
"redox_syscall 0.3.5", "redox_syscall",
"smallvec", "smallvec",
"windows-targets 0.48.0", "windows-targets 0.48.0",
] ]
@ -1229,12 +1268,14 @@ dependencies = [
] ]
[[package]] [[package]]
name = "redox_syscall" name = "redox_users"
version = "0.4.1" version = "0.4.5"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4722d768eff46b75989dd134e5c353f0d6296e5aaa3132e776cbdb56be7731aa" checksum = "bd283d9651eeda4b2a83a43c1c91b266c40fd76ecd39a50a8c630ae69dc72891"
dependencies = [ dependencies = [
"bitflags 1.3.2", "getrandom",
"libredox",
"thiserror",
] ]
[[package]] [[package]]
@ -1816,13 +1857,12 @@ checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160"
[[package]] [[package]]
name = "tempfile" name = "tempfile"
version = "3.9.0" version = "3.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "01ce4141aa927a6d1bd34a041795abd0db1cccba5d5f24b009f694bdf3a1f3fa" checksum = "85b77fafb263dd9d05cbeac119526425676db3784113aa9295c88498cbf8bff1"
dependencies = [ dependencies = [
"cfg-if", "cfg-if",
"fastrand", "fastrand",
"redox_syscall 0.4.1",
"rustix 0.38.31", "rustix 0.38.31",
"windows-sys 0.52.0", "windows-sys 0.52.0",
] ]

View File

@ -18,6 +18,8 @@ uuid = { version = "1.7.0", features = ["v4"] }
tower-http = { version = "0.5.2", features = ["cors", "trace"] } tower-http = { version = "0.5.2", features = ["cors", "trace"] }
serde_json = "1.0.116" serde_json = "1.0.116"
bus = "2.4.1" bus = "2.4.1"
dirs = "5.0.1"
[dev-dependencies] [dev-dependencies]
similar-asserts = "1.5.0" similar-asserts = "1.5.0"
tempfile = "3.10.1"

View File

@ -1,11 +1,12 @@
use std::{collections::BTreeMap, sync::RwLock}; use std::collections::BTreeMap;
use serde::Serialize; use serde::Serialize;
use crate::{ use crate::{
engine::Engine,
events::Events, events::Events,
log::{GraphItem, ItemState}, log::{GraphItem, ItemState},
shared_engine::SharedEngine,
storage::Storage,
}; };
#[derive(Serialize, PartialEq, Eq, Debug, Clone)] #[derive(Serialize, PartialEq, Eq, Debug, Clone)]
@ -31,22 +32,30 @@ pub enum Command {
}, },
} }
#[derive(Default)]
pub struct Commander { pub struct Commander {
engine: RwLock<Engine>, engine: SharedEngine,
storage: Storage,
events: Events, events: Events,
} }
impl Commander { impl Commander {
pub fn new(engine: SharedEngine, storage: Storage, events: Events) -> anyhow::Result<Self> {
Ok(Self {
engine,
storage,
events,
})
}
pub fn execute(&self, cmd: Command) -> anyhow::Result<()> { pub fn execute(&self, cmd: Command) -> anyhow::Result<()> {
tracing::debug!("executing event: {}", serde_json::to_string(&cmd)?); tracing::debug!("executing event: {}", serde_json::to_string(&cmd)?);
match cmd { match cmd.clone() {
Command::CreateRoot { root } => { Command::CreateRoot { root } => {
self.engine.write().unwrap().create_root(&root)?; self.engine.create_root(&root)?;
} }
Command::CreateSection { root, path } => { Command::CreateSection { root, path } => {
self.engine.write().unwrap().create( self.engine.create(
&root, &root,
&path.iter().map(|p| p.as_str()).collect::<Vec<_>>(), &path.iter().map(|p| p.as_str()).collect::<Vec<_>>(),
GraphItem::Section(BTreeMap::default()), GraphItem::Section(BTreeMap::default()),
@ -58,7 +67,7 @@ impl Commander {
title, title,
description, description,
state, state,
} => self.engine.write().unwrap().create( } => self.engine.create(
&root, &root,
&path.iter().map(|p| p.as_str()).collect::<Vec<_>>(), &path.iter().map(|p| p.as_str()).collect::<Vec<_>>(),
GraphItem::Item { GraphItem::Item {
@ -67,13 +76,17 @@ impl Commander {
state, state,
}, },
)?, )?,
Command::Move { root, src, dest } => self.engine.write().unwrap().section_move( Command::Move { root, src, dest } => self.engine.section_move(
&root, &root,
&src.iter().map(|p| p.as_str()).collect::<Vec<_>>(), &src.iter().map(|p| p.as_str()).collect::<Vec<_>>(),
&dest.iter().map(|p| p.as_str()).collect::<Vec<_>>(), &dest.iter().map(|p| p.as_str()).collect::<Vec<_>>(),
)?, )?,
} }
self.storage.store(&self.engine)?;
self.events.enque_command(cmd)?;
Ok(()) Ok(())
} }
} }

View File

@ -1,6 +1,6 @@
use std::{collections::BTreeMap, fmt::Display}; use std::{collections::BTreeMap, fmt::Display};
use anyhow::anyhow; use anyhow::{anyhow, Context};
use crate::log::{Graph, GraphItem}; use crate::log::{Graph, GraphItem};
@ -10,6 +10,16 @@ pub struct Engine {
} }
impl Engine { impl Engine {
pub fn engine_from_str(input: &str) -> anyhow::Result<Self> {
let graph: Graph = serde_json::from_str(input)?;
Ok(Self { graph })
}
pub fn to_str(&self) -> anyhow::Result<String> {
serde_json::to_string_pretty(&self.graph).context("failed to serialize graph")
}
pub fn create_root(&mut self, root: &str) -> anyhow::Result<()> { pub fn create_root(&mut self, root: &str) -> anyhow::Result<()> {
self.graph self.graph
.try_insert(root.to_string(), GraphItem::User(BTreeMap::default())) .try_insert(root.to_string(), GraphItem::User(BTreeMap::default()))

View File

@ -1,4 +1,4 @@
use std::sync::Mutex; use std::sync::{Arc, Mutex};
use bus::{Bus, BusReader}; use bus::{Bus, BusReader};
@ -9,8 +9,9 @@ pub enum Event {
CommandEvent { command: Command }, CommandEvent { command: Command },
} }
#[derive(Clone)]
pub struct Events { pub struct Events {
bus: Mutex<Bus<Event>>, bus: Arc<Mutex<Bus<Event>>>,
} }
impl Default for Events { impl Default for Events {
@ -18,7 +19,7 @@ impl Default for Events {
let bus = Bus::new(10); let bus = Bus::new(10);
Self { Self {
bus: Mutex::new(bus), bus: Arc::new(Mutex::new(bus)),
} }
} }
} }

View File

@ -11,9 +11,14 @@ use sqlx::{Pool, Postgres};
use tower_http::trace::TraceLayer; use tower_http::trace::TraceLayer;
pub mod commander; pub mod commander;
pub mod querier;
pub mod engine; pub mod engine;
pub mod events; pub mod events;
pub mod log; pub mod log;
pub mod shared_engine;
pub mod state;
pub mod storage;
#[derive(Parser)] #[derive(Parser)]
#[command(author, version, about, long_about = None, subcommand_required = true)] #[command(author, version, about, long_about = None, subcommand_required = true)]
@ -28,6 +33,29 @@ enum Commands {
#[arg(env = "SERVICE_HOST", long, default_value = "127.0.0.1:3000")] #[arg(env = "SERVICE_HOST", long, default_value = "127.0.0.1:3000")]
host: SocketAddr, host: SocketAddr,
}, },
CreateRoot {
#[arg(long = "root")]
root: String,
},
CreateSection {
#[arg(long = "root")]
root: String,
#[arg(long = "path")]
path: Option<String>,
},
Get {
#[arg(long = "root")]
root: String,
#[arg(long = "path")]
path: Option<String>,
},
Info {},
} }
#[tokio::main] #[tokio::main]
@ -37,7 +65,10 @@ async fn main() -> anyhow::Result<()> {
let cli = Command::parse(); let cli = Command::parse();
if let Some(Commands::Serve { host }) = cli.command { let state = state::State::new()?;
match cli.command {
Some(Commands::Serve { host }) => {
tracing::info!("Starting service"); tracing::info!("Starting service");
let state = SharedState(Arc::new(State::new().await?)); let state = SharedState(Arc::new(State::new().await?));
@ -69,6 +100,38 @@ async fn main() -> anyhow::Result<()> {
.await .await
.unwrap(); .unwrap();
} }
Some(Commands::CreateRoot { root }) => state
.commander
.execute(commander::Command::CreateRoot { root })?,
Some(Commands::CreateSection { root, path }) => {
state.commander.execute(commander::Command::CreateSection {
root,
path: path
.unwrap_or_default()
.split('.')
.map(|s| s.to_string())
.filter(|s| !s.is_empty())
.collect::<Vec<String>>(),
})?
}
Some(Commands::Get { root, path }) => {
let res = state.querier.get(
&root,
path.unwrap_or_default()
.split('.')
.filter(|s| !s.is_empty()),
);
let output = serde_json::to_string_pretty(&res)?;
println!("{}", output);
}
Some(Commands::Info {}) => {
println!("graph stored at: {}", state.storage.info()?)
}
None => {}
}
Ok(()) Ok(())
} }

View File

@ -0,0 +1,24 @@
use crate::{log::GraphItem, shared_engine::SharedEngine};
pub struct Querier {
engine: SharedEngine,
}
impl Querier {
pub fn new(engine: SharedEngine) -> Self {
Self { engine }
}
pub fn get(
&self,
root: &str,
path: impl IntoIterator<Item = impl Into<String>>,
) -> Option<GraphItem> {
let path = path.into_iter().map(|i| i.into()).collect::<Vec<String>>();
tracing::debug!("quering: {}, len: ({}))", path.join("."), path.len());
self.engine
.get(root, &path.iter().map(|i| i.as_str()).collect::<Vec<_>>())
}
}

View File

@ -0,0 +1,46 @@
use std::sync::{Arc, RwLock};
use crate::{engine::Engine, log::GraphItem};
#[derive(Clone)]
pub struct SharedEngine {
inner: Arc<RwLock<Engine>>,
}
impl From<Engine> for SharedEngine {
fn from(value: Engine) -> Self {
SharedEngine {
inner: Arc::new(RwLock::new(value)),
}
}
}
impl SharedEngine {
pub fn to_str(&self) -> anyhow::Result<String> {
self.inner.read().unwrap().to_str()
}
pub fn create_root(&self, root: &str) -> anyhow::Result<()> {
self.inner.write().unwrap().create_root(root)
}
pub fn create(&self, root: &str, path: &[&str], item: GraphItem) -> anyhow::Result<()> {
self.inner.write().unwrap().create(root, path, item)
}
pub fn get(&self, root: &str, path: &[&str]) -> Option<GraphItem> {
self.inner.read().unwrap().get(root, path).cloned()
}
pub fn section_move(
&self,
root: &str,
src_path: &[&str],
dest_path: &[&str],
) -> anyhow::Result<()> {
self.inner
.write()
.unwrap()
.section_move(root, src_path, dest_path)
}
}

View File

@ -0,0 +1,31 @@
use crate::{
commander::Commander, events::Events, querier::Querier, shared_engine::SharedEngine,
storage::Storage,
};
pub struct State {
engine: SharedEngine,
pub storage: Storage,
events: Events,
pub commander: Commander,
pub querier: Querier,
}
impl State {
pub fn new() -> anyhow::Result<Self> {
let storage = Storage::new();
let engine = storage.load()?;
let events = Events::default();
let engine = SharedEngine::from(engine);
Ok(Self {
engine: engine.clone(),
storage: storage.clone(),
events: events.clone(),
commander: Commander::new(engine.clone(), storage, events)?,
querier: Querier::new(engine),
})
}
}

View File

@ -0,0 +1,253 @@
use std::{
path::{Path, PathBuf},
sync::{Arc, Mutex},
time::Duration,
};
use crate::{engine::Engine, shared_engine::SharedEngine};
pub struct LockFile(PathBuf);
impl Drop for LockFile {
fn drop(&mut self) {
std::fs::remove_file(&self.0).expect("to be able to delete lockfile")
}
}
impl From<PathBuf> for LockFile {
fn from(value: PathBuf) -> Self {
Self(value)
}
}
#[derive(Clone)]
pub struct Storage {
base: PathBuf,
lock_file: Arc<Mutex<Option<LockFile>>>,
}
impl Default for Storage {
fn default() -> Self {
Self::new()
}
}
impl Storage {
pub fn new() -> Self {
let data_dir = dirs::data_local_dir()
.ok_or(anyhow::anyhow!("failed to retrieve the users data dir"))
.expect("to be able to find config");
Self {
base: data_dir,
lock_file: Arc::new(Mutex::new(None)),
}
}
pub fn with_base(&mut self, base: &Path) {
self.base = base.to_path_buf();
}
pub fn store(&self, engine: &SharedEngine) -> anyhow::Result<()> {
let state_path = self.state()?;
std::fs::write(state_path, engine.to_str()?)?;
Ok(())
}
pub fn load(&self) -> anyhow::Result<Engine> {
let mut lock = self.lock_file.lock().unwrap();
if lock.is_none() {
let lock_file = self.state_lock_file()?;
*lock = Some(lock_file);
}
let engine = match self.state_file()? {
Some(contents) => Engine::engine_from_str(&contents)?,
None => Engine::default(),
};
Ok(engine)
}
pub fn unload(self) -> anyhow::Result<()> {
drop(self);
Ok(())
}
fn state(&self) -> anyhow::Result<PathBuf> {
self.cache().map(|c| c.join("graph.json"))
}
fn state_file(&self) -> anyhow::Result<Option<String>> {
let state_path = self.state()?;
if !state_path.exists() {
return Ok(None);
}
let contents = std::fs::read_to_string(&state_path)?;
Ok(Some(contents))
}
fn state_lock(&self) -> anyhow::Result<PathBuf> {
self.cache().map(|c| c.join("graph.lock"))
}
fn create_lock_file(&self) -> anyhow::Result<()> {
let lock_path = self.state_lock()?;
if let Some(parent) = lock_path.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(lock_path, "hyperlog-lock")?;
Ok(())
}
fn state_lock_file(&self) -> anyhow::Result<LockFile> {
let lock_path = self.state_lock()?;
if !lock_path.exists() {
self.create_lock_file()?;
return Ok(LockFile::from(lock_path));
}
if let Ok(modified) = lock_path.metadata()?.modified() {
if modified.elapsed()? > Duration::from_secs(86400) {
std::fs::remove_file(&lock_path)?;
self.create_lock_file()?;
return Ok(LockFile::from(lock_path));
}
}
anyhow::bail!("lock file exists and is valid. Aborting");
}
fn cache(&self) -> anyhow::Result<PathBuf> {
Ok(self.base.join("hyperlog"))
}
pub fn info(&self) -> anyhow::Result<String> {
Ok(format!("storage:\n\tgraph: {}", self.state()?.display()))
}
}
#[cfg(test)]
mod test {
use std::collections::BTreeMap;
use similar_asserts::assert_eq;
use crate::log::GraphItem;
use super::*;
#[test]
fn can_create_state() -> anyhow::Result<()> {
let tempdir = tempfile::tempdir()?;
let mut storage = Storage::default();
storage.with_base(tempdir.path());
let engine = SharedEngine::from(storage.load()?);
engine.create_root("can_create_state")?;
storage.store(&engine)?;
let graph = std::fs::read_to_string(tempdir.path().join("hyperlog").join("graph.json"))?;
let lock = std::fs::read_to_string(tempdir.path().join("hyperlog").join("graph.lock"))?;
assert_eq!(
r#"{
"can_create_state": {
"type": "user"
}
}"#
.to_string(),
graph
);
assert_eq!(r#"hyperlog-lock"#.to_string(), lock);
Ok(())
}
#[test]
fn lock_already_exists() -> anyhow::Result<()> {
let tempdir = tempfile::tempdir()?;
let mut storage = Storage::default();
storage.with_base(tempdir.path());
let _engine = storage.load()?;
let mut storage_should_fail = Storage::default();
storage_should_fail.with_base(tempdir.path());
let engine_should_fail = storage_should_fail.load();
assert!(engine_should_fail.is_err());
if let Err(e) = engine_should_fail {
assert_eq!(
"lock file exists and is valid. Aborting".to_string(),
e.to_string()
);
}
Ok(())
}
#[test]
fn lock_is_cleaned_up() -> anyhow::Result<()> {
let tempdir = tempfile::tempdir()?;
let mut storage = Storage::default();
storage.with_base(tempdir.path());
let engine = SharedEngine::from(storage.load()?);
engine.create_root("can_create_state")?;
storage.store(&engine)?;
storage.unload()?;
assert!(!tempdir.path().join("hyperlog").join("graph.lock").exists());
Ok(())
}
#[test]
fn can_load_state() -> anyhow::Result<()> {
let tempdir = tempfile::tempdir()?;
let mut storage = Storage::default();
storage.with_base(tempdir.path());
let engine = SharedEngine::from(storage.load()?);
engine.create_root("can_create_state")?;
storage.store(&engine)?;
let graph = std::fs::read_to_string(tempdir.path().join("hyperlog").join("graph.json"))?;
assert_eq!(
r#"{
"can_create_state": {
"type": "user"
}
}"#
.to_string(),
graph
);
let engine = storage.load()?;
let res = engine.get("can_create_state", &[]);
assert_eq!(Some(GraphItem::User(BTreeMap::default())), res.cloned());
Ok(())
}
}