use std::{
    fs,
    io::{self, Read},
    os::unix::fs::PermissionsExt,
    path::PathBuf,
    process::{Child, Command, ExitStatus},
    sync::atomic::{AtomicU64, Ordering},
    thread,
    time::Duration,
};

#[cfg(debug_assertions)]
use crank_artifacts::test_support::{
    FaultAction, clear_checkpoint, set_checkpoint, wait_until_held,
};
use crank_artifacts::{ArtifactError, ArtifactRef, ArtifactStore, MAX_ARTIFACT_BYTES, StaleTemp};
use sha2::{Digest, Sha256};
#[cfg(debug_assertions)]
use std::sync::{Mutex, OnceLock};

static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
#[cfg(debug_assertions)]
fn fault_guard() -> std::sync::MutexGuard<'static, ()> {
    static GUARD: OnceLock<Mutex<()>> = OnceLock::new();
    GUARD.get_or_init(|| Mutex::new(())).lock().unwrap()
}

struct TestRoot(PathBuf);

impl TestRoot {
    fn new(name: &str) -> Self {
        let path = std::env::temp_dir().join(format!(
            "crank-artifacts-{name}-{}-{}",
            std::process::id(),
            NEXT_ROOT.fetch_add(1, Ordering::Relaxed)
        ));
        fs::create_dir(&path).unwrap();
        fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap();
        Self(path)
    }
}

impl Drop for TestRoot {
    fn drop(&mut self) {
        let _ = fs::set_permissions(&self.0, fs::Permissions::from_mode(0o700));
        let _ = fs::remove_dir_all(&self.0);
    }
}

fn wait_for_child(mut child: Child, timeout: Duration) -> ExitStatus {
    let deadline = std::time::Instant::now() + timeout;
    loop {
        if let Some(status) = child.try_wait().unwrap() {
            return status;
        }
        if std::time::Instant::now() >= deadline {
            let _ = child.kill();
            let _ = child.wait();
            panic!("child process did not finish within {timeout:?}");
        }
        thread::sleep(Duration::from_millis(10));
    }
}

#[test]
fn stores_and_reads_canonical_digest() {
    let root = TestRoot::new("roundtrip");
    let store = ArtifactStore::open(&root.0).unwrap();
    let stored = store.put(b"openapi: 3.0.0\n").unwrap();

    assert_eq!(
        stored.artifact_ref.as_str(),
        "sha256:344e4b2f7f15b76b5606be45d8031fc43f473f8d63f0e02c61dbf89a97f85e69"
    );
    assert_eq!(
        store.read(&stored.artifact_ref).unwrap(),
        b"openapi: 3.0.0\n"
    );
}

#[test]
fn concurrent_writers_deduplicate_without_overwrite() {
    let root = TestRoot::new("race");
    let store = ArtifactStore::open(&root.0).unwrap();
    let writers = (0..8)
        .map(|_| {
            let store = store.clone();
            thread::spawn(move || store.put(b"same immutable bytes").unwrap().artifact_ref)
        })
        .collect::<Vec<_>>();
    let refs = writers
        .into_iter()
        .map(|writer| writer.join().unwrap())
        .collect::<Vec<_>>();

    assert!(refs.windows(2).all(|pair| pair[0] == pair[1]));
    assert_eq!(store.read(&refs[0]).unwrap(), b"same immutable bytes");
}

#[test]
fn independent_writer_processes_deduplicate_without_overwrite() {
    if let (Some(root), Some(ready), Some(index)) = (
        std::env::var_os("CRANK_ARTIFACTS_PROCESS_ROOT"),
        std::env::var_os("CRANK_ARTIFACTS_PROCESS_READY"),
        std::env::var_os("CRANK_ARTIFACTS_PROCESS_INDEX"),
    ) {
        let ready = PathBuf::from(ready);
        fs::write(ready.join(index), b"ready").unwrap();
        let start = ready.join("start");
        while !start.exists() {
            thread::sleep(Duration::from_millis(5));
        }
        let store = ArtifactStore::open(PathBuf::from(root)).unwrap();
        let stored = store.put(b"same process bytes").unwrap();
        let expected =
            ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"same process bytes")))
                .unwrap();
        assert_eq!(stored.artifact_ref, expected);
        return;
    }

    let root = TestRoot::new("process-race");
    let barrier = TestRoot::new("process-barrier");
    let executable = std::env::current_exe().unwrap();
    let mut children = (0..4)
        .map(|index| {
            Command::new(&executable)
                .args([
                    "--exact",
                    "independent_writer_processes_deduplicate_without_overwrite",
                    "--nocapture",
                ])
                .env("CRANK_ARTIFACTS_PROCESS_ROOT", &root.0)
                .env("CRANK_ARTIFACTS_PROCESS_READY", &barrier.0)
                .env("CRANK_ARTIFACTS_PROCESS_INDEX", index.to_string())
                .spawn()
                .unwrap()
        })
        .collect::<Vec<_>>();
    let deadline = std::time::Instant::now() + Duration::from_secs(3);
    while fs::read_dir(&barrier.0).unwrap().count() != children.len() {
        assert!(
            std::time::Instant::now() < deadline,
            "children missed barrier"
        );
        thread::sleep(Duration::from_millis(10));
    }
    fs::write(barrier.0.join("start"), b"start").unwrap();
    for child in children.drain(..) {
        assert!(wait_for_child(child, Duration::from_secs(3)).success());
    }

    let store = ArtifactStore::open(&root.0).unwrap();
    let expected =
        ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"same process bytes")))
            .unwrap();
    assert_eq!(store.read(&expected).unwrap(), b"same process bytes");
}

#[test]
#[cfg(debug_assertions)]
fn verified_dedupe_fsyncs_its_shard_before_reporting_success() {
    let _guard = fault_guard();
    let root = TestRoot::new("dedupe-fsync");
    let store = ArtifactStore::open(&root.0).unwrap();
    store.put(b"dedupe fsync bytes").unwrap();
    set_checkpoint("dedupe_directory_fsync", FaultAction::Fail);
    assert_eq!(
        store.put(b"dedupe fsync bytes").unwrap_err(),
        ArtifactError::Storage
    );
    clear_checkpoint();
    assert_eq!(
        store.put(b"dedupe fsync bytes").unwrap().size_bytes,
        b"dedupe fsync bytes".len()
    );
}

#[test]
#[cfg(debug_assertions)]
fn new_namespace_edges_require_parent_directory_fsync() {
    let _guard = fault_guard();
    let root = TestRoot::new("namespace-fsync");
    let store = ArtifactStore::open(&root.0).unwrap();
    set_checkpoint("mkdir_parent_fsync", FaultAction::Fail);

    assert_eq!(store.put(b"namespace bytes"), Err(ArtifactError::Storage));
    assert!(root.0.join("sha256").is_dir());
    clear_checkpoint();
    set_checkpoint("mkdir_parent_fsync", FaultAction::Hold);
    let retry_store = store.clone();
    let retry = thread::spawn(move || retry_store.put(b"namespace bytes"));
    assert!(wait_until_held(Duration::from_secs(2)));
    clear_checkpoint();
    let stored = retry.join().unwrap().unwrap();
    assert_eq!(
        store.read(&stored.artifact_ref).unwrap(),
        b"namespace bytes"
    );
}

#[test]
#[cfg(debug_assertions)]
fn live_temp_cannot_be_deleted_while_writer_holds_shared_root_lock() {
    let _guard = fault_guard();
    use std::sync::mpsc;
    set_checkpoint("write", FaultAction::Hold);
    let root = TestRoot::new("live-temp");
    let store = ArtifactStore::open(&root.0).unwrap();
    let writer_store = store.clone();
    let writer = thread::spawn(move || writer_store.put(b"live temp bytes"));
    assert!(wait_until_held(Duration::from_secs(2)));
    let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 16).unwrap();
    assert_eq!(candidates.len(), 1);
    let candidate = candidates.into_iter().next().unwrap();
    let deleter_store = store.clone();
    let (sent, received) = mpsc::channel();
    let deleter = thread::spawn(move || {
        let _ = sent.send(deleter_store.delete_stale_temp(candidate));
    });
    assert!(received.recv_timeout(Duration::from_millis(100)).is_err());
    clear_checkpoint();
    assert!(writer.join().unwrap().is_ok());
    assert!(received.recv_timeout(Duration::from_secs(2)).is_ok());
    deleter.join().unwrap();
}

#[test]
fn rejects_malformed_and_oversized_inputs() {
    assert!(ArtifactRef::parse(&format!("sha256:{}", "a".repeat(64))).is_ok());
    assert!(ArtifactRef::parse(&format!("sha256:{}", "A".repeat(64))).is_err());
    assert!(ArtifactRef::parse("sha256:ABC").is_err());
    assert!(ArtifactRef::parse("file:///tmp/a").is_err());
    let root = TestRoot::new("limits");
    let store = ArtifactStore::open(&root.0).unwrap();
    assert_eq!(store.put(b"").unwrap_err(), ArtifactError::EmptySource);
    assert_eq!(
        store.put(&vec![0; MAX_ARTIFACT_BYTES + 1]).unwrap_err(),
        ArtifactError::SourceTooLarge
    );
    let maximum = vec![b'x'; MAX_ARTIFACT_BYTES];
    let stored = store.put(&maximum).unwrap();
    assert_eq!(stored.size_bytes, MAX_ARTIFACT_BYTES);
    assert_eq!(store.read(&stored.artifact_ref).unwrap(), maximum);
}

#[test]
fn root_replacement_cannot_redirect_an_open_store() {
    let root = TestRoot::new("replacement");
    let store = ArtifactStore::open(&root.0).unwrap();
    let original = root.0.with_extension("original");
    fs::rename(&root.0, &original).unwrap();
    fs::create_dir(&root.0).unwrap();
    fs::set_permissions(&root.0, fs::Permissions::from_mode(0o700)).unwrap();
    // Replacing the configured pathname after opening cannot change the fd
    // authority; only the original inode receives the blob.
    store.put(b"pinned authority").unwrap();
    assert!(original.join("sha256").exists());
    assert!(!root.0.join("sha256").exists());
    let _ = fs::remove_dir_all(&original);
}

#[test]
fn stale_temp_cleanup_uses_a_bounded_capability() {
    let root = TestRoot::new("cleanup");
    let store = ArtifactStore::open(&root.0).unwrap();
    let stored = store.put(b"prepare shard").unwrap();
    let shard = root
        .0
        .join("sha256")
        .join(&stored.artifact_ref.digest_hex()[..2]);
    let temp = shard.join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-1");
    fs::write(&temp, b"partial").unwrap();
    let (report, candidates) = store.scan_stale_temps(Duration::ZERO, 16, 1).unwrap();
    assert!(report.scanned <= 16);
    assert_eq!(candidates.len(), 1);
    store
        .delete_stale_temp(candidates.into_iter().next().unwrap())
        .unwrap();
    assert!(!temp.exists());
}

#[test]
fn housekeeping_deletes_entries_older_than_positive_grace() {
    let root = TestRoot::new("cleanup-positive-grace");
    let store = ArtifactStore::open(&root.0).unwrap();
    let stored = store.put(b"positive grace shard").unwrap();
    let temp = root
        .0
        .join("sha256")
        .join(&stored.artifact_ref.digest_hex()[..2])
        .join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-9");
    fs::write(&temp, b"partial").unwrap();
    fs::File::options()
        .write(true)
        .open(&temp)
        .unwrap()
        .set_modified(std::time::SystemTime::now() - Duration::from_secs(5))
        .unwrap();

    let (_, candidates) = store
        .scan_stale_temps(Duration::from_secs(1), 128, 1)
        .unwrap();
    assert_eq!(candidates.len(), 1);
    store
        .delete_stale_temp(candidates.into_iter().next().unwrap())
        .unwrap();
    assert!(!temp.exists());
}

#[test]
#[cfg(debug_assertions)]
fn housekeeping_ignores_temp_that_disappears_after_readdir() {
    let _guard = fault_guard();
    let root = TestRoot::new("cleanup-disappearing");
    let store = ArtifactStore::open(&root.0).unwrap();
    let stored = store.put(b"disappearing shard").unwrap();
    let temp = root
        .0
        .join("sha256")
        .join(&stored.artifact_ref.digest_hex()[..2])
        .join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-11");
    fs::write(&temp, b"partial").unwrap();
    set_checkpoint("housekeeping_before_stat", FaultAction::Hold);
    let scan_store = store.clone();
    let scanner = thread::spawn(move || scan_store.scan_stale_temps(Duration::ZERO, 128, 16));
    assert!(wait_until_held(Duration::from_secs(2)));
    fs::remove_file(&temp).unwrap();
    clear_checkpoint();

    let (report, candidates) = scanner.join().unwrap().unwrap();
    assert!(report.scanned > 0);
    assert!(candidates.is_empty());
}

#[test]
fn housekeeping_enforces_result_limit_after_stale_validation() {
    let root = TestRoot::new("cleanup-result-limit");
    let store = ArtifactStore::open(&root.0).unwrap();
    let mut temp_paths = Vec::new();
    for (source, sequence) in [(b"limit-a".as_slice(), 1), (b"limit-b", 2)] {
        let stored = store.put(source).unwrap();
        let shard = root
            .0
            .join("sha256")
            .join(&stored.artifact_ref.digest_hex()[..2]);
        let temp = shard.join(format!(
            ".crank-artifact-tmp-v1-00000000000000000000000000000000-1-{sequence}"
        ));
        fs::write(&temp, b"partial").unwrap();
        temp_paths.push(temp);
    }

    let (limited, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 1).unwrap();
    assert_eq!(candidates.len(), 1);
    assert_eq!(limited.omitted, 1);
    let (zero, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 0).unwrap();
    assert!(candidates.is_empty());
    assert_eq!(zero.omitted, 2);
    assert!(temp_paths.iter().all(|path| path.exists()));
}

#[test]
fn housekeeping_revalidates_subsecond_mtime() {
    let root = TestRoot::new("cleanup-subsecond");
    let store = ArtifactStore::open(&root.0).unwrap();
    let stored = store.put(b"subsecond shard").unwrap();
    let temp = root
        .0
        .join("sha256")
        .join(&stored.artifact_ref.digest_hex()[..2])
        .join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-3");
    fs::write(&temp, b"partial").unwrap();
    let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 1).unwrap();
    assert_eq!(candidates.len(), 1);

    let original = fs::metadata(&temp)
        .unwrap()
        .modified()
        .unwrap()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap();
    let changed_nanos = if original.subsec_nanos() == 999_999_999 {
        999_999_998
    } else {
        original.subsec_nanos() + 1
    };
    fs::File::options()
        .write(true)
        .open(&temp)
        .unwrap()
        .set_modified(std::time::UNIX_EPOCH + Duration::new(original.as_secs(), changed_nanos))
        .unwrap();
    assert_eq!(
        store.delete_stale_temp(candidates.into_iter().next().unwrap()),
        Err(ArtifactError::UnsafeRoot)
    );
    assert!(temp.exists());
}

#[test]
fn housekeeping_respects_global_budgets_and_rejects_foreign_or_replaced_temps() {
    let root = TestRoot::new("housekeeping-audit");
    let store = ArtifactStore::open(&root.0).unwrap();
    for source in [b"a".as_slice(), b"b", b"c"] {
        let _ = store.put(source).unwrap();
    }
    let sha = root.0.join("sha256");
    let mut shards = fs::read_dir(&sha)
        .unwrap()
        .map(|entry| entry.unwrap().path())
        .collect::<Vec<_>>();
    shards.sort();
    for shard in &shards {
        fs::write(shard.join(".crank-artifact-tmp-v1-not-a-valid-name"), b"x").unwrap();
    }
    let replacement = shards[0].join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-1");
    fs::write(&replacement, b"x").unwrap();
    let (limited, candidates) = store.scan_stale_temps(Duration::ZERO, 5, 1).unwrap();
    assert_eq!(limited.scanned, 5);
    assert!(candidates.len() <= 1);
    let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 16).unwrap();
    assert_eq!(candidates.len(), 1);
    let candidate = candidates.into_iter().next().unwrap();
    fs::remove_file(&replacement).unwrap();
    fs::write(&replacement, b"replacement").unwrap();
    assert_eq!(
        store.delete_stale_temp(candidate).unwrap_err(),
        ArtifactError::UnsafeRoot
    );
    let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 16).unwrap();
    let foreign_candidate = candidates.into_iter().next().unwrap();
    let other = TestRoot::new("housekeeping-other");
    let other_store = ArtifactStore::open(&other.0).unwrap();
    assert_eq!(
        other_store
            .delete_stale_temp(foreign_candidate)
            .unwrap_err(),
        ArtifactError::UnsafeRoot
    );
    let (_, fresh) = store
        .scan_stale_temps(Duration::from_secs(3600), 128, 16)
        .unwrap();
    assert!(fresh.is_empty());
}

#[test]
fn open_rejects_non_private_root_and_existing_finals_must_be_immutable() {
    let root = TestRoot::new("root-audit");
    fs::set_permissions(&root.0, fs::Permissions::from_mode(0o755)).unwrap();
    assert_eq!(
        ArtifactStore::open(&root.0).unwrap_err(),
        ArtifactError::UnsafeRoot
    );
    fs::set_permissions(&root.0, fs::Permissions::from_mode(0o770)).unwrap();
    assert_eq!(
        ArtifactStore::open(&root.0).unwrap_err(),
        ArtifactError::UnsafeRoot
    );
    fs::set_permissions(&root.0, fs::Permissions::from_mode(0o700)).unwrap();
    let store = ArtifactStore::open(&root.0).unwrap();
    let stored = store.put(b"final audit").unwrap();
    let path = root
        .0
        .join("sha256")
        .join(&stored.artifact_ref.digest_hex()[..2])
        .join(stored.artifact_ref.digest_hex());
    fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap();
    assert_eq!(
        store.read(&stored.artifact_ref).unwrap_err(),
        ArtifactError::Integrity
    );
    fs::set_permissions(&path, fs::Permissions::from_mode(0o400)).unwrap();
    let linked = path.with_extension("link");
    fs::hard_link(&path, &linked).unwrap();
    assert_eq!(
        store.read(&stored.artifact_ref).unwrap_err(),
        ArtifactError::Integrity
    );
}

#[test]
fn open_rejects_symlinked_root() {
    let root = TestRoot::new("root-symlink-target");
    let link = root.0.with_extension("symlink");
    std::os::unix::fs::symlink(&root.0, &link).unwrap();
    assert!(ArtifactStore::open(&link).is_err());
    fs::remove_file(link).unwrap();
}

#[test]
fn corrupted_or_symlinked_final_never_returns_bytes() {
    let root = TestRoot::new("integrity");
    let store = ArtifactStore::open(&root.0).unwrap();
    let stored = store.put(b"safe bytes").unwrap();
    let path = root
        .0
        .join("sha256")
        .join(&stored.artifact_ref.digest_hex()[..2])
        .join(stored.artifact_ref.digest_hex());
    fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap();
    fs::write(&path, b"evil bytes").unwrap();
    fs::set_permissions(&path, fs::Permissions::from_mode(0o400)).unwrap();
    assert_eq!(
        store.read(&stored.artifact_ref).unwrap_err(),
        ArtifactError::Integrity
    );

    fs::remove_file(&path).unwrap();
    let outside = root.0.join("outside-read-only");
    fs::write(&outside, b"safe bytes").unwrap();
    fs::set_permissions(&outside, fs::Permissions::from_mode(0o400)).unwrap();
    std::os::unix::fs::symlink(&outside, &path).unwrap();
    assert!(store.read(&stored.artifact_ref).is_err());

    fs::remove_file(&path).unwrap();
    fs::create_dir(&path).unwrap();
    assert_eq!(
        store.read(&stored.artifact_ref).unwrap_err(),
        ArtifactError::Integrity
    );
}

#[test]
fn directory_at_final_digest_is_integrity_failure() {
    let root = TestRoot::new("directory-final");
    let store = ArtifactStore::open(&root.0).unwrap();
    let stored = store.put(b"directory final").unwrap();
    let path = root
        .0
        .join("sha256")
        .join(&stored.artifact_ref.digest_hex()[..2])
        .join(stored.artifact_ref.digest_hex());
    fs::remove_file(&path).unwrap();
    fs::create_dir(&path).unwrap();
    fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap();
    assert_eq!(
        store.read(&stored.artifact_ref).unwrap_err(),
        ArtifactError::Integrity
    );
}

#[test]
fn legacy_reader_is_pinned_to_its_explicit_root_and_digest() {
    let root = TestRoot::new("legacy");
    let legacy = root.0.join("legacy");
    fs::create_dir(&legacy).unwrap();
    fs::set_permissions(&legacy, fs::Permissions::from_mode(0o700)).unwrap();
    let source = legacy.join("source.yaml");
    fs::write(&source, b"openapi: 3.0.0").unwrap();
    fs::set_permissions(&source, fs::Permissions::from_mode(0o600)).unwrap();
    let expected =
        ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"openapi: 3.0.0"))).unwrap();
    let store = ArtifactStore::open(&root.0)
        .unwrap()
        .with_legacy_root(&legacy);

    assert_eq!(
        store
            .read_legacy_file_url(&format!("file://{}", source.display()), &expected)
            .unwrap(),
        b"openapi: 3.0.0"
    );
    assert!(matches!(
        store.read_legacy_file_url("file:///etc/passwd", &expected),
        Err(ArtifactError::UnsafeRoot)
    ));
    assert_eq!(
        store
            .read_legacy_file_url(
                &format!("file://{}", source.display()),
                &ArtifactRef::from_digest_hex(&"0".repeat(64)).unwrap()
            )
            .unwrap_err(),
        ArtifactError::Integrity
    );
    let outside = root.0.join("outside-legacy.yaml");
    fs::write(&outside, b"openapi: 3.0.0").unwrap();
    fs::remove_file(&source).unwrap();
    std::os::unix::fs::symlink(&outside, &source).unwrap();
    assert!(
        store
            .read_legacy_file_url(&format!("file://{}", source.display()), &expected)
            .is_err()
    );

    let maximum = legacy.join("maximum.bin");
    let maximum_bytes = vec![b'm'; MAX_ARTIFACT_BYTES];
    fs::write(&maximum, &maximum_bytes).unwrap();
    let maximum_ref =
        ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(&maximum_bytes))).unwrap();
    assert_eq!(
        store
            .read_legacy_file_url(&format!("file://{}", maximum.display()), &maximum_ref)
            .unwrap(),
        maximum_bytes
    );
    assert!(maximum.exists());
}

#[test]
fn legacy_reader_handles_lexical_parent_components_without_following_symlinks() {
    let root = TestRoot::new("legacy-normalized");
    let base = root.0.join("base");
    let child = base.join("child");
    let legacy = base.join("legacy");
    fs::create_dir_all(&child).unwrap();
    fs::create_dir(&legacy).unwrap();
    fs::set_permissions(&base, fs::Permissions::from_mode(0o700)).unwrap();
    fs::set_permissions(&child, fs::Permissions::from_mode(0o700)).unwrap();
    fs::set_permissions(&legacy, fs::Permissions::from_mode(0o700)).unwrap();
    let source = legacy.join("source.yaml");
    fs::write(&source, b"normalized legacy").unwrap();
    let expected =
        ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"normalized legacy")))
            .unwrap();
    let configured = child.join("..").join("legacy");
    let store = ArtifactStore::open(&root.0)
        .unwrap()
        .with_legacy_root(configured);

    assert_eq!(
        store
            .read_legacy_file_url(&format!("file://{}", source.display()), &expected)
            .unwrap(),
        b"normalized legacy"
    );

    let outside_dir = base.join("outside");
    fs::create_dir(&outside_dir).unwrap();
    fs::write(outside_dir.join("source.yaml"), b"normalized legacy").unwrap();
    std::os::unix::fs::symlink(&outside_dir, legacy.join("linked")).unwrap();
    assert!(
        store
            .read_legacy_file_url(
                &format!("file://{}/linked/source.yaml", legacy.display()),
                &expected,
            )
            .is_err()
    );
}

#[test]
fn legacy_oversize_is_rejected() {
    let root = TestRoot::new("legacy-max");
    let legacy = root.0.join("legacy");
    fs::create_dir(&legacy).unwrap();
    fs::set_permissions(&legacy, fs::Permissions::from_mode(0o700)).unwrap();
    let source = legacy.join("large.yaml");
    fs::write(&source, vec![b'x'; MAX_ARTIFACT_BYTES + 1]).unwrap();
    let expected = ArtifactRef::from_digest_hex(&format!(
        "{:x}",
        Sha256::digest(vec![b'x'; MAX_ARTIFACT_BYTES + 1])
    ))
    .unwrap();
    let store = ArtifactStore::open(&root.0)
        .unwrap()
        .with_legacy_root(&legacy);
    assert_eq!(
        store
            .read_legacy_file_url(&format!("file://{}", source.display()), &expected)
            .unwrap_err(),
        ArtifactError::Integrity
    );
}

#[test]
fn fifo_entries_fail_without_blocking_reads() {
    let root = TestRoot::new("fifo");
    let expected =
        ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"fifo bytes"))).unwrap();
    let sha = root.0.join("sha256");
    let shard = sha.join(&expected.digest_hex()[..2]);
    fs::create_dir(&sha).unwrap();
    fs::create_dir(&shard).unwrap();
    fs::set_permissions(&sha, fs::Permissions::from_mode(0o700)).unwrap();
    fs::set_permissions(&shard, fs::Permissions::from_mode(0o700)).unwrap();
    let blob_fifo = shard.join(expected.digest_hex());
    let legacy = root.0.join("legacy");
    fs::create_dir(&legacy).unwrap();
    fs::set_permissions(&legacy, fs::Permissions::from_mode(0o700)).unwrap();
    let legacy_fifo = legacy.join("source");
    for path in [&blob_fifo, &legacy_fifo] {
        let path = std::ffi::CString::new(path.as_os_str().as_encoded_bytes()).unwrap();
        assert_eq!(unsafe { libc::mkfifo(path.as_ptr(), 0o400) }, 0);
    }

    let executable = std::env::current_exe().unwrap();
    for kind in ["blob", "legacy"] {
        let child = Command::new(&executable)
            .args(["--exact", "fifo_read_child", "--nocapture"])
            .env("CRANK_ARTIFACTS_FIFO_ROOT", &root.0)
            .env("CRANK_ARTIFACTS_FIFO_KIND", kind)
            .env("CRANK_ARTIFACTS_FIFO_REF", expected.as_str())
            .spawn()
            .unwrap();
        assert!(wait_for_child(child, Duration::from_secs(2)).success());
    }
}

#[test]
fn fifo_read_child() {
    let Some(root) = std::env::var_os("CRANK_ARTIFACTS_FIFO_ROOT") else {
        return;
    };
    let root = PathBuf::from(root);
    let kind = std::env::var("CRANK_ARTIFACTS_FIFO_KIND").unwrap();
    let expected = ArtifactRef::parse(&std::env::var("CRANK_ARTIFACTS_FIFO_REF").unwrap()).unwrap();
    let store = ArtifactStore::open(&root).unwrap();
    let result = if kind == "blob" {
        store.read(&expected)
    } else {
        store
            .with_legacy_root(root.join("legacy"))
            .read_legacy_file_url(
                &format!("file://{}/legacy/source", root.display()),
                &expected,
            )
    };
    assert_eq!(result.unwrap_err(), ArtifactError::Integrity);
}

struct ShortReader {
    bytes: Vec<u8>,
    offset: usize,
    interrupted: bool,
}

struct FailingReader;

struct HousekeepingReader {
    store: ArtifactStore,
    candidate: Option<StaleTemp>,
    bytes: io::Cursor<Vec<u8>>,
}

impl Read for FailingReader {
    fn read(&mut self, _buf: &mut [u8]) -> io::Result<usize> {
        Err(io::Error::other("private reader detail"))
    }
}
impl Read for HousekeepingReader {
    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
        if let Some(candidate) = self.candidate.take() {
            self.store.delete_stale_temp(candidate).unwrap();
        }
        self.bytes.read(buffer)
    }
}
impl Read for ShortReader {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if !self.interrupted {
            self.interrupted = true;
            return Err(io::Error::from(io::ErrorKind::Interrupted));
        }
        if self.offset == self.bytes.len() {
            return Ok(0);
        }
        let amount = 1.min(buf.len()).min(self.bytes.len() - self.offset);
        buf[..amount].copy_from_slice(&self.bytes[self.offset..self.offset + amount]);
        self.offset += amount;
        Ok(amount)
    }
}

#[test]
fn streaming_input_handles_short_reads_and_interruption() {
    let root = TestRoot::new("stream");
    let store = ArtifactStore::open(&root.0).unwrap();
    let mut reader = ShortReader {
        bytes: b"short chunk source".to_vec(),
        offset: 0,
        interrupted: false,
    };
    let stored = store.put_reader(&mut reader).unwrap();
    assert_eq!(
        store.read(&stored.artifact_ref).unwrap(),
        b"short chunk source"
    );

    assert_eq!(
        store.put_reader(&mut FailingReader).unwrap_err(),
        ArtifactError::Storage
    );
    for error in [
        ArtifactError::InvalidReference,
        ArtifactError::Integrity,
        ArtifactError::Storage,
        ArtifactError::UnsafeRoot,
    ] {
        let diagnostic = format!("{error} {error:?}");
        assert!(!diagnostic.contains("private reader detail"));
        assert!(!diagnostic.contains(std::env::temp_dir().to_string_lossy().as_ref()));
    }
}

#[test]
fn user_reader_can_run_housekeeping_without_self_deadlock() {
    use std::sync::mpsc;

    let root = TestRoot::new("reader-housekeeping");
    let store = ArtifactStore::open(&root.0).unwrap();
    let stored = store.put(b"reader housekeeping shard").unwrap();
    let temp = root
        .0
        .join("sha256")
        .join(&stored.artifact_ref.digest_hex()[..2])
        .join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-10");
    fs::write(&temp, b"partial").unwrap();
    let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 1).unwrap();
    let mut reader = HousekeepingReader {
        store: store.clone(),
        candidate: candidates.into_iter().next(),
        bytes: io::Cursor::new(b"reader bytes".to_vec()),
    };
    let writer_store = store.clone();
    let (sender, receiver) = mpsc::channel();
    thread::spawn(move || {
        sender.send(writer_store.put_reader(&mut reader)).unwrap();
    });

    let stored = receiver
        .recv_timeout(Duration::from_secs(2))
        .expect("put_reader deadlocked while its reader ran housekeeping")
        .unwrap();
    assert_eq!(store.read(&stored.artifact_ref).unwrap(), b"reader bytes");
    assert!(!temp.exists());
}

#[test]
#[cfg(debug_assertions)]
fn forked_crash_checkpoints_publish_only_complete_or_absent_blobs() {
    let _guard = fault_guard();
    if let Some(root) = std::env::var_os("CRANK_ARTIFACTS_CHILD_ROOT") {
        set_checkpoint(
            std::env::var("CRANK_ARTIFACTS_TEST_STAGE").unwrap(),
            FaultAction::Exit,
        );
        let store = ArtifactStore::open(PathBuf::from(root)).unwrap();
        let _ = store.put(b"crash-consistent bytes");
        panic!("checkpoint did not terminate the child");
    }

    for stage in ["write", "file_fsync", "publish", "directory_fsync"] {
        let root = TestRoot::new(stage);
        let status = Command::new(std::env::current_exe().unwrap())
            .args([
                "--exact",
                "forked_crash_checkpoints_publish_only_complete_or_absent_blobs",
                "--nocapture",
            ])
            .env("CRANK_ARTIFACTS_CHILD_ROOT", &root.0)
            .env("CRANK_ARTIFACTS_TEST_STAGE", stage)
            .status()
            .unwrap();
        assert_eq!(status.code(), Some(86), "stage={stage}");

        let store = ArtifactStore::open(&root.0).unwrap();
        let expected = ArtifactRef::from_digest_hex(&format!(
            "{:x}",
            Sha256::digest(b"crash-consistent bytes")
        ))
        .unwrap();
        match store.read(&expected) {
            Ok(bytes) => assert_eq!(bytes, b"crash-consistent bytes", "stage={stage}"),
            Err(ArtifactError::NotFound) => {}
            Err(error) => panic!("stage={stage} exposed an invalid final blob: {error}"),
        }
        let stored = store.put(b"crash-consistent bytes").unwrap();
        assert_eq!(stored.artifact_ref, expected, "stage={stage}");
        assert_eq!(
            store.read(&expected).unwrap(),
            b"crash-consistent bytes",
            "stage={stage}"
        );

        let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 16).unwrap();
        for candidate in candidates {
            store.delete_stale_temp(candidate).unwrap();
        }
    }
}

#[test]
#[cfg(debug_assertions)]
fn disk_full_fault_seams_leave_a_retryable_store() {
    let _guard = fault_guard();
    if let Some(root) = std::env::var_os("CRANK_ARTIFACTS_CHILD_ROOT") {
        set_checkpoint(
            std::env::var("CRANK_ARTIFACTS_TEST_STAGE").unwrap(),
            FaultAction::Fail,
        );
        let store = ArtifactStore::open(PathBuf::from(root)).unwrap();
        assert_eq!(
            store.put(b"disk-full seam bytes").unwrap_err(),
            ArtifactError::Storage
        );
        return;
    }

    for stage in ["write", "file_fsync", "publish", "directory_fsync"] {
        let root = TestRoot::new(&format!("full-{stage}"));
        let status = Command::new(std::env::current_exe().unwrap())
            .args([
                "--exact",
                "disk_full_fault_seams_leave_a_retryable_store",
                "--nocapture",
            ])
            .env("CRANK_ARTIFACTS_CHILD_ROOT", &root.0)
            .env("CRANK_ARTIFACTS_TEST_STAGE", stage)
            .status()
            .unwrap();
        assert!(status.success(), "stage={stage}");

        let store = ArtifactStore::open(&root.0).unwrap();
        let expected =
            ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"disk-full seam bytes")))
                .unwrap();
        match store.read(&expected) {
            Ok(bytes) => assert_eq!(bytes, b"disk-full seam bytes", "stage={stage}"),
            Err(ArtifactError::NotFound) => {}
            Err(error) => panic!("stage={stage} exposed an invalid final blob: {error}"),
        }
        assert_eq!(
            store.put(b"disk-full seam bytes").unwrap().artifact_ref,
            expected,
            "stage={stage}"
        );
        assert_eq!(
            store.read(&expected).unwrap(),
            b"disk-full seam bytes",
            "stage={stage}"
        );
    }
}

#[test]
fn real_process_write_limit_leaves_no_partial_final_blob() {
    if let Some(root) = std::env::var_os("CRANK_ARTIFACTS_CHILD_ROOT") {
        // Keep the test's host filesystem untouched while making the actual
        // write syscall fail as it would for a quota/disk-full condition.
        unsafe {
            assert_ne!(libc::signal(libc::SIGXFSZ, libc::SIG_IGN), libc::SIG_ERR);
            let limit = libc::rlimit {
                rlim_cur: 1,
                rlim_max: 1,
            };
            assert_eq!(libc::setrlimit(libc::RLIMIT_FSIZE, &limit), 0);
        }
        let store = ArtifactStore::open(PathBuf::from(root)).unwrap();
        assert_eq!(
            store.put(b"real write limit bytes").unwrap_err(),
            ArtifactError::Storage
        );
        return;
    }

    let root = TestRoot::new("real-write-limit");
    let status = Command::new(std::env::current_exe().unwrap())
        .args([
            "--exact",
            "real_process_write_limit_leaves_no_partial_final_blob",
            "--nocapture",
        ])
        .env("CRANK_ARTIFACTS_CHILD_ROOT", &root.0)
        .status()
        .unwrap();
    assert!(status.success());

    let store = ArtifactStore::open(&root.0).unwrap();
    let expected =
        ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"real write limit bytes")))
            .unwrap();
    assert_eq!(store.read(&expected).unwrap_err(), ArtifactError::NotFound);
    assert_eq!(
        store.put(b"real write limit bytes").unwrap().artifact_ref,
        expected
    );
    assert_eq!(store.read(&expected).unwrap(), b"real write limit bytes");
}