Admin-only crawler dashboard backed by an SSE live-status stream,
coordinated browser restart, runtime PHPSESSID refresh, dead-letter
requeue, and a batch of reliability fixes. Closes everything from
the two-pass audit (10 commits' worth) and bumps 0.52.0 -> 0.55.0.
Backend:
- New /admin/crawler/* surface (cookie-auth, RequireAdmin) split
into status / control / dead_jobs / backlog modules. SSE stream
composes in-memory status with DB-derived queue counts, memoizes
the counts for 1s and debounces watch pokes for 250ms (~10x QPS
reduction per subscriber). One-shot GET /admin/crawler shares the
same compose path.
- POST /admin/crawler/run gated by manual_pass_lock try_lock_owned
(409 Conflict on overlapping click); browser restart goes through
the coordinated_restart gate (drain + relaunch + auto-clear of the
sticky session_expired flag on Ok).
- Runtime PHPSESSID refresh via SessionController (allow-list
validation, never logged, audit row carries SHA-256 fingerprint).
Storage layer is repo::crawler::runtime_session_{load,persist}.
- Dead-letter requeue with four scopes (all/manga/chapter/job);
scope=all requires confirm:true; DISTINCT ON dedup keeps the
partial unique index from rejecting requeues for chapters with
multiple dead rows. SQL is four &'static str constants per scope.
- StatusHandle + ChapterGuard / CoverGuard RAII model survives
panics; last-writer-wins on cover so concurrent dispatches don't
clobber each other's slot. Pure functions (should_stop /
should_mark_clean_exit / should_abort_pass) with named regression
tests.
- Reliability bundle: per-lease heartbeat, jitter on retries,
per-job timeout, circuit breaker on consecutive failures, BrowserManager
coordinated restart gate, request fingerprint changes.
- Streaming page download: Storage::put_stream trait method,
LocalStorage impl atomic via temp + fsync + UUID-suffixed rename.
Pages stream through with peak memory ~one HTTP chunk + 64-byte
sniff prefix instead of one full image per dispatch.
- New partial indexes (migration 0022): mangas_missing_cover_idx
and crawler_jobs_dead_idx, both ordered by updated_at DESC to
match the dashboard's LIMIT/OFFSET reads.
- Security hardening: admin_csrf_guard (Origin/Referer allowlist
on /admin/* mutations, opt-in via ADMIN_ALLOWED_ORIGINS),
admin_no_store_guard (Cache-Control: no-store on admin
responses), audit rows carry per-scope target_id.
Frontend:
- /admin/crawler page decomposed into lib/components/crawler/
(11 components: ProgressBar, SearchBar, CrawlerHero,
CrawlerControls, ActiveChaptersCard, ActiveJobsTable,
MissingCoversTable, DeadJobsTable, RestartConfirmModal,
RequeueAllConfirmModal, SessionModal). Page is 532 LOC of
orchestration; each component 22-148 LOC.
- EventSource lifecycle wired to visibilitychange / pagehide /
pageshow (BFCache); after 5 consecutive errors probes the status
endpoint so a 401 routes through the global on401Hook instead of
infinite silent reconnects.
- Backlog $effect refetches debounced 500ms with per-loader
AbortControllers; refresh after a control action only runs when
the SSE stream is dead.
- Inline requeue button on /admin/mangas patches the affected row's
sync_state locally (no full chapter-list refetch); proper
aria-label. Requeue-all gets its own confirm modal; both confirm
modals autofocus Cancel.
- SvelteKit reverse proxy bypasses its 5-minute AbortController
for Accept: text/event-stream; pure shouldBypassProxyTimeout
helper covered by unit tests.
Config / docs:
- New env vars (.env.example): ADMIN_ALLOWED_ORIGINS,
CRAWLER_JOB_TIMEOUT_SECS, CRAWLER_METADATA_MAX_CONSECUTIVE_FAILURES,
CRAWLER_BROWSER_RESTART_THRESHOLD.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
816 lines
32 KiB
Rust
816 lines
32 KiB
Rust
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
|
|
|
|
use anyhow::Context;
|
|
use async_trait::async_trait;
|
|
use axum::extract::{DefaultBodyLimit, FromRequestParts, Request, State};
|
|
use axum::http::{HeaderName, HeaderValue, Method};
|
|
use axum::middleware::{self, Next};
|
|
use axum::response::Response;
|
|
use axum::Router;
|
|
use sqlx::postgres::PgPoolOptions;
|
|
use sqlx::PgPool;
|
|
use tokio_util::sync::CancellationToken;
|
|
use tower_http::cors::{AllowOrigin, CorsLayer};
|
|
use tower_http::trace::TraceLayer;
|
|
|
|
use crate::auth::extractor::CurrentUser;
|
|
use crate::auth::rate_limit::AuthRateLimiter;
|
|
use crate::error::AppError;
|
|
use crate::config::{AuthConfig, Config, CrawlerConfig, UploadConfig};
|
|
use crate::crawler::browser_manager::{self, BrowserManager};
|
|
use crate::crawler::content::{self, SyncOutcome};
|
|
use crate::crawler::daemon::{self, ChapterDispatcher, DaemonConfig, MetadataPass};
|
|
use crate::crawler::jobs::JobPayload;
|
|
use crate::crawler::pipeline::{self, MetadataStats};
|
|
use crate::crawler::rate_limit::HostRateLimiters;
|
|
use crate::crawler::resync::{RealResyncService, ResyncService};
|
|
use crate::crawler::safety::DownloadAllowlist;
|
|
use crate::crawler::session;
|
|
use crate::repo;
|
|
use crate::storage::{LocalStorage, Storage};
|
|
|
|
#[derive(Clone)]
|
|
pub struct AppState {
|
|
pub db: PgPool,
|
|
pub storage: Arc<dyn Storage>,
|
|
pub auth: AuthConfig,
|
|
pub upload: UploadConfig,
|
|
/// Shared rate limiter guarding the `/auth/*` mutation endpoints.
|
|
/// One instance per AppState so tests stay isolated across the
|
|
/// same process.
|
|
pub auth_limiter: Arc<AuthRateLimiter>,
|
|
/// Admin-triggered force resync. `None` when the crawler daemon
|
|
/// is disabled (`CRAWLER_DAEMON=false`); admin handlers gate on
|
|
/// `.is_some()` and return 503 otherwise. Set by [`build`] from the
|
|
/// same wiring that builds the daemon's chapter dispatcher, so a
|
|
/// force resync uses the daemon's BrowserManager + rate limiters.
|
|
pub resync: Option<Arc<dyn ResyncService>>,
|
|
/// Crawler observability + control handle (live status, coordinated
|
|
/// browser restart, runtime session, manual run). `None` when the
|
|
/// daemon is disabled; admin handlers gate on `.is_some()` → 503.
|
|
pub crawler: Option<Arc<CrawlerControl>>,
|
|
/// Browser origins permitted to issue mutating requests to
|
|
/// `/api/v1/admin/*`. See [`crate::config::Config::admin_allowed_origins`]
|
|
/// for the policy. Cloned per-request into the CSRF middleware; the
|
|
/// `Arc` keeps the clone cheap. Empty list → check is skipped
|
|
/// (operator opt-out documented in `.env.example`).
|
|
pub admin_allowed_origins: Arc<Vec<String>>,
|
|
}
|
|
|
|
/// Shared handle the admin crawler endpoints use to observe and control
|
|
/// the running daemon. Bundled so the handlers take one optional field on
|
|
/// `AppState` rather than many.
|
|
pub struct CrawlerControl {
|
|
pub browser_manager: Arc<BrowserManager>,
|
|
pub session: Arc<crate::crawler::session_control::SessionController>,
|
|
pub status: crate::crawler::status::StatusHandle,
|
|
/// Used by the "run metadata pass now" endpoint; `None` when no
|
|
/// `CRAWLER_START_URL` is configured (cron disabled).
|
|
pub metadata_pass: Option<Arc<dyn MetadataPass>>,
|
|
/// Drain budget for a manually-triggered coordinated browser restart.
|
|
pub drain_deadline: std::time::Duration,
|
|
/// Held for the duration of a `/admin/crawler/run` pass so a second
|
|
/// click returns 409 instead of fanning N overlapping metadata passes
|
|
/// onto the single browser lease. The daemon's daily cron does NOT
|
|
/// take this lock — cron and operator-triggered are different
|
|
/// trust modes (cron is single-fire by definition; the lock only
|
|
/// dedups operator clicks). `Arc` so the spawned task can hold an
|
|
/// owned guard past the request boundary.
|
|
pub manual_pass_lock: Arc<tokio::sync::Mutex<()>>,
|
|
}
|
|
|
|
/// Bundle returned by [`build`]. The router is what `axum::serve` consumes;
|
|
/// the daemon (when enabled) outlives the HTTP server and is awaited via
|
|
/// [`AppHandle::shutdown`] after the listener has finished gracefully.
|
|
pub struct AppHandle {
|
|
pub router: Router,
|
|
pub daemon: Option<daemon::DaemonHandle>,
|
|
}
|
|
|
|
impl AppHandle {
|
|
pub async fn shutdown(self) {
|
|
if let Some(d) = self.daemon {
|
|
d.shutdown().await;
|
|
}
|
|
}
|
|
}
|
|
|
|
pub async fn build(config: Config) -> anyhow::Result<AppHandle> {
|
|
let db = PgPoolOptions::new()
|
|
.max_connections(10)
|
|
.connect(&config.database_url)
|
|
.await?;
|
|
sqlx::migrate!("./migrations").run(&db).await?;
|
|
|
|
if let Some((username, password)) = config.admin_bootstrap.as_ref() {
|
|
repo::user::bootstrap_admin(&db, username, password)
|
|
.await
|
|
.context("bootstrap_admin from ADMIN_USERNAME/ADMIN_PASSWORD env")?;
|
|
tracing::info!(admin_username = %username, "admin bootstrap ensured");
|
|
}
|
|
|
|
let storage: Arc<dyn Storage> = Arc::new(LocalStorage::new(config.storage_dir.clone()));
|
|
|
|
let (daemon, resync, crawler) = if config.crawler.daemon_enabled {
|
|
let spawned = spawn_crawler_daemon(db.clone(), Arc::clone(&storage), &config.crawler).await?;
|
|
(Some(spawned.handle), Some(spawned.resync), Some(spawned.crawler))
|
|
} else {
|
|
tracing::info!("crawler daemon disabled (CRAWLER_DAEMON=false)");
|
|
(None, None, None)
|
|
};
|
|
|
|
let auth_limiter = Arc::new(AuthRateLimiter::new(config.auth.rate_limit));
|
|
let state = AppState {
|
|
db,
|
|
storage,
|
|
auth: config.auth.clone(),
|
|
upload: config.upload.clone(),
|
|
auth_limiter,
|
|
resync,
|
|
crawler,
|
|
admin_allowed_origins: Arc::new(config.admin_allowed_origins.clone()),
|
|
};
|
|
let router = router(state).layer(cors_layer(&config.cors_allowed_origins));
|
|
Ok(AppHandle { router, daemon })
|
|
}
|
|
|
|
/// Bundle returned by [`spawn_crawler_daemon`]. The handle owns the
|
|
/// daemon's tasks; `resync` is the operator-trigger service shared with
|
|
/// `AppState` so admin endpoints can call into the same browser /
|
|
/// rate-limit machinery.
|
|
struct SpawnedDaemon {
|
|
handle: daemon::DaemonHandle,
|
|
resync: Arc<dyn ResyncService>,
|
|
crawler: Arc<CrawlerControl>,
|
|
}
|
|
|
|
async fn spawn_crawler_daemon(
|
|
db: PgPool,
|
|
storage: Arc<dyn Storage>,
|
|
cfg: &CrawlerConfig,
|
|
) -> anyhow::Result<SpawnedDaemon> {
|
|
// Reqwest client with a shared cookie jar so CDN image fetches include
|
|
// PHPSESSID. The same `Arc<Jar>` is held by the SessionController, so a
|
|
// runtime session refresh rewrites it in place. Initial value: a
|
|
// persisted runtime session (survives restart) takes precedence over
|
|
// CRAWLER_PHPSESSID env.
|
|
let cookie_jar = Arc::new(reqwest::cookie::Jar::default());
|
|
let initial_sid = crate::crawler::session_control::SessionController::load_persisted(&db)
|
|
.await
|
|
.or_else(|| cfg.phpsessid.clone());
|
|
if let (Some(sid), Some(domain), Some(start_url)) =
|
|
(&initial_sid, &cfg.cookie_domain, &cfg.start_url)
|
|
{
|
|
let cookie_str = format!("PHPSESSID={sid}; Domain={domain}; Path=/");
|
|
let seed_url = reqwest::Url::parse(start_url)
|
|
.context("parse CRAWLER_START_URL for cookie seed")?;
|
|
cookie_jar.add_cookie_str(&cookie_str, &seed_url);
|
|
}
|
|
let mut http_builder = reqwest::Client::builder()
|
|
.timeout(std::time::Duration::from_secs(30))
|
|
.no_proxy()
|
|
.cookie_provider(Arc::clone(&cookie_jar));
|
|
if let Some(ua) = &cfg.user_agent {
|
|
http_builder = http_builder.user_agent(ua);
|
|
}
|
|
if let Some(proxy) = &cfg.proxy {
|
|
http_builder = http_builder
|
|
.proxy(reqwest::Proxy::all(proxy).with_context(|| format!("parse proxy: {proxy}"))?);
|
|
}
|
|
let http = http_builder.build().context("build crawler reqwest")?;
|
|
|
|
let mut rate = HostRateLimiters::new(std::time::Duration::from_millis(cfg.rate_ms));
|
|
if let Some(host) = &cfg.cdn_host {
|
|
rate = rate.with_override(host, std::time::Duration::from_millis(cfg.cdn_rate_ms));
|
|
}
|
|
let rate = Arc::new(rate);
|
|
|
|
let tor = crate::crawler::tor::TorController::from_parts(
|
|
cfg.tor_control_url.as_deref(),
|
|
cfg.tor_control_password.as_deref(),
|
|
cfg.tor_control_cookie_path.as_deref(),
|
|
)
|
|
.context("build TorController from CRAWLER_TOR_CONTROL_* env")?
|
|
.map(Arc::new);
|
|
if let Some(t) = &tor {
|
|
tracing::info!(?t, "TOR control configured; transient pages will trigger NEWNYM");
|
|
}
|
|
let tor_recircuit_max = cfg.tor_recircuit_max_attempts;
|
|
|
|
// Session controller + sticky session-expired flag. Created before the
|
|
// browser so the on_launch hook can read the *current* session value
|
|
// (rather than a value captured at startup), and so a runtime refresh
|
|
// updates the cookie everywhere.
|
|
let session_expired = Arc::new(AtomicBool::new(false));
|
|
let session_controller = crate::crawler::session_control::SessionController::new(
|
|
initial_sid,
|
|
Arc::clone(&cookie_jar),
|
|
cfg.cookie_domain.clone(),
|
|
cfg.start_url.clone(),
|
|
db.clone(),
|
|
Arc::clone(&session_expired),
|
|
);
|
|
|
|
// Live status surface, sized to the worker count.
|
|
let status = crate::crawler::status::StatusHandle::new(cfg.chapter_workers);
|
|
|
|
// Browser manager. on_launch re-injects PHPSESSID on every fresh
|
|
// chromium spawn so an idle teardown followed by re-launch stays
|
|
// authenticated without operator action.
|
|
let mut launch_opts = cfg.browser.clone();
|
|
if let Some(proxy) = &cfg.proxy {
|
|
let chromium_proxy = crate::crawler::url_utils::chromium_proxy_arg(proxy);
|
|
launch_opts.extra_args.push(format!("--proxy-server={chromium_proxy}"));
|
|
}
|
|
let on_launch = match (&cfg.cookie_domain, &cfg.start_url) {
|
|
(Some(domain), Some(start_url)) => {
|
|
let domain = domain.clone();
|
|
let start_url = start_url.clone();
|
|
let tor_for_launch = tor.as_ref().map(Arc::clone);
|
|
let sc = Arc::clone(&session_controller);
|
|
let on_launch: browser_manager::OnLaunch = Arc::new(move |browser| {
|
|
let domain = domain.clone();
|
|
let start_url = start_url.clone();
|
|
let tor_for_launch = tor_for_launch.as_ref().map(Arc::clone);
|
|
let sc = Arc::clone(&sc);
|
|
Box::pin(async move {
|
|
// Read the *current* session each launch so a runtime
|
|
// refresh is picked up on the next (re)launch. No session
|
|
// configured → run unauthenticated (metadata needs no auth).
|
|
let Some(sid) = sc.current().await else {
|
|
tracing::info!("on_launch: no session set — skipping inject + probe");
|
|
return Ok(());
|
|
};
|
|
session::inject_phpsessid(&browser, &sid, &domain)
|
|
.await
|
|
.context("on_launch: inject_phpsessid")?;
|
|
session::verify_session_with_recircuit(
|
|
&browser,
|
|
&start_url,
|
|
tor_for_launch.as_deref(),
|
|
tor_recircuit_max,
|
|
)
|
|
.await
|
|
.context("on_launch: verify_session")?;
|
|
Ok(())
|
|
})
|
|
});
|
|
on_launch
|
|
}
|
|
_ => browser_manager::noop_on_launch(),
|
|
};
|
|
let browser_manager = BrowserManager::new(launch_opts, cfg.idle_timeout, on_launch);
|
|
|
|
let metadata_pass: Option<Arc<dyn MetadataPass>> = cfg.start_url.as_ref().map(|url| {
|
|
let m: Arc<dyn MetadataPass> = Arc::new(RealMetadataPass {
|
|
browser_manager: Arc::clone(&browser_manager),
|
|
db: db.clone(),
|
|
storage: Arc::clone(&storage),
|
|
http: http.clone(),
|
|
rate: Arc::clone(&rate),
|
|
start_url: url.clone(),
|
|
manga_limit: cfg.manga_limit,
|
|
download_allowlist: cfg.download_allowlist.clone(),
|
|
max_image_bytes: cfg.max_image_bytes,
|
|
metadata_max_consecutive_failures: cfg.metadata_max_consecutive_failures,
|
|
status: status.clone(),
|
|
tor: tor.as_ref().map(Arc::clone),
|
|
});
|
|
m
|
|
});
|
|
|
|
let dispatcher: Arc<dyn ChapterDispatcher> = Arc::new(RealChapterDispatcher {
|
|
browser_manager: Arc::clone(&browser_manager),
|
|
db: db.clone(),
|
|
storage: Arc::clone(&storage),
|
|
http: http.clone(),
|
|
rate: Arc::clone(&rate),
|
|
download_allowlist: cfg.download_allowlist.clone(),
|
|
max_image_bytes: cfg.max_image_bytes,
|
|
transient_failures: Arc::new(AtomicU32::new(0)),
|
|
restart_threshold: cfg.browser_restart_threshold,
|
|
drain_deadline: cfg.job_timeout,
|
|
status: status.clone(),
|
|
tor: tor.as_ref().map(Arc::clone),
|
|
});
|
|
|
|
let resync: Arc<dyn ResyncService> = Arc::new(RealResyncService {
|
|
browser_manager: Arc::clone(&browser_manager),
|
|
db: db.clone(),
|
|
storage: Arc::clone(&storage),
|
|
http,
|
|
rate: Arc::clone(&rate),
|
|
download_allowlist: cfg.download_allowlist.clone(),
|
|
max_image_bytes: cfg.max_image_bytes,
|
|
tor: tor.as_ref().map(Arc::clone),
|
|
});
|
|
|
|
// Shared cancellation: daemon shutdown cancels the BrowserManager's
|
|
// idle reaper too. Reaper itself is added to the daemon's extra_tasks
|
|
// so DaemonHandle::shutdown awaits its completion.
|
|
let cancel = CancellationToken::new();
|
|
let reaper_task = browser_manager::spawn_idle_reaper(
|
|
Arc::clone(&browser_manager),
|
|
cancel.clone(),
|
|
);
|
|
// Also close the browser explicitly on shutdown so we don't rely on
|
|
// kill-on-drop when other Arc<Browser> holders may still exist.
|
|
let shutdown_task = {
|
|
let cancel = cancel.clone();
|
|
let mgr = Arc::clone(&browser_manager);
|
|
tokio::spawn(async move {
|
|
cancel.cancelled().await;
|
|
mgr.shutdown().await;
|
|
})
|
|
};
|
|
|
|
let daemon_handle = daemon::spawn(
|
|
db,
|
|
cancel,
|
|
DaemonConfig {
|
|
metadata_pass: metadata_pass.clone(),
|
|
dispatcher,
|
|
chapter_workers: cfg.chapter_workers,
|
|
daily_at: cfg.daily_at,
|
|
tz: cfg.tz,
|
|
retention_days: cfg.retention_days,
|
|
session_expired,
|
|
status: status.clone(),
|
|
job_timeout: cfg.job_timeout,
|
|
extra_tasks: vec![reaper_task, shutdown_task],
|
|
},
|
|
);
|
|
|
|
let crawler = Arc::new(CrawlerControl {
|
|
browser_manager: Arc::clone(&browser_manager),
|
|
session: session_controller,
|
|
status,
|
|
metadata_pass,
|
|
drain_deadline: cfg.job_timeout,
|
|
manual_pass_lock: Arc::new(tokio::sync::Mutex::new(())),
|
|
});
|
|
|
|
Ok(SpawnedDaemon {
|
|
handle: daemon_handle,
|
|
resync,
|
|
crawler,
|
|
})
|
|
}
|
|
|
|
// Real impls of the daemon traits, owning the browser manager + I/O. Kept
|
|
// in app.rs because they need the same builder-side env wiring that
|
|
// AppState gets — the daemon module itself stays free of reqwest / storage
|
|
// details so its tests don't pull them in.
|
|
|
|
struct RealMetadataPass {
|
|
browser_manager: Arc<BrowserManager>,
|
|
db: PgPool,
|
|
storage: Arc<dyn Storage>,
|
|
http: reqwest::Client,
|
|
rate: Arc<HostRateLimiters>,
|
|
start_url: String,
|
|
manga_limit: usize,
|
|
download_allowlist: DownloadAllowlist,
|
|
max_image_bytes: usize,
|
|
metadata_max_consecutive_failures: u32,
|
|
status: crate::crawler::status::StatusHandle,
|
|
tor: Option<Arc<crate::crawler::tor::TorController>>,
|
|
}
|
|
|
|
#[async_trait]
|
|
impl MetadataPass for RealMetadataPass {
|
|
async fn run(&self) -> anyhow::Result<MetadataStats> {
|
|
let result = pipeline::run_metadata_pass(
|
|
&self.browser_manager,
|
|
&self.db,
|
|
self.storage.as_ref(),
|
|
&self.http,
|
|
&self.rate,
|
|
&self.start_url,
|
|
self.manga_limit,
|
|
false,
|
|
&self.download_allowlist,
|
|
self.max_image_bytes,
|
|
self.metadata_max_consecutive_failures,
|
|
Some(&self.status),
|
|
self.tor.as_deref(),
|
|
)
|
|
.await;
|
|
if let Err(e) = &result {
|
|
if crate::crawler::nav::anyhow_looks_browser_dead(e) {
|
|
self.browser_manager.invalidate().await;
|
|
}
|
|
}
|
|
// Cover backfill follows the metadata pass even when the pass
|
|
// errored — the early-stop walk can complete its work and bail
|
|
// late, and a transient browser failure shouldn't cancel the
|
|
// residual cover backlog. The backfill has its own per-call cap
|
|
// so a runaway error stream can't monopolise the tick. It sets the
|
|
// CoverBackfill{index,total} phase + current_cover per entry.
|
|
match pipeline::backfill_missing_covers(
|
|
&self.browser_manager,
|
|
&self.db,
|
|
self.storage.as_ref(),
|
|
&self.http,
|
|
&self.rate,
|
|
pipeline::COVER_BACKFILL_DEFAULT_MAX,
|
|
&self.download_allowlist,
|
|
self.max_image_bytes,
|
|
Some(&self.status),
|
|
self.tor.as_deref(),
|
|
)
|
|
.await
|
|
{
|
|
Ok(stats) => {
|
|
if stats.considered > 0 {
|
|
tracing::info!(?stats, "cover backfill complete");
|
|
}
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "cover backfill failed");
|
|
if crate::crawler::nav::anyhow_looks_browser_dead(&e) {
|
|
self.browser_manager.invalidate().await;
|
|
}
|
|
}
|
|
}
|
|
result
|
|
}
|
|
}
|
|
|
|
struct RealChapterDispatcher {
|
|
browser_manager: Arc<BrowserManager>,
|
|
db: PgPool,
|
|
storage: Arc<dyn Storage>,
|
|
http: reqwest::Client,
|
|
rate: Arc<HostRateLimiters>,
|
|
download_allowlist: DownloadAllowlist,
|
|
max_image_bytes: usize,
|
|
/// Consecutive transient chapter failures; resets on any success.
|
|
/// Drives the automatic coordinated browser restart.
|
|
transient_failures: Arc<std::sync::atomic::AtomicU32>,
|
|
/// Consecutive-failure count that triggers an auto restart.
|
|
restart_threshold: u32,
|
|
/// How long a coordinated restart waits for in-flight leases to drain.
|
|
drain_deadline: std::time::Duration,
|
|
/// Live status surface — the dispatcher registers each chapter it
|
|
/// crawls (with a realtime page count) here.
|
|
status: crate::crawler::status::StatusHandle,
|
|
tor: Option<Arc<crate::crawler::tor::TorController>>,
|
|
}
|
|
|
|
#[async_trait]
|
|
impl ChapterDispatcher for RealChapterDispatcher {
|
|
async fn dispatch(&self, payload: JobPayload) -> anyhow::Result<SyncOutcome> {
|
|
match payload {
|
|
JobPayload::SyncChapterContent {
|
|
source_id: _,
|
|
chapter_id,
|
|
source_chapter_key: _,
|
|
} => {
|
|
let row = repo::chapter::dispatch_target(&self.db, chapter_id)
|
|
.await
|
|
.context("look up chapter for dispatch")?;
|
|
let Some((manga_id, source_url, manga_title, chapter_number)) = row else {
|
|
// Chapter (or its source row) is gone — ack done.
|
|
return Ok(SyncOutcome::Skipped);
|
|
};
|
|
// Register the chapter as crawling now (live status). The
|
|
// guard removes it on every exit path — success, panic, or
|
|
// the worker's outer-timeout drop.
|
|
let _active = self.status.begin_chapter(crate::crawler::status::ActiveChapter {
|
|
manga_id,
|
|
manga_title,
|
|
chapter_id,
|
|
chapter_number,
|
|
pages_done: 0,
|
|
pages_total: None,
|
|
});
|
|
let lease = self.browser_manager.acquire().await?;
|
|
let result = content::sync_chapter_content(
|
|
&lease,
|
|
&self.db,
|
|
self.storage.as_ref(),
|
|
&self.http,
|
|
&self.rate,
|
|
chapter_id,
|
|
manga_id,
|
|
&source_url,
|
|
false,
|
|
&self.download_allowlist,
|
|
self.max_image_bytes,
|
|
self.tor.as_deref(),
|
|
Some(&self.status),
|
|
)
|
|
.await;
|
|
drop(lease);
|
|
match result {
|
|
Ok(outcome) => {
|
|
// Any successful dispatch (including a clean Skipped)
|
|
// means the browser is healthy — reset the streak.
|
|
self.transient_failures.store(0, Ordering::Release);
|
|
Ok(outcome)
|
|
}
|
|
Err(e) => {
|
|
let streak = self.transient_failures.fetch_add(1, Ordering::AcqRel) + 1;
|
|
if crate::crawler::nav::anyhow_looks_browser_dead(&e) {
|
|
// Hard browser-dead: lazy invalidate (next acquire
|
|
// relaunches). Reset the streak — we're recovering.
|
|
self.browser_manager.invalidate().await;
|
|
self.transient_failures.store(0, Ordering::Release);
|
|
} else if self.restart_threshold > 0 && streak >= self.restart_threshold {
|
|
// Persistent transients that TOR recircuit couldn't
|
|
// fix — proactively restart Chromium.
|
|
tracing::warn!(
|
|
streak,
|
|
threshold = self.restart_threshold,
|
|
"auto browser restart: consecutive transient chapter failures"
|
|
);
|
|
let _ = self
|
|
.browser_manager
|
|
.coordinated_restart(self.drain_deadline)
|
|
.await;
|
|
self.transient_failures.store(0, Ordering::Release);
|
|
}
|
|
Err(e)
|
|
}
|
|
}
|
|
}
|
|
// Other payload kinds aren't dispatched by this daemon yet —
|
|
// SyncManga / SyncChapterList are handled inline by the cron's
|
|
// metadata pass.
|
|
_ => Ok(SyncOutcome::Skipped),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Build a router from a pre-assembled state. Used by integration tests
|
|
/// so they can swap in a test DB pool and a `tempfile`-backed storage.
|
|
pub fn router(state: AppState) -> Router {
|
|
let max_request_bytes = state.upload.max_request_bytes;
|
|
Router::new()
|
|
.nest("/api/v1", crate::api::routes())
|
|
.layer(middleware::from_fn(admin_no_store_guard))
|
|
.layer(middleware::from_fn_with_state(
|
|
state.clone(),
|
|
admin_csrf_guard,
|
|
))
|
|
.layer(middleware::from_fn_with_state(
|
|
state.clone(),
|
|
private_mode_guard,
|
|
))
|
|
.layer(DefaultBodyLimit::max(max_request_bytes))
|
|
.with_state(state)
|
|
.layer(TraceLayer::new_for_http())
|
|
}
|
|
|
|
/// Path prefix the admin-only middlewares scope themselves to. The router
|
|
/// already nests `/api/v1`, so callers see `/api/v1/admin/...`.
|
|
const ADMIN_PATH_PREFIX: &str = "/api/v1/admin/";
|
|
|
|
/// CSRF defence for cookie-authenticated admin mutations. The session
|
|
/// cookie is `SameSite=Lax`, which still permits top-level form-POSTs
|
|
/// from a malicious page — this middleware rejects such requests by
|
|
/// comparing the request's `Origin` (with `Referer` as fallback) against
|
|
/// the configured allowlist. Safe methods (`GET`/`HEAD`/`OPTIONS`) are
|
|
/// always allowed. Requests with neither `Origin` nor `Referer` are
|
|
/// allowed (non-browser callers like curl can't be a CSRF vector). When
|
|
/// the allowlist is empty the check is skipped entirely (operator
|
|
/// opt-out — documented in `.env.example`).
|
|
async fn admin_csrf_guard(
|
|
State(state): State<AppState>,
|
|
req: Request,
|
|
next: Next,
|
|
) -> Result<Response, AppError> {
|
|
if !req.uri().path().starts_with(ADMIN_PATH_PREFIX) {
|
|
return Ok(next.run(req).await);
|
|
}
|
|
if matches!(
|
|
*req.method(),
|
|
Method::GET | Method::HEAD | Method::OPTIONS
|
|
) {
|
|
return Ok(next.run(req).await);
|
|
}
|
|
if state.admin_allowed_origins.is_empty() {
|
|
return Ok(next.run(req).await);
|
|
}
|
|
let headers = req.headers();
|
|
let origin = headers.get("origin").and_then(|v| v.to_str().ok());
|
|
let referer = headers.get("referer").and_then(|v| v.to_str().ok());
|
|
// No Origin AND no Referer → server-to-server / curl / extension.
|
|
// Browsers always send one or the other on a cross-site POST.
|
|
let Some(candidate) = origin.or(referer) else {
|
|
return Ok(next.run(req).await);
|
|
};
|
|
if origin_in_allowlist(candidate, &state.admin_allowed_origins) {
|
|
return Ok(next.run(req).await);
|
|
}
|
|
tracing::warn!(
|
|
candidate = %truncate_for_log(candidate, 64),
|
|
path = %req.uri().path(),
|
|
"admin CSRF: rejecting mutation with disallowed origin"
|
|
);
|
|
Err(AppError::Forbidden)
|
|
}
|
|
|
|
/// Match `candidate` (an `Origin` value, or a `Referer` URL whose
|
|
/// origin we'll extract) against `allowed`. `Origin` is `scheme://host[:port]`
|
|
/// with no path; `Referer` is a full URL — compare by parsing both and
|
|
/// matching scheme + host + port.
|
|
fn origin_in_allowlist(candidate: &str, allowed: &[String]) -> bool {
|
|
let cand_origin = parse_origin(candidate);
|
|
let Some(cand) = cand_origin else { return false };
|
|
allowed
|
|
.iter()
|
|
.filter_map(|a| parse_origin(a))
|
|
.any(|a| a == cand)
|
|
}
|
|
|
|
/// Extract the origin (`scheme://host[:port]`) from an `Origin` header
|
|
/// value or a `Referer` URL. Returns `None` when the input doesn't parse
|
|
/// as a URL with a host.
|
|
fn parse_origin(raw: &str) -> Option<String> {
|
|
let url = reqwest::Url::parse(raw.trim()).ok()?;
|
|
let host = url.host_str()?;
|
|
let scheme = url.scheme();
|
|
let port_str = match (url.port(), scheme) {
|
|
(Some(p), "http") if p == 80 => String::new(),
|
|
(Some(p), "https") if p == 443 => String::new(),
|
|
(Some(p), _) => format!(":{p}"),
|
|
(None, _) => String::new(),
|
|
};
|
|
Some(format!("{scheme}://{host}{port_str}"))
|
|
}
|
|
|
|
fn truncate_for_log(s: &str, max: usize) -> &str {
|
|
let end = s
|
|
.char_indices()
|
|
.take(max)
|
|
.last()
|
|
.map(|(i, c)| i + c.len_utf8())
|
|
.unwrap_or(0);
|
|
&s[..end.min(s.len())]
|
|
}
|
|
|
|
/// Forbids intermediaries (CDN, browser bfcache, reverse proxy with a
|
|
/// permissive default) from caching admin responses. Defence-in-depth
|
|
/// for cookie-authenticated reads — even though the responses already
|
|
/// vary on cookie, a misconfigured cache layer in front of the
|
|
/// SvelteKit container could leak a logged-in admin's view to another
|
|
/// session. Headers added on response so the rest of the API is
|
|
/// unaffected.
|
|
async fn admin_no_store_guard(req: Request, next: Next) -> Response {
|
|
let is_admin_path = req.uri().path().starts_with(ADMIN_PATH_PREFIX);
|
|
let mut resp = next.run(req).await;
|
|
if is_admin_path {
|
|
resp.headers_mut().insert(
|
|
axum::http::header::CACHE_CONTROL,
|
|
HeaderValue::from_static("no-store"),
|
|
);
|
|
}
|
|
resp
|
|
}
|
|
|
|
/// Paths reachable anonymously even when `PRIVATE_MODE=true`. Login and
|
|
/// logout are needed for the auth flow itself; `/health` is reserved
|
|
/// for load-balancer probes; `/auth/config` lets the frontend decide
|
|
/// whether to render the login form or its anonymous alternatives;
|
|
/// `/auth/register` is exempted from the gate so the handler can
|
|
/// return its informative `registration_disabled` 403 (the same code
|
|
/// public-mode deployments use when `ALLOW_SELF_REGISTER=false`) —
|
|
/// the handler itself force-blocks the request body in private mode,
|
|
/// so no account ever gets created here. Everything else demands a
|
|
/// valid session cookie or bearer token.
|
|
fn is_public_in_private_mode(path: &str) -> bool {
|
|
matches!(
|
|
path,
|
|
"/api/v1/health"
|
|
| "/api/v1/auth/config"
|
|
| "/api/v1/auth/login"
|
|
| "/api/v1/auth/logout"
|
|
| "/api/v1/auth/register"
|
|
)
|
|
}
|
|
|
|
/// Site-wide auth gate for `PRIVATE_MODE=true`. With the flag off this
|
|
/// is a no-op pass-through, so public deployments take no extra DB
|
|
/// hit. With it on, the guard reuses [`CurrentUser`] — the same
|
|
/// session-cookie-then-bearer-token logic the per-handler extractor
|
|
/// uses — so the two paths can never drift.
|
|
async fn private_mode_guard(
|
|
State(state): State<AppState>,
|
|
req: Request,
|
|
next: Next,
|
|
) -> Result<Response, AppError> {
|
|
if !state.auth.private_mode {
|
|
return Ok(next.run(req).await);
|
|
}
|
|
if is_public_in_private_mode(req.uri().path()) {
|
|
return Ok(next.run(req).await);
|
|
}
|
|
let (mut parts, body) = req.into_parts();
|
|
match CurrentUser::from_request_parts(&mut parts, &state).await {
|
|
Ok(_) => {
|
|
let req = Request::from_parts(parts, body);
|
|
Ok(next.run(req).await)
|
|
}
|
|
Err(_) => Err(AppError::Unauthenticated),
|
|
}
|
|
}
|
|
|
|
pub(crate) fn cors_layer(allowed_origins: &[String]) -> CorsLayer {
|
|
if allowed_origins.is_empty() {
|
|
// Same-origin only — no CORS headers emitted.
|
|
return CorsLayer::new();
|
|
}
|
|
let origins: Vec<HeaderValue> = allowed_origins
|
|
.iter()
|
|
.filter_map(|o| HeaderValue::from_str(o).ok())
|
|
.collect();
|
|
CorsLayer::new()
|
|
.allow_origin(AllowOrigin::list(origins))
|
|
.allow_credentials(true)
|
|
.allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE])
|
|
.allow_headers([
|
|
HeaderName::from_static("content-type"),
|
|
HeaderName::from_static("authorization"),
|
|
])
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use axum::body::Body;
|
|
use axum::http::Request;
|
|
use axum::routing::get;
|
|
use tower::ServiceExt;
|
|
|
|
fn test_router() -> Router {
|
|
Router::new().route("/", get(|| async { "ok" }))
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn allowlist_preflight_emits_credentialed_headers() {
|
|
let app = test_router().layer(cors_layer(&["https://app.example.com".to_string()]));
|
|
let resp = app
|
|
.oneshot(
|
|
Request::builder()
|
|
.method(Method::OPTIONS)
|
|
.uri("/")
|
|
.header("origin", "https://app.example.com")
|
|
.header("access-control-request-method", "POST")
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
resp.headers().get("access-control-allow-origin").unwrap(),
|
|
"https://app.example.com"
|
|
);
|
|
assert_eq!(
|
|
resp.headers().get("access-control-allow-credentials").unwrap(),
|
|
"true"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn allowlist_rejects_unlisted_origin() {
|
|
let app = test_router().layer(cors_layer(&["https://app.example.com".to_string()]));
|
|
let resp = app
|
|
.oneshot(
|
|
Request::builder()
|
|
.method(Method::OPTIONS)
|
|
.uri("/")
|
|
.header("origin", "https://evil.example.org")
|
|
.header("access-control-request-method", "POST")
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
// Browsers will refuse the response when the allow-origin header
|
|
// is absent (or doesn't echo the requesting origin).
|
|
assert!(resp.headers().get("access-control-allow-origin").is_none());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn empty_allowlist_is_same_origin_only() {
|
|
let app = test_router().layer(cors_layer(&[]));
|
|
let resp = app
|
|
.oneshot(
|
|
Request::builder()
|
|
.method(Method::OPTIONS)
|
|
.uri("/")
|
|
.header("origin", "https://app.example.com")
|
|
.header("access-control-request-method", "POST")
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert!(resp.headers().get("access-control-allow-origin").is_none());
|
|
assert!(resp.headers().get("access-control-allow-credentials").is_none());
|
|
}
|
|
}
|