WIP: prevent duplicate alerts

Closes kemitix/git-next#128

Cargo.lock

@@ -945,8 +945,6 @@ dependencies = [
  "actix-rt",
  "anyhow",
  "assert2",
- "async-trait",
- "base64 0.22.1",
  "bytes",
  "clap",
  "derive-with",
@@ -964,7 +962,6 @@ dependencies = [
  "rand",
  "secrecy",
  "sendmail",
- "serde",
  "serde_json",
  "standardwebhooks",
  "test-log",
@@ -982,7 +979,6 @@ name = "git-next-core"
 version = "0.13.0"
 dependencies = [
  "actix",
- "actix-rt",
  "assert2",
  "async-trait",
  "derive-with",
@@ -999,9 +995,9 @@ dependencies = [
  "serde_json",
  "test-log",
  "thiserror",
+ "time",
  "toml",
  "tracing",
- "tracing-subscriber",
  "ulid",
 ]
 
@@ -1011,9 +1007,6 @@ version = "0.13.0"
 dependencies = [
  "assert2",
  "async-trait",
- "base64 0.22.1",
- "bytes",
- "derive_more",
  "git-next-core",
  "kxio",
  "rand",
@@ -1021,9 +1014,7 @@ dependencies = [
  "serde",
  "serde_json",
  "tokio",
- "toml",
  "tracing",
- "ulid",
 ]
 
 [[package]]
@@ -1032,8 +1023,6 @@ version = "0.13.0"
 dependencies = [
  "assert2",
  "async-trait",
- "base64 0.22.1",
- "bytes",
  "clap",
  "derive_more",
  "git-next-core",
@@ -1046,9 +1035,7 @@ dependencies = [
  "serde_json",
  "sha2",
  "tokio",
- "toml",
  "tracing",
- "ulid",
 ]
 
 [[package]]

crates/cli/Cargo.toml

@@ -31,18 +31,12 @@ kxio = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 
-# git
-async-trait = { workspace = true }
-
 # Conventional Commit check
 git-conventional = { workspace = true }
 
 # TOML parsing
 toml = { workspace = true }
 
-# base64 decoding
-base64 = { workspace = true }
-
 # Actors
 actix = { workspace = true }
 actix-rt = { workspace = true }
@@ -54,7 +48,6 @@ anyhow = { workspace = true }
 thiserror = { workspace = true }
 
 # Webhooks
-serde = { workspace = true }
 serde_json = { workspace = true }
 ulid = { workspace = true }
 time = { workspace = true }

crates/cli/src/alerts/desktop.rs

@@ -1,8 +1,9 @@
 //
-use crate::alerts::{messages::NotifyUser, short_message};
+use crate::alerts::short_message;
+use git_next_core::git::UserNotification;
 
-pub(super) fn send_desktop_notification(msg: &NotifyUser) {
+pub(super) fn send_desktop_notification(user_notification: &UserNotification) {
-    let message = short_message(msg);
+    let message = short_message(user_notification);
     if let Err(err) = notifica::notify("git-next", &message) {
         tracing::warn!(?err, "failed to send desktop notification");
     }

crates/cli/src/alerts/email.rs

@@ -1,7 +1,10 @@
 //
-use git_next_core::server::{EmailConfig, SmtpConfig};
+use git_next_core::{
+    git::UserNotification,
+    server::{EmailConfig, SmtpConfig},
+};
 
-use crate::alerts::{full_message, messages::NotifyUser, short_message};
+use crate::alerts::{full_message, short_message};
 
 #[derive(Debug)]
 struct EmailMessage {
@@ -11,12 +14,12 @@ struct EmailMessage {
     body: String,
 }
 
-pub(super) fn send_email(msg: &NotifyUser, email_config: &EmailConfig) {
+pub(super) fn send_email(user_notification: &UserNotification, email_config: &EmailConfig) {
     let email_message = EmailMessage {
         from: email_config.from().to_string(),
         to: email_config.to().to_string(),
-        subject: short_message(msg),
+        subject: short_message(user_notification),
-        body: full_message(msg),
+        body: full_message(user_notification),
     };
     match email_config.smtp() {
         Some(smtp) => send_email_smtp(email_message, smtp),

crates/cli/src/alerts/handlers/notify_user.rs

@@ -1,5 +1,6 @@
 //
 use actix::prelude::*;
+
 use tracing::{info, instrument, Instrument as _};
 
 use crate::alerts::{
@@ -18,19 +19,21 @@ impl Handler<NotifyUser> for AlertsActor {
         };
         let net = self.net.clone();
         let shout = shout.clone();
+        if let Some(user_notification) = self.history.sendable(msg.unwrap()) {
             async move {
                 if let Some(webhook_config) = shout.webhook() {
-                send_webhook(&msg, webhook_config, &net).await;
+                    send_webhook(&user_notification, webhook_config, &net).await;
                 }
                 if let Some(email_config) = shout.email() {
-                send_email(&msg, email_config);
+                    send_email(&user_notification, email_config);
                 }
                 if shout.desktop() {
-                send_desktop_notification(&msg);
+                    send_desktop_notification(&user_notification);
                 }
             }
             .in_current_span()
             .into_actor(self)
             .wait(ctx);
         }
+    }
 }

crates/cli/src/alerts/history.rs

@@ -1,2 +1,47 @@
+//
+use git_next_core::git::UserNotification;
+
+use std::{
+    collections::HashMap,
+    time::{Duration, Instant},
+};
+
 #[derive(Debug, Default)]
-pub struct History {}
+pub struct History {
+    /// The maximum age of an item in the history.
+    ///
+    /// Items older than this will be dropped.
+    max_age_seconds: Duration,
+
+    /// Maps a user notification to when it was last seen.
+    ///
+    /// The user notification will not be sent until after max_age_seconds from last seen.
+    ///
+    /// Each time we see a given user notification, the last seen time will be updated.
+    items: HashMap<UserNotification, Instant>,
+}
+impl History {
+    pub fn new(max_age_seconds: Duration) -> Self {
+        Self {
+            max_age_seconds,
+            items: HashMap::default(),
+        }
+    }
+
+    pub fn sendable(&mut self, user_notification: UserNotification) -> Option<UserNotification> {
+        let now = Instant::now();
+        self.prune(&now); // remove expired items first
+        let contains_key = self.items.contains_key(&user_notification);
+        self.items.insert(user_notification.clone(), now);
+        if contains_key {
+            return None;
+        }
+        Some(user_notification)
+    }
+
+    pub fn prune(&mut self, now: &Instant) {
+        if let Some(threshold) = now.checked_sub(self.max_age_seconds) {
+            self.items.retain(|_, last_seen| *last_seen > threshold)
+        };
+    }
+}

crates/cli/src/alerts/messages.rs

@@ -1,86 +1,6 @@
 //
-use derive_more::Deref as _;
 use git_next_core::{git::UserNotification, message, server::Shout};
-use serde_json::json;
 
 message!(UpdateShout: Shout: "Updated Shout configuration");
 
 message!(NotifyUser: UserNotification: "Request to send the message payload to the notification webhook");
-impl NotifyUser {
-    pub fn as_json(&self, timestamp: time::OffsetDateTime) -> serde_json::Value {
-        let timestamp = timestamp.unix_timestamp().to_string();
-        match self.deref() {
-            UserNotification::CICheckFailed {
-                forge_alias,
-                repo_alias,
-                commit,
-            } => json!({
-                "type": "cicheck.failed",
-                "timestamp": timestamp,
-                "data": {
-                    "forge_alias": forge_alias,
-                    "repo_alias": repo_alias,
-                    "commit": {
-                        "sha": commit.sha(),
-                        "message": commit.message()
-                    }
-                }
-            }),
-            UserNotification::RepoConfigLoadFailure {
-                forge_alias,
-                repo_alias,
-                reason,
-            } => json!({
-                "type": "config.load.failed",
-                "timestamp": timestamp,
-                "data": {
-                    "forge_alias": forge_alias,
-                    "repo_alias": repo_alias,
-                    "reason": reason
-                }
-            }),
-            UserNotification::WebhookRegistration {
-                forge_alias,
-                repo_alias,
-                reason,
-            } => json!({
-                "type": "webhook.registration.failed",
-                "timestamp": timestamp,
-                "data": {
-                    "forge_alias": forge_alias,
-                    "repo_alias": repo_alias,
-                    "reason": reason
-                }
-            }),
-            UserNotification::DevNotBasedOnMain {
-                forge_alias,
-                repo_alias,
-                dev_branch,
-                main_branch,
-                dev_commit,
-                main_commit,
-            } => json!({
-                "type": "branch.dev.not-on-main",
-                "timestamp": timestamp,
-                "data": {
-                    "forge_alias": forge_alias,
-                    "repo_alias": repo_alias,
-                    "branches": {
-                        "dev": dev_branch,
-                        "main": main_branch
-                    },
-                    "commits": {
-                        "dev": {
-                            "sha": dev_commit.sha(),
-                            "message": dev_commit.message()
-                        },
-                        "main": {
-                            "sha": main_commit.sha(),
-                            "message": main_commit.message()
-                        }
-                    }
-                }
-            }),
-        }
-    }
-}

crates/cli/src/alerts/mod.rs

@@ -1,5 +1,3 @@
-use std::ops::Deref as _;
-
 //
 use actix::prelude::*;
 
@@ -8,7 +6,6 @@ use derive_more::derive::Constructor;
 use git_next_core::{git::UserNotification, server::Shout};
 
 pub use history::History;
-use messages::NotifyUser;
 
 mod desktop;
 mod email;
@@ -17,10 +14,12 @@ mod history;
 pub mod messages;
 mod webhook;
 
+#[cfg(test)]
+mod tests;
+
 #[derive(Debug, Constructor)]
 pub struct AlertsActor {
     shout: Option<Shout>, // config for sending alerts to users
-    #[allow(dead_code)] // TODO (#128) Prevent duplicate user notifications
     history: History,     // record of alerts sent recently (e.g. 24 hours)
     net: kxio::network::Network,
 }
@@ -29,8 +28,8 @@ impl Actor for AlertsActor {
     type Context = Context<Self>;
 }
 
-fn short_message(msg: &NotifyUser) -> String {
+fn short_message(user_notification: &UserNotification) -> String {
-    let tail = match msg.deref() {
+    let tail = match user_notification {
         UserNotification::CICheckFailed {
             forge_alias,
             repo_alias,
@@ -58,8 +57,8 @@ fn short_message(msg: &NotifyUser) -> String {
     format!("[git-next] {tail}")
 }
 
-fn full_message(msg: &NotifyUser) -> String {
+fn full_message(user_notification: &UserNotification) -> String {
-    match msg.deref() {
+    match user_notification {
         UserNotification::CICheckFailed {
             forge_alias,
             repo_alias,

crates/cli/src/alerts/tests/history.rs

@@ -0,0 +1,66 @@
+use std::time::Duration;
+
+use assert2::let_assert;
+use git_next_core::git::UserNotification;
+
+use crate::{alerts::History, repo::tests::given};
+
+#[test]
+fn when_history_is_empty_then_message_is_passed() {
+    let mut history = History::new(Duration::from_millis(1));
+
+    let user_notification = UserNotification::RepoConfigLoadFailure {
+        forge_alias: given::a_forge_alias(),
+        repo_alias: given::a_repo_alias(),
+        reason: given::a_name(),
+    };
+
+    let result = history.sendable(user_notification);
+
+    assert!(result.is_some());
+}
+
+#[test]
+fn when_history_has_expired_then_message_is_passed() {
+    let dur = Duration::from_millis(1);
+    let mut history = History::new(dur);
+
+    let user_notification = UserNotification::RepoConfigLoadFailure {
+        forge_alias: given::a_forge_alias(),
+        repo_alias: given::a_repo_alias(),
+        reason: given::a_name(),
+    };
+
+    // add message to history
+    let result = history.sendable(user_notification);
+    let_assert!(Some(user_notification) = result);
+
+    std::thread::sleep(dur);
+
+    // after dur, message has expired, so is still valid
+    let result = history.sendable(user_notification);
+    assert!(result.is_some());
+}
+
+#[test]
+fn when_history_has_unexpired_then_message_is_blocked() {
+    let dur = Duration::from_millis(1);
+    let mut history = History::new(dur);
+
+    let user_notification = UserNotification::RepoConfigLoadFailure {
+        forge_alias: given::a_forge_alias(),
+        repo_alias: given::a_repo_alias(),
+        reason: given::a_name(),
+    };
+
+    // add message to history
+    let result = history.sendable(user_notification);
+    let_assert!(Some(user_notification) = result);
+
+    // no time passed
+    // std::thread::sleep(dur);
+
+    // after dur, message has expired, so is still valid
+    let result = history.sendable(user_notification);
+    assert!(result.is_none());
+}

crates/cli/src/alerts/tests/mod.rs

@@ -0,0 +1 @@
+mod history;

crates/cli/src/alerts/webhook.rs

@@ -1,12 +1,10 @@
 //
-use git_next_core::server::OutboundWebhook;
+use git_next_core::{git::UserNotification, server::OutboundWebhook};
 use secrecy::ExposeSecret as _;
 use standardwebhooks::Webhook;
 
-use crate::alerts::messages::NotifyUser;
-
 pub(super) async fn send_webhook(
-    msg: &NotifyUser,
+    user_notification: &UserNotification,
     webhook_config: &OutboundWebhook,
     net: &kxio::network::Network,
 ) {
@@ -16,18 +14,18 @@ pub(super) async fn send_webhook(
         tracing::warn!("Invalid notification configuration (signer) - can't sent notification");
         return;
     };
-    do_send_webhook(msg, webhook, webhook_config, net).await;
+    do_send_webhook(user_notification, webhook, webhook_config, net).await;
 }
 
 async fn do_send_webhook(
-    msg: &NotifyUser,
+    user_notification: &UserNotification,
     webhook: Webhook,
     webhook_config: &OutboundWebhook,
     net: &kxio::network::Network,
 ) {
     let message_id = format!("msg_{}", ulid::Ulid::new());
     let timestamp = time::OffsetDateTime::now_utc();
-    let payload = msg.as_json(timestamp);
+    let payload = user_notification.as_json(timestamp);
     let timestamp = timestamp.unix_timestamp();
     let to_sign = format!("{message_id}.{timestamp}.{payload}");
     tracing::info!(?to_sign, "");

crates/cli/src/repo/mod.rs

@@ -24,7 +24,7 @@ pub mod messages;
 mod notifications;
 
 #[cfg(test)]
-mod tests;
+pub mod tests;
 
 #[derive(Clone, Debug, Default)]
 pub struct RepoActorLog(std::sync::Arc<std::sync::RwLock<Vec<String>>>);

crates/core/Cargo.toml

@@ -14,14 +14,12 @@ github = []
 [dependencies]
 # logging
 tracing = { workspace = true }
-tracing-subscriber = { workspace = true }
 
 # fs/network
 kxio = { workspace = true }
 
 # Actors
 actix = { workspace = true }
-actix-rt = { workspace = true }
 
 # TOML parsing
 serde = { workspace = true }
@@ -37,6 +35,7 @@ async-trait = { workspace = true }
 
 # Webhooks
 ulid = { workspace = true }
+time = { workspace = true }
 
 # boilerplate
 derive_more = { workspace = true }

crates/core/src/config/branch_name.rs

@@ -1,3 +1,4 @@
+use derive_more::derive::Display;
 use serde::Serialize;
 
-crate::newtype!(BranchName: String, derive_more::Display, Default, Serialize: "The name of a Git branch");
+crate::newtype!(BranchName: String, Display, Default, Hash, Serialize: "The name of a Git branch");

crates/core/src/config/repo_alias.rs

@@ -3,6 +3,6 @@ use serde::Serialize;
 
 use crate::newtype;
 
-newtype!(RepoAlias: String, Display, Default, PartialOrd, Ord, Serialize: r#"The alias of a repo.
+newtype!(RepoAlias: String, Display, Default, Hash, PartialOrd, Ord, Serialize: r#"The alias of a repo.
 
 This is the alias for the repo within `git-next-server.toml`."#);

crates/core/src/git/user_notification.rs

@@ -1,7 +1,8 @@
 //
 use crate::{git::Commit, BranchName, ForgeAlias, RepoAlias};
+use serde_json::json;
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
 pub enum UserNotification {
     CICheckFailed {
         forge_alias: ForgeAlias,
@@ -27,3 +28,81 @@ pub enum UserNotification {
         main_commit: Commit,
     },
 }
+impl UserNotification {
+    pub fn as_json(&self, timestamp: time::OffsetDateTime) -> serde_json::Value {
+        let timestamp = timestamp.unix_timestamp().to_string();
+        match self {
+            Self::CICheckFailed {
+                forge_alias,
+                repo_alias,
+                commit,
+            } => json!({
+                "type": "cicheck.failed",
+                "timestamp": timestamp,
+                "data": {
+                    "forge_alias": forge_alias,
+                    "repo_alias": repo_alias,
+                    "commit": {
+                        "sha": commit.sha(),
+                        "message": commit.message()
+                    }
+                }
+            }),
+            Self::RepoConfigLoadFailure {
+                forge_alias,
+                repo_alias,
+                reason,
+            } => json!({
+                "type": "config.load.failed",
+                "timestamp": timestamp,
+                "data": {
+                    "forge_alias": forge_alias,
+                    "repo_alias": repo_alias,
+                    "reason": reason
+                }
+            }),
+            Self::WebhookRegistration {
+                forge_alias,
+                repo_alias,
+                reason,
+            } => json!({
+                "type": "webhook.registration.failed",
+                "timestamp": timestamp,
+                "data": {
+                    "forge_alias": forge_alias,
+                    "repo_alias": repo_alias,
+                    "reason": reason
+                }
+            }),
+            Self::DevNotBasedOnMain {
+                forge_alias,
+                repo_alias,
+                dev_branch,
+                main_branch,
+                dev_commit,
+                main_commit,
+            } => json!({
+                "type": "branch.dev.not-on-main",
+                "timestamp": timestamp,
+                "data": {
+                    "forge_alias": forge_alias,
+                    "repo_alias": repo_alias,
+                    "branches": {
+                        "dev": dev_branch,
+                        "main": main_branch
+                    },
+                    "commits": {
+                        "dev": {
+                            "sha": dev_commit.sha(),
+                            "message": dev_commit.message()
+                        },
+                        "main": {
+                            "sha": main_commit.sha(),
+                            "message": main_commit.message()
+                        }
+                    }
+                }
+            }),
+        }
+    }
+}

crates/forge-forgejo/Cargo.toml

@@ -12,9 +12,6 @@ git-next-core = { workspace = true }
 # logging
 tracing = { workspace = true }
 
-# base64 decoding
-base64 = { workspace = true }
-
 # git
 async-trait = { workspace = true }
 
@@ -24,18 +21,10 @@ kxio = { workspace = true }
 # TOML parsing
 serde = { workspace = true }
 serde_json = { workspace = true }
-toml = { workspace = true }
 
 # Secrets and Password
 secrecy = { workspace = true }
 
-# Webhooks
-bytes = { workspace = true }
-ulid = { workspace = true }
-
-# boilerplate
-derive_more = { workspace = true }
-
 # # Actors
 tokio = { workspace = true }
 

crates/forge-github/Cargo.toml

@@ -20,9 +20,6 @@ hmac = { workspace = true }
 sha2 = { workspace = true }
 hex = { workspace = true }
 
-# base64 decoding
-base64 = { workspace = true }
-
 # git
 async-trait = { workspace = true }
 
@@ -32,15 +29,10 @@ kxio = { workspace = true }
 # TOML parsing
 serde = { workspace = true }
 serde_json = { workspace = true }
-toml = { workspace = true }
 
 # Secrets and Password
 secrecy = { workspace = true }
 
-# Webhooks
-bytes = { workspace = true }
-ulid = { workspace = true }
-
 # boilerplate
 derive_more = { workspace = true }