//! Recovery of incomplete/cancelled/crashed production runs (design §7/§12.6).
//!
//! Every production run directory holds an exclusive `run.lock` while a
//! worker writes it and an append-only `progress.jsonl` journal with one
//! `{"path","sha256"}` line per published target PNG (fsynced after the PNG
//! itself was published). A run may be resumed only when its `render.json`
//! is byte-for-byte the same contract document that the current job would
//! write (project, sex/set, body/equipment SHA-256, target, FPS, framing,
//! pivot, camera, clip inventory, maps, counts). Before any rendering every
//! planned PNG is checked: journaled files whose bytes match their digest are
//! kept untouched; missing ones are regenerated; corrupt files and foreign
//! links are moved aside (`<name>.corrupt-aNNN-K`, never deleted); existing
//! files without a journal entry (crash between publish and journal) are
//! compared with the regenerated bytes and kept when identical.
use crate::digest;
use serde_json::Value;
use std::{
    collections::{BTreeMap, BTreeSet, HashSet},
    fs::{self, File, OpenOptions},
    io::{Read, Write},
    path::{Path, PathBuf},
};

pub const JOURNAL: &str = "progress.jsonl";
pub const LOCK: &str = "run.lock";
pub const RESUME_LOG: &str = "resume.jsonl";
const MAX_RENDER_JSON: u64 = 1024 * 1024;
const MAX_JOURNAL: u64 = 256 * 1024 * 1024;
const MAX_CANDIDATES: usize = 20;
const MAX_SCANNED_RUNS: usize = 2000;

/// Outcome of the pre-render verification of a resumed run.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ResumeReport {
    pub attempt: u32,
    pub planned: usize,
    pub verified: usize,
    pub missing: usize,
    pub quarantined: Vec<String>,
    pub unjournaled: usize,
    pub damaged_journal_lines: usize,
    pub previous_marker: Option<String>,
}

/// Open write session on one run directory (fresh or resumed).
pub struct RunSession {
    pub root: PathBuf,
    pub attempt: u32,
    /// Planned PNGs already on disk with a verified journal checksum.
    pub verified: BTreeMap<String, String>,
    /// Existing regular PNGs without a journal entry; compared on regeneration.
    pub unjournaled: BTreeSet<String>,
    pub report: ResumeReport,
    journal: File,
    _lock: File,
}

fn io(e: std::io::Error) -> String {
    e.to_string()
}

fn lock_run(root: &Path) -> Result<File, String> {
    let path = root.join(LOCK);
    if let Ok(meta) = fs::symlink_metadata(&path) {
        if !meta.is_file() {
            return Err("run.lock is geen regulier bestand".into());
        }
    }
    let file = OpenOptions::new()
        .create(true)
        .truncate(false)
        .write(true)
        .open(&path)
        .map_err(io)?;
    match file.try_lock() {
        Ok(()) => Ok(file),
        Err(fs::TryLockError::WouldBlock) => {
            Err("deze run wordt op dit moment al gerenderd (run.lock bezet)".into())
        }
        Err(fs::TryLockError::Error(e)) => Err(format!("run.lock: {e}")),
    }
}

fn is_locked(root: &Path) -> bool {
    let Ok(file) = OpenOptions::new().write(true).open(root.join(LOCK)) else {
        return false;
    };
    matches!(file.try_lock(), Err(fs::TryLockError::WouldBlock))
}

fn open_journal(root: &Path) -> Result<File, String> {
    let path = root.join(JOURNAL);
    if let Ok(meta) = fs::symlink_metadata(&path) {
        if !meta.is_file() {
            return Err("progress.jsonl is geen regulier bestand".into());
        }
    }
    OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .map_err(io)
}

fn read_regular(path: &Path, max: u64) -> Result<Option<Vec<u8>>, String> {
    let meta = match fs::symlink_metadata(path) {
        Ok(meta) => meta,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(e.to_string()),
    };
    if !meta.is_file() {
        return Err(format!("{} is geen regulier bestand", path.display()));
    }
    if meta.len() > max {
        return Err(format!("{} groter dan {max} bytes", path.display()));
    }
    let mut bytes = Vec::new();
    File::open(path)
        .map_err(io)?
        .take(max + 1)
        .read_to_end(&mut bytes)
        .map_err(io)?;
    if bytes.len() as u64 > max {
        return Err(format!("{} groter dan {max} bytes", path.display()));
    }
    Ok(Some(bytes))
}

/// Journal entries: last valid entry per path wins; returns (entries, damaged lines, ends with newline).
pub fn read_journal(root: &Path) -> Result<(BTreeMap<String, String>, usize, bool), String> {
    let Some(bytes) = read_regular(&root.join(JOURNAL), MAX_JOURNAL)? else {
        return Ok((BTreeMap::new(), 0, true));
    };
    let mut entries = BTreeMap::new();
    let mut damaged = 0;
    for line in bytes.split(|b| *b == b'\n') {
        if line.is_empty() {
            continue;
        }
        let parsed = serde_json::from_slice::<Value>(line).ok().and_then(|v| {
            let path = v.get("path")?.as_str()?.to_string();
            let sha = v.get("sha256")?.as_str()?.to_string();
            digest::is_sha256_hex(&sha).then_some((path, sha))
        });
        match parsed {
            Some((path, sha)) => {
                entries.insert(path, sha);
            }
            None => damaged += 1,
        }
    }
    Ok((entries, damaged, bytes.last().is_none_or(|b| *b == b'\n')))
}

/// Move a bad file/link aside inside the run without deleting or clobbering anything.
fn quarantine(root: &Path, rel: &str, attempt: u32) -> Result<String, String> {
    let path = root.join(rel);
    let name = path
        .file_name()
        .and_then(|n| n.to_str())
        .ok_or("ongeldige bestandsnaam")?
        .to_string();
    for n in 0..1000 {
        let target = path.with_file_name(format!("{name}.corrupt-a{attempt:03}-{n}"));
        match fs::hard_link(&path, &target) {
            Ok(()) => {
                fs::remove_file(&path).map_err(io)?;
                let rel_target = Path::new(rel)
                    .with_file_name(target.file_name().unwrap_or_default())
                    .to_string_lossy()
                    .replace('\\', "/");
                return Ok(rel_target);
            }
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(e) => return Err(format!("quarantaine van {rel} mislukt: {e}")),
        }
    }
    Err(format!("geen vrije quarantainenaam voor {rel}"))
}

/// Move a file to a fresh unique name (no clobber), returning the new file name.
fn archive(root: &Path, name: &str, prefix: &str) -> Result<String, String> {
    for n in 0..1000 {
        let target = if n == 0 {
            format!("{prefix}-{name}")
        } else {
            format!("{prefix}-{n}-{name}")
        };
        match fs::hard_link(root.join(name), root.join(&target)) {
            Ok(()) => {
                fs::remove_file(root.join(name)).map_err(io)?;
                return Ok(target);
            }
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(e) => return Err(format!("archiveren van {name} mislukt: {e}")),
        }
    }
    Err(format!("geen vrije archiefnaam voor {name}"))
}

/// Top-level keys whose values differ between two JSON objects.
pub fn contract_differences(existing: &Value, expected: &Value) -> Vec<String> {
    match (existing.as_object(), expected.as_object()) {
        (Some(a), Some(b)) => a
            .keys()
            .chain(b.keys())
            .collect::<BTreeSet<_>>()
            .into_iter()
            .filter(|k| a.get(*k) != b.get(*k))
            .cloned()
            .collect(),
        _ => vec!["render.json".into()],
    }
}

fn normalized(doc: &Value) -> Result<Value, String> {
    serde_json::from_str(&serde_json::to_string(doc).map_err(|e| e.to_string())?)
        .map_err(|e| e.to_string())
}

fn read_render(root: &Path) -> Result<Value, String> {
    let bytes =
        read_regular(&root.join("render.json"), MAX_RENDER_JSON)?.ok_or("render.json ontbreekt")?;
    serde_json::from_slice(&bytes).map_err(|e| format!("render.json: {e}"))
}

fn check_contract(root: &Path, render_doc: &Value) -> Result<(), String> {
    let existing = read_render(root)?;
    let expected = normalized(render_doc)?;
    if existing != expected {
        return Err(format!(
            "huidige instellingen/bronnen passen niet bij deze run (verschil in: {})",
            contract_differences(&existing, &expected).join(", ")
        ));
    }
    Ok(())
}

/// Direct `run-*` child directory of the output directory, not a link.
fn run_root(output: &Path, dir: &Path) -> Result<PathBuf, String> {
    let meta = fs::symlink_metadata(dir).map_err(io)?;
    if !meta.is_dir() {
        return Err("runmap is geen gewone map".into());
    }
    let out = output.canonicalize().map_err(io)?;
    let root = dir.canonicalize().map_err(io)?;
    let name_ok = root
        .file_name()
        .and_then(|n| n.to_str())
        .is_some_and(|n| n.starts_with("run-"));
    if root.parent() != Some(out.as_path()) || !name_ok {
        return Err("runmap ligt niet direct in de uitvoermap".into());
    }
    Ok(root)
}

impl RunSession {
    /// A new, empty run directory created by the caller.
    pub fn fresh(root: &Path) -> Result<Self, String> {
        let lock = lock_run(root)?;
        let journal = open_journal(root)?;
        Ok(Self {
            root: root.to_path_buf(),
            attempt: 0,
            verified: BTreeMap::new(),
            unjournaled: BTreeSet::new(),
            report: ResumeReport::default(),
            journal,
            _lock: lock,
        })
    }

    /// Lock, contract-check and verify an incomplete run before re-rendering.
    /// `planned` lists every relative target PNG of the job in render order.
    pub fn resume(
        output: &Path,
        dir: &Path,
        render_doc: &Value,
        planned: &[String],
        progress: &mut dyn FnMut(usize, usize),
    ) -> Result<Self, String> {
        let root = run_root(output, dir)?;
        if fs::symlink_metadata(root.join("complete.json")).is_ok() {
            return Err("deze run is al voltooid (complete.json bestaat)".into());
        }
        let lock = lock_run(&root)?;
        check_contract(&root, render_doc)?;
        let planned_set: HashSet<&str> = planned.iter().map(String::as_str).collect();
        let (journal_entries, damaged, newline) = read_journal(&root)?;
        if let Some(unknown) = journal_entries
            .keys()
            .find(|k| !planned_set.contains(k.as_str()))
        {
            return Err(format!("journal noemt een onbekend pad: {unknown}"));
        }
        let previous_attempts = read_regular(&root.join(RESUME_LOG), MAX_RENDER_JSON)?
            .map(|b| b.split(|c| *c == b'\n').filter(|l| !l.is_empty()).count())
            .unwrap_or(0);
        let attempt = previous_attempts as u32 + 1;
        let mut report = ResumeReport {
            attempt,
            planned: planned.len(),
            damaged_journal_lines: damaged,
            ..Default::default()
        };
        let mut verified = BTreeMap::new();
        let mut unjournaled = BTreeSet::new();
        for (i, rel) in planned.iter().enumerate() {
            if i % 256 == 0 {
                progress(i, planned.len());
            }
            let path = root.join(rel);
            let meta = match fs::symlink_metadata(&path) {
                Ok(meta) => meta,
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                    report.missing += 1;
                    continue;
                }
                Err(e) => return Err(e.to_string()),
            };
            if meta.is_dir() {
                return Err(format!("map staat op de plaats van PNG {rel}"));
            }
            if !meta.is_file() || meta.len() > crate::export_set::MAX_PNG_BYTES {
                report.quarantined.push(quarantine(&root, rel, attempt)?);
                continue;
            }
            match journal_entries.get(rel) {
                Some(expected) => {
                    let bytes = fs::read(&path).map_err(io)?;
                    if digest::sha256_hex(&bytes) == *expected {
                        verified.insert(rel.clone(), expected.clone());
                    } else {
                        report.quarantined.push(quarantine(&root, rel, attempt)?);
                    }
                }
                None => {
                    unjournaled.insert(rel.clone());
                }
            }
        }
        progress(planned.len(), planned.len());
        report.verified = verified.len();
        report.unjournaled = unjournaled.len();
        for marker in ["cancelled.json", "failed.json"] {
            if fs::symlink_metadata(root.join(marker)).is_ok() {
                let archived = archive(&root, marker, &format!("attempt-{:03}", attempt - 1))?;
                report.previous_marker = Some(archived);
            }
        }
        let mut journal = open_journal(&root)?;
        if !newline {
            journal.write_all(b"\n").map_err(io)?;
            journal.sync_data().map_err(io)?;
        }
        let entry = serde_json::json!({
            "attempt": attempt,
            "planned_png_files": report.planned,
            "verified": report.verified,
            "missing": report.missing,
            "quarantined": report.quarantined,
            "unjournaled_existing": report.unjournaled,
            "damaged_journal_lines": report.damaged_journal_lines,
            "previous_marker": report.previous_marker,
        });
        let mut log = OpenOptions::new()
            .create(true)
            .append(true)
            .open(root.join(RESUME_LOG))
            .map_err(io)?;
        log.write_all(format!("{entry}\n").as_bytes()).map_err(io)?;
        log.sync_data().map_err(io)?;
        Ok(Self {
            root,
            attempt,
            verified,
            unjournaled,
            report,
            journal,
            _lock: lock,
        })
    }

    fn record(&mut self, rel: &str, sha: &str) -> Result<(), String> {
        let line = serde_json::json!({"path": rel, "sha256": sha}).to_string() + "\n";
        self.journal.write_all(line.as_bytes()).map_err(io)?;
        self.journal.sync_data().map_err(io)
    }

    /// Publish encoded PNG bytes for a planned path, never overwriting a file in place.
    pub fn publish(&mut self, rel: &str, bytes: &[u8]) -> Result<String, String> {
        if let Some(sha) = self.verified.get(rel) {
            return Ok(sha.clone());
        }
        let sha = digest::sha256_hex(bytes);
        if self.unjournaled.remove(rel) {
            let path = self.root.join(rel);
            match fs::symlink_metadata(&path) {
                Ok(meta) if meta.is_file() => {
                    let existing = fs::read(&path).map_err(io)?;
                    if existing == bytes {
                        self.record(rel, &sha)?;
                        return Ok(sha);
                    }
                    let q = quarantine(&self.root, rel, self.attempt)?;
                    self.report.quarantined.push(q);
                }
                Ok(_) => {
                    let q = quarantine(&self.root, rel, self.attempt)?;
                    self.report.quarantined.push(q);
                }
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(e.to_string()),
            }
        }
        crate::production::publish_bytes(&self.root, Path::new(rel), bytes)?;
        self.record(rel, &sha)?;
        Ok(sha)
    }
}

/// An incomplete run of the current sex/set found in the output directory.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResumeCandidate {
    pub dir: PathBuf,
    pub name: String,
    /// Distinct planned PNGs recorded in the journal (not yet re-verified).
    pub journaled: usize,
    pub planned: u64,
    /// "geannuleerd", "mislukt" or "afgebroken" (no terminal marker: crash/kill).
    pub state: String,
    /// Why this run cannot be resumed now; None when resumable.
    pub blocked: Option<String>,
}

fn run_stamp(name: &str) -> u128 {
    name.strip_prefix("run-")
        .and_then(|r| r.split('-').next())
        .and_then(|n| n.parse().ok())
        .unwrap_or(0)
}

/// Incomplete runs of the same sex/set as `render_doc`, newest first.
pub fn candidates(output: &Path, render_doc: &Value) -> Result<Vec<ResumeCandidate>, String> {
    let expected = normalized(render_doc)?;
    let read = match fs::read_dir(output) {
        Ok(read) => read,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(e) => return Err(e.to_string()),
    };
    let mut found = Vec::new();
    for entry in read.take(MAX_SCANNED_RUNS) {
        let entry = entry.map_err(io)?;
        let name = entry.file_name().to_string_lossy().to_string();
        let Ok(kind) = entry.file_type() else {
            continue;
        };
        if !name.starts_with("run-") || !kind.is_dir() {
            continue;
        }
        let dir = entry.path();
        if fs::symlink_metadata(dir.join("complete.json")).is_ok() {
            continue;
        }
        let Ok(existing) = read_render(&dir) else {
            continue;
        };
        if existing.get("sex") != expected.get("sex") || existing.get("set") != expected.get("set")
        {
            continue;
        }
        let state = if dir.join("cancelled.json").exists() {
            "geannuleerd"
        } else if dir.join("failed.json").exists() {
            "mislukt"
        } else {
            "afgebroken"
        };
        let journaled = read_journal(&dir).map(|(e, _, _)| e.len()).unwrap_or(0);
        let blocked = if is_locked(&dir) {
            Some("wordt op dit moment gerenderd".to_string())
        } else if existing != expected {
            Some(format!(
                "past niet meer bij de huidige instellingen/bronnen (verschil in: {})",
                contract_differences(&existing, &expected).join(", ")
            ))
        } else {
            None
        };
        found.push(ResumeCandidate {
            dir,
            planned: existing
                .get("png_files")
                .and_then(Value::as_u64)
                .unwrap_or(0),
            name,
            journaled,
            state: state.into(),
            blocked,
        });
    }
    found.sort_by_key(|c| std::cmp::Reverse(run_stamp(&c.name)));
    found.truncate(MAX_CANDIDATES);
    Ok(found)
}