Handlers are _named operations_, which can be exposed via routes.

Each domain module that exposes handlers does so through a `handlers` child module, ideally as a top-level symbol that can be plugged directly into Axum's `MethodRouter`. Modules could make exceptions to this - kill the doctrinaire inside yourself, after all - but none of the API modules that actually exist need such exceptions, and consistency is useful.

The related details of request types, URL types, response types, errors, &c &c are then organized into modules under `handlers`, along with their respective tests.

7 files changed, 743 insertions, 0 deletions

diff --git a/src/channel/handlers/create/mod.rs b/src/channel/handlers/create/mod.rs
new file mode 100644
index 0000000..2c860fc
--- /dev/null
+++ b/src/channel/handlers/create/mod.rs
@@ -0,0 +1,67 @@
+use axum::{
+    extract::{Json, State},
+    http::StatusCode,
+    response::{self, IntoResponse},
+};
+
+use crate::{
+    app::App,
+    channel::{Channel, app},
+    clock::RequestedAt,
+    error::Internal,
+    name::Name,
+    token::extract::Identity,
+};
+
+#[cfg(test)]
+mod test;
+
+pub async fn handler(
+    State(app): State<App>,
+    _: Identity, // requires auth, but doesn't actually care who you are
+    RequestedAt(created_at): RequestedAt,
+    Json(request): Json<Request>,
+) -> Result<Response, Error> {
+    let channel = app
+        .channels()
+        .create(&request.name, &created_at)
+        .await
+        .map_err(Error)?;
+
+    Ok(Response(channel))
+}
+
+#[derive(serde::Deserialize)]
+pub struct Request {
+    pub name: Name,
+}
+
+#[derive(Debug)]
+pub struct Response(pub Channel);
+
+impl IntoResponse for Response {
+    fn into_response(self) -> response::Response {
+        let Self(channel) = self;
+        (StatusCode::ACCEPTED, Json(channel)).into_response()
+    }
+}
+
+#[derive(Debug)]
+pub struct Error(pub app::CreateError);
+
+impl IntoResponse for Error {
+    fn into_response(self) -> response::Response {
+        let Self(error) = self;
+        match error {
+            app::CreateError::DuplicateName(_) => {
+                (StatusCode::CONFLICT, error.to_string()).into_response()
+            }
+            app::CreateError::InvalidName(_) => {
+                (StatusCode::BAD_REQUEST, error.to_string()).into_response()
+            }
+            app::CreateError::Name(_) | app::CreateError::Database(_) => {
+                Internal::from(error).into_response()
+            }
+        }
+    }
+}
diff --git a/src/channel/handlers/create/test.rs b/src/channel/handlers/create/test.rs
new file mode 100644
index 0000000..3c770cf
--- /dev/null
+++ b/src/channel/handlers/create/test.rs
@@ -0,0 +1,242 @@
+use std::future;
+
+use axum::extract::{Json, State};
+use futures::stream::StreamExt as _;
+
+use crate::{
+    channel::app,
+    name::Name,
+    test::fixtures::{self, future::Expect as _},
+};
+
+#[tokio::test]
+async fn new_channel() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let creator = fixtures::identity::create(&app, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Call the endpoint
+
+    let name = fixtures::channel::propose();
+    let request = super::Request { name: name.clone() };
+    let super::Response(response) =
+        super::handler(State(app.clone()), creator, fixtures::now(), Json(request))
+            .await
+            .expect("creating a channel in an empty app succeeds");
+
+    // Verify the structure of the response
+
+    assert_eq!(name, response.name);
+
+    // Verify the semantics
+
+    let snapshot = app.boot().snapshot().await.expect("boot always succeeds");
+    assert!(snapshot.channels.iter().any(|channel| channel == &response));
+
+    let channel = app
+        .channels()
+        .get(&response.id)
+        .await
+        .expect("the newly-created channel exists");
+    assert_eq!(response, channel);
+
+    let mut events = app
+        .events()
+        .subscribe(resume_point)
+        .await
+        .expect("subscribing never fails")
+        .filter_map(fixtures::event::channel)
+        .filter_map(fixtures::event::channel::created)
+        .filter(|event| future::ready(event.channel == response));
+
+    let event = events.next().expect_some("creation event published").await;
+
+    assert_eq!(event.channel, response);
+}
+
+#[tokio::test]
+async fn duplicate_name() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let creator = fixtures::identity::create(&app, &fixtures::now()).await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+
+    // Call the endpoint
+
+    let request = super::Request {
+        name: channel.name.clone(),
+    };
+    let super::Error(error) =
+        super::handler(State(app.clone()), creator, fixtures::now(), Json(request))
+            .await
+            .expect_err("duplicate channel name should fail the request");
+
+    // Verify the structure of the response
+
+    assert!(matches!(
+        error,
+        app::CreateError::DuplicateName(name) if channel.name == name
+    ));
+}
+
+#[tokio::test]
+async fn conflicting_canonical_name() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let creator = fixtures::identity::create(&app, &fixtures::now()).await;
+
+    let existing_name = Name::from("rijksmuseum");
+    app.channels()
+        .create(&existing_name, &fixtures::now())
+        .await
+        .expect("creating a channel in an empty environment succeeds");
+
+    let conflicting_name = Name::from("r\u{0133}ksmuseum");
+
+    // Call the endpoint
+
+    let request = super::Request {
+        name: conflicting_name.clone(),
+    };
+    let super::Error(error) =
+        super::handler(State(app.clone()), creator, fixtures::now(), Json(request))
+            .await
+            .expect_err("duplicate channel name should fail the request");
+
+    // Verify the structure of the response
+
+    assert!(matches!(
+        error,
+        app::CreateError::DuplicateName(name) if conflicting_name == name
+    ));
+}
+
+#[tokio::test]
+async fn invalid_name() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let creator = fixtures::identity::create(&app, &fixtures::now()).await;
+
+    // Call the endpoint
+
+    let name = fixtures::channel::propose_invalid_name();
+    let request = super::Request { name: name.clone() };
+    let super::Error(error) = crate::channel::handlers::create::handler(
+        State(app.clone()),
+        creator,
+        fixtures::now(),
+        Json(request),
+    )
+    .await
+    .expect_err("invalid channel name should fail the request");
+
+    // Verify the structure of the response
+
+    assert!(matches!(
+        error,
+        app::CreateError::InvalidName(error_name) if name == error_name
+    ));
+}
+
+#[tokio::test]
+async fn name_reusable_after_delete() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let creator = fixtures::identity::create(&app, &fixtures::now()).await;
+    let name = fixtures::channel::propose();
+
+    // Call the endpoint (first time)
+
+    let request = super::Request { name: name.clone() };
+    let super::Response(response) = super::handler(
+        State(app.clone()),
+        creator.clone(),
+        fixtures::now(),
+        Json(request),
+    )
+    .await
+    .expect("new channel in an empty app");
+
+    // Delete the channel
+
+    app.channels()
+        .delete(&response.id, &fixtures::now())
+        .await
+        .expect("deleting a newly-created channel succeeds");
+
+    // Call the endpoint (second time)
+
+    let request = super::Request { name: name.clone() };
+    let super::Response(response) =
+        super::handler(State(app.clone()), creator, fixtures::now(), Json(request))
+            .await
+            .expect("creation succeeds after original channel deleted");
+
+    // Verify the structure of the response
+
+    assert_eq!(name, response.name);
+
+    // Verify the semantics
+
+    let channel = app
+        .channels()
+        .get(&response.id)
+        .await
+        .expect("the newly-created channel exists");
+    assert_eq!(response, channel);
+}
+
+#[tokio::test]
+async fn name_reusable_after_expiry() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let creator = fixtures::identity::create(&app, &fixtures::ancient()).await;
+    let name = fixtures::channel::propose();
+
+    // Call the endpoint (first time)
+
+    let request = super::Request { name: name.clone() };
+    let super::Response(_) = super::handler(
+        State(app.clone()),
+        creator.clone(),
+        fixtures::ancient(),
+        Json(request),
+    )
+    .await
+    .expect("new channel in an empty app");
+
+    // Delete the channel
+
+    app.channels()
+        .expire(&fixtures::now())
+        .await
+        .expect("expiry always succeeds");
+
+    // Call the endpoint (second time)
+
+    let request = super::Request { name: name.clone() };
+    let super::Response(response) =
+        super::handler(State(app.clone()), creator, fixtures::now(), Json(request))
+            .await
+            .expect("creation succeeds after original channel expired");
+
+    // Verify the structure of the response
+
+    assert_eq!(name, response.name);
+
+    // Verify the semantics
+
+    let channel = app
+        .channels()
+        .get(&response.id)
+        .await
+        .expect("the newly-created channel exists");
+    assert_eq!(response, channel);
+}
diff --git a/src/channel/handlers/delete/mod.rs b/src/channel/handlers/delete/mod.rs
new file mode 100644
index 0000000..b986bec
--- /dev/null
+++ b/src/channel/handlers/delete/mod.rs
@@ -0,0 +1,59 @@
+use axum::{
+    extract::{Json, Path, State},
+    http::StatusCode,
+    response::{self, IntoResponse},
+};
+
+use crate::{
+    app::App,
+    channel::{self, app, handlers::PathInfo},
+    clock::RequestedAt,
+    error::{Internal, NotFound},
+    token::extract::Identity,
+};
+
+#[cfg(test)]
+mod test;
+
+pub async fn handler(
+    State(app): State<App>,
+    Path(channel): Path<PathInfo>,
+    RequestedAt(deleted_at): RequestedAt,
+    _: Identity,
+) -> Result<Response, Error> {
+    app.channels().delete(&channel, &deleted_at).await?;
+
+    Ok(Response { id: channel })
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct Response {
+    pub id: channel::Id,
+}
+
+impl IntoResponse for Response {
+    fn into_response(self) -> response::Response {
+        (StatusCode::ACCEPTED, Json(self)).into_response()
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error(transparent)]
+pub struct Error(#[from] pub app::DeleteError);
+
+impl IntoResponse for Error {
+    fn into_response(self) -> response::Response {
+        let Self(error) = self;
+        match error {
+            app::DeleteError::NotFound(_) | app::DeleteError::Deleted(_) => {
+                NotFound(error).into_response()
+            }
+            app::DeleteError::NotEmpty(_) => {
+                (StatusCode::CONFLICT, error.to_string()).into_response()
+            }
+            app::DeleteError::Name(_) | app::DeleteError::Database(_) => {
+                Internal::from(error).into_response()
+            }
+        }
+    }
+}
diff --git a/src/channel/handlers/delete/test.rs b/src/channel/handlers/delete/test.rs
new file mode 100644
index 0000000..b1e42ea
--- /dev/null
+++ b/src/channel/handlers/delete/test.rs
@@ -0,0 +1,174 @@
+use axum::extract::{Path, State};
+
+use crate::{channel::app, test::fixtures};
+
+#[tokio::test]
+pub async fn valid_channel() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+
+    // Send the request
+
+    let deleter = fixtures::identity::create(&app, &fixtures::now()).await;
+    let response = super::handler(
+        State(app.clone()),
+        Path(channel.id.clone()),
+        fixtures::now(),
+        deleter,
+    )
+    .await
+    .expect("deleting a valid channel succeeds");
+
+    // Verify the response
+
+    assert_eq!(channel.id, response.id);
+
+    // Verify the semantics
+
+    let snapshot = app.boot().snapshot().await.expect("boot always succeeds");
+    assert!(!snapshot.channels.contains(&channel));
+}
+
+#[tokio::test]
+pub async fn invalid_channel_id() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+
+    // Send the request
+
+    let deleter = fixtures::identity::create(&app, &fixtures::now()).await;
+    let channel = fixtures::channel::fictitious();
+    let super::Error(error) = super::handler(
+        State(app.clone()),
+        Path(channel.clone()),
+        fixtures::now(),
+        deleter,
+    )
+    .await
+    .expect_err("deleting a nonexistent channel fails");
+
+    // Verify the response
+
+    assert!(matches!(error, app::DeleteError::NotFound(id) if id == channel));
+}
+
+#[tokio::test]
+pub async fn channel_deleted() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+
+    app.channels()
+        .delete(&channel.id, &fixtures::now())
+        .await
+        .expect("deleting a recently-sent channel succeeds");
+
+    // Send the request
+
+    let deleter = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Error(error) = super::handler(
+        State(app.clone()),
+        Path(channel.id.clone()),
+        fixtures::now(),
+        deleter,
+    )
+    .await
+    .expect_err("deleting a deleted channel fails");
+
+    // Verify the response
+
+    assert!(matches!(error, app::DeleteError::Deleted(id) if id == channel.id));
+}
+
+#[tokio::test]
+pub async fn channel_expired() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
+
+    app.channels()
+        .expire(&fixtures::now())
+        .await
+        .expect("expiring channels always succeeds");
+
+    // Send the request
+
+    let deleter = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Error(error) = super::handler(
+        State(app.clone()),
+        Path(channel.id.clone()),
+        fixtures::now(),
+        deleter,
+    )
+    .await
+    .expect_err("deleting an expired channel fails");
+
+    // Verify the response
+
+    assert!(matches!(error, app::DeleteError::Deleted(id) if id == channel.id));
+}
+
+#[tokio::test]
+pub async fn channel_purged() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
+
+    app.channels()
+        .expire(&fixtures::old())
+        .await
+        .expect("expiring channels always succeeds");
+
+    app.channels()
+        .purge(&fixtures::now())
+        .await
+        .expect("purging channels always succeeds");
+
+    // Send the request
+
+    let deleter = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Error(error) = super::handler(
+        State(app.clone()),
+        Path(channel.id.clone()),
+        fixtures::now(),
+        deleter,
+    )
+    .await
+    .expect_err("deleting a purged channel fails");
+
+    // Verify the response
+
+    assert!(matches!(error, app::DeleteError::NotFound(id) if id == channel.id));
+}
+
+#[tokio::test]
+pub async fn channel_not_empty() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let sender = fixtures::user::create(&app, &fixtures::now()).await;
+    fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
+
+    // Send the request
+
+    let deleter = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Error(error) = super::handler(
+        State(app.clone()),
+        Path(channel.id.clone()),
+        fixtures::now(),
+        deleter,
+    )
+    .await
+    .expect_err("deleting a channel with messages fails");
+
+    // Verify the response
+
+    assert!(matches!(error, app::DeleteError::NotEmpty(id) if id == channel.id));
+}
diff --git a/src/channel/handlers/mod.rs b/src/channel/handlers/mod.rs
new file mode 100644
index 0000000..f2ffd0d
--- /dev/null
+++ b/src/channel/handlers/mod.rs
@@ -0,0 +1,9 @@
+mod create;
+mod delete;
+mod send;
+
+pub use create::handler as create;
+pub use delete::handler as delete;
+pub use send::handler as send;
+
+type PathInfo = crate::channel::Id;
diff --git a/src/channel/handlers/send/mod.rs b/src/channel/handlers/send/mod.rs
new file mode 100644
index 0000000..aa241e2
--- /dev/null
+++ b/src/channel/handlers/send/mod.rs
@@ -0,0 +1,61 @@
+use axum::{
+    extract::{Json, Path, State},
+    http::StatusCode,
+    response::{self, IntoResponse},
+};
+
+use crate::channel::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(channel): Path<PathInfo>,
+    RequestedAt(sent_at): RequestedAt,
+    identity: Identity,
+    Json(request): Json<Request>,
+) -> Result<Response, Error> {
+    let message = app
+        .messages()
+        .send(&channel, &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::ChannelNotFound(_) => NotFound(error).into_response(),
+            SendError::Name(_) | SendError::Database(_) => Internal::from(error).into_response(),
+        }
+    }
+}
diff --git a/src/channel/handlers/send/test.rs b/src/channel/handlers/send/test.rs
new file mode 100644
index 0000000..f43f901
--- /dev/null
+++ b/src/channel/handlers/send/test.rs
@@ -0,0 +1,131 @@
+use axum::extract::{Json, Path, State};
+use futures::stream::{self, StreamExt as _};
+
+use crate::{
+    channel,
+    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 channel = fixtures::channel::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(channel.id.clone()),
+            sent_at.clone(),
+            sender.clone(),
+            Json(request),
+        )
+        .await
+        .expect("sending to a valid channel succeeds");
+    }
+
+    // Verify the semantics
+
+    let mut events = app
+        .events()
+        .subscribe(resume_point)
+        .await
+        .expect("subscribing to a valid channel succeeds")
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::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_channel() {
+    // 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 channel = channel::Id::generate();
+    let request = super::Request {
+        body: fixtures::message::propose(),
+    };
+    let super::Error(error) = super::handler(
+        State(app),
+        Path(channel.clone()),
+        sent_at,
+        sender,
+        Json(request),
+    )
+    .await
+    .expect_err("sending to a nonexistent channel fails");
+
+    // Verify the structure of the response
+
+    assert!(matches!(
+        error,
+        SendError::ChannelNotFound(error_channel) if channel == error_channel
+    ));
+}
+
+#[tokio::test]
+async fn deleted_channel() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let sender = fixtures::identity::create(&app, &fixtures::now()).await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+
+    app.channels()
+        .delete(&channel.id, &fixtures::now())
+        .await
+        .expect("deleting a new channel succeeds");
+
+    // Call the endpoint
+
+    let sent_at = fixtures::now();
+    let channel = channel::Id::generate();
+    let request = super::Request {
+        body: fixtures::message::propose(),
+    };
+    let super::Error(error) = super::handler(
+        State(app),
+        Path(channel.clone()),
+        sent_at,
+        sender,
+        Json(request),
+    )
+    .await
+    .expect_err("sending to a deleted channel fails");
+
+    // Verify the structure of the response
+
+    assert!(matches!(
+        error,
+        SendError::ChannelNotFound(error_channel) if channel == error_channel
+    ));
+}