1 files changed, 51 insertions, 5 deletions
diff --git a/src/channel/routes.rs b/src/channel/routes.rs
index 864f1b3..83c733c 100644
--- a/src/channel/routes.rs
+++ b/src/channel/routes.rs
@@ -1,14 +1,23 @@
 use axum::{
-    extract::{Form, State},
-    response::{IntoResponse, Redirect},
-    routing::post,
+    extract::{Form, Path, State},
+    http::StatusCode,
+    response::{
+        sse::{self, Sse},
+        IntoResponse, Redirect,
+    },
+    routing::{get, post},
     Router,
 };
+use futures::stream::{StreamExt as _, TryStreamExt as _};
 
-use crate::{app::App, error::InternalError, login::repo::logins::Login};
+use super::repo::channels::Id as ChannelId;
+use crate::{app::App, clock::RequestedAt, error::InternalError, login::repo::logins::Login};
 
 pub fn router() -> Router<App> {
-    Router::new().route("/create", post(on_create))
+    Router::new()
+        .route("/create", post(on_create))
+        .route("/:channel/send", post(on_send))
+        .route("/:channel/events", get(on_events))
 }
 
 #[derive(serde::Deserialize)]
@@ -25,3 +34,40 @@ async fn on_create(
 
     Ok(Redirect::to("/"))
 }
+
+#[derive(serde::Deserialize)]
+struct SendRequest {
+    message: String,
+}
+
+async fn on_send(
+    Path(channel): Path<ChannelId>,
+    RequestedAt(sent_at): RequestedAt,
+    State(app): State<App>,
+    login: Login,
+    Form(form): Form<SendRequest>,
+) -> Result<impl IntoResponse, InternalError> {
+    app.channels()
+        .send(&login, &channel, &form.message, &sent_at)
+        .await?;
+
+    Ok(StatusCode::ACCEPTED)
+}
+
+async fn on_events(
+    Path(channel): Path<ChannelId>,
+    State(app): State<App>,
+    _: Login, // requires auth, but doesn't actually care who you are
+) -> Result<impl IntoResponse, InternalError> {
+    let stream = app
+        .channels()
+        .events(&channel)
+        .await?
+        .map(|msg| match msg {
+            Ok(msg) => Ok(serde_json::to_string(&msg)?),
+            Err(err) => Err(err),
+        })
+        .map_ok(|msg| sse::Event::default().data(&msg));
+
+    Ok(Sse::new(stream).keep_alive(sse::KeepAlive::default()))
+}