//! Chapter content sync — fetch a logged-in chapter page, extract its
//! image URLs in `pageN` order, download each to storage, and atomically
//! persist a `pages` row per image plus the chapter's `page_count`.
//!
//! Only chapters belonging to a manga someone has bookmarked are
//! candidates. The crawler scans bookmarks at the start of each run and
//! enqueues unfetched chapters; the API also enqueues at bookmark-time
//! so users get instant feedback. Both feed into the same queue and
//! dedup by chapter id.

// Implementation lands in the next commits in this branch. Module is
// declared so other crates can `use crawler::content` without breaking
// builds while iteration is in progress.

use anyhow::Context;
use sqlx::PgPool;
use uuid::Uuid;

use crate::crawler::detect::PageError;
use crate::crawler::rate_limit::HostRateLimiters;
use crate::crawler::safety::{ensure_public_target, fetch_stream, looks_like_image, DownloadAllowlist};
use crate::crawler::session::{self, ChapterProbe};
use crate::storage::{Storage, StorageError};

/// Parse the chapter page DOM and return the page images in `pageN`
/// order. Filters out the loader `<img class="loading">` and any
/// `<img>` without a numeric `id="pageN"`.
///
/// Reader pages don't render the site's `#logo` element, so the
/// universal logo-sentinel can't apply here — instead we assert
/// `a#pic_container` is present. Its absence means the response is the
/// transient broken-page response (or a redirect to some other layout)
/// and the caller should retry.
pub fn parse_chapter_pages(html: &str) -> Result<Vec<ChapterImage>, PageError> {
    let doc = scraper::Html::parse_document(html);
    let container_sel = scraper::Selector::parse("a#pic_container").unwrap();
    if doc.select(&container_sel).next().is_none() {
        return Err(PageError::transient("reader: a#pic_container missing"));
    }
    let sel = scraper::Selector::parse("a#pic_container img:not(.loading)").unwrap();
    let mut pages: Vec<ChapterImage> = doc
        .select(&sel)
        .filter_map(|img| {
            let id = img.value().id()?;
            let n: i32 = id.strip_prefix("page")?.parse().ok()?;
            let src = img.value().attr("src")?.trim().to_string();
            if src.is_empty() {
                return None;
            }
            Some(ChapterImage { page_number: n, url: src })
        })
        .collect();
    pages.sort_by_key(|p| p.page_number);
    Ok(pages)
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChapterImage {
    pub page_number: i32,
    pub url: String,
}

/// Outcome of a single chapter sync — surfaced to callers for logging
/// and exit-code decisions.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncOutcome {
    /// All images downloaded and stored, chapter row updated.
    Fetched { pages: usize },
    /// `page_count > 0` already — no-op unless force_refetch is set.
    Skipped,
    /// Session probe failed mid-sync (avatar selector missing on the
    /// chapter page). Caller should abort the whole crawler run.
    SessionExpired,
    /// The headless browser could not be acquired (down or mid-restart).
    /// Produced only by the queue dispatcher (which calls `acquire()`); it is
    /// an *infrastructure* outage, not a job failure, so the daemon returns the
    /// job to `pending` WITHOUT burning a retry attempt. `sync_chapter_content`
    /// itself never returns this — callers that already hold a lease can treat
    /// it as unreachable.
    BrowserUnavailable,
}

/// Per-chapter max fetch attempts when TOR is configured. `N = 3` means
/// up to 3 total page fetches with 2 NEWNYM signals between them. When
/// TOR is not configured the effective budget collapses to 1 (single
/// attempt, no retry, no recircuit — bit-for-bit pre-TOR behavior).
const CHAPTER_RECIRCUIT_MAX_ATTEMPTS: u32 = 3;

/// Outcome of [`fetch_chapter_html_with_recircuit`]. `Ok` carries the
/// final reader HTML; the other two map to `sync_chapter_content`'s
/// existing failure modes.
#[derive(Debug)]
enum ChapterFetchOutcome {
    Ok(String),
    /// `ChapterProbe::Unauthenticated` after exhausting recircuit
    /// budget (or with budget=0). Caller returns
    /// `SyncOutcome::SessionExpired`.
    SessionExpired,
    /// `ChapterProbe::Transient` after exhausting recircuit budget
    /// (or with budget=0). Caller bails so the dispatcher does
    /// exponential backoff.
    PersistentTransient,
}

/// Refuse to navigate Chromium at a chapter URL that points inside the
/// deployment. The image-download path goes through `is_safe_url`, but
/// `new_page` is a second network surface: a `chapter_sources` row whose
/// host resolves to (or was crafted to be) a private IP would otherwise let
/// the headless browser probe `postgres:5432`, the cloud metadata service,
/// etc. Reuses the allowlist-free `ensure_public_target` (scheme +
/// private-IP literal check) since there is no per-host allowlist for the
/// scraped catalog itself.
fn guard_nav_url(source_url: &str) -> anyhow::Result<()> {
    ensure_public_target(source_url)
        .map_err(|e| anyhow::anyhow!("refuse to navigate unsafe chapter URL {source_url}: {e}"))
}

/// Single rate-limited Chromium navigation to the chapter URL,
/// returning the page HTML. Extracted from `sync_chapter_content` so
/// the recircuit loop can call it once per attempt.
async fn fetch_chapter_html_once(
    browser: &chromiumoxide::Browser,
    rate: &HostRateLimiters,
    source_url: &str,
) -> anyhow::Result<String> {
    guard_nav_url(source_url)?;
    rate.wait_for(source_url).await?;
    let page = crate::crawler::intercept::open_page(browser, source_url)
        .await
        .with_context(|| format!("open chapter page {source_url}"))?;
    // Close the tab on every exit path — a `?` on wait_for_nav / content()
    // would otherwise leak it (chromiumoxide doesn't close on drop).
    let closer = page.clone();
    crate::crawler::nav::close_after(
        async move {
            closer.close().await.ok();
        },
        async {
            crate::crawler::nav::wait_for_nav(&page)
                .await
                .context("wait for chapter nav")?;
            // Best-effort wait for the reader marker — same partial-render
            // race that bit the chapter-list parser can hit here. Timeout is
            // not an error; the chapter probe + parser sentinels still catch
            // real failures.
            let _ = crate::crawler::nav::wait_for_selector(
                &page,
                "a#pic_container",
                crate::crawler::nav::SELECTOR_TIMEOUT,
            )
            .await;
            page.content().await.context("read chapter html")
        },
    )
    .await
}

/// Pure-over-IO loop: fetch + classify, up to `max_attempts` total
/// fetches. Between attempts, `recircuit` is invoked (a no-op when
/// TOR isn't configured). `max_attempts = 1` collapses to the
/// original single-shot behavior — `Unauthenticated` →
/// `SessionExpired`, `Transient` → `PersistentTransient` on the first
/// hit, no recircuit.
///
/// Semantics match [`crate::crawler::detect::retry_on_transient`] and
/// [`run_session_probe_loop`]: `N` is **total attempts including the
/// first**, so `N = 3` means 3 fetches and up to 2 NEWNYM calls.
/// `Unauthenticated` and `Transient` share the budget — the loop
/// doesn't distinguish, so a sequence like Transient → Unauth → Ok
/// counts as 3 attempts.
async fn fetch_chapter_html_with_recircuit<F, Fut, R, RFut>(
    mut fetch: F,
    mut recircuit: R,
    max_attempts: u32,
    source_url_for_msg: &str,
) -> anyhow::Result<ChapterFetchOutcome>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = anyhow::Result<String>>,
    R: FnMut() -> RFut,
    RFut: std::future::Future<Output = ()>,
{
    debug_assert!(max_attempts >= 1, "max_attempts must be at least 1");
    let mut attempt = 0u32;
    loop {
        attempt += 1;
        let html = fetch().await?;
        match session::classify_chapter_probe(&html) {
            ChapterProbe::Ok => return Ok(ChapterFetchOutcome::Ok(html)),
            ChapterProbe::Unauthenticated => {
                if attempt >= max_attempts {
                    return Ok(ChapterFetchOutcome::SessionExpired);
                }
                tracing::warn!(
                    attempt,
                    max = max_attempts,
                    url = source_url_for_msg,
                    "chapter probe Unauthenticated; signaling TOR NEWNYM and retrying"
                );
                recircuit().await;
            }
            ChapterProbe::Transient => {
                if attempt >= max_attempts {
                    return Ok(ChapterFetchOutcome::PersistentTransient);
                }
                tracing::warn!(
                    attempt,
                    max = max_attempts,
                    url = source_url_for_msg,
                    "chapter probe Transient; signaling TOR NEWNYM and retrying"
                );
                recircuit().await;
            }
        }
    }
}

/// Fetch one chapter's images and persist them. Each image is
/// streamed straight to storage via `Storage::put_stream` after a
/// short prefix is peeked off the body for content-type sniffing —
/// peak memory per concurrent dispatch is one HTTP chunk plus the
/// sniff prefix, not a full multi-MB image. The per-image size cap
/// (`CRAWLER_MAX_IMAGE_BYTES`) is enforced inside the stream so a
/// server that omits Content-Length still can't exhaust memory. The
/// page rows + `page_count` are then written in one short transaction.
/// On any failure the chapter stays at `page_count = 0` (no partial
/// rows) and the blobs already written are deleted best-effort by
/// [`cleanup_orphans`], so a retry starts clean.
/// Timing wrapper over [`sync_chapter_content_inner`]: records one `chapter`
/// row in `crawl_metrics` with the wall-clock and outcome. Best-effort — a
/// metrics-write failure never fails the chapter sync. `Skipped` /
/// `SessionExpired` perform no fetch, so they aren't timed.
#[allow(clippy::too_many_arguments)]
pub async fn sync_chapter_content(
    browser: &chromiumoxide::Browser,
    db: &PgPool,
    storage: &dyn Storage,
    http: &reqwest::Client,
    rate: &HostRateLimiters,
    chapter_id: Uuid,
    manga_id: Uuid,
    source_url: &str,
    force_refetch: bool,
    allowlist: &DownloadAllowlist,
    max_image_bytes: usize,
    max_images_per_chapter: usize,
    tor: Option<&crate::crawler::tor::TorController>,
    progress: Option<&crate::crawler::status::StatusHandle>,
    enqueue_analysis: bool,
) -> anyhow::Result<SyncOutcome> {
    let started = std::time::Instant::now();
    let result = sync_chapter_content_inner(
        browser, db, storage, http, rate, chapter_id, manga_id, source_url,
        force_refetch, allowlist, max_image_bytes, max_images_per_chapter, tor, progress,
        enqueue_analysis,
    )
    .await;
    let duration_ms = started.elapsed().as_millis() as i64;
    match &result {
        Ok(SyncOutcome::Fetched { pages }) => {
            let _ = crate::repo::crawl_metrics::record(
                db,
                crate::repo::crawl_metrics::OP_CHAPTER,
                Some(manga_id),
                Some(chapter_id),
                "ok",
                duration_ms,
                Some(*pages as i32),
                None,
            )
            .await;
        }
        Err(e) => {
            let _ = crate::repo::crawl_metrics::record(
                db,
                crate::repo::crawl_metrics::OP_CHAPTER,
                Some(manga_id),
                Some(chapter_id),
                "failed",
                duration_ms,
                None,
                Some(&format!("{e:#}")),
            )
            .await;
        }
        // Skipped / SessionExpired: no fetch performed → not a timed op.
        Ok(_) => {}
    }
    result
}

#[allow(clippy::too_many_arguments)]
async fn sync_chapter_content_inner(
    browser: &chromiumoxide::Browser,
    db: &PgPool,
    storage: &dyn Storage,
    http: &reqwest::Client,
    rate: &HostRateLimiters,
    chapter_id: Uuid,
    manga_id: Uuid,
    source_url: &str,
    force_refetch: bool,
    allowlist: &DownloadAllowlist,
    max_image_bytes: usize,
    max_images_per_chapter: usize,
    tor: Option<&crate::crawler::tor::TorController>,
    // Optional live-status sink for the realtime page counter. The daemon
    // dispatcher passes the shared handle (the chapter has already been
    // registered via `begin_chapter`); the CLI / admin resync pass `None`.
    progress: Option<&crate::crawler::status::StatusHandle>,
    // When `true`, enqueue an `analyze_page` job per persisted page. The
    // daemon dispatcher passes the configured flag; CLI / resync pass
    // `false`.
    enqueue_analysis: bool,
) -> anyhow::Result<SyncOutcome> {
    // Skip if already fetched, unless caller explicitly forces.
    if !force_refetch {
        let (page_count,): (i32,) =
            sqlx::query_as("SELECT page_count FROM chapters WHERE id = $1")
                .bind(chapter_id)
                .fetch_one(db)
                .await
                .context("read chapter page_count")?;
        if page_count > 0 {
            return Ok(SyncOutcome::Skipped);
        }
    }

    // Fetch + classify. With TOR configured, allow up to
    // CHAPTER_RECIRCUIT_MAX_ATTEMPTS total page fetches with NEWNYM
    // between each. Without TOR, collapse to 1 attempt (no retry, no
    // recircuit) — matches the pre-TOR single-shot behavior bit-for-bit.
    let max_attempts = if tor.is_some() { CHAPTER_RECIRCUIT_MAX_ATTEMPTS } else { 1 };
    let html = match fetch_chapter_html_with_recircuit(
        || fetch_chapter_html_once(browser, rate, source_url),
        || async {
            if let Some(t) = tor {
                if let Err(e) = t.new_identity().await {
                    tracing::warn!(error = %e, "TOR NEWNYM failed; continuing with same circuit");
                }
            }
        },
        max_attempts,
        source_url,
    )
    .await?
    {
        ChapterFetchOutcome::Ok(html) => html,
        ChapterFetchOutcome::SessionExpired => return Ok(SyncOutcome::SessionExpired),
        ChapterFetchOutcome::PersistentTransient => {
            // Surface as a typed Err so the dispatcher path runs
            // ack_failed with exponential backoff (rather than the
            // session-expired sticky flag).
            anyhow::bail!(
                "chapter page at {source_url} returned a transient response after \
                 {max_attempts} attempt(s); will retry"
            );
        }
    };

    let images = parse_chapter_pages(&html)
        .with_context(|| format!("parse chapter pages at {source_url}"))?;
    if images.is_empty() {
        anyhow::bail!("no page images parsed from {source_url}");
    }
    // Bound total disk per chapter: the per-image byte cap doesn't stop a
    // hostile reader page from listing thousands of <img> tags. Ack failed
    // (the caller records it and backs off) rather than downloading them.
    if let Some(over) = image_count_over_cap(images.len(), max_images_per_chapter) {
        anyhow::bail!(
            "chapter at {source_url} lists {} page images, over the {} cap",
            over.count,
            over.cap
        );
    }

    // Resolve image URLs against the chapter URL (they may be relative).
    let base = reqwest::Url::parse(source_url).context("parse chapter URL")?;

    // Stream each image straight to storage as it's fetched, capping peak
    // memory at a single image rather than the whole chapter. Track the
    // keys written so they can be rolled back if a later page (or the
    // final DB commit) fails — preserving the all-or-nothing guarantee
    // without holding a DB transaction open across the network puts
    // (which matters once `Storage` is backed by S3).
    let total = images.len();
    // Publish the now-known page total so the dashboard shows "0/N".
    if let Some(p) = progress {
        p.set_chapter_pages(chapter_id, 0, Some(total));
    }
    let mut written_keys: Vec<String> = Vec::with_capacity(total);
    let mut stored: Vec<StoredPage> = Vec::with_capacity(total);
    for img in &images {
        match download_and_store_page(
            storage,
            http,
            rate,
            &base,
            source_url,
            manga_id,
            chapter_id,
            img,
            allowlist,
            max_image_bytes,
        )
        .await
        {
            Ok(page) => {
                written_keys.push(page.storage_key.clone());
                stored.push(page);
                // Live page counter: push the climbing count to subscribers.
                if let Some(p) = progress {
                    p.set_chapter_pages(chapter_id, stored.len(), Some(total));
                }
            }
            Err(e) => {
                cleanup_orphans(storage, &written_keys).await;
                return Err(e);
            }
        }
    }

    // Short transaction: page rows + page_count only, no network I/O. On
    // failure, roll back the stored bytes so the chapter stays at
    // page_count=0 and is retried cleanly next run.
    if let Err(e) = persist_pages(db, chapter_id, &stored, enqueue_analysis).await {
        cleanup_orphans(storage, &written_keys).await;
        return Err(e);
    }

    Ok(SyncOutcome::Fetched { pages: stored.len() })
}

/// A page image that has been written to storage and is awaiting its DB
/// row. Carries everything `persist_pages` needs.
pub(crate) struct StoredPage {
    page_number: i32,
    storage_key: String,
    content_type: String,
    /// Bytes written to storage, captured from `put_stream`'s return so
    /// storage-usage stats are a pure DB SUM.
    size_bytes: i64,
}

/// Bytes accumulated for content-type sniffing. `infer` only needs the
/// first few bytes for image formats (the longest signature in our
/// allow-list is AVIF at 12 bytes), but we read up to this many so a
/// fragmented TCP frame still produces a confident sniff and the
/// "first 16 bytes" diagnostic in the error path is useful.
const SNIFF_PREFIX_BYTES: usize = 64;

/// Bytes still admissible for the streaming tail after the sniff prefix
/// has been drained. The prefix already counts against the per-image cap,
/// so the tail budget is `max_image_bytes - prefix_len` using the
/// **actual** drained length — never a constant — so `prefix_len + tail`
/// can never exceed the cap.
fn remaining_after_prefix(max_image_bytes: usize, prefix_len: usize) -> usize {
    max_image_bytes.saturating_sub(prefix_len)
}

/// A rejected over-cap image count, carrying the numbers for the error.
struct ImageCountOverCap {
    count: usize,
    cap: usize,
}

/// `Some(..)` when `count` exceeds the per-chapter image cap. A `cap` of
/// `0` disables the check (unbounded), matching the config's "0 = no cap"
/// contract.
fn image_count_over_cap(count: usize, cap: usize) -> Option<ImageCountOverCap> {
    (cap != 0 && count > cap).then_some(ImageCountOverCap { count, cap })
}

/// Download a single page image, validate it's really an image, and
/// stream it to storage. Returns the storage key + content type. Does
/// not touch the DB — persistence is batched into one short transaction
/// afterward.
///
/// Streaming path: we peek the first [`SNIFF_PREFIX_BYTES`] from the
/// HTTP body to determine the file extension (and thus the storage
/// key), then re-emit those bytes followed by the rest of the response
/// stream via `Storage::put_stream`. Peak memory per concurrent
/// dispatch is one HTTP chunk (~16 KiB) plus the sniff prefix, not a
/// full multi-MB image. The per-image cap is enforced as bytes flow.
#[allow(clippy::too_many_arguments)]
async fn download_and_store_page(
    storage: &dyn Storage,
    http: &reqwest::Client,
    rate: &HostRateLimiters,
    base: &reqwest::Url,
    source_url: &str,
    manga_id: Uuid,
    chapter_id: Uuid,
    img: &ChapterImage,
    allowlist: &DownloadAllowlist,
    max_image_bytes: usize,
) -> anyhow::Result<StoredPage> {
    use futures_util::StreamExt as _;
    let url = base
        .join(&img.url)
        .with_context(|| format!("join image URL {} onto {source_url}", img.url))?;
    rate.wait_for(url.as_str()).await?;
    let resp = fetch_stream(http, url.as_str(), Some(source_url), allowlist).await?;
    let mut body = resp.bytes_stream();

    // Drain chunks until we have enough bytes to sniff confidently
    // (or the body is shorter than the prefix). Enforces the per-image
    // cap on the prefix accumulation too.
    let mut prefix = bytes::BytesMut::new();
    while prefix.len() < SNIFF_PREFIX_BYTES {
        let Some(chunk) = body.next().await else { break };
        let chunk = chunk
            .with_context(|| format!("stream chunk for {url}"))?;
        if prefix.len().saturating_add(chunk.len()) > max_image_bytes {
            anyhow::bail!(
                "image {url} exceeds {max_image_bytes}-byte cap (received >{}+{})",
                prefix.len(),
                chunk.len()
            );
        }
        prefix.extend_from_slice(&chunk);
    }
    let prefix = prefix.freeze();

    // Reject any non-image response: the only valid output of an image
    // URL is an image. `infer` returns None on truncated bytes too,
    // which is also a failure not a silent `.bin` extension.
    if !looks_like_image(&prefix) {
        anyhow::bail!(
            "image URL {url} returned non-image bytes \
             (first 16: {:?}); refusing to store as binary blob",
            &prefix.get(..16.min(prefix.len()))
        );
    }
    let ext = infer::get(&prefix)
        .map(|k| k.extension())
        .expect("looks_like_image asserted infer succeeded");
    let key = format!(
        "mangas/{manga_id}/chapters/{chapter_id}/pages/{:04}.{ext}",
        img.page_number
    );

    // Build a single stream of (prefix + remaining body) and pipe it
    // straight to storage. The cap is enforced via a running total in
    // the stream adapter so a server that omits Content-Length still
    // can't exhaust memory.
    // Budget the streaming tail against the bytes *actually* drained into
    // the prefix, not the constant `SNIFF_PREFIX_BYTES`. The prefix loop
    // appends whole chunks, so a single 16 KiB first chunk fills the
    // 64-byte sniff window in one drain — charging only 64 bytes would
    // then let the tail add another full `max_image_bytes`, storing up to
    // ~2× the cap. Capture the length before `prefix` is moved into the
    // stream below.
    let prefix_len = prefix.len();
    let prefix_stream = futures_util::stream::once(async move {
        Ok::<bytes::Bytes, StorageError>(prefix)
    });
    let mut remaining = remaining_after_prefix(max_image_bytes, prefix_len);
    let url_for_err = url.clone();
    let rest_stream = body.map(move |frame| match frame {
        Ok(chunk) => {
            if chunk.len() > remaining {
                return Err(StorageError::Io(std::io::Error::other(format!(
                    "image {url_for_err} exceeds {max_image_bytes}-byte cap"
                ))));
            }
            remaining -= chunk.len();
            Ok(chunk)
        }
        Err(e) => Err(StorageError::Io(std::io::Error::other(format!(
            "stream chunk for {url_for_err}: {e}"
        )))),
    });
    let combined = prefix_stream.chain(rest_stream);
    let size_bytes = storage
        .put_stream(&key, Box::pin(combined))
        .await
        .with_context(|| format!("put_stream {key}"))?;
    Ok(StoredPage {
        page_number: img.page_number,
        storage_key: key,
        content_type: format!("image/{ext}"),
        size_bytes: size_bytes as i64,
    })
}

/// Persist the page rows + chapter `page_count` in one short transaction.
/// `(chapter_id, page_number)` is unique so re-runs are idempotent.
pub(crate) async fn persist_pages(
    db: &PgPool,
    chapter_id: Uuid,
    stored: &[StoredPage],
    enqueue_analysis: bool,
) -> anyhow::Result<()> {
    let mut tx = db.begin().await.context("open chapter sync tx")?;
    let mut page_ids: Vec<Uuid> = Vec::with_capacity(stored.len());
    let mut page_numbers: Vec<i32> = Vec::with_capacity(stored.len());
    for page in stored {
        let (id,): (Uuid,) = sqlx::query_as(
            "INSERT INTO pages (chapter_id, page_number, storage_key, content_type, size_bytes)
             VALUES ($1, $2, $3, $4, $5)
             ON CONFLICT (chapter_id, page_number) DO UPDATE
             SET storage_key = EXCLUDED.storage_key,
                 content_type = EXCLUDED.content_type,
                 size_bytes = EXCLUDED.size_bytes
             RETURNING id",
        )
        .bind(chapter_id)
        .bind(page.page_number)
        .bind(&page.storage_key)
        .bind(&page.content_type)
        .bind(page.size_bytes)
        .fetch_one(&mut *tx)
        .await
        .with_context(|| format!("insert page row {}", page.page_number))?;
        page_ids.push(id);
        page_numbers.push(page.page_number);
    }
    // Drop any rows left over from a prior, larger crawl of this chapter
    // (re-crawl that now yields fewer pages). Without this the stale rows —
    // and their stale size_bytes — would inflate the storage aggregates
    // and the page_count would disagree with the row count. Cascades to
    // collection_pages / page_tags by design (re-upload drops saved-page
    // references).
    sqlx::query("DELETE FROM pages WHERE chapter_id = $1 AND page_number <> ALL($2::int[])")
        .bind(chapter_id)
        .bind(&page_numbers)
        .execute(&mut *tx)
        .await
        .context("prune stale page rows")?;
    sqlx::query("UPDATE chapters SET page_count = $1 WHERE id = $2")
        .bind(stored.len() as i32)
        .bind(chapter_id)
        .execute(&mut *tx)
        .await
        .context("update page_count")?;
    tx.commit().await.context("commit chapter sync")?;

    // Enqueue AI content-analysis for the crawled pages once their rows
    // are committed. Best-effort: a failed enqueue is logged, never fatal
    // (the admin re-enqueue endpoint can backfill). Gated by the caller so
    // jobs don't pile up when the analysis worker is disabled.
    if enqueue_analysis {
        for page_id in page_ids {
            if let Err(e) =
                crate::repo::page_analysis::enqueue_for_page(db, page_id, false).await
            {
                tracing::warn!(%page_id, error = %e, "failed to enqueue page analysis after crawl");
            }
        }
    }
    Ok(())
}

/// Best-effort delete of partially-written page blobs after a chapter sync
/// fails, so a retry doesn't accumulate orphans. Errors are logged, not
/// raised — a leftover blob is harmless and a future reaper can sweep it.
pub(crate) async fn cleanup_orphans(storage: &dyn Storage, keys: &[String]) {
    for key in keys {
        if let Err(e) = storage.delete(key).await {
            tracing::warn!(
                %key,
                error = ?e,
                "failed to delete orphaned page blob after chapter sync failure"
            );
        }
    }
}

// Suppress unused-import warning for `session::registrable_domain`
// until the bin/crawler wiring lands in this branch and uses it
// through this module.
#[allow(dead_code)]
fn _keep_session_in_scope() {
    let _ = session::registrable_domain;
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::LocalStorage;

    #[test]
    fn guard_nav_url_rejects_private_and_loopback_targets() {
        // A chapter_sources row that resolves to / was crafted as an
        // internal target must be refused before Chromium navigates.
        for url in [
            "http://127.0.0.1:5432/",
            "http://169.254.169.254/latest/meta-data/",
            "http://10.0.0.1/chapter/1",
            "http://localhost:8080/",
            "file:///etc/passwd",
        ] {
            assert!(guard_nav_url(url).is_err(), "must reject {url}");
        }
    }

    #[test]
    fn guard_nav_url_allows_public_chapter_urls() {
        assert!(guard_nav_url("https://reader.example.com/chapter/42").is_ok());
        assert!(guard_nav_url("http://manga-host.test/c/1/p/2").is_ok());
    }

    #[test]
    fn tail_budget_uses_actual_prefix_length_not_constant() {
        // A single 16 KiB first chunk fills the 64-byte sniff window in one
        // drain, so the streaming tail budget must be `cap - 16KiB`, not
        // `cap - 64`. The constant-64 math previously admitted a further
        // ~full cap on top of the already-drained prefix (~2× overshoot).
        let cap = 20 * 1024;
        let prefix_len = 16 * 1024; // one real chunk, well over SNIFF_PREFIX_BYTES
        let remaining = remaining_after_prefix(cap, prefix_len);
        assert_eq!(remaining, cap - prefix_len);
        // Invariant: prefix + admitted tail never exceeds the cap.
        assert!(prefix_len + remaining <= cap);
        // And it's strictly tighter than the old constant-64 budget.
        assert!(remaining < cap - SNIFF_PREFIX_BYTES);
    }

    #[test]
    fn image_count_cap_rejects_only_over_cap_and_respects_disable() {
        // Under / at the cap: accepted.
        assert!(image_count_over_cap(0, 2000).is_none());
        assert!(image_count_over_cap(2000, 2000).is_none());
        // Over the cap: rejected, carrying the numbers for the error.
        let over = image_count_over_cap(2001, 2000).expect("over cap");
        assert_eq!(over.count, 2001);
        assert_eq!(over.cap, 2000);
        // `0` disables the cap entirely (unbounded), matching config contract.
        assert!(image_count_over_cap(1_000_000, 0).is_none());
    }

    #[test]
    fn tail_budget_saturates_when_prefix_hits_cap() {
        // Body shorter than the sniff window: prefix drained == cap, tail
        // budget is zero (no negative underflow).
        assert_eq!(remaining_after_prefix(50, 50), 0);
        assert_eq!(remaining_after_prefix(50, 64), 0);
    }

    #[tokio::test]
    async fn cleanup_orphans_deletes_written_keys() {
        let dir = tempfile::tempdir().unwrap();
        let storage = LocalStorage::new(dir.path());
        let keys = vec![
            "mangas/m/chapters/c/pages/0001.jpg".to_string(),
            "mangas/m/chapters/c/pages/0002.jpg".to_string(),
        ];
        for k in &keys {
            storage.put(k, b"\xff\xd8\xff\xe0 jpeg-ish").await.unwrap();
            assert!(storage.exists(k).await.unwrap());
        }
        cleanup_orphans(&storage, &keys).await;
        for k in &keys {
            assert!(!storage.exists(k).await.unwrap(), "{k} should be deleted");
        }
    }

    #[tokio::test]
    async fn cleanup_orphans_tolerates_missing_keys() {
        // A key that was never written (e.g. the put itself failed) must
        // not make cleanup error — it's best-effort.
        let dir = tempfile::tempdir().unwrap();
        let storage = LocalStorage::new(dir.path());
        cleanup_orphans(&storage, &["never/written.jpg".to_string()]).await;
    }

    #[sqlx::test(migrations = "./migrations")]
    async fn persist_pages_inserts_rows_and_sets_page_count(pool: PgPool) {
        let manga_id = Uuid::new_v4();
        let chapter_id = Uuid::new_v4();
        sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, 'T')")
            .bind(manga_id)
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)")
            .bind(chapter_id)
            .bind(manga_id)
            .execute(&pool)
            .await
            .unwrap();

        let stored = vec![
            StoredPage {
                page_number: 1,
                storage_key: "k/0001.jpg".into(),
                content_type: "image/jpeg".into(),
                size_bytes: 111,
            },
            StoredPage {
                page_number: 2,
                storage_key: "k/0002.jpg".into(),
                content_type: "image/jpeg".into(),
                size_bytes: 222,
            },
        ];
        persist_pages(&pool, chapter_id, &stored, false).await.unwrap();

        let page_count: i32 =
            sqlx::query_scalar("SELECT page_count FROM chapters WHERE id = $1")
                .bind(chapter_id)
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(page_count, 2);
        let rows: i64 =
            sqlx::query_scalar("SELECT COUNT(*) FROM pages WHERE chapter_id = $1")
                .bind(chapter_id)
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(rows, 2);
        // Sizes are persisted from the StoredPage so SUM(size_bytes) is
        // the chapter's storage usage.
        let total: i64 =
            sqlx::query_scalar("SELECT COALESCE(SUM(size_bytes),0)::bigint FROM pages WHERE chapter_id = $1")
                .bind(chapter_id)
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(total, 333);

        // Idempotent re-run (force refetch path): same rows, page_count stable.
        persist_pages(&pool, chapter_id, &stored, false).await.unwrap();
        let rows2: i64 =
            sqlx::query_scalar("SELECT COUNT(*) FROM pages WHERE chapter_id = $1")
                .bind(chapter_id)
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(rows2, 2, "re-run is idempotent via ON CONFLICT");
    }

    #[sqlx::test(migrations = "./migrations")]
    async fn persist_pages_prunes_stale_rows_on_smaller_recrawl(pool: PgPool) {
        // A re-crawl that yields fewer pages must drop the leftover
        // high-numbered rows so page_count, the row count, and the storage
        // aggregates stay consistent.
        let manga_id = Uuid::new_v4();
        let chapter_id = Uuid::new_v4();
        sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, 'T')")
            .bind(manga_id)
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)")
            .bind(chapter_id)
            .bind(manga_id)
            .execute(&pool)
            .await
            .unwrap();

        let three = vec![
            StoredPage { page_number: 1, storage_key: "k/1.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 10 },
            StoredPage { page_number: 2, storage_key: "k/2.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 20 },
            StoredPage { page_number: 3, storage_key: "k/3.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 30 },
        ];
        persist_pages(&pool, chapter_id, &three, false).await.unwrap();

        // Re-crawl now yields only 2 pages.
        let two = vec![
            StoredPage { page_number: 1, storage_key: "k/1.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 11 },
            StoredPage { page_number: 2, storage_key: "k/2.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 22 },
        ];
        persist_pages(&pool, chapter_id, &two, false).await.unwrap();

        let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pages WHERE chapter_id = $1")
            .bind(chapter_id)
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(rows, 2, "stale page 3 should be pruned");
        let page_count: i32 = sqlx::query_scalar("SELECT page_count FROM chapters WHERE id = $1")
            .bind(chapter_id)
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(page_count, 2);
        let total: i64 = sqlx::query_scalar(
            "SELECT COALESCE(SUM(size_bytes),0)::bigint FROM pages WHERE chapter_id = $1",
        )
        .bind(chapter_id)
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(total, 33, "size reflects only the 2 re-crawled pages");
    }

    #[sqlx::test(migrations = "./migrations")]
    async fn persist_pages_enqueues_analysis_only_when_flag_set(pool: PgPool) {
        let manga_id = Uuid::new_v4();
        let chapter_id = Uuid::new_v4();
        sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, 'T')")
            .bind(manga_id)
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)")
            .bind(chapter_id)
            .bind(manga_id)
            .execute(&pool)
            .await
            .unwrap();
        let stored = vec![StoredPage {
            page_number: 1,
            storage_key: "k/0001.jpg".into(),
            content_type: "image/jpeg".into(),
            size_bytes: 100,
        }];

        // Flag off: no analyze_page jobs.
        persist_pages(&pool, chapter_id, &stored, false).await.unwrap();
        let jobs_off: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(jobs_off, 0, "flag off must not enqueue analysis");

        // Flag on: one analyze_page job for the upserted page.
        persist_pages(&pool, chapter_id, &stored, true).await.unwrap();
        let jobs_on: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(jobs_on, 1, "flag on enqueues one job per persisted page");
    }

    #[test]
    fn parse_chapter_pages_skips_loader_and_sorts_by_id() {
        // Loader image, two real pages out of order, and one with no id.
        let html = r#"
          <html><body id="body"><a id="pic_container">
            <img class="loading" src="/images/ajax-loader2.gif">
            <img id="page2" class="page2" src="https://cdn/2.jpg">
            <img id="page1" class="page1" src="https://cdn/1.jpg">
            <img src="https://cdn/orphan.jpg">
            <img id="not-a-page" src="https://cdn/not-a-page.jpg">
          </a></body></html>
        "#;
        let pages = parse_chapter_pages(html).expect("parse");
        assert_eq!(pages.len(), 2);
        assert_eq!(pages[0].page_number, 1);
        assert_eq!(pages[0].url, "https://cdn/1.jpg");
        assert_eq!(pages[1].page_number, 2);
        assert_eq!(pages[1].url, "https://cdn/2.jpg");
    }

    #[test]
    fn parse_chapter_pages_drops_images_without_src() {
        let html = r#"
          <a id="pic_container">
            <img id="page1" src="">
            <img id="page2" src="https://cdn/2.jpg">
          </a>
        "#;
        let pages = parse_chapter_pages(html).expect("parse");
        assert_eq!(pages.len(), 1);
        assert_eq!(pages[0].page_number, 2);
    }

    #[test]
    fn parse_chapter_pages_handles_three_digit_page_ids() {
        let html = r#"
          <a id="pic_container">
            <img id="page126" src="https://cdn/126.jpg">
            <img id="page9" src="https://cdn/9.jpg">
            <img id="page50" src="https://cdn/50.jpg">
          </a>
        "#;
        let pages = parse_chapter_pages(html).expect("parse");
        assert_eq!(
            pages.iter().map(|p| p.page_number).collect::<Vec<_>>(),
            vec![9, 50, 126]
        );
    }

    #[test]
    fn parse_chapter_pages_returns_transient_when_container_missing() {
        // Reader doesn't render #logo, so the universal logo sentinel
        // can't be used here — a#pic_container is the reader-specific
        // marker. Broken-page response trips this.
        let html = "<html><body>\
            <p>we're sorry, the request file are not found.</p>\
            </body></html>";
        let err = parse_chapter_pages(html).expect_err("expected Transient");
        assert!(err.is_transient(), "got non-transient: {err}");
    }

    // --- fetch_chapter_html_with_recircuit -------------------------------

    const OK_HTML: &str = r#"<html><body><a id="pic_container"><img id="page1" src="x"/></a></body></html>"#;
    const UNAUTH_HTML: &str = r#"<html><body><header><div id="logo">x</div></header><main>please log in</main></body></html>"#;
    const TRANSIENT_HTML: &str = "<html><body><p>we're sorry, the request file are not found.</p></body></html>";

    #[tokio::test]
    async fn recircuit_loop_ok_first_attempt() {
        let mut recircuits = 0u32;
        let mut fetches = 0u32;
        let outcome = fetch_chapter_html_with_recircuit(
            || {
                fetches += 1;
                async { Ok(OK_HTML.to_string()) }
            },
            || {
                recircuits += 1;
                async {}
            },
            3,
            "https://example/c",
        )
        .await
        .expect("ok");
        assert!(matches!(outcome, ChapterFetchOutcome::Ok(_)));
        assert_eq!(fetches, 1);
        assert_eq!(recircuits, 0);
    }

    #[tokio::test]
    async fn recircuit_loop_unauth_with_single_attempt_returns_session_expired() {
        // max_attempts=1 = TOR disabled, fail-fast on first Unauthenticated.
        let mut recircuits = 0u32;
        let mut fetches = 0u32;
        let outcome = fetch_chapter_html_with_recircuit(
            || {
                fetches += 1;
                async { Ok(UNAUTH_HTML.to_string()) }
            },
            || {
                recircuits += 1;
                async {}
            },
            1,
            "https://example/c",
        )
        .await
        .expect("ok-result");
        assert!(matches!(outcome, ChapterFetchOutcome::SessionExpired));
        assert_eq!(fetches, 1);
        assert_eq!(recircuits, 0, "no recircuit when budget is 1 (TOR disabled)");
    }

    #[tokio::test]
    async fn recircuit_loop_unauth_then_ok_within_budget() {
        // max_attempts=3 = up to 3 fetches with 2 recircuits between.
        let mut recircuits = 0u32;
        let mut fetch_n = 0u32;
        let outcome = fetch_chapter_html_with_recircuit(
            || {
                fetch_n += 1;
                let n = fetch_n;
                async move {
                    if n == 1 {
                        Ok(UNAUTH_HTML.to_string())
                    } else {
                        Ok(OK_HTML.to_string())
                    }
                }
            },
            || {
                recircuits += 1;
                async {}
            },
            3,
            "https://example/c",
        )
        .await
        .expect("ok");
        assert!(matches!(outcome, ChapterFetchOutcome::Ok(_)));
        assert_eq!(fetch_n, 2);
        assert_eq!(recircuits, 1);
    }

    #[tokio::test]
    async fn recircuit_loop_unauth_exhausts_budget_returns_session_expired() {
        let mut recircuits = 0u32;
        let mut fetch_n = 0u32;
        let outcome = fetch_chapter_html_with_recircuit(
            || {
                fetch_n += 1;
                async { Ok(UNAUTH_HTML.to_string()) }
            },
            || {
                recircuits += 1;
                async {}
            },
            3,
            "https://example/c",
        )
        .await
        .expect("ok-result");
        assert!(matches!(outcome, ChapterFetchOutcome::SessionExpired));
        assert_eq!(fetch_n, 3, "max_attempts=3 → 3 fetches total");
        assert_eq!(recircuits, 2, "2 recircuits between 3 fetches");
    }

    #[tokio::test]
    async fn recircuit_loop_transient_then_ok_within_budget() {
        let mut recircuits = 0u32;
        let mut fetch_n = 0u32;
        let outcome = fetch_chapter_html_with_recircuit(
            || {
                fetch_n += 1;
                let n = fetch_n;
                async move {
                    if n < 3 {
                        Ok(TRANSIENT_HTML.to_string())
                    } else {
                        Ok(OK_HTML.to_string())
                    }
                }
            },
            || {
                recircuits += 1;
                async {}
            },
            3,
            "https://example/c",
        )
        .await
        .expect("ok");
        assert!(matches!(outcome, ChapterFetchOutcome::Ok(_)));
        assert_eq!(fetch_n, 3);
        assert_eq!(recircuits, 2);
    }

    #[tokio::test]
    async fn recircuit_loop_transient_exhausts_budget_returns_persistent() {
        let mut recircuits = 0u32;
        let mut fetch_n = 0u32;
        let outcome = fetch_chapter_html_with_recircuit(
            || {
                fetch_n += 1;
                async { Ok(TRANSIENT_HTML.to_string()) }
            },
            || {
                recircuits += 1;
                async {}
            },
            3,
            "https://example/c",
        )
        .await
        .expect("ok-result");
        assert!(matches!(outcome, ChapterFetchOutcome::PersistentTransient));
        assert_eq!(fetch_n, 3, "max_attempts=3 → 3 fetches total");
        assert_eq!(recircuits, 2, "2 recircuits between 3 fetches");
    }

    #[tokio::test]
    async fn recircuit_loop_mixed_transient_then_unauth_then_ok_shares_budget() {
        // Audit-prompted regression: outcomes share the attempt counter.
        // Sequence: Transient (attempt 1) → Unauth (attempt 2) → Ok (3).
        let mut recircuits = 0u32;
        let mut fetch_n = 0u32;
        let outcome = fetch_chapter_html_with_recircuit(
            || {
                fetch_n += 1;
                let n = fetch_n;
                async move {
                    match n {
                        1 => Ok(TRANSIENT_HTML.to_string()),
                        2 => Ok(UNAUTH_HTML.to_string()),
                        _ => Ok(OK_HTML.to_string()),
                    }
                }
            },
            || {
                recircuits += 1;
                async {}
            },
            3,
            "https://example/c",
        )
        .await
        .expect("ok");
        assert!(matches!(outcome, ChapterFetchOutcome::Ok(_)));
        assert_eq!(fetch_n, 3);
        assert_eq!(recircuits, 2);
    }

    #[tokio::test]
    async fn recircuit_loop_propagates_fetch_errors() {
        let mut fetch_n = 0u32;
        let err = fetch_chapter_html_with_recircuit(
            || {
                fetch_n += 1;
                async { Err(anyhow::anyhow!("nav timeout")) }
            },
            || async {},
            3,
            "https://example/c",
        )
        .await
        .expect_err("fetch error bubbles");
        assert_eq!(fetch_n, 1);
        assert!(format!("{err:#}").contains("nav timeout"));
    }
}