refactor: migrate infra to mytheclipse crates (retry, cache, event, ratelimit, config)
Deploy Scraper / build-and-deploy (push) Canceled after 0s
Deploy Scraper / build-and-deploy (push) Canceled after 0s
Replace hand-rolled infrastructure with the custom mytheclipse library: - retry (src/infrastructure/scraping/retry.rs): backoff crate -> mytheclipse RetryConfig + retry() with retry_all predicate; same call-site helpers. - cache (src/infrastructure/cache): deadpool raw AsyncCommands -> mytheclipse_cache::RedisCache bridge (src/infrastructure/cache/mytheclipse.rs), typed JSON wrapper keeps the Cache<'a> API used by use cases. - events (src/events/bus.rs): custom broadcast pub/sub -> mytheclipse_event InMemoryEventBus + TypedEventBus alias. - ratelimit (src/presentation/middleware/ratelimit.rs): hand-rolled window -> mytheclipse::RateLimiter token bucket. - config (src/config/mod.rs): config crate -> mytheclipse_config ConfigLoader, preserving env-only fallback (config files optional). No route/API/Redis-key changes. mytheclipse-cache path-dep points at local source (redis 0.32 aligned with deadpool). Removed backoff/config deps.
This commit is contained in:
@@ -2,12 +2,11 @@
|
||||
// Updated for sync Redis API, reqwest API changes, and concurrency optimization.
|
||||
|
||||
use dashmap::DashMap;
|
||||
use redis::AsyncCommands;
|
||||
use std::sync::LazyLock;
|
||||
use tokio::sync::broadcast;
|
||||
use tracing::{debug, error, warn};
|
||||
|
||||
use crate::infrastructure::cache::redis_pool::get_redis_conn;
|
||||
use crate::infrastructure::cache::mytheclipse;
|
||||
use crate::infrastructure::utils::cache_ttl::CACHE_TTL_VERY_SHORT;
|
||||
use crate::infrastructure::utils::http::common_headers;
|
||||
use crate::infrastructure::utils::http::is_internet_baik_block_page;
|
||||
@@ -52,13 +51,13 @@ fn get_fetch_cache_key(slug: &str) -> String {
|
||||
}
|
||||
|
||||
async fn get_cached_fetch(slug: &str) -> Result<Option<FetchResult>, AppError> {
|
||||
let mut conn = get_redis_conn().await?;
|
||||
let key = get_fetch_cache_key(slug);
|
||||
let bytes = mytheclipse::get(&key)
|
||||
.await
|
||||
.map_err(|e| AppError::Internal(format!("Cache get failed for {}: {}", slug, e)))?;
|
||||
|
||||
let cached: Option<String> = conn.get(&key).await?;
|
||||
|
||||
if let Some(cached_str) = cached {
|
||||
match serde_json::from_str::<FetchResult>(&cached_str) {
|
||||
if let Some(bytes) = bytes {
|
||||
match serde_json::from_slice::<FetchResult>(&bytes) {
|
||||
Ok(parsed) => {
|
||||
debug!("[fetchWithProxy] Returning cached response for {}", slug);
|
||||
Ok(Some(parsed))
|
||||
@@ -71,14 +70,17 @@ async fn get_cached_fetch(slug: &str) -> Result<Option<FetchResult>, AppError> {
|
||||
}
|
||||
|
||||
async fn set_cached_fetch(slug: &str, value: &FetchResult) -> Result<(), AppError> {
|
||||
let mut conn = get_redis_conn().await?;
|
||||
let key = get_fetch_cache_key(slug);
|
||||
let json_string = serde_json::to_string(value)?;
|
||||
let json = serde_json::to_vec(value)?;
|
||||
|
||||
// Use standardized TTL
|
||||
conn.set_ex::<_, _, ()>(&key, &json_string, CACHE_TTL_VERY_SHORT)
|
||||
.await?;
|
||||
Ok(())
|
||||
mytheclipse::set(
|
||||
&key,
|
||||
json,
|
||||
Some(std::time::Duration::from_secs(CACHE_TTL_VERY_SHORT)),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| AppError::Internal(format!("Cache set failed for {}: {}", slug, e)))
|
||||
}
|
||||
// --- REDIS CACHE WRAPPER END ---
|
||||
|
||||
|
||||
Reference in New Issue
Block a user