WIP: feat: Implement LDAP sync cache

Cargo.toml

@@ -114,4 +114,5 @@ verbose_file_reads = "warn"
 
 [dev-dependencies]
 ldap3 = { version = "0.11.1", default-features = false, features = ["tls-rustls"] }
+tempfile = "3.10.1"
 test-log = { version = "0.2.16", features = ["trace", "unstable"] }

src/config.rs

@@ -68,7 +68,7 @@ pub struct Config {
 	/// Opt-in features
 	pub feature_flags: Set<FeatureFlag>,
 	/// Where to cache the last known LDAP state
-	pub cache_path: String,
+	pub cache_path: PathBuf,
 	/// The sync tool log level
 	pub log_level: Option<String>,
 }

src/lib.rs

@@ -1,9 +1,11 @@
 //! Simple LDAP -> Famedly Zitadel sync tool to match users between
 //! clients and our infrastructure.
-use anyhow::{Context, Result};
+use std::path::Path;
+
+use anyhow::{bail, Context, Result};
 
 use ldap_poller::ldap3::SearchEntry;
-use ldap_poller::{ldap::EntryStatus, Ldap};
+use ldap_poller::{ldap::EntryStatus, Cache, Ldap};
 use tokio::sync::mpsc::Receiver;
 use tokio_stream::{wrappers::ReceiverStream, StreamExt};
 
@@ -16,18 +18,30 @@ use zitadel::Zitadel;
 
 /// Run the sync
 pub async fn do_the_thing(config: Config) -> Result<()> {
+	let cache = read_cache(&config.cache_path).await?;
 	let zitadel = Zitadel::new(&config).await?;
+	let (mut ldap_client, ldap_receiver) = Ldap::new(config.clone().ldap.into(), cache);
+
+	let sync_handle: tokio::task::JoinHandle<Result<_>> = tokio::spawn(async move {
+		ldap_client.sync_once(None).await.context("failed to sync/fetch data from LDAP")?;
+		let cache = ldap_client.persist_cache().await;
+		tokio::fs::write(
+			&config.cache_path,
+			bincode::serialize(&cache).context("failed to serialize cache")?,
+		)
+		.await
+		.context("failed to write cache")?;
 
-	let (mut ldap_client, ldap_receiver) = Ldap::new(config.clone().ldap.into(), None);
-	tokio::spawn(async move {
-		ldap_client.sync_once(None).await.context("Failed to sync/fetch data from LDAP")
+		Ok(())
 	});
 
 	let (added, _changed, _removed) = get_user_changes(ldap_receiver).await;
 	tracing::info!("Finished syncing LDAP data");
 
 	zitadel.import_new_users(added).await?;
 
+	let _ = sync_handle.await?;
+
 	Ok(())
 }
 
@@ -49,3 +63,18 @@ async fn get_user_changes(
 		})
 		.await
 }
+
+/// Read the ldap sync cache
+async fn read_cache(path: &Path) -> Result<Option<Cache>> {
+	Ok(match tokio::fs::read(path).await {
+		Ok(data) => Some(bincode::deserialize(&data).context("cache deserialization failed")?),
+		Err(err) => {
+			if err.kind() == std::io::ErrorKind::NotFound {
+				tracing::info!("LDAP sync cache missing");
+				None
+			} else {
+				bail!(err)
+			}
+		}
+	})
+}

tests/e2e.rs

@@ -3,13 +3,15 @@
 use std::{collections::HashSet, path::Path, time::Duration};
 
 use ldap3::{Ldap as LdapClient, LdapConnAsync, LdapConnSettings};
+use tempfile::TempDir;
 use test_log::test;
 use tokio::sync::OnceCell;
 use zitadel_rust_client::{Config as ZitadelConfig, Type, Zitadel};
 
 use ldap_sync::{do_the_thing, Config};
 
 static CONFIG: OnceCell<Config> = OnceCell::const_new();
+static TEMPDIR: OnceCell<TempDir> = OnceCell::const_new();
 
 #[test(tokio::test)]
 #[test_log(default_log_filter = "debug")]
@@ -219,9 +221,17 @@ async fn open_zitadel_connection() -> Zitadel {
 async fn config() -> &'static Config {
 	CONFIG
 		.get_or_init(|| async {
-			Config::from_file(Path::new("tests/environment/config.yaml"))
+			let mut config = Config::from_file(Path::new("tests/environment/config.yaml"))
 				.await
-				.expect("failed to parse test env file")
+				.expect("failed to parse test env file");
+
+			let tempdir = TEMPDIR
+				.get_or_init(|| async { TempDir::new().expect("failed to initialize cache dir") })
+				.await;
+
+			config.cache_path = tempdir.path().join("cache.bin");
+
+			config
 		})
 		.await
 }