1use super::common::*;
2use super::import_shared::{
3 IntegrationPolicyImportPlan, IntegrationPolicyImportTarget, import_remote_error,
4 relevant_name_owners,
5};
6use super::import_validation::validate_import_plan;
7use super::normalization::normalize;
8
9#[derive(Debug, Clone, PartialEq, Serialize)]
10pub struct IntegrationPolicyImportReport {
11 pub applied: bool,
12 pub succeeded: Vec<Value>,
13 pub unchanged: Vec<Value>,
14 pub skipped: Vec<Value>,
15 pub failed: Vec<Value>,
16 pub total: usize,
17 pub affected_agents: u64,
18 pub package_installs: Vec<String>,
19}
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22enum ImportAction {
23 Create,
24 Replace,
25 Unchanged,
26}
27
28pub async fn apply_import(
29 transport: &Transport,
30 plan: &IntegrationPolicyImportPlan,
31) -> Result<IntegrationPolicyImportReport> {
32 validate_import_plan(plan)?;
33 if plan.host != transport.kibana_url() || plan.space != transport.space() {
34 return Err(Error::new(
35 ErrorKind::Conflict,
36 "integration import target changed since preview",
37 ));
38 }
39 let mut succeeded = Vec::new();
40 let mut unchanged = Vec::new();
41 let mut failed = Vec::new();
42 let mut expected_groups = plan.package_groups.clone();
43 let mut expected_parents = plan.parent_snapshots.clone();
44 let mut blocked_packages = BTreeMap::<String, String>::new();
45 let mut affected_parents = BTreeMap::<String, u64>::new();
46 let mut observed_installs = BTreeSet::new();
47
48 for target in &plan.targets {
49 let package_name = &target.effective.package.name;
50 if let Some(error) = blocked_packages.get(package_name) {
51 failed.push(import_failed_row(
52 &target.effective.id,
53 false,
54 format!("package dependency is unavailable: {error}"),
55 ));
56 continue;
57 }
58
59 let action = match recheck_import_object(transport, target).await {
60 Ok(action) => action,
61 Err(error) => {
62 failed.push(import_failed_row(
63 &target.effective.id,
64 false,
65 error.message,
66 ));
67 continue;
68 }
69 };
70 if let Err(error) = recheck_import_name_owner(transport, target, &plan.name_owners).await {
71 failed.push(import_failed_row(
72 &target.effective.id,
73 false,
74 error.message,
75 ));
76 continue;
77 }
78 if let Err(error) = recheck_import_parents(transport, target, &expected_parents).await {
79 failed.push(import_failed_row(
80 &target.effective.id,
81 false,
82 error.message,
83 ));
84 continue;
85 }
86
87 let group = expected_groups
88 .get_mut(package_name)
89 .expect("validated target package group");
90 let actual_state = match read_dependencies(transport, &target.effective.package).await {
91 Ok(state) if state == group.state => state,
92 Ok(_) => {
93 let message = "package changed since preview".to_owned();
94 blocked_packages.insert(package_name.clone(), message.clone());
95 failed.push(import_failed_row(&target.effective.id, false, message));
96 continue;
97 }
98 Err(error) => {
99 let message = import_remote_error(error, "apply package read").message;
100 blocked_packages.insert(package_name.clone(), message.clone());
101 failed.push(import_failed_row(&target.effective.id, false, message));
102 continue;
103 }
104 };
105 debug_assert_eq!(actual_state, group.state);
106
107 if action == ImportAction::Unchanged {
108 unchanged.push(json!({"id": target.effective.id}));
109 continue;
110 }
111
112 let (label, applied, route_error) = match action {
113 ImportAction::Create => {
114 match integration_policies::create(transport, &target.effective).await {
115 Ok(_) => ("created", true, None),
116 Err(error) => (
117 "created",
118 false,
119 Some(import_remote_error(error, "create request").message),
120 ),
121 }
122 }
123 ImportAction::Replace => {
124 let _body = target
125 .replacement_body
126 .as_ref()
127 .expect("validated replacement body");
128 match integration_policies::update(
129 transport,
130 &target.effective.id,
131 &target.effective,
132 )
133 .await
134 {
135 Ok(_) => ("replaced", true, None),
136 Err(error) => (
137 "replaced",
138 false,
139 Some(import_remote_error(error, "update request").message),
140 ),
141 }
142 }
143 ImportAction::Unchanged => unreachable!("unchanged rows continue above"),
144 };
145
146 if applied {
147 record_affected_parents(&mut affected_parents, target);
148 advance_parent_snapshots(&mut expected_parents, target);
149 }
150
151 let mut observed_after_create = None;
152 let mut package_observation_error = None;
153 if action == ImportAction::Create
154 && matches!(group.state.state, PackageDependencyState::NotInstalled)
155 {
156 match read_dependencies(transport, &target.effective.package).await {
157 Ok(after) => {
158 if is_exact_installed(&after, &target.effective.package) {
159 group.state = after.clone();
160 observed_installs.insert(format!(
161 "{}@{}",
162 target.effective.package.name, target.effective.package.version
163 ));
164 } else if !matches!(after.state, PackageDependencyState::NotInstalled) {
165 let message =
166 "package installed a different version after create".to_owned();
167 blocked_packages.insert(package_name.clone(), message.clone());
168 package_observation_error = Some(message);
169 }
170 observed_after_create = Some(after);
171 }
172 Err(error) => {
173 let message = import_remote_error(error, "post-create package read").message;
174 blocked_packages.insert(package_name.clone(), message.clone());
175 package_observation_error = Some(message);
176 }
177 }
178 }
179
180 let mut errors = route_error.into_iter().collect::<Vec<_>>();
181 if applied {
182 if let Err(error) = verify_import_stored(transport, &target.effective).await {
183 errors.push(error.message);
184 }
185 let package_result = match observed_after_create.as_ref() {
186 Some(after) => verify_exact_installed(after, &target.effective.package),
187 None => match read_dependencies(transport, &target.effective.package).await {
188 Ok(after) => verify_exact_installed(&after, &target.effective.package),
189 Err(error) => Err(import_remote_error(error, "package verification").message),
190 },
191 };
192 if let Err(message) = package_result {
193 blocked_packages
194 .entry(package_name.clone())
195 .or_insert_with(|| message.clone());
196 errors.push(message);
197 }
198 }
199 if let Some(error) = package_observation_error
200 && !errors.contains(&error)
201 {
202 errors.push(error);
203 }
204
205 if errors.is_empty() {
206 succeeded.push(json!({"id": target.effective.id, "action": label}));
207 } else {
208 failed.push(import_failed_row(
209 &target.effective.id,
210 applied,
211 errors.join("; "),
212 ));
213 }
214 }
215
216 Ok(IntegrationPolicyImportReport {
217 applied: true,
218 succeeded,
219 unchanged,
220 skipped: plan.skipped.clone(),
221 failed,
222 total: plan.total,
223 affected_agents: affected_parents.values().sum(),
224 package_installs: observed_installs.into_iter().collect(),
225 })
226}
227
228async fn recheck_import_object(
229 transport: &Transport,
230 target: &IntegrationPolicyImportTarget,
231) -> Result<ImportAction> {
232 match &target.current {
233 None => match integration_policies::get(transport, &target.effective.id).await {
234 Err(error) if error.kind == ErrorKind::NotFound => Ok(ImportAction::Create),
235 Ok(_) => Err(Error::new(
236 ErrorKind::Conflict,
237 "integration policy appeared since preview",
238 )),
239 Err(error) => Err(import_remote_error(error, "apply integration-policy read")),
240 },
241 Some(expected) => match integration_policies::get(transport, &target.effective.id).await {
242 Ok(actual) if actual.item == expected.item => {
243 if expected.spec == target.effective {
244 Ok(ImportAction::Unchanged)
245 } else {
246 Ok(ImportAction::Replace)
247 }
248 }
249 Ok(_) => Err(Error::new(
250 ErrorKind::Conflict,
251 "integration policy changed since preview",
252 )),
253 Err(error) if error.kind == ErrorKind::NotFound => Err(Error::new(
254 ErrorKind::Conflict,
255 "integration policy disappeared since preview",
256 )),
257 Err(error) => Err(import_remote_error(error, "apply integration-policy read")),
258 },
259 }
260}
261
262async fn recheck_import_name_owner(
263 transport: &Transport,
264 target: &IntegrationPolicyImportTarget,
265 expected_owners: &BTreeMap<String, BTreeSet<String>>,
266) -> Result<()> {
267 let names = BTreeSet::from([target.effective.name.clone()]);
268 let owners = relevant_name_owners(transport, &names)
269 .await
270 .map_err(|error| import_remote_error(error, "apply name read"))?;
271 if owners.get(&target.effective.name) == expected_owners.get(&target.effective.name) {
272 Ok(())
273 } else {
274 Err(Error::new(
275 ErrorKind::Conflict,
276 "integration policy name ownership changed since preview",
277 ))
278 }
279}
280
281async fn recheck_import_parents(
282 transport: &Transport,
283 target: &IntegrationPolicyImportTarget,
284 expected_parents: &BTreeMap<String, agent_policy_ops::AgentPolicyParentSnapshot>,
285) -> Result<()> {
286 for parent_id in target.parents.keys() {
287 let expected = expected_parents.get(parent_id).ok_or_else(|| {
288 Error::new(
289 ErrorKind::Error,
290 "integration import lost a shared parent snapshot",
291 )
292 })?;
293 let actual = agent_policy_ops::read_parent_snapshot(transport, parent_id)
294 .await
295 .map_err(|error| import_remote_error(error, "apply parent read"))?;
296 if actual != *expected {
297 return Err(Error::new(
298 ErrorKind::Conflict,
299 "integration policy parent changed since preview",
300 ));
301 }
302 }
303 Ok(())
304}
305
306fn record_affected_parents(
307 affected: &mut BTreeMap<String, u64>,
308 target: &IntegrationPolicyImportTarget,
309) {
310 for parent in target.parents.values() {
311 affected.entry(parent.id.clone()).or_insert(parent.agents);
312 }
313}
314
315fn advance_parent_snapshots(
316 parents: &mut BTreeMap<String, agent_policy_ops::AgentPolicyParentSnapshot>,
317 target: &IntegrationPolicyImportTarget,
318) {
319 let desired = target
320 .effective
321 .policy_ids
322 .iter()
323 .map(String::as_str)
324 .collect::<BTreeSet<_>>();
325 for parent_id in target.parents.keys() {
326 let parent = parents
327 .get_mut(parent_id)
328 .expect("validated shared parent snapshot");
329 if desired.contains(parent_id.as_str()) {
330 if parent
331 .attached_integrations
332 .binary_search_by(|attached| attached.as_str().cmp(&target.effective.id))
333 .is_err()
334 {
335 parent
336 .attached_integrations
337 .push(target.effective.id.clone());
338 parent.attached_integrations.sort();
339 }
340 } else {
341 parent
342 .attached_integrations
343 .retain(|attached| attached != &target.effective.id);
344 }
345 }
346}
347
348async fn verify_import_stored(
349 transport: &Transport,
350 desired: &IntegrationPolicySpec,
351) -> Result<()> {
352 let stored = integration_policies::get(transport, &desired.id)
353 .await
354 .map_err(|error| import_remote_error(error, "stored-policy read"))?;
355 let stored = normalize(&stored.item, transport.space())
356 .map_err(|error| import_remote_error(error, "stored-policy read"))?;
357 if stored == *desired {
358 Ok(())
359 } else {
360 Err(Error::new(
361 ErrorKind::Http,
362 "server stored a different integration-policy spec",
363 ))
364 }
365}
366
367fn import_failed_row(id: &str, applied: bool, error: impl Into<String>) -> Value {
368 json!({"id": id, "applied": applied, "error": error.into()})
369}