Represent channels and messages using a split "History" and "Snapshot" model.

This separates the code that figures out what happened to an entity from the code that represents it to a user, and makes it easier to compute a snapshot at a point in time (for things like bootstrap). It also makes the internal logic a bit easier to follow, since it's easier to tell whether you're working with a point in time or with the whole recorded history.

This hefty.

1 files changed, 61 insertions, 37 deletions

diff --git a/src/event/routes/test.rs b/src/event/routes/test.rs
index 68b55cc..ba9953e 100644
--- a/src/event/routes/test.rs
+++ b/src/event/routes/test.rs
@@ -6,7 +6,7 @@ use futures::{
 };
 
 use crate::{
-    event::{routes, Sequence},
+    event::{routes, Sequenced as _},
     test::fixtures::{self, future::Immediately as _},
 };
 
@@ -17,7 +17,7 @@ async fn includes_historical_message() {
     let app = fixtures::scratch_app().await;
     let sender = fixtures::login::create(&app).await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
-    let message = fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await;
+    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
 
     // Call the endpoint
 
@@ -36,7 +36,7 @@ async fn includes_historical_message() {
         .await
         .expect("delivered stored message");
 
-    assert_eq!(message, event);
+    assert!(fixtures::event::message_sent(&event, &message));
 }
 
 #[tokio::test]
@@ -58,7 +58,7 @@ async fn includes_live_message() {
     // Verify the semantics
 
     let sender = fixtures::login::create(&app).await;
-    let message = fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await;
+    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
 
     let event = events
         .filter(fixtures::filter::messages())
@@ -67,7 +67,7 @@ async fn includes_live_message() {
         .await
         .expect("delivered live message");
 
-    assert_eq!(message, event);
+    assert!(fixtures::event::message_sent(&event, &message));
 }
 
 #[tokio::test]
@@ -87,7 +87,7 @@ async fn includes_multiple_channels() {
             let app = app.clone();
             let sender = sender.clone();
             let channel = channel.clone();
-            async move { fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await }
+            async move { fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await }
         })
         .collect::<Vec<_>>()
         .await;
@@ -110,7 +110,9 @@ async fn includes_multiple_channels() {
         .await;
 
     for message in &messages {
-        assert!(events.iter().any(|event| { event == message }));
+        assert!(events
+            .iter()
+            .any(|event| fixtures::event::message_sent(event, message)));
     }
 }
 
@@ -123,9 +125,9 @@ async fn sequential_messages() {
     let sender = fixtures::login::create(&app).await;
 
     let messages = vec![
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
     ];
 
     // Call the endpoint
@@ -138,7 +140,13 @@ async fn sequential_messages() {
 
     // Verify the structure of the response.
 
-    let mut events = events.filter(|event| future::ready(messages.contains(event)));
+    let mut events = events.filter(|event| {
+        future::ready(
+            messages
+                .iter()
+                .any(|message| fixtures::event::message_sent(event, message)),
+        )
+    });
 
     // Verify delivery in order
     for message in &messages {
@@ -148,7 +156,7 @@ async fn sequential_messages() {
             .await
             .expect("undelivered messages remaining");
 
-        assert_eq!(message, &event);
+        assert!(fixtures::event::message_sent(&event, message));
     }
 }
 
@@ -160,11 +168,11 @@ async fn resumes_from() {
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
     let sender = fixtures::login::create(&app).await;
 
-    let initial_message = fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await;
+    let initial_message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
 
     let later_messages = vec![
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
     ];
 
     // Call the endpoint
@@ -190,9 +198,9 @@ async fn resumes_from() {
             .await
             .expect("delivered events");
 
-        assert_eq!(initial_message, event);
+        assert!(fixtures::event::message_sent(&event, &initial_message));
 
-        Sequence::from(&event)
+        event.sequence()
     };
 
     // Resume after disconnect
@@ -214,7 +222,9 @@ async fn resumes_from() {
         .await;
 
     for message in &later_messages {
-        assert!(events.iter().any(|event| event == message));
+        assert!(events
+            .iter()
+            .any(|event| fixtures::event::message_sent(event, message)));
     }
 }
 
@@ -249,8 +259,8 @@ async fn serial_resume() {
 
     let resume_at = {
         let initial_messages = [
-            fixtures::message::send(&app, &sender, &channel_a, &fixtures::now()).await,
-            fixtures::message::send(&app, &sender, &channel_b, &fixtures::now()).await,
+            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await,
         ];
 
         // First subscription
@@ -271,12 +281,14 @@ async fn serial_resume() {
             .await;
 
         for message in &initial_messages {
-            assert!(events.iter().any(|event| event == message));
+            assert!(events
+                .iter()
+                .any(|event| fixtures::event::message_sent(event, message)));
         }
 
         let event = events.last().expect("this vec is non-empty");
 
-        Sequence::from(event)
+        event.sequence()
     };
 
     // Resume after disconnect
@@ -285,8 +297,8 @@ async fn serial_resume() {
             // Note that channel_b does not appear here. The buggy behaviour
             // would be masked if channel_b happened to send a new message
             // into the resumed event stream.
-            fixtures::message::send(&app, &sender, &channel_a, &fixtures::now()).await,
-            fixtures::message::send(&app, &sender, &channel_a, &fixtures::now()).await,
+            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
         ];
 
         // Second subscription
@@ -307,12 +319,14 @@ async fn serial_resume() {
             .await;
 
         for message in &resume_messages {
-            assert!(events.iter().any(|event| event == message));
+            assert!(events
+                .iter()
+                .any(|event| fixtures::event::message_sent(event, message)));
         }
 
         let event = events.last().expect("this vec is non-empty");
 
-        Sequence::from(event)
+        event.sequence()
     };
 
     // Resume after disconnect a second time
@@ -321,8 +335,8 @@ async fn serial_resume() {
         // problem. The resume point should before both of these messages, but
         // after _all_ prior messages.
         let final_messages = [
-            fixtures::message::send(&app, &sender, &channel_a, &fixtures::now()).await,
-            fixtures::message::send(&app, &sender, &channel_b, &fixtures::now()).await,
+            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await,
         ];
 
         // Third subscription
@@ -345,7 +359,9 @@ async fn serial_resume() {
         // This set of messages, in particular, _should not_ include any prior
         // messages from `initial_messages` or `resume_messages`.
         for message in &final_messages {
-            assert!(events.iter().any(|event| event == message));
+            assert!(events
+                .iter()
+                .any(|event| fixtures::event::message_sent(event, message)));
         }
     };
 }
@@ -378,13 +394,17 @@ async fn terminates_on_token_expiry() {
 
     // These should not be delivered.
     let messages = [
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
     ];
 
     assert!(events
-        .filter(|event| future::ready(messages.contains(event)))
+        .filter(|event| future::ready(
+            messages
+                .iter()
+                .any(|message| fixtures::event::message_sent(event, message))
+        ))
         .next()
         .immediately()
         .await
@@ -425,13 +445,17 @@ async fn terminates_on_logout() {
 
     // These should not be delivered.
     let messages = [
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
-        fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
     ];
 
     assert!(events
-        .filter(|event| future::ready(messages.contains(event)))
+        .filter(|event| future::ready(
+            messages
+                .iter()
+                .any(|message| fixtures::event::message_sent(event, message))
+        ))
         .next()
         .immediately()
         .await