d155166
use cache::Cache;
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::sync::Arc;
#[derive(Clone)]
pub struct ObjectCache {
pub prefix: String,
pub cache: Arc<Cache>,
pub expire_ttl: u64,
}
/// Sanitize a cache key so it cannot collide across namespaces or inject
/// separator characters into the underlying cache storage.
pub(crate) fn sanitize_key(k: &str) -> String {
let mut out = String::with_capacity(k.len());
for b in k.bytes() {
if b.is_ascii_alphanumeric() || b == b'-' || b == b'_' || b == b'.' {
out.push(b as char);
} else {
out.push('_');
}
}
out
}
impl ObjectCache {
pub async fn get<T: DeserializeOwned + Send + Sync>(&self, k: &str) -> Option<T> {
let k = format!("{}.{}", self.prefix, sanitize_key(k));
serde_json::from_slice(self.cache.get(&k).await?.as_slice()).ok()
}
pub async fn set<V: Serialize>(&self, k: &str, v: V) -> Option<()> {
let k = format!("{}.{}", self.prefix, sanitize_key(k));
let bytes = serde_json::to_vec(&v).ok()?;
self.cache.set(&k, bytes, self.expire_ttl).await;
Some(())
}
pub async fn delete(&self, k: &str) -> Option<()> {
let k = format!("{}.{}", self.prefix, sanitize_key(k));
self.cache.delete(&k).await;
Some(())
}
pub async fn object_pool_get<T: DeserializeOwned + Send + Sync>(&self, k: &str) -> Option<T> {
let k = format!("object.pool.{}", sanitize_key(k));
serde_json::from_slice(self.cache.get(&k).await?.as_slice()).ok()
}
pub async fn object_pool_set<T: Serialize>(&self, k: &str, v: T) -> Option<()> {
let k = format!("object.pool.{}", sanitize_key(k));
let bytes = serde_json::to_vec(&v).ok()?;
self.cache.set(&k, bytes, self.expire_ttl).await;
Some(())
}
pub async fn object_pool_delete(&self, k: &str) -> Option<()> {
let k = format!("object.pool.{}", sanitize_key(k));
self.cache.delete(&k).await;
Some(())
}
// ---- Sync bridge -----------------------------------------------------
//
// The trait methods in `lib/git/*/usecase.rs` are sync (`fn`), but the
// cache backend is async. We bridge by blocking on the future. If a tokio
// runtime is already running on this thread we reuse its handle; otherwise
// we build a fresh single-thread runtime. This avoids panics when callers
// invoke these trait methods from inside an async context.
pub fn get_sync<T: DeserializeOwned + Send + Sync>(&self, k: &str) -> Option<T> {
let cache = self.cache.clone();
let prefix = self.prefix.clone();
let key = format!("{}.{}", prefix, sanitize_key(k));
block_on_safe(move || {
let cache = cache.clone();
let key = key.clone();
async move {
let bytes = cache.get(&key).await?;
serde_json::from_slice(bytes.as_slice()).ok()
}
})
}
pub fn set_sync<V: Serialize>(&self, k: &str, v: V) {
let cache = self.cache.clone();
let prefix = self.prefix.clone();
let key = format!("{}.{}", prefix, sanitize_key(k));
let bytes = match serde_json::to_vec(&v) {
Ok(b) => b,
Err(_) => return,
};
let ttl = self.expire_ttl;
block_on_safe(move || {
let cache = cache.clone();
let key = key.clone();
async move {
cache.set(&key, bytes, ttl).await;
}
});
}
pub fn delete_sync(&self, k: &str) {
let cache = self.cache.clone();
let prefix = self.prefix.clone();
let key = format!("{}.{}", prefix, sanitize_key(k));
block_on_safe(move || {
let cache = cache.clone();
let key = key.clone();
async move {
cache.delete(&key).await;
}
});
}
pub fn object_pool_get_sync<T: DeserializeOwned + Send + Sync>(&self, k: &str) -> Option<T> {
let cache = self.cache.clone();
let key = format!("object.pool.{}", sanitize_key(k));
block_on_safe(move || {
let cache = cache.clone();
let key = key.clone();
async move {
let bytes = cache.get(&key).await?;
serde_json::from_slice(bytes.as_slice()).ok()
}
})
}
pub fn object_pool_set_sync<T: Serialize>(&self, k: &str, v: T) {
let cache = self.cache.clone();
let key = format!("object.pool.{}", sanitize_key(k));
let bytes = match serde_json::to_vec(&v) {
Ok(b) => b,
Err(_) => return,
};
let ttl = self.expire_ttl;
block_on_safe(move || {
let cache = cache.clone();
let key = key.clone();
async move {
cache.set(&key, bytes, ttl).await;
}
});
}
/// Sync variant of `set` with explicit TTL. Use this for content-addressed
/// entries (TTL=0 = never expires).
pub fn set_sync_with_ttl<V: Serialize>(&self, k: &str, v: V, ttl: u64) {
let cache = self.cache.clone();
let prefix = self.prefix.clone();
let key = format!("{}.{}", prefix, sanitize_key(k));
let bytes = match serde_json::to_vec(&v) {
Ok(b) => b,
Err(_) => return,
};
block_on_safe(move || {
let cache = cache.clone();
let key = key.clone();
async move {
cache.set(&key, bytes, ttl).await;
}
});
}
}
fn block_on_safe<F, T>(f: impl FnOnce() -> F) -> T
where
F: std::future::Future<Output = T>,
{
match tokio::runtime::Handle::try_current() {
Ok(handle) => handle.block_on(f()),
Err(_) => tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("cache runtime")
.block_on(f()),
}
}