aion-server 0.31.0

use aion_package::AwlSource;

use super::{DocumentStoreError, prune, stage};

type TestResult = Result<(), Box<dyn std::error::Error>>;

/// The snapshot root under a scratch home, spelled once so the tests and the
/// production `root()` cannot drift about where snapshots live.
fn root_of(home: &tempfile::TempDir) -> std::path::PathBuf {
    home.path().join("workers/documents")
}

fn source(document: &str) -> AwlSource {
    AwlSource::new(
        "probe.awl",
        document,
        [(
            "schemas/brief.schema.json",
            b"{\"type\":\"object\"}".to_vec(),
        )],
    )
}

/// The document and every schema it imports are staged where the record's argv
/// can name them, with the import's document-relative path preserved: a worker
/// re-checks `schema("schemas/brief.schema.json")` against this tree exactly as
/// the deploy did.
#[test]
fn the_document_and_its_schema_imports_land_at_their_declared_paths() -> TestResult {
    let home = tempfile::tempdir()?;
    let staged = stage(&root_of(&home), &source("workflow probe\n"), "probe")?;

    assert!(staged.path.is_absolute());
    assert_eq!(
        staged.path.file_name().and_then(|n| n.to_str()),
        Some("probe.awl")
    );
    assert_eq!(std::fs::read_to_string(&staged.path)?, "workflow probe\n");

    let root = staged
        .path
        .parent()
        .ok_or("the document must have a parent")?;
    assert_eq!(
        std::fs::read(root.join("schemas/brief.schema.json"))?,
        b"{\"type\":\"object\"}"
    );
    assert!(
        root.starts_with(root_of(&home)),
        "{root:?} must live under the server's own document root, not the studio's tree"
    );
    Ok(())
}

/// 🔴 CONTENT-ADDRESSED BY THE DOCUMENT. Identical bytes resolve to the SAME
/// path (so a redeploy leaves the argv — and the running worker — alone) and
/// different bytes resolve to a DIFFERENT one (so a re-mint has a new argv to
/// notice). Both halves, because either alone is satisfied by a constant.
#[test]
fn identical_bytes_share_a_path_and_changed_bytes_do_not() -> TestResult {
    let home = tempfile::tempdir()?;
    let first = stage(&root_of(&home), &source("workflow probe\n"), "probe")?;
    let again = stage(&root_of(&home), &source("workflow probe\n"), "probe")?;
    let changed = stage(
        &root_of(&home),
        &source("workflow probe\n// edited\n"),
        "probe",
    )?;

    assert_eq!(first, again);
    assert_ne!(first.digest, changed.digest);
    assert_ne!(first.path, changed.path);
    // Staging alone never removes the superseded document: a record minted
    // against it may still be replaying that argv until its own re-mint
    // converges. Reclaiming is `prune`'s job, and it needs the live set.
    assert!(first.path.exists() && changed.path.exists());
    Ok(())
}

/// 🔴 THE LEAK, CLOSED — in the safe direction only. A snapshot a live record
/// still names survives; one nothing names is reclaimed. Both halves, because
/// a prune that kept everything and a prune that deleted everything each
/// satisfy one of them.
#[test]
fn prune_reclaims_only_the_snapshots_no_record_names() -> TestResult {
    let home = tempfile::tempdir()?;
    let root = root_of(&home);
    let kept = stage(&root, &source("workflow probe\n"), "probe")?;
    let dropped = stage(&root, &source("workflow probe\n// superseded\n"), "probe")?;

    let live = std::collections::BTreeSet::from([kept.path.clone()]);
    let removed = prune(&root, &live)?;

    assert_eq!(removed.len(), 1, "{removed:?}");
    assert!(kept.path.exists(), "a snapshot a record still replays");
    assert!(!dropped.path.exists(), "a snapshot nothing names");
    assert!(
        std::fs::read_to_string(&kept.path)? == "workflow probe\n",
        "the surviving snapshot must be untouched"
    );
    Ok(())
}

/// A prune with NOTHING live is still bounded by the same rule and clears the
/// root — and an absent root is not a failure. Both are the shapes a first
/// boot and a fully-retired document take.
#[test]
fn prune_over_an_empty_live_set_and_an_absent_root() -> TestResult {
    let home = tempfile::tempdir()?;
    let root = root_of(&home);
    assert!(prune(&root, &std::collections::BTreeSet::new())?.is_empty());
    let staged = stage(&root, &source("workflow probe\n"), "probe")?;
    assert_eq!(prune(&root, &std::collections::BTreeSet::new())?.len(), 1);
    assert!(!staged.path.exists());
    Ok(())
}

/// A crash mid-stage must never leave a tree that LOOKS complete. The staging
/// directory is built beside the final name and renamed in one step, so a
/// leftover `.partial` is visibly not a snapshot — and a later stage of the
/// same bytes clears it rather than inheriting it.
#[test]
fn an_interrupted_stage_leaves_no_snapshot_and_is_cleaned_by_the_next() -> TestResult {
    let home = tempfile::tempdir()?;
    let root = root_of(&home);
    let staged = stage(&root, &source("workflow probe\n"), "probe")?;
    let version = staged
        .path
        .parent()
        .and_then(|parent| parent.file_name())
        .and_then(|name| name.to_str())
        .ok_or("the staged document must sit in a named directory")?
        .to_owned();

    // Simulate the crash: a half-written staging tree left behind, and the
    // real snapshot removed as if the rename never happened.
    std::fs::remove_dir_all(staged.path.parent().ok_or("parent")?)?;
    let partial = root.join(format!("{version}.partial"));
    std::fs::create_dir_all(&partial)?;
    std::fs::write(partial.join("probe.awl"), "half written")?;

    let live = std::collections::BTreeSet::new();
    let removed = prune(&root, &live)?;
    assert_eq!(removed.len(), 1, "a `.partial` tree is not a snapshot");

    // And staging the same bytes again produces the real thing.
    std::fs::create_dir_all(&partial)?;
    let restaged = stage(&root, &source("workflow probe\n"), "probe")?;
    assert_eq!(restaged, staged);
    assert!(!partial.exists(), "the stale staging tree must be cleared");
    assert_eq!(std::fs::read_to_string(&restaged.path)?, "workflow probe\n");
    Ok(())
}

/// An archive is input from whoever built it. A name that climbs out of the
/// staging directory is refused BY NAME, before the capability would have
/// refused it as a bare permission error against a path nobody wrote.
#[test]
fn an_escaping_archive_name_is_refused_by_name() -> TestResult {
    let home = tempfile::tempdir()?;
    let escaping = AwlSource::new(
        "probe.awl",
        "workflow probe\n",
        [("../../escaped.json", b"x".to_vec())],
    );
    match stage(&root_of(&home), &escaping, "probe") {
        Err(DocumentStoreError::UnsafeName { name }) => {
            assert_eq!(name, "../../escaped.json");
        }
        other => return Err(format!("an escaping schema name must refuse: {other:?}").into()),
    }
    let named_escape = AwlSource::new(
        "../probe.awl",
        "workflow probe\n",
        Vec::<(String, Vec<u8>)>::new(),
    );
    match stage(&root_of(&home), &named_escape, "probe") {
        Err(DocumentStoreError::UnsafeName { name }) => {
            assert_eq!(name, "../probe.awl");
            Ok(())
        }
        other => Err(format!("an escaping document name must refuse: {other:?}").into()),
    }
}

/// A workflow type is an AWL identifier by the time a package exists — but this
/// reads an archive a stranger may have built, and a directory name is not the
/// place to find that out. Every non-identifier character is folded, and the
/// digest still distinguishes the versions.
#[test]
fn a_hostile_workflow_type_cannot_shape_the_directory_name() -> TestResult {
    let home = tempfile::tempdir()?;
    let staged = stage(&root_of(&home), &source("workflow probe\n"), "../../../etc")?;
    let directory = staged
        .path
        .parent()
        .and_then(|parent| parent.file_name())
        .and_then(|name| name.to_str())
        .ok_or("the staged document must sit in a named directory")?;
    assert!(!directory.contains('/'), "{directory}");
    assert!(!directory.contains(".."), "{directory}");
    assert!(directory.ends_with(&staged.digest), "{directory}");
    assert!(staged.path.starts_with(root_of(&home)), "{staged:?}");
    Ok(())
}