server: add support for updating user attributes

server/src/domain/handler.rs

@@ -1,8 +1,8 @@
 use crate::domain::{
     error::Result,
     types::{
-        AttributeType, Group, GroupDetails, GroupId, JpegPhoto, User, UserAndGroups, UserColumn,
+        AttributeType, AttributeValue, Group, GroupDetails, GroupId, JpegPhoto, User,
-        UserId, Uuid,
+        UserAndGroups, UserColumn, UserId, Uuid,
     },
 };
 use async_trait::async_trait;
@@ -115,6 +115,8 @@ pub struct UpdateUserRequest {
     pub first_name: Option<String>,
     pub last_name: Option<String>,
     pub avatar: Option<JpegPhoto>,
+    pub delete_attributes: Vec<String>,
+    pub insert_attributes: Vec<AttributeValue>,
 }
 
 #[derive(PartialEq, Eq, Debug, Serialize, Deserialize, Clone)]

server/src/domain/sql_schema_backend_handler.rs

@@ -1,43 +1,56 @@
 use crate::domain::{
-    error::Result,
+    error::{DomainError, Result},
-    handler::{AttributeSchema, Schema, SchemaBackendHandler},
+    handler::{AttributeList, AttributeSchema, Schema, SchemaBackendHandler},
     model,
     sql_backend_handler::SqlBackendHandler,
 };
 use async_trait::async_trait;
-use sea_orm::{EntityTrait, QueryOrder};
+use sea_orm::{DatabaseTransaction, EntityTrait, QueryOrder, TransactionTrait};
-
-use super::handler::AttributeList;
 
 #[async_trait]
 impl SchemaBackendHandler for SqlBackendHandler {
     async fn get_schema(&self) -> Result<Schema> {
-        Ok(Schema {
+        Ok(self
-            user_attributes: AttributeList {
+            .sql_pool
-                attributes: self.get_user_attributes().await?,
+            .transaction::<_, Schema, DomainError>(|transaction| {
-            },
+                Box::pin(async move { Self::get_schema_with_transaction(transaction).await })
-            group_attributes: AttributeList {
+            })
-                attributes: self.get_group_attributes().await?,
+            .await?)
-            },
-        })
     }
 }
 
 impl SqlBackendHandler {
-    async fn get_user_attributes(&self) -> Result<Vec<AttributeSchema>> {
+    pub(crate) async fn get_schema_with_transaction(
+        transaction: &DatabaseTransaction,
+    ) -> Result<Schema> {
+        Ok(Schema {
+            user_attributes: AttributeList {
+                attributes: Self::get_user_attributes(transaction).await?,
+            },
+            group_attributes: AttributeList {
+                attributes: Self::get_group_attributes(transaction).await?,
+            },
+        })
+    }
+
+    async fn get_user_attributes(
+        transaction: &DatabaseTransaction,
+    ) -> Result<Vec<AttributeSchema>> {
         Ok(model::UserAttributeSchema::find()
             .order_by_asc(model::UserAttributeSchemaColumn::AttributeName)
-            .all(&self.sql_pool)
+            .all(transaction)
             .await?
             .into_iter()
             .map(|m| m.into())
             .collect())
     }
 
-    async fn get_group_attributes(&self) -> Result<Vec<AttributeSchema>> {
+    async fn get_group_attributes(
+        transaction: &DatabaseTransaction,
+    ) -> Result<Vec<AttributeSchema>> {
         Ok(model::GroupAttributeSchema::find()
             .order_by_asc(model::GroupAttributeSchemaColumn::AttributeName)
-            .all(&self.sql_pool)
+            .all(transaction)
             .await?
             .into_iter()
             .map(|m| m.into())

server/src/domain/sql_user_backend_handler.rs

@@ -13,8 +13,8 @@ use sea_orm::{
     sea_query::{
         query::OnConflict, Alias, Cond, Expr, Func, IntoColumnRef, IntoCondition, SimpleExpr,
     },
-    ActiveModelTrait, ActiveValue, ColumnTrait, EntityTrait, IntoActiveValue, ModelTrait,
+    ActiveModelTrait, ActiveValue, ColumnTrait, DatabaseTransaction, EntityTrait, IntoActiveValue,
-    QueryFilter, QueryOrder, QuerySelect, QueryTrait, Set, TransactionTrait,
+    ModelTrait, QueryFilter, QueryOrder, QuerySelect, QueryTrait, Set, TransactionTrait,
 };
 use std::collections::HashSet;
 use tracing::instrument;
@@ -154,6 +154,101 @@ impl UserListerBackendHandler for SqlBackendHandler {
     }
 }
 
+impl SqlBackendHandler {
+    async fn update_user_with_transaction(
+        transaction: &DatabaseTransaction,
+        request: UpdateUserRequest,
+    ) -> Result<()> {
+        let update_user = model::users::ActiveModel {
+            user_id: ActiveValue::Set(request.user_id.clone()),
+            email: request.email.map(ActiveValue::Set).unwrap_or_default(),
+            display_name: to_value(&request.display_name),
+            ..Default::default()
+        };
+        let to_serialized_value = |s: &Option<String>| match s.as_ref().map(|s| s.as_str()) {
+            None => None,
+            Some("") => Some(ActiveValue::NotSet),
+            Some(s) => Some(ActiveValue::Set(Serialized::from(s))),
+        };
+        let mut update_user_attributes = Vec::new();
+        let mut remove_user_attributes = Vec::new();
+        let mut process_serialized =
+            |value: ActiveValue<Serialized>, attribute_name: &str| match &value {
+                ActiveValue::NotSet => {
+                    remove_user_attributes.push(attribute_name.to_owned());
+                }
+                ActiveValue::Set(_) => {
+                    update_user_attributes.push(model::user_attributes::ActiveModel {
+                        user_id: Set(request.user_id.clone()),
+                        attribute_name: Set(attribute_name.to_owned()),
+                        value,
+                    })
+                }
+                _ => unreachable!(),
+            };
+        if let Some(value) = to_serialized_value(&request.first_name) {
+            process_serialized(value, "first_name");
+        }
+        if let Some(value) = to_serialized_value(&request.last_name) {
+            process_serialized(value, "last_name");
+        }
+        if let Some(avatar) = request.avatar {
+            process_serialized(avatar.into_active_value(), "avatar");
+        }
+        let schema = Self::get_schema_with_transaction(transaction).await?;
+        for attribute in request.insert_attributes {
+            if schema
+                .user_attributes
+                .get_attribute_type(&attribute.name)
+                .is_some()
+            {
+                process_serialized(ActiveValue::Set(attribute.value), &attribute.name);
+            } else {
+                return Err(DomainError::InternalError(format!(
+                    "Attribute name {} doesn't exist in the schema, yet was attempted to be inserted in the database",
+                    &attribute.name
+                )));
+            }
+        }
+        for attribute in request.delete_attributes {
+            if schema
+                .user_attributes
+                .get_attribute_type(&attribute)
+                .is_some()
+            {
+                remove_user_attributes.push(attribute);
+            } else {
+                return Err(DomainError::InternalError(format!(
+                    "Attribute name {} doesn't exist in the schema, yet was attempted to be removed from the database",
+                    attribute
+                )));
+            }
+        }
+        update_user.update(transaction).await?;
+        if !remove_user_attributes.is_empty() {
+            model::UserAttributes::delete_many()
+                .filter(model::UserAttributesColumn::UserId.eq(&request.user_id))
+                .filter(model::UserAttributesColumn::AttributeName.is_in(remove_user_attributes))
+                .exec(transaction)
+                .await?;
+        }
+        if !update_user_attributes.is_empty() {
+            model::UserAttributes::insert_many(update_user_attributes)
+                .on_conflict(
+                    OnConflict::columns([
+                        model::UserAttributesColumn::UserId,
+                        model::UserAttributesColumn::AttributeName,
+                    ])
+                    .update_column(model::UserAttributesColumn::Value)
+                    .to_owned(),
+                )
+                .exec(transaction)
+                .await?;
+        }
+        Ok(())
+    }
+}
+
 #[async_trait]
 impl UserBackendHandler for SqlBackendHandler {
     #[instrument(skip_all, level = "debug", ret, fields(user_id = ?user_id.as_str()))]
@@ -240,71 +335,11 @@ impl UserBackendHandler for SqlBackendHandler {
 
     #[instrument(skip(self), level = "debug", err, fields(user_id = ?request.user_id.as_str()))]
     async fn update_user(&self, request: UpdateUserRequest) -> Result<()> {
-        let update_user = model::users::ActiveModel {
-            user_id: ActiveValue::Set(request.user_id.clone()),
-            email: request.email.map(ActiveValue::Set).unwrap_or_default(),
-            display_name: to_value(&request.display_name),
-            ..Default::default()
-        };
-        let mut update_user_attributes = Vec::new();
-        let mut remove_user_attributes = Vec::new();
-        let to_serialized_value = |s: &Option<String>| match s.as_ref().map(|s| s.as_str()) {
-            None => None,
-            Some("") => Some(ActiveValue::NotSet),
-            Some(s) => Some(ActiveValue::Set(Serialized::from(s))),
-        };
-        let mut process_serialized =
-            |value: ActiveValue<Serialized>, attribute_name: &str| match &value {
-                ActiveValue::NotSet => {
-                    remove_user_attributes.push(attribute_name.to_owned());
-                }
-                ActiveValue::Set(_) => {
-                    update_user_attributes.push(model::user_attributes::ActiveModel {
-                        user_id: Set(request.user_id.clone()),
-                        attribute_name: Set(attribute_name.to_owned()),
-                        value,
-                    })
-                }
-                _ => unreachable!(),
-            };
-        if let Some(value) = to_serialized_value(&request.first_name) {
-            process_serialized(value, "first_name");
-        }
-        if let Some(value) = to_serialized_value(&request.last_name) {
-            process_serialized(value, "last_name");
-        }
-        if let Some(avatar) = request.avatar {
-            process_serialized(avatar.into_active_value(), "avatar");
-        }
         self.sql_pool
             .transaction::<_, (), DomainError>(|transaction| {
-                Box::pin(async move {
+                Box::pin(
-                    update_user.update(transaction).await?;
+                    async move { Self::update_user_with_transaction(transaction, request).await },
-                    if !update_user_attributes.is_empty() {
+                )
-                        model::UserAttributes::insert_many(update_user_attributes)
-                            .on_conflict(
-                                OnConflict::columns([
-                                    model::UserAttributesColumn::UserId,
-                                    model::UserAttributesColumn::AttributeName,
-                                ])
-                                .update_column(model::UserAttributesColumn::Value)
-                                .to_owned(),
-                            )
-                            .exec(transaction)
-                            .await?;
-                    }
-                    if !remove_user_attributes.is_empty() {
-                        model::UserAttributes::delete_many()
-                            .filter(model::UserAttributesColumn::UserId.eq(&request.user_id))
-                            .filter(
-                                model::UserAttributesColumn::AttributeName
-                                    .is_in(remove_user_attributes),
-                            )
-                            .exec(transaction)
-                            .await?;
-                    }
-                    Ok(())
-                })
             })
             .await?;
         Ok(())
@@ -714,6 +749,8 @@ mod tests {
                 first_name: Some("first_name".to_string()),
                 last_name: Some("last_name".to_string()),
                 avatar: Some(JpegPhoto::for_tests()),
+                delete_attributes: Vec::new(),
+                insert_attributes: Vec::new(),
             })
             .await
             .unwrap();
@@ -781,6 +818,118 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_update_user_insert_attribute() {
+        let fixture = TestFixture::new().await;
+
+        fixture
+            .handler
+            .update_user(UpdateUserRequest {
+                user_id: UserId::new("bob"),
+                first_name: None,
+                last_name: None,
+                avatar: None,
+                insert_attributes: vec![AttributeValue {
+                    name: "first_name".to_owned(),
+                    value: Serialized::from("new first"),
+                }],
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        let user = fixture
+            .handler
+            .get_user_details(&UserId::new("bob"))
+            .await
+            .unwrap();
+        assert_eq!(
+            user.attributes,
+            vec![
+                AttributeValue {
+                    name: "first_name".to_owned(),
+                    value: Serialized::from("new first")
+                },
+                AttributeValue {
+                    name: "last_name".to_owned(),
+                    value: Serialized::from("last bob")
+                }
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn test_update_user_delete_attribute() {
+        let fixture = TestFixture::new().await;
+
+        fixture
+            .handler
+            .update_user(UpdateUserRequest {
+                user_id: UserId::new("bob"),
+                first_name: None,
+                last_name: None,
+                avatar: None,
+                delete_attributes: vec!["first_name".to_owned()],
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        let user = fixture
+            .handler
+            .get_user_details(&UserId::new("bob"))
+            .await
+            .unwrap();
+        assert_eq!(
+            user.attributes,
+            vec![AttributeValue {
+                name: "last_name".to_owned(),
+                value: Serialized::from("last bob")
+            }]
+        );
+    }
+
+    #[tokio::test]
+    async fn test_update_user_replace_attribute() {
+        let fixture = TestFixture::new().await;
+
+        fixture
+            .handler
+            .update_user(UpdateUserRequest {
+                user_id: UserId::new("bob"),
+                first_name: None,
+                last_name: None,
+                avatar: None,
+                delete_attributes: vec!["first_name".to_owned()],
+                insert_attributes: vec![AttributeValue {
+                    name: "first_name".to_owned(),
+                    value: Serialized::from("new first"),
+                }],
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        let user = fixture
+            .handler
+            .get_user_details(&UserId::new("bob"))
+            .await
+            .unwrap();
+        assert_eq!(
+            user.attributes,
+            vec![
+                AttributeValue {
+                    name: "first_name".to_owned(),
+                    value: Serialized::from("new first")
+                },
+                AttributeValue {
+                    name: "last_name".to_owned(),
+                    value: Serialized::from("last bob")
+                },
+            ]
+        );
+    }
+
     #[tokio::test]
     async fn test_update_user_delete_avatar() {
         let fixture = TestFixture::new().await;

server/src/infra/graphql/mutation.rs

@@ -161,6 +161,7 @@ impl<Handler: BackendHandler> Mutation<Handler> {
                 first_name: user.first_name,
                 last_name: user.last_name,
                 avatar,
+                ..Default::default()
             })
             .instrument(span)
             .await?;