1 files changed, 28 insertions, 94 deletions

diff --git a/src/event/app.rs b/src/event/app.rs
index 5e9e79a..e58bea9 100644
--- a/src/event/app.rs
+++ b/src/event/app.rs
@@ -1,22 +1,15 @@
-use chrono::TimeDelta;
 use futures::{
     future,
     stream::{self, StreamExt as _},
     Stream,
 };
+use itertools::Itertools as _;
 use sqlx::sqlite::SqlitePool;
 
-use super::{
-    broadcaster::Broadcaster,
-    repo::message::Provider as _,
-    types::{self, ChannelEvent},
-};
+use super::{broadcaster::Broadcaster, Event, Sequence, Sequenced};
 use crate::{
     channel::{self, repo::Provider as _},
-    clock::DateTime,
-    db::NotFound as _,
-    event::{repo::Provider as _, Sequence},
-    login::Login,
+    message::{self, repo::Provider as _},
 };
 
 pub struct Events<'a> {
@@ -29,111 +22,52 @@ impl<'a> Events<'a> {
         Self { db, events }
     }
 
-    pub async fn send(
-        &self,
-        login: &Login,
-        channel: &channel::Id,
-        body: &str,
-        sent_at: &DateTime,
-    ) -> Result<types::ChannelEvent, EventsError> {
-        let mut tx = self.db.begin().await?;
-        let channel = tx
-            .channels()
-            .by_id(channel)
-            .await
-            .not_found(|| EventsError::ChannelNotFound(channel.clone()))?;
-        let sent = tx.sequence().next(sent_at).await?;
-        let event = tx
-            .message_events()
-            .create(login, &channel, &sent, body)
-            .await?;
-        tx.commit().await?;
-
-        self.events.broadcast(&event);
-        Ok(event)
-    }
-
-    pub async fn expire(&self, relative_to: &DateTime) -> Result<(), sqlx::Error> {
-        // Somewhat arbitrarily, expire after 90 days.
-        let expire_at = relative_to.to_owned() - TimeDelta::days(90);
-
-        let mut tx = self.db.begin().await?;
-        let expired = tx.message_events().expired(&expire_at).await?;
-
-        let mut events = Vec::with_capacity(expired.len());
-        for (channel, message) in expired {
-            let deleted = tx.sequence().next(relative_to).await?;
-            let event = tx
-                .message_events()
-                .delete(&channel, &message, &deleted)
-                .await?;
-            events.push(event);
-        }
-
-        tx.commit().await?;
-
-        for event in events {
-            self.events.broadcast(&event);
-        }
-
-        Ok(())
-    }
-
     pub async fn subscribe(
         &self,
         resume_at: Option<Sequence>,
-    ) -> Result<impl Stream<Item = types::ChannelEvent> + std::fmt::Debug, sqlx::Error> {
+    ) -> Result<impl Stream<Item = Event> + std::fmt::Debug, sqlx::Error> {
         // Subscribe before retrieving, to catch messages broadcast while we're
         // querying the DB. We'll prune out duplicates later.
         let live_messages = self.events.subscribe();
 
         let mut tx = self.db.begin().await?;
-        let channels = tx.channels().replay(resume_at).await?;
 
+        let channels = tx.channels().replay(resume_at).await?;
         let channel_events = channels
-            .into_iter()
-            .map(ChannelEvent::created)
-            .filter(move |event| {
-                resume_at.map_or(true, |resume_at| Sequence::from(event) > resume_at)
-            });
-
-        let message_events = tx.message_events().replay(resume_at).await?;
-
-        let mut replay_events = channel_events
-            .into_iter()
-            .chain(message_events.into_iter())
+            .iter()
+            .map(channel::History::events)
+            .kmerge_by(|a, b| a.instant.sequence < b.instant.sequence)
+            .filter(Sequence::after(resume_at))
+            .map(Event::from);
+
+        let messages = tx.messages().replay(resume_at).await?;
+        let message_events = messages
+            .iter()
+            .map(message::History::events)
+            .kmerge_by(|a, b| a.instant.sequence < b.instant.sequence)
+            .filter(Sequence::after(resume_at))
+            .map(Event::from);
+
+        let replay_events = channel_events
+            .merge_by(message_events, |a, b| {
+                a.instant.sequence < b.instant.sequence
+            })
             .collect::<Vec<_>>();
-        replay_events.sort_by_key(|event| Sequence::from(event));
-        let resume_live_at = replay_events.last().map(Sequence::from);
+        let resume_live_at = replay_events.last().map(Sequenced::sequence);
 
         let replay = stream::iter(replay_events);
 
-        // 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 live_messages = live_messages
             // Filtering on the broadcast resume point filters out messages
             // before resume_at, and filters out messages duplicated from
-            // stored_messages.
+            // `replay_events`.
             .filter(Self::resume(resume_live_at));
 
         Ok(replay.chain(live_messages))
     }
 
-    fn resume(
-        resume_at: Option<Sequence>,
-    ) -> impl for<'m> FnMut(&'m types::ChannelEvent) -> future::Ready<bool> {
-        move |event| future::ready(resume_at < Some(Sequence::from(event)))
+    fn resume(resume_at: Option<Sequence>) -> impl for<'m> FnMut(&'m Event) -> future::Ready<bool> {
+        let filter = Sequence::after(resume_at);
+        move |event| future::ready(filter(event))
     }
 }
-
-#[derive(Debug, thiserror::Error)]
-pub enum EventsError {
-    #[error("channel {0} not found")]
-    ChannelNotFound(channel::Id),
-    #[error(transparent)]
-    DatabaseError(#[from] sqlx::Error),
-}