summaryrefslogtreecommitdiff
path: root/src/channel/handlers
diff options
context:
space:
mode:
Diffstat (limited to 'src/channel/handlers')
-rw-r--r--src/channel/handlers/create/mod.rs67
-rw-r--r--src/channel/handlers/create/test.rs242
-rw-r--r--src/channel/handlers/delete/mod.rs59
-rw-r--r--src/channel/handlers/delete/test.rs174
-rw-r--r--src/channel/handlers/mod.rs9
-rw-r--r--src/channel/handlers/send/mod.rs61
-rw-r--r--src/channel/handlers/send/test.rs131
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
+ ));
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-01-02 19:30:40 +0000