use std::collections::BTreeSet;
use anyhow::{Result, anyhow};
use colored::Colorize;
use serde_json::Value;
use rigg_core::normalize::normalize_for_push;
use rigg_core::registry;
use rigg_core::resources::{ResourceKind, ResourceRef};
use rigg_core::store::{ProjectState, Store, SyncClass, assert_exclusive_ownership};
use rigg_core::workspace::{Project, ResolvedEnv, Workspace};
use crate::cli::PullArgs;
use crate::commands::remote::{Remote, ensure_any_connection};
use crate::commands::{
CommandError, GlobalContext, interactive, load_workspace, resolve_env, select_projects,
};
pub async fn run(ctx: &GlobalContext, args: PullArgs) -> Result<()> {
if args.watch {
loop {
pull_once(ctx, &args).await?;
tokio::time::sleep(std::time::Duration::from_secs(args.interval)).await;
}
}
pull_once(ctx, &args).await
}
async fn pull_once(ctx: &GlobalContext, args: &PullArgs) -> Result<()> {
let ws = load_workspace()?;
let env = resolve_env(&ws, ctx)?;
assert_exclusive_ownership(&ws, &env.name)?;
let projects = select_projects(&ws, args.project.as_deref(), args.all)?;
let mut owned_by_any: BTreeSet<String> = BTreeSet::new();
for project in &ws.projects {
for (r, _) in Store::new(project, &env.name).list()? {
owned_by_any.insert(r.key());
}
let state = ProjectState::load(&ws, &env.name, &project.name);
owned_by_any.extend(state.baselines.keys().cloned());
}
let mut any_conflict = false;
for project in projects {
any_conflict |= pull_project(ctx, &ws, &env, project, &owned_by_any).await?;
}
if any_conflict {
return Err(anyhow!(CommandError::DriftOrConflict(
"conflicts detected during pull; resolve interactively or with --yes".to_string()
)));
}
Ok(())
}
async fn pull_project(
ctx: &GlobalContext,
ws: &Workspace,
env: &ResolvedEnv,
project: &Project,
owned_by_any: &BTreeSet<String>,
) -> Result<bool> {
let store = Store::new(project, &env.name);
let remote = Remote::for_project(env, project);
ensure_any_connection(&remote, project)?;
let mut state = ProjectState::load(ws, &env.name, &project.name);
println!(
"{} project '{}' (env: {})",
"Pull".bold(),
project.name.bold(),
env.name
);
remote.print_targets();
let snapshot = remote.snapshot().await?;
let auto_created = registry::auto_created_by(&snapshot);
let remote_keys: BTreeSet<String> = snapshot.iter().map(|(r, _)| r.key()).collect();
let local_files = store.list()?;
let mut owned_here: BTreeSet<String> = local_files.iter().map(|(r, _)| r.key()).collect();
owned_here.extend(state.baselines.keys().cloned());
let mut any_conflict = false;
let mut unmanaged = 0usize;
let mut written = 0usize;
for (r, doc) in &snapshot {
let key = r.key();
let owned_by_this = owned_here.contains(&key);
let owned_elsewhere = !owned_by_this && owned_by_any.contains(&key);
if owned_elsewhere {
continue; }
if !owned_by_this {
if !registry::is_platform_managed(r.kind, doc) && !auto_created.contains_key(&key) {
unmanaged += 1;
}
continue;
}
let local = store.read(r).ok();
match state.classify(r, local.as_ref(), Some(doc)) {
SyncClass::InSync => {
state.set_baseline(r, doc);
}
SyncClass::RemoteAhead | SyncClass::RemoteOnly => {
if store.write(r, doc)? {
println!(" {} updated {}", "~".cyan(), r);
written += 1;
}
state.set_baseline(r, doc);
}
SyncClass::LocalAhead => {
println!(" {} {} (local ahead — push pending)", "≠".yellow(), r);
}
SyncClass::Untracked | SyncClass::Conflict => {
let summary = local
.as_ref()
.map(|l| conflict_summary(r.kind, l, doc))
.unwrap_or_else(|| "differs locally and remotely".to_string());
if ctx.yes {
store.write(r, doc)?;
state.set_baseline(r, doc);
println!(" {} overwrote {}", "~".cyan(), r);
written += 1;
} else if ctx.interactive() {
println!(" {} {} — {}", "conflict".red().bold(), r, summary);
const OVERWRITE: &str = "overwrite local with remote";
const KEEP: &str = "keep local";
const DIFF: &str = "show diff";
const ABORT: &str = "abort pull";
let mut show_diff_option = true;
loop {
let mut opts = vec![OVERWRITE.to_string(), KEEP.to_string()];
if show_diff_option {
opts.push(DIFF.to_string());
}
opts.push(ABORT.to_string());
let choice = match interactive::select("Resolve:", opts, ctx.no_color) {
Ok(c) => c,
Err(_) => ABORT.to_string(),
};
match choice.as_str() {
OVERWRITE => {
store.write(r, doc)?;
state.set_baseline(r, doc);
println!(" {} overwrote {}", "~".cyan(), r);
written += 1;
break;
}
KEEP => {
println!(" kept local {r}");
break;
}
DIFF => {
if let Some(l) = &local {
let result = rigg_diff::semantic::diff(
&normalize_for_push(r.kind, doc),
&normalize_for_push(r.kind, l),
"name",
);
let labels = rigg_diff::output::SideLabels {
new_side: "local".to_string(),
old_side: format!("Azure ({})", env.name),
};
println!();
print!(
"{}",
rigg_diff::output::format_text(
&result,
&r.to_string(),
&labels
)
);
println!();
}
show_diff_option = false;
}
_ => {
state.save(ws, &env.name, &project.name)?;
return Err(anyhow!("aborted"));
}
}
}
} else {
println!(
" {} {} — {} (run `rigg diff {}` to inspect; pass --yes to overwrite)",
"conflict".red().bold(),
r,
summary,
project.name
);
any_conflict = true;
}
}
SyncClass::LocalOnly => unreachable!("remote doc was provided"),
}
}
let deleted: Vec<ResourceRef> = local_files
.iter()
.map(|(r, _)| r.clone())
.filter(|r| {
state.has_baseline(r)
&& remote.supported_kinds().contains(&r.kind)
&& !remote_keys.contains(&r.key())
})
.collect();
for r in deleted {
if ctx.interactive() {
println!(" {} {} was deleted remotely", "!".red().bold(), r);
if interactive::confirm_default_no(
"Delete the local file too? (no = keep for re-push)",
ctx.no_color,
)? {
store.delete(&r)?;
state.clear_baseline(&r);
println!(" {} removed local {}", "✓".green(), r);
} else {
state.clear_baseline(&r);
println!(" kept {r} (push will re-create it in Azure)");
}
} else {
println!(
" {} {} deleted remotely (kept locally; delete the file or push to re-create)",
"!".red().bold(),
r
);
state.clear_baseline(&r);
any_conflict = true;
}
}
state.save(ws, &env.name, &project.name)?;
if unmanaged > 0 {
println!(
" {} {unmanaged} unmanaged remote resource(s) — adopt with `rigg adopt {} <selector>` (e.g. `all`, `indexes`, `agents/name`)",
"i".blue(),
project.name,
);
}
if written == 0 {
println!(" {} up to date", "✓".green());
}
Ok(any_conflict)
}
fn conflict_summary(kind: ResourceKind, local: &Value, remote: &Value) -> String {
let result = rigg_diff::semantic::diff(
&normalize_for_push(kind, remote),
&normalize_for_push(kind, local),
"name",
);
let n = result.changes.len();
let mut paths: Vec<&str> = result.changes.iter().map(|c| c.path.as_str()).collect();
paths.sort_unstable();
let mut fields: Vec<&str> = paths.into_iter().take(3).collect();
if n > 3 {
fields.push("…");
}
format!("{n} field(s) differ ({})", fields.join(", "))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn conflict_summary_counts_and_names_fields() {
let local = serde_json::json!({"name": "a", "model": "x", "p": 1, "q": 2, "r": 3});
let remote = serde_json::json!({"name": "a", "model": "y", "p": 9, "q": 8, "r": 7});
let s = conflict_summary(ResourceKind::Agent, &local, &remote);
assert!(s.starts_with("4 field(s) differ ("), "{s}");
assert!(s.contains("model"), "{s}");
assert!(
s.ends_with(", …)") || s.matches(',').count() >= 2,
"at most 3 named: {s}"
);
}
}