feat: debounce notifications

Author: link2xt

src/debouncer.rs

@@ -0,0 +1,81 @@
+//! # Debouncer for notifications.
+//!
+//! Sometimes the client application may be reinstalled
+//! while keeping the notification token.
+//! In this case the same token is stored twice
+//! for the same mailbox on a chatmail relay
+//! and is notified twice for the same message.
+//! Since it is not possible for the chatmail relay
+//! to deduplicate the tokens in this case
+//! as only the notification gateway
+//! can decrypt them, notification gateway needs
+//! to debounce notifications to the same token.
+
+use std::time::{Instant, Duration};
+use std::collections::{HashSet, BinaryHeap};
+use std::cmp::Reverse;
+use std::sync::Mutex;
+
+#[derive(Default)]
+pub(crate) struct Debouncer {
+    state: Mutex<DebouncerState>
+}
+
+#[derive(Default)]
+struct DebouncerState {
+    /// Set of recently notified tokens.
+    ///
+    /// The tokens are stored in plaintext,
+    /// not hashed or encrypted.
+    /// No token is stored for a long time anyway.
+    tokens: HashSet<String>,
+
+    /// Binary heap storing tokens
+    /// sorted by the timestamp of the recent notifications.
+    ///
+    /// `Reverse` is used to turn max-heap into min-heap.
+    heap: BinaryHeap<Reverse<(Instant, String)>>,
+}
+
+impl DebouncerState {
+    /// Removes old entries for tokens that can be notified again.
+    fn cleanup(&mut self) {
+        let now = Instant::now();
+        loop {
+            let Some(Reverse((timestamp, token))) = self.heap.pop() else {
+                break;
+            };
+
+            if timestamp.duration_since(now) < Duration::from_secs(10) {
+                self.heap.push(Reverse((timestamp, token)));
+                break;
+            }
+
+            self.tokens.remove(&token);
+        }
+    }
+
+    /// Returns true if the token was notified recently
+    /// and should not be notified again.
+    fn is_debounced(&mut self, token: &String) -> bool {
+        self.cleanup();
+        self.tokens.contains(token)
+    }
+
+    fn notify(&mut self, token: String) {
+        if self.tokens.insert(token.clone()) {
+            self.heap.push(Reverse((Instant::now(), token)));
+        }
+    }
+}
+
+impl Debouncer {
+    pub(crate) fn is_debounced(&self, token: &String) -> bool {
+        let mut state = self.state.lock().unwrap();
+        state.is_debounced(token)
+    }
+
+    pub(crate) fn notify(&self, token: String) {
+        self.state.lock().unwrap().notify(token);
+    }
+}

src/lib.rs

@@ -1,3 +1,4 @@
+mod debouncer;
 pub mod metrics;
 pub mod notifier;
 mod openpgp;

src/server.rs

@@ -293,6 +293,10 @@ async fn notify_device(
     }
 
     info!("Got direct notification for {device_token}.");
+    if state.debouncer().is_debounced(&device_token) {
+        return Ok(StatusCode::OK);
+    }
+    state.debouncer().notify(device_token.clone());
     let device_token: NotificationToken = device_token.as_str().parse()?;
 
     let status_code = match device_token {

src/state.rs

@@ -9,6 +9,7 @@ use anyhow::{Context as _, Result};
 use crate::metrics::Metrics;
 use crate::openpgp::PgpDecryptor;
 use crate::schedule::Schedule;
+use crate::debouncer::Debouncer;
 
 #[derive(Clone)]
 pub struct State {
@@ -36,6 +37,8 @@ pub struct InnerState {
     /// Decryptor for incoming tokens
     /// storing the secret keyring inside.
     openpgp_decryptor: PgpDecryptor,
+
+    debouncer: Debouncer,
 }
 
 impl State {
@@ -88,6 +91,7 @@ impl State {
                 interval,
                 fcm_authenticator,
                 openpgp_decryptor,
+                debouncer: Default::default(),
             }),
         })
     }
@@ -134,4 +138,8 @@ impl State {
     pub fn openpgp_decryptor(&self) -> &PgpDecryptor {
         &self.inner.openpgp_decryptor
     }
+
+    pub(crate) fn debouncer(&self) -> &Debouncer {
+        &self.inner.debouncer
+    }
 }