use std::path::PathBuf;
use quilt_rs::io::remote::HostConfig;
use quilt_uri::Namespace;
use crate::cli::Error;
use crate::cli::model::Commands;
use crate::cli::output::Render;
use crate::cli::output::Std;
use crate::cli::text::printable;
use crate::cli::text::printable_line;
#[derive(Debug)]
pub struct Input {
pub namespace: Namespace,
pub host_config: Option<HostConfig>,
}
#[derive(Debug)]
pub struct Output {
pub report: Option<quilt_rs::flow::PullReport>,
}
fn group(f: &mut std::fmt::Formatter<'_>, heading: &str, paths: &[PathBuf]) -> std::fmt::Result {
if paths.is_empty() {
return Ok(());
}
let plural = if paths.len() == 1 { "" } else { "s" };
write!(f, "\n{} file{plural} {heading}:", paths.len())?;
for path in paths {
write!(f, "\n {}", printable_line(&path.display().to_string()))?;
}
Ok(())
}
impl std::fmt::Display for Output {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let Some(report) = &self.report else {
return write!(f, "Already up to date");
};
write!(f, r#"Revision "{}" pulled"#, report.manifest_uri.hash)?;
group(f, "new", &report.added)?;
group(f, "new, not downloaded", &report.added_not_fetched)?;
group(f, "updated", &report.updated)?;
group(f, "removed", &report.removed)?;
if let Some(message) = report.message.as_deref() {
write!(f, "\nLatest revision: {}", printable(message))?;
}
Ok(())
}
}
impl Render for Output {
fn to_json(&self) -> serde_json::Value {
let paths = |list: &[PathBuf]| {
list.iter()
.map(|path| path.display().to_string())
.collect::<Vec<_>>()
};
let Some(report) = &self.report else {
return serde_json::json!({
"hash": null,
"manifest_uri": null,
"added": [],
"added_not_fetched": [],
"updated": [],
"removed": [],
"message": null,
});
};
serde_json::json!({
"hash": &report.manifest_uri.hash,
"manifest_uri": report.manifest_uri.to_string(),
"added": paths(&report.added),
"added_not_fetched": paths(&report.added_not_fetched),
"updated": paths(&report.updated),
"removed": paths(&report.removed),
"message": report.message.as_deref(),
})
}
}
pub async fn command(m: impl Commands, args: Input) -> Std {
Std::from_result(m.pull(args).await)
}
async fn pull_package(
local_domain: &quilt_rs::LocalDomain,
namespace: Namespace,
host_config: Option<HostConfig>,
) -> Result<quilt_rs::flow::Pulled, Error> {
match local_domain.get_installed_package(&namespace).await? {
Some(installed_package) => {
let locked = installed_package.lock().await?;
let scope = installed_package.lineage().await?.sync_scope;
Ok(Box::pin(locked.pull(host_config, scope)).await?)
}
None => Err(Error::NamespaceNotFound(namespace)),
}
}
pub async fn model(
local_domain: &quilt_rs::LocalDomain,
Input {
namespace,
host_config,
}: Input,
) -> Result<Output, Error> {
let report = match pull_package(local_domain, namespace, host_config).await? {
quilt_rs::flow::Pulled::UpToDate => None,
quilt_rs::flow::Pulled::Applied(report) => Some(report),
};
Ok(Output { report })
}
#[cfg(test)]
mod tests {
use super::*;
use test_log::test;
use crate::cli::fixtures::packages::outdated as pkg;
use crate::cli::model::install_package_into_temp_dir;
#[test(tokio::test)]
async fn live_model() -> Result<(), Error> {
let uri = pkg::URI;
let (m, _, _temp_dir) = install_package_into_temp_dir(uri).await?;
{
let local_domain = m.get_local_domain();
let output = model(
local_domain,
Input {
namespace: pkg::NAMESPACE.into(),
host_config: None,
},
)
.await?;
assert_eq!(
output.report.unwrap().manifest_uri.hash,
pkg::LATEST_TOP_HASH
);
}
Ok(())
}
#[test(tokio::test)]
async fn live_pull_follows_the_stored_folders_scope() -> Result<(), Error> {
use crate::cli::fixtures::packages::revision_report as report_pkg;
use crate::cli::keep;
const KEPT: &str =
"add/deeply/nested/directory/with/a/very-long-name/summary-of-everything.parquet";
const OUTSIDE: &str = "add/one.txt";
let (m, installed, _temp_dir) = install_package_into_temp_dir(report_pkg::R1_URI).await?;
let namespace: Namespace = report_pkg::NAMESPACE_STR.try_into()?;
m.keep(keep::Input {
namespace: namespace.clone(),
scope: Some(quilt_rs::lineage::SyncScope::folders(["add/deeply"])?),
download: false,
})
.await?;
let report = model(
m.get_local_domain(),
Input {
namespace,
host_config: None,
},
)
.await?
.report
.expect("r1 is behind");
let home = installed.package_home().await?;
assert!(
home.join(KEPT).exists(),
"the kept folder's new file is downloaded"
);
assert!(!home.join(OUTSIDE).exists(), "a new file outside it is not");
assert!(report.added.contains(&PathBuf::from(KEPT)), "{report:?}");
assert!(
report.added_not_fetched.contains(&PathBuf::from(OUTSIDE)),
"{report:?}"
);
Ok(())
}
fn manifest_uri(hash: &str) -> quilt_uri::ManifestUri {
quilt_uri::ManifestUri {
origin: None,
bucket: "bucket".to_string(),
namespace: ("demo", "sales").into(),
hash: hash.to_string(),
}
}
#[test]
fn json_always_lists_four_groups_and_a_full_uri() {
let hash = "0123456789abcdef";
let output = Output {
report: Some(quilt_rs::flow::PullReport {
manifest_uri: manifest_uri(hash),
added: vec![std::path::PathBuf::from("new.csv")],
added_not_fetched: Vec::new(),
updated: Vec::new(),
removed: Vec::new(),
message: None,
}),
};
assert_eq!(
output.to_json().to_string(),
r#"{"hash":"0123456789abcdef","manifest_uri":"quilt+s3://bucket#package=demo/sales@0123456789abcdef","added":["new.csv"],"added_not_fetched":[],"updated":[],"removed":[],"message":null}"#
);
let up_to_date = Output { report: None };
assert_eq!(
up_to_date.to_json().to_string(),
r#"{"hash":null,"manifest_uri":null,"added":[],"added_not_fetched":[],"updated":[],"removed":[],"message":null}"#
);
assert_eq!(up_to_date.to_string(), "Already up to date");
}
}