💰 refactor(core): Jobs & state cleanup + add Iranian coins API

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rust_rest_api"
-version = "0.2.1"
+version = "0.3.0"
 edition = "2024"
 license = "MIT"
 authors = ["Habibi-Dev"]
@@ -22,8 +22,10 @@ sea-orm = { version = "1.1.16", features = ["sqlx-sqlite", "runtime-tokio-rustls
 once_cell = "1.21.3"
 rust-embed = "8.7.2"
 askama = { version = "0.14", features = ["full"] }
-http = "1.3.1"
 mime_guess = "2.0.5"
+reqwest = { version = "0.12.23", features = ["__rustls", "json", "rustls-tls"] }
+uuid = { version = "1.18.1", features = ["v4"] }
+anyhow = "1.0.100"
 
 [dependencies.migration]
 path = "./migration"

src/app.rs

@@ -0,0 +1,63 @@
+use crate::core::config::Config;
+use crate::jobs::{cache_flush::CacheFlushJob, coin_sync::CoinSyncJob};
+use crate::server;
+use crate::services::{cache::StatsCache, routes::Routes};
+use crate::state::{self, APP_STATE, AppState};
+use anyhow::{Context, Result};
+use migration::{Migrator, MigratorTrait};
+use sea_orm::Database;
+use std::sync::Arc;
+
+pub async fn run() -> Result<()> {
+    let config = load_config()?;
+    let db = setup_database().await?;
+    let cache = Arc::new(StatsCache::new(db.clone()));
+
+    state::State::init(db, Arc::clone(&cache));
+
+    start_background_jobs(&cache)?;
+
+    let app_state = get_app_state()?;
+    let app = Routes::routes(app_state);
+
+    server::http::start_http(app, &config)
+        .await
+        .map_err(|e| anyhow::anyhow!("Failed to start HTTP server: {}", e))?;
+
+    Ok(())
+}
+
+fn load_config() -> Result<Config> {
+    Ok(Config::from_env())
+}
+
+async fn setup_database() -> Result<sea_orm::DatabaseConnection> {
+    let db_url =
+        std::env::var("DATABASE_URL").context("DATABASE_URL environment variable is required")?;
+
+    let db = Database::connect(&db_url)
+        .await
+        .context("Failed to connect to database")?;
+
+    Migrator::up(&db, None)
+        .await
+        .context("Database migration failed")?;
+
+    Ok(db)
+}
+
+fn start_background_jobs(cache: &Arc<StatsCache>) -> Result<()> {
+    let flush_job = CacheFlushJob::new(Arc::clone(cache));
+    let coin_job = CoinSyncJob::new();
+
+    crate::services::jobs::FlushJob::start(vec![flush_job.into_task(), coin_job.into_task()], None);
+
+    Ok(())
+}
+
+fn get_app_state() -> Result<AppState> {
+    APP_STATE
+        .get()
+        .cloned()
+        .context("Application state not initialized")
+}

src/jobs/cache_flush.rs

@@ -0,0 +1,26 @@
+use crate::jobs::JobTask;
+use crate::services::cache::StatsCache;
+use std::sync::Arc;
+
+pub struct CacheFlushJob {
+    cache: Arc<StatsCache>,
+}
+
+impl CacheFlushJob {
+    pub fn new(cache: Arc<StatsCache>) -> Self {
+        Self { cache }
+    }
+
+    pub fn into_task(self) -> JobTask {
+        let cache = self.cache;
+
+        Arc::new(move || {
+            let cache_clone = Arc::clone(&cache);
+            Box::pin(async move {
+                if let Err(e) = cache_clone.flush_to_db().await {
+                    eprintln!("Cache flush job failed: {}", e);
+                }
+            })
+        })
+    }
+}

src/jobs/coin_sync.rs

@@ -0,0 +1,36 @@
+use crate::jobs::JobTask;
+use crate::services::coin_ir::request::RequestCoinIr;
+use crate::state::APP_STATE;
+use std::sync::Arc;
+
+pub struct CoinSyncJob;
+
+impl CoinSyncJob {
+    pub fn new() -> Self {
+        Self
+    }
+
+    pub fn into_task(self) -> JobTask {
+        Arc::new(|| {
+            Box::pin(async {
+                if let Err(e) = sync_coins().await {
+                    eprintln!("Coin sync job failed: {}", e);
+                }
+            })
+        })
+    }
+}
+
+async fn sync_coins() -> Result<(), Box<dyn std::error::Error>> {
+    let service = RequestCoinIr::new();
+    let coins = service.coins_typed().await?;
+
+    let state = APP_STATE.get().ok_or("Application state not available")?;
+
+    state
+        .coin_tx
+        .send(coins)
+        .map_err(|_| "Failed to send coins to channel")?;
+
+    Ok(())
+}

src/jobs/mod.rs

@@ -0,0 +1,6 @@
+pub mod cache_flush;
+pub mod coin_sync;
+
+use std::{future::Future, pin::Pin, sync::Arc};
+
+pub type JobTask = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;

src/main.rs

@@ -1,42 +1,18 @@
+mod app;
 mod core;
 mod entities;
+mod jobs;
 mod middleware;
 mod repository;
 mod server;
 mod services;
+mod state;
 mod utility;
 
-use crate::services::{cache::StatsCache, jobs::FlushJob, routes::Routes};
-use core::config::Config;
-use migration::{Migrator, MigratorTrait};
-use sea_orm::Database;
-use std::{sync::OnceLock, time::Instant};
-
-static START: OnceLock<Instant> = OnceLock::new();
-
 #[tokio::main]
 async fn main() {
-    START.set(Instant::now()).ok();
-
-    let config = Config::from_env();
-    let db_url = std::env::var("DATABASE_URL").expect("DATABASE_URL is required");
-
-    let db = Database::connect(&db_url).await.expect("DB connect failed");
-    Migrator::up(&db, None).await.expect("Migration failed");
-
-    let cache = StatsCache::new(db.clone());
-
-    // Start background flush job
-    FlushJob::start(cache.clone());
-
-    let state = services::AppState {
-        _db: db,
-        stats_cache: cache,
-    };
-    let app = Routes::routes(state);
-
-    if let Err(e) = server::http::start_http(app, &config).await {
-        eprintln!("Fatal server error: {}", e);
+    if let Err(e) = app::run().await {
+        eprintln!("Application failed to start: {}", e);
         std::process::exit(1);
     }
 }

src/middleware/visit_event.rs

@@ -1,6 +1,7 @@
-use crate::services::AppState;
+use crate::state::AppState;
 use axum::extract::{Request, State};
 use axum::{middleware::Next, response::Response};
+
 pub async fn visit_event(
     State(_state): State<AppState>, // underscore if unused
     req: Request,

src/services/coin_ir/data.rs

@@ -0,0 +1,66 @@
+use crate::services::coin_ir::de_num_opt;
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+#[derive(Debug, Clone, Serialize)]
+pub struct CoinInfo {
+    pub current: Option<u64>,
+    pub highest: Option<u64>,
+    pub lowest: Option<u64>,
+    pub updated_at: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct CoinData {
+    pub emami: Option<CoinInfo>,
+    pub bahar: Option<CoinInfo>,
+    pub nim: Option<CoinInfo>,
+    pub rob: Option<CoinInfo>,
+    pub gerami: Option<CoinInfo>,
+    pub sync_at: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ApiResponse {
+    pub current: HashMap<String, Quote>,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct Quote {
+    #[serde(deserialize_with = "de_num_opt")]
+    pub p: Option<u64>,
+    #[serde(deserialize_with = "de_num_opt")]
+    pub h: Option<u64>,
+    #[serde(deserialize_with = "de_num_opt")]
+    pub l: Option<u64>,
+    pub ts: Option<String>,
+}
+
+impl From<&HashMap<String, Quote>> for CoinData {
+    fn from(m: &HashMap<String, Quote>) -> Self {
+        let coin_info = Self::coin_info(m);
+        Self {
+            emami: Some(coin_info("sekee")),
+            bahar: Some(coin_info("sekeb")),
+            nim: Some(coin_info("nim")),
+            rob: Some(coin_info("rob")),
+            gerami: Some(coin_info("gerami")),
+            sync_at: Some(Utc::now().to_rfc3339()),
+        }
+    }
+}
+
+impl CoinData {
+    pub fn coin_info<'a>(m: &'a HashMap<String, Quote>) -> impl Fn(&str) -> CoinInfo + 'a {
+        move |k: &str| {
+            let q = m.get(k);
+            CoinInfo {
+                current: q.and_then(|q| q.p),
+                highest: q.and_then(|q| q.h),
+                lowest: q.and_then(|q| q.l),
+                updated_at: q.and_then(|q| q.ts.clone()),
+            }
+        }
+    }
+}

src/services/coin_ir/helper.rs

@@ -0,0 +1,10 @@
+use crate::utility::clean_num;
+use serde::Deserialize;
+
+pub fn de_num_opt<'de, D>(d: D) -> Result<Option<u64>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let v = Option::<String>::deserialize(d)?;
+    Ok(v.and_then(|s| clean_num(&s)))
+}

src/services/coin_ir/mod.rs

@@ -0,0 +1,12 @@
+pub(crate) mod data;
+pub mod helper;
+pub mod request;
+mod response;
+
+use crate::services::coin_ir::response::response_coin;
+use axum::routing::{MethodRouter, get};
+pub use helper::de_num_opt;
+
+pub fn routers_list() -> Vec<(&'static str, MethodRouter)> {
+    Vec::from([("/ir", get(response_coin))])
+}

src/services/coin_ir/request.rs

@@ -0,0 +1,38 @@
+use crate::services::coin_ir::data::{ApiResponse, CoinData};
+use reqwest::Client;
+use uuid::Uuid;
+
+pub struct RequestCoinIr {
+    client: Client,
+}
+
+impl RequestCoinIr {
+    pub fn new() -> Self {
+        let client = Client::builder()
+            .use_rustls_tls()
+            .timeout(std::time::Duration::from_secs(10))
+            .build()
+            .expect("reqwest client build failed");
+        Self { client }
+    }
+
+    pub async fn start(&self) -> Result<ApiResponse, reqwest::Error> {
+        let uuid = Uuid::new_v4();
+        let url = format!("https://call1.tgju.org/ajax.json?rev={}", uuid);
+        let resp = self
+            .client
+            .get(url)
+            .header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")
+            .send()
+            .await?
+            .error_for_status()?; // convert 4xx/5xx into Err
+
+        let api: ApiResponse = resp.json().await?;
+        Ok(api)
+    }
+
+    pub async fn coins_typed(&self) -> Result<CoinData, reqwest::Error> {
+        let api = self.start().await?;
+        Ok(CoinData::from(&api.current))
+    }
+}

src/services/coin_ir/response.rs

@@ -0,0 +1,11 @@
+use crate::core::response::ApiResponse;
+use crate::state::APP_STATE;
+use axum::Json;
+use serde_json::json;
+
+pub async fn response_coin() -> Json<ApiResponse<serde_json::Value>> {
+    let state = APP_STATE.get();
+    let payload = json!(state.unwrap().coin_rx.borrow().clone());
+
+    Json(ApiResponse::success(payload))
+}

src/services/health/mod.rs

@@ -1,6 +1,6 @@
 pub(crate) mod response;
 
-use crate::services::AppState;
+use crate::state::AppState;
 use axum::Router;
 use axum::routing::get;
 use response::{fallback, health_check, init};