Make `/api/events` a firehose endpoint.

It now includes events for all channels. Clients are responsible for filtering.

The schema for channel events has changed; it now includes a channel name and ID, in the same format as the sender's name and ID. They also now include a `"type"` field, whose only valid value (as of this writing) is `"message"`.

This is groundwork for delivering message deletion (expiry) events to clients, and notifying clients of channel lifecycle events.

1 files changed, 18 insertions, 71 deletions

diff --git a/src/channel/routes/test/on_send.rs b/src/channel/routes/test/on_send.rs
index 93a5480..5d87bdc 100644
--- a/src/channel/routes/test/on_send.rs
+++ b/src/channel/routes/test/on_send.rs
@@ -1,65 +1,14 @@
-use axum::{
-    extract::{Json, Path, State},
-    http::StatusCode,
-};
+use axum::extract::{Json, Path, State};
 use futures::stream::StreamExt;
 
 use crate::{
     channel::routes,
-    events::app,
+    events::{app, types},
     repo::channel,
     test::fixtures::{self, future::Immediately as _},
 };
 
 #[tokio::test]
-async fn channel_exists() {
-    // Set up the environment
-
-    let app = fixtures::scratch_app().await;
-    let sender = fixtures::login::create(&app).await;
-    let channel = fixtures::channel::create(&app).await;
-
-    // Call the endpoint
-
-    let sent_at = fixtures::now();
-    let request = routes::SendRequest {
-        message: fixtures::message::propose(),
-    };
-    let status = routes::on_send(
-        State(app.clone()),
-        Path(channel.id.clone()),
-        sent_at.clone(),
-        sender.clone(),
-        Json(request.clone()),
-    )
-    .await
-    .expect("sending to a valid channel");
-
-    // Verify the structure of the response
-
-    assert_eq!(StatusCode::ACCEPTED, status);
-
-    // Verify the semantics
-
-    let subscribed_at = fixtures::now();
-    let mut events = app
-        .events()
-        .subscribe(&channel.id, &subscribed_at, None)
-        .await
-        .expect("subscribing to a valid channel");
-
-    let event = events
-        .next()
-        .immediately()
-        .await
-        .expect("event received by subscribers");
-
-    assert_eq!(request.message, event.body);
-    assert_eq!(sender, event.sender);
-    assert_eq!(*sent_at, event.sent_at);
-}
-
-#[tokio::test]
 async fn messages_in_order() {
     // Set up the environment
 
@@ -70,21 +19,15 @@ async fn messages_in_order() {
     // Call the endpoint (twice)
 
     let requests = vec![
-        (
-            fixtures::now(),
-            routes::SendRequest {
-                message: fixtures::message::propose(),
-            },
-        ),
-        (
-            fixtures::now(),
-            routes::SendRequest {
-                message: fixtures::message::propose(),
-            },
-        ),
+        (fixtures::now(), fixtures::message::propose()),
+        (fixtures::now(), fixtures::message::propose()),
     ];
 
-    for (sent_at, request) in &requests {
+    for (sent_at, message) in &requests {
+        let request = routes::SendRequest {
+            message: message.clone(),
+        };
+
         routes::on_send(
             State(app.clone()),
             Path(channel.id.clone()),
@@ -101,17 +44,21 @@ async fn messages_in_order() {
     let subscribed_at = fixtures::now();
     let events = app
         .events()
-        .subscribe(&channel.id, &subscribed_at, None)
+        .subscribe(&subscribed_at, types::ResumePoint::default())
         .await
         .expect("subscribing to a valid channel")
         .take(requests.len());
 
     let events = events.collect::<Vec<_>>().immediately().await;
 
-    for ((sent_at, request), event) in requests.into_iter().zip(events) {
-        assert_eq!(request.message, event.body);
-        assert_eq!(sender, event.sender);
-        assert_eq!(*sent_at, event.sent_at);
+    for ((sent_at, message), types::ResumableEvent(_, event)) in requests.into_iter().zip(events) {
+        assert_eq!(*sent_at, event.at);
+        assert!(matches!(
+            event.data,
+            types::ChannelEventData::Message(event_message)
+                if event_message.sender == sender
+                    && event_message.body == message
+        ));
     }
 }