Annotate channel events with channel ID at the router, not intrinsically.

This bugged me aesthetically. At `app.channel().events(channel)`, the caller knows the channel ID; they don't need to be told. Having the same info come back out in the returned events felt bad.

2 files changed, 24 insertions, 10 deletions

diff --git a/src/channel/repo/messages.rs b/src/channel/repo/messages.rs
index e15a02a..2c89724 100644
--- a/src/channel/repo/messages.rs
+++ b/src/channel/repo/messages.rs
@@ -26,7 +26,6 @@ pub struct Messages<'t>(&'t mut SqliteConnection);
 pub struct BroadcastMessage {
     pub id: Id,
     pub sender: Login,
-    pub channel: ChannelId,
     pub body: String,
     pub sent_at: DateTime,
 }
@@ -51,7 +50,6 @@ impl<'c> Messages<'c> {
                 returning
                     id as "id: Id",
                     sender as "sender: LoginId",
-                    channel as "channel: ChannelId",
                     body,
                     sent_at as "sent_at: DateTime"
             "#,
@@ -66,7 +64,6 @@ impl<'c> Messages<'c> {
             BroadcastMessage {
                 id: row.id,
                 sender: sender.clone(),
-                channel: row.channel,
                 body: row.body,
                 sent_at: row.sent_at,
             }
@@ -88,7 +85,6 @@ impl<'c> Messages<'c> {
                     message.id as "id: Id",
                     login.id as "sender_id: LoginId",
                     login.name as sender_name,
-                    message.channel as "channel: ChannelId",
                     message.body,
                     message.sent_at as "sent_at: DateTime"
                 from message
@@ -106,7 +102,6 @@ impl<'c> Messages<'c> {
                 id: row.sender_id,
                 name: row.sender_name,
             },
-            channel: row.channel,
             body: row.body,
             sent_at: row.sent_at,
         })
diff --git a/src/events.rs b/src/events.rs
index 2d1e1f8..51f06b4 100644
--- a/src/events.rs
+++ b/src/events.rs
@@ -49,21 +49,33 @@ async fn on_events(
     let streams = stream::iter(query.channels)
         .then(|channel| {
             let app = app.clone();
-            async move { app.channels().events(&channel, resume_at.as_ref()).await }
+            async move {
+                let events = app
+                    .channels()
+                    .events(&channel, resume_at.as_ref())
+                    .await?
+                    .map_ok(move |message| ChannelEvent {
+                        channel: channel.clone(),
+                        message,
+                    });
+
+                Ok::<_, BoxedError>(events)
+            }
         })
         .try_collect::<Vec<_>>()
         .await?;
 
-    let stream = stream::select_all(streams).and_then(|msg| future::ready(to_event(msg)));
+    let stream = stream::select_all(streams).and_then(|msg| future::ready(to_sse_event(msg)));
     let sse = Sse::new(stream).keep_alive(sse::KeepAlive::default());
 
     Ok(sse)
 }
 
-fn to_event(msg: BroadcastMessage) -> Result<sse::Event, BoxedError> {
-    let data = serde_json::to_string(&msg)?;
+fn to_sse_event(event: ChannelEvent<BroadcastMessage>) -> Result<sse::Event, BoxedError> {
+    let data = serde_json::to_string(&event)?;
     let event = sse::Event::default()
-        .id(msg
+        .id(event
+            .message
             .sent_at
             .to_rfc3339_opts(SecondsFormat::AutoSi, /* use_z */ true))
         .data(&data);
@@ -71,6 +83,13 @@ fn to_event(msg: BroadcastMessage) -> Result<sse::Event, BoxedError> {
     Ok(event)
 }
 
+#[derive(serde::Serialize)]
+struct ChannelEvent<M> {
+    channel: ChannelId,
+    #[serde(flatten)]
+    message: M,
+}
+
 pub struct LastEventId(pub String);
 
 static LAST_EVENT_ID: HeaderName = HeaderName::from_static("last-event-id");