1110 lines
41 KiB
Rust
1110 lines
41 KiB
Rust
//! 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,
|
|
}
|
|
|
|
/// 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 = browser
|
|
.new_page(source_url)
|
|
.await
|
|
.with_context(|| format!("open chapter page {source_url}"))?;
|
|
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;
|
|
let html = page.content().await.context("read chapter html")?;
|
|
page.close().await.ok();
|
|
Ok(html)
|
|
}
|
|
|
|
/// 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,
|
|
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, 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,
|
|
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}");
|
|
}
|
|
|
|
// 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;
|
|
|
|
/// 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.
|
|
let prefix_stream = futures_util::stream::once(async move {
|
|
Ok::<bytes::Bytes, StorageError>(prefix)
|
|
});
|
|
let prefix_len = SNIFF_PREFIX_BYTES.min(max_image_bytes);
|
|
let mut remaining = max_image_bytes.saturating_sub(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());
|
|
}
|
|
|
|
#[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"));
|
|
}
|
|
}
|