pigeon-cli 0.4.24

//! Finds files already landed under `<local_output>/source/` (ADR-0112),
//! used as one of the pipeline's two dispatch sources alongside the pull
//! subprocess's live tail (`transform::worker`'s module doc comment).
//! "Already done" is no longer decided here -- `transform` runs on a
//! freshly-provisioned, ephemeral VM per invocation, so a local checkpoint
//! never survives a VM replacement and was proven dead weight in production
//! (ADR-0120). `worker::enqueue` now answers "already done" by checking
//! `destination::list_existing_filenames` instead, keyed off each file's
//! deterministic `placement::compute_destination_name`.

use std::path::{Path, PathBuf};

use super::format::InputFileType;
use crate::core::data::collect_files;

/// One pending file discovered under `<local_output>/source/`. `relative_path`
/// is relative to `source_dir`, using `/`-separated components regardless of
/// host OS -- this is also half of `placement`'s destination-name hash input
/// (ADR-0112 Decision §5).
pub(crate) struct PendingFile {
    pub relative_path: String,
    pub absolute_path: PathBuf,
}

/// `path`'s location relative to `base`, rendered with `/`-separated
/// components regardless of host OS -- used as half of
/// `placement::compute_destination_name`'s hash input, so it must be stable
/// across platforms for the same logical file.
fn relative_path_string(base: &Path, path: &Path) -> Option<String> {
    let relative = path.strip_prefix(base).ok()?;
    let components: Vec<&str> = relative
        .components()
        .filter_map(|c| c.as_os_str().to_str())
        .collect();
    if components.is_empty() {
        return None;
    }
    Some(components.join("/"))
}

/// Walks `source_dir` (everything the bulk pull has landed so far), keeping
/// only files whose extension is one of `input_file_type`'s spellings
/// (defensive -- the source listing's own `--include` filters should
/// already guarantee this; anything else found here is unexpected, not
/// fatal, and is skipped with a `tracing::warn!`). Used as the pipeline's
/// initial dispatch source (`transform::worker`), run once before the pull
/// subprocess spawns. Whether a given file is actually already done is
/// decided downstream by `worker::enqueue`, not here.
///
/// The check is a membership test over `InputFileType::extensions`, not an
/// equality test against one canonical string: a `.jpg` file under a
/// `--input-file-type=jpeg` run is legitimate and must not be dropped
/// (ADR-0139). This was a second, independent gate failing the same way as
/// the source listing's filter, and would have silently skipped any `.jpg`
/// already staged on local disk even once that filter was fixed.
pub(crate) fn gather_pending(
    source_dir: &Path,
    input_file_type: InputFileType,
) -> Result<Vec<PendingFile>, String> {
    let mut pending = Vec::new();
    for absolute_path in collect_files(source_dir)? {
        let Some(relative_path) = relative_path_string(source_dir, &absolute_path) else {
            continue;
        };
        let actual_extension = absolute_path
            .extension()
            .and_then(|ext| ext.to_str())
            .unwrap_or_default();
        if !input_file_type.matches_extension(actual_extension) {
            tracing::warn!(
                path = %absolute_path.display(),
                expected = ?input_file_type.extensions(),
                "transform: found an unexpected extension under source/, skipping"
            );
            continue;
        }
        pending.push(PendingFile {
            relative_path,
            absolute_path,
        });
    }
    Ok(pending)
}

#[cfg(test)]
mod tests {
    use std::fs;

    use super::*;

    #[test]
    fn gather_pending_filters_by_extension() {
        let dir = tempfile::tempdir().unwrap();
        let source_dir = dir.path().join("source");
        fs::create_dir_all(&source_dir).unwrap();
        fs::write(source_dir.join("a.png"), b"a").unwrap();
        fs::write(source_dir.join("stray.txt"), b"not a png").unwrap();

        let pending = gather_pending(&source_dir, InputFileType::Png).unwrap();
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].relative_path, "a.png");
    }

    #[test]
    fn gather_pending_includes_a_nested_relative_path() {
        let dir = tempfile::tempdir().unwrap();
        let source_dir = dir.path().join("source");
        fs::create_dir_all(source_dir.join("screenshots")).unwrap();
        fs::write(source_dir.join("screenshots/img.heic"), b"x").unwrap();

        let pending = gather_pending(&source_dir, InputFileType::Heic).unwrap();
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].relative_path, "screenshots/img.heic");
    }

    /// ADR-0139's second gate: a `.jpg` staged on local disk under a
    /// `jpeg` run must be picked up, not warned-and-skipped. Before the
    /// membership test, this gate dropped them even once the source
    /// listing's filter was fixed.
    #[test]
    fn gather_pending_accepts_every_spelling_of_one_input_kind() {
        let dir = tempfile::tempdir().unwrap();
        let source_dir = dir.path().join("source");
        fs::create_dir_all(&source_dir).unwrap();
        fs::write(source_dir.join("a.jpg"), b"a").unwrap();
        fs::write(source_dir.join("b.jpeg"), b"b").unwrap();
        fs::write(source_dir.join("c.JPG"), b"c").unwrap();
        fs::write(source_dir.join("stray.png"), b"a different format").unwrap();

        let pending = gather_pending(&source_dir, InputFileType::Jpeg).unwrap();
        let mut found: Vec<String> = pending.into_iter().map(|file| file.relative_path).collect();
        found.sort();
        assert_eq!(found, vec!["a.jpg", "b.jpeg", "c.JPG"]);
    }

    #[test]
    fn gather_pending_filters_by_extension_for_a_video_input_kind() {
        let dir = tempfile::tempdir().unwrap();
        let source_dir = dir.path().join("source");
        fs::create_dir_all(&source_dir).unwrap();
        fs::write(source_dir.join("a.mp4"), b"a").unwrap();
        fs::write(source_dir.join("stray.mov"), b"not an mp4").unwrap();

        let pending = gather_pending(&source_dir, InputFileType::Mp4).unwrap();
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].relative_path, "a.mp4");
    }

    #[test]
    fn gather_pending_includes_every_matching_file_regardless_of_prior_state() {
        let dir = tempfile::tempdir().unwrap();
        let source_dir = dir.path().join("source");
        fs::create_dir_all(&source_dir).unwrap();
        fs::write(source_dir.join("a.png"), b"a").unwrap();
        fs::write(source_dir.join("b.png"), b"b").unwrap();

        // No checkpoint concept exists anymore -- gather_pending reports
        // every matching file every time; "already done" is decided by
        // `worker::enqueue` against the destination listing instead.
        let pending = gather_pending(&source_dir, InputFileType::Png).unwrap();
        assert_eq!(pending.len(), 2);
    }
}