Spaces:
Paused
Paused
| use std::net::SocketAddr; | |
| use std::sync::Arc; | |
| use std::time::Duration; | |
| use ab_glyph::FontArc; | |
| use dashmap::DashMap; | |
| use sea_orm::{ConnectionTrait, DatabaseConnection}; | |
| use tracing_subscriber::EnvFilter; | |
| use openposterdb_api::cache::MemCacheEntry; | |
| use openposterdb_api::config::Config; | |
| use openposterdb_api::handlers; | |
| use openposterdb_api::services::db; | |
| use openposterdb_api::services::fanart::FanartClient; | |
| use openposterdb_api::services::mdblist::MdblistClient; | |
| use openposterdb_api::services::omdb::OmdbClient; | |
| use openposterdb_api::services::tmdb::TmdbClient; | |
| use openposterdb_api::services::trakt::TraktClient; | |
| use openposterdb_api::{build_app, upgrade, AppState, FONT_BYTES, MIGRATIONS, SCHEMA_SQL}; | |
| async fn main() { | |
| // Load .env file if present (ignored if missing) | |
| dotenvy::dotenv().ok(); | |
| eprintln!("WARNING: test-support feature is enabled — rate limiting is disabled. Do not use in production."); | |
| tracing_subscriber::fmt() | |
| .with_env_filter( | |
| EnvFilter::try_from_default_env() | |
| .unwrap_or_else(|_| "info,sea_orm=warn,sea_orm_migration=warn,sqlx=warn".into()), | |
| ) | |
| .init(); | |
| let config = Config::from_env(); | |
| let http = reqwest::Client::builder() | |
| .user_agent(concat!( | |
| "openposterdb/", | |
| env!("CARGO_PKG_VERSION"), | |
| " (+https://github.com/PNRxA/openposterdb)" | |
| )) | |
| .timeout(Duration::from_secs(30)) | |
| .connect_timeout(Duration::from_secs(10)) | |
| .build() | |
| .expect("failed to build HTTP client"); | |
| let font = FontArc::try_from_slice(FONT_BYTES).expect("failed to load font"); | |
| let omdb = config | |
| .omdb_api_key | |
| .as_ref() | |
| .map(|key| OmdbClient::new(key.clone(), http.clone())); | |
| let mdblist = config | |
| .mdblist_api_key | |
| .as_ref() | |
| .map(|key| MdblistClient::new(key.clone(), http.clone())); | |
| let fanart = config | |
| .fanart_api_key | |
| .as_ref() | |
| .map(|key| FanartClient::new(key.clone(), http.clone())); | |
| let trakt = config | |
| .trakt_client_id | |
| .as_ref() | |
| .map(|key| TraktClient::new(key.clone(), http.clone())); | |
| // Load JWT secret | |
| let jwt_secret = db::load_secret_from_env("JWT_SECRET"); | |
| let secure_cookies = std::env::var("COOKIE_SECURE") | |
| .map(|v| v != "false" && v != "0") | |
| .unwrap_or(true); | |
| // Log startup configuration | |
| tracing::info!( | |
| mdblist = config.mdblist_api_key.is_some(), | |
| omdb = config.omdb_api_key.is_some(), | |
| fanart = config.fanart_api_key.is_some(), | |
| trakt = config.trakt_client_id.is_some(), | |
| "rating providers configured" | |
| ); | |
| tracing::info!( | |
| cache_dir = %config.cache_dir, | |
| db_dir = %config.db_dir, | |
| image_quality = config.image_quality, | |
| mem_cache_mb = config.image_mem_cache_mb, | |
| secure_cookies, | |
| cdn_redirects = config.enable_cdn_redirects, | |
| external_cache_only = config.external_cache_only, | |
| free_key_enabled = ?config.free_key_enabled, | |
| "server configuration" | |
| ); | |
| if config.external_cache_only && !config.enable_cdn_redirects { | |
| tracing::warn!( | |
| "EXTERNAL_CACHE_ONLY is enabled without ENABLE_CDN_REDIRECTS — \ | |
| every request after the in-memory cache expires will regenerate the image. \ | |
| Consider enabling CDN redirects so a CDN can absorb repeat traffic." | |
| ); | |
| } | |
| // Ensure cache directory exists (DB no longer needs a local directory — it's remote Postgres) | |
| if !config.external_cache_only { | |
| tokio::fs::create_dir_all(&config.cache_dir) | |
| .await | |
| .expect("failed to create cache dir"); | |
| } | |
| let database_url = std::env::var("DATABASE_URL") | |
| .expect("DATABASE_URL must be set for Postgres connection"); | |
| let pg_opts = database_url | |
| .parse::<sqlx::postgres::PgConnectOptions>() | |
| .expect("invalid DATABASE_URL"); | |
| tracing::info!( | |
| host = pg_opts.get_host(), | |
| port = pg_opts.get_port(), | |
| "connecting to database" | |
| ); | |
| let pool = sqlx::postgres::PgPoolOptions::new() | |
| .max_connections(20) | |
| .min_connections(4) | |
| .acquire_timeout(Duration::from_secs(5)) | |
| .idle_timeout(Duration::from_secs(300)) | |
| .connect_with(pg_opts) | |
| .await | |
| .expect("failed to connect to database"); | |
| let database: DatabaseConnection = | |
| sea_orm::SqlxPostgresConnector::from_sqlx_postgres_pool(pool); | |
| for sql in SCHEMA_SQL { | |
| database | |
| .execute_unprepared(sql) | |
| .await | |
| .expect("failed to create table"); | |
| } | |
| for (sql, expected_err) in MIGRATIONS { | |
| match database.execute_unprepared(sql).await { | |
| Ok(_) => {} | |
| Err(e) if e.to_string().to_lowercase().contains(expected_err) => { | |
| tracing::debug!("Migration already applied: {e}"); | |
| } | |
| Err(e) => panic!("migration failed: {e}\n SQL: {sql}"), | |
| } | |
| } | |
| // Run data upgrades (cache key migrations, etc.) | |
| upgrade::run(&database, &config.cache_dir, config.external_cache_only) | |
| .await | |
| .expect("data upgrade failed"); | |
| // Clean up expired refresh tokens | |
| match db::delete_expired_refresh_tokens(&database).await { | |
| Ok(count) if count > 0 => tracing::info!("Cleaned up {count} expired refresh tokens"), | |
| Ok(_) => {} | |
| Err(e) => tracing::warn!("Failed to clean up expired refresh tokens: {e}"), | |
| } | |
| // Seed admin user from env if no admins exist | |
| if let (Ok(username), Ok(password)) = ( | |
| std::env::var("ADMIN_USERNAME"), | |
| std::env::var("ADMIN_PASSWORD"), | |
| ) { | |
| if !username.is_empty() && !password.is_empty() { | |
| match db::count_admin_users(&database).await { | |
| Ok(0) => { | |
| let hash = handlers::auth::hash_password(&password) | |
| .expect("failed to hash admin password"); | |
| match db::create_admin_user(&database, &username, &hash).await { | |
| Ok(_) => tracing::info!("Seeded admin user '{username}' from environment"), | |
| Err(e) => tracing::error!("Failed to seed admin user: {e}"), | |
| } | |
| } | |
| Ok(_) => { | |
| tracing::debug!("Admin user already exists, skipping seed"); | |
| } | |
| Err(e) => tracing::error!("Failed to check admin users: {e}"), | |
| } | |
| } | |
| } | |
| // Initialize caches | |
| let api_key_cache = moka::future::Cache::builder() | |
| .max_capacity(10_000) | |
| .time_to_live(Duration::from_secs(300)) | |
| .build(); | |
| let image_inflight = moka::future::Cache::builder() | |
| .max_capacity(1_000) | |
| .time_to_live(Duration::from_secs(30)) | |
| .build(); | |
| let id_cache = moka::future::Cache::builder() | |
| .max_capacity(50_000) | |
| .time_to_live(Duration::from_secs(3600)) | |
| .build(); | |
| let ratings_cache = moka::future::Cache::builder() | |
| .max_capacity(50_000) | |
| .time_to_live(Duration::from_secs(1800)) | |
| .build(); | |
| // Reload persisted cache snapshots from a previous run (non-expired entries only). | |
| openposterdb_api::cache::persistence::load_id_cache_snapshot(&database, &id_cache).await; | |
| openposterdb_api::cache::persistence::load_ratings_cache_snapshot(&database, &ratings_cache).await; | |
| let image_mem_cache = moka::future::Cache::builder() | |
| .weigher(|_key: &String, value: &MemCacheEntry| -> u32 { | |
| u32::try_from(value.bytes.len()).unwrap_or(u32::MAX).saturating_add(64) | |
| }) | |
| .max_capacity(config.image_mem_cache_mb * 1024 * 1024) | |
| .time_to_live(Duration::from_secs(3600)) | |
| .time_to_idle(Duration::from_secs(1800)) | |
| .build(); | |
| let refresh_locks = moka::sync::Cache::builder() | |
| .max_capacity(10_000) | |
| .time_to_live(Duration::from_secs(300)) | |
| .build(); | |
| let fanart_cache = moka::future::Cache::builder() | |
| .max_capacity(10_000) | |
| .time_to_live(Duration::from_secs(3600)) | |
| .build(); | |
| let fanart_negative = moka::future::Cache::builder() | |
| .max_capacity(10_000) | |
| .time_to_live(Duration::from_secs(3600)) | |
| .build(); | |
| let tmdb_images_cache = moka::future::Cache::builder() | |
| .max_capacity(10_000) | |
| .time_to_live(Duration::from_secs(1800)) | |
| .build(); | |
| let settings_cache = moka::future::Cache::builder() | |
| .max_capacity(10_000) | |
| .time_to_live(Duration::from_secs(300)) | |
| .build(); | |
| let global_settings_cache = moka::future::Cache::builder() | |
| .max_capacity(1) | |
| .time_to_live(Duration::from_secs(300)) | |
| .build(); | |
| let preview_cache = moka::future::Cache::builder() | |
| .max_capacity(500) | |
| .time_to_live(Duration::from_secs(3600)) | |
| .build(); | |
| let free_api_key_cache = moka::future::Cache::builder() | |
| .max_capacity(1) | |
| .time_to_live(Duration::from_secs(60)) | |
| .build(); | |
| let settings_hash_registry = moka::future::Cache::builder() | |
| .max_capacity(10_000) | |
| .time_to_live(Duration::from_secs(300)) | |
| .build(); | |
| let available_ratings_cache = moka::future::Cache::builder() | |
| .max_capacity(50_000) | |
| .time_to_live(Duration::from_secs(3600)) | |
| .build(); | |
| let available_cpus = std::thread::available_parallelism() | |
| .map(|n| n.get()) | |
| .unwrap_or(2); | |
| tracing::info!(cpus = available_cpus, "detected available CPUs"); | |
| let render_concurrency: usize = std::env::var("RENDER_CONCURRENCY") | |
| .ok() | |
| .and_then(|v| v.parse().ok()) | |
| .unwrap_or(available_cpus * 2); | |
| let render_semaphore = Arc::new(tokio::sync::Semaphore::new(render_concurrency)); | |
| tracing::info!(permits = render_concurrency, "render semaphore initialized"); | |
| let cross_id_concurrency: usize = std::env::var("CROSS_ID_CONCURRENCY") | |
| .ok() | |
| .and_then(|v| v.parse().ok()) | |
| .unwrap_or(available_cpus); | |
| let cross_id_semaphore = Arc::new(tokio::sync::Semaphore::new(cross_id_concurrency)); | |
| tracing::info!(permits = cross_id_concurrency, "cross-id semaphore initialized"); | |
| let pending_last_used: Arc<DashMap<i32, ()>> = Arc::new(DashMap::new()); | |
| let state = Arc::new(AppState { | |
| http: http.clone(), | |
| tmdb: TmdbClient::new(config.tmdb_api_key.clone(), http), | |
| omdb, | |
| mdblist, | |
| font, | |
| refresh_locks, | |
| db: database, | |
| jwt_secret, | |
| secure_cookies, | |
| api_key_cache, | |
| image_inflight, | |
| id_cache, | |
| ratings_cache, | |
| image_mem_cache, | |
| pending_last_used: pending_last_used.clone(), | |
| fanart, | |
| trakt, | |
| fanart_cache, | |
| fanart_negative, | |
| tmdb_images_cache, | |
| settings_cache, | |
| global_settings_cache, | |
| preview_cache, | |
| free_api_key_cache, | |
| render_semaphore, | |
| cross_id_semaphore, | |
| settings_hash_registry, | |
| available_ratings_cache, | |
| config: config.clone(), | |
| }); | |
| // Spawn background task for the 7-day image_meta/available_ratings cleanup job. | |
| // Checks hourly whether a run is due (persisted timestamp), so a run missed | |
| // while the container was asleep past the 7-day mark catches up promptly on wake. | |
| { | |
| let state = state.clone(); | |
| tokio::spawn(async move { | |
| let mut interval = tokio::time::interval(Duration::from_secs(3600)); | |
| loop { | |
| interval.tick().await; | |
| openposterdb_api::cleanup::run_if_due(&state).await; | |
| } | |
| }); | |
| } | |
| // Spawn background flush task for batched last_used_at updates | |
| { | |
| let db = state.db.clone(); | |
| let pending = pending_last_used.clone(); | |
| tokio::spawn(async move { | |
| let mut interval = tokio::time::interval(Duration::from_secs(60)); | |
| interval.tick().await; | |
| loop { | |
| interval.tick().await; | |
| let ids: Vec<i32> = pending.iter().map(|r| *r.key()).collect(); | |
| pending.clear(); | |
| if !ids.is_empty() | |
| && let Err(e) = db::batch_update_last_used(&db, &ids).await | |
| { | |
| tracing::warn!(error = %e, "failed to batch update last_used_at"); | |
| } | |
| } | |
| }); | |
| } | |
| let app = build_app(state.clone()); | |
| let listener = tokio::net::TcpListener::bind(&config.listen_addr) | |
| .await | |
| .expect("failed to bind"); | |
| tracing::info!(addr = %config.listen_addr, "server listening"); | |
| let shutdown_pending = pending_last_used; | |
| let shutdown_db = state.db.clone(); | |
| let shutdown_signal = async move { | |
| let ctrl_c = async { | |
| tokio::signal::ctrl_c() | |
| .await | |
| .expect("failed to install Ctrl+C handler"); | |
| }; | |
| let terminate = async { | |
| tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) | |
| .expect("failed to install SIGTERM handler") | |
| .recv() | |
| .await; | |
| }; | |
| let terminate = std::future::pending::<()>(); | |
| tokio::select! { | |
| _ = ctrl_c => {}, | |
| _ = terminate => {}, | |
| } | |
| tracing::info!("shutdown signal received, flushing pending last_used updates"); | |
| let ids: Vec<i32> = shutdown_pending.iter().map(|r| *r.key()).collect(); | |
| shutdown_pending.clear(); | |
| if !ids.is_empty() | |
| && let Err(e) = db::batch_update_last_used(&shutdown_db, &ids).await | |
| { | |
| tracing::warn!(error = %e, "failed to flush last_used_at on shutdown"); | |
| } | |
| }; | |
| axum::serve( | |
| listener, | |
| app.into_make_service_with_connect_info::<SocketAddr>(), | |
| ) | |
| .with_graceful_shutdown(shutdown_signal) | |
| .await | |
| .expect("server error"); | |
| } |