//! Application entry point, tracing, config, pool, then hands off to `build_app`.

use axum::http::Request;
use sqlx::ConnectOptions;
use sqlx::postgres::PgPoolOptions;
use std::time::Duration;
use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer};
use tower_http::trace::TraceLayer;
use tower_sessions::cookie::SameSite;
use tower_sessions::cookie::time::Duration as CookieDuration;
use tower_sessions::{Expiry, SessionManagerLayer};
use tower_sessions_sqlx_store::PostgresStore;
use tracing_subscriber::{Layer, layer::SubscriberExt, util::SubscriberInitExt};

use makenotwork::config::Config;
use makenotwork::constants;
use makenotwork::email::{EmailClient, EmailConfig};
use makenotwork::payments::StripeClient;
use makenotwork::scanning::ScanPipeline;
use makenotwork::storage::S3Client;
use makenotwork::{AppState, AppStorage, build_app};
use webauthn_rs::WebauthnBuilder;

#[tokio::main]
async fn main() {
    dotenvy::dotenv().ok();

    // Before any TLS client is built; see the function's own docs for why this
    // cannot be left to rustls to work out.
    makenotwork::crypto::install_default_crypto_provider();

    // JSON in release, human-readable in dev
    tracing_subscriber::registry()
        .with(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| makenotwork::DEFAULT_LOG_FILTER.into()),
        )
        .with(if cfg!(debug_assertions) {
            tracing_subscriber::fmt::layer().boxed()
        } else {
            tracing_subscriber::fmt::layer().json().boxed()
        })
        .init();

    // Sando boot-smoke gate spawns the binary with SANDO_BOOT_SMOKE=1 to
    // verify it loads + links + survives. The real init loads
    // assumptions.toml and other runtime files that don't exist in the
    // sando build workspace. Short-circuit to a tiny axum server that
    // proves the binary, tokio runtime, axum, and TCP bind all work,
    // then idles until the gate kills it.
    if std::env::var("SANDO_BOOT_SMOKE").is_ok() {
        // Sando passes SANDO_BOOT_SMOKE_PORT so the gate can actually probe
        // GET /health (readiness), not just check the process stays up. A bad
        // value or unset falls back to an ephemeral port (liveness-only, the
        // historical behavior) rather than failing the smoke for a config typo.
        let port = std::env::var("SANDO_BOOT_SMOKE_PORT")
            .ok()
            .and_then(|p| p.parse::<u16>().ok())
            .unwrap_or(0);
        tracing::info!(port, "SANDO_BOOT_SMOKE=1; running minimal smoke server");
        let app = axum::Router::new().route("/health", axum::routing::get(|| async { "ok" }));
        let listener = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, port))
            .await
            .expect("smoke: bind 127.0.0.1");
        axum::serve(listener, app).await.expect("smoke: serve");
        return;
    }

    // Config-only validation mode. Loads Config::from_env() with the process
    // environment and exits, no DB connection, no migrations, no socket bind,
    // so a required var missing on a target (e.g. CDN_BASE_URL) is caught
    // *before* a deploy swaps the `current` symlink and restarts into a crash
    // loop. Sando's pre-swap config-drift guard runs `MNW_CHECK_CONFIG=1
    // <binary>` on the target with its env sourced; also handy by hand. Prints a
    // stable sentinel line so a caller can distinguish a genuine config error
    // (exit 1) from a binary too old to support this mode (which would ignore
    // the var and fall through to normal startup).
    if std::env::var("MNW_CHECK_CONFIG").is_ok() {
        match Config::from_env() {
            Ok(_) => {
                println!("MNW_CONFIG_CHECK: ok");
                std::process::exit(0);
            }
            Err(e) => {
                eprintln!("MNW_CONFIG_CHECK: error: {e}");
                std::process::exit(1);
            }
        }
    }

    // Docs integrity check mode. Loads the doc corpus with the exact production
    // config and reports broken internal links, then exits: no DB, no socket,
    // so it is cheap. Sando's `code_smoke` gate runs `MNW_CHECK_DOCS=1 <binary>`
    // as its first step, before the throwaway-DB boot, so a rotted internal link
    // fails the pipeline early rather than serving a live 404 in prod. Broken
    // links fail the check; slug collisions are reported but do not fail it
    // (pre-existing collisions shouldn't red the pipeline). Prints a stable
    // sentinel line, like MNW_CHECK_CONFIG.
    if std::env::var("MNW_CHECK_DOCS").is_ok() {
        let assumptions = match makenotwork::site_docs::load_assumptions() {
            Ok(a) => a,
            Err(e) => {
                eprintln!("MNW_CHECK_DOCS: error: {e}");
                std::process::exit(1);
            }
        };
        let docs = makenotwork::site_docs::build_doc_loader(assumptions);
        let broken = docs.broken_links();
        for b in broken {
            eprintln!(
                "  broken link: {} -> {} (no page serves this slug)",
                b.source_slug, b.target_slug
            );
        }
        for c in docs.collisions() {
            eprintln!(
                "  slug collision: {} ({} displaced by {})",
                c.slug, c.displaced_section, c.winning_section
            );
        }
        if broken.is_empty() {
            println!(
                "MNW_CHECK_DOCS: ok ({} collision(s) reported)",
                docs.collisions().len()
            );
            std::process::exit(0);
        }
        println!("MNW_CHECK_DOCS: {} broken link(s)", broken.len());
        std::process::exit(1);
    }

    let config = Config::from_env().expect("Failed to load configuration");
    tracing::info!("Configuration loaded");

    // A box serving some screens from the description layer and the rest from
    // Askama is worth saying out loud. The failure this guards against is a
    // QUASI_SCREENS left set on a host nobody meant to convert, which otherwise
    // looks exactly like the site behaving oddly for one page.
    if config.quasi_screens.any() {
        tracing::warn!(
            screens = %std::env::var("QUASI_SCREENS").unwrap_or_default(),
            "QUASI_SCREENS is set: these screens serve from the description layer, not Askama"
        );
    }

    // Create database connection pool with health checks and lifecycle limits.
    // - test_before_acquire: validates connections before use (catches stale/broken conns)
    // - max_lifetime: rotates connections to prevent long-lived session issues
    // - idle_timeout: prunes idle connections to free server resources
    // - min_connections: keeps warm connections ready for immediate use
    // - log_slow_statements: logs queries exceeding 100ms at WARN level
    let connect_options: sqlx::postgres::PgConnectOptions =
        config.database_url.parse().expect("Invalid DATABASE_URL");
    let connect_options = connect_options
        .log_statements(log::LevelFilter::Trace)
        .log_slow_statements(log::LevelFilter::Warn, Duration::from_millis(100));

    // Bound every query on every pooled connection: `statement_timeout` caps
    // run time so a wedged query can't pin its connection forever (25 of those
    // would exhaust the pool with no recovery, acquire_timeout only bounds
    // getting a connection, not running one), and `lock_timeout` fails a
    // lock-contended query fast instead of blocking indefinitely.
    let statement_timeout_ms = constants::DB_STATEMENT_TIMEOUT_SECS * 1000;
    let lock_timeout_ms = constants::DB_LOCK_TIMEOUT_SECS * 1000;
    let db = PgPoolOptions::new()
        .max_connections(constants::DB_POOL_MAX_CONNECTIONS)
        .min_connections(constants::DB_POOL_MIN_CONNECTIONS)
        .acquire_timeout(Duration::from_secs(constants::DB_ACQUIRE_TIMEOUT_SECS))
        .max_lifetime(Duration::from_secs(constants::DB_MAX_LIFETIME_SECS))
        .idle_timeout(Duration::from_secs(constants::DB_IDLE_TIMEOUT_SECS))
        .test_before_acquire(true)
        .after_connect(move |conn, _meta| {
            Box::pin(async move {
                use sqlx::Executor;
                conn.execute(
                    format!(
                        "SET statement_timeout = {statement_timeout_ms}; SET lock_timeout = {lock_timeout_ms}"
                    )
                    .as_str(),
                )
                .await?;
                Ok(())
            })
        })
        .connect_with(connect_options)
        .await
        .expect("Failed to connect to database");

    tracing::info!("Database connected");

    // Run migrations, exit cleanly (code 2) on failure instead of panicking.
    // This prevents systemd from crash-looping on migration errors (e.g. version
    // conflicts after a bad deploy). WAM ticket alerts the operator.
    //
    // `sqlx::migrate!` embeds every migration as one array literal, so the lint
    // measures the whole `migrations/` directory. It crossed 16KB when the
    // mailing-list tables landed, which is why this is new and why it is not a
    // finding: the array is built once at startup and the alternative is
    // runtime migration discovery, which is strictly worse for a service that
    // must not start against the wrong schema.
    #[allow(
        clippy::large_stack_arrays,
        reason = "sqlx::migrate! embeds the whole directory"
    )]
    if let Err(e) = sqlx::migrate!("./migrations").run(&db).await {
        tracing::error!(error = %e, "Migration failed, exiting without restart");

        // Best-effort WAM alert (DB is up, WAM may be reachable)
        if let Ok(wam_url) = std::env::var("WAM_URL") {
            let body = format!("Migration error on startup:\n{e}");
            let mut req = reqwest::Client::new()
                .post(format!("{wam_url}/tickets"))
                .json(&serde_json::json!({
                    "title": "Migration failure, server not starting",
                    "body": body,
                    "priority": "critical",
                    "source": "migration-failure",
                }))
                .timeout(std::time::Duration::from_secs(5));
            if let Ok(token) = std::env::var("WAM_TOKEN")
                && !token.trim().is_empty()
            {
                req = req.bearer_auth(token.trim());
            }
            let _ = req.send().await;
        }

        // Exit code 2, systemd configured not to restart on this code
        std::process::exit(2);
    }

    tracing::info!("Migrations complete");

    // Example-marketplace seed for staging (testnot.work). Runs against the
    // already-migrated DB, then exits, never during normal boot. Layered guards
    // (opt-in env, host allowlist, no-real-users) make it refuse against prod.
    // See `makenotwork::seed` and `_private/docs/mnw/testnot-example-seed.md`.
    if std::env::args().any(|a| a == "--seed-examples") {
        let opts = makenotwork::seed::SeedOptions::from_env(&config.host_url);

        // Build the storage handles the media phase needs (main + public/CDN
        // buckets). Both are `None` when S3 is unconfigured, the media phase then
        // no-ops and items stay hidden. Done here, inside the one-shot seed block,
        // so normal boot (which builds these later) is untouched.
        async fn seed_client(
            cfg: Option<&makenotwork::config::StorageConfig>,
            host_url: &str,
        ) -> Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> {
            let cfg = cfg?;
            match S3Client::new(cfg, host_url).await {
                Ok(client) => Some(std::sync::Arc::new(client) as _),
                Err(e) => {
                    tracing::warn!(error = ?e, "seed: failed to init S3 storage; media skipped");
                    None
                }
            }
        }
        let media = makenotwork::seed::SeedMedia {
            s3: seed_client(config.storage.as_ref(), &config.host_url).await,
            public_s3: seed_client(config.public_storage.as_ref(), &config.host_url).await,
            cdn_base_url: Some(config.cdn_base_url.clone()),
            assets: makenotwork::seed::manifest::ResolvedAssets::default(),
        };

        // Fetch the curated public-domain media before touching the database, so
        // a manifest that cannot be resolved leaves the existing catalog alone.
        let media = match media.with_manifest().await {
            Ok(media) => media,
            Err(e) => {
                tracing::error!(error = %e, "example seed: media manifest could not be resolved");
                std::process::exit(1);
            }
        };

        match makenotwork::seed::run(&db, &opts, &media).await {
            Ok(()) => {
                tracing::info!("example seed complete");
                std::process::exit(0);
            }
            Err(e) => {
                tracing::error!(error = %e, "example seed refused or failed");
                std::process::exit(1);
            }
        }
    }

    // Create PostgreSQL-backed session store (persists across restarts)
    let session_store = PostgresStore::new(db.clone());
    session_store
        .migrate()
        .await
        .expect("Failed to migrate session store");

    // Continuously delete expired rows from the tower-sessions table. A session
    // row is minted on every anonymous page render, so without this the table
    // grows without bound under bot/crawler traffic and is read on every
    // request. (The app's own `user_sessions` table is pruned by the daily
    // scheduler; this covers the tower-sessions store, which the scheduler does
    // not own.) Runs hourly on a background task.
    {
        use tower_sessions::ExpiredDeletion;
        let deletion_store = session_store.clone();
        tokio::task::spawn(async move {
            if let Err(e) = deletion_store
                .continuously_delete_expired(tokio::time::Duration::from_hours(1))
                .await
            {
                tracing::error!(error = ?e, "tower-sessions expired-deletion task exited");
            }
        });
    }

    // In release mode, require HTTPS for session cookies (override with INSECURE_COOKIES=1 for staging)
    let secure_cookies =
        !cfg!(debug_assertions) && std::env::var("INSECURE_COOKIES").unwrap_or_default() != "1";
    if secure_cookies {
        tracing::info!("Session cookies configured for HTTPS (secure=true)");
    } else if !cfg!(debug_assertions) {
        tracing::warn!("Session cookies set to insecure (INSECURE_COOKIES=1)");
    }

    let session_layer = SessionManagerLayer::new(session_store)
        .with_secure(secure_cookies)
        .with_http_only(true)
        .with_same_site(SameSite::Lax) // Lax allows session on top-level navigations (OAuth redirects)
        .with_expiry(Expiry::OnInactivity(CookieDuration::days(
            constants::SESSION_EXPIRY_DAYS,
        )));

    tracing::info!("PostgreSQL session store initialized");

    // Initialize S3 client if storage is configured
    let s3: Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> = if let Some(
        ref storage_config,
    ) = config.storage
    {
        match S3Client::new(storage_config, &config.host_url).await {
            Ok(client) => {
                tracing::info!(bucket = %storage_config.bucket, "S3 storage initialized");
                Some(std::sync::Arc::new(client))
            }
            Err(e) => {
                tracing::warn!(error = ?e, "failed to initialize S3 storage, file uploads unavailable");
                None
            }
        }
    } else {
        tracing::info!("S3 storage not configured. File uploads will be unavailable.");
        None
    };

    // Initialize SyncKit blob S3 client if configured (separate bucket)
    let synckit_s3: Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> = if let Some(
        ref synckit_storage_config,
    ) =
        config.synckit_storage
    {
        match S3Client::new(synckit_storage_config, &config.host_url).await {
            Ok(client) => {
                tracing::info!(bucket = %synckit_storage_config.bucket, "SyncKit blob storage initialized");
                Some(std::sync::Arc::new(client))
            }
            Err(e) => {
                tracing::warn!(error = ?e, "Failed to initialize SyncKit blob storage");
                None
            }
        }
    } else {
        tracing::info!("SyncKit blob storage not configured");
        None
    };

    // Initialize public (CDN-served) bucket client if configured. Holds only
    // promoted image content; the scan worker copies Clean covers/gallery here.
    let public_s3: Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> = if let Some(
        ref public_storage_config,
    ) =
        config.public_storage
    {
        match S3Client::new(public_storage_config, &config.host_url).await {
            Ok(client) => {
                tracing::info!(bucket = %public_storage_config.bucket, "Public S3 bucket initialized");
                Some(std::sync::Arc::new(client))
            }
            Err(e) => {
                tracing::warn!(error = ?e, "Failed to initialize public S3 bucket");
                None
            }
        }
    } else {
        tracing::info!("Public S3 bucket not configured");
        None
    };

    // Initialize Stripe client if configured
    let stripe: Option<std::sync::Arc<dyn makenotwork::payments::PaymentProvider>> = if let Some(
        ref stripe_config,
    ) =
        config.stripe
    {
        match StripeClient::new(stripe_config) {
            Ok(client) => {
                tracing::info!("Stripe payments initialized");
                Some(std::sync::Arc::new(client)
                    as std::sync::Arc<dyn makenotwork::payments::PaymentProvider>)
            }
            Err(e) => {
                // Invalid client config is a boot invariant violation, exit
                // without restart rather than run with payments silently broken.
                tracing::error!(error = %e, "Failed to build Stripe client, exiting without restart");
                std::process::exit(2);
            }
        }
    } else {
        tracing::info!("Stripe not configured. Payments will be unavailable.");
        None
    };

    // Initialize email client (logs in dev mode when POSTMARK_TOKEN is not set)
    let email = EmailClient::new(EmailConfig::from_env(), Some(db.clone()));

    // Load business assumptions (single source of truth for figures in the docs).
    // Validation failure aborts startup, we don't want to serve stale numbers.
    let assumptions = makenotwork::site_docs::load_assumptions().unwrap_or_else(|e| panic!("{e}"));

    // Load documentation pages from disk. The same builder backs the DB-free
    // MNW_CHECK_DOCS integrity check above, so what boots and what is checked
    // are the same corpus under the same config.
    let docs = std::sync::Arc::new(makenotwork::site_docs::build_doc_loader(
        assumptions.clone(),
    ));

    // Initialize file scanning pipeline (optional). If scanning is configured,
    // assert at least one AV layer is actually live, otherwise the FailOpen
    // policy on ClamAV silently passes everything as Clean and the deploy
    // looks healthy while shipping zero real coverage. Refuse to boot.
    let scanner = if let Some(ref scan_config) = config.scan {
        match ScanPipeline::new(scan_config) {
            Ok(s) => match s.assert_live().await {
                Ok(()) => {
                    tracing::info!("File scanning enabled");
                    Some(std::sync::Arc::new(s))
                }
                Err(e) => panic!("Scanning configured but no live AV layer: {e}"),
            },
            Err(e) => {
                tracing::warn!(error = %e, "File scanning disabled");
                None
            }
        }
    } else {
        tracing::info!("File scanning not configured");
        None
    };

    // Initialize WebAuthn from HOST_URL (passkey / passwordless login)
    let rp_origin = url::Url::parse(&config.host_url).expect("HOST_URL is not a valid URL");
    let rp_id = rp_origin
        .host_str()
        .expect("HOST_URL has no host")
        .to_string();
    let webauthn = std::sync::Arc::new(
        WebauthnBuilder::new(&rp_id, &rp_origin)
            .expect("Failed to create WebauthnBuilder")
            .rp_name("Makenotwork")
            .build()
            .expect("Failed to build Webauthn"),
    );
    tracing::info!("WebAuthn initialized (rp_id={rp_id})");

    // Initialize syntax highlighter if git repos path is configured
    let syntax = if config.build.git_repos_path.is_some() {
        tracing::info!(path = ?config.build.git_repos_path, "Git source browser enabled");
        Some(std::sync::Arc::new(
            makenotwork::git::SyntaxHighlighter::new(),
        ))
    } else {
        tracing::info!("Git source browser not configured (GIT_REPOS_PATH unset)");
        None
    };

    // Construct MT client when both mt_base_url and internal_shared_secret are set
    let mt_client = match (
        &config.integrations.mt_base_url,
        &config.integrations.internal_shared_secret,
    ) {
        (Some(base_url), Some(secret)) => {
            tracing::info!(base_url = %base_url, "MT integration enabled");
            Some(makenotwork::mt_client::MtClient::new(
                base_url.clone(),
                secret.clone(),
            ))
        }
        _ => {
            tracing::info!(
                "MT integration not configured (need MT_BASE_URL + INTERNAL_SHARED_SECRET)"
            );
            None
        }
    };

    // WAM ticket manager client (tailnet-only, for operational alerts). The
    // optional WAM_TOKEN is the shared secret WAM enforces when set.
    let wam = config.integrations.wam_url.as_ref().map(|url| {
        tracing::info!(url = %url, "WAM integration enabled");
        makenotwork::wam_client::WamClient::new(url.clone(), std::env::var("WAM_TOKEN").ok())
    });

    // Warm custom domain cache
    let domain_cache = std::sync::Arc::new(dashmap::DashMap::new());
    match makenotwork::db::custom_domains::get_all_verified_domains(&db).await {
        Ok(domains) => {
            for d in &domains {
                domain_cache.insert(d.domain.clone(), d.user_id);
            }
            if !domains.is_empty() {
                tracing::info!(count = domains.len(), "Custom domain cache warmed");
            }
        }
        Err(e) => {
            tracing::warn!(error = ?e, "Failed to warm custom domain cache");
        }
    }

    // Shutdown broadcast: dropped at graceful-shutdown time to signal the
    // background pool, monitor, scheduler, and scan workers to drain and stop.
    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(());
    let page_view_tx = makenotwork::db::page_views::spawn_batcher(db.clone());
    let (bg, bg_handle) = makenotwork::background::spawn_pool(shutdown_tx.subscribe());

    // Derive pricing from the loaded assumptions. Installing the process-global
    // TierPrices is an explicit, ordered step here (not buried in AppState
    // construction): CreatorTier accessors (price_cents, max_file_bytes,
    // max_storage_bytes) read the global, so it must be installed before any code
    // path constructs a CreatorTier and calls one of them.
    let tier_prices = makenotwork::tier_prices::TierPrices::from_assumptions(&assumptions);
    tier_prices.clone().install_global();
    let runway_config = makenotwork::tier_prices::RunwayConfig::from_assumptions(&assumptions);
    let fee_calculator = makenotwork::fee_calculator::FeeCalculator::load(
        makenotwork::site_docs::assumptions_path(),
    );

    let state = AppState::build(makenotwork::AppStateParts {
        db,
        config: config.clone(),
        storage: AppStorage {
            s3,
            synckit_s3,
            public_s3,
        },
        stripe,
        email,
        docs,
        tier_prices,
        runway_config,
        fee_calculator,
        scanner,
        webauthn,
        syntax,
        mt_client,
        wam,
        domain_cache,
        metrics_handle: Some(makenotwork::metrics::init()),
        page_view_tx,
        bg,
    });

    // Log active features at startup
    tracing::info!(
        s3 = state.storage.s3.is_some(),
        synckit_s3 = state.storage.synckit_s3.is_some(),
        stripe = state.stripe.is_some(),
        scanner = state.scanner.is_some(),
        mt = state.mt_client.is_some(),
        wam = state.wam.is_some(),
        git = state.config.build.git_repos_path.is_some(),
        "Active features"
    );

    // Start background health monitor and scheduler
    let _monitor_handle =
        makenotwork::monitor::spawn_monitor(axum::extract::FromRef::from_ref(&state), shutdown_rx);
    let scheduler_shutdown_rx = shutdown_tx.subscribe();
    let scheduler_handle =
        makenotwork::scheduler::spawn_scheduler(state.clone(), scheduler_shutdown_rx);

    // Per-instance (not scheduler-gated, so every instance in a rolling deploy
    // agrees): decides whether the footer links to /changelog at all.
    let _changelog_handle =
        makenotwork::changelog::spawn_refresher(state.db.clone(), shutdown_tx.subscribe());

    // Start scan worker pool. Only meaningful if a scanner is configured;
    // otherwise enqueue_scan_for never enqueues (trust-gate fast path).
    if let (Some(scanner), Some(s3_for_workers)) = (state.scanner.clone(), state.storage.s3.clone())
    {
        let scan_ctx = std::sync::Arc::new(makenotwork::scanning::worker::WorkerContext {
            db: state.db.clone(),
            s3: s3_for_workers,
            pipeline: scanner,
            scan_semaphore: state.limiters.scan_semaphore.clone(),
            wam: state.wam.clone(),
            bg: state.bg.clone(),
            cloudflare: makenotwork::cloudflare::CloudflarePurger::from_env(),
            cdn_base_url: std::sync::Arc::from(state.config.cdn_base_url.as_str()),
            synckit_s3: state.storage.synckit_s3.clone(),
            public_s3: state.storage.public_s3.clone(),
            config: state.config.clone(),
        });
        let worker_count = makenotwork::constants::SCAN_WORKER_COUNT;
        let worker_shutdown_rx = shutdown_tx.subscribe();
        makenotwork::scanning::worker::spawn_pool(worker_count, &scan_ctx, worker_shutdown_rx);
        tracing::info!(worker_count, "scan worker pool started");

        let report = makenotwork::scanning::spool::reap_all(std::path::Path::new(
            makenotwork::constants::SCAN_SPOOL_DIR,
        ));
        if report.deleted > 0 || report.errors > 0 {
            tracing::info!(
                deleted = report.deleted,
                errors = report.errors,
                "scan spool startup reaper completed"
            );
        }
    }

    // Security signals need somewhere to send an alert. Before the router, so
    // nothing the server answers is counted against an uninstalled sink.
    makenotwork::security_signals::install(state.db.clone(), state.email.clone());

    // Build router (shared with integration tests via lib.rs)
    let app = build_app(&state, session_layer)
        // Outside every per-route limiter, so a 429 the governor returned is
        // seen here. Counting only; the alert, if any, is spawned off the
        // request path.
        .layer(axum::middleware::from_fn(
            |req: Request<axum::body::Body>, next: axum::middleware::Next| async move {
                // Read the header rather than calling `extract_client_ip`: that
                // helper counts absences to warn about a missing Cloudflare
                // proxy, and a per-response call would double every count.
                let ip = req
                    .headers()
                    .get("cf-connecting-ip")
                    .and_then(|v| v.to_str().ok())
                    .and_then(|s| s.split(',').next())
                    .map(|s| s.trim().to_string())
                    .filter(|s| !s.is_empty());
                let response = next.run(req).await;
                makenotwork::security_signals::note_response(
                    response.status().as_u16(),
                    ip.as_deref(),
                );
                response
            },
        ))
        // Request ID: propagate → trace → set (Axum applies inside-out)
        .layer(PropagateRequestIdLayer::x_request_id())
        .layer(
            TraceLayer::new_for_http()
                .make_span_with(|request: &Request<_>| {
                    let request_id = request
                        .headers()
                        .get("x-request-id")
                        .and_then(|v| v.to_str().ok())
                        .unwrap_or("-");

                    tracing::info_span!(
                        "request",
                        method = %request.method(),
                        uri = %request.uri(),
                        request_id = %request_id,
                        user_id = tracing::field::Empty,
                    )
                })
                .on_response(
                    |response: &axum::http::Response<_>,
                     latency: Duration,
                     _span: &tracing::Span| {
                        let status = response.status().as_u16();
                        if status >= 500 {
                            tracing::error!(
                                status,
                                latency_ms = latency.as_millis() as u64,
                                "response"
                            );
                        } else if status >= 400 {
                            tracing::warn!(
                                status,
                                latency_ms = latency.as_millis() as u64,
                                "response"
                            );
                        } else {
                            tracing::debug!(
                                status,
                                latency_ms = latency.as_millis() as u64,
                                "response"
                            );
                        }
                    },
                ),
        )
        .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid));

    let addr = config.socket_addr();
    tracing::info!(addr = %addr, "listening");

    let listener = tokio::net::TcpListener::bind(addr)
        .await
        .expect("Failed to bind to address");

    // Use into_make_service_with_connect_info to provide peer IP for rate limiting.
    // After the shutdown signal fires, in-flight requests drain, then the
    // scheduler + background pool drain concurrently; the hard-exit timer
    // (SHUTDOWN_HARD_EXIT_SECS, armed at signal receipt) is the backstop that
    // frees a hung restart. The budget is reconciled so the pool drain fits
    // before the hard exit (see the shutdown constants below).
    // Bound the in-flight request drain. `with_graceful_shutdown` alone waits for
    // in-flight requests with NO deadline, so a single hung request stalls
    // shutdown until the hard-exit kills the process, taking the post-serve pool
    // drain with it (audit Run 24 A4). Fire a one-shot the instant the signal
    // arrives, then give the drain at most SHUTDOWN_REQUEST_DRAIN_SECS from that
    // point before abandoning it and proceeding to the pool drain. (The timer must
    // start at signal time, not server-start time, hence the one-shot.)
    let (signal_tx, signal_rx) = tokio::sync::oneshot::channel::<()>();
    let signal_tx = std::sync::Mutex::new(Some(signal_tx));
    let serve = axum::serve(
        listener,
        app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
    )
    .with_graceful_shutdown(async move {
        shutdown_signal().await;
        if let Some(tx) = signal_tx.lock().unwrap().take() {
            let _ = tx.send(());
        }
    })
    .into_future();
    tokio::pin!(serve);

    let drain_deadline = async {
        let _ = signal_rx.await; // resolves only once the signal fires
        tokio::time::sleep(Duration::from_secs(SHUTDOWN_REQUEST_DRAIN_SECS)).await;
    };
    tokio::select! {
        r = &mut serve => r.expect("Server error"),
        () = drain_deadline => tracing::warn!(
            "in-flight request drain exceeded {SHUTDOWN_REQUEST_DRAIN_SECS}s; \
             abandoning it and proceeding to the pool drain"
        ),
    }

    drop(shutdown_tx); // signal monitor, scheduler, and scan workers to stop

    // Drain the scheduler's in-flight tick (platform-credit settlement,
    // pending-S3-deletion retries, webhook re-drives) and the fire-and-forget
    // background pool (queued emails / cache purges / MT thread creations)
    // together. Both are engineered idempotent, so a truncated drain is
    // recoverable, but awaiting a bounded window lets in-progress work finish
    // cleanly. They run CONCURRENTLY under one shared budget so the pool drain
    // isn't starved by a slow scheduler tick before the hard-exit fires: the old
    // sequential ≤5s + ≤5s could reach 10s and let the hard-exit truncate the
    // second drain (audit Run 23 F4). If the shared window overruns we proceed
    // and rely on idempotency as the backstop.
    let drains = async { tokio::join!(scheduler_handle, bg_handle) };
    match tokio::time::timeout(Duration::from_secs(SHUTDOWN_POOL_DRAIN_SECS), drains).await {
        Ok((scheduler_res, bg_res)) => {
            match scheduler_res {
                Ok(()) => tracing::info!("scheduler drained"),
                Err(e) => tracing::warn!(error = ?e, "scheduler task ended abnormally on shutdown"),
            }
            match bg_res {
                Ok(()) => tracing::info!("background pool drained"),
                Err(e) => {
                    tracing::warn!(error = ?e, "background pool task ended abnormally on shutdown");
                }
            }
        }
        Err(_) => tracing::warn!(
            "scheduler + background pool did not drain within {SHUTDOWN_POOL_DRAIN_SECS}s of \
             shutdown; proceeding (work is idempotent)"
        ),
    }

    tracing::info!("Server shut down gracefully");
}

/// Filesystem path to the business-assumptions TOML, `ASSUMPTIONS_PATH` or its
/// default. Single source for the default so every reader agrees.
/// Shutdown budget, reconciled so the background-pool drain can't be truncated
/// by the hard-exit deadline (audit Run 23 F4). The hard-exit timer arms at
/// signal receipt and must outlast the in-flight request drain PLUS the
/// post-serve drain window, otherwise the later drain is cut off mid-flight.
/// The two post-serve drains run *concurrently* (see `join!` below), so they
/// share the pool-drain window rather than summing to twice it.
const SHUTDOWN_REQUEST_DRAIN_SECS: u64 = 10;
const SHUTDOWN_POOL_DRAIN_SECS: u64 = 5;
/// Hard-exit backstop, defined as its own ceiling (not the sum of the two
/// drains) with margin above them, the process is force-killed only if BOTH
/// bounded drains overrun their budgets. The margin also makes the invariant
/// below a real check rather than the tautology it was when this equalled the
/// sum (audit Run 24 A4).
const SHUTDOWN_HARD_EXIT_SECS: u64 = 20;
const _: () = assert!(
    SHUTDOWN_HARD_EXIT_SECS > SHUTDOWN_REQUEST_DRAIN_SECS + SHUTDOWN_POOL_DRAIN_SECS,
    "hard-exit must outlast the bounded request + pool drains, with margin"
);

/// Wait for SIGINT (Ctrl-C) or SIGTERM, then return to trigger graceful shutdown.
/// In-flight requests get up to `SHUTDOWN_REQUEST_DRAIN_SECS` to complete, then
/// the post-serve drains get `SHUTDOWN_POOL_DRAIN_SECS`; the hard-exit timer
/// (`SHUTDOWN_HARD_EXIT_SECS`) is the backstop that frees a hung restart.
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 SIGTERM handler")
            .recv()
            .await;
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        () = ctrl_c => tracing::info!("Received SIGINT, starting graceful shutdown"),
        () = terminate => tracing::info!("Received SIGTERM, starting graceful shutdown"),
    }

    // Spawn the hard-exit backstop: if the full graceful drain (request drain +
    // concurrent pool drain) overruns its reconciled budget, force exit so a hung
    // connection can't block a deployment restart indefinitely.
    tokio::spawn(async {
        tokio::time::sleep(Duration::from_secs(SHUTDOWN_HARD_EXIT_SECS)).await;
        eprintln!("Graceful shutdown timed out after {SHUTDOWN_HARD_EXIT_SECS}s, forcing exit");
        std::process::exit(1);
    });
}