Pass common args to run_add_group function

1 files changed, 162 insertions, 156 deletions

diff --git a/src/main.rs b/src/main.rs
index 3249cff..8fdc1f7 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -196,7 +196,7 @@ async fn main() -> Result<()> {
             run_sync(&common_args, &args).await?;
         }
         Commands::Add(args) => {
-            // run_add_groups(args).await?;
+            run_add_groups(&common_args, args).await?;
             println!("add");
         } // Commands::Del(args) => {
           //     let common = CommonOperationArgs {
@@ -292,29 +292,35 @@ async fn run_sync(
     Ok(())
 }
 
-// async fn run_add_groups(args: AddArgs) -> Result<()> {
-//     // info!(
-//     //     pool_id = %args.pool_id,
-//     //     emails_file = ?args.emails_file,
-//     //     concurrency = ?args.concurrency,
-//     //     timeout = ?args.timeout,
-//     //     "add groups operation requested (not implemented yet)"
-//     // );
-
-//     let config = aws_config::load_from_env().await;
-//     let client = CognitoClient::new(&config);
-
-//     add_users_to_groups_from_files(
-//         &args.pool_id,
-//         &args.sync_file,
-//         &args.emails_file,
-//         &args.groups,
-//         args.concurrency,
-//     )
-//     .await?;
+async fn run_add_groups(
+    common_args: &CommonOperationArgs,
+    cmd_args: AddArgs,
+) -> Result<()> {
+    info!(
+        pool_id = %common_args.pool_id,
+        emails_file = ?cmd_args.emails_file,
+        sync_file = ?cmd_args.sync_file,
+        groups = ?cmd_args.groups,
+        concurrency = ?common_args.concurrency,
+        timeout = ?common_args.timeout,
+        "add groups operation requested"
+    );
 
-//     Ok(())
-// }
+    let config = aws_config::load_from_env().await;
+    let client = CognitoClient::new(&config);
+
+    add_users_to_groups_from_files(
+        &client,
+        &common_args.pool_id,
+        &cmd_args.sync_file,
+        &cmd_args.emails_file,
+        &cmd_args.groups,
+        common_args.concurrency,
+    )
+    .await?;
+
+    Ok(())
+}
 
 // #[allow(dead_code)]
 // async fn run_remove_groups(args: CommonOperationArgs) -> Result<()> {
@@ -511,141 +517,141 @@ pub async fn load_email_list<P: AsRef<Path>>(path: P) -> Result<Vec<String>> {
     Ok(emails)
 }
 
-// /// Add a Cognito user to one or more groups using the Admin API.
-// ///
-// /// This function assumes AWS credentials and permissions allow admin operations.
-// pub async fn admin_add_user_to_groups(
-//     client: &CognitoClient,
-//     pool_id: &str,
-//     username: &str,
-//     groups: &[String],
-// ) -> Result<()> {
-//     for group in groups {
-//         info!(%username, %group, %pool_id, "adding user to Cognito group");
-
-//         client
-//             .admin_add_user_to_group()
-//             .user_pool_id(pool_id)
-//             .username(username)
-//             .group_name(group)
-//             .send()
-//             .await
-//             .with_context(|| {
-//                 format!(
-//                     "failed to add user '{}' to group '{}'",
-//                     username, group
-//                 )
-//             })?;
-
-//         info!(%username, %group, "user successfully added to group");
-//     }
+/// Add a Cognito user to one or more groups using the Admin API.
+///
+/// This function assumes AWS credentials and permissions allow admin operations.
+pub async fn admin_add_user_to_groups(
+    client: &CognitoClient,
+    pool_id: &str,
+    username: &str,
+    groups: &[String],
+) -> Result<()> {
+    for group in groups {
+        info!(%username, %group, %pool_id, "adding user to Cognito group");
+
+        client
+            .admin_add_user_to_group()
+            .user_pool_id(pool_id)
+            .username(username)
+            .group_name(group)
+            .send()
+            .await
+            .with_context(|| {
+                format!(
+                    "failed to add user '{}' to group '{}'",
+                    username, group
+                )
+            })?;
+
+        info!(%username, %group, "user successfully added to group");
+    }
 
-//     Ok(())
-// }
+    Ok(())
+}
 
-// /// High-level flow:
-// /// 1. Load sync CSV into HashMap<email, username>.
-// /// 2. Load target e-mails (one per line).
-// /// 3. For each email, find username in HashMap.
-// /// 4. Call `admin_add_user_to_groups` with concurrency limit.
-// /// 5. Log success or failure for each email.
-// ///
-// /// `sync_csv_path`: CSV generated by `sync` (username,email).
-// /// `emails_list_path`: TXT file with one e-mail per line (users to be added).
-// pub async fn add_users_to_groups_from_files(
-//     client: &CognitoClient,
-//     pool_id: &str,
-//     sync_csv_path: &Path,
-//     emails_list_path: &Path,
-//     groups: &[String],
-//     concurrency: usize,
-// ) -> Result<()> {
-//     let concurrency = std::cmp::max(1, concurrency);
-
-//     // 1. Load sync CSV into map: email -> username
-//     let email_to_username = read_sync_file_to_map(sync_csv_path).await?;
-//     info!(
-//         total_entries = email_to_username.len(),
-//         "loaded sync map (email -> username)"
-//     );
+/// High-level flow:
+/// 1. Load sync CSV into HashMap<email, username>.
+/// 2. Load target e-mails (one per line).
+/// 3. For each email, find username in HashMap.
+/// 4. Call `admin_add_user_to_groups` with concurrency limit.
+/// 5. Log success or failure for each email.
+///
+/// `sync_csv_path`: CSV generated by `sync` (username,email).
+/// `emails_list_path`: TXT file with one e-mail per line (users to be added).
+pub async fn add_users_to_groups_from_files(
+    client: &CognitoClient,
+    pool_id: &str,
+    sync_csv_path: &Path,
+    emails_list_path: &Path,
+    groups: &[String],
+    concurrency: usize,
+) -> Result<()> {
+    let concurrency = std::cmp::max(1, concurrency);
 
-//     // 2. Load target e-mails
-//     let emails = load_email_list(emails_list_path).await?;
-//     let total_emails = emails.len();
-//     info!(total_emails, "loaded target e-mail list");
-
-//     let pb = Arc::new({
-//         let bar = ProgressBar::new(total_emails as u64);
-//         bar.set_style(
-//             ProgressStyle::with_template(
-//                 "[{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} {msg}",
-//             )
-//             .unwrap_or_else(|_| ProgressStyle::default_bar()),
-//         );
-//         bar.set_message("processing users");
-//         bar
-//     });
-
-//     let semaphore = Arc::new(Semaphore::new(concurrency));
-//     let mut handles: Vec<JoinHandle<()>> = Vec::with_capacity(emails.len());
-
-//     for email in emails {
-//         let username = match email_to_username.get(&email) {
-//             Some(u) => u.clone(),
-//             None => {
-//                 error!(%email, "e-mail not found in sync map, skipping");
-//                 continue;
-//             }
-//         };
-
-//         let permit = semaphore.clone().acquire_owned().await?;
-//         let client_clone = client.clone();
-//         let pool_id = pool_id.to_string();
-//         let groups = groups.to_vec();
-//         let email_clone = email.clone();
-//         let pb_clone = pb.clone();
-
-//         let handle = tokio::spawn(async move {
-//             let _permit = permit; // keep permit alive for the duration of this task
-
-//             if let Err(err) = admin_add_user_to_groups(
-//                 &client_clone,
-//                 &pool_id,
-//                 &username,
-//                 &groups,
-//             )
-//             .await
-//             {
-//                 error!(
-//                     email = %email_clone,
-//                     %username,
-//                     error = ?err,
-//                     "failed to add user to one or more groups"
-//                 );
-//             } else {
-//                 info!(
-//                     email = %email_clone,
-//                     %username,
-//                     groups = ?groups,
-//                     "user successfully processed for all groups"
-//                 );
-//             }
-
-//             pb_clone.inc(1);
-//         });
-
-//         handles.push(handle);
-//     }
+    // 1. Load sync CSV into map: email -> username
+    let email_to_username = read_sync_file_to_map(sync_csv_path).await?;
+    info!(
+        total_entries = email_to_username.len(),
+        "loaded sync map (email -> username)"
+    );
 
-//     // Wait for all tasks to complete
-//     for handle in handles {
-//         // Ignore panics here, just surface as error logs.
-//         if let Err(join_err) = handle.await {
-//             error!(error = ?join_err, "join error while processing a user");
-//         }
-//     }
+    // 2. Load target e-mails
+    let emails = load_email_list(emails_list_path).await?;
+    let total_emails = emails.len();
+    info!(total_emails, "loaded target e-mail list");
 
-//     pb.finish_with_message("done");
-//     info!("finished processing all users for add-groups operation");
-//     Ok(())
-// }
+    let pb = Arc::new({
+        let bar = ProgressBar::new(total_emails as u64);
+        bar.set_style(
+            ProgressStyle::with_template(
+                "[{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} {msg}",
+            )
+            .unwrap_or_else(|_| ProgressStyle::default_bar()),
+        );
+        bar.set_message("processing users");
+        bar
+    });
+
+    let semaphore = Arc::new(Semaphore::new(concurrency));
+    let mut handles: Vec<JoinHandle<()>> = Vec::with_capacity(emails.len());
+
+    for email in emails {
+        let username = match email_to_username.get(&email) {
+            Some(u) => u.clone(),
+            None => {
+                error!(%email, "e-mail not found in sync map, skipping");
+                continue;
+            }
+        };
+
+        let permit = semaphore.clone().acquire_owned().await?;
+        let client_clone = client.clone();
+        let pool_id = pool_id.to_string();
+        let groups = groups.to_vec();
+        let email_clone = email.clone();
+        let pb_clone = pb.clone();
+
+        let handle = tokio::spawn(async move {
+            let _permit = permit; // keep permit alive for the duration of this task
+
+            if let Err(err) = admin_add_user_to_groups(
+                &client_clone,
+                &pool_id,
+                &username,
+                &groups,
+            )
+            .await
+            {
+                error!(
+                    email = %email_clone,
+                    %username,
+                    error = ?err,
+                    "failed to add user to one or more groups"
+                );
+            } else {
+                info!(
+                    email = %email_clone,
+                    %username,
+                    groups = ?groups,
+                    "user successfully processed for all groups"
+                );
+            }
+
+            pb_clone.inc(1);
+        });
+
+        handles.push(handle);
+    }
+
+    // Wait for all tasks to complete
+    for handle in handles {
+        // Ignore panics here, just surface as error logs.
+        if let Err(join_err) = handle.await {
+            error!(error = ?join_err, "join error while processing a user");
+        }
+    }
+
+    pb.finish_with_message("done");
+    info!("finished processing all users for add-groups operation");
+    Ok(())
+}