use std::borrow::Cow;
use quilt_rs::lineage::PackageLineage;
use quilt_rs::lineage::PackageState;
use quilt_rs::lineage::PackageStateReport;
use quilt_rs::lineage::UpstreamState;
use quilt_uri::Namespace;
use tabled::settings::Modify;
use tabled::settings::Span;
use tracing::log;
use crate::cli::Error;
use crate::cli::model::Commands;
use crate::cli::output::Render;
use crate::cli::output::Std;
const NO_BUCKET: &str = "∅";
const LAST_SYNCED_HINT: &str = "Statuses are as of each package's last install, pull or push. \
Run `quilt status --namespace <namespace>` to check the remote now.";
const FETCHED_HINT: &str = "Statuses were checked against each package's remote just now, \
except where marked unreachable. Changed files are ones not committed yet.";
const UNREACHABLE_SUFFIX: &str = " (remote unreachable)";
const FETCH_CONCURRENCY: usize = 8;
#[derive(Debug, Default)]
pub struct Input {
pub fetch: bool,
}
pub struct PackageEntry {
pub bucket: Option<String>,
pub namespace: Namespace,
pub status: UpstreamState,
pub fetched: Option<Fetched>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Fetched {
pub state: PackageState,
pub changed_files: Option<usize>,
pub access_denied: bool,
pub remote_checked: bool,
}
impl Fetched {
fn failed() -> Self {
Self {
state: PackageState::Unknown,
changed_files: None,
access_denied: false,
remote_checked: false,
}
}
}
impl From<PackageStateReport> for Fetched {
fn from(report: PackageStateReport) -> Self {
Self {
state: report.state,
changed_files: report.changed_files,
access_denied: false,
remote_checked: report.latest_refreshed,
}
}
}
pub struct Output {
installed_packages_list: Vec<PackageEntry>,
fetched: bool,
}
impl Output {
fn new(mut installed_packages_list: Vec<PackageEntry>, fetched: bool) -> Self {
installed_packages_list.sort_by(|a, b| {
(a.bucket.is_none(), &a.bucket, &a.namespace).cmp(&(
b.bucket.is_none(),
&b.bucket,
&b.namespace,
))
});
Self {
installed_packages_list,
fetched,
}
}
fn column(&self) -> &'static str {
if self.fetched {
"status"
} else {
"last synced"
}
}
fn hint(&self) -> &'static str {
if self.fetched {
FETCHED_HINT
} else {
LAST_SYNCED_HINT
}
}
}
fn last_synced(entry: &PackageEntry) -> &'static str {
match entry.status {
UpstreamState::UpToDate => "synced",
UpstreamState::Ahead => "unpushed commit",
UpstreamState::Behind => "behind",
UpstreamState::Diverged => "diverged",
UpstreamState::Local if entry.bucket.is_some() => "never pushed",
UpstreamState::Local => "local only",
UpstreamState::Error => "unknown",
}
}
fn fetched_state(fetched: &Fetched) -> Cow<'static, str> {
if fetched.access_denied {
return "no access".into();
}
match fetched.state {
PackageState::Latest => "latest".into(),
PackageState::PendingCommit => "revision not published".into(),
PackageState::PendingChanges { files: 1 } => "1 file changed".into(),
PackageState::PendingChanges { files } => format!("{files} files changed").into(),
PackageState::Behind => "not the latest".into(),
PackageState::Diverged => "changed in both places".into(),
PackageState::Unpublished => "not published yet".into(),
PackageState::NoRemote => "no S3 bucket yet".into(),
PackageState::Unknown => "unknown".into(),
}
}
fn state(entry: &PackageEntry) -> Cow<'static, str> {
match &entry.fetched {
Some(fetched) if fetched.changed_files.is_some() && !fetched.remote_checked => {
format!("{}{UNREACHABLE_SUFFIX}", fetched_state(fetched)).into()
}
Some(fetched) => fetched_state(fetched),
None => last_synced(entry).into(),
}
}
impl std::fmt::Display for Output {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
if self.installed_packages_list.is_empty() {
return write!(f, "No installed packages");
}
let mut builder = tabled::builder::Builder::default();
builder.push_record(["bucket", "namespace", self.column()]);
for entry in &self.installed_packages_list {
builder.push_record([
Cow::Borrowed(entry.bucket.as_deref().unwrap_or(NO_BUCKET)),
Cow::Owned(entry.namespace.to_string()),
state(entry),
]);
}
let mut table = builder.build();
let mut row = 1; for group in self
.installed_packages_list
.chunk_by(|a, b| a.bucket == b.bucket)
{
if group.len() > 1 {
table.with(Modify::new((row, 0)).with(Span::row(group.len().cast_signed())));
}
row += group.len();
}
write!(f, "{table}\n{}", self.hint())
}
}
impl Render for Output {
fn to_json(&self) -> serde_json::Value {
let packages: Vec<_> = self
.installed_packages_list
.iter()
.map(|package| {
let mut row = serde_json::json!({
"bucket": package.bucket.as_deref(),
"namespace": package.namespace.to_string(),
"status": package.status,
});
if let Some(fetched) = &package.fetched {
row["fetched"] = fetched.remote_checked.into();
if let Some(files) = fetched.changed_files {
row["changed_files"] = files.into();
}
if fetched.access_denied {
row["access_denied"] = true.into();
}
}
row
})
.collect();
serde_json::json!({ "packages": packages })
}
}
pub async fn command(m: impl Commands, args: Input) -> Std {
Std::from_result(m.list(args).await)
}
fn bucket(lineage: &PackageLineage) -> Option<String> {
lineage
.remote_uri
.as_ref()
.map(|remote| remote.bucket.clone())
.filter(|bucket| !bucket.is_empty())
}
pub async fn model(local_domain: &quilt_rs::LocalDomain, args: Input) -> Result<Output, Error> {
let domain_lineage = local_domain.get_lineage().await?;
let mut installed_packages_list = Vec::with_capacity(domain_lineage.packages.len());
if !args.fetch {
for (namespace, lineage) in domain_lineage.packages {
installed_packages_list.push(PackageEntry {
bucket: bucket(&lineage),
status: UpstreamState::from(lineage),
namespace,
fetched: None,
});
}
return Ok(Output::new(installed_packages_list, false));
}
let limit = std::sync::Arc::new(tokio::sync::Semaphore::new(FETCH_CONCURRENCY));
let mut tasks = tokio::task::JoinSet::new();
let mut spawned = std::collections::HashMap::new();
for (namespace, lineage) in domain_lineage.packages {
let installed_package = local_domain.create_installed_package(namespace.clone());
let bucket = bucket(&lineage);
let fallback = PackageEntry {
bucket: bucket.clone(),
namespace: namespace.clone(),
status: UpstreamState::Error,
fetched: Some(Fetched::failed()),
};
let limit = limit.clone();
let task = tasks.spawn(async move {
let _permit = limit.acquire_owned().await;
fetch_entry(namespace, bucket, || installed_package.state()).await
});
spawned.insert(task.id(), fallback);
}
while let Some(joined) = tasks.join_next_with_id().await {
match joined {
Ok((id, entry)) => {
spawned.remove(&id);
installed_packages_list.push(entry);
}
Err(err) => {
if let Some(fallback) = spawned.remove(&err.id()) {
log::warn!("Failed to check {}: {err}", fallback.namespace);
installed_packages_list.push(fallback);
}
}
}
}
Ok(Output::new(installed_packages_list, true))
}
async fn fetch_entry<F, Fut>(namespace: Namespace, bucket: Option<String>, state: F) -> PackageEntry
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = Result<PackageStateReport, quilt_rs::Error>>,
{
let (status, fetched) = match state().await {
Ok(report) => (report.upstream_state, Fetched::from(report)),
Err(err) if err.is_access_denied() => (
UpstreamState::Error,
Fetched {
access_denied: true,
..Fetched::failed()
},
),
Err(err) => {
log::warn!("Failed to check {namespace}: {err}");
(UpstreamState::Error, Fetched::failed())
}
};
PackageEntry {
bucket,
namespace,
status,
fetched: Some(fetched),
}
}
#[cfg(test)]
mod tests {
use super::*;
use test_log::test;
use crate::cli::fixtures::packages::default as pkg;
use crate::cli::model::Commands;
use crate::cli::model::create_model_in_temp_dir;
use crate::cli::model::install_package_into_temp_dir;
fn entry(bucket: Option<&str>, name: &str, status: UpstreamState) -> PackageEntry {
PackageEntry {
bucket: bucket.map(str::to_owned),
namespace: ("example", name).into(),
status,
fetched: None,
}
}
fn fetched_entry(
bucket: Option<&str>,
name: &str,
status: UpstreamState,
fetched: Fetched,
) -> PackageEntry {
PackageEntry {
fetched: Some(fetched),
..entry(bucket, name, status)
}
}
#[test(tokio::test)]
async fn test_empty_list() -> Result<(), Error> {
let (m, _temp_dir) = create_model_in_temp_dir().await?;
{
let local_domain = m.get_local_domain();
let empty_output = model(local_domain, Input::default()).await?;
assert!(empty_output.installed_packages_list.is_empty());
assert_eq!(format!("{empty_output}"), "No installed packages");
}
Ok(())
}
#[test]
fn test_json_empty_list() {
let output = Output::new(Vec::new(), false);
assert_eq!(output.to_json().to_string(), r#"{"packages":[]}"#);
}
#[test]
fn test_json_includes_bucket_namespace_and_status() {
let output = Output::new(
vec![
entry(Some("acme-research"), "one", UpstreamState::UpToDate),
entry(None, "scratch", UpstreamState::Local),
],
false,
);
assert_eq!(
output.to_json().to_string(),
r#"{"packages":[{"bucket":"acme-research","namespace":"example/one","status":"up_to_date"},{"bucket":null,"namespace":"example/scratch","status":"local"}]}"#
);
}
#[test]
fn test_display_groups_interleaved_buckets() {
let output = Output::new(
vec![
entry(Some("zulu-bucket"), "late", UpstreamState::Behind),
entry(Some("acme-research"), "two", UpstreamState::UpToDate),
entry(None, "scratch", UpstreamState::Local),
entry(Some("acme-research"), "one", UpstreamState::UpToDate),
entry(Some("zulu-bucket"), "early", UpstreamState::Ahead),
],
false,
);
let rendered = format!("{output}");
assert_eq!(rendered.matches("acme-research").count(), 1, "one span");
assert_eq!(rendered.matches("zulu-bucket").count(), 1, "one span");
let positions = |name: &str| rendered.find(name).expect("{name} is listed");
assert!(positions("acme-research") < positions("zulu-bucket"));
assert!(positions("example/one") < positions("example/two"));
assert!(positions("example/early") < positions("example/late"));
assert!(positions("zulu-bucket") < positions("example/scratch"));
}
#[test]
fn test_display_spans_repeated_buckets() {
let output = Output::new(
vec![
entry(Some("acme-research"), "one", UpstreamState::UpToDate),
entry(Some("acme-research"), "two", UpstreamState::UpToDate),
entry(Some("zulu-bucket"), "late", UpstreamState::Behind),
entry(None, "scratch", UpstreamState::Local),
],
false,
);
let rendered = format!("{output}");
assert_eq!(rendered.matches("acme-research").count(), 1, "named once");
assert_eq!(rendered.matches("zulu-bucket").count(), 1);
assert_eq!(rendered.matches(NO_BUCKET).count(), 1);
for name in [
"example/one",
"example/two",
"example/late",
"example/scratch",
] {
assert!(rendered.contains(name), "{name} is listed");
}
assert_eq!(rendered.matches("last synced").count(), 1);
assert_eq!(rendered.matches("| synced").count(), 2);
}
#[test]
fn test_display_words_each_state_as_of_last_sync() {
let cases = [
(Some("acme-research"), UpstreamState::UpToDate, "synced"),
(
Some("acme-research"),
UpstreamState::Ahead,
"unpushed commit",
),
(Some("acme-research"), UpstreamState::Behind, "behind"),
(Some("acme-research"), UpstreamState::Diverged, "diverged"),
(Some("acme-research"), UpstreamState::Local, "never pushed"),
(None, UpstreamState::Local, "local only"),
(Some("acme-research"), UpstreamState::Error, "unknown"),
];
for (bucket, status, expected) in cases {
assert_eq!(last_synced(&entry(bucket, "one", status)), expected);
}
}
#[test]
fn test_display_ends_with_last_synced_hint() {
let output = Output::new(
vec![entry(Some("acme-research"), "one", UpstreamState::UpToDate)],
false,
);
assert!(format!("{output}").ends_with(LAST_SYNCED_HINT));
let empty = Output::new(Vec::new(), false);
assert!(!format!("{empty}").contains(LAST_SYNCED_HINT));
}
fn fetched(state: PackageState, changed_files: Option<usize>) -> Fetched {
Fetched {
state,
changed_files,
remote_checked: changed_files.is_some(),
..Fetched::failed()
}
}
fn unreachable(state: PackageState, changed_files: usize) -> Fetched {
Fetched {
remote_checked: false,
..fetched(state, Some(changed_files))
}
}
fn denied() -> Fetched {
Fetched {
access_denied: true,
..Fetched::failed()
}
}
#[test]
fn test_display_words_each_fetched_state_as_quilt_sync_does() {
let cases = [
(fetched(PackageState::Latest, Some(0)), "latest"),
(fetched(PackageState::Latest, None), "latest"),
(
fetched(PackageState::PendingCommit, Some(0)),
"revision not published",
),
(
fetched(PackageState::PendingChanges { files: 1 }, Some(1)),
"1 file changed",
),
(
fetched(PackageState::PendingChanges { files: 3 }, Some(3)),
"3 files changed",
),
(fetched(PackageState::Behind, Some(2)), "not the latest"),
(
fetched(PackageState::Diverged, Some(2)),
"changed in both places",
),
(
fetched(PackageState::Unpublished, Some(2)),
"not published yet",
),
(fetched(PackageState::NoRemote, None), "no S3 bucket yet"),
(fetched(PackageState::Unknown, None), "unknown"),
(denied(), "no access"),
];
for (fetched, expected) in cases {
let entry = fetched_entry(Some("acme-research"), "one", UpstreamState::Error, fetched);
assert_eq!(state(&entry), expected, "{:?}", entry.fetched);
}
}
#[test]
fn test_display_marks_rows_whose_remote_was_unreachable() {
let cases = [
(
unreachable(PackageState::Latest, 0),
"latest (remote unreachable)",
),
(
unreachable(PackageState::PendingChanges { files: 1 }, 1),
"1 file changed (remote unreachable)",
),
(
unreachable(PackageState::Behind, 0),
"not the latest (remote unreachable)",
),
(fetched(PackageState::NoRemote, None), "no S3 bucket yet"),
(fetched(PackageState::Unknown, None), "unknown"),
(denied(), "no access"),
];
for (fetched, expected) in cases {
let entry = fetched_entry(Some("acme-research"), "one", UpstreamState::Error, fetched);
assert_eq!(state(&entry), expected, "{:?}", entry.fetched);
}
}
#[test]
fn test_display_names_the_fetched_column_and_hint() {
let output = Output::new(
vec![fetched_entry(
Some("acme-research"),
"one",
UpstreamState::UpToDate,
fetched(PackageState::Latest, Some(0)),
)],
true,
);
let rendered = format!("{output}");
assert!(rendered.ends_with(FETCHED_HINT));
assert!(!rendered.contains(LAST_SYNCED_HINT));
assert!(!rendered.contains("last synced"));
assert!(rendered.contains("| status"));
assert!(rendered.contains("| latest"));
let empty = Output::new(Vec::new(), true);
assert_eq!(format!("{empty}"), "No installed packages");
}
#[test]
fn test_json_adds_fetched_fields() {
let output = Output::new(
vec![
fetched_entry(
Some("acme-research"),
"one",
UpstreamState::UpToDate,
fetched(PackageState::PendingChanges { files: 2 }, Some(2)),
),
fetched_entry(Some("acme-research"), "two", UpstreamState::Error, denied()),
fetched_entry(
Some("acme-research"),
"three",
UpstreamState::UpToDate,
unreachable(PackageState::PendingChanges { files: 1 }, 1),
),
fetched_entry(
None,
"scratch",
UpstreamState::Local,
fetched(PackageState::NoRemote, None),
),
],
true,
);
assert_eq!(
output.to_json().to_string(),
r#"{"packages":[{"bucket":"acme-research","namespace":"example/one","status":"up_to_date","fetched":true,"changed_files":2},{"bucket":"acme-research","namespace":"example/three","status":"up_to_date","fetched":false,"changed_files":1},{"bucket":"acme-research","namespace":"example/two","status":"error","fetched":false,"access_denied":true},{"bucket":null,"namespace":"example/scratch","status":"local","fetched":false}]}"#
);
}
fn s3_error(kind: quilt_rs::S3ErrorKind) -> quilt_rs::Error {
quilt_rs::Error::S3(quilt_rs::S3Error::new(kind))
}
#[test(tokio::test)]
async fn test_fetch_entry_takes_the_checks_answer() {
let entry = fetch_entry(
("example", "one").into(),
Some("acme-research".to_string()),
|| async {
Ok(PackageStateReport {
state: PackageState::Behind,
upstream_state: UpstreamState::Behind,
changed_files: Some(2),
latest_refreshed: true,
})
},
)
.await;
assert_eq!(entry.bucket.as_deref(), Some("acme-research"));
assert_eq!(entry.status, UpstreamState::Behind);
assert_eq!(entry.fetched, Some(fetched(PackageState::Behind, Some(2))));
assert_eq!(state(&entry), "not the latest");
let stale = fetch_entry(
("example", "one").into(),
Some("acme-research".to_string()),
|| async {
Ok(PackageStateReport {
state: PackageState::Latest,
upstream_state: UpstreamState::UpToDate,
changed_files: Some(0),
latest_refreshed: false,
})
},
)
.await;
assert_eq!(stale.fetched, Some(unreachable(PackageState::Latest, 0)));
assert_eq!(state(&stale), "latest (remote unreachable)");
}
#[test(tokio::test)]
async fn test_fetch_entry_falls_back_per_row_on_error() {
let denied_row = fetch_entry(
("example", "one").into(),
Some("acme-research".to_string()),
|| async {
Err(s3_error(quilt_rs::S3ErrorKind::AccessDenied(
"denied".to_string(),
)))
},
)
.await;
assert_eq!(denied_row.status, UpstreamState::Error);
assert_eq!(denied_row.fetched, Some(denied()));
assert_eq!(state(&denied_row), "no access");
for err in [
s3_error(quilt_rs::S3ErrorKind::GetObject("boom".to_string())),
quilt_rs::Error::Login(quilt_rs::LoginError::NoSession(None)),
] {
let failed = fetch_entry(
("example", "one").into(),
Some("acme-research".to_string()),
|| async { Err(err) },
)
.await;
assert_eq!(failed.status, UpstreamState::Error);
assert_eq!(failed.fetched, Some(Fetched::failed()));
assert_eq!(state(&failed), "unknown");
}
}
#[test(tokio::test)]
async fn test_model_fetch_with_local_package() -> Result<(), Error> {
let (m, _temp_dir) = create_model_in_temp_dir().await?;
m.create(crate::cli::create::Input {
namespace: ("example", "local").into(),
source: None,
message: None,
})
.await?;
let output = m.list(Input { fetch: true }).await?;
assert_eq!(output.installed_packages_list.len(), 1);
let rendered = format!("{output}");
assert!(rendered.contains("| no S3 bucket yet"));
assert!(rendered.ends_with(FETCHED_HINT));
Ok(())
}
#[test(tokio::test)]
async fn test_model_with_local_package() -> Result<(), Error> {
let (m, _temp_dir) = create_model_in_temp_dir().await?;
m.create(crate::cli::create::Input {
namespace: ("example", "local").into(),
source: None,
message: None,
})
.await?;
let output = model(m.get_local_domain(), Input::default()).await?;
let entry = &output.installed_packages_list[0];
assert_eq!(entry.bucket, None);
assert_eq!(entry.namespace, ("example", "local").into());
assert_eq!(entry.status, UpstreamState::Local);
let output = format!("{output}");
assert!(output.contains("example/local"));
assert!(output.contains(NO_BUCKET));
Ok(())
}
#[test(tokio::test)]
async fn live_model() -> Result<(), Error> {
let uri = format!("{}&path={}", pkg::URI_LATEST, pkg::README_LK_ESCAPED);
let (m, _, _temp_dir) = install_package_into_temp_dir(&uri).await?;
{
let local_domain = m.get_local_domain();
let output = model(local_domain, Input::default()).await?;
let entry = &output.installed_packages_list[0];
assert_eq!(entry.bucket.as_deref(), Some(pkg::BUCKET));
assert_eq!(entry.namespace, pkg::NAMESPACE.into());
assert_eq!(entry.status, UpstreamState::UpToDate);
let output = format!("{output}");
assert!(output.contains("bucket"));
assert!(output.contains("namespace"));
assert!(output.contains("last synced"));
assert!(output.contains(pkg::BUCKET));
assert!(output.contains("| synced"));
}
Ok(())
}
#[test(tokio::test)]
async fn live_command_with_package() -> Result<(), Error> {
let uri = format!("{}&path={}", pkg::URI_LATEST, pkg::README_LK_ESCAPED);
let (m, _, _temp_dir) = install_package_into_temp_dir(&uri).await?;
if let Std::Out(output) = command(m, Input::default()).await {
let output = output.to_string();
assert!(output.contains(pkg::BUCKET));
assert!(output.contains(pkg::NAMESPACE_STR));
assert!(output.contains("| synced"));
} else {
return Err(Error::Test("Failed to list packages".to_string()));
}
Ok(())
}
}