Mangalord/backend/src/app.rs

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::RwLock as StdRwLock;

use anyhow::Context;
use async_trait::async_trait;
use axum::extract::{DefaultBodyLimit, FromRequestParts, Request, State};
use axum::http::{HeaderName, HeaderValue, Method};
use axum::middleware::{self, Next};
use axum::response::Response;
use axum::Router;
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
use tokio_util::sync::CancellationToken;
use tower_http::cors::{AllowOrigin, CorsLayer};
use tower_http::trace::TraceLayer;

use crate::auth::extractor::CurrentUser;
use crate::auth::rate_limit::AuthRateLimiter;
use crate::error::AppError;
use crate::config::{AnalysisConfig, AuthConfig, Config, CrawlerConfig, UploadConfig};
use crate::crawler::browser_manager::{self, BrowserManager};
use crate::crawler::content::{self, SyncOutcome};
use crate::crawler::daemon::{self, ChapterDispatcher, DaemonConfig, MetadataPass};
use crate::crawler::jobs::JobPayload;
use crate::crawler::pipeline::{self, MetadataStats};
use crate::crawler::rate_limit::HostRateLimiters;
use crate::crawler::resync::{RealResyncService, ResyncService};
use crate::crawler::safety::DownloadAllowlist;
use crate::crawler::session;
use crate::repo;
use crate::storage::{LocalStorage, Storage};

#[derive(Clone)]
pub struct AppState {
    pub db: PgPool,
    pub storage: Arc<dyn Storage>,
    pub auth: AuthConfig,
    pub upload: UploadConfig,
    /// Shared rate limiter guarding the `/auth/*` mutation endpoints.
    /// One instance per AppState so tests stay isolated across the
    /// same process.
    pub auth_limiter: Arc<AuthRateLimiter>,
    /// Runtime-swappable controls (crawler resync/control handles + the
    /// analysis enable gate). Shared with the [`Supervisors`] so a config
    /// reload that respawns a daemon updates what handlers see, live —
    /// without a restart. In tests this starts empty (no daemons).
    pub runtime: Arc<RuntimeControls>,
    /// Applies a persisted settings change to the running daemons
    /// (graceful stop + respawn). `Some` in production ([`build`]); `None`
    /// in the test harness, where settings still persist but no daemon is
    /// spawned. See [`DaemonReloader`].
    pub reloader: Option<Arc<dyn DaemonReloader>>,
    /// Env-derived crawler config — the **base** the stored settings DTO is
    /// overlaid onto (carries env-only fields: browser, proxy, TOR, session).
    pub crawler_base: CrawlerConfig,
    /// Env-derived analysis config base (carries the env-only vision API key).
    pub analysis_base: AnalysisConfig,
    /// Browser origins permitted to issue mutating requests to
    /// `/api/v1/admin/*`. See [`crate::config::Config::admin_allowed_origins`]
    /// for the policy. Cloned per-request into the CSRF middleware; the
    /// `Arc` keeps the clone cheap. Empty list → check is skipped
    /// (operator opt-out documented in `.env.example`).
    pub admin_allowed_origins: Arc<Vec<String>>,
    /// Live analysis-event broadcaster. Always present (the worker and the
    /// admin enqueue path publish here; the SSE endpoint subscribes). When
    /// the worker is disabled the channel simply stays quiet.
    pub analysis_events: Arc<crate::analysis::events::AnalysisEvents>,
}

impl AppState {
    /// Current crawler control handle, or `None` when the daemon is stopped.
    pub fn crawler(&self) -> Option<Arc<CrawlerControl>> {
        self.runtime.crawler_control()
    }

    /// Current force-resync service, or `None` when the daemon is stopped.
    pub fn resync(&self) -> Option<Arc<dyn ResyncService>> {
        self.runtime.resync()
    }

    /// Whether AI page analysis is currently enabled (read live).
    pub fn analysis_enabled(&self) -> bool {
        self.runtime.analysis_enabled()
    }
}

/// The crawler control handles, swapped atomically when the daemon is
/// (re)spawned or stopped. Cloned cheaply on every read.
#[derive(Clone, Default)]
struct CrawlerControls {
    resync: Option<Arc<dyn ResyncService>>,
    control: Option<Arc<CrawlerControl>>,
}

/// Shared, runtime-mutable surface read by handlers and mutated by the
/// [`Supervisors`] on a config reload. The analysis gate is an `Arc<AtomicBool>`
/// so the crawler's chapter dispatcher (which enqueues `analyze_page` jobs)
/// sees toggles live without being respawned.
pub struct RuntimeControls {
    crawler: StdRwLock<CrawlerControls>,
    analysis_enabled: Arc<AtomicBool>,
}

impl RuntimeControls {
    /// Empty controls (no crawler daemon) with the given initial analysis gate.
    pub fn new(analysis_enabled: bool) -> Self {
        Self {
            crawler: StdRwLock::new(CrawlerControls::default()),
            analysis_enabled: Arc::new(AtomicBool::new(analysis_enabled)),
        }
    }

    pub fn analysis_enabled(&self) -> bool {
        self.analysis_enabled.load(Ordering::Relaxed)
    }

    /// Flip the analysis gate. Used by the [`Supervisors`] on reload and by
    /// the test harness's stub reloader.
    pub fn set_analysis_enabled(&self, v: bool) {
        self.analysis_enabled.store(v, Ordering::Relaxed);
    }

    /// The shared gate handle, passed to the crawler chapter dispatcher.
    fn analysis_gate(&self) -> Arc<AtomicBool> {
        Arc::clone(&self.analysis_enabled)
    }

    pub fn crawler_control(&self) -> Option<Arc<CrawlerControl>> {
        self.crawler.read().unwrap().control.clone()
    }

    pub fn resync(&self) -> Option<Arc<dyn ResyncService>> {
        self.crawler.read().unwrap().resync.clone()
    }

    /// Test helper: install a resync service without a running daemon.
    pub fn set_resync(&self, resync: Option<Arc<dyn ResyncService>>) {
        self.crawler.write().unwrap().resync = resync;
    }

    fn set_crawler(
        &self,
        resync: Option<Arc<dyn ResyncService>>,
        control: Option<Arc<CrawlerControl>>,
    ) {
        let mut g = self.crawler.write().unwrap();
        g.resync = resync;
        g.control = control;
    }
}

/// Applies a validated settings change to the running daemons by gracefully
/// stopping and respawning them with the new effective config (and updating
/// the shared [`RuntimeControls`]). Implemented by [`Supervisors`] in
/// production; the settings handlers call it after persisting.
#[async_trait]
pub trait DaemonReloader: Send + Sync {
    async fn reload_crawler(&self, cfg: CrawlerConfig) -> anyhow::Result<()>;
    async fn reload_analysis(&self, cfg: AnalysisConfig) -> anyhow::Result<()>;
}

/// Shared handle the admin crawler endpoints use to observe and control
/// the running daemon. Bundled so the handlers take one optional field on
/// `AppState` rather than many.
pub struct CrawlerControl {
    pub browser_manager: Arc<BrowserManager>,
    pub session: Arc<crate::crawler::session_control::SessionController>,
    pub status: crate::crawler::status::StatusHandle,
    /// Used by the "run metadata pass now" endpoint; `None` when no
    /// `CRAWLER_START_URL` is configured (cron disabled).
    pub metadata_pass: Option<Arc<dyn MetadataPass>>,
    /// Used by the "reconcile missing" endpoint; `None` when no
    /// `CRAWLER_START_URL` is configured.
    pub reconcile_pass: Option<Arc<dyn crate::crawler::daemon::ReconcilePass>>,
    /// Drain budget for a manually-triggered coordinated browser restart.
    pub drain_deadline: std::time::Duration,
    /// Held for the duration of a `/admin/crawler/run` pass so a second
    /// click returns 409 instead of fanning N overlapping metadata passes
    /// onto the single browser lease. The daemon's daily cron does NOT
    /// take this lock — cron and operator-triggered are different
    /// trust modes (cron is single-fire by definition; the lock only
    /// dedups operator clicks). `Arc` so the spawned task can hold an
    /// owned guard past the request boundary.
    pub manual_pass_lock: Arc<tokio::sync::Mutex<()>>,
}

/// Bundle returned by [`build`]. The router is what `axum::serve` consumes;
/// the [`Supervisors`] own the background daemons (when enabled), outlive the
/// HTTP server, and are awaited via [`AppHandle::shutdown`] after the listener
/// has finished gracefully.
pub struct AppHandle {
    pub router: Router,
    pub supervisors: Arc<Supervisors>,
}

impl AppHandle {
    pub async fn shutdown(self) {
        self.supervisors.shutdown().await;
    }
}

/// Owns the crawler + analysis daemon handles and respawns them on a config
/// reload. Holds the build-time inputs (db, storage, env-base configs, the
/// shared event bus and [`RuntimeControls`]) needed to spawn from scratch.
pub struct Supervisors {
    db: PgPool,
    storage: Arc<dyn Storage>,
    runtime: Arc<RuntimeControls>,
    analysis_events: Arc<crate::analysis::events::AnalysisEvents>,
    /// Serializes reloads + owns the live crawler daemon handle.
    crawler_handle: tokio::sync::Mutex<Option<daemon::DaemonHandle>>,
    /// Serializes reloads + owns the live analysis daemon handle.
    analysis_handle: tokio::sync::Mutex<Option<crate::analysis::daemon::AnalysisDaemonHandle>>,
}

impl Supervisors {
    pub async fn shutdown(&self) {
        if let Some(h) = self.crawler_handle.lock().await.take() {
            h.shutdown().await;
        }
        if let Some(h) = self.analysis_handle.lock().await.take() {
            h.shutdown().await;
        }
    }
}

#[async_trait]
impl DaemonReloader for Supervisors {
    async fn reload_crawler(&self, cfg: CrawlerConfig) -> anyhow::Result<()> {
        // Serialize reloads; do the (possibly slow) graceful shutdown under
        // the handle lock but outside the read-path RwLock so status reads
        // never block on a draining browser.
        let mut guard = self.crawler_handle.lock().await;
        if let Some(h) = guard.take() {
            h.shutdown().await;
        }
        self.runtime.set_crawler(None, None);
        if cfg.daemon_enabled {
            let spawned = spawn_crawler_daemon(
                self.db.clone(),
                Arc::clone(&self.storage),
                &cfg,
                self.runtime.analysis_gate(),
            )
            .await?;
            self.runtime
                .set_crawler(Some(spawned.resync), Some(spawned.crawler));
            *guard = Some(spawned.handle);
            tracing::info!("crawler daemon (re)started from settings");
        } else {
            tracing::info!("crawler daemon stopped (disabled in settings)");
        }
        Ok(())
    }

    async fn reload_analysis(&self, cfg: AnalysisConfig) -> anyhow::Result<()> {
        let mut guard = self.analysis_handle.lock().await;
        if let Some(h) = guard.take() {
            h.shutdown().await;
        }
        self.runtime.set_analysis_enabled(cfg.enabled);
        if cfg.enabled {
            let handle = spawn_analysis_daemon(
                self.db.clone(),
                Arc::clone(&self.storage),
                &cfg,
                Arc::clone(&self.analysis_events),
            )
            .await?;
            *guard = Some(handle);
            tracing::info!(model = %cfg.model, "analysis daemon (re)started from settings");
        } else {
            tracing::info!("analysis daemon stopped (disabled in settings)");
        }
        Ok(())
    }
}

/// How often the background reaper sweeps expired sessions. Hourly is ample:
/// the sweep is a single indexed DELETE and expired rows are already invisible
/// to auth, so this is purely storage hygiene.
const SESSION_GC_INTERVAL: std::time::Duration = std::time::Duration::from_secs(3600);

pub async fn build(config: Config) -> anyhow::Result<AppHandle> {
    let db = PgPoolOptions::new()
        .max_connections(config.db.max_connections)
        .acquire_timeout(config.db.acquire_timeout)
        .connect(&config.database_url)
        .await?;
    sqlx::migrate!("./migrations").run(&db).await?;

    if let Some((username, password)) = config.admin_bootstrap.as_ref() {
        repo::user::bootstrap_admin(&db, username, password)
            .await
            .context("bootstrap_admin from ADMIN_USERNAME/ADMIN_PASSWORD env")?;
        tracing::info!(admin_username = %username, "admin bootstrap ensured");
    }

    let storage: Arc<dyn Storage> = Arc::new(LocalStorage::new(config.storage_dir.clone()));

    // Seed the settings rows from env on first boot, then load the effective
    // config (DB overlaid on the env base). The DB is the source of truth
    // after the first boot; env only fills a missing row.
    let crawler_cfg = load_effective_crawler(&db, &config.crawler).await?;
    let analysis_cfg = load_effective_analysis(&db, &config.analysis).await?;

    // Live-event bus shared by the worker (progress) and the admin enqueue
    // path; the SSE endpoint subscribes. Created unconditionally so the
    // stream exists even with the worker disabled.
    let analysis_events = Arc::new(crate::analysis::events::AnalysisEvents::new());

    let runtime = Arc::new(RuntimeControls::new(analysis_cfg.enabled));
    let supervisors = Arc::new(Supervisors {
        db: db.clone(),
        storage: Arc::clone(&storage),
        runtime: Arc::clone(&runtime),
        analysis_events: Arc::clone(&analysis_events),
        crawler_handle: tokio::sync::Mutex::new(None),
        analysis_handle: tokio::sync::Mutex::new(None),
    });

    // Initial spawn goes through the same reload path so there's one code
    // path for "bring the daemon up with this config". A spawn failure is
    // logged but does NOT abort startup: the persisted settings are the
    // source of truth and a bad value (e.g. a config that won't launch the
    // browser) must not wedge the whole server into a boot loop — the
    // server comes up with that daemon stopped so an admin can fix it via
    // the settings UI.
    if crawler_cfg.daemon_enabled {
        if let Err(e) = supervisors.reload_crawler(crawler_cfg).await {
            tracing::error!(?e, "crawler daemon failed to start; continuing with it stopped");
        }
    } else {
        tracing::info!("crawler daemon disabled");
    }
    if analysis_cfg.enabled {
        if let Err(e) = supervisors.reload_analysis(analysis_cfg).await {
            tracing::error!(?e, "analysis worker failed to start; continuing with it stopped");
        }
    } else {
        tracing::info!("analysis worker disabled");
    }

    // Periodic reaper for lapsed sessions. `find_active` already ignores
    // expired rows, so this only reclaims storage — without it the table grows
    // unbounded as sessions lapse. Detached and best-effort: a failed sweep is
    // logged and retried next tick. Runs regardless of crawler/analysis config
    // since sessions exist in every deployment.
    {
        let db = db.clone();
        tokio::spawn(async move {
            let mut ticker = tokio::time::interval(SESSION_GC_INTERVAL);
            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
            loop {
                ticker.tick().await;
                match repo::session::delete_expired(&db).await {
                    Ok(0) => {}
                    Ok(n) => tracing::info!(reaped = n, "session gc: removed expired sessions"),
                    Err(e) => tracing::warn!(?e, "session gc sweep failed; retrying next tick"),
                }
            }
        });
    }

    let auth_limiter = Arc::new(AuthRateLimiter::new(config.auth.rate_limit));
    let state = AppState {
        db,
        storage,
        auth: config.auth.clone(),
        upload: config.upload.clone(),
        auth_limiter,
        runtime,
        reloader: Some(Arc::clone(&supervisors) as Arc<dyn DaemonReloader>),
        crawler_base: config.crawler.clone(),
        analysis_base: config.analysis.clone(),
        admin_allowed_origins: Arc::new(config.admin_allowed_origins.clone()),
        analysis_events,
    };
    let router = router(state).layer(cors_layer(&config.cors_allowed_origins));
    Ok(AppHandle { router, supervisors })
}

/// Seed the crawler settings row from env when absent, then return the
/// effective [`CrawlerConfig`] (DB DTO overlaid on the env base). A stored
/// DTO that fails to convert (corruption / drift) falls back to the env base
/// rather than blocking startup.
async fn load_effective_crawler(
    db: &PgPool,
    base: &CrawlerConfig,
) -> anyhow::Result<CrawlerConfig> {
    use crate::settings::{CrawlerSettings, KEY_CRAWLER};
    let dto = match repo::app_settings::get(db, KEY_CRAWLER).await? {
        Some(v) => serde_json::from_value::<CrawlerSettings>(v)
            .unwrap_or_else(|_| CrawlerSettings::from_config(base)),
        None => {
            let dto = CrawlerSettings::from_config(base);
            repo::app_settings::seed_if_absent(db, KEY_CRAWLER, &serde_json::to_value(&dto)?)
                .await?;
            tracing::info!("seeded crawler settings from env");
            dto
        }
    };
    Ok(dto.to_config(base).unwrap_or_else(|e| {
        tracing::warn!(?e, "stored crawler settings invalid; using env base");
        base.clone()
    }))
}

/// Analysis counterpart of [`load_effective_crawler`].
async fn load_effective_analysis(
    db: &PgPool,
    base: &AnalysisConfig,
) -> anyhow::Result<AnalysisConfig> {
    use crate::settings::{AnalysisSettings, KEY_ANALYSIS};
    let dto = match repo::app_settings::get(db, KEY_ANALYSIS).await? {
        Some(v) => serde_json::from_value::<AnalysisSettings>(v)
            .unwrap_or_else(|_| AnalysisSettings::from_config(base)),
        None => {
            let dto = AnalysisSettings::from_config(base);
            repo::app_settings::seed_if_absent(db, KEY_ANALYSIS, &serde_json::to_value(&dto)?)
                .await?;
            tracing::info!("seeded analysis settings from env");
            dto
        }
    };
    Ok(dto.to_config(base).unwrap_or_else(|e| {
        tracing::warn!(?e, "stored analysis settings invalid; using env base");
        base.clone()
    }))
}

/// Spawn the AI content-analysis worker daemon with the given config. Returns
/// its handle. Independent of the crawler daemon (works for uploads with the
/// crawler off). Uses a plain reqwest client — no cookie jar / proxy.
async fn spawn_analysis_daemon(
    db: PgPool,
    storage: Arc<dyn Storage>,
    cfg: &AnalysisConfig,
    events: Arc<crate::analysis::events::AnalysisEvents>,
) -> anyhow::Result<crate::analysis::daemon::AnalysisDaemonHandle> {
    // Reclaim jobs orphaned by a previous crash/kill. `crawler::jobs::reclaim_orphaned`
    // is keyed only on `state='running' AND leased_until < now()`, so it
    // covers any kind that uses the table — including this daemon's
    // `analyze_page` jobs. Previously this ran only inside
    // `spawn_crawler_daemon`, which meant a deploy with the crawler off
    // (analysis-only) never refunded orphaned analysis leases until the
    // lease expiry path lazily picked them up. Safe under multi-replica
    // (only expired leases are touched) and idempotent (the crawler-side
    // call happens-before this if both are running).
    match crate::crawler::jobs::reclaim_orphaned(&db).await {
        Ok(0) => {}
        Ok(n) => tracing::info!(
            reclaimed = n,
            "analysis: reclaimed orphaned in-flight jobs at startup"
        ),
        Err(e) => tracing::warn!(?e, "analysis: reclaim_orphaned at startup failed"),
    }
    // Pick the engine. The OCR backend runs in-process (no network, no
    // readiness gate); the vision backend talks to a local LLM server.
    let (dispatcher, readiness): (
        Arc<dyn crate::analysis::daemon::AnalyzeDispatcher>,
        Option<Arc<dyn crate::analysis::daemon::VisionReadiness>>,
    ) = match cfg.effective_backend() {
        crate::config::AnalysisBackend::Ocr => {
            // Load the `.rten` models once; a bad path is a loud boot error.
            let engine = crate::analysis::ocr::OcrsEngine::from_model_paths(
                &cfg.ocr_detection_model,
                &cfg.ocr_recognition_model,
                cfg.ocr_max_decode_pixels,
            )
            .context("build ocrs engine")?;
            // Cap concurrent CPU-bound OCR runs across all workers so a high
            // ANALYSIS_WORKERS can't oversubscribe the blocking pool.
            let cores = std::thread::available_parallelism()
                .map(|n| n.get())
                .unwrap_or(1);
            let permits =
                crate::analysis::ocr::ocr_concurrency_limit(cfg.workers, cores);
            let dispatcher = Arc::new(crate::analysis::ocr::OcrAnalyzeDispatcher {
                db: db.clone(),
                storage,
                engine: Arc::new(engine),
                max_image_bytes: cfg.max_image_bytes,
                ocr_permits: Arc::new(tokio::sync::Semaphore::new(permits)),
            });
            // In-process engine is always ready — no gate.
            (dispatcher, None)
        }
        crate::config::AnalysisBackend::Vision => {
            let http = reqwest::Client::builder()
                .timeout(cfg.request_timeout)
                // Refuse to honour ambient HTTP_PROXY / HTTPS_PROXY container
                // env. The vision call carries an env-managed bearer token +
                // page image bytes; a stray upstream proxy would exfiltrate
                // both. Mirrors the crawler client's `.no_proxy()` (see
                // `spawn_crawler_daemon`).
                .no_proxy()
                // Re-validate redirect hops so a hostile/compromised vision
                // endpoint can't 302 the bearer token + page bytes into the
                // deployment's internal network. No allowlist here (the
                // endpoint is a single admin-configured URL), so the policy
                // enforces scheme + private-IP only.
                .redirect(crate::crawler::safety::public_redirect_policy())
                // NOTE: deliberately no `.dns_resolver(safe_dns_resolver())`.
                // The vision endpoint is a single operator-configured internal
                // service (e.g. `mangalord-vision`) that legitimately resolves
                // to a private Docker IP; a private-IP-rejecting resolver drops
                // every call. Redirect hops stay guarded by the policy above,
                // and the URL is operator-set, not attacker-controlled input.
                .build()
                .context("build analysis http client")?;
            let vision = crate::analysis::vision::VisionClient::new(http, cfg);
            let dispatcher = Arc::new(crate::analysis::daemon::RealAnalyzeDispatcher {
                db: db.clone(),
                storage,
                vision,
                model: cfg.model.clone(),
                max_image_bytes: cfg.max_image_bytes,
            });
            // When a readiness URL is configured, gate leasing on it so an
            // autoscaler that idle-stops the vision container never lets a job
            // burn its retries. A dedicated short-timeout client keeps the
            // probe snappy and independent of the (long) per-request timeout.
            let readiness: Option<Arc<dyn crate::analysis::daemon::VisionReadiness>> =
                match &cfg.vision_health_url {
                    Some(url) if !url.is_empty() => {
                        let probe = reqwest::Client::builder()
                            .timeout(std::time::Duration::from_secs(5))
                            // Same reasoning as the main analysis client: do
                            // not honour ambient HTTP_PROXY env. The readiness
                            // probe is unauthenticated but a hostile upstream
                            // still gets a useful side-channel on backend
                            // uptime + vision health.
                            .no_proxy()
                            .redirect(crate::crawler::safety::public_redirect_policy())
                            // No private-IP resolver: same operator-configured
                            // internal vision host as the analysis client above.
                            .build()
                            .context("build vision readiness http client")?;
                        Some(Arc::new(crate::analysis::daemon::HttpVisionReadiness {
                            http: probe,
                            health_url: url.clone(),
                        }))
                    }
                    _ => None,
                };
            (dispatcher, readiness)
        }
    };
    let handle = crate::analysis::daemon::spawn(
        db,
        CancellationToken::new(),
        crate::analysis::daemon::AnalysisDaemonConfig {
            dispatcher,
            workers: cfg.workers,
            job_timeout: cfg.job_timeout,
            events,
            readiness,
        },
    );
    // Log the *effective* backend (what the worker actually dispatches
    // through), not the raw `cfg.backend` — with vision dormant they diverge
    // when an operator requests `vision`, and a misleading line here was a
    // real observability footgun. The model label tracks the effective engine.
    let effective_backend = cfg.effective_backend();
    let effective_model: &str = match effective_backend {
        crate::config::AnalysisBackend::Ocr => crate::analysis::ocr::OCR_MODEL_LABEL,
        crate::config::AnalysisBackend::Vision => &cfg.model,
    };
    tracing::info!(
        workers = cfg.workers,
        backend = ?effective_backend,
        model = %effective_model,
        "analysis worker daemon started"
    );
    Ok(handle)
}

/// Bundle returned by [`spawn_crawler_daemon`]. The handle owns the
/// daemon's tasks; `resync` is the operator-trigger service shared with
/// `AppState` so admin endpoints can call into the same browser /
/// rate-limit machinery.
struct SpawnedDaemon {
    handle: daemon::DaemonHandle,
    resync: Arc<dyn ResyncService>,
    crawler: Arc<CrawlerControl>,
}

async fn spawn_crawler_daemon(
    db: PgPool,
    storage: Arc<dyn Storage>,
    cfg: &CrawlerConfig,
    analysis_enabled: Arc<AtomicBool>,
) -> anyhow::Result<SpawnedDaemon> {
    // Publish the opt-in browser SSRF-interception toggle so every headless
    // navigation (via `intercept::open_page`) honours it. Off by default.
    crate::crawler::intercept::set_enabled(cfg.ssrf_intercept);

    // Reqwest client with a shared cookie jar so CDN image fetches include
    // PHPSESSID. The same `Arc<Jar>` is held by the SessionController, so a
    // runtime session refresh rewrites it in place. Initial value: a
    // persisted runtime session (survives restart) takes precedence over
    // CRAWLER_PHPSESSID env.
    let cookie_jar = Arc::new(reqwest::cookie::Jar::default());
    let initial_sid = crate::crawler::session_control::SessionController::load_persisted(&db)
        .await
        .or_else(|| cfg.phpsessid.clone());
    if let (Some(sid), Some(domain), Some(start_url)) =
        (&initial_sid, &cfg.cookie_domain, &cfg.start_url)
    {
        let cookie_str = format!("PHPSESSID={sid}; Domain={domain}; Path=/");
        let seed_url = reqwest::Url::parse(start_url)
            .context("parse CRAWLER_START_URL for cookie seed")?;
        cookie_jar.add_cookie_str(&cookie_str, &seed_url);
    }
    let mut http_builder = reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(30))
        .no_proxy()
        // Re-validate every redirect hop against the download allowlist:
        // reqwest's default policy follows up to 10 redirects, and
        // `is_safe_url` only guards the initial URL, so an allowlisted CDN
        // 302ing to a private IP would otherwise be followed (SSRF).
        .redirect(crate::crawler::safety::safe_redirect_policy(
            cfg.download_allowlist.clone(),
        ))
        .cookie_provider(Arc::clone(&cookie_jar));
    if let Some(ua) = &cfg.user_agent {
        http_builder = http_builder.user_agent(ua);
    }
    if let Some(proxy) = &cfg.proxy {
        http_builder = http_builder
            .proxy(reqwest::Proxy::all(proxy).with_context(|| format!("parse proxy: {proxy}"))?);
    }
    // DNS-rebinding guard: reject hosts that resolve to a private/internal IP,
    // complementing the string-level allowlist check which can't see
    // post-resolution addresses. Attached on the direct path AND on http(s)
    // proxies (reqwest resolves the target itself there). Skipped only for SOCKS
    // proxies, where the proxy — not reqwest — resolves the target, so the only
    // name this resolver would see is the proxy's OWN host (legitimately on a
    // private Docker IP, e.g. `tor` → 172.x); attaching it there rejected every
    // fetch for zero gain. See `should_attach_safe_resolver`.
    if crate::crawler::safety::should_attach_safe_resolver(cfg.proxy.as_deref()) {
        http_builder = http_builder.dns_resolver(crate::crawler::safety::safe_dns_resolver());
    }
    let http = http_builder.build().context("build crawler reqwest")?;

    let mut rate = HostRateLimiters::new(std::time::Duration::from_millis(cfg.rate_ms));
    if let Some(host) = &cfg.cdn_host {
        rate = rate.with_override(host, std::time::Duration::from_millis(cfg.cdn_rate_ms));
    }
    let rate = Arc::new(rate);

    let tor = crate::crawler::tor::TorController::from_parts(
        cfg.tor_control_url.as_deref(),
        cfg.tor_control_password.as_deref(),
        cfg.tor_control_cookie_path.as_deref(),
    )
    .context("build TorController from CRAWLER_TOR_CONTROL_* env")?
    .map(Arc::new);
    if let Some(t) = &tor {
        tracing::info!(?t, "TOR control configured; transient pages will trigger NEWNYM");
    }
    let tor_recircuit_max = cfg.tor_recircuit_max_attempts;

    // Session controller + sticky session-expired flag. Created before the
    // browser so the on_launch hook can read the *current* session value
    // (rather than a value captured at startup), and so a runtime refresh
    // updates the cookie everywhere.
    let session_expired = Arc::new(AtomicBool::new(false));
    let session_controller = crate::crawler::session_control::SessionController::new(
        initial_sid,
        Arc::clone(&cookie_jar),
        cfg.cookie_domain.clone(),
        cfg.start_url.clone(),
        db.clone(),
        Arc::clone(&session_expired),
    );

    // Live status surface, sized to the worker count.
    let status = crate::crawler::status::StatusHandle::new(cfg.chapter_workers);

    // Browser manager. on_launch re-injects PHPSESSID on every fresh
    // chromium spawn so an idle teardown followed by re-launch stays
    // authenticated without operator action.
    let mut launch_opts = cfg.browser.clone();
    if let Some(proxy) = &cfg.proxy {
        let chromium_proxy = crate::crawler::url_utils::chromium_proxy_arg(proxy);
        launch_opts.extra_args.push(format!("--proxy-server={chromium_proxy}"));
    }
    let on_launch = match (&cfg.cookie_domain, &cfg.start_url) {
        (Some(domain), Some(start_url)) => {
            let domain = domain.clone();
            let start_url = start_url.clone();
            let tor_for_launch = tor.as_ref().map(Arc::clone);
            let sc = Arc::clone(&session_controller);
            let on_launch: browser_manager::OnLaunch = Arc::new(move |browser| {
                let domain = domain.clone();
                let start_url = start_url.clone();
                let tor_for_launch = tor_for_launch.as_ref().map(Arc::clone);
                let sc = Arc::clone(&sc);
                Box::pin(async move {
                    // Read the *current* session each launch so a runtime
                    // refresh is picked up on the next (re)launch. No session
                    // configured → run unauthenticated (metadata needs no auth).
                    let Some(sid) = sc.current().await else {
                        tracing::info!("on_launch: no session set — skipping inject + probe");
                        return Ok(());
                    };
                    session::inject_phpsessid(&browser, &sid, &domain)
                        .await
                        .context("on_launch: inject_phpsessid")?;
                    session::verify_session_with_recircuit(
                        &browser,
                        &start_url,
                        tor_for_launch.as_deref(),
                        tor_recircuit_max,
                    )
                    .await
                    .context("on_launch: verify_session")?;
                    Ok(())
                })
            });
            on_launch
        }
        _ => browser_manager::noop_on_launch(),
    };
    let browser_manager = BrowserManager::new(launch_opts, cfg.idle_timeout, on_launch);

    let metadata_pass: Option<Arc<dyn MetadataPass>> = cfg.start_url.as_ref().map(|url| {
        let m: Arc<dyn MetadataPass> = Arc::new(RealMetadataPass {
            browser_manager: Arc::clone(&browser_manager),
            db: db.clone(),
            storage: Arc::clone(&storage),
            http: http.clone(),
            rate: Arc::clone(&rate),
            start_url: url.clone(),
            manga_limit: cfg.manga_limit,
            download_allowlist: cfg.download_allowlist.clone(),
            max_image_bytes: cfg.max_image_bytes,
            metadata_max_consecutive_failures: cfg.metadata_max_consecutive_failures,
            status: status.clone(),
            tor: tor.as_ref().map(Arc::clone),
        });
        m
    });

    let reconcile_pass: Option<Arc<dyn crate::crawler::daemon::ReconcilePass>> =
        cfg.start_url.as_ref().map(|url| {
            let m: Arc<dyn crate::crawler::daemon::ReconcilePass> = Arc::new(RealReconcilePass {
                browser_manager: Arc::clone(&browser_manager),
                db: db.clone(),
                rate: Arc::clone(&rate),
                start_url: url.clone(),
                status: status.clone(),
                tor: tor.as_ref().map(Arc::clone),
            });
            m
        });

    let dispatcher: Arc<dyn ChapterDispatcher> = Arc::new(RealChapterDispatcher {
        browser_manager: Arc::clone(&browser_manager),
        db: db.clone(),
        storage: Arc::clone(&storage),
        http: http.clone(),
        rate: Arc::clone(&rate),
        download_allowlist: cfg.download_allowlist.clone(),
        max_image_bytes: cfg.max_image_bytes,
        max_images_per_chapter: cfg.max_images_per_chapter,
        analysis_enabled,
        transient_failures: Arc::new(AtomicU32::new(0)),
        restart_threshold: cfg.browser_restart_threshold,
        drain_deadline: cfg.job_timeout,
        status: status.clone(),
        tor: tor.as_ref().map(Arc::clone),
    });

    let resync: Arc<dyn ResyncService> = Arc::new(RealResyncService {
        browser_manager: Arc::clone(&browser_manager),
        db: db.clone(),
        storage: Arc::clone(&storage),
        http,
        rate: Arc::clone(&rate),
        download_allowlist: cfg.download_allowlist.clone(),
        max_image_bytes: cfg.max_image_bytes,
        max_images_per_chapter: cfg.max_images_per_chapter,
        tor: tor.as_ref().map(Arc::clone),
    });

    // Shared cancellation: daemon shutdown cancels the BrowserManager's
    // idle reaper too. Reaper itself is added to the daemon's extra_tasks
    // so DaemonHandle::shutdown awaits its completion.
    let cancel = CancellationToken::new();
    let reaper_task = browser_manager::spawn_idle_reaper(
        Arc::clone(&browser_manager),
        cancel.clone(),
    );
    // Also close the browser explicitly on shutdown so we don't rely on
    // kill-on-drop when other Arc<Browser> holders may still exist.
    let shutdown_task = {
        let cancel = cancel.clone();
        let mgr = Arc::clone(&browser_manager);
        tokio::spawn(async move {
            cancel.cancelled().await;
            mgr.shutdown().await;
        })
    };

    // Reclaim jobs orphaned by a previous crash/kill (running with an
    // expired lease) before workers start, so recovery is immediate instead
    // of waiting a full lease window for `lease`'s expiry clause. Safe under
    // multi-replica: only already-expired leases are touched. Best-effort —
    // a failure here just defers recovery to the lazy lease path.
    match crate::crawler::jobs::reclaim_orphaned(&db).await {
        Ok(0) => {}
        Ok(n) => tracing::info!(reclaimed = n, "crawler: reclaimed orphaned in-flight jobs at startup"),
        Err(e) => tracing::warn!(?e, "crawler: reclaim_orphaned at startup failed"),
    }

    let daemon_handle = daemon::spawn(
        db,
        cancel,
        DaemonConfig {
            metadata_pass: metadata_pass.clone(),
            dispatcher,
            chapter_workers: cfg.chapter_workers,
            daily_at: cfg.daily_at,
            tz: cfg.tz,
            retention_days: cfg.retention_days,
            metrics_retention_days: cfg.metrics_retention_days,
            session_expired,
            status: status.clone(),
            job_timeout: cfg.job_timeout,
            extra_tasks: vec![reaper_task, shutdown_task],
        },
    );

    let crawler = Arc::new(CrawlerControl {
        browser_manager: Arc::clone(&browser_manager),
        session: session_controller,
        status,
        metadata_pass,
        reconcile_pass,
        drain_deadline: cfg.job_timeout,
        manual_pass_lock: Arc::new(tokio::sync::Mutex::new(())),
    });

    Ok(SpawnedDaemon {
        handle: daemon_handle,
        resync,
        crawler,
    })
}

// Real impls of the daemon traits, owning the browser manager + I/O. Kept
// in app.rs because they need the same builder-side env wiring that
// AppState gets — the daemon module itself stays free of reqwest / storage
// details so its tests don't pull them in.

struct RealMetadataPass {
    browser_manager: Arc<BrowserManager>,
    db: PgPool,
    storage: Arc<dyn Storage>,
    http: reqwest::Client,
    rate: Arc<HostRateLimiters>,
    start_url: String,
    manga_limit: usize,
    download_allowlist: DownloadAllowlist,
    max_image_bytes: usize,
    metadata_max_consecutive_failures: u32,
    status: crate::crawler::status::StatusHandle,
    tor: Option<Arc<crate::crawler::tor::TorController>>,
}

#[async_trait]
impl MetadataPass for RealMetadataPass {
    async fn run(&self) -> anyhow::Result<MetadataStats> {
        let result = pipeline::run_metadata_pass(
            &self.browser_manager,
            &self.db,
            self.storage.as_ref(),
            &self.http,
            &self.rate,
            &self.start_url,
            self.manga_limit,
            false,
            &self.download_allowlist,
            self.max_image_bytes,
            self.metadata_max_consecutive_failures,
            Some(&self.status),
            self.tor.as_deref(),
        )
        .await;
        if let Err(e) = &result {
            if crate::crawler::nav::anyhow_looks_browser_dead(e) {
                self.browser_manager.invalidate().await;
            }
        }
        // Cover backfill follows the metadata pass even when the pass
        // errored — the early-stop walk can complete its work and bail
        // late, and a transient browser failure shouldn't cancel the
        // residual cover backlog. The backfill has its own per-call cap
        // so a runaway error stream can't monopolise the tick. It sets the
        // CoverBackfill{index,total} phase + current_cover per entry.
        match pipeline::backfill_missing_covers(
            &self.browser_manager,
            &self.db,
            self.storage.as_ref(),
            &self.http,
            &self.rate,
            pipeline::COVER_BACKFILL_DEFAULT_MAX,
            &self.download_allowlist,
            self.max_image_bytes,
            Some(&self.status),
            self.tor.as_deref(),
        )
        .await
        {
            Ok(stats) => {
                if stats.considered > 0 {
                    tracing::info!(?stats, "cover backfill complete");
                }
            }
            Err(e) => {
                tracing::warn!(error = ?e, "cover backfill failed");
                if crate::crawler::nav::anyhow_looks_browser_dead(&e) {
                    self.browser_manager.invalidate().await;
                }
            }
        }
        result
    }
}

struct RealReconcilePass {
    browser_manager: Arc<BrowserManager>,
    db: PgPool,
    rate: Arc<HostRateLimiters>,
    start_url: String,
    status: crate::crawler::status::StatusHandle,
    tor: Option<Arc<crate::crawler::tor::TorController>>,
}

#[async_trait]
impl crate::crawler::daemon::ReconcilePass for RealReconcilePass {
    async fn run(&self) -> anyhow::Result<crate::crawler::reconcile::ReconcileStats> {
        let result = crate::crawler::reconcile::reconcile_missing(
            &self.browser_manager,
            &self.db,
            &self.rate,
            &self.start_url,
            Some(&self.status),
            self.tor.as_deref(),
        )
        .await;
        if let Err(e) = &result {
            if crate::crawler::nav::anyhow_looks_browser_dead(e) {
                self.browser_manager.invalidate().await;
            }
        }
        result
    }
}

struct RealChapterDispatcher {
    browser_manager: Arc<BrowserManager>,
    db: PgPool,
    storage: Arc<dyn Storage>,
    http: reqwest::Client,
    rate: Arc<HostRateLimiters>,
    download_allowlist: DownloadAllowlist,
    max_image_bytes: usize,
    /// Per-chapter image-count cap (see `CrawlerConfig::max_images_per_chapter`).
    max_images_per_chapter: usize,
    /// Enqueue `analyze_page` jobs for freshly-crawled pages. Shared gate
    /// (read live) so toggling analysis at runtime takes effect without a
    /// crawler respawn. Mirrors the analysis enable setting.
    analysis_enabled: Arc<AtomicBool>,
    /// Consecutive transient chapter failures; resets on any success.
    /// Drives the automatic coordinated browser restart.
    transient_failures: Arc<std::sync::atomic::AtomicU32>,
    /// Consecutive-failure count that triggers an auto restart.
    restart_threshold: u32,
    /// How long a coordinated restart waits for in-flight leases to drain.
    drain_deadline: std::time::Duration,
    /// Live status surface — the dispatcher registers each chapter it
    /// crawls (with a realtime page count) here.
    status: crate::crawler::status::StatusHandle,
    tor: Option<Arc<crate::crawler::tor::TorController>>,
}

#[async_trait]
impl ChapterDispatcher for RealChapterDispatcher {
    async fn dispatch(&self, payload: JobPayload) -> anyhow::Result<SyncOutcome> {
        match payload {
            JobPayload::SyncChapterContent {
                source_id: _,
                chapter_id,
                source_chapter_key: _,
            } => {
                let row = repo::chapter::dispatch_target(&self.db, chapter_id)
                    .await
                    .context("look up chapter for dispatch")?;
                let Some((manga_id, source_url, manga_title, chapter_number)) = row else {
                    // Chapter (or its source row) is gone — ack done.
                    return Ok(SyncOutcome::Skipped);
                };
                // Register the chapter as crawling now (live status). The
                // guard removes it on every exit path — success, panic, or
                // the worker's outer-timeout drop.
                let _active = self.status.begin_chapter(crate::crawler::status::ActiveChapter {
                    manga_id,
                    manga_title,
                    chapter_id,
                    chapter_number,
                    pages_done: 0,
                    pages_total: None,
                });
                let lease = match self.browser_manager.acquire().await {
                    Ok(l) => l,
                    Err(e) => {
                        // Browser down / mid-restart: defer the job WITHOUT
                        // burning an attempt (the daemon releases it back to
                        // pending) rather than counting an infrastructure
                        // outage as a per-job failure.
                        tracing::warn!(error = ?e, "dispatch: browser unavailable — deferring job");
                        return Ok(SyncOutcome::BrowserUnavailable);
                    }
                };
                let result = content::sync_chapter_content(
                    &lease,
                    &self.db,
                    self.storage.as_ref(),
                    &self.http,
                    &self.rate,
                    chapter_id,
                    manga_id,
                    &source_url,
                    false,
                    &self.download_allowlist,
                    self.max_image_bytes,
                    self.max_images_per_chapter,
                    self.tor.as_deref(),
                    Some(&self.status),
                    self.analysis_enabled.load(Ordering::Relaxed),
                )
                .await;
                drop(lease);
                match result {
                    Ok(outcome) => {
                        // Any successful dispatch (including a clean Skipped)
                        // means the browser is healthy — reset the streak.
                        self.transient_failures.store(0, Ordering::Release);
                        Ok(outcome)
                    }
                    Err(e) => {
                        let streak = self.transient_failures.fetch_add(1, Ordering::AcqRel) + 1;
                        if crate::crawler::nav::anyhow_looks_browser_dead(&e) {
                            // Hard browser-dead: lazy invalidate (next acquire
                            // relaunches). Reset the streak — we're recovering.
                            self.browser_manager.invalidate().await;
                            self.transient_failures.store(0, Ordering::Release);
                        } else if self.restart_threshold > 0 && streak >= self.restart_threshold {
                            // Persistent transients that TOR recircuit couldn't
                            // fix — proactively restart Chromium.
                            tracing::warn!(
                                streak,
                                threshold = self.restart_threshold,
                                "auto browser restart: consecutive transient chapter failures"
                            );
                            let _ = self
                                .browser_manager
                                .coordinated_restart(self.drain_deadline)
                                .await;
                            self.transient_failures.store(0, Ordering::Release);
                        }
                        Err(e)
                    }
                }
            }
            // Reconcile-enqueued manga-detail sync: fetch the detail page,
            // upsert metadata, sync chapters — the identical per-ref work the
            // cron metadata pass runs inline, via the shared
            // `pipeline::process_manga_ref`.
            JobPayload::SyncManga {
                source_id: _,
                source_manga_key,
                url,
                title,
            } => {
                let source = crate::crawler::source::target::TargetSource::new(url.clone());
                let r = crate::crawler::source::SourceMangaRef {
                    source_manga_key,
                    title,
                    url,
                };
                // Scope the lease so it (and the borrowing FetchContext) drop
                // before any browser-restart handling in the match below.
                let result = {
                    let lease = match self.browser_manager.acquire().await {
                        Ok(l) => l,
                        Err(e) => {
                            // See the SyncChapterContent arm: defer without
                            // burning an attempt when the browser is unavailable.
                            tracing::warn!(error = ?e, "dispatch: browser unavailable — deferring job");
                            return Ok(SyncOutcome::BrowserUnavailable);
                        }
                    };
                    let ctx = crate::crawler::source::FetchContext {
                        browser: &lease,
                        rate: &self.rate,
                        tor: self.tor.as_deref(),
                    };
                    pipeline::process_manga_ref(
                        &ctx,
                        &source,
                        &self.db,
                        self.storage.as_ref(),
                        &self.http,
                        &self.rate,
                        &r,
                        false, // chapters ON — we want chapter rows synced
                        &self.download_allowlist,
                        self.max_image_bytes,
                        Some(&self.status),
                    )
                    .await
                };
                match result {
                    Ok(p) => {
                        self.transient_failures.store(0, Ordering::Release);
                        tracing::info!(
                            manga_id = %p.manga_id,
                            key = %r.source_manga_key,
                            "SyncManga: manga synced"
                        );
                        Ok(SyncOutcome::Fetched { pages: 0 })
                    }
                    Err(pipeline::RefError::Fetch(e)) | Err(pipeline::RefError::Skip(e)) => {
                        let streak = self.transient_failures.fetch_add(1, Ordering::AcqRel) + 1;
                        if crate::crawler::nav::anyhow_looks_browser_dead(&e) {
                            self.browser_manager.invalidate().await;
                            self.transient_failures.store(0, Ordering::Release);
                        } else if self.restart_threshold > 0 && streak >= self.restart_threshold {
                            tracing::warn!(
                                streak,
                                threshold = self.restart_threshold,
                                "auto browser restart: consecutive transient sync_manga failures"
                            );
                            let _ = self
                                .browser_manager
                                .coordinated_restart(self.drain_deadline)
                                .await;
                            self.transient_failures.store(0, Ordering::Release);
                        }
                        Err(e)
                    }
                }
            }
            // Other payload kinds aren't dispatched by this daemon —
            // SyncChapterList is handled inline by the cron's metadata pass;
            // analyze_page is owned by the analysis daemon.
            _ => Ok(SyncOutcome::Skipped),
        }
    }
}

/// Build a router from a pre-assembled state. Used by integration tests
/// so they can swap in a test DB pool and a `tempfile`-backed storage.
pub fn router(state: AppState) -> Router {
    let max_request_bytes = state.upload.max_request_bytes;
    Router::new()
        .nest("/api/v1", crate::api::routes())
        .layer(middleware::from_fn(admin_no_store_guard))
        .layer(middleware::from_fn_with_state(
            state.clone(),
            admin_csrf_guard,
        ))
        .layer(middleware::from_fn_with_state(
            state.clone(),
            private_mode_guard,
        ))
        .layer(DefaultBodyLimit::max(max_request_bytes))
        .with_state(state)
        .layer(TraceLayer::new_for_http())
}

/// Path prefix the admin-only middlewares scope themselves to. The router
/// already nests `/api/v1`, so callers see `/api/v1/admin/...`.
const ADMIN_PATH_PREFIX: &str = "/api/v1/admin/";

/// CSRF defence for cookie-authenticated admin mutations. The session
/// cookie is `SameSite=Lax`, which still permits top-level form-POSTs
/// from a malicious page — this middleware rejects such requests by
/// comparing the request's `Origin` (with `Referer` as fallback) against
/// the configured allowlist. Safe methods (`GET`/`HEAD`/`OPTIONS`) are
/// always allowed.
///
/// **Bearer-token-only requests** (`Authorization: Bearer …` with NO
/// session cookie) are bot API callers — they can't be a CSRF vector
/// because the browser never attaches the Authorization header
/// automatically. Skip the check for them.
///
/// **Bearer + cookie (the "cookie-ride")** is treated as cookie-auth.
/// An attacker page can mint `Authorization: Bearer junk` on a
/// credentialed cross-site POST; the cookie carries the actual
/// authority. Closing this hole means cookie precedence: as soon as a
/// session cookie is present, the CSRF gate fires regardless of the
/// Authorization header. (0.87.10 closed a 0.87.2 regression here.)
///
/// **Cookie-auth requests** must:
///   * be from an allowed origin (`Origin` then `Referer`), AND
///   * have at least one of those headers present (a browser always
///     sends one on a cross-site POST; a missing pair on a cookie-auth
///     request is the exact niche an extension / no-referrer-policy
///     CSRF would try to exploit).
///
/// When the allowlist is empty we fail-closed for cookie-auth requests
/// — this used to silently let everything through, so an operator who
/// forgot to set `ADMIN_ALLOWED_ORIGINS` shipped an unguarded admin
/// surface. Operators on a pure-bot-token deploy aren't impacted (their
/// requests carry only `Authorization: Bearer …` with no session cookie
/// and skip the gate via the bearer-only branch above).
///
/// **No auth at all** (no cookie, no bearer): bypass the CSRF gate so
/// the auth extractor returns a clean 401 instead of a confusing 403.
async fn admin_csrf_guard(
    State(state): State<AppState>,
    req: Request,
    next: Next,
) -> Result<Response, AppError> {
    if !req.uri().path().starts_with(ADMIN_PATH_PREFIX) {
        return Ok(next.run(req).await);
    }
    if matches!(
        *req.method(),
        Method::GET | Method::HEAD | Method::OPTIONS
    ) {
        return Ok(next.run(req).await);
    }
    let headers = req.headers();

    // Detect both auth modes BEFORE choosing a bypass. The previous
    // version short-circuited on "is_bearer" regardless of cookie state,
    // which let an attacker page do a credentialed cross-site POST with
    // a forged `Authorization: Bearer junk` header: the header existed,
    // CSRF bypassed, then the auth extractor authenticated the victim
    // via the session cookie. Cookie precedence here re-anchors the
    // gate to the actual authority being ridden.
    // Drive the parse off the auth-module constant so a rename of
    // `SESSION_COOKIE_NAME` propagates here instead of silently
    // reopening the cookie-ride. Split on `=` (not a prefix match) so
    // an attacker can't shadow with `mangalord_session_x=` etc.
    let has_session_cookie = headers
        .get(axum::http::header::COOKIE)
        .and_then(|v| v.to_str().ok())
        .is_some_and(|raw| {
            raw.split(';').any(|p| {
                p.trim_start().split('=').next()
                    == Some(crate::auth::extractor::SESSION_COOKIE_NAME)
            })
        });

    // Bearer-token-only requests are bot callers and bypass the gate —
    // a browser can't set Authorization on a cross-site POST. But ONLY
    // when no session cookie is also attached; with both present we
    // must treat the request as cookie-auth to defeat the cookie-ride.
    let is_bearer = headers
        .get(axum::http::header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .is_some_and(|v| v.trim_start().to_ascii_lowercase().starts_with("bearer "));
    if is_bearer && !has_session_cookie {
        return Ok(next.run(req).await);
    }

    // No auth context at all → let the auth extractor return a clean 401
    // ("log in first"). A no-auth request can't be a CSRF vector — there's
    // no authority to ride. Without this, an anonymous curl POST to an
    // admin endpoint surfaces 403 with our CSRF message instead of the
    // expected 401, which is confusing for both operators and tests.
    if !has_session_cookie {
        return Ok(next.run(req).await);
    }
    let origin = headers.get("origin").and_then(|v| v.to_str().ok());
    let referer = headers.get("referer").and_then(|v| v.to_str().ok());
    let candidate = origin.or(referer);

    if state.admin_allowed_origins.is_empty() {
        // Fail-closed: cookie-auth admin mutations without an allowlist
        // configuration are refused. Set `ADMIN_ALLOWED_ORIGINS` for the
        // browser deployment, or call with `Authorization: Bearer …`.
        tracing::warn!(
            path = %req.uri().path(),
            "admin CSRF: ADMIN_ALLOWED_ORIGINS is empty — cookie-auth admin mutations are refused (set the env var for browser-exposed deploys)"
        );
        return Err(AppError::Forbidden);
    }

    // Cookie-auth path with an allowlist configured: a browser always
    // sends Origin or Referer on a cross-site POST. A missing pair is
    // either a no-referrer-policy edge case or a tool deliberately
    // hiding origin — refuse rather than risk it.
    let Some(candidate) = candidate else {
        tracing::warn!(
            path = %req.uri().path(),
            "admin CSRF: cookie-auth request with neither Origin nor Referer — refusing"
        );
        return Err(AppError::Forbidden);
    };
    if origin_in_allowlist(candidate, &state.admin_allowed_origins) {
        return Ok(next.run(req).await);
    }
    tracing::warn!(
        candidate = %truncate_for_log(candidate, 64),
        path = %req.uri().path(),
        "admin CSRF: rejecting mutation with disallowed origin"
    );
    Err(AppError::Forbidden)
}

/// Match `candidate` (an `Origin` value, or a `Referer` URL whose
/// origin we'll extract) against `allowed`. `Origin` is `scheme://host[:port]`
/// with no path; `Referer` is a full URL — compare by parsing both and
/// matching scheme + host + port.
fn origin_in_allowlist(candidate: &str, allowed: &[String]) -> bool {
    let cand_origin = parse_origin(candidate);
    let Some(cand) = cand_origin else { return false };
    allowed
        .iter()
        .filter_map(|a| parse_origin(a))
        .any(|a| a == cand)
}

/// Extract the origin (`scheme://host[:port]`) from an `Origin` header
/// value or a `Referer` URL. Returns `None` when the input doesn't parse
/// as a URL with a host.
fn parse_origin(raw: &str) -> Option<String> {
    let url = reqwest::Url::parse(raw.trim()).ok()?;
    let host = url.host_str()?;
    let scheme = url.scheme();
    let port_str = match (url.port(), scheme) {
        (Some(80), "http") => String::new(),
        (Some(443), "https") => String::new(),
        (Some(p), _) => format!(":{p}"),
        (None, _) => String::new(),
    };
    Some(format!("{scheme}://{host}{port_str}"))
}

fn truncate_for_log(s: &str, max: usize) -> &str {
    let end = s
        .char_indices()
        .take(max)
        .last()
        .map(|(i, c)| i + c.len_utf8())
        .unwrap_or(0);
    &s[..end.min(s.len())]
}

/// Forbids intermediaries (CDN, browser bfcache, reverse proxy with a
/// permissive default) from caching admin responses. Defence-in-depth
/// for cookie-authenticated reads — even though the responses already
/// vary on cookie, a misconfigured cache layer in front of the
/// SvelteKit container could leak a logged-in admin's view to another
/// session. Headers added on response so the rest of the API is
/// unaffected.
async fn admin_no_store_guard(req: Request, next: Next) -> Response {
    let is_admin_path = req.uri().path().starts_with(ADMIN_PATH_PREFIX);
    let mut resp = next.run(req).await;
    if is_admin_path {
        resp.headers_mut().insert(
            axum::http::header::CACHE_CONTROL,
            HeaderValue::from_static("no-store"),
        );
    }
    resp
}

/// Paths reachable anonymously even when `PRIVATE_MODE=true`. Login and
/// logout are needed for the auth flow itself; `/health` is reserved
/// for load-balancer probes; `/auth/config` lets the frontend decide
/// whether to render the login form or its anonymous alternatives;
/// `/auth/register` is exempted from the gate so the handler can
/// return its informative `registration_disabled` 403 (the same code
/// public-mode deployments use when `ALLOW_SELF_REGISTER=false`) —
/// the handler itself force-blocks the request body in private mode,
/// so no account ever gets created here. Everything else demands a
/// valid session cookie or bearer token.
fn is_public_in_private_mode(path: &str) -> bool {
    matches!(
        path,
        "/api/v1/health"
            | "/api/v1/auth/config"
            | "/api/v1/auth/login"
            | "/api/v1/auth/logout"
            | "/api/v1/auth/register"
    )
}

/// Site-wide auth gate for `PRIVATE_MODE=true`. With the flag off this
/// is a no-op pass-through, so public deployments take no extra DB
/// hit. With it on, the guard reuses [`CurrentUser`] — the same
/// session-cookie-then-bearer-token logic the per-handler extractor
/// uses — so the two paths can never drift.
async fn private_mode_guard(
    State(state): State<AppState>,
    req: Request,
    next: Next,
) -> Result<Response, AppError> {
    if !state.auth.private_mode {
        return Ok(next.run(req).await);
    }
    if is_public_in_private_mode(req.uri().path()) {
        return Ok(next.run(req).await);
    }
    let (mut parts, body) = req.into_parts();
    match CurrentUser::from_request_parts(&mut parts, &state).await {
        Ok(_) => {
            let req = Request::from_parts(parts, body);
            Ok(next.run(req).await)
        }
        Err(_) => Err(AppError::Unauthenticated),
    }
}

pub(crate) fn cors_layer(allowed_origins: &[String]) -> CorsLayer {
    if allowed_origins.is_empty() {
        // Same-origin only — no CORS headers emitted.
        return CorsLayer::new();
    }
    let origins: Vec<HeaderValue> = allowed_origins
        .iter()
        .filter_map(|o| HeaderValue::from_str(o).ok())
        .collect();
    CorsLayer::new()
        .allow_origin(AllowOrigin::list(origins))
        .allow_credentials(true)
        .allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE])
        .allow_headers([
            HeaderName::from_static("content-type"),
            HeaderName::from_static("authorization"),
        ])
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;
    use axum::http::Request;
    use axum::routing::get;
    use tower::ServiceExt;

    fn test_router() -> Router {
        Router::new().route("/", get(|| async { "ok" }))
    }

    #[tokio::test]
    async fn allowlist_preflight_emits_credentialed_headers() {
        let app = test_router().layer(cors_layer(&["https://app.example.com".to_string()]));
        let resp = app
            .oneshot(
                Request::builder()
                    .method(Method::OPTIONS)
                    .uri("/")
                    .header("origin", "https://app.example.com")
                    .header("access-control-request-method", "POST")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(
            resp.headers().get("access-control-allow-origin").unwrap(),
            "https://app.example.com"
        );
        assert_eq!(
            resp.headers().get("access-control-allow-credentials").unwrap(),
            "true"
        );
    }

    #[tokio::test]
    async fn allowlist_rejects_unlisted_origin() {
        let app = test_router().layer(cors_layer(&["https://app.example.com".to_string()]));
        let resp = app
            .oneshot(
                Request::builder()
                    .method(Method::OPTIONS)
                    .uri("/")
                    .header("origin", "https://evil.example.org")
                    .header("access-control-request-method", "POST")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        // Browsers will refuse the response when the allow-origin header
        // is absent (or doesn't echo the requesting origin).
        assert!(resp.headers().get("access-control-allow-origin").is_none());
    }

    #[tokio::test]
    async fn empty_allowlist_is_same_origin_only() {
        let app = test_router().layer(cors_layer(&[]));
        let resp = app
            .oneshot(
                Request::builder()
                    .method(Method::OPTIONS)
                    .uri("/")
                    .header("origin", "https://app.example.com")
                    .header("access-control-request-method", "POST")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert!(resp.headers().get("access-control-allow-origin").is_none());
        assert!(resp.headers().get("access-control-allow-credentials").is_none());
    }

    /// `spawn_analysis_daemon` runs `crawler::jobs::reclaim_orphaned` at
    /// startup so an analysis-only deploy refunds expired
    /// `analyze_page` leases that a crashed previous run left running.
    /// Without this, the call could be silently removed and the
    /// reclaim test in `tests/crawler_jobs.rs` would still pass
    /// (it covers the helper, not the wiring).
    ///
    /// Seed an expired-lease `analyze_page` row in `crawler_jobs`, call
    /// `spawn_analysis_daemon`, then assert the row went back to
    /// `pending` with the attempt refunded.
    #[sqlx::test(migrations = "./migrations")]
    async fn spawn_analysis_daemon_reclaims_orphaned_analyze_leases(pool: PgPool) {
        use crate::storage::LocalStorage;
        use std::time::Duration;
        use uuid::Uuid;

        // Seed a chapter + page so the worker has something it COULD lease.
        // We don't need the worker to actually run — we're testing the
        // reclaim that happens before workers start.
        let manga_id: Uuid =
            sqlx::query_scalar("INSERT INTO mangas (title) VALUES ('M') RETURNING id")
                .fetch_one(&pool)
                .await
                .unwrap();
        let chapter_id: Uuid = sqlx::query_scalar(
            "INSERT INTO chapters (manga_id, number) VALUES ($1, 1) RETURNING id",
        )
        .bind(manga_id)
        .fetch_one(&pool)
        .await
        .unwrap();
        let page_id: Uuid = sqlx::query_scalar(
            "INSERT INTO pages (chapter_id, page_number, storage_key, content_type) \
             VALUES ($1, 1, 'k/1.png', 'image/png') RETURNING id",
        )
        .bind(chapter_id)
        .fetch_one(&pool)
        .await
        .unwrap();

        // Plant an expired-lease analyze_page row that mimics a previous
        // worker that crashed mid-dispatch (attempts=1, leased_until in
        // the past, state=running).
        let payload = serde_json::json!({
            "kind": "analyze_page",
            "page_id": page_id,
            "force": false
        });
        sqlx::query(
            "INSERT INTO crawler_jobs (payload, state, attempts, leased_until) \
             VALUES ($1, 'running', 1, now() - interval '1 hour')",
        )
        .bind(payload)
        .execute(&pool)
        .await
        .unwrap();

        // Bind to a local so the TempDir lives for the rest of the test.
        // `tempfile::tempdir().unwrap().path()` would drop the TempDir
        // at end-of-expression and `LocalStorage` would hold a path to
        // a deleted directory.
        let storage_dir = tempfile::tempdir().unwrap();
        let storage: Arc<dyn Storage> =
            Arc::new(LocalStorage::new(storage_dir.path()));
        // The worker always runs OCR now (vision is dormant — see
        // `effective_backend`), and the `.rten` models aren't shipped to unit
        // CI. Point the engine at a path that can't exist so the *engine build*
        // fails deterministically. Reclaim runs at the very top of
        // `spawn_analysis_daemon`, before — and independently of — engine
        // readiness, so the row must still be reclaimed even though spawn
        // returns Err. That's exactly the regression this test guards (an
        // analysis-only deploy must reclaim orphaned leases at startup).
        let cfg = crate::config::AnalysisConfig {
            ocr_detection_model: "/nonexistent/text-detection.rten".to_string(),
            ocr_recognition_model: "/nonexistent/text-recognition.rten".to_string(),
            workers: 1,
            job_timeout: Duration::from_secs(1),
            ..Default::default()
        };
        let events = Arc::new(crate::analysis::events::AnalysisEvents::new());

        let spawned = spawn_analysis_daemon(pool.clone(), storage, &cfg, events).await;
        assert!(
            spawned.is_err(),
            "engine build must fail with a missing model path"
        );

        // The reclaim must have moved the row back to pending with the
        // attempt refunded (attempts goes from 1 → 0). Reaching it on the
        // initial running state would mean reclaim never ran.
        let (state, attempts): (String, i32) = sqlx::query_as(
            "SELECT state, attempts FROM crawler_jobs WHERE payload->>'page_id' = $1",
        )
        .bind(page_id.to_string())
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(
            state, "pending",
            "expired-lease analyze_page row must be reclaimed to pending"
        );
        assert_eq!(attempts, 0, "reclaim_orphaned refunds the attempt");
    }
}