1 files changed, 26 insertions, 18 deletions

diff --git a/src/boot/app.rs b/src/boot/app.rs
index cd45c38..89eec12 100644
--- a/src/boot/app.rs
+++ b/src/boot/app.rs
@@ -1,10 +1,11 @@
+use itertools::Itertools as _;
 use sqlx::sqlite::SqlitePool;
 
 use super::Snapshot;
 use crate::{
     channel::{self, repo::Provider as _},
-    event::{Heartbeat, repo::Provider as _},
-    message::repo::Provider as _,
+    event::{Event, Sequence, repo::Provider as _},
+    message::{self, repo::Provider as _},
     name,
     user::{self, repo::Provider as _},
 };
@@ -21,7 +22,6 @@ impl<'a> Boot<'a> {
     pub async fn snapshot(&self) -> Result<Snapshot, Error> {
         let mut tx = self.db.begin().await?;
         let resume_point = tx.sequence().current().await?;
-        let heartbeat = Heartbeat::TIMEOUT;
 
         let users = tx.users().all(resume_point).await?;
         let channels = tx.channels().all(resume_point).await?;
@@ -29,27 +29,35 @@ impl<'a> Boot<'a> {
 
         tx.commit().await?;
 
-        let users = users
-            .into_iter()
-            .filter_map(|user| user.as_of(resume_point))
-            .collect();
+        let user_events = users
+            .iter()
+            .map(user::History::events)
+            .kmerge_by(Sequence::merge)
+            .filter(Sequence::up_to(resume_point))
+            .map(Event::from);
 
-        let channels = channels
-            .into_iter()
-            .filter_map(|channel| channel.as_of(resume_point))
-            .collect();
+        let channel_events = channels
+            .iter()
+            .map(channel::History::events)
+            .kmerge_by(Sequence::merge)
+            .filter(Sequence::up_to(resume_point))
+            .map(Event::from);
+
+        let message_events = messages
+            .iter()
+            .map(message::History::events)
+            .kmerge_by(Sequence::merge)
+            .filter(Sequence::up_to(resume_point))
+            .map(Event::from);
 
-        let messages = messages
-            .into_iter()
-            .filter_map(|message| message.as_of(resume_point))
+        let events = user_events
+            .merge_by(channel_events, Sequence::merge)
+            .merge_by(message_events, Sequence::merge)
             .collect();
 
         Ok(Snapshot {
             resume_point,
-            heartbeat,
-            users,
-            channels,
-            messages,
+            events,
         })
     }
 }