feat: Delete disabled users

src/lib.rs

@@ -35,10 +35,11 @@ pub async fn do_the_thing(config: Config) -> Result<()> {
 		Ok(())
 	});
 
-	let (added, _changed, _removed) = get_user_changes(ldap_receiver).await;
+	let (added, changed, _removed) = get_user_changes(ldap_receiver).await;
 	tracing::info!("Finished syncing LDAP data");
 
 	zitadel.import_new_users(added).await?;
+	zitadel.update_users(changed).await?;
 
 	let _ = sync_handle.await?;
 
@@ -48,15 +49,15 @@ pub async fn do_the_thing(config: Config) -> Result<()> {
 /// Get user changes from an ldap receiver
 async fn get_user_changes(
 	ldap_receiver: Receiver<EntryStatus>,
-) -> (Vec<SearchEntry>, Vec<SearchEntry>, Vec<Vec<u8>>) {
+) -> (Vec<SearchEntry>, Vec<(SearchEntry, SearchEntry)>, Vec<Vec<u8>>) {
 	ReceiverStream::new(ldap_receiver)
 		.fold((vec![], vec![], vec![]), |(mut added, mut changed, mut removed), entry_status| {
 			match entry_status {
 				EntryStatus::New(entry) => {
 					tracing::debug!("New entry: {:?}", entry);
 					added.push(entry);
 				}
-				EntryStatus::Changed { old: _, new } => changed.push(new),
+				EntryStatus::Changed { old, new } => changed.push((old, new)),
 				EntryStatus::Removed(entry) => removed.push(entry),
 			};
 			(added, changed, removed)

src/zitadel.rs

@@ -1,5 +1,5 @@
 //! Helper functions for submitting data to Zitadel
-use anyhow::{anyhow, Result};
+use anyhow::{anyhow, bail, Result};
 use itertools::Itertools;
 use ldap_poller::ldap3::SearchEntry;
 use uuid::{uuid, Uuid};
@@ -64,6 +64,59 @@ impl Zitadel {
 		Ok(())
 	}
 
+	/// Update a list of old/new user maps
+	pub(crate) async fn update_users(&self, users: Vec<(SearchEntry, SearchEntry)>) -> Result<()> {
+		let (users, invalid): (Vec<_>, Vec<anyhow::Error>) = users
+			.into_iter()
+			.map(|(old, new)| {
+				let old = User::try_from_search_entry(old, &self.config)?;
+				let new = User::try_from_search_entry(new, &self.config)?;
+
+				Ok((old, new))
+			})
+			.partition_result();
+
+		if !invalid.is_empty() {
+			let messages = invalid
+				.into_iter()
+				.fold(String::default(), |acc, error| acc + error.to_string().as_str() + "\n");
+
+			tracing::warn!("Some users cannot be updated due to missing attributes:\n{}", messages);
+		}
+
+		let disabled: Vec<User> = users
+			.iter()
+			.filter(|&(old, new)| (old.enabled && !new.enabled))
+			.map(|(_, new)| new.clone())
+			.collect();
+
+		let _enabled: Vec<User> = users
+			.iter()
+			.filter(|(old, new)| (!old.enabled && new.enabled))
+			.map(|(_, new)| new.clone())
+			.collect();
+
+		for user in disabled {
+			let status = self.delete_user(&user).await;
+
+			if let Err(error) = status {
+				tracing::error!("Failed to delete user `{}`: {}`", user.ldap_id, error);
+			}
+		}
+
+		Ok(())
+	}
+
+	/// Delete a Zitadel user
+	async fn delete_user(&self, user: &User) -> Result<()> {
+		if let Some(user_id) = self.client.get_user_by_login_name(&user.email).await? {
+			self.client.remove_user(user_id.id).await?;
+		} else {
+			bail!("could not find user `{}` for deletion", user.email);
+		}
+		Ok(())
+	}
+
 	/// Import a user into Zitadel
 	async fn import_user(&self, user: &User) -> Result<()> {
 		let new_user_id = self

tests/e2e.rs

@@ -2,7 +2,7 @@
 
 use std::{collections::HashSet, path::Path, time::Duration};
 
-use ldap3::{Ldap as LdapClient, LdapConnAsync, LdapConnSettings};
+use ldap3::{Ldap as LdapClient, LdapConnAsync, LdapConnSettings, Mod};
 use tempfile::TempDir;
 use test_log::test;
 use tokio::sync::OnceCell;
@@ -144,7 +144,7 @@ async fn test_e2e_sync_change() {
 
 	do_the_thing(config().await.clone()).await.expect("syncing failed");
 
-	ldap.change_user("change", "telephoneNumber", "+12015550123").await;
+	ldap.change_user("change", vec![("telephoneNumber", HashSet::from(["+12015550123"]))]).await;
 
 	do_the_thing(config().await.clone()).await.expect("syncing failed");
 
@@ -181,17 +181,14 @@ async fn test_e2e_sync_disable() {
 
 	do_the_thing(config().await.clone()).await.expect("syncing failed");
 
-	ldap.disable_user("disable").await;
+	ldap.change_user("disable", vec![("shadowInactive", HashSet::from(["514"]))]).await;
 
 	do_the_thing(config().await.clone()).await.expect("syncing failed");
 
 	let zitadel = open_zitadel_connection().await;
-	let user = zitadel
-		.get_user_by_login_name("[email protected]")
-		.await
-		.expect("could not query Zitadel users");
+	let user = zitadel.get_user_by_login_name("[email protected]").await;
 
-	assert!(user.is_none());
+	assert!(user.is_err_and(|error| matches!(error, ZitadelError::TonicResponseError(status) if status.code() == TonicErrorCode::NotFound)));
 }
 
 struct Ldap {
@@ -262,12 +259,16 @@ impl Ldap {
 		tracing::info!("Successfully added test user");
 	}
 
-	async fn change_user(&self, cn: &str, attribute: &str, value: &str) {
-		todo!()
-	}
+	async fn change_user(&mut self, uid: &str, changes: Vec<(&str, HashSet<&str>)>) {
+		let mods = changes
+			.into_iter()
+			.map(|(attribute, changes)| Mod::Replace(attribute, changes))
+			.collect();
 
-	async fn disable_user(&self, cn: &str) {
-		todo!()
+		self.client
+			.modify(&format!("uid={},{}", uid, config().await.ldap.base_dn.as_str()), mods)
+			.await
+			.expect("failed to modify userg");
 	}
 }