use crate::namespace::OrbokCacheNamespace;
use localcache::{CacheEngine, ChangeDetectionMode, LocalFileCacheError};
use orbok_core::{CleanupAction, CleanupPlan, OrbokError, OrbokResult};
use orbok_db::{CACHE_FILE_NAME, Catalog};
use orbok_fs::ValidatedPath;
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::path::{Path, PathBuf};
use std::time::Duration;
#[derive(Debug, Clone, Default)]
pub struct EngineOptions {
pub ttl: Option<Duration>,
pub max_entries: Option<usize>,
}
#[derive(Debug, Clone, Default)]
pub struct CacheCleanupOutcome {
pub removed_entries: u64,
}
#[derive(Debug, Clone)]
pub struct NamespaceUsage {
pub namespace: String,
pub entries: u64,
pub payload_bytes: u64,
}
pub struct CacheService {
db_path: PathBuf,
}
impl CacheService {
pub fn new(data_dir: &Path) -> Self {
Self {
db_path: data_dir.join(CACHE_FILE_NAME),
}
}
pub fn db_path(&self) -> &Path {
&self.db_path
}
pub fn engine<T: Serialize + DeserializeOwned>(
&self,
catalog: &Catalog,
namespace: &OrbokCacheNamespace,
options: EngineOptions,
) -> OrbokResult<CacheEngine<T>> {
let mut builder = CacheEngine::<T>::builder()
.database(&self.db_path)
.namespace(namespace.as_namespace())
.payload_version(namespace.payload_version())
.change_detection(ChangeDetectionMode::MetadataThenFullHash)
.compress();
builder = match options.ttl {
Some(ttl) => builder.ttl(ttl),
None => builder.no_ttl(),
};
if let Some(n) = options.max_entries {
builder = builder.max_entries(n);
}
let engine = builder.build().map_err(cache_err)?;
self.register_engine::<T>(catalog, namespace, &options)?;
Ok(engine)
}
pub fn get_fresh<T: Serialize + DeserializeOwned>(
engine: &CacheEngine<T>,
path: &ValidatedPath,
) -> OrbokResult<Option<T>> {
Ok(engine
.get_if_fresh(&path.canonical)
.map_err(cache_err)?
.map(|entry| entry.payload))
}
pub fn put<T: Serialize + DeserializeOwned>(
engine: &CacheEngine<T>,
path: &ValidatedPath,
payload: &T,
) -> OrbokResult<()> {
engine.set(&path.canonical, payload).map_err(cache_err)
}
pub fn remove<T: Serialize + DeserializeOwned>(
engine: &CacheEngine<T>,
path: &ValidatedPath,
) -> OrbokResult<bool> {
engine.remove(&path.canonical).map_err(cache_err)
}
pub fn run_safe_cleanup(
&self,
catalog: &Catalog,
plan: &CleanupPlan,
) -> OrbokResult<CacheCleanupOutcome> {
plan.assert_safe_for_ordinary_cleanup()?;
let namespaces: Vec<OrbokCacheNamespace> = match plan.action {
CleanupAction::ClearTemporaryExtraction => vec![OrbokCacheNamespace::ExtractSegments],
CleanupAction::ClearSnippetCache => vec![OrbokCacheNamespace::PreviewCache],
CleanupAction::RemoveReplacedStaleIndexes => vec![OrbokCacheNamespace::ChunkBundle],
CleanupAction::ClearExpiredSearchCache => vec![],
_ => return Err(OrbokError::CleanupWouldTouchPersistentData),
};
let mut outcome = CacheCleanupOutcome::default();
for namespace in namespaces {
let engine = self.maintenance_engine(catalog, &namespace)?;
outcome.removed_entries += engine.cleanup_expired().map_err(cache_err)? as u64;
outcome.removed_entries += engine.cleanup_missing_files().map_err(cache_err)? as u64;
outcome.removed_entries += engine.purge_stale_versions().map_err(cache_err)? as u64;
tracing::debug!(
namespace = namespace.as_namespace(),
removed = outcome.removed_entries,
"cache cleanup pass"
);
}
Ok(outcome)
}
pub fn shrink(&self, catalog: &Catalog) -> OrbokResult<()> {
let engine = self.maintenance_engine(catalog, &OrbokCacheNamespace::PreviewCache)?;
engine.shrink_database().map_err(cache_err)
}
pub fn usage(
&self,
catalog: &Catalog,
namespaces: &[OrbokCacheNamespace],
) -> OrbokResult<Vec<NamespaceUsage>> {
let mut out = Vec::new();
for namespace in namespaces {
let engine = self.maintenance_engine(catalog, namespace)?;
let stats = engine.cache_stats().map_err(cache_err)?;
out.push(NamespaceUsage {
namespace: stats.namespace,
entries: stats.total_entries as u64,
payload_bytes: stats.total_payload_bytes,
});
}
Ok(out)
}
fn maintenance_engine(
&self,
catalog: &Catalog,
namespace: &OrbokCacheNamespace,
) -> OrbokResult<CacheEngine<serde_json::Value>> {
self.engine::<serde_json::Value>(catalog, namespace, EngineOptions::default())
}
fn register_engine<T>(
&self,
catalog: &Catalog,
namespace: &OrbokCacheNamespace,
options: &EngineOptions,
) -> OrbokResult<()> {
let data_class = match namespace.data_class() {
orbok_core::DataClass::RebuildableIndex => "rebuildable_index",
_ => "ephemeral_cache",
};
let id = format!("ce_{}", namespace.as_namespace().replace([':', '/'], "_"));
let now = orbok_core::now_iso8601();
let conn = catalog.lock();
conn.execute(
"INSERT INTO cache_engines (cache_engine_id, engine_kind, database_path, namespace, \
data_class, payload_type, payload_version, ttl_seconds, max_entries, status, \
created_at, updated_at) VALUES (?1,'localcache',?2,?3,?4,?5,?6,?7,?8,'active',?9,?9) \
ON CONFLICT(engine_kind, database_path, namespace) DO UPDATE SET \
payload_type = ?5, payload_version = ?6, ttl_seconds = ?7, max_entries = ?8, \
status = 'active', updated_at = ?9",
rusqlite::params![
id,
self.db_path.to_string_lossy(),
namespace.as_namespace(),
data_class,
std::any::type_name::<T>(),
namespace.payload_version(),
options.ttl.map(|d| d.as_secs() as i64),
options.max_entries.map(|n| n as i64),
now,
],
)
.map_err(|e| OrbokError::Database(e.to_string()))?;
Ok(())
}
}
fn cache_err(e: LocalFileCacheError) -> OrbokError {
OrbokError::Cache(e.to_string())
}