Fixed a bug with duplicate notifications

src/bin/server.rs

@@ -11,9 +11,9 @@ use tachikoma::telemetry::init_tracing;
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
-    let settings = get_config()?;
-    init_tracing();
     set_env();
+    init_tracing();
+    let settings = get_config()?;
 
     run_migrations(&settings.database).await?;
     let registry = Registry::new(&settings.database).await?;

src/bin/tachikoma.rs

@@ -22,9 +22,9 @@ use tachikoma::telemetry::init_tracing;
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
-    let settings = get_config()?;
-    init_tracing();
     set_env();
+    init_tracing();
+    let settings = get_config()?;
 
     tracing::info!("Starting tachikama");
     run_migrations(&settings.database).await?;

src/logic/release.rs

@@ -3,7 +3,7 @@ use std::{
     time::Duration,
 };
 
-use chrono::{DateTime, Utc};
+use chrono::Utc;
 
 use tokio::time::sleep;
 
@@ -20,7 +20,7 @@ use tracing::{debug, error};
 pub async fn hosts_release_timer<T: GetMessageSender>(registry: Registry, notifier: &Notifier<T>) {
     let mut release_timer = ReleaseTimer {
         registry,
-        last_expiration_soon_notification: HashMap::new(),
+        sent_hosts_notification: HashMap::new(),
         expiration_notify_delay_time: Duration::from_secs(30 * 60),
     };
     loop {
@@ -53,7 +53,7 @@ pub async fn hosts_release_timer<T: GetMessageSender>(registry: Registry, notifi
 
 struct ReleaseTimer {
     registry: Registry,
-    last_expiration_soon_notification: HashMap<UserId, (DateTime<Utc>, HashSet<HostId>)>,
+    sent_hosts_notification: HashMap<UserId, HashSet<HostId>>,
     expiration_notify_delay_time: Duration,
 }
 
@@ -62,6 +62,11 @@ impl ReleaseTimer {
         let mut tx = self.registry.begin().await?;
 
         let expired_hosts = tx.get_leased_until_hosts(Utc::now()).await?;
+        let mut users_hosts: HashMap<UserId, HashSet<HostId>> = HashMap::new();
+        expired_hosts.iter().for_each(|host| {
+            users_hosts.entry(host.user.id).or_default().insert(host.id);
+        });
+
         if !expired_hosts.is_empty() {
             tx.free_hosts(
                 expired_hosts
@@ -73,6 +78,16 @@ impl ReleaseTimer {
             .await?;
             tx.commit().await?;
         }
+
+        users_hosts
+            .drain()
+            .for_each(|(user_id, expired_hosts_ids)| {
+                self.sent_hosts_notification
+                    .entry(user_id)
+                    .and_modify(|hosts_ids| {
+                        hosts_ids.retain(|host_id| !expired_hosts_ids.contains(host_id));
+                    });
+            });
         Ok(expired_hosts)
     }
 
@@ -124,27 +139,22 @@ impl ReleaseTimer {
         });
 
         for (user_id, hosts_ids) in expire_soon_notifications.into_iter() {
-            if let Some((last_notify_time, last_hosts_ids)) =
-                self.last_expiration_soon_notification.get(&user_id)
-            {
-                let delta = Utc::now() - last_notify_time;
-                if delta.num_seconds() < self.expiration_notify_delay_time.as_secs() as i64
-                    && hosts_ids.eq(last_hosts_ids)
-                {
-                    continue;
-                }
+            let sent_host_ids = self.sent_hosts_notification.entry(user_id).or_default();
+            let hosts_ids: Vec<HostId> = hosts_ids.difference(sent_host_ids).copied().collect();
+
+            if hosts_ids.is_empty() {
+                continue;
             }
 
-            let notification =
-                Notification::ExpirationSoon(hosts_ids.clone().into_iter().collect());
+            let notification = Notification::ExpirationSoon(hosts_ids.clone());
             if let Err(err) = notifier.notify(user_id, &notification).await {
                 error!(
                     "Notification sending error {:?} {:?}: {err}",
                     user_id, notification
                 );
+            } else {
+                sent_host_ids.extend(hosts_ids.iter());
             }
-            self.last_expiration_soon_notification
-                .insert(user_id, (Utc::now(), hosts_ids));
         }
 
         Ok(())