use multithreaded::{AppState, config::Config, csrf};
use sqlx::postgres::PgPoolOptions;
use tokio::net::TcpListener;
use tower_http::services::ServeDir;
use tower_sessions::ExpiredDeletion;
use tower_sessions::SessionManagerLayer;
use tower_sessions::cookie::SameSite;
use tower_sessions_sqlx_store::PostgresStore;
use tracing_subscriber::EnvFilter;

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .init();

    dotenvy::dotenv().ok();

    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");

    // Explicit pool sizing for the handler pool. Some handlers fan out ~3
    // queries concurrently via `try_join!`, so 32 connections sustain ~10
    // concurrent such requests before acquisition queues. Session I/O and its
    // expiry sweep run on a *separate* pool (below), so they no longer contend
    // here (M-Pf2). Bound acquisition so a burst sheds fast rather than hanging.
    let pool = PgPoolOptions::new()
        .max_connections(32)
        .acquire_timeout(std::time::Duration::from_secs(10))
        .connect(&database_url)
        .await
        .expect("failed to connect to database");

    sqlx::migrate!()
        .run(&pool)
        .await
        .expect("failed to run migrations");

    tracing::info!("migrations applied");

    // Seed initial data if --seed flag is passed, then exit
    if std::env::args().any(|a| a == "--seed") {
        multithreaded::seed::run(&pool).await;
        tracing::info!("seed data inserted");
        return;
    }

    let config = Config::from_env();
    multithreaded::error_page::init(config.mnw_base_url.clone());

    // Optional S3 storage for image uploads
    let s3 = if let Some(ref s3_config) = config.s3 {
        match multithreaded::storage::S3Storage::new(s3_config).await {
            Ok(client) => {
                tracing::info!("S3 storage configured (bucket: {})", s3_config.bucket);
                Some(std::sync::Arc::new(client))
            }
            Err(e) => {
                tracing::warn!("S3 storage unavailable: {e}");
                None
            }
        }
    } else {
        tracing::info!("S3 storage not configured (image uploads disabled)");
        None
    };

    let state = AppState {
        db: pool.clone(),
        config,
        http: reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(15))
            .connect_timeout(std::time::Duration::from_secs(5))
            .build()
            .expect("failed to build HTTP client"),
        link_preview: multithreaded::link_preview::LinkPreviewFetcher::Http(
            multithreaded::link_preview::build_preview_client(),
        ),
        s3,
    };

    // Session store backed by PostgreSQL, on its own small pool. Every authed
    // request does session I/O and the hourly expiry sweep scans the session
    // table; keeping them off the handler pool means a handler-query burst can't
    // starve session reads (or vice versa) (M-Pf2).
    let session_pool = PgPoolOptions::new()
        .max_connections(6)
        .acquire_timeout(std::time::Duration::from_secs(10))
        .connect(&database_url)
        .await
        .expect("failed to connect session store pool");
    let session_store = PostgresStore::new(session_pool);
    session_store
        .migrate()
        .await
        .expect("failed to migrate session store");

    // Both background loops run under panic supervision: a panic inside a sweep
    // would otherwise silently kill the worker for the life of the process
    // (S3 orphans pile up / the session table grows unbounded) while the server
    // keeps serving. `supervise` restarts the worker after a panic.
    let deletion_store = session_store.clone();
    let deletion_task = supervise("session-expiry-sweep", move || {
        deletion_store
            .clone()
            .continuously_delete_expired(tokio::time::Duration::from_hours(1))
    });

    // Reconcile sweep: purge S3 objects for removed images (backlog + retries).
    // Only meaningful when S3 is configured.
    let purge_task = state.s3.as_ref().map(|s3| {
        let db = state.db.clone();
        let s3 = s3.clone();
        supervise("image-purge-sweep", move || {
            multithreaded::maintenance::continuously_purge_removed_images(
                db.clone(),
                s3.clone(),
                tokio::time::Duration::from_hours(6),
            )
        })
    });

    let session_layer = SessionManagerLayer::new(session_store)
        .with_name("mt_session")
        .with_same_site(SameSite::Lax)
        .with_expiry(tower_sessions::Expiry::OnInactivity(time::Duration::days(
            7,
        )))
        .with_secure(state.config.cookie_secure);

    // CSRF + session are scoped to the forum routes only; the internal API uses
    // HMAC auth and must not run them.
    let forum = multithreaded::routes::forum_routes(state.clone())
        .layer(axum::middleware::from_fn(csrf::csrf_middleware))
        .layer(session_layer);

    // Security headers (CSP, nosniff, X-Frame, cache-control) wrap the WHOLE app,
    // applied outermost so `/static` assets and the internal API get them too,
    // not just forum routes (the layers used to sit inside `forum_routes`, before
    // the merge/nest, so static responses shipped without nosniff/X-Frame/CSP).
    let app = forum
        // Internal API routes, HMAC auth only, no CSRF/session middleware
        .merge(multithreaded::routes::internal::internal_routes(state))
        .nest_service("/static", ServeDir::new("static"))
        .layer(tower_http::set_header::SetResponseHeaderLayer::overriding(
            axum::http::header::CONTENT_SECURITY_POLICY,
            axum::http::HeaderValue::from_static(
                "default-src 'self'; img-src 'self'; style-src 'self'; \
                 frame-ancestors 'none'; object-src 'none'; base-uri 'none'; \
                 form-action 'self'",
            ),
        ))
        .layer(tower_http::set_header::SetResponseHeaderLayer::overriding(
            axum::http::header::X_CONTENT_TYPE_OPTIONS,
            axum::http::HeaderValue::from_static("nosniff"),
        ))
        .layer(tower_http::set_header::SetResponseHeaderLayer::overriding(
            axum::http::header::X_FRAME_OPTIONS,
            axum::http::HeaderValue::from_static("DENY"),
        ))
        .layer(
            tower_http::set_header::SetResponseHeaderLayer::if_not_present(
                axum::http::header::CACHE_CONTROL,
                axum::http::HeaderValue::from_static("private, no-cache"),
            ),
        )
        // Outermost: a hard per-request deadline so a hung handler (e.g. a slow
        // query that outlives the pool acquire, or a stuck await) sheds instead of
        // pinning a connection forever. 30s clears any legitimate request, every
        // outbound call is bounded to 5-15s and the DB acquire to 10s; the only
        // request that transfers meaningful bytes is an image upload, capped at
        // ~5 MB (needs ~1.4 Mbit/s to finish inside the window). Returns 408.
        .layer(tower_http::timeout::TimeoutLayer::with_status_code(
            axum::http::StatusCode::REQUEST_TIMEOUT,
            std::time::Duration::from_secs(30),
        ));

    // Default to loopback. Rate limiting uses TrustedProxyKeyExtractor, which
    // honors CF-Connecting-IP / X-Forwarded-For only from a configured trusted
    // proxy (TRUSTED_PROXIES, default loopback) and otherwise keys on the direct
    // peer, so spoofed forwarding headers from a non-proxy client are ignored
    // regardless of bind address. Binding to
    // 127.0.0.1 is still the right default (only Caddy reaches the port); a
    // tailnet-direct staging box sets HOST and lists its proxy in TRUSTED_PROXIES.
    let host = std::env::var("HOST").unwrap_or_else(|_| "127.0.0.1".to_string());
    let port = std::env::var("PORT").unwrap_or_else(|_| "3400".to_string());
    let addr = format!("{host}:{port}");

    let listener = TcpListener::bind(&addr).await.expect("failed to bind");

    tracing::info!("listening on {}", listener.local_addr().unwrap());

    axum::serve(
        listener,
        app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
    )
    .with_graceful_shutdown(shutdown_signal())
    .await
    .expect("server error");

    deletion_task.abort();
    let _ = deletion_task.await;
    if let Some(task) = purge_task {
        task.abort();
        let _ = task.await;
    }
}

/// Spawn a never-returning background loop under panic supervision.
///
/// The inner future is run in its own task so a panic surfaces here as a
/// `JoinError` instead of silently killing the worker; on a panic, or an
/// unexpected normal return, it is logged and restarted after a short backoff.
/// `factory` is `Fn` so it can rebuild the future (cloning any captured handles)
/// on each restart. Aborting the returned handle (at shutdown) stops the loop.
fn supervise<F, Fut>(name: &'static str, factory: F) -> tokio::task::JoinHandle<()>
where
    F: Fn() -> Fut + Send + 'static,
    Fut: std::future::Future + Send + 'static,
    Fut::Output: Send,
{
    const RESTART_BACKOFF: std::time::Duration = std::time::Duration::from_secs(5);
    tokio::task::spawn(async move {
        loop {
            match tokio::task::spawn(factory()).await {
                Ok(_) => {
                    tracing::error!(
                        task = name,
                        "background task returned unexpectedly; restarting"
                    );
                }
                Err(e) if e.is_panic() => {
                    tracing::error!(task = name, "background task panicked; restarting");
                }
                // Cancelled: the supervisor itself is being aborted at shutdown.
                Err(_) => return,
            }
            tokio::time::sleep(RESTART_BACKOFF).await;
        }
    })
}

async fn shutdown_signal() {
    use tokio::signal;
    let ctrl_c = async {
        signal::ctrl_c()
            .await
            .expect("failed to install Ctrl+C handler");
    };
    #[cfg(unix)]
    let terminate = async {
        signal::unix::signal(signal::unix::SignalKind::terminate())
            .expect("failed to install signal handler")
            .recv()
            .await;
    };
    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();
    tokio::select! {
        () = ctrl_c => {},
        () = terminate => {},
    }
    tracing::info!("Shutdown signal received");
}