|
|
|
@@ -0,0 +1,188 @@
|
|
|
|
|
//! Filesystem-backed `RewindBlobRepository` implementation.
|
|
|
|
|
//!
|
|
|
|
|
//! Blob bytes are stored at `<session_dir>/blobs/<hex(key)>.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 `<session_dir>/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<String>,
|
|
|
|
|
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<Option<Vec<u8>>> {
|
|
|
|
|
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<Vec<String>> {
|
|
|
|
|
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<String> = Vec::new();
|
|
|
|
|
let mut latest_by_key: std::collections::HashMap<String, BlobIndexEntry> =
|
|
|
|
|
std::collections::HashMap::new();
|
|
|
|
|
for line in content.lines() {
|
|
|
|
|
let Ok(entry) = serde_json::from_str::<BlobIndexEntry>(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<BlobIndexEntry> = 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);
|
|
|
|
|
}
|
|
|
|
|
}
|
asepharyana