Spaces:
Paused
Paused
| use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, ActiveValue::Set}; | |
| use crate::entity::{id_cache_snapshot, ratings_cache_snapshot}; | |
| use crate::id::ResolvedId; | |
| use crate::services::ratings::RatingsResult; | |
| fn now_secs() -> i64 { | |
| std::time::SystemTime::now() | |
| .duration_since(std::time::UNIX_EPOCH) | |
| .unwrap_or_default() | |
| .as_secs() as i64 | |
| } | |
| /// Persists a fresh id_cache entry. Fire-and-forget: failures are logged, | |
| /// never propagated, since this is a best-effort warm-start optimization, | |
| /// not something a request should ever fail or wait on. | |
| pub async fn save_id_cache_entry(db: &sea_orm::DatabaseConnection, key: &str, value: &ResolvedId, ttl_secs: i64) { | |
| let Ok(json) = serde_json::to_string(value) else { | |
| tracing::warn!(key, "failed to serialize id_cache entry for persistence"); | |
| return; | |
| }; | |
| let model = id_cache_snapshot::ActiveModel { | |
| key: Set(key.to_string()), | |
| value: Set(json), | |
| expires_at: Set(now_secs() + ttl_secs), | |
| }; | |
| if let Err(e) = id_cache_snapshot::Entity::insert(model) | |
| .on_conflict( | |
| sea_orm::sea_query::OnConflict::column(id_cache_snapshot::Column::Key) | |
| .update_columns([id_cache_snapshot::Column::Value, id_cache_snapshot::Column::ExpiresAt]) | |
| .to_owned(), | |
| ) | |
| .exec(db) | |
| .await | |
| { | |
| tracing::warn!(key, error = %e, "failed to persist id_cache entry"); | |
| } | |
| } | |
| /// Persists a fresh ratings_cache entry. Same fire-and-forget semantics as | |
| /// `save_id_cache_entry`. | |
| pub async fn save_ratings_cache_entry(db: &sea_orm::DatabaseConnection, key: &str, value: &RatingsResult, ttl_secs: i64) { | |
| let Ok(json) = serde_json::to_string(value) else { | |
| tracing::warn!(key, "failed to serialize ratings_cache entry for persistence"); | |
| return; | |
| }; | |
| let model = ratings_cache_snapshot::ActiveModel { | |
| key: Set(key.to_string()), | |
| value: Set(json), | |
| expires_at: Set(now_secs() + ttl_secs), | |
| }; | |
| if let Err(e) = ratings_cache_snapshot::Entity::insert(model) | |
| .on_conflict( | |
| sea_orm::sea_query::OnConflict::column(ratings_cache_snapshot::Column::Key) | |
| .update_columns([ratings_cache_snapshot::Column::Value, ratings_cache_snapshot::Column::ExpiresAt]) | |
| .to_owned(), | |
| ) | |
| .exec(db) | |
| .await | |
| { | |
| tracing::warn!(key, error = %e, "failed to persist ratings_cache entry"); | |
| } | |
| } | |
| /// On startup: reloads non-expired id_cache_snapshot rows into the live | |
| /// moka cache. Expired rows are deleted rather than reloaded. | |
| pub async fn load_id_cache_snapshot(db: &sea_orm::DatabaseConnection, cache: &moka::future::Cache<String, ResolvedId>) { | |
| let now = now_secs(); | |
| let rows = match id_cache_snapshot::Entity::find().all(db).await { | |
| Ok(rows) => rows, | |
| Err(e) => { | |
| tracing::warn!(error = %e, "failed to load id_cache snapshot"); | |
| return; | |
| } | |
| }; | |
| let mut loaded = 0u64; | |
| let mut stale_keys = Vec::new(); | |
| for row in rows { | |
| if row.expires_at <= now { | |
| stale_keys.push(row.key); | |
| continue; | |
| } | |
| match serde_json::from_str::<ResolvedId>(&row.value) { | |
| Ok(value) => { | |
| cache.insert(row.key, value).await; | |
| loaded += 1; | |
| } | |
| Err(e) => { | |
| tracing::warn!(key = %row.key, error = %e, "failed to deserialize id_cache snapshot entry, dropping"); | |
| stale_keys.push(row.key); | |
| } | |
| } | |
| } | |
| if !stale_keys.is_empty() { | |
| let _ = id_cache_snapshot::Entity::delete_many() | |
| .filter(id_cache_snapshot::Column::Key.is_in(stale_keys)) | |
| .exec(db) | |
| .await; | |
| } | |
| tracing::info!(loaded, "id_cache snapshot reloaded"); | |
| } | |
| /// On startup: reloads non-expired ratings_cache_snapshot rows into the | |
| /// live moka cache. Expired rows are deleted rather than reloaded. | |
| pub async fn load_ratings_cache_snapshot(db: &sea_orm::DatabaseConnection, cache: &moka::future::Cache<String, RatingsResult>) { | |
| let now = now_secs(); | |
| let rows = match ratings_cache_snapshot::Entity::find().all(db).await { | |
| Ok(rows) => rows, | |
| Err(e) => { | |
| tracing::warn!(error = %e, "failed to load ratings_cache snapshot"); | |
| return; | |
| } | |
| }; | |
| let mut loaded = 0u64; | |
| let mut stale_keys = Vec::new(); | |
| for row in rows { | |
| if row.expires_at <= now { | |
| stale_keys.push(row.key); | |
| continue; | |
| } | |
| match serde_json::from_str::<RatingsResult>(&row.value) { | |
| Ok(value) => { | |
| cache.insert(row.key, value).await; | |
| loaded += 1; | |
| } | |
| Err(e) => { | |
| tracing::warn!(key = %row.key, error = %e, "failed to deserialize ratings_cache snapshot entry, dropping"); | |
| stale_keys.push(row.key); | |
| } | |
| } | |
| } | |
| if !stale_keys.is_empty() { | |
| let _ = ratings_cache_snapshot::Entity::delete_many() | |
| .filter(ratings_cache_snapshot::Column::Key.is_in(stale_keys)) | |
| .exec(db) | |
| .await; | |
| } | |
| tracing::info!(loaded, "ratings_cache snapshot reloaded"); | |
| } |