Track event sequences globally, not per channel.

Per-channel event sequences were a cute idea, but it made reasoning about event resumption much, much harder (case in point: recovering the order of events in a partially-ordered collection is quadratic, since it's basically graph sort). The minor overhead of a global sequence number is likely tolerable, and this simplifies both the API and the internals.

1 files changed, 36 insertions, 43 deletions

diff --git a/src/events/repo/message.rs b/src/events/repo/message.rs
index f8bae2b..3237553 100644
--- a/src/events/repo/message.rs
+++ b/src/events/repo/message.rs
@@ -2,11 +2,12 @@ use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction};
 
 use crate::{
     clock::DateTime,
-    events::types::{self, Sequence},
+    events::types,
     repo::{
         channel::{self, Channel},
         login::{self, Login},
         message::{self, Message},
+        sequence::Sequence,
     },
 };
 
@@ -27,34 +28,33 @@ impl<'c> Events<'c> {
         &mut self,
         sender: &Login,
         channel: &Channel,
-        body: &str,
         sent_at: &DateTime,
+        sent_sequence: Sequence,
+        body: &str,
     ) -> Result<types::ChannelEvent, sqlx::Error> {
-        let sequence = self.assign_sequence(channel).await?;
-
         let id = message::Id::generate();
 
         let message = sqlx::query!(
             r#"
 				insert into message
-					(id, channel, sequence, sender, body, sent_at)
+					(id, channel, sender, sent_at, sent_sequence, body)
 				values ($1, $2, $3, $4, $5, $6)
 				returning
 					id as "id: message::Id",
-                    sequence as "sequence: Sequence",
 					sender as "sender: login::Id",
-					body,
-					sent_at as "sent_at: DateTime"
+                    sent_at as "sent_at: DateTime",
+                    sent_sequence as "sent_sequence: Sequence",
+					body
 			"#,
             id,
             channel.id,
-            sequence,
             sender.id,
-            body,
             sent_at,
+            sent_sequence,
+            body,
         )
         .map(|row| types::ChannelEvent {
-            sequence: row.sequence,
+            sequence: row.sent_sequence,
             at: row.sent_at,
             data: types::MessageEvent {
                 channel: channel.clone(),
@@ -72,28 +72,12 @@ impl<'c> Events<'c> {
         Ok(message)
     }
 
-    pub async fn assign_sequence(&mut self, channel: &Channel) -> Result<Sequence, sqlx::Error> {
-        let next = sqlx::query_scalar!(
-            r#"
-                update channel
-                set last_sequence = last_sequence + 1
-                where id = $1
-                returning last_sequence as "next_sequence: Sequence"
-            "#,
-            channel.id,
-        )
-        .fetch_one(&mut *self.0)
-        .await?;
-
-        Ok(next)
-    }
-
-    pub async fn delete_expired(
+    pub async fn delete(
         &mut self,
         channel: &Channel,
         message: &message::Id,
-        sequence: Sequence,
         deleted_at: &DateTime,
+        deleted_sequence: Sequence,
     ) -> Result<types::ChannelEvent, sqlx::Error> {
         sqlx::query_scalar!(
             r#"
@@ -107,7 +91,7 @@ impl<'c> Events<'c> {
         .await?;
 
         Ok(types::ChannelEvent {
-            sequence,
+            sequence: deleted_sequence,
             at: *deleted_at,
             data: types::MessageDeletedEvent {
                 channel: channel.clone(),
@@ -127,6 +111,7 @@ impl<'c> Events<'c> {
                     channel.id as "channel_id: channel::Id",
                     channel.name as "channel_name",
                     channel.created_at as "channel_created_at: DateTime",
+                    channel.created_sequence as "channel_created_sequence: Sequence",
                     message.id as "message: message::Id"
                 from message
                     join channel on message.channel = channel.id
@@ -141,6 +126,7 @@ impl<'c> Events<'c> {
                     id: row.channel_id,
                     name: row.channel_name,
                     created_at: row.channel_created_at,
+                    created_sequence: row.channel_created_sequence,
                 },
                 row.message,
             )
@@ -153,32 +139,39 @@ impl<'c> Events<'c> {
 
     pub async fn replay(
         &mut self,
-        channel: &Channel,
         resume_at: Option<Sequence>,
     ) -> Result<Vec<types::ChannelEvent>, sqlx::Error> {
         let events = sqlx::query!(
             r#"
 				select
 					message.id as "id: message::Id",
-                    sequence as "sequence: Sequence",
-					login.id as "sender_id: login::Id",
-					login.name as sender_name,
-					message.body,
-					message.sent_at as "sent_at: DateTime"
+                    channel.id as "channel_id: channel::Id",
+                    channel.name as "channel_name",
+                    channel.created_at as "channel_created_at: DateTime",
+                    channel.created_sequence as "channel_created_sequence: Sequence",
+					sender.id as "sender_id: login::Id",
+					sender.name as sender_name,
+                    message.sent_at as "sent_at: DateTime",
+                    message.sent_sequence as "sent_sequence: Sequence",
+                    message.body
 				from message
-					join login on message.sender = login.id
-				where channel = $1
-					and coalesce(sequence > $2, true)
-				order by sequence asc
+                    join channel on message.channel = channel.id
+					join login as sender on message.sender = sender.id
+				where coalesce(message.sent_sequence > $1, true)
+				order by sent_sequence asc
 			"#,
-            channel.id,
             resume_at,
         )
         .map(|row| types::ChannelEvent {
-            sequence: row.sequence,
+            sequence: row.sent_sequence,
             at: row.sent_at,
             data: types::MessageEvent {
-                channel: channel.clone(),
+                channel: Channel {
+                    id: row.channel_id,
+                    name: row.channel_name,
+                    created_at: row.channel_created_at,
+                    created_sequence: row.channel_created_sequence,
+                },
                 sender: login::Login {
                     id: row.sender_id,
                     name: row.sender_name,