quilt-cli 0.37.0

use std::path::PathBuf;

use quilt_uri::Namespace;

use crate::cli::Error;
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,
}

#[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>,
}

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

impl Output {
    /// The requested paths that were installed: every one but the skipped.
    fn installed(&self) -> impl Iterator<Item = &PathBuf> {
        self.paths
            .iter()
            .filter(|path| !self.skipped.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();
        if installed.is_empty() {
            output.push("No paths installed".to_string());
        } else {
            for path in installed {
                output.push(format!("Path: \"{}\"", path.display()));
            }
        }
        for path in &self.skipped {
            output.push(format!(
                "Skipped: \"{}\" (no longer on the remote)",
                path.display()
            ));
        }
        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
                // skipped ones — 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<_>>(),
        })
    }
}

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?)
}

async fn install_paths(
    installed_package: &quilt_rs::InstalledPackage,
    paths: &[PathBuf],
) -> Result<Vec<PathBuf>, Error> {
    Ok(installed_package.install_paths(paths).await?.skipped)
}

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,
    }: 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 installed_package = install_package(local_domain, &uri, namespace).await?;
    let paths = get_entries(path, paths);

    let skipped = if paths.is_empty() {
        Vec::new()
    } else {
        install_paths(&installed_package, &paths).await?
    };

    Ok(Output {
        installed_package,
        paths,
        skipped,
    })
}

#[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,
                },
            )
            .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(),
                },
            )
            .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(),
            },
        )
        .await?;

        assert_eq!(output.to_json()["paths"], serde_json::json!([pkg::KEPT]));
        assert_eq!(
            output.to_json()["skipped"],
            serde_json::json!([pkg::REPLACED])
        );
        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(),
            },
        )
        .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"),
            ],
            skipped: vec![std::path::PathBuf::from("data/two.csv")],
        };

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

    /// A skipped path is listed once, as skipped, never as a `Path:` line: that
    /// list reads as the files now on disk.
    #[test(tokio::test)]
    async fn text_lists_a_skipped_path_only_as_skipped() -> 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")],
            skipped: vec![PathBuf::from("b.txt")],
        };

        assert!(
            format!("{output}").ends_with(concat!(
                "\nPath: \"a.txt\"",
                "\nSkipped: \"b.txt\" (no longer on the remote)"
            )),
            "{output}"
        );
        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(),
        };

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