⚡ Parallelize groups apply processing

Cargo.toml

@@ -29,3 +29,7 @@ harness = false
 [[bench]]
 name = "bench_inspect"
 harness = false
+
+[[bench]]
+name = "bench_apply_groups"
+harness = false

benches/bench_apply_groups.rs

@@ -0,0 +1,43 @@
+use app::client::KeycloakClient;
+use app::apply;
+use std::path::PathBuf;
+use tokio::runtime::Runtime;
+use std::fs;
+
+#[path = "../tests/common/mod.rs"]
+mod common;
+
+fn main() {
+    let rt = Runtime::new().unwrap();
+    rt.block_on(async {
+        let server_url = common::start_mock_server().await;
+        let mut client = KeycloakClient::new(server_url, "test-realm".to_string());
+        client
+            .login("admin-cli", None, Some("admin"), Some("admin"))
+            .await
+            .unwrap();
+
+        let temp_dir = tempfile::tempdir().unwrap();
+        let groups_dir = temp_dir.path().join("groups");
+        fs::create_dir_all(&groups_dir).unwrap();
+
+        for i in 0..100 {
+            let group_name = format!("group-{}", i);
+            let group_file = groups_dir.join(format!("{}.yaml", group_name));
+            let content = format!("name: {}\n", group_name);
+            fs::write(group_file, content).unwrap();
+        }
+
+        // Warm up
+        let _ = apply::run(&client, temp_dir.path().to_path_buf()).await;
+
+        let start = std::time::Instant::now();
+        for _ in 0..10 {
+            apply::run(&client, temp_dir.path().to_path_buf())
+                .await
+                .unwrap();
+        }
+        let elapsed = start.elapsed();
+        println!("Elapsed time for 10 iterations of 100 groups: {:?}", elapsed);
+    });
+}

src/apply.rs

@@ -272,40 +272,51 @@ async fn apply_groups(client: &KeycloakClient, input_dir: &std::path::Path) -> R
             .into_iter()
             .filter_map(|g| g.name.clone().map(|n| (n, g)))
             .collect();
+        let existing_groups_map = std::sync::Arc::new(existing_groups_map);
 
         let mut entries = async_fs::read_dir(&groups_dir).await?;
+        let mut set = JoinSet::new();
+
         while let Some(entry) = entries.next_entry().await? {
             let path = entry.path();
             if path.extension().is_some_and(|ext| ext == "yaml") {
-                let content = async_fs::read_to_string(&path).await?;
-                let mut val: serde_json::Value = serde_yaml::from_str(&content)?;
-                substitute_secrets(&mut val);
-                let mut group_rep: GroupRepresentation = serde_json::from_value(val)?;
-                let name = group_rep.name.as_deref().unwrap_or("");
+                let client = client.clone();
+                let existing_groups_map = existing_groups_map.clone();
+                set.spawn(async move {
+                    let content = async_fs::read_to_string(&path).await?;
+                    let mut val: serde_json::Value = serde_yaml::from_str(&content)?;
+                    substitute_secrets(&mut val);
+                    let mut group_rep: GroupRepresentation = serde_json::from_value(val)?;
+                    let name = group_rep.name.clone().unwrap_or_default();
 
-                if name.is_empty() {
-                    continue;
-                }
+                    if name.is_empty() {
+                        return Ok::<(), anyhow::Error>(());
+                    }
 
-                if let Some(existing) = existing_groups_map.get(name) {
-                    if let Some(id) = &existing.id {
-                        group_rep.id = Some(id.clone());
+                    if let Some(existing) = existing_groups_map.get(&name) {
+                        if let Some(id) = &existing.id {
+                            group_rep.id = Some(id.clone());
+                            client
+                                .update_group(id, &group_rep)
+                                .await
+                                .context(format!("Failed to update group {}", name))?;
+                            println!("Updated group {}", name);
+                        }
+                    } else {
+                        group_rep.id = None;
                         client
-                            .update_group(id, &group_rep)
+                            .create_group(&group_rep)
                             .await
-                            .context(format!("Failed to update group {}", name))?;
-                        println!("Updated group {}", name);
+                            .context(format!("Failed to create group {}", name))?;
+                        println!("Created group {}", name);
                     }
-                } else {
-                    group_rep.id = None;
-                    client
-                        .create_group(&group_rep)
-                        .await
-                        .context(format!("Failed to create group {}", name))?;
-                    println!("Created group {}", name);
-                }
+                    Ok::<(), anyhow::Error>(())
+                });
             }
         }
+        while let Some(res) = set.join_next().await {
+            res??;
+        }
     }
     Ok(())
 }