Blame for api/src/main.rs

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};

#[tokio::main]
async fn main() {
    // Load .env file if present (ignored if missing)
    dotenvy::dotenv().ok();

    #[cfg(feature = "test-support")]
    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");
        };

        #[cfg(unix)]
        let terminate = async {
            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
                .expect("failed to install SIGTERM handler")
                .recv()
                .await;
        };

        #[cfg(not(unix))]
        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");
}