girt 0.10.2

use std::ops::ControlFlow;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

use rstest::rstest;

use super::*;
use crate::{EntryMode, InitKind, ObjectFormat, Tree, TreeEntry};

fn source(format: ObjectFormat) -> (tempfile::TempDir, Repository, ObjectId) {
    let root = tempfile::tempdir().unwrap();
    let repo = Repository::init(format, root.path().join("source"), InitKind::Bare).unwrap();
    let loose = repo.loose_objects();
    let blob = loose.write_blob(b"shared contents").unwrap();
    let tree = Tree::new(
        format,
        vec![
            TreeEntry {
                mode: EntryMode::Blob,
                name: b"a".to_vec(),
                id: blob,
            },
            TreeEntry {
                mode: EntryMode::Blob,
                name: b"b".to_vec(),
                id: blob,
            },
        ],
    )
    .unwrap();
    let id = loose.write_tree(&tree).unwrap();
    std::fs::write(repo.git_dir().join("refs/heads/main"), format!("{id}\n")).unwrap();
    (root, repo, id)
}

#[rstest]
#[case::sha1(ObjectFormat::Sha1)]
#[case::sha256(ObjectFormat::Sha256)]
fn local_progress_counts_unique_reads_and_pack_entries(#[case] format: ObjectFormat) {
    let (_root, repo, id) = source(format);
    let mut events = Vec::new();
    let received = receive(
        &repo,
        |_| vec![id],
        &KnownHistory::default(),
        FetchLimits::default(),
        TransportControl::new(&AtomicBool::new(false)),
        |_| ControlFlow::Continue(()),
        |event| events.push(event),
    );
    assert!(received.is_ok());
    assert_eq!(
        events,
        [
            LocalFetchProgress::Reading { objects: 0 },
            LocalFetchProgress::Reading { objects: 1 },
            LocalFetchProgress::Reading { objects: 2 },
            LocalFetchProgress::Packing { objects: (0, 2) },
            LocalFetchProgress::Packing { objects: (1, 2) },
            LocalFetchProgress::Packing { objects: (2, 2) },
            LocalFetchProgress::Complete,
        ]
    );
}

#[test]
fn local_progress_no_wants_has_no_pack() {
    let (_root, repo, _id) = source(ObjectFormat::Sha1);
    let mut events = Vec::new();
    receive(
        &repo,
        |_| vec![],
        &KnownHistory::default(),
        FetchLimits::default(),
        TransportControl::new(&AtomicBool::new(false)),
        |_| ControlFlow::Continue(()),
        |event| events.push(event),
    )
    .unwrap();
    assert_eq!(
        events,
        [
            LocalFetchProgress::Reading { objects: 0 },
            LocalFetchProgress::Complete
        ]
    );
}

#[test]
fn local_progress_known_objects_are_not_read_or_packed() {
    let (_root, repo, id) = source(ObjectFormat::Sha1);
    let cancel = AtomicBool::new(false);
    let objects = repo.objects(Default::default()).unwrap();
    let known = KnownHistory::new(&objects, &[id], FetchLimits::default(), &cancel).unwrap();
    let mut events = Vec::new();
    receive(
        &repo,
        |_| vec![id],
        &known,
        FetchLimits::default(),
        TransportControl::new(&cancel),
        |_| ControlFlow::Continue(()),
        |event| events.push(event),
    )
    .unwrap();
    assert_eq!(
        events,
        [
            LocalFetchProgress::Reading { objects: 0 },
            LocalFetchProgress::Complete
        ]
    );
}

#[rstest]
#[case::before_reads(LocalFetchProgress::Reading { objects: 0 })]
#[case::during_reads(LocalFetchProgress::Reading { objects: 1 })]
#[case::before_pack(LocalFetchProgress::Packing { objects: (0, 2) })]
#[case::during_pack(LocalFetchProgress::Packing { objects: (1, 2) })]
#[case::after_entries(LocalFetchProgress::Packing { objects: (2, 2) })]
fn local_progress_callback_can_cancel_before_completion(#[case] stop: LocalFetchProgress) {
    let (_root, repo, id) = source(ObjectFormat::Sha1);
    let cancel = AtomicBool::new(false);
    let mut events = Vec::new();
    let result = receive(
        &repo,
        |_| vec![id],
        &KnownHistory::default(),
        FetchLimits::default(),
        TransportControl::new(&cancel),
        |_| ControlFlow::Continue(()),
        |event| {
            events.push(event);
            if event == stop {
                cancel.store(true, Ordering::Relaxed);
            }
        },
    );
    assert!(matches!(result, Err(FetchError::Cancelled)));
    assert!(!events.contains(&LocalFetchProgress::Complete));
    assert_eq!(events.last(), Some(&stop));
}

#[test]
fn local_progress_terminal_callback_cannot_undo_completion() {
    let (_root, repo, id) = source(ObjectFormat::Sha1);
    let cancel = AtomicBool::new(false);
    receive(
        &repo,
        |_| vec![id],
        &KnownHistory::default(),
        FetchLimits::default(),
        TransportControl::new(&cancel),
        |_| ControlFlow::Continue(()),
        |event| {
            if event == LocalFetchProgress::Complete {
                cancel.store(true, Ordering::Relaxed);
            }
        },
    )
    .unwrap();
    assert!(cancel.load(Ordering::Relaxed));
}

#[test]
fn local_progress_preserves_original_callback_cancellation() {
    let (_root, repo, id) = source(ObjectFormat::Sha1);
    let mut events = Vec::new();
    let result = receive(
        &repo,
        |_| vec![id],
        &KnownHistory::default(),
        FetchLimits::default(),
        TransportControl::new(&AtomicBool::new(false)),
        |_| ControlFlow::Break(()),
        |event| events.push(event),
    );
    assert!(matches!(result, Err(FetchError::Cancelled)));
    assert!(!events.contains(&LocalFetchProgress::Complete));
}

#[test]
fn local_progress_expired_deadline_has_no_notifications() {
    let (_root, repo, id) = source(ObjectFormat::Sha1);
    let mut events = Vec::new();
    let result = receive(
        &repo,
        |_| vec![id],
        &KnownHistory::default(),
        FetchLimits::default(),
        TransportControl {
            cancel: &AtomicBool::new(false),
            deadline: Some(Instant::now() - Duration::from_secs(1)),
        },
        |_| ControlFlow::Continue(()),
        |event| events.push(event),
    );
    assert!(matches!(result, Err(FetchError::Deadline)));
    assert!(events.is_empty());
}

#[test]
fn local_progress_pack_failure_has_no_completion() {
    let (_root, repo, id) = source(ObjectFormat::Sha1);
    let mut events = Vec::new();
    let result = receive(
        &repo,
        |_| vec![id],
        &KnownHistory::default(),
        FetchLimits {
            max_pack_bytes: 12,
            ..Default::default()
        },
        TransportControl::new(&AtomicBool::new(false)),
        |_| ControlFlow::Continue(()),
        |event| events.push(event),
    );
    assert!(matches!(result, Err(FetchError::PackWrite(_))));
    assert!(!events.contains(&LocalFetchProgress::Complete));
}

#[test]
fn local_progress_missing_source_has_no_completion() {
    let (_root, repo, _id) = source(ObjectFormat::Sha1);
    let missing = ObjectId::for_blob(ObjectFormat::Sha1, b"missing");
    std::fs::write(
        repo.git_dir().join("refs/heads/main"),
        format!("{missing}\n"),
    )
    .unwrap();
    let mut events = Vec::new();
    let result = receive(
        &repo,
        |_| vec![missing],
        &KnownHistory::default(),
        FetchLimits::default(),
        TransportControl::new(&AtomicBool::new(false)),
        |_| ControlFlow::Continue(()),
        |event| events.push(event),
    );
    assert!(matches!(result, Err(FetchError::Missing(_))));
    assert!(!events.contains(&LocalFetchProgress::Complete));
}

#[test]
fn local_clone_pack_is_spooled_and_removed_on_drop() {
    let (_root, repo, id) = source(ObjectFormat::Sha1);
    let received = receive(
        &repo,
        |_| vec![id],
        &KnownHistory::default(),
        FetchLimits::default(),
        TransportControl::new(&AtomicBool::new(false)),
        |_| ControlFlow::Continue(()),
        |_| {},
    )
    .unwrap();
    let path = received
        .temporary_pack_path()
        .expect("local transfer must spool")
        .to_owned();
    assert_eq!(
        std::fs::metadata(&path).unwrap().len() as usize,
        received.pack_bytes()
    );
    drop(received);
    assert!(!path.exists());
}

/// A Git-written history of nested trees across many commits, so the walk reads ahead.
fn git_history(format: ObjectFormat) -> (tempfile::TempDir, Repository, ObjectId) {
    let root = tempfile::tempdir().unwrap();
    let path = root.path();
    let git = |args: &[&str]| {
        let output = std::process::Command::new("git")
            .current_dir(path)
            .env("GIT_CONFIG_NOSYSTEM", "1")
            .env("GIT_CONFIG_GLOBAL", path.join("absent-config"))
            .env("GIT_AUTHOR_NAME", "A")
            .env("GIT_AUTHOR_EMAIL", "a@example.com")
            .env("GIT_COMMITTER_NAME", "A")
            .env("GIT_COMMITTER_EMAIL", "a@example.com")
            .args(args)
            .output()
            .unwrap();
        assert!(output.status.success(), "{args:?}");
        String::from_utf8(output.stdout).unwrap()
    };
    git(&["init", "-q", &format!("--object-format={format}"), "."]);
    // A background `gc --auto` would repack and delete loose objects while the test reads them.
    git(&["config", "gc.auto", "0"]);
    git(&["config", "maintenance.auto", "false"]);
    for round in 0..12 {
        for file in 0..40 {
            let dir = path.join(format!("d{}/e{}", file % 4, file % 3));
            std::fs::create_dir_all(&dir).unwrap();
            std::fs::write(
                dir.join(format!("f{file}")),
                format!("{file} {}\n", round * (file % 5)),
            )
            .unwrap();
        }
        git(&["add", "-A"]);
        git(&[
            "-c",
            "commit.gpgsign=false",
            "commit",
            "-qm",
            &format!("r{round}"),
        ]);
    }
    let head = git(&["rev-parse", "HEAD"]).trim().parse().unwrap();
    let repo = Repository::open(path.join(".git")).unwrap();
    (root, repo, head)
}

/// Every worker count, including none, reads the same objects in the same order and writes the
/// same pack. A small decode allowance is still refused while workers read ahead.
#[rstest]
#[case::sha1(ObjectFormat::Sha1)]
#[case::sha256(ObjectFormat::Sha256)]
fn worker_counts_select_identical_packs_and_keep_limits(#[case] format: ObjectFormat) {
    let (_root, repo, head) = git_history(format);
    let run = |max_threads, max_decode_bytes| {
        let mut events = Vec::new();
        let received = receive(
            &repo,
            |_| vec![head],
            &KnownHistory::default(),
            FetchLimits {
                max_threads,
                max_decode_bytes,
                ..FetchLimits::default()
            },
            TransportControl::new(&AtomicBool::new(false)),
            |_| ControlFlow::Continue(()),
            |event| events.push(event),
        );
        (received, events)
    };
    let (inline, inline_events) = run(1, FetchLimits::default().max_decode_bytes);
    let inline = inline.unwrap();
    let inline_pack = std::fs::read(inline.temporary_pack_path().unwrap()).unwrap();
    for threads in [0, 2, 8] {
        let (received, events) = run(threads, FetchLimits::default().max_decode_bytes);
        let received = received.unwrap();
        assert_eq!(events, inline_events, "{threads} threads");
        assert_eq!(
            std::fs::read(received.temporary_pack_path().unwrap()).unwrap(),
            inline_pack,
            "{threads} threads"
        );
    }
    for threads in [1, 8] {
        assert!(matches!(
            run(threads, 4096).0,
            Err(FetchError::Limit("local bytes"))
        ));
    }
}