//! Reconcile pass — find mangas the metadata pass missed.
//!
//! The interleaved metadata pass ([`crate::crawler::pipeline`]) walks the
//! source list newest-first and visits each manga inline, which makes it
//! vulnerable to list drift: a manga can slip a pagination slot while the
//! walk spends minutes on detail work, and never get upserted.
//!
//! Reconcile fixes that with a cheap, full, unconditional list-only walk
//! (refs only — no `fetch_manga`, no early stop), set-diffs the walked keys
//! against `manga_sources`, and enqueues the strictly-missing ones as
//! `SyncManga` jobs for the crawl worker to visit. A set-diff is immune to
//! drift: a manga only has to appear *somewhere* in the list, not survive a
//! specific slot during a slow walk.
//!
//! Identity matches the DB exactly because both sides derive the key the
//! same way: the walker's refs carry `source_manga_key =
//! derive_key_from_url(list_href)`, which is precisely what
//! `manga_sources.source_manga_key` stores.

use std::collections::HashSet;

use anyhow::Context;
use sqlx::PgPool;

use crate::crawler::browser_manager::BrowserManager;
use crate::crawler::jobs::{self, EnqueueResult, JobPayload};
use crate::crawler::rate_limit::HostRateLimiters;
use crate::crawler::source::target::TargetSource;
use crate::crawler::source::{FetchContext, Source, SourceMangaRef};
use crate::crawler::status::Phase;
use crate::repo;
use crate::repo::crawler::ensure_source;

/// Counters surfaced when a reconcile pass finishes.
#[derive(Debug, Default, Clone, Copy)]
pub struct ReconcileStats {
    /// Distinct refs collected from the full list walk.
    pub walked: usize,
    /// Walked keys with no `manga_sources` row (strict `NOT EXISTS`).
    pub missing: usize,
    /// Missing mangas newly enqueued as `SyncManga` jobs this pass.
    pub enqueued: usize,
    /// Missing mangas skipped because a blocking (pending/running/dead)
    /// `SyncManga` job already exists, or the enqueue lost an insert race.
    pub skipped: usize,
}

/// Pure set-diff: which walked refs should be enqueued. A ref is selected
/// when its key is neither already in the DB (`existing`) nor already has a
/// blocking job (`blocked`). Repeated keys in `walked` (source index drift
/// can surface the same manga twice) are de-duplicated — the first ref wins.
pub(crate) fn select_missing<'a>(
    walked: &'a [SourceMangaRef],
    existing: &HashSet<String>,
    blocked: &HashSet<String>,
) -> Vec<&'a SourceMangaRef> {
    let mut seen = HashSet::new();
    let mut out = Vec::new();
    for r in walked {
        if existing.contains(&r.source_manga_key) || blocked.contains(&r.source_manga_key) {
            continue;
        }
        if seen.insert(r.source_manga_key.clone()) {
            out.push(r);
        }
    }
    out
}

/// Run a full list-only walk, diff against the DB, and enqueue missing
/// mangas as `SyncManga` jobs. Holds the (exclusive) browser lease only for
/// the cheap walk; the enqueued jobs are drained later by the crawl worker.
#[allow(clippy::too_many_arguments)]
pub async fn reconcile_missing(
    browser_manager: &BrowserManager,
    db: &PgPool,
    rate: &HostRateLimiters,
    start_url: &str,
    status: Option<&crate::crawler::status::StatusHandle>,
    tor: Option<&crate::crawler::tor::TorController>,
) -> anyhow::Result<ReconcileStats> {
    let lease = browser_manager
        .acquire()
        .await
        .context("acquire browser lease for reconcile pass")?;
    let browser_ref: &chromiumoxide::Browser = &lease;
    if let Some(s) = status {
        s.set_phase(Phase::Reconciling {
            walked: 0,
            enqueued: 0,
        })
        .await;
    }

    // Chapter parsing is irrelevant here (we never call fetch_manga); the
    // metadata-only constructor is intent-revealing and harmless.
    let source = TargetSource::new(start_url.to_string()).without_chapter_parsing();
    let ctx = FetchContext {
        browser: browser_ref,
        rate,
        tor,
    };
    let source_id = source.id();
    ensure_source(
        db,
        source_id,
        "Target Site",
        &crate::crawler::url_utils::origin_of(start_url).unwrap_or_else(|| start_url.to_string()),
    )
    .await
    .context("ensure_source")?;

    tracing::info!("starting reconcile pass (full list-only walk)");
    let mut walker = source.discover(&ctx).await.context("discover failed")?;

    // Full unconditional walk: collect every ref, no early stop. Yield the
    // lease if a coordinated browser restart is pending so the drain isn't
    // stalled — the missing set we've gathered so far is still enqueued.
    let mut walked: Vec<SourceMangaRef> = Vec::new();
    'outer: loop {
        if browser_manager.is_restart_pending() {
            tracing::info!("reconcile pass: browser restart pending — yielding partial walk");
            break;
        }
        match walker.next_batch(&ctx).await? {
            Some(batch) => {
                for r in batch {
                    walked.push(r);
                }
                if let Some(s) = status {
                    s.set_phase(Phase::Reconciling {
                        walked: walked.len(),
                        enqueued: 0,
                    })
                    .await;
                }
                if browser_manager.is_restart_pending() {
                    tracing::info!(
                        "reconcile pass: browser restart pending mid-batch — yielding partial walk"
                    );
                    break 'outer;
                }
            }
            None => break,
        }
    }
    // Browser work is done — release before the (browser-free) diff+enqueue.
    drop(lease);

    let existing = repo::crawler::existing_source_keys(db, source_id)
        .await
        .context("existing_source_keys")?;
    let blocked = repo::crawler::sync_manga_keys_with_blocking_job(db, source_id)
        .await
        .context("sync_manga_keys_with_blocking_job")?;

    let missing = select_missing(&walked, &existing, &blocked);

    let mut stats = ReconcileStats {
        walked: walked.len(),
        missing: missing.len(),
        ..Default::default()
    };
    for r in missing {
        let payload = JobPayload::SyncManga {
            source_id: source_id.to_string(),
            source_manga_key: r.source_manga_key.clone(),
            url: r.url.clone(),
            title: r.title.clone(),
        };
        match jobs::enqueue(db, &payload).await {
            Ok(EnqueueResult::Inserted(_)) => stats.enqueued += 1,
            Ok(EnqueueResult::Skipped) => stats.skipped += 1,
            Err(e) => {
                tracing::warn!(key = %r.source_manga_key, error = ?e, "enqueue SyncManga failed");
                stats.skipped += 1;
            }
        }
        if let Some(s) = status {
            s.set_phase(Phase::Reconciling {
                walked: stats.walked,
                enqueued: stats.enqueued,
            })
            .await;
        }
    }

    tracing::info!(
        walked = stats.walked,
        missing = stats.missing,
        enqueued = stats.enqueued,
        skipped = stats.skipped,
        "reconcile pass complete"
    );
    Ok(stats)
}

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

    fn r(key: &str) -> SourceMangaRef {
        SourceMangaRef {
            source_manga_key: key.to_string(),
            title: format!("Title {key}"),
            url: format!("https://target.example/manga/{key}"),
        }
    }

    fn set(keys: &[&str]) -> HashSet<String> {
        keys.iter().map(|s| s.to_string()).collect()
    }

    #[test]
    fn select_missing_returns_keys_not_in_manga_sources() {
        let walked = vec![r("a"), r("b"), r("c")];
        let existing = set(&["b"]);
        let blocked = set(&[]);
        let got: Vec<&str> = select_missing(&walked, &existing, &blocked)
            .iter()
            .map(|r| r.source_manga_key.as_str())
            .collect();
        assert_eq!(got, vec!["a", "c"]);
    }

    #[test]
    fn select_missing_treats_dropped_rows_as_present() {
        // The caller passes dropped keys in `existing` (existing_source_keys
        // does not filter dropped_at), so a dropped manga is NOT re-enqueued.
        let walked = vec![r("a"), r("dropped")];
        let existing = set(&["dropped"]);
        let got: Vec<&str> = select_missing(&walked, &existing, &set(&[]))
            .iter()
            .map(|r| r.source_manga_key.as_str())
            .collect();
        assert_eq!(got, vec!["a"]);
    }

    #[test]
    fn select_missing_skips_keys_with_blocking_job() {
        let walked = vec![r("a"), r("queued"), r("dead")];
        let blocked = set(&["queued", "dead"]);
        let got: Vec<&str> = select_missing(&walked, &set(&[]), &blocked)
            .iter()
            .map(|r| r.source_manga_key.as_str())
            .collect();
        assert_eq!(got, vec!["a"]);
    }

    #[test]
    fn select_missing_dedups_repeated_walked_keys() {
        // Index drift can surface the same manga twice in one walk.
        let walked = vec![r("a"), r("a"), r("b")];
        let got: Vec<&str> = select_missing(&walked, &set(&[]), &set(&[]))
            .iter()
            .map(|r| r.source_manga_key.as_str())
            .collect();
        assert_eq!(got, vec!["a", "b"]);
    }
}
