//! Orchestration: per-channel pipeline, gofile/0807 loops, subscriptions,
//! --assume-done — Rust port of the logic in `version-python/dlp.py`.
//!
//! Failure policy (the core requirement): a failure in one URL must never
//! abort the run. Catch, record in state.json, continue. Cooldown pauses
//! are the only permitted stall.

use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::thread;
use std::time::Duration;

use anyhow::{Context as _, anyhow};
use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
use serde_json::Value;
use sha2::{Digest, Sha256};

use crate::discord::{self, FoundUrl};
use crate::globalstore::GlobalStore;
use crate::gofile::{GoFileClient, GoFileError};
use crate::manifest::{ManifestRow, append_manifest};
use crate::state::{FileEntry, State};
use crate::zero807::{self, Zero807Error};

/// Runtime options for a download pass (the subset of CLI flags the
/// orchestrator needs).
#[derive(Debug, Clone)]
pub struct RunOpts {
    pub retries: u32,
    pub retry_delay: u64,
    pub cooldown_failures: u32,
    pub cooldown_seconds: u64,
    pub hash: bool,
    pub extract: bool,
    pub password: Option<String>,
    pub limit: Option<usize>,
    pub shuffle: bool,
    pub only: Option<String>,
    /// only URLs posted at/after this point (Rust-only extension: "5d",
    /// "12h", "2026-10-04", or an RFC 3339 timestamp)
    pub since: Option<String>,
    pub dry_run: bool,
    pub fresh: bool,
}

impl Default for RunOpts {
    fn default() -> Self {
        RunOpts {
            retries: 3,
            retry_delay: 5,
            cooldown_failures: 3,
            cooldown_seconds: 60,
            hash: false,
            extract: false,
            password: None,
            limit: None,
            shuffle: false,
            only: None,
            since: None,
            dry_run: false,
            fresh: false,
        }
    }
}

/// Parse a `--since` value: relative ("5d", "12h", "30m") or absolute
/// (date, naive datetime, or RFC 3339). Naive times are read as UTC.
pub fn parse_since(s: &str) -> anyhow::Result<DateTime<Utc>> {
    let t = s.trim();
    for (suffix, dur) in [
        ("d", chrono::Duration::hours(24)),
        ("h", chrono::Duration::hours(1)),
        ("m", chrono::Duration::minutes(1)),
    ] {
        if let Some(num) = t.strip_suffix(suffix) {
            let n: i32 = num.parse().map_err(|_| anyhow!("bad --since value: {s}"))?;
            return Ok(Utc::now() - dur * n);
        }
    }
    if let Ok(dt) = DateTime::parse_from_rfc3339(t) {
        return Ok(dt.with_timezone(&Utc));
    }
    if let Ok(dt) = NaiveDateTime::parse_from_str(t, "%Y-%m-%dT%H:%M:%S%.f") {
        return Ok(dt.and_utc());
    }
    if let Ok(dt) = NaiveDateTime::parse_from_str(t, "%Y-%m-%d %H:%M:%S") {
        return Ok(dt.and_utc());
    }
    if let Ok(d) = NaiveDate::parse_from_str(t, "%Y-%m-%d") {
        return Ok(d.and_hms_opt(0, 0, 0).unwrap().and_utc());
    }
    anyhow::bail!("bad --since value: {s} (use e.g. 5d, 12h, 2026-10-04, or RFC 3339)")
}

/// Parse a message timestamp from the export; None -> keep-with-warning.
pub fn parse_message_ts(ts: &str) -> Option<DateTime<Utc>> {
    if let Ok(dt) = DateTime::parse_from_rfc3339(ts) {
        return Some(dt.with_timezone(&Utc));
    }
    if let Ok(dt) = NaiveDateTime::parse_from_str(ts, "%Y-%m-%dT%H:%M:%S%.f") {
        return Some(dt.and_utc());
    }
    if let Ok(d) = NaiveDate::parse_from_str(ts, "%Y-%m-%d") {
        return Some(d.and_hms_opt(0, 0, 0).unwrap().and_utc());
    }
    None
}

/// sha256 of a file, streamed (python sha256_of).
pub fn sha256_of(path: &Path) -> anyhow::Result<String> {
    let mut file =
        fs::File::open(path).with_context(|| format!("opening {} for hashing", path.display()))?;
    let mut hasher = Sha256::new();
    std::io::copy(&mut file, &mut hasher)?;
    Ok(hex::encode(hasher.finalize()))
}

/// Lines from a URL list file; skips blanks and # comments.
pub fn read_url_lines(path: &Path) -> anyhow::Result<Vec<String>> {
    let text = fs::read_to_string(path).with_context(|| format!("reading {}", path.display()))?;
    Ok(text
        .lines()
        .map(str::trim)
        .filter(|l| !l.is_empty() && !l.starts_with('#'))
        .map(str::to_string)
        .collect())
}

/// Load a plain URL list (e.g. failed_urls.txt); no attribution available.
pub fn load_urls_file(path: &Path) -> anyhow::Result<Vec<FoundUrl>> {
    let mut found: HashMap<String, FoundUrl> = HashMap::new();
    for url in read_url_lines(path)? {
        let Some(host) = discord::classify(&url) else {
            log::warn!("Unsupported host in {}: {url}", path.display());
            continue;
        };
        found.entry(url.clone()).or_insert(FoundUrl {
            url,
            host,
            message_id: None,
            timestamp: None,
            author: None,
            encoded: false,
        });
    }
    let mut list: Vec<FoundUrl> = found.into_values().collect();
    list.sort_by(|a, b| a.url.cmp(&b.url));
    Ok(list)
}

/// Record a successful download: state.mark(ok) + global store + manifest
/// rows (python record_files).
pub fn record_files(
    state: &mut State,
    gstore: Option<&mut GlobalStore>,
    url: &str,
    files: &[Value],
    source: &SourceInfo,
    manifest_path: &Path,
) -> anyhow::Result<()> {
    let entries: Vec<FileEntry> = files
        .iter()
        .map(|f| FileEntry {
            path: f["path"].as_str().unwrap_or_default().to_string(),
            size: f["size"].as_u64(),
            extra: {
                let mut m = std::collections::BTreeMap::new();
                if let Some(n) = f.get("name") {
                    m.insert("name".to_string(), n.clone());
                }
                if let Some(h) = f.get("sha256") {
                    m.insert("sha256".to_string(), h.clone());
                }
                m
            },
        })
        .collect();
    state.mark(url, |r| {
        r.status = "ok".into();
        r.files = entries.clone();
        // python passes error=None which is *skipped* (never clears)
    })?;
    if let Some(gs) = gstore {
        gs.mark(
            url,
            Some(&source.channel),
            source.host.as_deref(),
            Some(files.len() as u64),
            None,
        )?;
    }
    for f in files {
        let row = ManifestRow {
            channel: source.channel.clone(),
            url: url.to_string(),
            host: source.host.clone(),
            message_id: source.message_id.clone(),
            timestamp: source.timestamp.clone(),
            author: source.author.clone(),
            path: f["path"].as_str().map(str::to_string),
            size: f["size"].as_u64(),
            extra: {
                let mut m = std::collections::BTreeMap::new();
                if let Some(n) = f.get("name") {
                    m.insert("name".to_string(), n.clone());
                }
                if let Some(h) = f.get("sha256") {
                    m.insert("sha256".to_string(), h.clone());
                }
                m
            },
        };
        append_manifest(manifest_path, &row)?;
    }
    Ok(())
}

/// Message-level attribution for one URL (python source_by_url values).
#[derive(Debug, Clone, Default)]
pub struct SourceInfo {
    pub channel: String,
    pub host: Option<String>,
    pub message_id: Option<Value>,
    pub timestamp: Option<String>,
    pub author: Option<String>,
    pub manifest: PathBuf,
}

fn source_of(channel_id: &str, f: &FoundUrl, manifest: &Path) -> SourceInfo {
    SourceInfo {
        channel: channel_id.to_string(),
        host: Some(f.host.to_string()),
        message_id: f.message_id.clone().map(Value::String),
        timestamp: f.timestamp.clone(),
        author: f.author.clone(),
        manifest: manifest.to_path_buf(),
    }
}

/// Optional: unzip each downloaded archive next to itself (python
/// extract_zips in dlp.py).
pub fn extract_zips(files: &[Value]) {
    for f in files {
        let Some(path_str) = f["path"].as_str() else {
            continue;
        };
        let path = PathBuf::from(path_str);
        if path.extension().map(|e| e.to_ascii_lowercase()) != Some("zip".into()) || !path.exists()
        {
            continue;
        }
        let target = path.with_extension("");
        if target.exists()
            && fs::read_dir(&target)
                .map(|d| d.count() > 0)
                .unwrap_or(false)
        {
            log::info!("Skip extracting {} (target exists)", path.display());
            continue;
        }
        log::info!("Extracting {} -> {}", path.display(), target.display());
        if let Err(exc) = fs::create_dir_all(&target).and_then(|_| unzip_to(&path, &target)) {
            log::warn!("Extraction of {} failed: {exc}", path.display());
        }
    }
}

fn unzip_to(zip_path: &Path, target: &Path) -> std::io::Result<()> {
    let file = fs::File::open(zip_path)?;
    let mut zf = zip::ZipArchive::new(file)?;
    for i in 0..zf.len() {
        let mut entry = zf.by_index(i)?;
        if entry.is_dir() {
            fs::create_dir_all(target.join(entry.name()))?;
            continue;
        }
        let out_path = target.join(entry.name());
        if let Some(parent) = out_path.parent() {
            fs::create_dir_all(parent)?;
        }
        let mut out = fs::File::create(&out_path)?;
        std::io::copy(&mut entry, &mut out)?;
    }
    Ok(())
}

pub fn format_size(n: u64) -> String {
    let mut size = n as f64;
    for unit in ["B", "KiB", "MiB", "GiB", "TiB"] {
        if size < 1024.0 || unit == "TiB" {
            return format!("{size:.1} {unit}");
        }
        size /= 1024.0;
    }
    format!("{n} B")
}

pub fn write_failed_urls(out_dir: &Path, state: &State) -> anyhow::Result<PathBuf> {
    let dest = out_dir.join("failed_urls.txt");
    if let Some(parent) = dest.parent() {
        fs::create_dir_all(parent)?;
    }
    let failures = state.failures();
    let text = if failures.is_empty() {
        String::new()
    } else {
        format!("{}\n", failures.join("\n"))
    };
    fs::write(&dest, text)?;
    Ok(dest)
}

/// Record URLs as already downloaded (no files) so they are never fetched.
///
/// Manual import step for the pre-dlp.py era: point it at an old URL list
/// you know is fully downloaded. There is no verification - garbage in,
/// skipped forever.
pub fn mark_assume_done(path: &Path, gstore: &mut GlobalStore) -> anyhow::Result<usize> {
    if gstore.path.is_none() {
        anyhow::bail!("--assume-done needs the global store (drop --no-global)");
    }
    let mut count = 0;
    for url in read_url_lines(path)? {
        gstore.mark(
            &url,
            None,
            discord::classify(&url),
            Some(0),
            Some("assumed done via --assume-done"),
        )?;
        count += 1;
    }
    log::info!(
        "Marked {count} URLs as done in the global store ({})",
        gstore
            .path
            .as_ref()
            .map(|p| p.display().to_string())
            .unwrap_or_default()
    );
    Ok(count)
}

/// state.is_done(url) or the global store says done (python is_done closure).
fn is_done(state: &State, gstore: Option<&GlobalStore>, url: &str) -> bool {
    state.is_done(url) || gstore.is_some_and(|g| g.is_done(url))
}

/// Torrent links are never downloaded (user policy): the channel hosts huge
/// user-created torrents; this tool is for direct-download packs. A skipped
/// torrent is recorded as done (with a note) so it never comes back.
fn is_torrent_url(url: &str) -> bool {
    url.to_ascii_lowercase().ends_with(".torrent")
}

fn is_torrent_file(name: &str) -> bool {
    name.to_ascii_lowercase().ends_with(".torrent")
}

/// One channel's pipeline; returns the number of failures (python
/// process_channel).
pub fn process_channel(
    channel_id: &str,
    out_dir: &Path,
    opts: &RunOpts,
    token: Option<&str>,
    mut gstore: Option<&mut GlobalStore>,
    urls_file: Option<&Path>,
    force_export: bool,
) -> anyhow::Result<u64> {
    let mut state = State::load(out_dir.join("state.json"));
    let manifest = out_dir.join("manifest.jsonl");

    let found = if let Some(urls_file) = urls_file {
        load_urls_file(urls_file)?
    } else {
        let token = token.ok_or_else(|| anyhow!("internal error: no token resolved"))?;
        let export_path =
            discord::export_channel(channel_id, out_dir, token, force_export || opts.fresh)?;
        discord::extract_urls(&export_path)?
    };

    let found: Vec<FoundUrl> = found
        .into_iter()
        .filter(|f| {
            if is_torrent_url(&f.url) {
                log::info!("Skip torrent (never downloaded): {}", f.url);
                if let Err(e) = state.mark(&f.url, |r| {
                    r.status = "ok".into();
                    r.extra
                        .insert("note".into(), Value::from("torrent link skipped (policy)"));
                }) {
                    log::error!("Could not record torrent skip for {}: {e}", f.url);
                }
                return false;
            }
            true
        })
        .filter(|f| opts.only.as_ref().is_none_or(|o| o.as_str() == f.host))
        .collect();

    let found: Vec<FoundUrl> = if let Some(since) = &opts.since {
        let cutoff = parse_since(since)?;
        let before = found.len();
        let kept: Vec<FoundUrl> = found
            .into_iter()
            .filter(|f| match &f.timestamp {
                Some(ts) => match parse_message_ts(ts) {
                    Some(dt) => dt >= cutoff,
                    None => {
                        log::warn!("Unparseable timestamp {ts:?} for {} - keeping it", f.url);
                        true
                    }
                },
                None => {
                    // e.g. --urls-file lists carry no attribution
                    log::warn!("No timestamp for {} - keeping it", f.url);
                    true
                }
            })
            .collect();
        log::info!(
            "--since {}: kept {} of {} URLs (rest are older)",
            since,
            kept.len(),
            before
        );
        kept
    } else {
        found
    };

    let mut gofile_urls: Vec<FoundUrl> = Vec::new();
    let mut z0807_urls: Vec<FoundUrl> = Vec::new();
    for f in &found {
        match f.host {
            "gofile" => gofile_urls.push(f.clone()),
            "0807" => z0807_urls.push(f.clone()),
            // classify() already filtered, but be defensive
            _ => {
                log::warn!("Skipping unsupported URL: {}", f.url);
                state.mark(&f.url, |r| {
                    r.status = "failed".into();
                    r.error = Some("unsupported host".into());
                })?;
            }
        }
    }

    let mut source_by_url: HashMap<String, SourceInfo> = HashMap::new();
    for f in &gofile_urls {
        source_by_url.insert(f.url.clone(), source_of(channel_id, f, &manifest));
    }
    for f in &z0807_urls {
        source_by_url.insert(f.url.clone(), source_of(channel_id, f, &manifest));
    }

    // --limit applies per host class to the *pending* URLs, so a test run
    // downloads exactly N new things per host (done ones don't use up the
    // quota). Python replaces the bucket with pending[:limit].
    if let Some(limit) = opts.limit {
        for list in [&mut gofile_urls, &mut z0807_urls] {
            let mut pending: Vec<FoundUrl> = list
                .iter()
                .filter(|f| !is_done(&state, gstore.as_deref(), &f.url))
                .cloned()
                .collect();
            if opts.shuffle {
                fisher_yates(&mut pending);
            }
            pending.truncate(limit);
            *list = pending;
        }
    }

    let total = gofile_urls.len() + z0807_urls.len();
    let pending_len = gofile_urls
        .iter()
        .chain(&z0807_urls)
        .filter(|f| !is_done(&state, gstore.as_deref(), &f.url))
        .count();
    log::info!(
        "{cid}: {n} URLs ({g} gofile, {z} 0807); {d} done, {p} to download",
        cid = channel_id,
        n = total,
        g = gofile_urls.len(),
        z = z0807_urls.len(),
        d = total - pending_len,
        p = pending_len,
    );

    if opts.dry_run {
        let selected: std::collections::HashSet<&str> = gofile_urls
            .iter()
            .chain(&z0807_urls)
            .map(|f| f.url.as_str())
            .collect();
        for f in &found {
            let flag = if selected.contains(f.url.as_str()) {
                "download"
            } else if is_done(&state, gstore.as_deref(), &f.url) {
                "done"
            } else {
                "skip" // cut by --limit / --only
            };
            println!("[{:<6}] [{:<8}] {}", f.host, flag, f.url);
        }
        return Ok(0);
    }

    if !gofile_urls.is_empty() {
        run_gofile(
            &gofile_urls,
            opts,
            out_dir,
            &mut state,
            &source_by_url,
            gstore.as_deref_mut(),
            &manifest,
        )?;
    }
    if !z0807_urls.is_empty() {
        run_0807(
            &z0807_urls,
            opts,
            out_dir,
            &mut state,
            &source_by_url,
            gstore.as_deref_mut(),
            &manifest,
        )?;
    }

    let failed_file = write_failed_urls(out_dir, &state)?;

    let ok = state.records.values().filter(|r| r.status == "ok").count();
    let bad = state.failures();
    let dead = state
        .records
        .values()
        .filter(|r| {
            r.error
                .as_deref()
                .map(|e| e.contains("dead link"))
                .unwrap_or(false)
        })
        .count();
    let total_files: u64 = state.records.values().map(|r| r.files.len() as u64).sum();
    let total_size: u64 = state
        .records
        .values()
        .flat_map(|r| r.files.iter())
        .map(|f| f.size.unwrap_or(0))
        .sum();
    log::info!(
        "{}: {} ok ({} files, {}) / {} failed ({} dead links) / {} seen",
        out_dir.display(),
        ok,
        total_files,
        format_size(total_size),
        bad.len(),
        dead,
        state.records.len(),
    );
    if !bad.is_empty() {
        log::info!(
            "Failed URLs written to {} - re-run this tool or use \
             dlphelper dlp <id> <out> --urls-file {} to retry them.",
            failed_file.display(),
            failed_file
                .file_name()
                .map(|n| n.to_string_lossy().into_owned())
                .unwrap_or_default()
        );
    }
    Ok(bad.len() as u64)
}

/// Fisher-Yates over pending URL pointers (python random.shuffle).
fn fisher_yates<T>(items: &mut [T]) {
    use rand::Rng;
    let mut rng = rand::rng();
    for i in (1..items.len()).rev() {
        let j = rng.random_range(..=i);
        items.swap(i, j);
    }
}

/// GoFile pipeline (python run_gofile): one guest account for the whole
/// pass, per-URL retry loop with dead-link/password/empty no-retry branches,
/// re-auth on auth errors, cooldowns on consecutive failures - the run never
/// stops.
pub fn run_gofile(
    urls: &[FoundUrl],
    opts: &RunOpts,
    out_dir: &Path,
    state: &mut State,
    source_by_url: &HashMap<String, SourceInfo>,
    mut gstore: Option<&mut GlobalStore>,
    manifest: &Path,
) -> anyhow::Result<()> {
    if !urls
        .iter()
        .any(|f| !is_done(state, gstore.as_deref(), &f.url))
    {
        return Ok(()); // nothing pending: no guest account creation, no API calls
    }
    let mut client = GoFileClient::new(opts.password.as_deref());
    if let Err(exc) = client.ensure_token() {
        log::error!("{exc} - marking all pending gofile links as failed.");
        for f in urls {
            if is_done(state, gstore.as_deref(), &f.url) {
                continue;
            }
            state.mark(&f.url, |r| {
                r.status = "failed".into();
                r.error = Some(exc.to_string());
                r.files = Vec::new();
            })?;
        }
        return Ok(());
    }

    let mut consecutive_failures: u32 = 0;
    for f in urls {
        if is_done(state, gstore.as_deref(), &f.url) {
            log::info!("Skip (already downloaded): {}", f.url);
            continue;
        }
        let source = source_by_url.get(&f.url).cloned().unwrap_or_default();
        let mut resolved = false;
        for attempt in 1..=opts.retries {
            // One attempt: list, download all files, hash. GoFileError maps
            // onto the python exception handlers below; anything else is an
            // unexpected bug and is recorded as such. The immediately-invoked
            // closure ends the shared borrow of `client` before the handler
            // arms below may re-authenticate.
            let files = (|| -> anyhow::Result<Vec<Value>> {
                let files_spec = client.list_files(&f.url).map_err(anyhow::Error::new)?;
                if files_spec.is_empty() {
                    return Err(anyhow::Error::new(GoFileError::Empty(
                        "content exists but contains no files".into(),
                    )));
                }
                let mut files: Vec<Value> = Vec::new();
                let mut skipped_torrents = 0usize;
                for (relpath, name, link) in files_spec {
                    if is_torrent_file(&name) {
                        skipped_torrents += 1;
                        log::info!("Skip torrent file in content (policy): {name}");
                        continue;
                    }
                    let dest = out_dir
                        .join("downloads")
                        .join("gofile")
                        .join(&relpath)
                        .join(&name);
                    let size = client
                        .download_file(&link, &dest)
                        .map_err(anyhow::Error::new)?;
                    let mut entry = json_entry(&dest, size, &name);
                    if opts.hash {
                        let h = sha256_of(&dest)?;
                        entry["sha256"] = Value::String(h);
                    }
                    files.push(entry);
                }
                if files.is_empty() && skipped_torrents > 0 {
                    // content holds only torrents: nothing to fetch, ever
                    log::info!(
                        "OK {} (0 files, {} torrent(s) skipped by policy)",
                        f.url,
                        skipped_torrents
                    );
                    return Ok(Vec::new());
                }
                Ok(files)
            })();
            match files {
                Ok(files) => {
                    log::info!("OK {} ({} files)", f.url, files.len());
                    record_files(
                        state,
                        gstore.as_deref_mut(),
                        &f.url,
                        &files,
                        &source,
                        manifest,
                    )?;
                    if files.is_empty() {
                        state.mark(&f.url, |r| {
                            r.extra.insert(
                                "note".into(),
                                Value::from("only .torrent files - skipped"),
                            );
                        })?;
                    }
                    if opts.extract {
                        extract_zips(&files);
                    }
                    consecutive_failures = 0;
                    resolved = true;
                    break;
                }
                Err(err) => {
                    let Some(ge) = err.downcast_ref::<GoFileError>() else {
                        // a bug must not stop the run
                        log::error!("Unexpected error for {}: {err:#}", f.url);
                        let msg = format!("unexpected: {err}");
                        state.mark(&f.url, |r| {
                            r.error = Some(msg);
                            r.files = Vec::new();
                        })?;
                        if attempt < opts.retries {
                            thread::sleep(Duration::from_secs(opts.retry_delay * attempt as u64));
                        }
                        continue;
                    };
                    match ge {
                        GoFileError::PasswordRequired(msg) => {
                            // Retrying without a password never helps
                            log::error!("PASSWORD REQUIRED {}: {msg}", f.url);
                            state.mark(&f.url, |r| {
                                r.status = "failed".into();
                                r.error = Some(msg.clone());
                                r.files = Vec::new();
                            })?;
                            resolved = true;
                            break;
                        }
                        GoFileError::NotFound(msg) => {
                            // Dead links never come back - no retry
                            log::error!("DEAD LINK {}: {msg}", f.url);
                            state.mark(&f.url, |r| {
                                r.status = "failed".into();
                                r.error = Some(msg.clone());
                                r.files = Vec::new();
                            })?;
                            resolved = true;
                            break;
                        }
                        GoFileError::Empty(msg) => {
                            // Emptied content is stable - no retry
                            log::error!("EMPTY CONTENT {}: {msg}", f.url);
                            state.mark(&f.url, |r| {
                                r.status = "failed".into();
                                r.error = Some(msg.clone());
                                r.files = Vec::new();
                            })?;
                            resolved = true;
                            break;
                        }
                        GoFileError::AuthError(msg) => {
                            // Token may have died mid-run (rate limit); re-auth
                            log::warn!("Auth error for {}: {msg} - re-authenticating", f.url);
                            state.mark(&f.url, |r| {
                                r.error = Some(msg.clone());
                                r.files = Vec::new();
                            })?;
                            client.reset();
                            if let Err(reauth) = client.ensure_token() {
                                log::error!("Re-authentication failed: {reauth}");
                            }
                            if attempt < opts.retries {
                                thread::sleep(Duration::from_secs(
                                    opts.retry_delay * attempt as u64,
                                ));
                            }
                        }
                        GoFileError::Error(msg) => {
                            log::warn!(
                                "Attempt {}/{} failed for {}: {msg}",
                                attempt,
                                opts.retries,
                                f.url
                            );
                            state.mark(&f.url, |r| {
                                r.error = Some(msg.clone());
                                r.files = Vec::new();
                            })?;
                            if attempt < opts.retries {
                                thread::sleep(Duration::from_secs(
                                    opts.retry_delay * attempt as u64,
                                ));
                            }
                        }
                    }
                }
            }
        }
        if !resolved {
            let err = state
                .records
                .get(&f.url)
                .and_then(|r| r.error.clone())
                .unwrap_or_else(|| "unknown".into());
            state.mark(&f.url, |r| {
                r.status = "failed".into();
                r.error = Some(err);
                r.files = Vec::new();
            })?;
            consecutive_failures += 1;
            log::error!("FAILED {} after {} attempts", f.url, opts.retries);
            if consecutive_failures >= opts.cooldown_failures {
                log::warn!(
                    "{} failures in a row - cooling down {}s before continuing \
                     (rate limit?). The run does not stop.",
                    consecutive_failures,
                    opts.cooldown_seconds,
                );
                thread::sleep(Duration::from_secs(opts.cooldown_seconds));
            }
        }
    }
    Ok(())
}

fn json_entry(dest: &Path, size: u64, name: &str) -> Value {
    serde_json::json!({
        "path": dest.display().to_string(),
        "size": size,
        "name": name,
    })
}

/// 0807 pipeline (python run_0807).
pub fn run_0807(
    urls: &[FoundUrl],
    opts: &RunOpts,
    out_dir: &Path,
    state: &mut State,
    source_by_url: &HashMap<String, SourceInfo>,
    mut gstore: Option<&mut GlobalStore>,
    manifest: &Path,
) -> anyhow::Result<()> {
    if !urls
        .iter()
        .any(|f| !is_done(state, gstore.as_deref(), &f.url))
    {
        return Ok(());
    }
    let out_dir = out_dir.join("downloads").join("0807");
    let mut consecutive_failures: u32 = 0;
    for f in urls {
        if is_done(state, gstore.as_deref(), &f.url) {
            log::info!("Skip (already downloaded): {}", f.url);
            continue;
        }
        let source = source_by_url.get(&f.url).cloned().unwrap_or_default();
        let mut resolved = false;
        for attempt in 1..=opts.retries {
            let downloaded = zero807::download(&f.url, &out_dir)
                .map_err(anyhow::Error::new)
                .and_then(|(dest, size)| -> anyhow::Result<(Value, u64)> {
                    let mut entry = json_entry(
                        &dest,
                        size,
                        &dest
                            .file_name()
                            .map(|n| n.to_string_lossy().into_owned())
                            .unwrap_or_default(),
                    );
                    if opts.hash {
                        let h = sha256_of(&dest)?;
                        entry["sha256"] = Value::String(h);
                    }
                    Ok((entry, size))
                });
            match downloaded {
                Ok((entry, size)) => {
                    log::info!("OK {} ({})", f.url, format_size(size));
                    record_files(
                        state,
                        gstore.as_deref_mut(),
                        &f.url,
                        std::slice::from_ref(&entry),
                        &source,
                        manifest,
                    )?;
                    if opts.extract {
                        extract_zips(std::slice::from_ref(&entry));
                    }
                    consecutive_failures = 0;
                    resolved = true;
                    break;
                }
                Err(err) => {
                    let Some(ze) = err.downcast_ref::<Zero807Error>() else {
                        // a bug must not stop the run
                        log::error!("Unexpected error for {}: {err:#}", f.url);
                        let msg = format!("unexpected: {err}");
                        state.mark(&f.url, |r| {
                            r.error = Some(msg);
                            r.files = Vec::new();
                        })?;
                        if attempt < opts.retries {
                            thread::sleep(Duration::from_secs(opts.retry_delay * attempt as u64));
                        }
                        continue;
                    };
                    match ze {
                        Zero807Error::NotFound(msg) => {
                            // Dead links never come back - no retry
                            log::error!("DEAD LINK {}: {msg}", f.url);
                            state.mark(&f.url, |r| {
                                r.status = "failed".into();
                                r.error = Some(msg.clone());
                                r.files = Vec::new();
                            })?;
                            resolved = true;
                            break;
                        }
                        Zero807Error::Error(msg) => {
                            log::warn!(
                                "Attempt {}/{} failed for {}: {msg}",
                                attempt,
                                opts.retries,
                                f.url
                            );
                            state.mark(&f.url, |r| {
                                r.error = Some(msg.clone());
                                r.files = Vec::new();
                            })?;
                            if attempt < opts.retries {
                                thread::sleep(Duration::from_secs(
                                    opts.retry_delay * attempt as u64,
                                ));
                            }
                        }
                    }
                }
            }
        }
        if !resolved {
            let err = state
                .records
                .get(&f.url)
                .and_then(|r| r.error.clone())
                .unwrap_or_else(|| "unknown".into());
            state.mark(&f.url, |r| {
                r.status = "failed".into();
                r.error = Some(err);
                r.files = Vec::new();
            })?;
            consecutive_failures += 1;
            log::error!("FAILED {} after {} attempts", f.url, opts.retries);
            if consecutive_failures >= opts.cooldown_failures {
                log::warn!(
                    "{} failures in a row - cooling down {}s before continuing \
                     (HTTP 403 storm?). The run does not stop.",
                    consecutive_failures,
                    opts.cooldown_seconds,
                );
                thread::sleep(Duration::from_secs(opts.cooldown_seconds));
            }
        }
    }
    Ok(())
}

/// Load subscriptions.json: [{"name", "channel_id", "out_dir"}, ...].
pub fn load_subscriptions(path: &Path) -> anyhow::Result<Vec<(String, String, Option<String>)>> {
    let data: Value = serde_json::from_str(
        &fs::read_to_string(path).with_context(|| format!("reading {}", path.display()))?,
    )?;
    let entries = match &data {
        Value::Object(o) => o
            .get("subscriptions")
            .cloned()
            .unwrap_or_else(|| Value::Array(vec![])),
        Value::Array(_) => data.clone(),
        _ => anyhow::bail!(
            "{}: expected a list or {{\"subscriptions\": [...]}}",
            path.display()
        ),
    };
    let mut subs = Vec::new();
    for entry in entries.as_array().context("subscriptions must be a list")? {
        let bad = || {
            anyhow!(
                "{}: each subscription needs 'channel_id' and 'out_dir' (got: {entry})",
                path.display()
            )
        };
        let channel_id = entry
            .get("channel_id")
            .and_then(Value::as_str)
            .ok_or_else(bad)?;
        let out_dir = entry
            .get("out_dir")
            .and_then(Value::as_str)
            .ok_or_else(bad)?;
        let name = entry
            .get("name")
            .and_then(Value::as_str)
            .map(str::to_string);
        subs.push((channel_id.to_string(), out_dir.to_string(), name));
    }
    Ok(subs)
}

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

    #[test]
    fn format_size_matches_python() {
        // python format_size values
        assert_eq!(format_size(0), "0.0 B");
        assert_eq!(format_size(512), "512.0 B");
        assert_eq!(format_size(1024), "1.0 KiB");
        assert_eq!(format_size(1536), "1.5 KiB");
        assert_eq!(format_size(1024 * 1024), "1.0 MiB");
        assert_eq!(format_size(1024_u64 * 1024 * 1024), "1.0 GiB");
        assert_eq!(format_size(1024_u64 * 1024 * 1024 * 1024), "1.0 TiB");
    }

    #[test]
    fn read_url_lines_skips_comments_and_blanks() {
        let dir = tempfile::tempdir().unwrap();
        let p = dir.path().join("urls.txt");
        fs::write(&p, "\n# comment\nhttps://a \n\nhttps://b\n# x\n").unwrap();
        let lines = read_url_lines(&p).unwrap();
        assert_eq!(lines, vec!["https://a", "https://b"]);
    }

    #[test]
    fn load_urls_file_sorts_and_dedupes() {
        let dir = tempfile::tempdir().unwrap();
        let p = dir.path().join("urls.txt");
        fs::write(
            &p,
            "https://gofile.io/d/b\nhttps://0807.st/d/x\nhttps://gofile.io/d/a\n",
        )
        .unwrap();
        let found = load_urls_file(&p).unwrap();
        let urls: Vec<&str> = found.iter().map(|f| f.url.as_str()).collect();
        assert_eq!(
            urls,
            vec![
                "https://0807.st/d/x",
                "https://gofile.io/d/a",
                "https://gofile.io/d/b"
            ]
        );
        assert!(found.iter().all(|f| f.message_id.is_none()));
    }

    #[test]
    fn load_urls_file_warns_unsupported() {
        let dir = tempfile::tempdir().unwrap();
        let p = dir.path().join("urls.txt");
        fs::write(&p, "https://example.com/x\n").unwrap();
        let found = load_urls_file(&p).unwrap();
        assert!(found.is_empty());
    }

    #[test]
    fn write_failed_urls_empty_and_filled() {
        let dir = tempfile::tempdir().unwrap();
        let mut state = State::load(dir.path().join("state.json"));
        let dest = write_failed_urls(dir.path(), &state).unwrap();
        assert_eq!(fs::read_to_string(&dest).unwrap(), "");
        state
            .mark("https://gofile.io/d/a", |r| r.status = "failed".into())
            .unwrap();
        state
            .mark("https://gofile.io/d/b", |r| r.status = "ok".into())
            .unwrap();
        let dest = write_failed_urls(dir.path(), &state).unwrap();
        assert_eq!(
            fs::read_to_string(&dest).unwrap(),
            "https://gofile.io/d/a\n"
        );
    }

    #[test]
    fn record_files_writes_state_global_and_manifest() {
        let dir = tempfile::tempdir().unwrap();
        let mut state = State::load(dir.path().join("state.json"));
        let manifest = dir.path().join("manifest.jsonl");
        let mut gs = GlobalStore::load(dir.path().join("urls.json"));
        let source = SourceInfo {
            channel: "123".into(),
            host: Some("gofile".into()),
            message_id: Some(Value::from(42)),
            timestamp: Some("2024-01-01".into()),
            author: Some("ben".into()),
            manifest: manifest.clone(),
        };
        let files = vec![serde_json::json!({"path": "/tmp/x.7z", "size": 10, "name": "x.7z"})];
        record_files(
            &mut state,
            Some(&mut gs),
            "https://gofile.io/d/a",
            &files,
            &source,
            &manifest,
        )
        .unwrap();
        assert_eq!(state.records["https://gofile.io/d/a"].status, "ok");
        assert_eq!(state.records["https://gofile.io/d/a"].files.len(), 1);
        assert!(gs.is_done("https://gofile.io/d/a"));
        let mtext = fs::read_to_string(&manifest).unwrap();
        let row: Value = serde_json::from_str(mtext.trim()).unwrap();
        assert_eq!(row["channel"], "123");
        assert_eq!(row["message_id"], 42);
        assert_eq!(row["path"], "/tmp/x.7z");
        assert_eq!(row["name"], "x.7z");
    }

    #[test]
    fn extract_zips_skips_missing_and_non_zip() {
        // non-zip entries are ignored, no panic
        extract_zips(&[serde_json::json!({"path": "/definitely/not/there.7z"})]);
        extract_zips(&[]);
    }

    #[test]
    fn fisher_yates_is_permutation() {
        let mut v: Vec<u32> = (0..50).collect();
        fisher_yates(&mut v);
        let mut sorted = v.clone();
        sorted.sort_unstable();
        assert_eq!(sorted, (0..50).collect::<Vec<_>>());
    }

    #[test]
    fn load_subscriptions_both_shapes() {
        let dir = tempfile::tempdir().unwrap();
        let p = dir.path().join("subs.json");
        fs::write(
            &p,
            r#"{"subscriptions": [{"name": "A", "channel_id": "1", "out_dir": "out_a"}]}"#,
        )
        .unwrap();
        let subs = load_subscriptions(&p).unwrap();
        assert_eq!(
            subs,
            vec![("1".to_string(), "out_a".to_string(), Some("A".to_string()))]
        );
        fs::write(&p, r#"[{"channel_id": "2", "out_dir": "out_b"}]"#).unwrap();
        let subs = load_subscriptions(&p).unwrap();
        assert_eq!(subs, vec![("2".to_string(), "out_b".to_string(), None)]);
    }

    #[test]
    fn load_subscriptions_rejects_missing_fields() {
        let dir = tempfile::tempdir().unwrap();
        let p = dir.path().join("subs.json");
        fs::write(&p, r#"[{"channel_id": "2"}]"#).unwrap();
        assert!(load_subscriptions(&p).is_err());
    }

    #[test]
    fn parse_since_relative_and_absolute() {
        let now = Utc::now();
        let cutoff = parse_since("5d").unwrap();
        assert!(cutoff <= now);
        assert!(cutoff > now - chrono::Duration::days(6));
        assert!(parse_since("12h").unwrap() > now - chrono::Duration::hours(13));
        assert_eq!(
            parse_since("2026-10-04").unwrap().to_rfc3339(),
            "2026-10-04T00:00:00+00:00"
        );
        assert_eq!(
            parse_since("2026-10-04T09:30:00Z").unwrap().to_rfc3339(),
            "2026-10-04T09:30:00+00:00"
        );
        assert_eq!(
            parse_since("2026-10-04 09:30:00").unwrap().to_rfc3339(),
            "2026-10-04T09:30:00+00:00"
        );
        assert!(parse_since("nonsense").is_err());
    }

    #[test]
    fn torrent_detection() {
        assert!(is_torrent_url("https://0807.st/rh0BzdD.torrent"));
        assert!(is_torrent_url("https://0807.st/x.TORRENT"));
        assert!(!is_torrent_url("https://0807.st/rh0BzdD.zip"));
        assert!(is_torrent_file("Lord of Skulls.torrent"));
        assert!(!is_torrent_file("Lord of Skulls.rar"));
    }

    #[test]
    fn parse_message_ts_handles_export_formats() {
        assert_eq!(
            parse_message_ts("2026-01-01T00:00:00")
                .unwrap()
                .to_rfc3339(),
            "2026-01-01T00:00:00+00:00"
        );
        assert!(parse_message_ts("2026-01-01T00:00:00.123000+00:00").is_some());
        assert!(parse_message_ts("garbage").is_none());
    }
}