quilt-cli 0.40.0

use std::path::PathBuf;

use quilt_rs::lineage::SyncScope;
use quilt_uri::Namespace;

use crate::cli::Error;
use crate::cli::keep;
use crate::cli::model::Commands;
use crate::cli::output::Render;
use crate::cli::output::Std;

#[derive(Debug)]
pub struct Input {
    pub namespace: Option<Namespace>,
    pub paths: Option<Vec<PathBuf>>,
    pub uri: String,
    /// The scope to keep, set after the install.
    pub keep: Option<SyncScope>,
    /// Whether to download the files `keep` covers.
    pub download: bool,
}

#[derive(Debug)]
pub struct Output {
    installed_package: quilt_rs::InstalledPackage,
    paths: Vec<std::path::PathBuf>,
    /// Requested paths not installed because the remote no longer holds their
    /// bytes (an unversioned bucket, overwritten since).
    skipped: Vec<std::path::PathBuf>,
    /// Requested paths not installed because their bytes do not match the
    /// hash the package records.
    hash_mismatch: Vec<std::path::PathBuf>,
    /// Requested paths the copy already tracked, left as they are.
    already_installed: Vec<std::path::PathBuf>,
    /// What `--keep-all` or `--keep-folder` set and downloaded.
    keep: Option<keep::Output>,
}

#[cfg(test)]
impl Output {
    pub fn get_installed_package(self) -> quilt_rs::InstalledPackage {
        self.installed_package
    }
}

impl Output {
    /// The requested paths that were installed now: every one but the
    /// skipped, the hash-mismatched and the already installed.
    fn installed(&self) -> impl Iterator<Item = &PathBuf> {
        self.paths.iter().filter(|path| {
            !self.skipped.contains(path)
                && !self.hash_mismatch.contains(path)
                && !self.already_installed.contains(path)
        })
    }
}

impl std::fmt::Display for Output {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut output = vec![format!("{}", self.installed_package)];
        let installed: Vec<_> = self.installed().collect();
        // A keep result says what it downloaded.
        if installed.is_empty() && self.keep.is_none() {
            output.push("No paths installed".to_string());
        } else {
            for path in installed {
                output.push(format!("Path: \"{}\"", path.display()));
            }
        }
        for path in &self.already_installed {
            output.push(format!("Already installed: \"{}\"", path.display()));
        }
        for path in &self.skipped {
            output.push(format!(
                "Skipped: \"{}\" (no longer on the remote)",
                path.display()
            ));
        }
        for path in &self.hash_mismatch {
            output.push(format!(
                "Hash mismatch: \"{}\" (the package may be corrupt)",
                path.display()
            ));
        }
        if let Some(keep) = &self.keep {
            output.push(keep.to_string());
        }
        write!(f, "{}", output.join("\n"))
    }
}

impl Render for Output {
    fn to_json(&self) -> serde_json::Value {
        serde_json::json!({
            "namespace": self.installed_package.namespace.to_string(),
            "paths": self
                .installed()
                // Unlike every other payload's keys, these are the caller's
                // own `&path=` and `--path` values echoed back, less the
                // ones left out and already installed — never matched against the manifest. A non-UTF-8 argument
                // renders lossily here; it also matches no manifest key, so
                // it installs nothing.
                .map(|path| path.display().to_string())
                .collect::<Vec<_>>(),
            "skipped": self
                .skipped
                .iter()
                .map(|path| path.display().to_string())
                .collect::<Vec<_>>(),
            "hash_mismatch": self
                .hash_mismatch
                .iter()
                .map(|path| path.display().to_string())
                .collect::<Vec<_>>(),
            "already_installed": self
                .already_installed
                .iter()
                .map(|path| path.display().to_string())
                .collect::<Vec<_>>(),
            "keep": self.keep.as_ref().map(Render::to_json),
        })
    }
}

pub async fn command(m: impl Commands, args: Input) -> Std {
    Std::from_result(m.install(args).await)
}

async fn install_package(
    local_domain: &quilt_rs::LocalDomain,
    uri: &quilt_uri::S3PackageUri,
    namespace: Namespace,
) -> Result<quilt_rs::InstalledPackage, Error> {
    let remote = local_domain.get_remote();
    if let Some(installed_package) = local_domain.get_installed_package(&namespace).await? {
        // TODO: check the actual remote_manifest
        return Ok(installed_package);
    }
    let manifest_uri =
        quilt_rs::io::manifest::resolve_manifest_uri(remote, uri.catalog.as_ref(), uri).await?;
    Ok(local_domain.install_package(&manifest_uri).await?)
}

fn get_entries(
    uri_path: Option<PathBuf>,
    arg_paths: Option<Vec<PathBuf>>,
) -> Vec<std::path::PathBuf> {
    let mut paths = Vec::new();
    if let Some(logical_key) = uri_path {
        paths.push(logical_key);
    }
    if let Some(logical_keys) = arg_paths {
        for logical_key in logical_keys {
            paths.push(logical_key);
        }
    }
    paths
}

pub async fn model(
    local_domain: &quilt_rs::LocalDomain,
    Input {
        namespace,
        paths,
        uri,
        keep,
        download,
    }: Input,
) -> Result<Output, Error> {
    let uri: quilt_uri::S3PackageUri = uri.parse()?;
    let path = uri.path.clone();

    let namespace = namespace.unwrap_or(uri.namespace.clone());
    let paths = get_entries(path, paths);
    if keep.is_some() && !download && !paths.is_empty() {
        return Err(Error::NoDownloadWithPath);
    }
    let installed_package = install_package(local_domain, &uri, namespace).await?;

    let report = installed_package.install_paths(&paths).await?;

    let keep = match keep {
        Some(scope) => Some(keep::keep(&installed_package, Some(scope), download).await?),
        None => None,
    };

    Ok(Output {
        installed_package,
        paths,
        skipped: report.skipped,
        hash_mismatch: report.hash_mismatch,
        already_installed: report.already_installed,
        keep,
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    use std::path::PathBuf;

    use test_log::test;

    use quilt_rs::io::storage::LocalStorage;
    use quilt_rs::io::storage::Storage;
    use quilt_rs::paths::DomainPaths;

    use crate::cli::fixtures::packages::default as pkg;
    use crate::cli::model::create_model_in_temp_dir;

    /// Verifies the installation process in CLI with valid data:
    ///   * lineage is updated with the installed package and tracked paths
    ///   * the working directory contains tracked mutable files
    ///   * `.quilt/objects` contains immutable files from the package
    ///   * `.quilt/installed` contains the installed manifest under the namespace directory
    ///   * `.quilt/packages` contains the cached manifest under the bucket directory
    /// Uses an actual manifest from Quilt without mocks.
    #[allow(
        clippy::too_many_lines,
        reason = "large integration test; allowed per-test so new large tests stay flagged"
    )]
    #[test(tokio::test)]
    async fn live_model() -> Result<(), Error> {
        let uri = format!("{}&path={}", pkg::URI, pkg::README_LK_ESCAPED);

        let readme_logical_key = PathBuf::from(pkg::README_LK);
        let timestamp_logical_key = PathBuf::from(pkg::TIMESTAMP_LK);

        let (m, temp_dir) = create_model_in_temp_dir().await?;
        let working_dir = temp_dir.path().join(pkg::NAMESPACE_STR);
        {
            let local_domain = m.get_local_domain();

            let output = model(
                local_domain,
                Input {
                    namespace: None,
                    paths: Some(vec![timestamp_logical_key.clone()]),
                    uri,
                    keep: None,
                    download: true,
                },
            )
            .await?;

            assert_eq!(
                format!("{output}"),
                format!(
                    "Installed package \"{}\"\nPath: \"{}\"\nPath: \"{}\"",
                    pkg::NAMESPACE_STR,
                    pkg::README_LK,
                    pkg::TIMESTAMP_LK,
                )
            );

            let installed_package = output.installed_package;
            assert_eq!(installed_package.namespace, (pkg::NAMESPACE).into());
            assert!(
                installed_package
                    .lineage()
                    .await?
                    .paths
                    .contains_key(&readme_logical_key)
            );

            assert_eq!(
                installed_package.package_home().await?,
                PathBuf::from(temp_dir.as_ref()).join(pkg::NAMESPACE_STR)
            );
            assert_eq!(
                output.paths,
                vec![readme_logical_key.clone(), timestamp_logical_key.clone()]
            );
        }

        let storage = LocalStorage::new();
        assert!(storage.exists(working_dir.join(&readme_logical_key)).await);
        assert!(
            storage
                .exists(working_dir.join(&timestamp_logical_key))
                .await
        );

        let domain_paths = DomainPaths::new(temp_dir.path().to_path_buf());

        assert!(
            storage
                .exists(domain_paths.installed_manifest(&pkg::NAMESPACE.into(), pkg::TOP_HASH))
                .await
        );

        assert!(
            storage
                .exists(
                    domain_paths
                        .cached_manifests_dir("data-yaml-spec-tests")
                        .join(pkg::TOP_HASH)
                )
                .await
        );

        // README.md
        assert!(
            storage
                .exists(
                    domain_paths
                        .objects_dir()
                        .join("3e5e75033079a0b5bfaeff79c8f10dbc3f461e283ad8126c333cd74792e62ea7")
                )
                .await
        );
        // timestamp.txt
        assert!(
            storage
                .exists(
                    domain_paths
                        .objects_dir()
                        .join("dc3ea61d9a4aaf7d822eed1de089db83d46aa29f3fbdd99466f7e5e216c91c8a")
                )
                .await
        );

        {
            let local_domain = m.get_local_domain();
            let install_once_more = model(
                local_domain,
                Input {
                    namespace: None,
                    paths: None,
                    uri: pkg::URI.to_string(),
                    keep: None,
                    download: true,
                },
            )
            .await?;

            // No paths installed during this call
            assert_eq!(
                format!("{install_once_more}"),
                format!(
                    "Installed package \"{}\"\nNo paths installed",
                    pkg::NAMESPACE_STR,
                )
            );

            assert_eq!(
                install_once_more.installed_package.namespace,
                pkg::NAMESPACE.into(),
            );

            // However paths are still tracked, because we didn't install a package anew,
            // but re-use installed package from the previous call.
            let paths = install_once_more.installed_package.lineage().await?.paths;
            assert!(paths.contains_key(&readme_logical_key));
            assert!(paths.contains_key(&timestamp_logical_key));
        }

        Ok(())
    }

    /// On a bucket without versioning, an older revision's key may name bytes a
    /// later revision wrote. Installing r1 must not put r2's `b.txt` under r1's
    /// row, where it would read as Modified. It skips `b.txt`, reports it,
    /// installs `a.txt`, and leaves status clean. Asking again still skips it,
    /// because the wrong bytes never entered the object store.
    #[test(tokio::test)]
    async fn live_unversioned_install_skips_a_replaced_file_and_installs_the_rest()
    -> Result<(), Error> {
        use crate::cli::fixtures::packages::unversioned as pkg;

        let (m, temp_dir) = create_model_in_temp_dir().await?;
        let working_dir = temp_dir.path().join(pkg::NAMESPACE_STR);
        let requested = vec![PathBuf::from(pkg::KEPT), PathBuf::from(pkg::REPLACED)];

        let output = model(
            m.get_local_domain(),
            Input {
                namespace: None,
                paths: Some(requested.clone()),
                uri: pkg::R1_URI.to_string(),
                keep: None,
                download: true,
            },
        )
        .await?;

        assert_eq!(output.to_json()["paths"], serde_json::json!([pkg::KEPT]));
        assert_eq!(
            output.to_json()["skipped"],
            serde_json::json!([pkg::REPLACED])
        );
        assert_eq!(output.to_json()["hash_mismatch"], serde_json::json!([]));
        assert!(
            format!("{output}").ends_with(&format!(
                "Skipped: \"{}\" (no longer on the remote)",
                pkg::REPLACED
            )),
            "{output}"
        );
        assert_eq!(
            tokio::fs::read(working_dir.join(pkg::KEPT)).await?,
            pkg::KEPT_R1_BODY
        );
        assert!(
            !working_dir.join(pkg::REPLACED).exists(),
            "r2's bytes were placed under r1's row"
        );

        let installed = output.get_installed_package();
        let lineage = installed.lineage().await?;
        assert_eq!(
            lineage.paths.keys().collect::<Vec<_>>(),
            vec![&PathBuf::from(pkg::KEPT)]
        );
        let status = installed.status(None).await?;
        assert!(
            status.changes.is_empty(),
            "untouched files read as changed: {:?}",
            status.changes.keys().collect::<Vec<_>>()
        );

        let retry = model(
            m.get_local_domain(),
            Input {
                namespace: None,
                paths: Some(vec![PathBuf::from(pkg::REPLACED)]),
                uri: pkg::R1_URI.to_string(),
                keep: None,
                download: true,
            },
        )
        .await?;
        assert_eq!(
            retry.to_json()["skipped"],
            serde_json::json!([pkg::REPLACED])
        );
        assert!(!working_dir.join(pkg::REPLACED).exists());
        Ok(())
    }

    #[test(tokio::test)]
    async fn json_carries_namespace_and_paths() -> Result<(), Error> {
        use crate::cli::create;
        use crate::cli::model::create_model_in_temp_dir;

        let (m, _temp_dir) = create_model_in_temp_dir().await?;
        let created = m
            .create(create::Input {
                namespace: ("test", "pkg").into(),
                source: None,
                message: None,
            })
            .await?;

        let output = Output {
            installed_package: created.installed_package,
            paths: vec![
                std::path::PathBuf::from("data/one.csv"),
                std::path::PathBuf::from("data/two.csv"),
                std::path::PathBuf::from("data/three.csv"),
            ],
            skipped: vec![std::path::PathBuf::from("data/two.csv")],
            hash_mismatch: vec![std::path::PathBuf::from("data/three.csv")],
            already_installed: Vec::new(),
            keep: None,
        };

        assert_eq!(
            output.to_json().to_string(),
            r#"{"namespace":"test/pkg","paths":["data/one.csv"],"skipped":["data/two.csv"],"hash_mismatch":["data/three.csv"],"already_installed":[],"keep":null}"#
        );
        Ok(())
    }

    /// A tracked path keeps its local edit and is reported as already
    /// installed; the other requested path installs.
    #[test(tokio::test)]
    async fn live_tracked_path_is_already_installed_and_the_rest_install() -> Result<(), Error> {
        let (m, temp_dir) = create_model_in_temp_dir().await?;
        let working_dir = temp_dir.path().join(pkg::NAMESPACE_STR);
        let readme = PathBuf::from(pkg::README_LK);
        let timestamp = PathBuf::from(pkg::TIMESTAMP_LK);

        model(
            m.get_local_domain(),
            Input {
                namespace: None,
                paths: Some(vec![readme.clone()]),
                uri: pkg::URI.to_string(),
                keep: None,
                download: true,
            },
        )
        .await?;
        tokio::fs::write(working_dir.join(&readme), b"users edit").await?;

        let output = model(
            m.get_local_domain(),
            Input {
                namespace: None,
                paths: Some(vec![readme.clone(), timestamp.clone()]),
                uri: pkg::URI.to_string(),
                keep: None,
                download: true,
            },
        )
        .await?;

        assert_eq!(
            output.to_json()["paths"],
            serde_json::json!([pkg::TIMESTAMP_LK])
        );
        assert_eq!(
            output.to_json()["already_installed"],
            serde_json::json!([pkg::README_LK])
        );
        assert!(
            format!("{output}").ends_with(&format!(
                "Path: \"{}\"\nAlready installed: \"{}\"",
                pkg::TIMESTAMP_LK,
                pkg::README_LK
            )),
            "{output}"
        );
        assert_eq!(
            tokio::fs::read(working_dir.join(&readme)).await?,
            b"users edit"
        );
        assert!(working_dir.join(&timestamp).exists());
        Ok(())
    }

    /// A skipped or hash-mismatched path is listed once, as such, never as a
    /// `Path:` line: that list reads as the files now on disk.
    #[test(tokio::test)]
    async fn text_lists_a_path_left_out_only_as_left_out() -> Result<(), Error> {
        use crate::cli::create;
        use crate::cli::model::create_model_in_temp_dir;

        let (m, _temp_dir) = create_model_in_temp_dir().await?;
        let created = m
            .create(create::Input {
                namespace: ("test", "pkg").into(),
                source: None,
                message: None,
            })
            .await?;

        let output = Output {
            installed_package: created.installed_package,
            paths: vec![
                PathBuf::from("a.txt"),
                PathBuf::from("b.txt"),
                PathBuf::from("d.txt"),
            ],
            skipped: vec![PathBuf::from("b.txt")],
            hash_mismatch: vec![PathBuf::from("d.txt")],
            already_installed: vec![PathBuf::from("c.txt")],
            keep: None,
        };

        assert!(
            format!("{output}").ends_with(concat!(
                "\nPath: \"a.txt\"",
                "\nAlready installed: \"c.txt\"",
                "\nSkipped: \"b.txt\" (no longer on the remote)",
                "\nHash mismatch: \"d.txt\" (the package may be corrupt)"
            )),
            "{output}"
        );
        Ok(())
    }

    /// Install `pkg` keeping the folder `one`, spelled `./one/`.
    async fn install_keeping_one(download: bool) -> Result<(Output, tempfile::TempDir), Error> {
        let (m, temp_dir) = create_model_in_temp_dir().await?;
        let output = model(
            m.get_local_domain(),
            Input {
                namespace: None,
                paths: None,
                uri: pkg::URI.to_string(),
                keep: Some(SyncScope::folders(["./one/"])?),
                download,
            },
        )
        .await?;
        Ok((output, temp_dir))
    }

    /// The package's one file under `one` is downloaded; `timestamp.txt`,
    /// outside it, is not.
    #[test(tokio::test)]
    async fn live_install_keeping_a_folder_downloads_its_files() -> Result<(), Error> {
        let (output, _temp_dir) = install_keeping_one(true).await?;

        let lineage = output.installed_package.lineage().await?;
        assert_eq!(
            lineage.sync_scope,
            SyncScope::Folders(vec!["one".to_string()])
        );
        assert_eq!(
            lineage.paths.into_keys().collect::<Vec<_>>(),
            vec![PathBuf::from(pkg::README_LK)]
        );
        assert_eq!(output.to_json()["keep"]["outstanding"], 0);
        Ok(())
    }

    #[test(tokio::test)]
    async fn live_install_keeping_a_folder_without_download_fetches_nothing() -> Result<(), Error> {
        let (output, _temp_dir) = install_keeping_one(false).await?;

        let lineage = output.installed_package.lineage().await?;
        assert_eq!(
            lineage.sync_scope,
            SyncScope::Folders(vec!["one".to_string()])
        );
        assert_eq!(lineage.paths, quilt_rs::lineage::LineagePaths::new());
        assert_eq!(output.to_json()["keep"]["outstanding"], 1);
        Ok(())
    }

    /// Checked before anything is installed.
    #[test(tokio::test)]
    async fn no_download_refuses_a_path_in_the_uri() -> Result<(), Error> {
        use crate::cli::model::create_model_in_temp_dir;

        let (m, _temp_dir) = create_model_in_temp_dir().await?;
        let err = model(
            m.get_local_domain(),
            Input {
                namespace: None,
                paths: None,
                uri: "quilt+s3://bucket#package=a/b&path=data.csv".to_string(),
                keep: Some(SyncScope::EntirePackage),
                download: false,
            },
        )
        .await
        .unwrap_err();
        assert!(matches!(err, Error::NoDownloadWithPath), "{err}");
        Ok(())
    }

    /// A copy that installed no paths reports `[]`, not a missing key.
    #[test(tokio::test)]
    async fn json_paths_are_an_empty_list_when_none_installed() -> Result<(), Error> {
        use crate::cli::create;
        use crate::cli::model::create_model_in_temp_dir;

        let (m, _temp_dir) = create_model_in_temp_dir().await?;
        let created = m
            .create(create::Input {
                namespace: ("test", "bare").into(),
                source: None,
                message: None,
            })
            .await?;

        let output = Output {
            installed_package: created.installed_package,
            paths: Vec::new(),
            skipped: Vec::new(),
            hash_mismatch: Vec::new(),
            already_installed: Vec::new(),
            keep: None,
        };

        assert_eq!(
            output.to_json().to_string(),
            r#"{"namespace":"test/bare","paths":[],"skipped":[],"hash_mismatch":[],"already_installed":[],"keep":null}"#
        );
        Ok(())
    }
}