Code organization changes considered during implementation of vector-of-sequence-numbers stream resume.

3 files changed, 11 insertions, 134 deletions

diff --git a/src/channel/app.rs b/src/channel/app.rs
index 2da25d2..bb87734 100644
--- a/src/channel/app.rs
+++ b/src/channel/app.rs
@@ -1,22 +1,8 @@
-use chrono::TimeDelta;
-use futures::{
-    future,
-    stream::{self, StreamExt as _},
-    Stream,
-};
 use sqlx::sqlite::SqlitePool;
 
 use crate::{
-    clock::DateTime,
-    events::{
-        app::Broadcaster,
-        repo::broadcast::{self, Provider as _},
-    },
-    repo::{
-        channel::{self, Channel, Provider as _},
-        error::NotFound as _,
-        login::Login,
-    },
+    events::app::Broadcaster,
+    repo::channel::{Channel, Provider as _},
 };
 
 pub struct Channels<'a> {
@@ -49,107 +35,6 @@ impl<'a> Channels<'a> {
 
         Ok(channels)
     }
-
-    pub async fn send(
-        &self,
-        login: &Login,
-        channel: &channel::Id,
-        body: &str,
-        sent_at: &DateTime,
-    ) -> Result<broadcast::Message, EventsError> {
-        let mut tx = self.db.begin().await?;
-        let channel = tx
-            .channels()
-            .by_id(channel)
-            .await
-            .not_found(|| EventsError::ChannelNotFound(channel.clone()))?;
-        let message = tx
-            .broadcast()
-            .create(login, &channel, body, sent_at)
-            .await?;
-        tx.commit().await?;
-
-        self.broadcaster.broadcast(&channel.id, &message);
-        Ok(message)
-    }
-
-    pub async fn events(
-        &self,
-        channel: &channel::Id,
-        subscribed_at: &DateTime,
-        resume_at: Option<broadcast::Sequence>,
-    ) -> Result<impl Stream<Item = broadcast::Message> + std::fmt::Debug, EventsError> {
-        // Somewhat arbitrarily, expire after 90 days.
-        let expire_at = subscribed_at.to_owned() - TimeDelta::days(90);
-
-        let mut tx = self.db.begin().await?;
-        let channel = tx
-            .channels()
-            .by_id(channel)
-            .await
-            .not_found(|| EventsError::ChannelNotFound(channel.clone()))?;
-
-        // Subscribe before retrieving, to catch messages broadcast while we're
-        // querying the DB. We'll prune out duplicates later.
-        let live_messages = self.broadcaster.listen(&channel.id);
-
-        tx.broadcast().expire(&expire_at).await?;
-        let stored_messages = tx.broadcast().replay(&channel, resume_at).await?;
-        tx.commit().await?;
-
-        let resume_broadcast_at = stored_messages
-            .last()
-            .map(|message| message.sequence)
-            .or(resume_at);
-
-        // This should always be the case, up to integer rollover, primarily
-        // because every message in stored_messages has a sequence not less
-        // than `resume_at`, or `resume_at` is None. We use the last message
-        // (if any) to decide when to resume the `live_messages` stream.
-        //
-        // It probably simplifies to assert!(resume_at <= resume_broadcast_at), but
-        // this form captures more of the reasoning.
-        assert!(
-            (resume_at.is_none() && resume_broadcast_at.is_none())
-                || (stored_messages.is_empty() && resume_at == resume_broadcast_at)
-                || resume_at < resume_broadcast_at
-        );
-
-        // no skip_expired or resume transforms for stored_messages, as it's
-        // constructed not to contain messages meeting either criterion.
-        //
-        // * skip_expired is redundant with the `tx.broadcasts().expire(…)` call;
-        // * resume is redundant with the resume_at argument to
-        //   `tx.broadcasts().replay(…)`.
-        let stored_messages = stream::iter(stored_messages);
-        let live_messages = live_messages
-            // Sure, it's temporally improbable that we'll ever skip a message
-            // that's 90 days old, but there's no reason not to be thorough.
-            .filter(Self::skip_expired(&expire_at))
-            // Filtering on the broadcast resume point filters out messages
-            // before resume_at, and filters out messages duplicated from
-            // stored_messages.
-            .filter(Self::resume(resume_broadcast_at));
-
-        Ok(stored_messages.chain(live_messages))
-    }
-
-    fn resume(
-        resume_at: Option<broadcast::Sequence>,
-    ) -> impl for<'m> FnMut(&'m broadcast::Message) -> future::Ready<bool> {
-        move |msg| {
-            future::ready(match resume_at {
-                None => true,
-                Some(resume_at) => msg.sequence > resume_at,
-            })
-        }
-    }
-    fn skip_expired(
-        expire_at: &DateTime,
-    ) -> impl for<'m> FnMut(&'m broadcast::Message) -> future::Ready<bool> {
-        let expire_at = expire_at.to_owned();
-        move |msg| future::ready(msg.sent_at > expire_at)
-    }
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -177,13 +62,3 @@ pub enum InternalError {
     #[error(transparent)]
     DatabaseError(#[from] sqlx::Error),
 }
-
-#[derive(Debug, thiserror::Error)]
-pub enum EventsError {
-    #[error("channel {0} not found")]
-    ChannelNotFound(channel::Id),
-    #[error(transparent)]
-    ResumeAtError(#[from] chrono::ParseError),
-    #[error(transparent)]
-    DatabaseError(#[from] sqlx::Error),
-}
diff --git a/src/channel/routes.rs b/src/channel/routes.rs
index 674c876..bb6cde6 100644
--- a/src/channel/routes.rs
+++ b/src/channel/routes.rs
@@ -6,11 +6,12 @@ use axum::{
     Router,
 };
 
-use super::app::{self, EventsError};
+use super::app;
 use crate::{
     app::App,
     clock::RequestedAt,
     error::InternalError,
+    events::app::EventsError,
     repo::{
         channel::{self, Channel},
         login::Login,
@@ -89,7 +90,7 @@ async fn on_send(
     login: Login,
     Json(request): Json<SendRequest>,
 ) -> Result<StatusCode, ErrorResponse> {
-    app.channels()
+    app.events()
         .send(&login, &channel, &request.message, &sent_at)
         .await
         // Could impl `From` here, but it's more code and this is used once.
diff --git a/src/channel/routes/test/on_send.rs b/src/channel/routes/test/on_send.rs
index eab7c32..6690374 100644
--- a/src/channel/routes/test/on_send.rs
+++ b/src/channel/routes/test/on_send.rs
@@ -5,7 +5,8 @@ use axum::{
 use futures::stream::StreamExt;
 
 use crate::{
-    channel::{app, routes},
+    channel::routes,
+    events::app,
     repo::channel,
     test::fixtures::{self, future::Immediately as _},
 };
@@ -42,8 +43,8 @@ async fn channel_exists() {
 
     let subscribed_at = fixtures::now();
     let mut events = app
-        .channels()
-        .events(&channel.id, &subscribed_at, None)
+        .events()
+        .subscribe(&channel.id, &subscribed_at, None)
         .await
         .expect("subscribing to a valid channel");
 
@@ -99,8 +100,8 @@ async fn messages_in_order() {
 
     let subscribed_at = fixtures::now();
     let events = app
-        .channels()
-        .events(&channel.id, &subscribed_at, None)
+        .events()
+        .subscribe(&channel.id, &subscribed_at, None)
         .await
         .expect("subscribing to a valid channel")
         .take(requests.len());