Sync changes

services/headless-lms/models/src/users.rs

@@ -214,6 +214,43 @@ AND deleted_at IS NULL
     Ok(res.iter().map(|x| x.id).collect::<Vec<_>>())
 }
 
+pub async fn update_email_for_user(
+    conn: &mut PgConnection,
+    upstream_id: &i32,
+    new_email: String,
+) -> ModelResult<()> {
+    info!("Updating user (Upstream id: {upstream_id})");
+    let mut tx = conn.begin().await?;
+
+    let user = sqlx::query_as!(
+        User,
+        "SELECT * FROM users WHERE upstream_id = $1",
+        upstream_id
+    )
+    .fetch_one(&mut *tx)
+    .await?;
+
+    sqlx::query!(
+        "UPDATE user_details SET email = $1 WHERE user_id = $2",
+        new_email,
+        user.id,
+    )
+    .execute(&mut *tx)
+    .await?;
+
+    let email_domain = new_email.trim().split('@').last();
+    sqlx::query!(
+        "UPDATE users SET email_domain = $1 WHERE id = $2",
+        email_domain,
+        user.id,
+    )
+    .execute(&mut *tx)
+    .await?;
+
+    info!("Email change succeeded");
+    Ok(())
+}
+
 pub async fn delete_user(conn: &mut PgConnection, id: Uuid) -> ModelResult<()> {
     info!("Deleting user {id}");
     let mut tx = conn.begin().await?;

services/headless-lms/server/src/programs/sync_tmc_users.rs

@@ -6,9 +6,10 @@ use std::env;
 use crate::setup_tracing;
 use anyhow::Context;
 
+use chrono::DateTime;
 use dotenv::dotenv;
 use headless_lms_models as models;
-use models::users::get_users_ids_in_db_from_upstream_ids;
+use models::users::{get_users_ids_in_db_from_upstream_ids, update_email_for_user};
 
 use serde::{Deserialize, Serialize};
 use sqlx::{PgConnection, PgPool};
@@ -40,6 +41,60 @@ pub async fn main() -> anyhow::Result<()> {
     let db_pool = PgPool::connect(&database_url).await?;
     let mut conn = db_pool.acquire().await?;
     delete_users(&mut conn, &recent_changes).await?;
+    update_users(&mut conn, &recent_changes).await?;
+    Ok(())
+}
+
+pub async fn update_users(
+    conn: &mut PgConnection,
+    recent_changes: &TMCRecentChanges,
+) -> anyhow::Result<()> {
+    let mut email_update_list = recent_changes
+        .changes
+        .iter()
+        .filter(|c| c.change_type == "email_changed")
+        .collect::<Vec<_>>();
+
+    info!("Updating emails for {} users", email_update_list.len());
+    email_update_list.sort_by(|a, b| {
+        let date_a = match DateTime::parse_from_rfc3339(a.created_at.as_str()) {
+            Ok(val) => val,
+            Err(e) => {
+                error!("Error converting date: '{}'", a.created_at);
+                error!("Error: {}", e);
+                DateTime::parse_from_str("01.01.1450", "%d.%m.%Y").unwrap()
+            }
+        };
+
+        let date_b = match DateTime::parse_from_rfc3339(b.created_at.as_str()) {
+            Ok(val) => val,
+            Err(e) => {
+                error!("Error converting date: '{}'", b.created_at);
+                error!("Error: {}", e);
+                DateTime::parse_from_str("01.01.1450", "%d.%m.%Y").unwrap()
+            }
+        };
+
+        date_a.cmp(&date_b)
+    });
+
+    for user in email_update_list {
+        match update_email_for_user(
+            &mut *conn,
+            &user.id,
+            user.new_value.as_deref().unwrap_or("unknown").to_string(),
+        )
+        .await
+        {
+            Ok(email) => email,
+            Err(e) => {
+                error!("Error updating user with id {}", user.id);
+                error!("Error: {}", e);
+            }
+        };
+    }
+
+    info!("Update done");
     Ok(())
 }