Test receiving push events when backgrounded

And thus also displaying notifications.

6 files changed, 88 insertions, 25 deletions

diff --git a/src/push/app.rs b/src/push/app.rs
index 2d6e15c..ed8bf31 100644
--- a/src/push/app.rs
+++ b/src/push/app.rs
@@ -46,18 +46,42 @@ impl<'a> Push<'a> {
         Ok(id)
     }
 
+    pub async fn broadcast(
+        &self,
+        message: &str,
+    ) -> Result<(), EchoError> {
+        let mut tx = self.db.begin().await?;
+        let subscriptions = tx
+            .subscriptions()
+            .all()
+            .await?;
+
+        tx.commit().await?;
+
+        for subscription in subscriptions {
+            // We don't care if any of these error, for now.
+            // Eventually, we should remove rows that cause certain error conditions.
+            println!("Sending to {:#?}", subscription.info.endpoint);
+            self.send(&subscription.info, message).await.unwrap_or_else(|err| {
+                println!("Error with {:#?}: {}", subscription.info.endpoint, err);
+            })
+        }
+
+        Ok(())
+    }
+
     pub async fn echo(
         &self,
         user: &User,
-        subscription: &Id,
+        endpoint: &String,
         message: &str,
     ) -> Result<(), EchoError> {
         let mut tx = self.db.begin().await?;
         let subscription = tx
             .subscriptions()
-            .by_id(subscription)
+            .by_endpoint(endpoint)
             .await
-            .not_found(|| EchoError::NotFound(subscription.clone()))?;
+            .not_found(|| EchoError::NotFound(endpoint.clone()))?;
         if subscription.user != user.id {
             return Err(EchoError::NotSubscriber(subscription.id, user.id.clone()));
         }
@@ -89,13 +113,13 @@ impl<'a> Push<'a> {
         Ok(())
     }
 
-    pub async fn unregister(&self, user: &User, subscription: &Id) -> Result<(), UnregisterError> {
+    pub async fn unregister(&self, user: &User, endpoint: &String) -> Result<(), UnregisterError> {
         let mut tx = self.db.begin().await?;
         let subscription = tx
             .subscriptions()
-            .by_id(subscription)
+            .by_endpoint(endpoint)
             .await
-            .not_found(|| UnregisterError::NotFound(subscription.clone()))?;
+            .not_found(|| UnregisterError::NotFound(endpoint.clone()))?;
         if subscription.user != user.id {
             return Err(UnregisterError::NotSubscriber(
                 subscription.id,
@@ -118,7 +142,7 @@ pub enum RegisterError {
 #[derive(Debug, thiserror::Error)]
 pub enum EchoError {
     #[error("subscription {0} not found")]
-    NotFound(Id),
+    NotFound(String),
     #[error("user {1} is not the subscriber for subscription {0}")]
     NotSubscriber(Id, user::Id),
     #[error(transparent)]
@@ -130,7 +154,7 @@ pub enum EchoError {
 #[derive(Debug, thiserror::Error)]
 pub enum UnregisterError {
     #[error("subscription {0} not found")]
-    NotFound(Id),
+    NotFound(String),
     #[error("user {1} is not the subscriber for subscription {0}")]
     NotSubscriber(Id, user::Id),
     #[error(transparent)]
diff --git a/src/push/handlers/echo.rs b/src/push/handlers/echo.rs
index 4b4de57..a6c7be2 100644
--- a/src/push/handlers/echo.rs
+++ b/src/push/handlers/echo.rs
@@ -1,10 +1,10 @@
 use axum::extract::{Json, State};
 
-use crate::{app::App, push::Id, token::extract::Identity};
+use crate::{app::App, token::extract::Identity};
 
 #[derive(serde::Deserialize)]
 pub struct Request {
-    subscription: Id,
+    endpoint: String,
     msg: String,
 }
 
@@ -13,8 +13,18 @@ pub async fn handler(
     identity: Identity,
     Json(request): Json<Request>,
 ) -> Result<(), crate::error::Internal> {
-    let Request { subscription, msg } = request;
-    app.push().echo(&identity.user, &subscription, &msg).await?;
+    let Request { endpoint, msg } = request;
+    app.push().echo(&identity.user, &endpoint, &msg).await?;
+
+    Ok(())
+}
+
+pub async fn broadcast(
+    State(app): State<App>,
+    Json(request): Json<Request>,
+) -> Result<(), crate::error::Internal> {
+    let Request { endpoint: _, msg } = request;
+    app.push().broadcast(&msg).await?;
 
     Ok(())
 }
diff --git a/src/push/handlers/mod.rs b/src/push/handlers/mod.rs
index 90edaa7..4b27ff5 100644
--- a/src/push/handlers/mod.rs
+++ b/src/push/handlers/mod.rs
@@ -7,6 +7,7 @@ mod register;
 mod unregister;
 
 pub use echo::handler as echo;
+pub use echo::broadcast as broadcast;
 pub use register::handler as register;
 pub use unregister::handler as unregister;
 
diff --git a/src/push/handlers/unregister.rs b/src/push/handlers/unregister.rs
index a00ee92..b35dbd5 100644
--- a/src/push/handlers/unregister.rs
+++ b/src/push/handlers/unregister.rs
@@ -1,16 +1,23 @@
 use axum::{
-    extract::{Path, State},
+    extract::{Json, State},
     http::StatusCode,
 };
 
-use crate::{app::App, error::Internal, push::Id, token::extract::Identity};
+use crate::{app::App, error::Internal, token::extract::Identity};
+
+
+#[derive(serde::Deserialize)]
+pub struct Request {
+    endpoint: String,
+}
 
 pub async fn handler(
     State(app): State<App>,
     identity: Identity,
-    Path(subscription): Path<Id>,
+    Json(request): Json<Request>,
 ) -> Result<StatusCode, Internal> {
-    app.push().unregister(&identity.user, &subscription).await?;
+    let Request { endpoint } = request;
+    app.push().unregister(&identity.user, &endpoint).await?;
 
     Ok(StatusCode::NO_CONTENT)
 }
diff --git a/src/push/repo.rs b/src/push/repo.rs
index ddef706..56bbc3d 100644
--- a/src/push/repo.rs
+++ b/src/push/repo.rs
@@ -41,7 +41,30 @@ impl Subscriptions<'_> {
         Ok(id)
     }
 
-    pub async fn by_id(&mut self, id: &Id) -> Result<Subscription, sqlx::Error> {
+    pub async fn all(&mut self) -> Result<Vec<Subscription>, sqlx::Error> {
+        let subscriptions = sqlx::query!(
+            r#"
+                select
+                    id as "id: Id",
+                    user as "user: user::Id",
+                    endpoint,
+                    key_p256dh,
+                    key_auth
+                from subscription
+            "#
+        )
+        .map(|row| Subscription {
+            id: row.id,
+            user: row.user,
+            info: SubscriptionInfo::new(row.endpoint, row.key_p256dh, row.key_auth),
+        })
+        .fetch_all(&mut *self.0)
+        .await?;
+
+        Ok(subscriptions)
+    }
+
+    pub async fn by_endpoint(&mut self, endpoint: &String) -> Result<Subscription, sqlx::Error> {
         let subscription = sqlx::query!(
             r#"
                 select
@@ -51,9 +74,9 @@ impl Subscriptions<'_> {
                     key_p256dh,
                     key_auth
                 from subscription
-                where id = $1
+                where endpoint = $1
             "#,
-            id,
+            endpoint,
         )
         .map(|row| Subscription {
             id: row.id,
@@ -70,9 +93,9 @@ impl Subscriptions<'_> {
         sqlx::query!(
             r#"
                 delete from subscription
-                where id = $1
+                where endpoint = $1
             "#,
-            subscription.id,
+            subscription.info.endpoint,
         )
         .execute(&mut *self.0)
         .await?;
diff --git a/src/routes.rs b/src/routes.rs
index ce60835..3e2fc04 100644
--- a/src/routes.rs
+++ b/src/routes.rs
@@ -27,10 +27,8 @@ pub fn routes(app: &App) -> Router<App> {
         .route("/api/setup", post(setup::handlers::setup))
         .route("/api/vapid", get(push::handlers::vapid))
         .route("/api/push", post(push::handlers::register))
-        .route(
-            "/api/push/:subscription",
-            delete(push::handlers::unregister),
-        )
+        .route("/api/push", delete(push::handlers::unregister))
+        .route("/api/broadcast", post(push::handlers::broadcast))
         .route("/api/echo", post(push::handlers::echo));
 
     // API routes that require the administrator to complete setup first.