Mangalord/backend/src/bin/crawler.rs

//! Crawler binary.
//!
//! Now an ops escape hatch sitting alongside the in-process daemon: walks
//! the source's manga listing (all pages), fetches each manga's metadata +
//! chapter list, downloads covers, reconciles chapters — and then, for any
//! chapter belonging to a bookmarked manga whose `page_count` is still 0,
//! fetches the chapter pages inline. The daemon does the same work through
//! `crawler_jobs`; the CLI is kept around for force-refetches and manual
//! backfills.
//!
//! Configuration mirrors the daemon's `CRAWLER_*` env vars (see
//! `crate::config::CrawlerConfig`) plus the CLI-only:
//! - **Start URL**: first CLI positional arg, else `$CRAWLER_START_URL`.
//! - **Skip chapters / chapter content / force re-fetch / keep browser**:
//!   `CRAWLER_SKIP_CHAPTERS`, `CRAWLER_SKIP_CHAPTER_CONTENT`,
//!   `CRAWLER_FORCE_REFETCH_CHAPTERS`, `CRAWLER_KEEP_BROWSER_OPEN`.
//! - **Limit**: `CRAWLER_LIMIT` (max manga detail fetches per run).
//!
//! See `crawler::pipeline::run_metadata_pass` for the shared metadata
//! flow.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{anyhow, Context};
use futures_util::stream::{self, StreamExt};
use mangalord::crawler::browser::{BrowserMode, LaunchOptions};
use mangalord::crawler::browser_manager::{self, BrowserManager};
use mangalord::crawler::content::{self, SyncOutcome};
use mangalord::crawler::pipeline;
use mangalord::crawler::rate_limit::HostRateLimiters;
use mangalord::crawler::session;
use mangalord::storage::{LocalStorage, Storage};
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
use tracing_subscriber::EnvFilter;
use uuid::Uuid;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    dotenvy::dotenv().ok();
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env().unwrap_or_else(|_| {
                "info,mangalord=debug,chromiumoxide::conn=off,chromiumoxide::handler=off"
                    .into()
            }),
        )
        .init();

    let start_url = resolve_start_url()?;
    let database_url = std::env::var("DATABASE_URL")
        .map_err(|_| anyhow!("DATABASE_URL must be set"))?;
    let storage_dir: PathBuf = std::env::var("STORAGE_DIR")
        .unwrap_or_else(|_| "./data/storage".to_string())
        .into();
    let rate_ms = env_u64("CRAWLER_RATE_MS", 1000);
    let cdn_host = std::env::var("CRAWLER_CDN_HOST")
        .ok()
        .filter(|s| !s.trim().is_empty());
    let cdn_rate_ms = env_u64("CRAWLER_CDN_RATE_MS", rate_ms);
    let limit = env_u64("CRAWLER_LIMIT", 0) as usize;
    let skip_chapters = env_bool("CRAWLER_SKIP_CHAPTERS", false);
    let skip_chapter_content = env_bool("CRAWLER_SKIP_CHAPTER_CONTENT", false);
    let chapter_workers = env_u64("CRAWLER_CHAPTER_WORKERS", 1).max(1) as usize;
    let force_refetch_chapters = env_bool("CRAWLER_FORCE_REFETCH_CHAPTERS", false);
    let phpsessid = std::env::var("CRAWLER_PHPSESSID")
        .ok()
        .filter(|s| !s.trim().is_empty());
    let cookie_domain = std::env::var("CRAWLER_COOKIE_DOMAIN")
        .ok()
        .filter(|s| !s.trim().is_empty())
        .or_else(|| session::registrable_domain(&start_url));
    let user_agent = std::env::var("CRAWLER_USER_AGENT")
        .ok()
        .filter(|s| !s.trim().is_empty());
    let proxy_url = std::env::var("CRAWLER_PROXY")
        .ok()
        .filter(|s| !s.trim().is_empty());
    let tor_control_url = std::env::var("CRAWLER_TOR_CONTROL_URL")
        .ok()
        .filter(|s| !s.trim().is_empty());
    let tor_control_password = std::env::var("CRAWLER_TOR_CONTROL_PASSWORD")
        .ok()
        .filter(|s| !s.trim().is_empty());
    let tor_control_cookie_path = std::env::var("CRAWLER_TOR_CONTROL_COOKIE_PATH")
        .ok()
        .filter(|s| !s.trim().is_empty())
        .map(std::path::PathBuf::from);
    let tor_recircuit_max_attempts: u32 = std::env::var("CRAWLER_TOR_RECIRCUIT_MAX_ATTEMPTS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(3)
        .max(1);
    let keep_browser_open = env_bool("CRAWLER_KEEP_BROWSER_OPEN", false);

    let db = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .context("connect to database")?;
    sqlx::migrate!("./migrations").run(&db).await?;

    let storage: Arc<dyn Storage> = Arc::new(LocalStorage::new(&storage_dir));

    let cookie_jar = Arc::new(reqwest::cookie::Jar::default());
    if let (Some(sid), Some(domain)) = (&phpsessid, &cookie_domain) {
        let cookie_str = format!("PHPSESSID={sid}; Domain={domain}; Path=/");
        let seed_url =
            reqwest::Url::parse(&start_url).context("parse start URL for cookie seed")?;
        cookie_jar.add_cookie_str(&cookie_str, &seed_url);
        tracing::info!(domain, "seeded PHPSESSID into reqwest cookie jar");
    }
    // SSRF defence: only download from the catalog host + CDN host (plus
    // optional CRAWLER_DOWNLOAD_ALLOWLIST extras). Built here so the same
    // allowlist guards both the redirect policy and the per-image check.
    let allowlist = Arc::new(build_download_allowlist(&start_url, cdn_host.as_deref()));
    let mut http_builder = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .no_proxy()
        // Re-validate every redirect hop against the allowlist — reqwest's
        // default follows up to 10 redirects and `is_safe_url` only guards
        // the initial URL, so a 302 to a private IP would otherwise pivot
        // inside the deployment (SSRF).
        .redirect(mangalord::crawler::safety::safe_redirect_policy(
            (*allowlist).clone(),
        ))
        // Reject hosts that resolve to private/internal IPs (DNS rebinding).
        .dns_resolver(mangalord::crawler::safety::safe_dns_resolver())
        .cookie_provider(cookie_jar);
    if let Some(ua) = &user_agent {
        http_builder = http_builder.user_agent(ua);
    }
    if let Some(proxy) = &proxy_url {
        http_builder = http_builder
            .proxy(reqwest::Proxy::all(proxy).with_context(|| format!("parse proxy URL: {proxy}"))?);
    }
    let http = http_builder.build().context("build http client")?;

    let mut options = LaunchOptions::from_env();
    if let Some(proxy) = &proxy_url {
        let chromium_proxy = mangalord::crawler::url_utils::chromium_proxy_arg(proxy);
        options.extra_args.push(format!("--proxy-server={chromium_proxy}"));
    }
    let keep_open = match (keep_browser_open, options.mode) {
        (true, BrowserMode::Headed) => true,
        (true, BrowserMode::Headless) => {
            tracing::warn!(
                "CRAWLER_KEEP_BROWSER_OPEN ignored in headless mode (no window to inspect)"
            );
            false
        }
        _ => false,
    };
    tracing::info!(
        ?options,
        %start_url,
        rate_ms,
        cdn_host = ?cdn_host,
        cdn_rate_ms,
        limit,
        skip_chapters,
        skip_chapter_content,
        chapter_workers,
        force_refetch_chapters,
        phpsessid_set = phpsessid.is_some(),
        cookie_domain = ?cookie_domain,
        user_agent = ?user_agent,
        proxy = ?proxy_url,
        keep_open,
        storage_dir = %storage_dir.display(),
        "starting crawler"
    );

    let tor = mangalord::crawler::tor::TorController::from_parts(
        tor_control_url.as_deref(),
        tor_control_password.as_deref(),
        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");
    }

    // BrowserManager with idle_timeout = ZERO so the CLI keeps Chromium
    // alive for the entire run — same lifecycle as the old direct
    // `browser::launch()` flow. on_launch re-injects PHPSESSID + runs the
    // session probe; bad cookies fail fast before any real work happens.
    let on_launch: browser_manager::OnLaunch = match (&phpsessid, &cookie_domain) {
        (Some(sid), Some(domain)) => {
            let sid = sid.clone();
            let domain = domain.clone();
            let start_url_clone = start_url.clone();
            let tor_for_launch = tor.as_ref().map(Arc::clone);
            Arc::new(move |browser| {
                let sid = sid.clone();
                let domain = domain.clone();
                let start_url = start_url_clone.clone();
                let tor_for_launch = tor_for_launch.as_ref().map(Arc::clone);
                Box::pin(async move {
                    session::inject_phpsessid(&browser, &sid, &domain)
                        .await
                        .context("inject_phpsessid")?;
                    session::verify_session_with_recircuit(
                        &browser,
                        &start_url,
                        tor_for_launch.as_deref(),
                        tor_recircuit_max_attempts,
                    )
                    .await
                    .context("verify_session")?;
                    Ok(())
                })
            })
        }
        _ => browser_manager::noop_on_launch(),
    };
    let session_ready = phpsessid.is_some() && cookie_domain.is_some();
    let manager = BrowserManager::new(options, Duration::ZERO, on_launch);

    let result = run(
        Arc::clone(&manager),
        &db,
        Arc::clone(&storage),
        &http,
        &start_url,
        rate_ms,
        cdn_host.as_deref(),
        cdn_rate_ms,
        Arc::clone(&allowlist),
        limit,
        skip_chapters,
        skip_chapter_content || !session_ready,
        chapter_workers,
        force_refetch_chapters,
        tor.clone(),
    )
    .await;

    if keep_open {
        tracing::info!(
            "crawler finished; browser kept open. Press Ctrl+C to close and exit."
        );
        let _ = tokio::signal::ctrl_c().await;
        tracing::info!("Ctrl+C received; closing browser");
    }
    manager.shutdown().await;
    result
}

#[allow(clippy::too_many_arguments)]
async fn run(
    manager: Arc<BrowserManager>,
    db: &PgPool,
    storage: Arc<dyn Storage>,
    http: &reqwest::Client,
    start_url: &str,
    rate_ms: u64,
    cdn_host: Option<&str>,
    cdn_rate_ms: u64,
    allowlist: Arc<mangalord::crawler::safety::DownloadAllowlist>,
    limit: usize,
    skip_chapters: bool,
    skip_chapter_content: bool,
    chapter_workers: usize,
    force_refetch_chapters: bool,
    tor: Option<Arc<mangalord::crawler::tor::TorController>>,
) -> anyhow::Result<()> {
    let mut rate = HostRateLimiters::new(Duration::from_millis(rate_ms));
    if let Some(host) = cdn_host {
        rate = rate.with_override(host, Duration::from_millis(cdn_rate_ms));
    }
    let rate = Arc::new(rate);

    // Per-image download cap (the allowlist is built in `main` and passed in
    // so the HTTP client's redirect policy and this check share one source).
    let max_image_bytes: usize = std::env::var("CRAWLER_MAX_IMAGE_BYTES")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(mangalord::crawler::safety::DEFAULT_MAX_IMAGE_BYTES);
    // Per-chapter image *count* cap — bounds total disk against a hostile
    // reader page listing thousands of <img> tags. `0` disables it.
    let max_images_per_chapter: usize = std::env::var("CRAWLER_MAX_IMAGES_PER_CHAPTER")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(2000);

    let stats = pipeline::run_metadata_pass(
        manager.as_ref(),
        db,
        storage.as_ref(),
        http,
        rate.as_ref(),
        start_url,
        limit,
        skip_chapters,
        allowlist.as_ref(),
        max_image_bytes,
        // Circuit-breaker disabled for the operator-driven CLI: a manual
        // sweep should push through transient failures, not self-abort.
        0,
        // No live status surface for the one-shot CLI.
        None,
        tor.as_deref(),
    )
    .await?;
    tracing::info!(?stats, "metadata pass complete");

    if !skip_chapter_content {
        sync_bookmarked_chapter_content(
            Arc::clone(&manager),
            db,
            Arc::clone(&storage),
            http,
            Arc::clone(&rate),
            "target",
            chapter_workers,
            force_refetch_chapters,
            Arc::clone(&allowlist),
            max_image_bytes,
            max_images_per_chapter,
            tor.clone(),
        )
        .await?;
    }

    Ok(())
}

/// Find every chapter whose manga is bookmarked by at least one user and
/// that hasn't been content-synced yet, then fan them out across `workers`
/// concurrent tasks. Same as before except the browser comes from a
/// BrowserManager lease so it interleaves cleanly with the metadata pass.
///
/// A `SessionExpired` result aborts the phase.
#[allow(clippy::too_many_arguments)]
async fn sync_bookmarked_chapter_content(
    manager: Arc<BrowserManager>,
    db: &PgPool,
    storage: Arc<dyn Storage>,
    http: &reqwest::Client,
    rate: Arc<HostRateLimiters>,
    source_id: &str,
    workers: usize,
    force_refetch: bool,
    allowlist: Arc<mangalord::crawler::safety::DownloadAllowlist>,
    max_image_bytes: usize,
    max_images_per_chapter: usize,
    tor: Option<Arc<mangalord::crawler::tor::TorController>>,
) -> anyhow::Result<()> {
    let pending: Vec<(Uuid, Uuid, String)> = sqlx::query_as(
        r#"
        SELECT id, manga_id, source_url FROM (
            SELECT DISTINCT c.id, c.manga_id, c.created_at, cs.source_url
              FROM chapters c
              JOIN bookmarks b ON b.manga_id = c.manga_id
              JOIN chapter_sources cs ON cs.chapter_id = c.id
             WHERE cs.source_id = $1
               AND cs.dropped_at IS NULL
               AND (c.page_count = 0 OR $2)
        ) sub
        ORDER BY manga_id, created_at ASC
        "#,
    )
    .bind(source_id)
    .bind(force_refetch)
    .fetch_all(db)
    .await
    .context("query pending chapter content")?;

    if pending.is_empty() {
        tracing::info!("chapter content: nothing pending");
        return Ok(());
    }
    tracing::info!(count = pending.len(), workers, "chapter content phase starting");

    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let stats = std::sync::Mutex::new(WorkerStats::default());

    stream::iter(pending.into_iter())
        .for_each_concurrent(workers.max(1), |(chapter_id, manga_id, source_url)| {
            let session_expired = Arc::clone(&session_expired);
            let storage = Arc::clone(&storage);
            let rate = Arc::clone(&rate);
            let manager = Arc::clone(&manager);
            let allowlist = Arc::clone(&allowlist);
            let tor = tor.clone();
            let stats = &stats;
            async move {
                if session_expired.load(std::sync::atomic::Ordering::Relaxed) {
                    return;
                }
                let lease = match manager.acquire().await {
                    Ok(l) => l,
                    Err(e) => {
                        tracing::error!(%chapter_id, error = ?e, "browser acquire failed");
                        let mut s = stats.lock().unwrap();
                        s.failed += 1;
                        return;
                    }
                };
                let outcome = content::sync_chapter_content(
                    &lease,
                    db,
                    storage.as_ref(),
                    http,
                    rate.as_ref(),
                    chapter_id,
                    manga_id,
                    &source_url,
                    force_refetch,
                    allowlist.as_ref(),
                    max_image_bytes,
                    max_images_per_chapter,
                    tor.as_deref(),
                    // CLI one-shot — no live status surface.
                    None,
                    // Standalone CLI doesn't drive the analysis worker.
                    false,
                )
                .await;
                drop(lease);
                let mut s = stats.lock().unwrap();
                match outcome {
                    Ok(SyncOutcome::Fetched { pages }) => {
                        tracing::info!(%chapter_id, pages, "chapter content fetched");
                        s.fetched += 1;
                    }
                    Ok(SyncOutcome::Skipped) => s.skipped += 1,
                    Ok(SyncOutcome::SessionExpired) => {
                        tracing::error!(
                            %chapter_id,
                            "session expired mid-run — refresh CRAWLER_PHPSESSID and re-run"
                        );
                        session_expired
                            .store(true, std::sync::atomic::Ordering::Relaxed);
                    }
                    Err(e) => {
                        tracing::warn!(
                            %chapter_id, error = ?e, "chapter content sync failed"
                        );
                        s.failed += 1;
                    }
                }
            }
        })
        .await;

    let total = stats.into_inner().unwrap();
    tracing::info!(
        fetched = total.fetched,
        skipped = total.skipped,
        failed = total.failed,
        "chapter content phase done"
    );

    if session_expired.load(std::sync::atomic::Ordering::Relaxed) {
        anyhow::bail!("session expired during chapter content phase");
    }
    Ok(())
}

#[derive(Default, Clone, Copy)]
struct WorkerStats {
    fetched: usize,
    skipped: usize,
    failed: usize,
}

fn resolve_start_url() -> anyhow::Result<String> {
    if let Some(arg) = std::env::args().nth(1) {
        return Ok(arg);
    }
    std::env::var("CRAWLER_START_URL").map_err(|_| {
        anyhow!(
            "start URL is required — pass as first CLI arg or set $CRAWLER_START_URL"
        )
    })
}

fn env_u64(name: &str, default: u64) -> u64 {
    std::env::var(name)
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(default)
}

fn env_bool(name: &str, default: bool) -> bool {
    match std::env::var(name).ok().as_deref() {
        Some("1") | Some("true") | Some("TRUE") | Some("yes") => true,
        Some("0") | Some("false") | Some("FALSE") | Some("no") => false,
        _ => default,
    }
}

/// Build the crawler download allowlist from env + the catalog/CDN hosts.
/// Shared by the HTTP client's redirect policy and the per-image safety
/// check so both agree on which hosts are reachable.
///
/// `CRAWLER_ALLOW_ANY_HOST=true` short-circuits the host check for
/// sharded-CDN sources; private-IP and scheme guards still apply.
fn build_download_allowlist(
    start_url: &str,
    cdn_host: Option<&str>,
) -> mangalord::crawler::safety::DownloadAllowlist {
    use mangalord::crawler::safety::DownloadAllowlist;
    if env_bool("CRAWLER_ALLOW_ANY_HOST", false) {
        return DownloadAllowlist::allow_any();
    }
    let mut allow = DownloadAllowlist::new();
    if let Ok(parsed) = reqwest::Url::parse(start_url) {
        if let Some(h) = parsed.host_str() {
            allow = allow.allow(h);
        }
    }
    if let Some(host) = cdn_host {
        allow = allow.allow(host);
    }
    if let Ok(extras) = std::env::var("CRAWLER_DOWNLOAD_ALLOWLIST") {
        for piece in extras.split(',') {
            let trimmed = piece.trim();
            if !trimmed.is_empty() {
                allow = allow.allow(trimmed);
            }
        }
    }
    allow
}