From 22dd6fdda7d0c8eabfce296bbad53587204c69fa Mon Sep 17 00:00:00 2001 From: asepharyana Date: Fri, 17 Jul 2026 04:21:18 +0700 Subject: [PATCH] feat(cms): implement RewindBlobRepository for managing binary blobs --- Cargo.lock | 1 + crates/zesdex-cms/Cargo.toml | 1 + crates/zesdex-cms/src/domain/repository.rs | 13 ++ .../src/infrastructure/persistence/mod.rs | 2 + .../persistence/rewind_blob_repo.rs | 188 ++++++++++++++++++ 5 files changed, 205 insertions(+) create mode 100644 crates/zesdex-cms/src/infrastructure/persistence/rewind_blob_repo.rs diff --git a/Cargo.lock b/Cargo.lock index b8758d2..818fac7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4727,6 +4727,7 @@ dependencies = [ "anyhow", "chrono", "dirs", + "hex", "serde", "serde_json", "tracing", diff --git a/crates/zesdex-cms/Cargo.toml b/crates/zesdex-cms/Cargo.toml index 691693a..4b289dd 100644 --- a/crates/zesdex-cms/Cargo.toml +++ b/crates/zesdex-cms/Cargo.toml @@ -11,6 +11,7 @@ anyhow.workspace = true chrono.workspace = true uuid.workspace = true tracing.workspace = true +hex.workspace = true dirs.workspace = true zesdex-entities.workspace = true zesdex-utils.workspace = true diff --git a/crates/zesdex-cms/src/domain/repository.rs b/crates/zesdex-cms/src/domain/repository.rs index d1b7dbc..18a7bd5 100644 --- a/crates/zesdex-cms/src/domain/repository.rs +++ b/crates/zesdex-cms/src/domain/repository.rs @@ -63,6 +63,19 @@ pub trait MemoryRepository { fn delete(&self, memory_dir: &Path, name: &str) -> Result<()>; } +/// Repository for rewind-snapshot binary blobs, keyed by an arbitrary +/// caller-supplied key (e.g. a tool-call id) within a session. +pub trait RewindBlobRepository { + /// Store (or overwrite) a blob under `blob_key` for this session. + fn store_blob(&self, session_dir: &Path, blob_key: &str, data: &[u8], mime_type: Option<&str>) -> anyhow::Result<()>; + + /// Retrieve a blob's bytes by key, or `None` if not found. + fn retrieve_blob(&self, session_dir: &Path, blob_key: &str) -> anyhow::Result>>; + + /// List all blob keys for this session, oldest first. + fn list_blob_keys(&self, session_dir: &Path) -> anyhow::Result>; +} + /// Persistence contract for `EditLog`. pub trait EditLogRepository { /// Open (or start tracking) the edit log for a session directory. diff --git a/crates/zesdex-cms/src/infrastructure/persistence/mod.rs b/crates/zesdex-cms/src/infrastructure/persistence/mod.rs index 34bc2ef..b121443 100644 --- a/crates/zesdex-cms/src/infrastructure/persistence/mod.rs +++ b/crates/zesdex-cms/src/infrastructure/persistence/mod.rs @@ -11,10 +11,12 @@ pub mod app_config_repo; pub mod conversation_repo; pub mod edit_log_repo; pub mod memory_repo; +pub mod rewind_blob_repo; pub mod settings_repo; pub use app_config_repo::JsonAppConfigRepository; pub use conversation_repo::JsonConversationRepository; pub use edit_log_repo::JsonlEditLogRepository; pub use memory_repo::MarkdownMemoryRepository; +pub use rewind_blob_repo::FileRewindBlobRepository; pub use settings_repo::JsonSettingsRepository; diff --git a/crates/zesdex-cms/src/infrastructure/persistence/rewind_blob_repo.rs b/crates/zesdex-cms/src/infrastructure/persistence/rewind_blob_repo.rs new file mode 100644 index 0000000..c9718d5 --- /dev/null +++ b/crates/zesdex-cms/src/infrastructure/persistence/rewind_blob_repo.rs @@ -0,0 +1,188 @@ +//! Filesystem-backed `RewindBlobRepository` implementation. +//! +//! Blob bytes are stored at `/blobs/.bin` (the key +//! is hex-encoded as the filename to sidestep any path-traversal/invalid- +//! filename-character concerns entirely, mirroring the simplicity of +//! `Memory::slugify` elsewhere in this crate but without needing a +//! human-readable filename). Key/ordering/mime-type metadata lives in an +//! append-only `/blobs/index.jsonl`, one JSON line per +//! `store_blob` call — the same JSONL-index pattern already used by +//! `EditLogRepository`. `list_blob_keys` de-duplicates by keeping each +//! key's *last* index line (so overwriting a key doesn't produce a +//! duplicate listing entry) and returns keys ordered by first-seen +//! `created_at` ascending (oldest first), matching the previous +//! `SQLite`-backed `ORDER BY created_at ASC` behavior. + +use std::io::Write; +use std::path::Path; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; + +use crate::domain::repository::RewindBlobRepository; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct BlobIndexEntry { + key: String, + mime_type: Option, + created_at: i64, +} + +/// Concrete filesystem rewind-blob repository. +#[derive(Debug, Clone, Default)] +pub struct FileRewindBlobRepository; + +impl FileRewindBlobRepository { + /// Create a new filesystem rewind-blob repository. + pub fn new() -> Self { + Self + } + + fn blobs_dir(session_dir: &Path) -> std::path::PathBuf { + session_dir.join("blobs") + } + + fn blob_file_path(session_dir: &Path, blob_key: &str) -> std::path::PathBuf { + Self::blobs_dir(session_dir).join(format!("{}.bin", hex::encode(blob_key.as_bytes()))) + } + + fn index_path(session_dir: &Path) -> std::path::PathBuf { + Self::blobs_dir(session_dir).join("index.jsonl") + } +} + +impl RewindBlobRepository for FileRewindBlobRepository { + fn store_blob( + &self, + session_dir: &Path, + blob_key: &str, + data: &[u8], + mime_type: Option<&str>, + ) -> Result<()> { + let blobs_dir = Self::blobs_dir(session_dir); + std::fs::create_dir_all(&blobs_dir) + .with_context(|| format!("failed to create blobs dir '{}'", blobs_dir.display()))?; + + let path = Self::blob_file_path(session_dir, blob_key); + let tmp = path.with_extension("bin.tmp"); + std::fs::write(&tmp, data)?; + let f = std::fs::File::open(&tmp)?; + f.sync_all()?; + std::fs::rename(&tmp, &path)?; + + let entry = BlobIndexEntry { + key: blob_key.to_string(), + mime_type: mime_type.map(String::from), + created_at: chrono::Utc::now().timestamp_millis(), + }; + let index_path = Self::index_path(session_dir); + let mut f = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&index_path) + .with_context(|| format!("failed to open blob index '{}'", index_path.display()))?; + writeln!(f, "{}", serde_json::to_string(&entry)?)?; + f.sync_all()?; + + Ok(()) + } + + fn retrieve_blob(&self, session_dir: &Path, blob_key: &str) -> Result>> { + let path = Self::blob_file_path(session_dir, blob_key); + if !path.exists() { + return Ok(None); + } + Ok(Some(std::fs::read(&path)?)) + } + + fn list_blob_keys(&self, session_dir: &Path) -> Result> { + let index_path = Self::index_path(session_dir); + let Ok(content) = std::fs::read_to_string(&index_path) else { + return Ok(Vec::new()); + }; + // Keep only the last occurrence of each key (later overwrites win), + // but remember first-seen order for the final ascending sort. + let mut first_seen_order: Vec = Vec::new(); + let mut latest_by_key: std::collections::HashMap = + std::collections::HashMap::new(); + for line in content.lines() { + let Ok(entry) = serde_json::from_str::(line) else { + continue; + }; + if !latest_by_key.contains_key(&entry.key) { + first_seen_order.push(entry.key.clone()); + } + latest_by_key.insert(entry.key.clone(), entry); + } + let mut entries: Vec = first_seen_order + .into_iter() + .filter_map(|k| latest_by_key.get(&k).cloned()) + .collect(); + entries.sort_by_key(|e| e.created_at); + Ok(entries.into_iter().map(|e| e.key).collect()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tmp_dir() -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!( + "zesdex-cms-blob-test-{}", + uuid::Uuid::new_v4() + )); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + #[test] + fn store_and_retrieve_roundtrip() { + let dir = tmp_dir(); + let repo = FileRewindBlobRepository::new(); + repo.store_blob(&dir, "tool-call-1", b"hello world", Some("text/plain")) + .unwrap(); + let bytes = repo.retrieve_blob(&dir, "tool-call-1").unwrap(); + assert_eq!(bytes, Some(b"hello world".to_vec())); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn retrieve_missing_key_returns_none() { + let dir = tmp_dir(); + let repo = FileRewindBlobRepository::new(); + assert_eq!(repo.retrieve_blob(&dir, "no-such-key").unwrap(), None); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn list_blob_keys_returns_oldest_first() { + let dir = tmp_dir(); + let repo = FileRewindBlobRepository::new(); + repo.store_blob(&dir, "first", b"a", None).unwrap(); + std::thread::sleep(std::time::Duration::from_millis(5)); + repo.store_blob(&dir, "second", b"b", None).unwrap(); + let keys = repo.list_blob_keys(&dir).unwrap(); + assert_eq!(keys, vec!["first".to_string(), "second".to_string()]); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn overwriting_a_key_keeps_only_the_latest_entry_in_the_listing() { + let dir = tmp_dir(); + let repo = FileRewindBlobRepository::new(); + repo.store_blob(&dir, "k", b"v1", None).unwrap(); + repo.store_blob(&dir, "k", b"v2", None).unwrap(); + let keys = repo.list_blob_keys(&dir).unwrap(); + assert_eq!( + keys, + vec!["k".to_string()], + "key must appear exactly once even after being overwritten" + ); + assert_eq!( + repo.retrieve_blob(&dir, "k").unwrap(), + Some(b"v2".to_vec()) + ); + let _ = std::fs::remove_dir_all(&dir); + } +}