Spaces:
Paused
Paused
File size: 5,268 Bytes
1ad3b12 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | 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");
} |