use std::{ path::Path, sync::Arc, time::{Duration, SystemTime, UNIX_EPOCH}, }; use litellm_cache::{ BaseCache, BatchCache, BatchEntry, CacheCodec, CounterCache, DeleteCache, DisconnectCache, Error, ExactCacheContext, FlushCache, }; use crate::{DiskStore, DiskcacheSqliteStore, PythonDiskCacheAdapter, StoredValue, ValueAdapter}; pub struct DiskCache { store: Arc, adapter: Arc, codec: S, } impl DiskCache { pub fn open(directory: impl AsRef, codec: S) -> Result { Ok(Self { store: Arc::new(DiskcacheSqliteStore::open(directory)?), adapter: Arc::new(PythonDiskCacheAdapter), codec, }) } } impl DiskCache { pub fn with_store(store: D, codec: S) -> Self { Self { store: Arc::new(store), adapter: Arc::new(PythonDiskCacheAdapter), codec, } } } impl DiskCache { pub fn with_adapter(store: D, adapter: A, codec: S) -> Self { Self { store: Arc::new(store), adapter: Arc::new(adapter), codec, } } pub fn directory(&self) -> &Path { self.store.directory() } fn decode_stored(&self, value: StoredValue) -> Result, Error> { let Some(bytes) = self.adapter.read(value)? else { return Ok(None); }; self.codec.decode(&bytes).map(Some) } async fn run_blocking(store: Arc, operation: F) -> Result where T: Send + 'static, F: FnOnce(&D) -> Result + Send + 'static, { tokio::task::spawn_blocking(move || operation(&store)) .await .map_err(|_| Error::Unavailable)? } } impl BaseCache for DiskCache { type Value = S::Value; type Context = ExactCacheContext; fn get_ttl(&self, context: &Self::Context) -> Option { context.ttl } fn set_cache( &self, key: &str, value: Self::Value, context: &Self::Context, ) -> Result<(), Error> { let value = self.adapter.write(self.codec.encode(&value)?); let expire_time = context.ttl.map(|ttl| unix_now() + ttl.as_secs_f64()); self.store.set(key, value, expire_time, unix_now()) } fn get_cache(&self, key: &str, _: &Self::Context) -> Result, Error> { self.store .get(key, unix_now())? .map(|value| self.decode_stored(value)) .transpose() .map(|value| value.flatten()) } async fn async_set_cache( &self, key: &str, value: Self::Value, context: ExactCacheContext, ) -> Result<(), Error> { let value = self.adapter.write(self.codec.encode(&value)?); let ttl = context.ttl; let key = key.to_string(); Self::run_blocking(Arc::clone(&self.store), move |store| { let expire_time = ttl.map(|ttl| unix_now() + ttl.as_secs_f64()); store.set(&key, value, expire_time, unix_now()) }) .await } async fn async_get_cache( &self, key: &str, _: &ExactCacheContext, ) -> Result, Error> { let key = key.to_string(); let value = Self::run_blocking(Arc::clone(&self.store), move |store| { store.get(&key, unix_now()) }) .await?; value .map(|value| self.decode_stored(value)) .transpose() .map(|value| value.flatten()) } async fn async_set_cache_pipeline( &self, entries: Vec<(String, Self::Value)>, context: ExactCacheContext, ) -> Result<(), Error> { let entries = entries .into_iter() .map(|(key, value)| { self.codec .encode(&value) .map(|value| (key, self.adapter.write(value))) }) .collect::, _>>()?; let expire_after = context.ttl; Self::run_blocking(Arc::clone(&self.store), move |store| { for (key, value) in entries { let expire_time = expire_after.map(|ttl| unix_now() + ttl.as_secs_f64()); store.set(&key, value, expire_time, unix_now())?; } Ok(()) }) .await } } impl BatchCache for DiskCache { fn batch_get_cache( &self, keys: &[String], context: &ExactCacheContext, ) -> Result>, Error> { keys.iter() .map(|key| match self.get_cache(key, context) { Ok(Some(value)) => Ok(BatchEntry::Hit(value)), Ok(None) => Ok(BatchEntry::Miss), Err(Error::InvalidEntry) => Ok(BatchEntry::Invalid), Err(error) => Err(error), }) .collect() } async fn async_batch_get_cache( &self, keys: Vec, _: ExactCacheContext, ) -> Result>, Error> { let values = Self::run_blocking(Arc::clone(&self.store), move |store| { keys.into_iter() .map(|key| store.get(&key, unix_now()).map(|value| (key, value))) .collect::, _>>() }) .await?; values .into_iter() .map(|(_, value)| match value { None => Ok(BatchEntry::Miss), Some(value) => match self.decode_stored(value) { Ok(Some(value)) => Ok(BatchEntry::Hit(value)), Ok(None) => Ok(BatchEntry::Miss), Err(Error::InvalidEntry) => Ok(BatchEntry::Invalid), Err(error) => Err(error), }, }) .collect() } } impl DeleteCache for DiskCache { fn delete_cache(&self, key: &str) -> Result<(), Error> { self.store.pop(key, unix_now()).map(|_| ()) } async fn async_delete_cache(&self, key: &str) -> Result<(), Error> { let key = key.to_string(); Self::run_blocking(Arc::clone(&self.store), move |store| { store.pop(&key, unix_now()).map(|_| ()) }) .await } } impl FlushCache for DiskCache { fn flush_cache(&self) -> Result<(), Error> { self.store.clear() } async fn async_flush_cache(&self) -> Result<(), Error> { Self::run_blocking(Arc::clone(&self.store), |store| store.clear()).await } } impl DisconnectCache for DiskCache { async fn disconnect(&self) -> Result<(), Error> { Ok(()) } } impl CounterCache for DiskCache { fn increment_cache( &self, key: &str, amount: f64, context: ExactCacheContext, ) -> Result { increment( self.adapter.as_ref(), self.store.as_ref(), key, amount, context.ttl, ) } async fn async_increment( &self, key: &str, amount: f64, context: ExactCacheContext, _refresh_ttl: bool, ) -> Result { let key = key.to_string(); let adapter = Arc::clone(&self.adapter); Self::run_blocking(Arc::clone(&self.store), move |store| { increment(adapter.as_ref(), store, &key, amount, context.ttl) }) .await } } fn increment( adapter: &A, store: &D, key: &str, amount: f64, ttl: Option, ) -> Result { let mut result = None; let mut apply = |current: Option| { let initial = adapter.counter_seed(current)?; let value = initial + amount; let stored = adapter.counter_value(value); result = Some(value); Ok((stored, ttl.map(|ttl| unix_now() + ttl.as_secs_f64()))) }; store.update(key, unix_now(), &mut apply)?; result.ok_or(Error::InvalidEntry) } fn unix_now() -> f64 { SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_secs_f64() }