1use super::diff::{ItemOp, ListOp};
2use super::reports::{DanglingPointer, Mirror, PushReport, StackIdentity};
3use crate::diff::{Change, Drift};
4use crate::exceptions;
5use crate::model::{ListKey, Rule};
6use crate::normalize;
7use crate::report::{ChangeReport, ReportEntry};
8use crate::rules as api;
9use elasticctl_core::{Error, ErrorKind, Result, Transport};
10use serde_json::Value;
11use std::collections::{BTreeMap, BTreeSet};
12use std::path::Path;
13
14mod references;
15mod summary;
16use references::inject_list_ids;
17use summary::{ExceptionCounts, push_summary, update_value_summary};
18
19#[derive(Debug, Clone)]
20pub struct PushPlan {
21 pub preview_action: String,
22 pub preview_details: Vec<String>,
23 pub report: ChangeReport,
24 pub summary: PushReport,
25 desired: BTreeMap<String, Rule>,
26 list_ops: Vec<ListOp>,
27 item_ops: Vec<ItemOp>,
28 pub value_lists: Vec<crate::value_lists::ListPlan>,
29 pub value_list_outcomes: Vec<crate::value_lists::ListOutcome>,
30 pub value_list_drift: Option<super::values::ValueListDrift>,
31 values: super::values::ValuePlan,
32}
33
34pub async fn plan_push(
35 t: &Transport,
36 dir: &Path,
37 selectors: &[String],
38 tag: Option<&str>,
39 search: Option<&str>,
40 source: crate::rules::RuleSource,
41 identity: &StackIdentity,
42) -> Result<PushPlan> {
43 let Mirror {
44 rules: local_all,
45 lists,
46 items,
47 value_lists,
48 has_value_lists,
49 } = super::mirror::read_mirror(dir)?;
50 let scope = super::scope_of(t, selectors, tag, search, source, &local_all, "apply").await?;
51 let (local, out_of_scope) = if scope.is_scoped() {
52 (scope.narrow(local_all), 0)
53 } else {
54 scope.split_by_source(local_all)
55 };
56
57 let remote = scope.remote(t).await?;
58 let drift = Drift::compute(&local, &remote)?;
59
60 let plan =
61 super::diff::exception_plan(t, lists.clone(), items.clone(), &local, &remote).await?;
62 let values = super::values::plan(
63 t,
64 super::values::Inputs {
65 pairs: &value_lists,
66 has_root: has_value_lists,
67 lists: &lists,
68 items: &items,
69 local: &local,
70 remote: &remote,
71 exception_plan: &plan,
72 },
73 true,
74 )
75 .await?;
76 let exceptions = plan.drift;
77 let list_ops = plan.list_ops;
78 let item_ops = plan.item_ops;
79 let resolvable = plan.resolvable;
80
81 let by_id = |id: &str| local.iter().find(|r| r.rule_id().ok() == Some(id)).cloned();
82 let remote_by_id = |id: &str| {
83 remote
84 .iter()
85 .find(|r| r.rule_id().ok() == Some(id))
86 .cloned()
87 };
88
89 let actionable = drift.actionable();
90 let actionable_ids: BTreeSet<String> =
91 actionable.iter().map(|c| c.rule_id().to_string()).collect();
92
93 let mut repairs: Vec<DanglingPointer> = Vec::new();
94 let mut repaired_ids: BTreeSet<String> = BTreeSet::new();
95 for dangling in &exceptions.dangling {
96 if actionable_ids.contains(&dangling.rule_id)
97 || by_id(&dangling.rule_id).is_none()
98 || !repaired_ids.insert(dangling.rule_id.clone())
99 {
100 continue;
101 }
102 repairs.push(dangling.clone());
103 }
104
105 let mut preview_details = Vec::new();
106
107 for op in &list_ops {
108 match op {
109 ListOp::Create(list) => {
110 preview_details.push(format!("{} {} create", list.list_id()?, list.name()))
111 }
112 ListOp::Update { after, .. } => {
113 preview_details.push(format!("{} {} update", after.list_id()?, after.name()))
114 }
115 }
116 }
117 for op in &item_ops {
118 match op {
119 ItemOp::Create(item) => {
120 preview_details.push(format!("{} {} create", item.item_id()?, item.list_id()?))
121 }
122 ItemOp::Update { after, .. } => preview_details.push(format!(
123 "{} {} update",
124 after.item_id()?,
125 after.list_id()?
126 )),
127 ItemOp::Remove { before, .. } => preview_details.push(format!(
128 "{} {} delete",
129 before.item_id()?,
130 before.list_id()?
131 )),
132 }
133 }
134 for change in &actionable {
135 let line = match change {
136 Change::Added { rule_id, name } => format!("{rule_id} {name} create"),
137 Change::Modified {
138 rule_id,
139 name,
140 fields,
141 } => {
142 let names: Vec<&str> = fields.iter().map(|f| f.field.as_str()).collect();
143 format!("{rule_id} {name} update ({})", names.join(", "))
144 }
145 _ => String::new(),
146 };
147 if !line.is_empty() {
148 preview_details.push(line);
149 }
150 }
151 for dangling in &repairs {
152 let name = by_id(&dangling.rule_id)
153 .map(|r| r.name().to_string())
154 .unwrap_or_default();
155 preview_details.push(format!("{} {} update (pointer)", dangling.rule_id, name));
156 }
157
158 if let Some(drift) = &values.drift {
159 for id in &drift.unreferenced {
160 preview_details.push(format!("value list {id} is unreferenced; left unchanged"));
161 }
162 for warning in &drift.warnings {
163 preview_details.push(format!("value list {}: {}", warning.id, warning.message));
164 }
165 }
166
167 let mut entries: Vec<ReportEntry> = Vec::new();
168 let mut desired: BTreeMap<String, Rule> = BTreeMap::new();
169
170 for change in &drift.changes {
171 if let Change::RemoteOnly { rule_id, name } = change {
172 entries.push(ReportEntry {
173 rule_id: rule_id.clone(),
174 name: name.clone(),
175 action: "skipped_remote_only".into(),
176 value_list: None,
177 before: remote_by_id(rule_id).map(|r| normalize::canonical(&r).into_value()),
178 after: None,
179 applied: false,
180 error: None,
181 });
182 }
183 }
184
185 for change in &actionable {
186 let (rule_id, name, action) = match change {
187 Change::Added { rule_id, name } => (rule_id.clone(), name.clone(), "create"),
188 Change::Modified { rule_id, name, .. } => (rule_id.clone(), name.clone(), "update"),
189 _ => continue,
190 };
191
192 let Some(desired_rule) = by_id(&rule_id) else {
193 continue;
194 };
195 let before = remote_by_id(&rule_id).map(|r| normalize::canonical(&r).into_value());
196
197 desired.insert(rule_id.clone(), desired_rule.clone());
198
199 entries.push(ReportEntry {
200 rule_id,
201 name,
202 action: action.into(),
203 value_list: None,
204 before,
205 after: Some(normalize::canonical(&desired_rule).into_value()),
206 applied: false,
207 error: None,
208 });
209 }
210
211 for dangling in &repairs {
212 let desired_rule = by_id(&dangling.rule_id).expect("repair rule was found above");
213 let before = remote_by_id(&dangling.rule_id).map(|r| normalize::canonical(&r).into_value());
214 desired.insert(dangling.rule_id.clone(), desired_rule.clone());
215 entries.push(ReportEntry {
216 rule_id: dangling.rule_id.clone(),
217 name: desired_rule.name().to_string(),
218 action: "update".into(),
219 value_list: None,
220 before,
221 after: Some(normalize::canonical(&desired_rule).into_value()),
222 applied: false,
223 error: None,
224 });
225 }
226
227 let desired_rules: Vec<Rule> = desired.values().cloned().collect();
228 let unresolved: Vec<String> = super::referenced_keys(&desired_rules)
229 .into_iter()
230 .filter(|key| !resolvable.contains(key))
231 .map(|key| format!("\"{}\" ({})", key.list_id, key.namespace_type))
232 .collect();
233 if !unresolved.is_empty() {
234 return Err(Error::new(
235 ErrorKind::NotFound,
236 format!(
237 "rule(s) reference exception list(s) that do not exist on this stack and are \
238 not in the mirror: {}",
239 unresolved.join(", ")
240 ),
241 ));
242 }
243
244 let item_removals = item_ops
245 .iter()
246 .filter(|op| matches!(op, ItemOp::Remove { .. }))
247 .count();
248 let item_writes = item_ops.len() - item_removals;
249 let mut preview_action = format!(
250 "Push {} rule change(s), {} exception list(s) and {} item(s)",
251 actionable.len() + repairs.len(),
252 list_ops.len(),
253 item_writes,
254 );
255 if item_removals > 0 {
256 preview_action.push_str(&format!(", {} item deletion(s)", item_removals));
257 }
258 preview_action.push_str(&format!(" from {}{}", dir.display(), scope.describe()));
259
260 let report = ChangeReport {
261 profile: identity.profile.clone(),
262 host: identity.host.clone(),
263 space: identity.space.clone(),
264 applied: false,
265 entries,
266 };
267 let summary = push_summary(
268 &report,
269 scope.is_scoped().then(|| scope.selected()),
270 scope.is_scoped().then_some(scope.local_total),
271 out_of_scope,
272 ExceptionCounts::default(),
273 );
274
275 Ok(PushPlan {
276 preview_action,
277 preview_details,
278 report,
279 summary,
280 desired,
281 list_ops,
282 item_ops,
283 value_lists: values.plans.clone(),
284 value_list_outcomes: Vec::new(),
285 value_list_drift: values.drift.clone(),
286 values,
287 })
288}
289
290pub async fn apply_push(t: &Transport, plan: PushPlan) -> Result<PushPlan> {
291 apply_push_with_clock(t, plan, &crate::value_lists::SystemClock::default()).await
292}
293
294async fn apply_push_with_clock<C: crate::value_lists::Clock>(
295 t: &Transport,
296 mut plan: PushPlan,
297 clock: &C,
298) -> Result<PushPlan> {
299 let desired_rules: Vec<Rule> = plan.desired.values().cloned().collect();
300 let wanted: Vec<ListKey> = super::referenced_keys(&desired_rules).into_iter().collect();
301 let mut resolved = exceptions::resolve_ids(t, &wanted).await?;
302
303 plan.value_list_outcomes = crate::value_lists::apply_plans(t, &plan.values.plans, clock).await;
304 let value_entries: Vec<_> = plan
305 .value_list_outcomes
306 .iter()
307 .flat_map(|outcome| {
308 outcome.evidence.iter().cloned().map(move |mut evidence| {
309 evidence.error = outcome.error.clone();
310 evidence
311 })
312 })
313 .map(ReportEntry::from)
314 .collect();
315 if plan
316 .value_list_outcomes
317 .iter()
318 .any(|o| o.outcome != crate::value_lists::Outcome::Applied)
319 {
320 plan.report.entries.splice(0..0, value_entries);
321 plan.report.applied = true;
322 update_value_summary(&mut plan);
323 return Ok(plan);
324 }
325 let mut counts = ExceptionCounts::default();
326 let mut exception_entries = value_entries;
327
328 for op in &plan.list_ops {
329 let failure = match op {
330 ListOp::Create(list) => match exceptions::create_list(t, list).await {
331 Ok(created) => {
332 if let Some(id) = created.as_map().get("id").and_then(Value::as_str) {
333 resolved.insert(list.key()?, id.to_string());
334 }
335 counts.lists_created += 1;
336 exception_entries.push(ReportEntry {
337 rule_id: list.list_id().unwrap_or("<unreadable>").to_string(),
338 name: list.name().to_string(),
339 action: "create_list".into(),
340 value_list: None,
341 before: None,
342 after: Some(normalize::canonical_list(&created).into_value()),
343 applied: true,
344 error: None,
345 });
346 None
347 }
348 Err(e) => Some(ReportEntry {
349 rule_id: list.list_id().unwrap_or("<unreadable>").to_string(),
350 name: list.name().to_string(),
351 action: "create_list".into(),
352 value_list: None,
353 before: None,
354 after: None,
355 applied: false,
356 error: Some(e.message),
357 }),
358 },
359 ListOp::Update { before, after } => match exceptions::update_list(t, after).await {
360 Ok(applied) => {
361 counts.lists_updated += 1;
362 exception_entries.push(ReportEntry {
363 rule_id: after.list_id().unwrap_or("<unreadable>").to_string(),
364 name: after.name().to_string(),
365 action: "update_list".into(),
366 value_list: None,
367 before: Some(before.clone().into_value()),
368 after: Some(normalize::canonical_list(&applied).into_value()),
369 applied: true,
370 error: None,
371 });
372 None
373 }
374 Err(e) => Some(ReportEntry {
375 rule_id: after.list_id().unwrap_or("<unreadable>").to_string(),
376 name: after.name().to_string(),
377 action: "update_list".into(),
378 value_list: None,
379 before: Some(before.clone().into_value()),
380 after: None,
381 applied: false,
382 error: Some(e.message),
383 }),
384 },
385 };
386 if let Some(failed_entry) = failure {
387 return Ok(finish_after_exception_failure(
388 plan,
389 exception_entries,
390 failed_entry,
391 counts,
392 ));
393 }
394 }
395
396 if let Err(error) = super::values::reread(t, &plan.values).await {
397 let failed_entry = ReportEntry {
398 rule_id: String::new(),
399 name: String::new(),
400 action: "check_value_lists".into(),
401 before: None,
402 after: None,
403 applied: false,
404 error: Some(error.message),
405 value_list: None,
406 };
407 return Ok(finish_after_exception_failure(
408 plan,
409 exception_entries,
410 failed_entry,
411 counts,
412 ));
413 }
414
415 for op in &plan.item_ops {
416 let failure = match op {
417 ItemOp::Create(item) => match exceptions::create_item(t, item).await {
418 Ok(applied) => {
419 counts.items_created += 1;
420 exception_entries.push(ReportEntry {
421 rule_id: item.item_id().unwrap_or("<unreadable>").to_string(),
422 name: item.list_id().unwrap_or("<unreadable>").to_string(),
423 action: "create_item".into(),
424 value_list: None,
425 before: None,
426 after: Some(normalize::canonical_item(&applied).into_value()),
427 applied: true,
428 error: None,
429 });
430 None
431 }
432 Err(e) => Some(ReportEntry {
433 rule_id: item.item_id().unwrap_or("<unreadable>").to_string(),
434 name: item.list_id().unwrap_or("<unreadable>").to_string(),
435 action: "create_item".into(),
436 value_list: None,
437 before: None,
438 after: None,
439 applied: false,
440 error: Some(e.message),
441 }),
442 },
443 ItemOp::Update { before, after } => match exceptions::update_item(t, after).await {
444 Ok(applied) => {
445 counts.items_updated += 1;
446 exception_entries.push(ReportEntry {
447 rule_id: after.item_id().unwrap_or("<unreadable>").to_string(),
448 name: after.list_id().unwrap_or("<unreadable>").to_string(),
449 action: "update_item".into(),
450 value_list: None,
451 before: Some(before.clone().into_value()),
452 after: Some(normalize::canonical_item(&applied).into_value()),
453 applied: true,
454 error: None,
455 });
456 None
457 }
458 Err(e) => Some(ReportEntry {
459 rule_id: after.item_id().unwrap_or("<unreadable>").to_string(),
460 name: after.list_id().unwrap_or("<unreadable>").to_string(),
461 action: "update_item".into(),
462 value_list: None,
463 before: Some(before.clone().into_value()),
464 after: None,
465 applied: false,
466 error: Some(e.message),
467 }),
468 },
469 ItemOp::Remove {
470 before,
471 namespace_type,
472 } => match exceptions::delete_item(
473 t,
474 before.item_id().unwrap_or("<unreadable>"),
475 namespace_type,
476 )
477 .await
478 {
479 Ok(_) => {
480 counts.items_removed += 1;
481 exception_entries.push(ReportEntry {
482 rule_id: before.item_id().unwrap_or("<unreadable>").to_string(),
483 name: before.list_id().unwrap_or("<unreadable>").to_string(),
484 action: "delete_item".into(),
485 value_list: None,
486 before: Some(before.clone().into_value()),
487 after: None,
488 applied: true,
489 error: None,
490 });
491 None
492 }
493 Err(e) => Some(ReportEntry {
494 rule_id: before.item_id().unwrap_or("<unreadable>").to_string(),
495 name: before.list_id().unwrap_or("<unreadable>").to_string(),
496 action: "delete_item".into(),
497 value_list: None,
498 before: Some(before.clone().into_value()),
499 after: None,
500 applied: false,
501 error: Some(e.message),
502 }),
503 },
504 };
505 if let Some(failed_entry) = failure {
506 return Ok(finish_after_exception_failure(
507 plan,
508 exception_entries,
509 failed_entry,
510 counts,
511 ));
512 }
513 }
514
515 let mut entries = exception_entries;
516 entries.reserve(plan.report.entries.len());
517 for entry in plan.report.entries {
518 if entry.action != "create" && entry.action != "update" {
519 entries.push(entry);
520 continue;
521 }
522
523 let Some(desired) = plan.desired.get(&entry.rule_id) else {
524 let missing = entry.rule_id.clone();
525 entries.push(ReportEntry {
526 rule_id: entry.rule_id,
527 name: entry.name,
528 action: entry.action,
529 value_list: None,
530 before: entry.before,
531 after: None,
532 applied: false,
533 error: Some(format!("the plan has no desired rule for \"{missing}\"")),
534 });
535 continue;
536 };
537
538 let mut to_write = desired.clone();
539 if let Err(e) = inject_list_ids(&mut to_write, &resolved) {
540 entries.push(ReportEntry {
541 rule_id: entry.rule_id,
542 name: entry.name,
543 action: entry.action,
544 value_list: None,
545 before: entry.before,
546 after: None,
547 applied: false,
548 error: Some(e.message),
549 });
550 continue;
551 }
552 let before = entry.before;
553 let is_create = entry.action == "create";
554
555 let outcome = if is_create {
556 api::create(t, &to_write).await
557 } else {
558 api::update(t, &to_write).await
559 };
560
561 match outcome {
562 Ok(applied) => entries.push(ReportEntry {
563 rule_id: entry.rule_id,
564 name: entry.name,
565 action: entry.action,
566 value_list: None,
567 before,
568 after: Some(normalize::canonical(&applied).into_value()),
569 applied: true,
570 error: None,
571 }),
572 Err(e) => entries.push(ReportEntry {
573 rule_id: entry.rule_id,
574 name: entry.name,
575 action: entry.action,
576 value_list: None,
577 before,
578 after: None,
579 applied: false,
580 error: Some(e.message),
581 }),
582 }
583 }
584
585 let (selected, local_total, out_of_scope) = (
586 plan.summary.selected,
587 plan.summary.local_total,
588 plan.summary.out_of_scope,
589 );
590 plan.report.entries = entries;
591 plan.report.applied = true;
592 plan.summary = push_summary(&plan.report, selected, local_total, out_of_scope, counts);
593 update_value_summary(&mut plan);
594 Ok(plan)
595}
596
597fn finish_after_exception_failure(
598 mut plan: PushPlan,
599 mut exception_entries: Vec<ReportEntry>,
600 failed_entry: ReportEntry,
601 counts: ExceptionCounts,
602) -> PushPlan {
603 exception_entries.push(failed_entry);
604 exception_entries.append(&mut plan.report.entries);
605 plan.report.entries = exception_entries;
606 plan.report.applied = true;
607 let (selected, local_total, out_of_scope) = (
608 plan.summary.selected,
609 plan.summary.local_total,
610 plan.summary.out_of_scope,
611 );
612 plan.summary = push_summary(&plan.report, selected, local_total, out_of_scope, counts);
613 update_value_summary(&mut plan);
614 plan
615}
616
617#[cfg(test)]
618mod tests;