Reorganize and consolidate HTTP routes.

HTTP routes are now defined in a single, unified module, pulling them out of the topical modules they were formerly part of.

This is intended to improve the navigability of the codebase. Previously, finding the handler corresponding to a specific endpoint required prior familiarity, though in practice you could usually guess from topic area. Now, all routes are defined in one place; if you know the path, you can read down the list to find the handler.

Handlers themselves live with the domain they are most appropriately "part of," generally (in this version, universally) in a `handlers` submodule. The handlers themselves have been flattened down; rather than representing a path and a method, they now represent a named operation (which is suspiciously similar to the path in most cases). This means that we no longer have constructs like `crate::ui::routes::ch::channel` - it's now `crate::ui::handlers::channel` instead.

## Disclaimer

I Solemnly Swear I Didn't Change Any Handlers.

## Prior art

I've inadvertently reinvented Django's `urls.py` convention, and I've opted to lean into that.

Merges flatter-routes-reorg into main.

1 files changed, 227 insertions, 0 deletions

diff --git a/src/event/handlers/stream/test/resume.rs b/src/event/handlers/stream/test/resume.rs
new file mode 100644
index 0000000..34fee4d
--- /dev/null
+++ b/src/event/handlers/stream/test/resume.rs
@@ -0,0 +1,227 @@
+use std::future;
+
+use axum::extract::State;
+use axum_extra::extract::Query;
+use futures::stream::{self, StreamExt as _};
+
+use crate::{
+    event::Sequenced as _,
+    test::fixtures::{self, future::Expect as _},
+};
+
+#[tokio::test]
+async fn resume() {
+    // 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;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    let initial_message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
+
+    let later_messages = vec![
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+    ];
+
+    // Call the endpoint
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+
+    let resume_at = {
+        // First subscription
+        let super::Response(events) = super::handler(
+            State(app.clone()),
+            subscriber.clone(),
+            None,
+            Query(super::QueryParams { resume_point }),
+        )
+        .await
+        .expect("subscribe never fails");
+
+        let event = events
+            .filter_map(fixtures::event::message)
+            .filter_map(fixtures::event::message::sent)
+            .filter(|event| future::ready(event.message == initial_message))
+            .next()
+            .expect_some("delivered event for initial message")
+            .await;
+
+        event.sequence()
+    };
+
+    // Resume after disconnect
+    let super::Response(resumed) = super::handler(
+        State(app),
+        subscriber,
+        Some(resume_at.into()),
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Verify final events
+
+    let mut events = resumed
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::sent)
+        .zip(stream::iter(later_messages));
+
+    while let Some((event, message)) = events.next().expect_ready("event ready").await {
+        assert_eq!(message, event.message);
+    }
+}
+
+// This test verifies a real bug I hit developing the vector-of-sequences
+// approach to resuming events. A small omission caused the event IDs in a
+// resumed stream to _omit_ channels that were in the original stream until
+// those channels also appeared in the resumed stream.
+//
+// Clients would see something like
+// * In the original stream, Cfoo=5,Cbar=8
+// * In the resumed stream, Cfoo=6 (no Cbar sequence number)
+//
+// Disconnecting and reconnecting a second time, using event IDs from that
+// initial period of the first resume attempt, would then cause the second
+// resume attempt to restart all other channels from the beginning, and not
+// from where the first disconnection happened.
+//
+// As we have switched to a single global event sequence number, this scenario
+// can no longer arise, but this test is preserved because the actual behaviour
+// _is_ a valid way for clients to behave, and should work. We might as well
+// keep testing it.
+#[tokio::test]
+async fn serial_resume() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let sender = fixtures::user::create(&app, &fixtures::now()).await;
+    let channel_a = fixtures::channel::create(&app, &fixtures::now()).await;
+    let channel_b = fixtures::channel::create(&app, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Call the endpoint
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+
+    let resume_at = {
+        let initial_messages = [
+            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await,
+        ];
+
+        // First subscription
+
+        let super::Response(events) = super::handler(
+            State(app.clone()),
+            subscriber.clone(),
+            None,
+            Query(super::QueryParams { resume_point }),
+        )
+        .await
+        .expect("subscribe never fails");
+
+        // Check for expected events
+
+        let events = events
+            .filter_map(fixtures::event::message)
+            .filter_map(fixtures::event::message::sent)
+            .zip(stream::iter(initial_messages))
+            .collect::<Vec<_>>()
+            .expect_ready("zipping a finite list of events is ready immediately")
+            .await;
+
+        assert!(
+            events
+                .iter()
+                .all(|(event, message)| message == &event.message)
+        );
+
+        let (event, _) = events.last().expect("this vec is non-empty");
+
+        // Take the last one's resume point
+
+        event.sequence()
+    };
+
+    // Resume after disconnect
+    let resume_at = {
+        let resume_messages = [
+            // Note that channel_b does not appear here. The buggy behaviour
+            // would be masked if channel_b happened to send a new message
+            // into the resumed event stream.
+            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
+        ];
+
+        // Second subscription
+        let super::Response(events) = super::handler(
+            State(app.clone()),
+            subscriber.clone(),
+            Some(resume_at.into()),
+            Query(super::QueryParams { resume_point }),
+        )
+        .await
+        .expect("subscribe never fails");
+
+        // Check for expected events
+
+        let events = events
+            .filter_map(fixtures::event::message)
+            .filter_map(fixtures::event::message::sent)
+            .zip(stream::iter(resume_messages))
+            .collect::<Vec<_>>()
+            .expect_ready("zipping a finite list of events is ready immediately")
+            .await;
+
+        assert!(
+            events
+                .iter()
+                .all(|(event, message)| message == &event.message)
+        );
+
+        let (event, _) = events.last().expect("this vec is non-empty");
+
+        // Take the last one's resume point
+
+        event.sequence()
+    };
+
+    // Resume after disconnect a second time
+    {
+        // At this point, we can send on either channel and demonstrate the
+        // problem. The resume point should before both of these messages, but
+        // after _all_ prior messages.
+        let final_messages = [
+            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await,
+        ];
+
+        // Third subscription
+        let super::Response(events) = super::handler(
+            State(app.clone()),
+            subscriber.clone(),
+            Some(resume_at.into()),
+            Query(super::QueryParams { resume_point }),
+        )
+        .await
+        .expect("subscribe never fails");
+
+        // Check for expected events
+
+        let events = events
+            .filter_map(fixtures::event::message)
+            .filter_map(fixtures::event::message::sent)
+            .zip(stream::iter(final_messages))
+            .collect::<Vec<_>>()
+            .expect_ready("zipping a finite list of events is ready immediately")
+            .await;
+
+        assert!(
+            events
+                .iter()
+                .all(|(event, message)| message == &event.message)
+        );
+    };
+}