3 files changed, 114 insertions, 2 deletions
diff --git a/src/event/handlers/stream/mod.rs b/src/event/handlers/stream/mod.rs
index 63bfff3..8b89c31 100644
--- a/src/event/handlers/stream/mod.rs
+++ b/src/event/handlers/stream/mod.rs
@@ -18,8 +18,8 @@ use crate::{
 #[cfg(test)]
 mod test;
 
-pub async fn handler(
-    State(app): State<App>,
+pub async fn handler<P>(
+    State(app): State<App<P>>,
     identity: Identity,
     last_event_id: Option<LastEventId<Sequence>>,
     Query(query): Query<QueryParams>,
diff --git a/src/event/handlers/stream/test/mod.rs b/src/event/handlers/stream/test/mod.rs
index 3bc634f..c3a6ce6 100644
--- a/src/event/handlers/stream/test/mod.rs
+++ b/src/event/handlers/stream/test/mod.rs
@@ -4,5 +4,6 @@ mod message;
 mod resume;
 mod setup;
 mod token;
+mod vapid;
 
 use super::{QueryParams, Response, handler};
diff --git a/src/event/handlers/stream/test/vapid.rs b/src/event/handlers/stream/test/vapid.rs
new file mode 100644
index 0000000..dbc3929
--- /dev/null
+++ b/src/event/handlers/stream/test/vapid.rs
@@ -0,0 +1,111 @@
+use axum::extract::State;
+use axum_extra::extract::Query;
+use futures::StreamExt as _;
+
+use crate::test::{fixtures, fixtures::future::Expect as _};
+
+#[tokio::test]
+async fn live_vapid_key_changes() {
+    // Set up the context
+    let app = fixtures::scratch_app().await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Subscribe to events
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Rotate the VAPID key
+
+    app.vapid()
+        .refresh_key(&fixtures::now())
+        .await
+        .expect("refreshing the vapid key always succeeds");
+
+    // Verify that there's a key rotation event
+
+    events
+        .filter_map(fixtures::event::stream::vapid)
+        .filter_map(fixtures::event::stream::vapid::changed)
+        .next()
+        .expect_some("a vapid key change event is sent")
+        .await;
+}
+
+#[tokio::test]
+async fn stored_vapid_key_changes() {
+    // Set up the context
+    let app = fixtures::scratch_app().await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Rotate the VAPID key
+
+    app.vapid()
+        .refresh_key(&fixtures::now())
+        .await
+        .expect("refreshing the vapid key always succeeds");
+
+    // Subscribe to events
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Verify that there's a key rotation event
+
+    events
+        .filter_map(fixtures::event::stream::vapid)
+        .filter_map(fixtures::event::stream::vapid::changed)
+        .next()
+        .expect_some("a vapid key change event is sent")
+        .await;
+}
+
+#[tokio::test]
+async fn no_past_vapid_key_changes() {
+    // Set up the context
+    let app = fixtures::scratch_app().await;
+
+    // Rotate the VAPID key
+
+    app.vapid()
+        .refresh_key(&fixtures::now())
+        .await
+        .expect("refreshing the vapid key always succeeds");
+
+    // Subscribe to events
+
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Verify that there's a key rotation event
+
+    events
+        .filter_map(fixtures::event::stream::vapid)
+        .filter_map(fixtures::event::stream::vapid::changed)
+        .next()
+        .expect_wait("a vapid key change event is not sent")
+        .await;
+}