1 files changed, 84 insertions, 37 deletions
diff --git a/src/event/routes/test/message.rs b/src/event/routes/test/message.rs
index 63a3f43..fafaeb3 100644
--- a/src/event/routes/test/message.rs
+++ b/src/event/routes/test/message.rs
@@ -16,14 +16,19 @@ async fn sending() {
 
     let app = fixtures::scratch_app().await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Call the endpoint
 
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let get::Response(events) =
-        get::handler(State(app.clone()), subscriber, None, Query::default())
-            .await
-            .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Send a message
 
@@ -56,6 +61,7 @@ async fn previously_sent() {
 
     let app = fixtures::scratch_app().await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Send a message
 
@@ -74,10 +80,14 @@ async fn previously_sent() {
     // Call the endpoint
 
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let get::Response(events) =
-        get::handler(State(app.clone()), subscriber, None, Query::default())
-            .await
-            .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Verify that an event is delivered
 
@@ -96,6 +106,7 @@ async fn sent_in_multiple_channels() {
 
     let app = fixtures::scratch_app().await;
     let sender = fixtures::login::create(&app, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
 
     let channels = [
         fixtures::channel::create(&app, &fixtures::now()).await,
@@ -115,9 +126,14 @@ async fn sent_in_multiple_channels() {
     // Call the endpoint
 
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let get::Response(events) = get::handler(State(app), subscriber, None, Query::default())
-        .await
-        .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Verify the structure of the response.
 
@@ -141,6 +157,7 @@ async fn sent_sequentially() {
     let app = fixtures::scratch_app().await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
     let sender = fixtures::login::create(&app, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
 
     let messages = vec![
         fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
@@ -151,9 +168,14 @@ async fn sent_sequentially() {
     // Subscribe
 
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let get::Response(events) = get::handler(State(app), subscriber, None, Query::default())
-        .await
-        .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Verify the expected events in the expected order
 
@@ -180,14 +202,19 @@ async fn expiring() {
     let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
     let sender = fixtures::login::create(&app, &fixtures::ancient()).await;
     let message = fixtures::message::send(&app, &channel, &sender, &fixtures::ancient()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Subscribe
 
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let get::Response(events) =
-        get::handler(State(app.clone()), subscriber, None, Query::default())
-            .await
-            .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Expire messages
 
@@ -214,6 +241,7 @@ async fn previously_expired() {
     let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
     let sender = fixtures::login::create(&app, &fixtures::ancient()).await;
     let message = fixtures::message::send(&app, &channel, &sender, &fixtures::ancient()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Expire messages
 
@@ -225,10 +253,14 @@ async fn previously_expired() {
     // Subscribe
 
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let get::Response(events) =
-        get::handler(State(app.clone()), subscriber, None, Query::default())
-            .await
-            .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Check for expiry event
     let _ = events
@@ -248,19 +280,24 @@ async fn deleting() {
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
     let sender = fixtures::login::create(&app, &fixtures::now()).await;
     let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Subscribe
 
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let get::Response(events) =
-        get::handler(State(app.clone()), subscriber, None, Query::default())
-            .await
-            .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Delete the message
 
     app.messages()
-        .delete(&message.id, &fixtures::now())
+        .delete(&sender, &message.id, &fixtures::now())
         .await
         .expect("deleting a valid message succeeds");
 
@@ -282,21 +319,26 @@ async fn previously_deleted() {
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
     let sender = fixtures::login::create(&app, &fixtures::now()).await;
     let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Delete the message
 
     app.messages()
-        .delete(&message.id, &fixtures::now())
+        .delete(&sender, &message.id, &fixtures::now())
         .await
         .expect("deleting a valid message succeeds");
 
     // Subscribe
 
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let get::Response(events) =
-        get::handler(State(app.clone()), subscriber, None, Query::default())
-            .await
-            .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Check for delete event
     let _ = events
@@ -316,11 +358,12 @@ async fn previously_purged() {
     let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
     let sender = fixtures::login::create(&app, &fixtures::ancient()).await;
     let message = fixtures::message::send(&app, &channel, &sender, &fixtures::ancient()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Purge the message
 
     app.messages()
-        .delete(&message.id, &fixtures::ancient())
+        .delete(&sender, &message.id, &fixtures::ancient())
         .await
         .expect("deleting a valid message succeeds");
 
@@ -332,10 +375,14 @@ async fn previously_purged() {
     // Subscribe
 
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let get::Response(events) =
-        get::handler(State(app.clone()), subscriber, None, Query::default())
-            .await
-            .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Check for delete event