5 files changed, 55 insertions, 21 deletions

diff --git a/src/channel/app.rs b/src/channel/app.rs
index 6bad158..1eeca79 100644
--- a/src/channel/app.rs
+++ b/src/channel/app.rs
@@ -1,25 +1,33 @@
 use sqlx::sqlite::SqlitePool;
 
-use crate::repo::channel::{Channel, Provider as _};
+use crate::{
+    clock::DateTime,
+    events::{broadcaster::Broadcaster, types::ChannelEvent},
+    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) -> Self {
-        Self { db }
+    pub const fn new(db: &'a SqlitePool, broadcaster: &'a Broadcaster) -> Self {
+        Self { db, broadcaster }
     }
 
-    pub async fn create(&self, name: &str) -> Result<Channel, CreateError> {
+    pub async fn create(&self, name: &str, created_at: &DateTime) -> Result<Channel, CreateError> {
         let mut tx = self.db.begin().await?;
         let channel = tx
             .channels()
-            .create(name)
+            .create(name, created_at)
             .await
             .map_err(|err| CreateError::from_duplicate_name(err, name))?;
         tx.commit().await?;
 
+        self.broadcaster
+            .broadcast(&ChannelEvent::created(channel.clone()));
+
         Ok(channel)
     }
 
diff --git a/src/channel/routes.rs b/src/channel/routes.rs
index f524e62..1f8db5a 100644
--- a/src/channel/routes.rs
+++ b/src/channel/routes.rs
@@ -52,11 +52,12 @@ struct CreateRequest {
 async fn on_create(
     State(app): State<App>,
     _: Login, // requires auth, but doesn't actually care who you are
+    RequestedAt(created_at): RequestedAt,
     Json(form): Json<CreateRequest>,
 ) -> Result<Json<Channel>, CreateError> {
     let channel = app
         .channels()
-        .create(&form.name)
+        .create(&form.name, &created_at)
         .await
         .map_err(CreateError)?;
 
diff --git a/src/channel/routes/test/list.rs b/src/channel/routes/test/list.rs
index f7f7b44..bc94024 100644
--- a/src/channel/routes/test/list.rs
+++ b/src/channel/routes/test/list.rs
@@ -26,7 +26,7 @@ async fn one_channel() {
 
     let app = fixtures::scratch_app().await;
     let viewer = fixtures::login::create(&app).await;
-    let channel = fixtures::channel::create(&app).await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
 
     // Call the endpoint
 
@@ -46,8 +46,8 @@ async fn multiple_channels() {
     let app = fixtures::scratch_app().await;
     let viewer = fixtures::login::create(&app).await;
     let channels = vec![
-        fixtures::channel::create(&app).await,
-        fixtures::channel::create(&app).await,
+        fixtures::channel::create(&app, &fixtures::now()).await,
+        fixtures::channel::create(&app, &fixtures::now()).await,
     ];
 
     // Call the endpoint
diff --git a/src/channel/routes/test/on_create.rs b/src/channel/routes/test/on_create.rs
index 23885c0..bb6697f 100644
--- a/src/channel/routes/test/on_create.rs
+++ b/src/channel/routes/test/on_create.rs
@@ -1,8 +1,10 @@
 use axum::extract::{Json, State};
+use futures::{future, stream::StreamExt as _};
 
 use crate::{
     channel::{app, routes},
-    test::fixtures,
+    events::types,
+    test::fixtures::{self, future::Immediately as _},
 };
 
 #[tokio::test]
@@ -16,10 +18,14 @@ async fn new_channel() {
 
     let name = fixtures::channel::propose();
     let request = routes::CreateRequest { name };
-    let Json(response_channel) =
-        routes::on_create(State(app.clone()), creator, Json(request.clone()))
-            .await
-            .expect("new channel in an empty app");
+    let Json(response_channel) = routes::on_create(
+        State(app.clone()),
+        creator,
+        fixtures::now(),
+        Json(request.clone()),
+    )
+    .await
+    .expect("new channel in an empty app");
 
     // Verify the structure of the response
 
@@ -28,8 +34,23 @@ async fn new_channel() {
     // Verify the semantics
 
     let channels = app.channels().all().await.expect("always succeeds");
-
     assert!(channels.contains(&response_channel));
+
+    let mut events = app
+        .events()
+        .subscribe(&fixtures::now(), types::ResumePoint::default())
+        .await
+        .expect("subscribing never fails")
+        .filter(|types::ResumableEvent(_, event)| future::ready(event.channel == response_channel));
+
+    let types::ResumableEvent(_, event) = events
+        .next()
+        .immediately()
+        .await
+        .expect("creation event published");
+
+    assert_eq!(types::Sequence::default(), event.sequence);
+    assert_eq!(types::ChannelEventData::Created, event.data);
 }
 
 #[tokio::test]
@@ -38,15 +59,19 @@ async fn duplicate_name() {
 
     let app = fixtures::scratch_app().await;
     let creator = fixtures::login::create(&app).await;
-    let channel = fixtures::channel::create(&app).await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
 
     // Call the endpoint
 
     let request = routes::CreateRequest { name: channel.name };
-    let routes::CreateError(error) =
-        routes::on_create(State(app.clone()), creator, Json(request.clone()))
-            .await
-            .expect_err("duplicate channel name");
+    let routes::CreateError(error) = routes::on_create(
+        State(app.clone()),
+        creator,
+        fixtures::now(),
+        Json(request.clone()),
+    )
+    .await
+    .expect_err("duplicate channel name");
 
     // Verify the structure of the response
 
diff --git a/src/channel/routes/test/on_send.rs b/src/channel/routes/test/on_send.rs
index 5d87bdc..e4de0f1 100644
--- a/src/channel/routes/test/on_send.rs
+++ b/src/channel/routes/test/on_send.rs
@@ -14,7 +14,7 @@ async fn messages_in_order() {
 
     let app = fixtures::scratch_app().await;
     let sender = fixtures::login::create(&app).await;
-    let channel = fixtures::channel::create(&app).await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
 
     // Call the endpoint (twice)