2 files changed, 193 insertions, 0 deletions
diff --git a/src/conversation/handlers/send/mod.rs b/src/conversation/handlers/send/mod.rs
new file mode 100644
index 0000000..9ec020a
--- /dev/null
+++ b/src/conversation/handlers/send/mod.rs
@@ -0,0 +1,63 @@
+use axum::{
+    extract::{Json, Path, State},
+    http::StatusCode,
+    response::{self, IntoResponse},
+};
+
+use crate::conversation::handlers::PathInfo;
+use crate::{
+    app::App,
+    clock::RequestedAt,
+    error::{Internal, NotFound},
+    message::{Body, Message, app::SendError},
+    token::extract::Identity,
+};
+
+#[cfg(test)]
+mod test;
+
+pub async fn handler(
+    State(app): State<App>,
+    Path(conversation): Path<PathInfo>,
+    RequestedAt(sent_at): RequestedAt,
+    identity: Identity,
+    Json(request): Json<Request>,
+) -> Result<Response, Error> {
+    let message = app
+        .messages()
+        .send(&conversation, &identity.user, &sent_at, &request.body)
+        .await?;
+
+    Ok(Response(message))
+}
+
+#[derive(serde::Deserialize)]
+pub struct Request {
+    pub body: Body,
+}
+
+#[derive(Debug)]
+pub struct Response(pub Message);
+
+impl IntoResponse for Response {
+    fn into_response(self) -> response::Response {
+        let Self(message) = self;
+        (StatusCode::ACCEPTED, Json(message)).into_response()
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error(transparent)]
+pub struct Error(#[from] pub SendError);
+
+impl IntoResponse for Error {
+    fn into_response(self) -> response::Response {
+        let Self(error) = self;
+        match error {
+            SendError::ConversationNotFound(_) | SendError::ConversationDeleted(_) => {
+                NotFound(error).into_response()
+            }
+            SendError::Name(_) | SendError::Database(_) => Internal::from(error).into_response(),
+        }
+    }
+}
diff --git a/src/conversation/handlers/send/test.rs b/src/conversation/handlers/send/test.rs
new file mode 100644
index 0000000..bd32510
--- /dev/null
+++ b/src/conversation/handlers/send/test.rs
@@ -0,0 +1,130 @@
+use axum::extract::{Json, Path, State};
+use futures::stream::{self, StreamExt as _};
+
+use crate::{
+    conversation,
+    event::Sequenced,
+    message::app::SendError,
+    test::fixtures::{self, future::Expect as _},
+};
+
+#[tokio::test]
+async fn messages_in_order() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let sender = fixtures::identity::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Call the endpoint (twice)
+
+    let requests = vec![
+        (fixtures::now(), fixtures::message::propose()),
+        (fixtures::now(), fixtures::message::propose()),
+    ];
+
+    for (sent_at, body) in &requests {
+        let request = super::Request { body: body.clone() };
+
+        let _ = super::handler(
+            State(app.clone()),
+            Path(conversation.id.clone()),
+            sent_at.clone(),
+            sender.clone(),
+            Json(request),
+        )
+        .await
+        .expect("sending to a valid conversation succeeds");
+    }
+
+    // Verify the semantics
+
+    let mut events = app
+        .events()
+        .subscribe(resume_point)
+        .await
+        .expect("subscribing always succeeds")
+        .filter_map(fixtures::event::stream::message)
+        .filter_map(fixtures::event::stream::message::sent)
+        .zip(stream::iter(requests));
+
+    while let Some((event, (sent_at, body))) = events
+        .next()
+        .expect_ready("an event should be ready for each message")
+        .await
+    {
+        assert_eq!(*sent_at, event.at());
+        assert_eq!(sender.user.id, event.message.sender);
+        assert_eq!(body, event.message.body);
+    }
+}
+
+#[tokio::test]
+async fn nonexistent_conversation() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let sender = fixtures::identity::create(&app, &fixtures::now()).await;
+
+    // Call the endpoint
+
+    let sent_at = fixtures::now();
+    let conversation = conversation::Id::generate();
+    let request = super::Request {
+        body: fixtures::message::propose(),
+    };
+    let super::Error(error) = super::handler(
+        State(app),
+        Path(conversation.clone()),
+        sent_at,
+        sender,
+        Json(request),
+    )
+    .await
+    .expect_err("sending to a nonexistent conversation fails");
+
+    // Verify the structure of the response
+
+    assert!(matches!(
+        error,
+        SendError::ConversationNotFound(error_conversation) if conversation == error_conversation
+    ));
+}
+
+#[tokio::test]
+async fn deleted_conversation() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let sender = fixtures::identity::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
+
+    app.conversations()
+        .delete(&conversation.id, &fixtures::now())
+        .await
+        .expect("deleting a new conversation succeeds");
+
+    // Call the endpoint
+
+    let sent_at = fixtures::now();
+    let request = super::Request {
+        body: fixtures::message::propose(),
+    };
+    let super::Error(error) = super::handler(
+        State(app),
+        Path(conversation.id.clone()),
+        sent_at,
+        sender,
+        Json(request),
+    )
+    .await
+    .expect_err("sending to a deleted conversation fails");
+
+    // Verify the structure of the response
+
+    assert!(matches!(
+        error,
+        SendError::ConversationDeleted(error_conversation) if conversation.id == error_conversation
+    ));
+}