feat: Add redis based reliability reporting

.circleci/config.yml

@@ -165,6 +165,10 @@ jobs:
           username: $DOCKER_USER
           password: $DOCKER_PASS
         command: gcloud beta emulators bigtable start --host-port=localhost:8086
+      # - image: redis/redis-stack-server
+      #  auth:
+      #    username: $DOCKER_USER
+      #    password: $DOCKER_PASS
     resource_class: large
     environment:
       BIGTABLE_EMULATOR_HOST: localhost:8086

autoconnect/Cargo.toml

@@ -57,3 +57,9 @@ default = ["bigtable"]
 bigtable = ["autopush_common/bigtable", "autoconnect_settings/bigtable"]
 emulator = ["bigtable"]
 log_vapid = []
+reliable_report = [
+    "autoconnect_settings/reliable_report",
+    "autoconnect_web/reliable_report",
+    "autoconnect_ws/reliable_report",
+    "autopush_common/reliable_report",
+]

autoconnect/autoconnect-common/src/broadcast.rs

@@ -199,11 +199,11 @@ impl BroadcastChangeTracker {
             }
             *ver = broadcast.version;
         } else {
-            trace!("📢 Not found: {}", &b_id);
+            trace!("📢 Not found: {b_id}");
             return Err(ApcErrorKind::BroadcastError("Broadcast not found".into()).into());
         }
 
-        trace!("📢 New version of {}", &b_id);
+        trace!("📢 New version of {b_id}");
         // Check to see if this broadcast has been updated since initialization
         let bcast_index = self
             .broadcast_list

autoconnect/autoconnect-common/src/protocol.rs

@@ -90,6 +90,8 @@ pub struct ClientAck {
     pub channel_id: Uuid,
     // The corresponding version number for the message.
     pub version: String,
+    #[serde(default)]
+    pub reliability_id: Option<String>,
 }
 
 #[derive(Debug, Serialize)]

autoconnect/autoconnect-settings/Cargo.toml

@@ -25,3 +25,4 @@ autopush_common.workspace = true
 # specify the default via the calling crate, in order to simplify default chains.
 bigtable = ["autopush_common/bigtable"]
 emulator = ["bigtable"]
+reliable_report = ["autopush_common/reliable_report"]

autoconnect/autoconnect-settings/src/app_state.rs

@@ -12,6 +12,8 @@ use autoconnect_common::{
     registry::ClientRegistry,
 };
 use autopush_common::db::{client::DbClient, DbSettings, StorageType};
+#[cfg(feature = "reliable_report")]
+use autopush_common::reliability::PushReliability;
 
 use crate::{Settings, ENV_PREFIX};
 
@@ -32,6 +34,9 @@ pub struct AppState {
     pub settings: Settings,
     pub router_url: String,
     pub endpoint_url: String,
+
+    #[cfg(feature = "reliable_report")]
+    pub reliability: Arc<PushReliability>,
 }
 
 impl AppState {
@@ -84,6 +89,13 @@ impl AppState {
                 ENV_PREFIX.to_uppercase()
             ),
         };
+
+        #[cfg(feature = "reliable_report")]
+        let reliability = Arc::new(
+            PushReliability::new(&settings.reliability_dsn, &Some(db.clone())).map_err(|e| {
+                ConfigError::Message(format!("Could not start Reliability connection: {:?}", e))
+            })?,
+        );
         let http = reqwest::Client::builder()
             .timeout(Duration::from_secs(1))
             .build()
@@ -103,6 +115,8 @@ impl AppState {
             settings,
             router_url,
             endpoint_url,
+            #[cfg(feature = "reliable_report")]
+            reliability,
         })
     }
 

autoconnect/autoconnect-settings/src/lib.rs

@@ -108,6 +108,8 @@ pub struct Settings {
     ///
     /// By default, the number of available physical CPUs is used as the worker count.
     pub actix_workers: Option<usize>,
+    #[cfg(feature = "reliable_report")]
+    pub reliability_dsn: Option<String>,
 }
 
 impl Default for Settings {
@@ -139,6 +141,8 @@ impl Default for Settings {
             msg_limit: 150,
             actix_max_connections: None,
             actix_workers: None,
+            #[cfg(feature = "reliable_report")]
+            reliability_dsn: None,
         }
     }
 }

autoconnect/autoconnect-web/Cargo.toml

@@ -36,3 +36,7 @@ tokio.workspace = true
 autoconnect_common = { workspace = true, features = ["test-support"] }
 
 [features]
+reliable_report = [
+    "autopush_common/reliable_report",
+    "autoconnect_ws/reliable_report",
+]

autoconnect/autoconnect-web/src/routes.rs

@@ -21,16 +21,59 @@ pub async fn push_route(
     notif: web::Json<Notification>,
     app_state: web::Data<AppState>,
 ) -> HttpResponse {
+    #[allow(unused_mut)] // Needed for "reliable_report"
+    let mut notif = notif.into_inner();
+    #[cfg(feature = "reliable_report")]
+    {
+        notif.reliable_state = app_state
+            .reliability
+            .record(
+                &notif.reliability_id,
+                autopush_common::reliability::PushReliabilityState::IntAccepted,
+                &notif.reliable_state,
+                Some(notif.timestamp + notif.ttl),
+            )
+            .await;
+    }
     trace!(
-        "⏩ push_route, uaid: {} channel_id: {}",
+        "⏩ in push_route, uaid: {} channel_id: {}",
         uaid,
-        notif.channel_id
+        notif.channel_id,
     );
+    #[cfg(feature = "reliable_report")]
+    let expiry = {
+        // Set "transmitted" a bit early since we can't do this inside of `notify`.
+        notif.reliable_state = app_state
+            .reliability
+            .record(
+                &notif.reliability_id,
+                autopush_common::reliability::PushReliabilityState::Transmitted,
+                &notif.reliable_state,
+                Some(notif.timestamp + notif.ttl),
+            )
+            .await;
+        Some(notif.timestamp + notif.ttl)
+    };
+    // Attempt to send the notification to the UA using WebSocket protocol, or store on failure.
     let result = app_state
         .clients
-        .notify(uaid.into_inner(), notif.into_inner())
+        .notify(uaid.into_inner(), notif.clone())
         .await;
     if result.is_ok() {
+        #[cfg(feature = "reliable_report")]
+        {
+            // Set "transmitted" a bit early since we can't do this inside of `notify`.
+            notif.reliable_state = app_state
+                .reliability
+                .record(
+                    &notif.reliability_id,
+                    autopush_common::reliability::PushReliabilityState::Accepted,
+                    &notif.reliable_state,
+                    expiry,
+                )
+                .await;
+        }
+
         HttpResponse::Ok().finish()
     } else {
         HttpResponse::NotFound().body("Client not available")

autoconnect/autoconnect-ws/Cargo.toml

@@ -35,3 +35,7 @@ ctor.workspace = true
 autoconnect_common = { workspace = true, features = ["test-support"] }
 
 [features]
+reliable_report = [
+    "autopush_common/reliable_report",
+    "autoconnect_ws_sm/reliable_report",
+]

autoconnect/autoconnect-ws/autoconnect-ws-sm/Cargo.toml

@@ -32,3 +32,4 @@ serde_json.workspace = true
 autoconnect_common = { workspace = true, features = ["test-support"] }
 
 [features]
+reliable_report = []

autoconnect/autoconnect-ws/autoconnect-ws-sm/src/identified/mod.rs

@@ -138,6 +138,11 @@ impl WebPushClient {
         &self.app_state.settings
     }
 
+    #[cfg(feature = "reliable_report")]
+    pub fn app_reliability(&self) -> &autopush_common::reliability::PushReliability {
+        &self.app_state.reliability
+    }
+
     /// Connect this `WebPushClient` to the `ClientRegistry`
     ///
     /// Returning a `Stream` of `ServerNotification`s from the `ClientRegistry`

autoconnect/autoconnect-ws/autoconnect-ws-sm/src/identified/on_server_notif.rs

@@ -163,6 +163,31 @@ impl WebPushClient {
         Ok(smsgs)
     }
 
+    #[cfg(feature = "reliable_report")]
+    /// Record and transition the state for trackable messages.
+    async fn record_state(
+        &self,
+        messages: &mut Vec<Notification>,
+        state: autopush_common::reliability::PushReliabilityState,
+    ) {
+        // *Note* because `.map()` is sync
+        // we can't call the async func without additional hoops.
+        // I'm guessing that there's a more elegant way to do this, but this works for now.
+        for message in messages {
+            let expiry = message.timestamp + message.ttl;
+            message.reliable_state = self
+                .app_state
+                .reliability
+                .record(
+                    &message.reliability_id,
+                    state,
+                    &message.reliable_state,
+                    Some(expiry),
+                )
+                .await;
+        }
+    }
+
     /// Read a chunk (max count 10 returned) of Notifications from storage
     ///
     /// This alternates between reading Topic Notifications and Timestamp
@@ -186,10 +211,20 @@ impl WebPushClient {
         let topic_resp = if self.flags.include_topic {
             trace!("🗄️ WebPushClient::do_check_storage: fetch_topic_messages");
             // Get the most recent max 11 messages.
-            self.app_state
+            #[allow(unused_mut)]
+            let mut messages = self
+                .app_state
                 .db
                 .fetch_topic_messages(&self.uaid, 11)
-                .await?
+                .await?;
+            #[cfg(feature = "reliable_report")]
+            // Since we pulled these from storage, mark them as "retrieved"
+            self.record_state(
+                &mut messages.messages,
+                autopush_common::reliability::PushReliabilityState::Retreived,
+            )
+            .await;
+            messages
         } else {
             Default::default()
         };
@@ -226,7 +261,8 @@ impl WebPushClient {
             "🗄️ WebPushClient::do_check_storage: fetch_timestamp_messages timestamp: {:?}",
             timestamp
         );
-        let timestamp_resp = self
+        #[allow(unused_mut)]
+        let mut timestamp_resp = self
             .app_state
             .db
             .fetch_timestamp_messages(&self.uaid, timestamp, 10)
@@ -244,6 +280,13 @@ impl WebPushClient {
                 )
                 .with_tag("topic", "false")
                 .send();
+            #[cfg(feature = "reliable_report")]
+            // Since we pulled these from storage, mark them as "retrieved"
+            self.record_state(
+                &mut timestamp_resp.messages,
+                autopush_common::reliability::PushReliabilityState::Retreived,
+            )
+            .await;
         }
 
         Ok(CheckStorageResponse {

autoendpoint/Cargo.toml

@@ -82,3 +82,5 @@ emulator = ["bigtable"]
 stub = []
 # Verbosely log vapid assertions (NOT ADVISED FOR WIDE PRODUCTION USE)
 log_vapid = []
+
+reliable_report = ["autopush_common/reliable_report"]