feat: complete (mostly) the WebPushClient

SYNC-3688

Cargo.toml

@@ -53,6 +53,7 @@ httparse = "1.3"
 hyper = "0.14"
 lazy_static = "1.4"
 log = { version = "0.4", features = ["max_level_debug", "release_max_level_info"] }
+mockall = "0.8.3"  # 0.9+ requires reworking tests
 mozsvc-common = "0.2"
 openssl = "0.10"
 rand = "0.8"

autoconnect/autoconnect-common/src/test_support.rs

@@ -1,6 +1,6 @@
 use uuid::Uuid;
 
-use autopush_common::db::{mock::MockDbClient, HelloResponse};
+use autopush_common::db::{mock::MockDbClient, User};
 
 pub const UA: &str =
     "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:109.0) Gecko/20100101 Firefox/110.0";
@@ -15,20 +15,37 @@ pub const HELLO: &str = r#"{"messageType": "hello", "use_webpush": true}"#;
 pub const HELLO_AGAIN: &str = r#"{"messageType": "hello", "use_webpush": true,
                                   "uaid": "deadbeef-0000-0000-deca-fbad00000000"}"#;
 
+pub const CURRENT_MONTH: &str = "message_2023_5";
+
 /// Return a simple MockDbClient that responds to hello (once) with a new uaid.
 pub fn hello_db() -> MockDbClient {
-    hello_again_db(uuid::Uuid::new_v4())
+    let mut db = MockDbClient::new();
+    db.expect_message_table()
+        .times(1)
+        .return_const(CURRENT_MONTH.to_owned());
+    db
 }
 
 /// Return a simple MockDbClient that responds to hello (once) with the
 /// specified uaid.
 pub fn hello_again_db(uaid: Uuid) -> MockDbClient {
     let mut db = MockDbClient::new();
-    db.expect_hello().times(1).return_once(move |_, _, _, _| {
-        Ok(HelloResponse {
-            uaid: Some(uaid),
+    db.expect_get_user().times(1).return_once(move |_| {
+        Ok(Some(User {
+            uaid,
+            current_month: Some(CURRENT_MONTH.to_owned()),
             ..Default::default()
-        })
+        }))
     });
+    db.expect_message_table()
+        .times(1)
+        .return_const(CURRENT_MONTH.to_owned());
+    db.expect_update_user().times(1).return_once(|_| Ok(()));
+    db.expect_fetch_messages()
+        .times(1)
+        .return_once(|_, _| Ok(Default::default()));
+    db.expect_fetch_timestamp_messages()
+        .times(1)
+        .return_once(|_, _, _| Ok(Default::default()));
     db
 }

autoconnect/autoconnect-settings/Cargo.toml

@@ -10,6 +10,7 @@ config.workspace = true
 fernet.workspace = true
 lazy_static.workspace = true
 mozsvc-common.workspace = true
+reqwest.workspace = true
 serde.workspace = true
 serde_derive.workspace = true
 slog.workspace = true

autoconnect/autoconnect-settings/src/app_state.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
 use cadence::StatsdClient;
 use fernet::{Fernet, MultiFernet};
@@ -17,6 +17,7 @@ pub struct AppState {
     /// Handle to the data storage object
     pub db: Box<dyn DbClient>,
     pub metrics: Arc<StatsdClient>,
+    pub http: reqwest::Client,
 
     /// Encryption object for the endpoint URL
     pub fernet: MultiFernet,
@@ -63,13 +64,18 @@ impl AppState {
             StorageType::DynamoDb => Box::new(DdbClientImpl::new(metrics.clone(), &db_settings)?),
             StorageType::INVALID => panic!("Invalid Storage type. Check {}_DB_DSN.", ENV_PREFIX),
         };
+        let http = reqwest::Client::builder()
+            .timeout(Duration::from_secs(1))
+            .build()
+            .unwrap_or_else(|e| panic!("Error while building reqwest::Client: {}", e));
 
         let router_url = settings.router_url();
         let endpoint_url = settings.endpoint_url();
 
         Ok(Self {
             db,
             metrics,
+            http,
             fernet,
             clients: Arc::new(ClientRegistry::default()),
             settings,

autoconnect/autoconnect-web/src/lib.rs

@@ -8,13 +8,12 @@ extern crate slog_scope;
 pub mod client;
 pub mod dockerflow;
 pub mod metrics;
+pub mod routes;
 #[cfg(test)]
 mod test;
 
 use actix_web::web;
 
-use autoconnect_ws::ws_handler;
-
 /// Requires import of the `config` function also in this module to use.
 #[macro_export]
 macro_rules! build_app {
@@ -38,12 +37,14 @@ macro_rules! build_app {
 pub fn config(cfg: &mut web::ServiceConfig) {
     cfg
         // Websocket Handler
-        .route("/", web::get().to(ws_handler))
-        // TODO: Internode Message handler
-        //.service(web::resource("/push/{uaid}").route(web::push().to(autoconnect_web::route::InterNode::put))
+        .route("/", web::get().to(autoconnect_ws::ws_handler))
+        .service(web::resource("/push/{uaid}").route(web::put().to(crate::routes::push_route)))
+        .service(
+            web::resource("/notif/{uaid}").route(web::put().to(crate::routes::check_storage_route)),
+        )
         .service(web::resource("/status").route(web::get().to(dockerflow::status_route)))
         .service(web::resource("/health").route(web::get().to(dockerflow::health_route)))
-        .service(web::resource("/v1/err").route(web::get().to(dockerflow::log_check)))
+        .service(web::resource("/v1/err/crit").route(web::get().to(dockerflow::log_check)))
         // standardized
         .service(web::resource("/__error__").route(web::get().to(dockerflow::log_check)))
         // Dockerflow

autoconnect/autoconnect-web/src/routes.rs

@@ -0,0 +1,42 @@
+/// Handle incoming notifications from autoendpoint
+use actix_web::{web, HttpResponse};
+use uuid::Uuid;
+
+use autoconnect_settings::AppState;
+use autopush_common::notification::Notification;
+
+/// Deliver a Push notification directly to a connected client
+pub async fn push_route(
+    uaid: web::Path<Uuid>,
+    notif: web::Json<Notification>,
+    state: web::Data<AppState>,
+) -> HttpResponse {
+    trace!(
+        "push_route, uaid: {} channel_id: {}",
+        uaid,
+        notif.channel_id
+    );
+    let result = state
+        .clients
+        .notify(uaid.into_inner(), notif.into_inner())
+        .await;
+    if result.is_ok() {
+        HttpResponse::Ok().finish()
+    } else {
+        HttpResponse::NotFound().body("Client not available")
+    }
+}
+
+/// Notify a connected client to check storage for new notifications
+pub async fn check_storage_route(
+    uaid: web::Path<Uuid>,
+    state: web::Data<AppState>,
+) -> HttpResponse {
+    trace!("check_storage_route, uaid: {}", uaid);
+    let result = state.clients.check_storage(uaid.into_inner()).await;
+    if result.is_ok() {
+        HttpResponse::Ok().finish()
+    } else {
+        HttpResponse::NotFound().body("Client not available")
+    }
+}

autoconnect/autoconnect-web/src/test.rs

@@ -124,6 +124,7 @@ pub async fn malformed_webpush_message() {
         panic!("Expected Close(Some(..)) not {:#?}", item);
     };
     assert_eq!(close_reason.code, actix_http::ws::CloseCode::Error);
+    assert_eq!(close_reason.description.unwrap(), "Json");
     assert!(framed.next().await.is_none());
 }