Mangalord/backend/src/crawler/status.rs

//! Live, in-process crawler status.
//!
//! The metadata pass runs inline in the cron tick (it is not a
//! `crawler_jobs` row), so without this surface "what is the crawler doing
//! right now" is unanswerable from the dashboard. The daemon publishes its
//! current [`Phase`], the chapters being crawled right now (with a live
//! page count), and the cover being fetched into a shared [`StatusHandle`];
//! the admin endpoint reads a [`CrawlerStatus`] snapshot and composes it
//! with DB-derived counts + the session/browser flags.
//!
//! NOTE: this is per-process state. The deployment is a single server
//! (see CLAUDE.md), so an in-memory handle is sufficient; durable signals
//! (last-pass summary, runtime session) are persisted in `crawler_state`.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use chrono::{DateTime, Utc};
use serde::Serialize;
use tokio::sync::{watch, RwLock};
use uuid::Uuid;

use crate::crawler::pipeline::MetadataStats;

/// What the daemon's metadata pass is doing right now. Serialised with an
/// internal `state` tag so the frontend can switch on it.
#[derive(Clone, Debug, Serialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum Phase {
    /// Sleeping until the next scheduled metadata pass.
    Idle { next_fire: Option<DateTime<Utc>> },
    /// Walking the source catalog list pages.
    WalkingList,
    /// Fetching one manga's metadata. `index`/`total` drive a progress bar
    /// (`total` is `None` when the source size is unknown / uncapped).
    FetchingMetadata {
        index: usize,
        total: Option<usize>,
        title: String,
    },
    /// Backfilling covers that failed on first attempt. `index`/`total`
    /// track progress through this tick's batch.
    CoverBackfill { index: usize, total: usize },
    /// Reconcile pass: walking the full source list (refs only, no detail
    /// visit) to find mangas missing from the DB. `walked` is refs seen so
    /// far this walk; `enqueued` is missing mangas queued as `SyncManga`.
    Reconciling { walked: usize, enqueued: usize },
}

/// A chapter being downloaded right now, with a live page count. Keyed in
/// the status by `chapter_id`; inserted by the dispatcher when a job starts
/// and removed (via an RAII guard) when it finishes, panics, or times out.
#[derive(Clone, Debug, Serialize)]
pub struct ActiveChapter {
    pub manga_id: Uuid,
    pub manga_title: String,
    pub chapter_id: Uuid,
    pub chapter_number: i32,
    pub pages_done: usize,
    /// `None` until the chapter page list has been parsed.
    pub pages_total: Option<usize>,
}

/// The manga whose cover is being downloaded right now.
#[derive(Clone, Debug, Serialize)]
pub struct CoverTarget {
    pub manga_id: Uuid,
    pub manga_title: String,
}

/// Summary of the most recent metadata pass (persisted across restarts in
/// `crawler_state` by the cron; mirrored here for the live read).
#[derive(Clone, Debug, Serialize, Default)]
pub struct LastPass {
    pub at: Option<DateTime<Utc>>,
    pub discovered: usize,
    pub upserted: usize,
    pub covers_fetched: usize,
    pub mangas_failed: usize,
}

/// A point-in-time snapshot returned by [`StatusHandle::snapshot`]. The
/// session/browser/queue fields are composed at read time by the endpoint
/// (they live elsewhere), so they are not stored here.
#[derive(Clone, Debug, Serialize)]
pub struct CrawlerStatus {
    pub phase: Phase,
    /// Number of configured chapter workers (for "N busy / M workers").
    pub worker_count: usize,
    /// Chapters being downloaded right now, with live page counts.
    pub active_chapters: Vec<ActiveChapter>,
    pub last_pass: LastPass,
    /// The cover being downloaded right now, if any.
    pub current_cover: Option<CoverTarget>,
}

/// Scalar status state held under the async `RwLock`. Active chapters and
/// the current cover live in separate sync maps so per-page updates and
/// RAII removal don't need to `.await` (removal happens in `Drop`).
#[derive(Clone, Debug)]
struct Scalar {
    phase: Phase,
    worker_count: usize,
    last_pass: LastPass,
}

/// Cloneable handle the daemon tasks use to publish status. Cheap to clone
/// (`Arc`). All writers funnel through the helper methods so locking stays
/// localised. Every mutation bumps a `watch` version so SSE subscribers
/// get pushed an update instead of polling.
#[derive(Clone)]
pub struct StatusHandle {
    scalar: Arc<RwLock<Scalar>>,
    /// Currently-downloading chapters keyed by `chapter_id`. A sync mutex so
    /// the RAII [`ChapterGuard`]'s `Drop` can remove without `.await`.
    active: Arc<Mutex<HashMap<Uuid, ActiveChapter>>>,
    /// The cover being downloaded right now (if any). Sync mutex so the
    /// RAII [`CoverGuard`]'s `Drop` can clear without `.await`, which is
    /// what makes the cleared-on-panic guarantee hold.
    current_cover: Arc<Mutex<Option<CoverTarget>>>,
    /// Monotonic version bumped on every change. SSE handlers `subscribe()`
    /// and `await .changed()` for instant pushes; `watch` has no
    /// lost-wakeup so a change between snapshots is never missed.
    version: Arc<watch::Sender<u64>>,
}

/// Lock the active map, recovering from a poisoned mutex. The map values
/// are plain structs and we never hold the lock across a panic-prone
/// section, so resuming on poison is safe — but log it so a real poison
/// (which signals a panic-in-critical-section bug somewhere) doesn't pass
/// in silence.
fn lock_active(
    m: &Mutex<HashMap<Uuid, ActiveChapter>>,
) -> std::sync::MutexGuard<'_, HashMap<Uuid, ActiveChapter>> {
    m.lock().unwrap_or_else(|e| {
        tracing::warn!(
            "status::lock_active recovered from a poisoned mutex — \
             this implies a panic somewhere holding the lock"
        );
        e.into_inner()
    })
}

/// Same shape as [`lock_active`] but for the single-slot cover mutex.
fn lock_cover(
    m: &Mutex<Option<CoverTarget>>,
) -> std::sync::MutexGuard<'_, Option<CoverTarget>> {
    m.lock().unwrap_or_else(|e| {
        tracing::warn!(
            "status::lock_cover recovered from a poisoned mutex — \
             this implies a panic somewhere holding the lock"
        );
        e.into_inner()
    })
}

impl StatusHandle {
    pub fn new(num_workers: usize) -> Self {
        let (version, _rx) = watch::channel(0u64);
        Self {
            scalar: Arc::new(RwLock::new(Scalar {
                phase: Phase::Idle { next_fire: None },
                worker_count: num_workers.max(1),
                last_pass: LastPass::default(),
            })),
            active: Arc::new(Mutex::new(HashMap::new())),
            current_cover: Arc::new(Mutex::new(None)),
            version: Arc::new(version),
        }
    }

    fn bump(&self) {
        self.version.send_modify(|v| *v = v.wrapping_add(1));
    }

    /// A receiver whose `.changed()` resolves on the next status change.
    pub fn subscribe(&self) -> watch::Receiver<u64> {
        self.version.subscribe()
    }

    /// Signal a change without mutating in-memory state — used when an
    /// *external* signal the live snapshot reflects (browser phase,
    /// session-expired flag, queue counts) has changed, so subscribers
    /// recompose promptly.
    pub fn poke(&self) {
        self.bump();
    }

    pub async fn set_phase(&self, phase: Phase) {
        self.scalar.write().await.phase = phase;
        self.bump();
    }

    /// Register a cover-fetch as in flight; returns a guard that clears
    /// the current cover when dropped (on completion, panic-unwind, or
    /// any future early-return). Last-writer-wins: a guard only clears
    /// the slot when it still holds the cover it set (so overlapping
    /// guards — not used today, but defensive — don't clobber each
    /// other).
    pub fn begin_cover(&self, target: CoverTarget) -> CoverGuard {
        let manga_id = target.manga_id;
        *lock_cover(&self.current_cover) = Some(target);
        self.bump();
        CoverGuard {
            current_cover: Arc::clone(&self.current_cover),
            version: Arc::clone(&self.version),
            manga_id,
        }
    }

    /// Register a chapter as crawling now; returns a guard that removes it
    /// when dropped (on completion, panic-unwind, or timeout-drop).
    pub fn begin_chapter(&self, chapter: ActiveChapter) -> ChapterGuard {
        let id = chapter.chapter_id;
        lock_active(&self.active).insert(id, chapter);
        self.bump();
        ChapterGuard {
            active: Arc::clone(&self.active),
            version: Arc::clone(&self.version),
            chapter_id: id,
        }
    }

    /// Update the live page count of an in-flight chapter. Sync (no
    /// `.await`) so it's cheap to call once per stored page.
    pub fn set_chapter_pages(&self, chapter_id: Uuid, done: usize, total: Option<usize>) {
        {
            let mut map = lock_active(&self.active);
            if let Some(c) = map.get_mut(&chapter_id) {
                c.pages_done = done;
                c.pages_total = total;
            }
        }
        self.bump();
    }

    /// Record a finished metadata pass. Stamps `at` with `now`.
    pub async fn record_pass(&self, stats: &MetadataStats, at: DateTime<Utc>) {
        self.scalar.write().await.last_pass = LastPass {
            at: Some(at),
            discovered: stats.discovered,
            upserted: stats.upserted,
            covers_fetched: stats.covers_fetched,
            mangas_failed: stats.mangas_failed,
        };
        self.bump();
    }

    /// Seed the last-pass summary from a persisted `crawler_state` value on
    /// startup so the dashboard isn't blank until the first tick.
    pub async fn set_last_pass(&self, last: LastPass) {
        self.scalar.write().await.last_pass = last;
        self.bump();
    }

    pub async fn snapshot(&self) -> CrawlerStatus {
        let scalar = self.scalar.read().await.clone();
        let mut active_chapters: Vec<ActiveChapter> =
            lock_active(&self.active).values().cloned().collect();
        // Stable, readable order: by chapter number then id.
        active_chapters.sort_by(|a, b| {
            a.chapter_number
                .cmp(&b.chapter_number)
                .then(a.chapter_id.cmp(&b.chapter_id))
        });
        let current_cover = lock_cover(&self.current_cover).clone();
        CrawlerStatus {
            phase: scalar.phase,
            worker_count: scalar.worker_count,
            active_chapters,
            last_pass: scalar.last_pass,
            current_cover,
        }
    }
}

/// RAII handle clearing the [`CoverTarget`] from the live status when the
/// cover-fetch finishes, panics, or is dropped on any early-return.
pub struct CoverGuard {
    current_cover: Arc<Mutex<Option<CoverTarget>>>,
    version: Arc<watch::Sender<u64>>,
    /// Manga id whose cover this guard registered. The drop only clears
    /// the slot when the stored value still matches — defends against a
    /// hypothetical newer guard clobbering this one's clear.
    manga_id: Uuid,
}

impl Drop for CoverGuard {
    fn drop(&mut self) {
        let mut slot = lock_cover(&self.current_cover);
        if slot.as_ref().map(|c| c.manga_id) == Some(self.manga_id) {
            *slot = None;
        }
        self.version.send_modify(|v| *v = v.wrapping_add(1));
    }
}

/// RAII handle removing an [`ActiveChapter`] from the live status when the
/// chapter dispatch finishes, panics, or is dropped on timeout.
pub struct ChapterGuard {
    active: Arc<Mutex<HashMap<Uuid, ActiveChapter>>>,
    version: Arc<watch::Sender<u64>>,
    chapter_id: Uuid,
}

impl Drop for ChapterGuard {
    fn drop(&mut self) {
        lock_active(&self.active).remove(&self.chapter_id);
        self.version.send_modify(|v| *v = v.wrapping_add(1));
    }
}

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

    fn sample_chapter(n: i32) -> ActiveChapter {
        ActiveChapter {
            manga_id: Uuid::new_v4(),
            manga_title: "M".into(),
            chapter_id: Uuid::new_v4(),
            chapter_number: n,
            pages_done: 0,
            pages_total: None,
        }
    }

    #[tokio::test]
    async fn begin_chapter_shows_in_snapshot_and_guard_removes_on_drop() {
        let h = StatusHandle::new(2);
        let chap = sample_chapter(7);
        let cid = chap.chapter_id;
        {
            let _guard = h.begin_chapter(chap);
            let snap = h.snapshot().await;
            assert_eq!(snap.active_chapters.len(), 1);
            assert_eq!(snap.active_chapters[0].chapter_id, cid);
            assert_eq!(snap.worker_count, 2);
        }
        // Guard dropped → entry removed.
        let snap = h.snapshot().await;
        assert!(snap.active_chapters.is_empty());
    }

    #[tokio::test]
    async fn set_chapter_pages_updates_live_count() {
        let h = StatusHandle::new(1);
        let chap = sample_chapter(1);
        let cid = chap.chapter_id;
        let _guard = h.begin_chapter(chap);
        h.set_chapter_pages(cid, 3, Some(20));
        let snap = h.snapshot().await;
        assert_eq!(snap.active_chapters[0].pages_done, 3);
        assert_eq!(snap.active_chapters[0].pages_total, Some(20));
        // Updating an unknown chapter is a no-op, not a panic.
        h.set_chapter_pages(Uuid::new_v4(), 9, Some(9));
    }

    #[tokio::test]
    async fn snapshot_sorts_active_chapters_by_number() {
        let h = StatusHandle::new(2);
        let _g1 = h.begin_chapter(sample_chapter(5));
        let _g2 = h.begin_chapter(sample_chapter(2));
        let snap = h.snapshot().await;
        assert_eq!(snap.active_chapters[0].chapter_number, 2);
        assert_eq!(snap.active_chapters[1].chapter_number, 5);
    }

    #[tokio::test]
    async fn cover_guard_sets_then_clears_on_drop() {
        let h = StatusHandle::new(1);
        let mid = Uuid::new_v4();
        {
            let _g = h.begin_cover(CoverTarget {
                manga_id: mid,
                manga_title: "One Piece".into(),
            });
            assert_eq!(
                h.snapshot().await.current_cover.map(|c| c.manga_id),
                Some(mid)
            );
        }
        assert!(h.snapshot().await.current_cover.is_none());
    }

    #[tokio::test]
    async fn cover_guard_clears_on_panic_drop() {
        // Simulate a download_and_store_cover panic: the guard is on the
        // stack, the panic unwinds, and the slot must still be cleared.
        let h = StatusHandle::new(1);
        let mid = Uuid::new_v4();
        let h2 = h.clone();
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _g = h2.begin_cover(CoverTarget {
                manga_id: mid,
                manga_title: "K-On!".into(),
            });
            panic!("simulated cover-download panic");
        }));
        assert!(result.is_err());
        assert!(h.snapshot().await.current_cover.is_none());
    }

    #[tokio::test]
    async fn cover_guard_does_not_clobber_a_newer_target() {
        // Defensive: if a *newer* begin_cover ran before the older
        // guard's drop fires, the drop must not clear the newer
        // target. (No code today produces overlapping guards, but the
        // invariant prevents a future caller from quietly breaking the
        // live-cover surface.)
        let h = StatusHandle::new(1);
        let older = Uuid::new_v4();
        let newer = Uuid::new_v4();
        let g_old = h.begin_cover(CoverTarget {
            manga_id: older,
            manga_title: "old".into(),
        });
        let _g_new = h.begin_cover(CoverTarget {
            manga_id: newer,
            manga_title: "new".into(),
        });
        drop(g_old);
        assert_eq!(
            h.snapshot().await.current_cover.map(|c| c.manga_id),
            Some(newer)
        );
    }

    #[tokio::test]
    async fn record_pass_captures_stats_and_timestamp() {
        let h = StatusHandle::new(1);
        let stats = MetadataStats {
            discovered: 5,
            upserted: 3,
            covers_fetched: 2,
            mangas_failed: 1,
        };
        let at = Utc::now();
        h.record_pass(&stats, at).await;
        let snap = h.snapshot().await;
        assert_eq!(snap.last_pass.discovered, 5);
        assert_eq!(snap.last_pass.upserted, 3);
        assert_eq!(snap.last_pass.at, Some(at));
    }

    #[tokio::test]
    async fn subscribe_resolves_on_mutation_poke_and_chapter_change() {
        let h = StatusHandle::new(1);
        let mut rx = h.subscribe();
        h.set_phase(Phase::WalkingList).await;
        rx.changed().await.unwrap();
        h.poke();
        rx.changed().await.unwrap();
        // begin_chapter + guard drop each bump the version.
        let g = h.begin_chapter(sample_chapter(1));
        rx.changed().await.unwrap();
        drop(g);
        rx.changed().await.unwrap();
    }
}