2 files changed, 21 insertions, 79 deletions

diff --git a/src/channel/app.rs b/src/channel/app.rs
index 793fa35..6bad158 100644
--- a/src/channel/app.rs
+++ b/src/channel/app.rs
@@ -1,18 +1,14 @@
 use sqlx::sqlite::SqlitePool;
 
-use crate::{
-    events::broadcaster::Broadcaster,
-    repo::channel::{Channel, Provider as _},
-};
+use crate::repo::channel::{Channel, Provider as _};
 
 pub struct Channels<'a> {
     db: &'a SqlitePool,
-    broadcaster: &'a Broadcaster,
 }
 
 impl<'a> Channels<'a> {
-    pub const fn new(db: &'a SqlitePool, broadcaster: &'a Broadcaster) -> Self {
-        Self { db, broadcaster }
+    pub const fn new(db: &'a SqlitePool) -> Self {
+        Self { db }
     }
 
     pub async fn create(&self, name: &str) -> Result<Channel, CreateError> {
@@ -22,7 +18,6 @@ impl<'a> Channels<'a> {
             .create(name)
             .await
             .map_err(|err| CreateError::from_duplicate_name(err, name))?;
-        self.broadcaster.register_channel(&channel.id);
         tx.commit().await?;
 
         Ok(channel)
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
+        ));
     }
 }