Implement storage of synchronized entities in terms of events, not state.

Conversations, users, messages, and all other "synchronized" entities now have an in-memory implementation of their lifecycle, rather than a database-backed one. These operations take a history, apply one lifecycle change to that history, and emit a new history. Storage is then implemented by applying the events in this new history to the database.

The storage methods in repo types, which process these events by emitting SQL statements, make necessary assumptions that the events being passed to them are coherent with the data already in storage. For example, the code to handle a conversation's delete event is allowed to assume that the database already contains a row for that conversation, inserted in response to a prior conversation creation event.

Data retrieval is not modified in this commit, and probably never will be without a more thorough storage rewrite. The whole intention of the data modelling approach I've been using is that a single row per entity represents its entire history, in turn so that the data in the database should be legible to people approaching it using normal SQL tools.

Developed as an aesthetic response to increasing unease with the lack of an ORM versus the boring-ness of our actual queries.

Merges event-based-storage into main.

1 files changed, 73 insertions, 75 deletions

diff --git a/src/message/repo.rs b/src/message/repo.rs
index 83bf0d5..4f66bdc 100644
--- a/src/message/repo.rs
+++ b/src/message/repo.rs
@@ -1,11 +1,14 @@
 use sqlx::{SqliteConnection, Transaction, sqlite::Sqlite};
 
-use super::{Body, History, Id, snapshot::Message};
+use super::{
+    Body, Event, History, Id, Message,
+    event::{Deleted, Sent},
+};
 use crate::{
     clock::DateTime,
-    conversation::{self, Conversation},
+    conversation,
     event::{Instant, Sequence},
-    user::{self, User},
+    user,
 };
 
 pub trait Provider {
@@ -21,50 +24,84 @@ impl Provider for Transaction<'_, Sqlite> {
 pub struct Messages<'t>(&'t mut SqliteConnection);
 
 impl Messages<'_> {
-    pub async fn create(
+    pub async fn record_events(
         &mut self,
-        conversation: &Conversation,
-        sender: &User,
-        sent: &Instant,
-        body: &Body,
-    ) -> Result<History, sqlx::Error> {
-        let id = Id::generate();
+        events: impl IntoIterator<Item = Event>,
+    ) -> Result<(), sqlx::Error> {
+        for event in events {
+            self.record_event(&event).await?;
+        }
+        Ok(())
+    }
 
-        let message = sqlx::query!(
+    pub async fn record_event(&mut self, event: &Event) -> Result<(), sqlx::Error> {
+        match event {
+            Event::Sent(sent) => self.record_sent(sent).await,
+            Event::Deleted(deleted) => self.record_deleted(deleted).await,
+        }
+    }
+
+    async fn record_sent(&mut self, sent: &Sent) -> Result<(), sqlx::Error> {
+        let Message {
+            id,
+            conversation,
+            sender,
+            body,
+            sent,
+            deleted: _,
+        } = &sent.message;
+
+        sqlx::query!(
             r#"
                 insert into message
-                    (id, conversation, sender, sent_at, sent_sequence, body, last_sequence)
-                values ($1, $2, $3, $4, $5, $6, $7)
-                returning
-                    id as "id: Id",
-                    conversation as "conversation: conversation::Id",
-                    sender as "sender: user::Id",
-                    sent_at as "sent_at: DateTime",
-                    sent_sequence as "sent_sequence: Sequence",
-                    body as "body: Body"
+                    (id, conversation, sender, body, sent_at, sent_sequence, last_sequence)
+                values ($1, $2, $3, $4, $5, $6, $6)
             "#,
             id,
-            conversation.id,
-            sender.id,
-            sent.at,
-            sent.sequence,
+            conversation,
+            sender,
             body,
+            sent.at,
             sent.sequence,
         )
-        .map(|row| History {
-            message: Message {
-                sent: Instant::new(row.sent_at, row.sent_sequence),
-                conversation: row.conversation,
-                sender: row.sender,
-                id: row.id,
-                body: row.body.unwrap_or_default(),
-                deleted: None,
-            },
-        })
-        .fetch_one(&mut *self.0)
+        .execute(&mut *self.0)
         .await?;
 
-        Ok(message)
+        Ok(())
+    }
+
+    async fn record_deleted(&mut self, deleted: &Deleted) -> Result<(), sqlx::Error> {
+        let Deleted { instant, id } = deleted;
+
+        sqlx::query!(
+            r#"
+                insert into message_deleted (id, deleted_at, deleted_sequence)
+                values ($1, $2, $3)
+            "#,
+            id,
+            instant.at,
+            instant.sequence,
+        )
+        .execute(&mut *self.0)
+        .await?;
+
+        // Small social responsibility hack here: when a message is deleted, its body is
+        // retconned to have been the empty string. Someone reading the event stream
+        // afterwards, or looking at messages in the conversation, cannot retrieve the
+        // "deleted" message by ignoring the deletion event.
+        sqlx::query!(
+            r#"
+                update message
+                set body = '', last_sequence = max(last_sequence, $1)
+                where id = $2
+            "#,
+            instant.sequence,
+            id,
+        )
+        .execute(&mut *self.0)
+        .await?;
+
+        Ok(())
     }
 
     pub async fn live(
@@ -178,45 +215,6 @@ impl Messages<'_> {
         Ok(message)
     }
 
-    pub async fn delete(
-        &mut self,
-        message: &Message,
-        deleted: &Instant,
-    ) -> Result<History, sqlx::Error> {
-        sqlx::query!(
-            r#"
-                insert into message_deleted (id, deleted_at, deleted_sequence)
-                values ($1, $2, $3)
-            "#,
-            message.id,
-            deleted.at,
-            deleted.sequence,
-        )
-        .execute(&mut *self.0)
-        .await?;
-
-        // Small social responsibility hack here: when a message is deleted, its body is
-        // retconned to have been the empty string. Someone reading the event stream
-        // afterwards, or looking at messages in the conversation, cannot retrieve the
-        // "deleted" message by ignoring the deletion event.
-        sqlx::query!(
-            r#"
-                update message
-                set body = '', last_sequence = max(last_sequence, $1)
-                where id = $2
-                returning id as "id: Id"
-            "#,
-            deleted.sequence,
-            message.id,
-        )
-        .fetch_one(&mut *self.0)
-        .await?;
-
-        let message = self.by_id(&message.id).await?;
-
-        Ok(message)
-    }
-
     pub async fn purge(&mut self, purge_at: &DateTime) -> Result<(), sqlx::Error> {
         let messages = sqlx::query_scalar!(
             r#"