feat: Introduce User Groups to Parseable

- Added support for user groups
- Migrated `PUT /user/{username}/role` to `PATCH /user/{username}/role/add` and `PATCH /user/{username}/role/remove`

Author: parmesant

src/handlers/http/cluster/mod.rs

@@ -160,20 +160,29 @@ pub async fn sync_streams_with_ingestors(
 pub async fn sync_users_with_roles_with_ingestors(
     username: &str,
     role: &HashSet<String>,
+    operation: &str,
 ) -> Result<(), RBACError> {
+    match operation {
+        "add" | "remove" => {}
+        _ => return Err(RBACError::InvalidSyncOperation(operation.to_string())),
+    }
+
     let role_data = to_vec(&role.clone()).map_err(|err| {
         error!("Fatal: failed to serialize role: {:?}", err);
         RBACError::SerdeError(err)
     })?;
 
     let username = username.to_owned();
 
+    let op = operation.to_string();
+
     for_each_live_ingestor(move |ingestor| {
         let url = format!(
-            "{}{}/user/{}/role/sync",
+            "{}{}/user/{}/role/sync/{}",
             ingestor.domain_name,
             base_path_without_preceding_slash(),
-            username
+            username,
+            op
         );
 
         let role_data = role_data.clone();

src/handlers/http/modal/ingest/ingestor_rbac.rs

@@ -48,7 +48,7 @@ pub async fn post_user(
         let _ = storage::put_staging_metadata(&metadata);
         let created_role = user.roles.clone();
         Users.put_user(user.clone());
-        Users.put_role(&username, created_role.clone());
+        Users.add_roles(&username, created_role.clone());
     }
 
     Ok(generated_password)
@@ -73,14 +73,45 @@ pub async fn delete_user(username: web::Path<String>) -> Result<impl Responder,
     Ok(format!("deleted user: {username}"))
 }
 
-// Handler PUT /user/{username}/roles => Put roles for user
-// Put roles for given user
-pub async fn put_role(
+// // Handler PUT /user/{username}/roles => Put roles for user
+// // Put roles for given user
+// pub async fn put_role(
+//     username: web::Path<String>,
+//     role: web::Json<HashSet<String>>,
+// ) -> Result<String, RBACError> {
+//     let username = username.into_inner();
+//     let role = role.into_inner();
+
+//     if !Users.contains(&username) {
+//         return Err(RBACError::UserDoesNotExist);
+//     };
+//     // update parseable.json first
+//     let mut metadata = get_metadata().await?;
+//     if let Some(user) = metadata
+//         .users
+//         .iter_mut()
+//         .find(|user| user.username() == username)
+//     {
+//         user.roles.clone_from(&role);
+//     } else {
+//         // should be unreachable given state is always consistent
+//         return Err(RBACError::UserDoesNotExist);
+//     }
+
+//     let _ = storage::put_staging_metadata(&metadata);
+//     // update in mem table
+//     Users.add_roles(&username.clone(), role.clone());
+
+//     Ok(format!("Roles updated successfully for {username}"))
+// }
+
+// Handler PATCH /user/{username}/role/sync/add => Add roles to a user
+pub async fn add_roles_to_user(
     username: web::Path<String>,
-    role: web::Json<HashSet<String>>,
+    roles_to_add: web::Json<HashSet<String>>,
 ) -> Result<String, RBACError> {
     let username = username.into_inner();
-    let role = role.into_inner();
+    let roles_to_add = roles_to_add.into_inner();
 
     if !Users.contains(&username) {
         return Err(RBACError::UserDoesNotExist);
@@ -92,15 +123,48 @@ pub async fn put_role(
         .iter_mut()
         .find(|user| user.username() == username)
     {
-        user.roles.clone_from(&role);
+        user.roles.extend(roles_to_add.clone());
     } else {
         // should be unreachable given state is always consistent
         return Err(RBACError::UserDoesNotExist);
     }
 
     let _ = storage::put_staging_metadata(&metadata);
     // update in mem table
-    Users.put_role(&username.clone(), role.clone());
+    Users.add_roles(&username.clone(), roles_to_add.clone());
+
+    Ok(format!("Roles updated successfully for {username}"))
+}
+
+// Handler PATCH /user/{username}/role/sync/add => Add roles to a user
+pub async fn remove_roles_from_user(
+    username: web::Path<String>,
+    roles_to_remove: web::Json<HashSet<String>>,
+) -> Result<String, RBACError> {
+    let username = username.into_inner();
+    let roles_to_remove = roles_to_remove.into_inner();
+
+    if !Users.contains(&username) {
+        return Err(RBACError::UserDoesNotExist);
+    };
+    // update parseable.json first
+    let mut metadata = get_metadata().await?;
+    if let Some(user) = metadata
+        .users
+        .iter_mut()
+        .find(|user| user.username() == username)
+    {
+        let diff: HashSet<String> =
+            HashSet::from_iter(user.roles.difference(&roles_to_remove).cloned());
+        user.roles = diff;
+    } else {
+        // should be unreachable given state is always consistent
+        return Err(RBACError::UserDoesNotExist);
+    }
+
+    let _ = storage::put_staging_metadata(&metadata);
+    // update in mem table
+    Users.remove_roles(&username.clone(), roles_to_remove.clone());
 
     Ok(format!("Roles updated successfully for {username}"))
 }

src/handlers/http/modal/ingest_server.rs

@@ -198,11 +198,21 @@ impl IngestServer {
                     .wrap(DisAllowRootUser),
             )
             .service(
-                web::resource("/{username}/role/sync")
-                    // PUT /user/{username}/roles => Put roles for user
+                web::resource("/{username}/role/sync/add")
+                    // PATCH /user/{username}/role/sync/add => Add roles to a user
                     .route(
-                        web::put()
-                            .to(ingestor_rbac::put_role)
+                        web::patch()
+                            .to(ingestor_rbac::add_roles_to_user)
+                            .authorize(Action::PutUserRoles)
+                            .wrap(DisAllowRootUser),
+                    ),
+            )
+            .service(
+                web::resource("/{username}/role/sync/remove")
+                    // PATCH /user/{username}/role/sync/remove => Remove roles from a user
+                    .route(
+                        web::patch()
+                            .to(ingestor_rbac::remove_roles_from_user)
                             .authorize(Action::PutUserRoles)
                             .wrap(DisAllowRootUser),
                     ),