Merge branch 'main' into wip/path-routing-for-channels

4 files changed, 52 insertions, 63 deletions

diff --git a/src/event/app.rs b/src/event/app.rs
index 141037d..951ce25 100644
--- a/src/event/app.rs
+++ b/src/event/app.rs
@@ -9,6 +9,7 @@ use sqlx::sqlite::SqlitePool;
 use super::{broadcaster::Broadcaster, Event, ResumePoint, Sequence, Sequenced};
 use crate::{
     channel::{self, repo::Provider as _},
+    login::{self, repo::Provider as _},
     message::{self, repo::Provider as _},
 };
 
@@ -33,6 +34,14 @@ impl<'a> Events<'a> {
 
         let mut tx = self.db.begin().await?;
 
+        let logins = tx.logins().replay(resume_at).await?;
+        let login_events = logins
+            .iter()
+            .map(login::History::events)
+            .kmerge_by(Sequence::merge)
+            .filter(Sequence::after(resume_at))
+            .map(Event::from);
+
         let channels = tx.channels().replay(resume_at).await?;
         let channel_events = channels
             .iter()
@@ -49,7 +58,8 @@ impl<'a> Events<'a> {
             .filter(Sequence::after(resume_at))
             .map(Event::from);
 
-        let replay_events = channel_events
+        let replay_events = login_events
+            .merge_by(channel_events, Sequence::merge)
             .merge_by(message_events, Sequence::merge)
             .collect::<Vec<_>>();
         let resume_live_at = replay_events.last().map(Sequenced::sequence);
diff --git a/src/event/mod.rs b/src/event/mod.rs
index e748d66..69c7a10 100644
--- a/src/event/mod.rs
+++ b/src/event/mod.rs
@@ -1,13 +1,14 @@
-use crate::{channel, message};
+use crate::{channel, login, message};
 
 pub mod app;
-pub mod broadcaster;
+mod broadcaster;
 mod extract;
 pub mod repo;
 mod routes;
 mod sequence;
 
 pub use self::{
+    broadcaster::Broadcaster,
     routes::router,
     sequence::{Instant, Sequence, Sequenced},
 };
@@ -15,63 +16,37 @@ pub use self::{
 pub type ResumePoint = Option<Sequence>;
 
 #[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
-pub struct Event {
-    #[serde(flatten)]
-    pub instant: Instant,
-    #[serde(flatten)]
-    pub kind: Kind,
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum Event {
+    Login(login::Event),
+    Channel(channel::Event),
+    Message(message::Event),
 }
 
 impl Sequenced for Event {
     fn instant(&self) -> Instant {
-        self.instant
-    }
-}
-
-impl From<channel::Event> for Event {
-    fn from(event: channel::Event) -> Self {
-        Self {
-            instant: event.instant,
-            kind: event.kind.into(),
+        match self {
+            Self::Login(event) => event.instant(),
+            Self::Channel(event) => event.instant(),
+            Self::Message(event) => event.instant(),
         }
     }
 }
 
-impl From<message::Event> for Event {
-    fn from(event: message::Event) -> Self {
-        Self {
-            instant: event.instant(),
-            kind: event.kind.into(),
-        }
+impl From<login::Event> for Event {
+    fn from(event: login::Event) -> Self {
+        Self::Login(event)
     }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
-#[serde(tag = "type", rename_all = "snake_case")]
-pub enum Kind {
-    #[serde(rename = "created")]
-    ChannelCreated(channel::event::Created),
-    #[serde(rename = "message")]
-    MessageSent(message::event::Sent),
-    MessageDeleted(message::event::Deleted),
-    #[serde(rename = "deleted")]
-    ChannelDeleted(channel::event::Deleted),
-}
-
-impl From<channel::event::Kind> for Kind {
-    fn from(kind: channel::event::Kind) -> Self {
-        match kind {
-            channel::event::Kind::Created(created) => Self::ChannelCreated(created),
-            channel::event::Kind::Deleted(deleted) => Self::ChannelDeleted(deleted),
-        }
+impl From<channel::Event> for Event {
+    fn from(event: channel::Event) -> Self {
+        Self::Channel(event)
     }
 }
 
-impl From<message::event::Kind> for Kind {
-    fn from(kind: message::event::Kind) -> Self {
-        match kind {
-            message::event::Kind::Sent(created) => Self::MessageSent(created),
-            message::event::Kind::Deleted(deleted) => Self::MessageDeleted(deleted),
-        }
+impl From<message::Event> for Event {
+    fn from(event: message::Event) -> Self {
+        Self::Message(event)
     }
 }
diff --git a/src/event/routes/test.rs b/src/event/routes/test.rs
index ba9953e..209a016 100644
--- a/src/event/routes/test.rs
+++ b/src/event/routes/test.rs
@@ -15,13 +15,13 @@ async fn includes_historical_message() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let sender = fixtures::login::create(&app).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
     let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
 
     // Call the endpoint
 
-    let subscriber_creds = fixtures::login::create_with_password(&app).await;
+    let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
     let routes::Events(events) = routes::events(State(app), subscriber, None, Query::default())
         .await
@@ -48,7 +48,7 @@ async fn includes_live_message() {
 
     // Call the endpoint
 
-    let subscriber_creds = fixtures::login::create_with_password(&app).await;
+    let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
     let routes::Events(events) =
         routes::events(State(app.clone()), subscriber, None, Query::default())
@@ -57,7 +57,7 @@ async fn includes_live_message() {
 
     // Verify the semantics
 
-    let sender = fixtures::login::create(&app).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
     let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
 
     let event = events
@@ -75,7 +75,7 @@ async fn includes_multiple_channels() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let sender = fixtures::login::create(&app).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
 
     let channels = [
         fixtures::channel::create(&app, &fixtures::now()).await,
@@ -94,7 +94,7 @@ async fn includes_multiple_channels() {
 
     // Call the endpoint
 
-    let subscriber_creds = fixtures::login::create_with_password(&app).await;
+    let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
     let routes::Events(events) = routes::events(State(app), subscriber, None, Query::default())
         .await
@@ -122,7 +122,7 @@ async fn sequential_messages() {
 
     let app = fixtures::scratch_app().await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
-    let sender = fixtures::login::create(&app).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
 
     let messages = vec![
         fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
@@ -132,7 +132,7 @@ async fn sequential_messages() {
 
     // Call the endpoint
 
-    let subscriber_creds = fixtures::login::create_with_password(&app).await;
+    let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
     let routes::Events(events) = routes::events(State(app), subscriber, None, Query::default())
         .await
@@ -166,7 +166,7 @@ async fn resumes_from() {
 
     let app = fixtures::scratch_app().await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
-    let sender = fixtures::login::create(&app).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
 
     let initial_message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
 
@@ -177,7 +177,7 @@ async fn resumes_from() {
 
     // Call the endpoint
 
-    let subscriber_creds = fixtures::login::create_with_password(&app).await;
+    let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
 
     let resume_at = {
@@ -248,13 +248,13 @@ async fn serial_resume() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let sender = fixtures::login::create(&app).await;
+    let sender = fixtures::login::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;
 
     // Call the endpoint
 
-    let subscriber_creds = fixtures::login::create_with_password(&app).await;
+    let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
 
     let resume_at = {
@@ -372,11 +372,11 @@ async fn terminates_on_token_expiry() {
 
     let app = fixtures::scratch_app().await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
-    let sender = fixtures::login::create(&app).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
 
     // Subscribe via the endpoint
 
-    let subscriber_creds = fixtures::login::create_with_password(&app).await;
+    let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await;
     let subscriber =
         fixtures::identity::identity(&app, &subscriber_creds, &fixtures::ancient()).await;
 
@@ -417,11 +417,11 @@ async fn terminates_on_logout() {
 
     let app = fixtures::scratch_app().await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
-    let sender = fixtures::login::create(&app).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
 
     // Subscribe via the endpoint
 
-    let subscriber_creds = fixtures::login::create_with_password(&app).await;
+    let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await;
     let subscriber_token =
         fixtures::identity::logged_in(&app, &subscriber_creds, &fixtures::now()).await;
     let subscriber =
diff --git a/src/event/sequence.rs b/src/event/sequence.rs
index ceb5bcb..bf6d5b8 100644
--- a/src/event/sequence.rs
+++ b/src/event/sequence.rs
@@ -72,6 +72,10 @@ impl Sequence {
 pub trait Sequenced {
     fn instant(&self) -> Instant;
 
+    fn at(&self) -> DateTime {
+        self.instant().at
+    }
+
     fn sequence(&self) -> Sequence {
         self.instant().into()
     }