av1converter 3.0.0

use crate::encoder::command_builder::{EncodingParams, build_ffmpeg_args};
use crate::i18n::{Msg, t};
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::thread;
use std::time::Duration;

static JOB_COUNTER: AtomicU64 = AtomicU64::new(0);

/// Scratch path an in-progress encode is written to, alongside the real output
/// and keeping its extension so `FFmpeg` still infers the container. Published
/// to the destination by [`publish_partial`] once the encode finishes.
fn partial_output_path(output: &str, tag: &str) -> String {
    let path = Path::new(output);
    let parent = path.parent().unwrap_or(Path::new("."));
    let stem = path
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("output");
    let name = match path.extension().and_then(|e| e.to_str()) {
        Some(ext) => format!("{stem}.part.{tag}.{ext}"),
        None => format!("{stem}.part.{tag}"),
    };
    parent.join(name).to_string_lossy().into_owned()
}

pub(crate) use crate::utils::child::pid_alive;

/// Scratch files left next to `output` by an encode that never finished, as
/// produced by [`partial_output_path`]. The match is by shape — both halves of
/// the `{pid}_{counter}` tag must be digits, and a file that merely contains
/// `.part.` is not scratch — and the tagged pid must no longer be running.
pub fn orphaned_partials(output: &Path) -> Vec<PathBuf> {
    let (Some(parent), Some(stem)) = (output.parent(), output.file_stem().and_then(|s| s.to_str()))
    else {
        return Vec::new();
    };
    let prefix = format!("{stem}.part.");
    let suffix = output
        .extension()
        .and_then(|e| e.to_str())
        .map(|ext| format!(".{ext}"));

    let Ok(entries) = std::fs::read_dir(parent) else {
        return Vec::new();
    };
    entries
        .flatten()
        .filter(|entry| {
            let name = entry.file_name();
            let Some(name) = name.to_str() else {
                return false;
            };
            let Some(rest) = name.strip_prefix(&prefix) else {
                return false;
            };
            let tag = match suffix.as_deref() {
                Some(suffix) => match rest.strip_suffix(suffix) {
                    Some(tag) => tag,
                    None => return false,
                },
                None => rest,
            };
            matches!(tag.split_once('_'), Some((pid, uid))
                if !pid.is_empty()
                    && !uid.is_empty()
                    && pid.bytes().all(|b| b.is_ascii_digit())
                    && uid.bytes().all(|b| b.is_ascii_digit())
                    && pid.parse::<u32>().is_ok_and(|pid| !pid_alive(pid)))
        })
        .map(|entry| entry.path())
        .collect()
}

/// Whether both paths designate the same file: literally equal, or resolving to
/// the same canonical target when both already exist.
fn is_same_file(a: &Path, b: &Path) -> bool {
    a == b
        || match (a.canonicalize(), b.canonicalize()) {
            (Ok(a), Ok(b)) => a == b,
            _ => false,
        }
}

fn path_occupied(path: &Path) -> bool {
    std::fs::symlink_metadata(path).is_ok()
}

/// Progress callback type
pub type ProgressCallback = Box<dyn FnMut(f64) + Send>;

/// Encoding result
#[derive(Debug)]
pub enum EncodeResult {
    /// Encoding completed successfully
    Success,
    /// Encoding was cancelled
    Cancelled,
    /// Encoding failed
    Error(String),
}

/// Encode a video file using `FFmpeg`
#[allow(clippy::too_many_lines)]
pub fn encode_video(
    params: &EncodingParams,
    progress_callback: Option<ProgressCallback>,
    cancel_flag: &AtomicBool,
    duration: f64,
    total_frames: f64,
) -> EncodeResult {
    // FFmpeg cannot edit a file in place; refuse before spawning.
    if is_same_file(Path::new(&params.input), Path::new(&params.output)) {
        return EncodeResult::Error(t(params.lang, Msg::ErrOutputIsInput).to_string());
    }
    if path_occupied(Path::new(&params.output)) {
        return EncodeResult::Error(t(params.lang, Msg::ErrOutputExists).to_string());
    }

    // Reserved with `create_new` before FFmpeg sees it; `-y` never erases a
    // real file sitting at the `.part` name.
    let (partial, tag) = loop {
        let uid = JOB_COUNTER.fetch_add(1, Ordering::Relaxed);
        let tag = format!("{}_{}", std::process::id(), uid);
        let partial = partial_output_path(&params.output, &tag);
        match OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&partial)
        {
            Ok(_) => break (partial, tag),
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {}
            Err(e) => {
                return EncodeResult::Error(format!("Failed to reserve temporary output: {e}"));
            }
        }
    };
    let mut encode_params = params.clone();
    encode_params.output.clone_from(&partial);
    let args = build_ffmpeg_args(&encode_params);

    // Create progress file
    let progress_file = match crate::utils::scratch_path(&format!("av1c_progress_{tag}.txt")) {
        Ok(path) => path,
        Err(e) => {
            let _ = std::fs::remove_file(&partial);
            return EncodeResult::Error(e);
        }
    };
    if File::create(&progress_file).is_err() {
        let _ = std::fs::remove_file(&partial);
        return EncodeResult::Error("Failed to create progress file".to_string());
    }

    // Insert progress args after -nostdin
    let mut args = args;
    args.insert(2, "-progress".to_string());
    args.insert(3, progress_file.to_string_lossy().to_string());

    // stderr goes to a temp file, not a pipe.
    let stderr_path = match crate::utils::scratch_path(&format!("av1c_stderr_{tag}.txt")) {
        Ok(path) => path,
        Err(e) => {
            let _ = std::fs::remove_file(&progress_file);
            let _ = std::fs::remove_file(&partial);
            return EncodeResult::Error(e);
        }
    };
    let stderr_file = match File::create(&stderr_path) {
        Ok(f) => f,
        Err(e) => {
            let _ = std::fs::remove_file(&progress_file);
            let _ = std::fs::remove_file(&partial);
            return EncodeResult::Error(format!("Failed to create stderr file: {e}"));
        }
    };

    // Start FFmpeg
    let mut ffmpeg = Command::new("ffmpeg");
    ffmpeg
        .args(&args)
        .stdout(Stdio::null())
        .stderr(Stdio::from(stderr_file));
    let (mut child, _child) = match crate::utils::child::spawn(&mut ffmpeg) {
        Ok(spawned) => spawned,
        Err(e) => {
            let _ = std::fs::remove_file(&progress_file);
            let _ = std::fs::remove_file(&stderr_path);
            let _ = std::fs::remove_file(&partial);
            return EncodeResult::Error(format!("Failed to start ffmpeg: {e}"));
        }
    };

    // A panic in the loop still kills ffmpeg and removes the scratch file.
    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        run_encode_loop(
            &mut child,
            &progress_file,
            duration,
            total_frames,
            progress_callback,
            cancel_flag,
            &partial,
            &stderr_path,
            params.remux_only.then(|| params.input.clone()).as_ref(),
            params.lang,
        )
    }))
    .unwrap_or_else(|_| {
        crate::utils::child::kill_and_wait(&mut child);
        let _ = std::fs::remove_file(&partial);
        EncodeResult::Error("encode panicked".to_string())
    });

    // Cleanup
    let _ = std::fs::remove_file(&progress_file);
    let _ = std::fs::remove_file(&stderr_path);

    // A cancel that wins the race against a successful ffmpeg exit is reported
    // as Cancelled and leaves no finished output behind.
    let result = if matches!(result, EncodeResult::Success) && cancel_flag.load(Ordering::Acquire) {
        let _ = std::fs::remove_file(&partial);
        EncodeResult::Cancelled
    } else {
        result
    };

    // The scratch file becomes the output only now, when the encode is known to
    // have succeeded. Failure and cancellation already removed it. `hard_link`
    // fails with `AlreadyExists` instead of replacing an existing destination;
    // filesystems without hard links fall back to an exclusive create+copy,
    // which never clobbers a destination that appears mid-publish.
    if matches!(result, EncodeResult::Success)
        && let Err(message) = publish_partial(&partial, &params.output, params.lang)
    {
        let _ = std::fs::remove_file(&partial);
        return EncodeResult::Error(message);
    }

    result
}

/// Move `partial` onto `output` without ever replacing an existing file.
fn publish_partial(partial: &str, output: &str, lang: crate::i18n::Language) -> Result<(), String> {
    match std::fs::hard_link(partial, output) {
        Ok(()) => {
            let _ = std::fs::remove_file(partial);
            Ok(())
        }
        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
            Err(t(lang, Msg::ErrOutputAppeared).to_string())
        }
        Err(_) => {
            // Exclusive create: fails atomically if the destination exists.
            match OpenOptions::new().write(true).create_new(true).open(output) {
                Ok(mut dest) => {
                    let copy = File::open(partial)
                        .and_then(|mut src| std::io::copy(&mut src, &mut dest).map(|_| ()));
                    match copy {
                        Ok(()) => {
                            let _ = std::fs::remove_file(partial);
                            Ok(())
                        }
                        Err(e) => {
                            let _ = std::fs::remove_file(output);
                            Err(format!("Failed to move the encoded file into place: {e}"))
                        }
                    }
                }
                Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                    Err(t(crate::i18n::Language::English, Msg::ErrOutputAppeared).to_string())
                }
                Err(e) => Err(format!("Failed to move the encoded file into place: {e}")),
            }
        }
    }
}

/// Run the encoding loop with progress updates
///
#[allow(clippy::too_many_arguments)]
fn run_encode_loop(
    child: &mut Child,
    progress_file: &Path,
    duration: f64,
    total_frames: f64,
    mut progress_callback: Option<ProgressCallback>,
    cancel_flag: &AtomicBool,
    output: &str,
    stderr_path: &Path,
    remux_input: Option<&String>,
    lang: crate::i18n::Language,
) -> EncodeResult {
    // For remux jobs, estimate the final output size from the source file size.
    let remux_input_size = remux_input
        .and_then(|input| std::fs::metadata(input).ok())
        .map(|m| m.len())
        .filter(|&len| len > 0);

    loop {
        // Check cancellation
        if cancel_flag.load(Ordering::Acquire) {
            crate::utils::child::kill_and_wait(child);
            let _ = std::fs::remove_file(output);
            return EncodeResult::Cancelled;
        }

        if let Some(input_size) = remux_input_size {
            // Remux: derive progress from output-size growth vs. source size.
            if let Ok(out_meta) = std::fs::metadata(output) {
                #[allow(clippy::cast_precision_loss)]
                let progress = (out_meta.len() as f64 / input_size as f64 * 100.0).min(99.0);
                if let Some(ref mut cb) = progress_callback {
                    cb(progress);
                }
            }
        } else if let Some(content) = read_file_tail(progress_file) {
            // Encode: derive progress from the processed timestamp vs. duration.
            // FFmpeg versions differ in which out_time field they emit.
            let progress = if let Some(time_secs) = latest_progress_time_secs(&content) {
                Some(if duration > 0.0 {
                    (time_secs / duration * 100.0).min(100.0)
                } else {
                    // Duration unknown: advances slowly, capped at 99%.
                    (time_secs / 7200.0 * 100.0).min(99.0)
                })
            } else if total_frames > 0.0 {
                // Some sources (e.g. Dolby Vision) make FFmpeg report out_time=N/A.
                // Fall back to the processed frame count vs. the total frame count.
                latest_progress_frame(&content)
                    .map(|frame| (frame / total_frames * 100.0).min(99.0))
            } else {
                None
            };

            if let Some(progress) = progress
                && let Some(ref mut cb) = progress_callback
            {
                cb(progress);
            }
        }

        // Check if FFmpeg finished
        match child.try_wait() {
            Ok(Some(status)) => {
                crate::utils::child::ChildGuard::unregister(child.id());
                // A zero exit covers encodes FFmpeg ended early after a
                // tolerated read error.
                let short = status
                    .success()
                    .then(|| {
                        encoded_seconds(output, progress_file, duration, total_frames, cancel_flag)
                            .and_then(|encoded| shortfall(duration, encoded))
                    })
                    .flatten();
                if let Some(missing) = short {
                    let _ = std::fs::remove_file(output);
                    return EncodeResult::Error(with_stderr(
                        &t(lang, Msg::ErrEncodeShort)
                            .replace("{missing}", &format!("{missing:.0}"))
                            .replace("{duration}", &format!("{duration:.0}")),
                        stderr_path,
                    ));
                }
                if !status.success() {
                    let _ = std::fs::remove_file(output);
                    return EncodeResult::Error(with_stderr(
                        &t(lang, Msg::ErrFfmpegFailed).replace("{status}", &status.to_string()),
                        stderr_path,
                    ));
                }
                // ffmpeg exited after the last cancel poll; honour a late cancel.
                if cancel_flag.load(Ordering::Acquire) {
                    let _ = std::fs::remove_file(output);
                    return EncodeResult::Cancelled;
                }
                return EncodeResult::Success;
            }
            Ok(None) => {
                // Remux copies finish quickly and are polled more often.
                let poll_ms = if remux_input_size.is_some() { 100 } else { 250 };
                thread::sleep(Duration::from_millis(poll_ms));
            }
            Err(e) => {
                crate::utils::child::kill_and_wait(child);
                let _ = std::fs::remove_file(output);
                return EncodeResult::Error(format!("Failed to check ffmpeg status: {e}"));
            }
        }
    }
}

/// `message`, followed by the last few lines `FFmpeg` wrote to stderr.
fn with_stderr(message: &str, stderr_path: &Path) -> String {
    let stderr = read_file_tail(stderr_path).unwrap_or_default();
    let last_lines: Vec<&str> = stderr.lines().rev().take(5).collect();
    if last_lines.is_empty() {
        return message.to_string();
    }
    format!(
        "{message}: {}",
        last_lines.into_iter().rev().collect::<Vec<_>>().join("\n")
    )
}

/// Seconds of source the encode covered, read from the finished file, and from
/// the `-progress` log when ffprobe reports no duration.
fn encoded_seconds(
    output: &str,
    progress_file: &Path,
    duration: f64,
    total_frames: f64,
    cancel: &AtomicBool,
) -> Option<f64> {
    if let Some(secs) = crate::analyzer::ffprobe::probe_duration_secs(output, cancel) {
        return Some(secs);
    }

    let content = read_file_tail(progress_file)?;
    match latest_progress_time_secs(&content) {
        Some(secs) => Some(secs),
        // Some sources (e.g. Dolby Vision) report out_time=N/A throughout.
        None if total_frames > 0.0 && duration > 0.0 => {
            Some(latest_progress_frame(&content)? / total_frames * duration)
        }
        None => None,
    }
}

/// Seconds of the source missing from a finished encode, beyond the tolerated
/// drift between a container's stated duration and its streams.
fn shortfall(duration: f64, encoded: f64) -> Option<f64> {
    if duration <= 0.0 {
        return None;
    }
    let slack = (duration * 0.005).max(2.0);
    (duration - encoded > slack).then_some(duration - encoded)
}

/// Read the last few progress blocks from `-progress` output. `FFmpeg` appends
/// a block roughly twice a second and never truncates; only the tail of the
/// file is read.
pub(crate) fn read_file_tail(path: &Path) -> Option<String> {
    /// Comfortably more than one block, which always covers a whole block.
    const WINDOW: usize = 8192;

    let mut file = File::open(path).ok()?;
    let len = file.metadata().ok()?.len();
    file.seek(SeekFrom::Start(len.saturating_sub(WINDOW as u64)))
        .ok()?;

    let mut buf = Vec::with_capacity(WINDOW);
    file.read_to_end(&mut buf).ok()?;
    // The window can start mid-line; parsing is per-line and a partial first
    // line is ignored.
    Some(String::from_utf8_lossy(&buf).into_owned())
}

fn latest_progress_time_secs(content: &str) -> Option<f64> {
    content
        .lines()
        .filter_map(parse_progress_time_secs)
        .next_back()
}

fn latest_progress_frame(content: &str) -> Option<f64> {
    content
        .lines()
        .filter_map(|line| {
            line.strip_prefix("frame=")?
                .trim()
                .parse::<f64>()
                .ok()
                .filter(|frame| *frame > 0.0)
        })
        .next_back()
}

fn parse_progress_time_secs(line: &str) -> Option<f64> {
    if let Some(value) = line
        .strip_prefix("out_time_us=")
        .or_else(|| line.strip_prefix("out_time_ms="))
    {
        return value
            .trim()
            .parse::<f64>()
            .ok()
            .filter(|time_us| *time_us > 0.0)
            .map(|time_us| time_us / 1_000_000.0);
    }

    parse_out_time_secs(line)
}

fn parse_out_time_secs(line: &str) -> Option<f64> {
    let value = line.strip_prefix("out_time=")?.trim();
    let mut parts = value.split(':');

    let hours = parts.next()?.parse::<f64>().ok()?;
    let minutes = parts.next()?.parse::<f64>().ok()?;
    let seconds = parts.next()?.parse::<f64>().ok()?;

    if parts.next().is_some() {
        return None;
    }

    let total = hours * 3600.0 + minutes * 60.0 + seconds;
    (total > 0.0).then_some(total)
}

#[cfg(test)]
mod tests {
    use super::{
        encoded_seconds, is_same_file, latest_progress_frame, latest_progress_time_secs,
        partial_output_path, publish_partial, read_file_tail, shortfall,
    };
    use std::path::Path;

    #[test]
    fn publishing_never_overwrites_an_existing_output() {
        let dir = std::env::temp_dir().join(format!("av1c_publish_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let partial = dir.join("out.part.mkv");
        let output = dir.join("out.mkv");
        std::fs::write(&partial, b"encoded").unwrap();
        std::fs::write(&output, b"someone else's file").unwrap();

        let result = publish_partial(
            partial.to_str().unwrap(),
            output.to_str().unwrap(),
            crate::i18n::Language::English,
        );
        assert_eq!(
            result,
            Err(crate::i18n::t(
                crate::i18n::Language::English,
                crate::i18n::Msg::ErrOutputAppeared
            )
            .to_string())
        );
        assert_eq!(std::fs::read(&output).unwrap(), b"someone else's file");

        std::fs::remove_file(&output).unwrap();
        publish_partial(
            partial.to_str().unwrap(),
            output.to_str().unwrap(),
            crate::i18n::Language::English,
        )
        .unwrap();
        assert_eq!(std::fs::read(&output).unwrap(), b"encoded");
        assert!(!partial.exists());

        let _ = std::fs::remove_dir_all(dir);
    }

    #[cfg(any(unix, windows))]
    #[test]
    fn the_current_process_counts_as_alive() {
        assert!(super::pid_alive(std::process::id()));
    }

    /// The scratch file sits next to the real output, keeps its extension, and
    /// is never the destination path itself.
    #[test]
    fn partial_path_is_a_distinct_sibling_with_the_same_extension() {
        let output = "/media/films/movie_av1.mkv";
        let partial = partial_output_path(output, "123_4");

        assert_eq!(
            Path::new(&partial),
            Path::new("/media/films/movie_av1.part.123_4.mkv")
        );
        assert_ne!(partial, output);
        assert_eq!(Path::new(&partial).parent(), Path::new(output).parent());
        assert_eq!(Path::new(&partial).extension().unwrap(), "mkv");
    }

    #[test]
    fn partial_path_handles_an_extensionless_output() {
        assert_eq!(
            Path::new(&partial_output_path("/tmp/movie", "123_4")),
            Path::new("/tmp/movie.part.123_4")
        );
    }

    #[test]
    fn same_file_detects_identical_paths() {
        assert!(is_same_file(
            Path::new("/tmp/a.mkv"),
            Path::new("/tmp/a.mkv")
        ));
        assert!(!is_same_file(
            Path::new("/tmp/a.mkv"),
            Path::new("/tmp/a_av1.mkv")
        ));
    }

    /// Two spellings of one existing file resolve to the same target.
    #[test]
    fn same_file_resolves_equivalent_spellings() {
        let dir = std::env::temp_dir();
        let path = dir.join("av1c_test_same_file.mkv");
        std::fs::write(&path, b"x").unwrap();
        let indirect = dir.join(".").join("av1c_test_same_file.mkv");

        assert!(is_same_file(&path, &indirect));
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn file_tail_stays_bounded_and_keeps_the_end() {
        let path =
            std::env::temp_dir().join(format!("av1c_test_file_tail_{}.log", std::process::id()));
        std::fs::write(&path, format!("{}the end", "x".repeat(9000))).unwrap();

        let tail = read_file_tail(&path).unwrap();
        assert_eq!(tail.len(), 8192);
        assert!(tail.ends_with("the end"));
        let _ = std::fs::remove_file(path);
    }

    fn progress_file(name: &str, content: &str) -> std::path::PathBuf {
        let path =
            std::env::temp_dir().join(format!("av1c_test_{name}_{}.txt", std::process::id()));
        std::fs::write(&path, content).unwrap();
        path
    }

    /// An encode that stopped minutes into an hour-long source is not a success.
    #[test]
    fn a_truncated_encode_is_detected() {
        assert_eq!(shortfall(3600.0, 240.0), Some(3360.0));
    }

    /// Two minutes missing from a two-hour source is a truncated encode.
    #[test]
    fn two_minutes_short_of_two_hours_is_detected() {
        assert_eq!(shortfall(7200.0, 7080.0), Some(120.0));
    }

    /// A complete encode, and one a hair short of the container duration, pass.
    #[test]
    fn a_complete_encode_reports_nothing_missing() {
        assert_eq!(shortfall(3600.0, 3600.0), None);
        assert_eq!(shortfall(3600.0, 3583.0), None);
        assert_eq!(shortfall(1.0, 0.958), None);
    }

    /// An unknown source duration leaves nothing to compare against.
    #[test]
    fn an_unknown_duration_skips_the_check() {
        assert_eq!(shortfall(0.0, 240.0), None);
    }

    /// With no output to probe, the progress log stands in — via `out_time`, or
    /// via the frame count when the source reports `out_time=N/A`.
    #[test]
    fn the_progress_log_stands_in_for_an_unprobeable_output() {
        let cancel = std::sync::atomic::AtomicBool::new(false);
        let missing = std::env::temp_dir().join(format!(
            "av1c_test_no_such_output_{}.mkv",
            std::process::id()
        ));
        let _ = std::fs::remove_file(&missing);
        let output = missing.to_string_lossy().into_owned();

        let timed = progress_file("timed", "out_time_us=240000000\nprogress=end\n");
        let framed = progress_file("framed", "frame=1440\nout_time=N/A\nprogress=end\n");

        assert_eq!(
            encoded_seconds(&output, &timed, 3600.0, 0.0, &cancel),
            Some(240.0)
        );
        assert_eq!(
            encoded_seconds(&output, &framed, 3600.0, 86_400.0, &cancel),
            Some(60.0)
        );

        let _ = std::fs::remove_file(timed);
        let _ = std::fs::remove_file(framed);
    }

    #[test]
    fn parses_latest_frame_count() {
        let progress = "frame=10\nout_time=N/A\nframe=42\nout_time=N/A\n";

        assert_eq!(latest_progress_frame(progress), Some(42.0));
    }

    #[test]
    fn ignores_zero_frame_count() {
        let progress = "frame=0\nout_time=N/A\n";

        assert_eq!(latest_progress_frame(progress), None);
    }

    #[test]
    fn parses_latest_out_time_us() {
        let progress = "out_time_us=1000000\nprogress=continue\nout_time_us=2500000\n";

        assert_eq!(latest_progress_time_secs(progress), Some(2.5));
    }

    #[test]
    fn parses_legacy_out_time_ms_as_microseconds() {
        let progress = "out_time_ms=1500000\nprogress=continue\n";

        assert_eq!(latest_progress_time_secs(progress), Some(1.5));
    }

    #[test]
    fn parses_textual_out_time() {
        let progress = "out_time=N/A\nout_time=01:02:03.500000\n";

        assert_eq!(latest_progress_time_secs(progress), Some(3723.5));
    }
}