Store `Conversation` instances using their events.

This replaces the approach of having the repo type know about conversation lifecycle in detail. Instead, the repo type accepts events and applies them to the DB blindly. The SQL written to implement each event does, however, embed assumptions about what order events will happen in.

1 files changed, 72 insertions, 66 deletions

diff --git a/src/conversation/repo.rs b/src/conversation/repo.rs
index 7e38b62..cb66bf8 100644
--- a/src/conversation/repo.rs
+++ b/src/conversation/repo.rs
@@ -1,9 +1,12 @@
 use futures::stream::{StreamExt as _, TryStreamExt as _};
 use sqlx::{SqliteConnection, Transaction, sqlite::Sqlite};
 
+use super::{
+    Conversation, Event, History, Id,
+    event::{Created, Deleted},
+};
 use crate::{
     clock::DateTime,
-    conversation::{Conversation, History, Id},
     db::NotFound,
     event::{Instant, Sequence},
     name::{self, Name},
@@ -22,22 +25,41 @@ impl Provider for Transaction<'_, Sqlite> {
 pub struct Conversations<'t>(&'t mut SqliteConnection);
 
 impl Conversations<'_> {
-    pub async fn create(&mut self, name: &Name, created: &Instant) -> Result<History, sqlx::Error> {
-        let id = Id::generate();
-        let name = name.clone();
+    pub async fn record_events(
+        &mut self,
+        events: impl IntoIterator<Item = Event>,
+    ) -> Result<(), sqlx::Error> {
+        for event in events {
+            self.record_event(&event).await?;
+        }
+        Ok(())
+    }
+
+    pub async fn record_event(&mut self, event: &Event) -> Result<(), sqlx::Error> {
+        match event {
+            Event::Created(created) => self.record_created(created).await,
+            Event::Deleted(deleted) => self.record_deleted(deleted).await,
+        }
+    }
+
+    async fn record_created(&mut self, created: &Created) -> Result<(), sqlx::Error> {
+        let Conversation {
+            id,
+            created,
+            name,
+            deleted: _,
+        } = &created.conversation;
         let display_name = name.display();
         let canonical_name = name.canonical();
-        let created = *created;
 
         sqlx::query!(
             r#"
                 insert into conversation (id, created_at, created_sequence, last_sequence)
-                values ($1, $2, $3, $4)
+                values ($1, $2, $3, $3)
             "#,
             id,
             created.at,
             created.sequence,
-            created.sequence,
         )
         .execute(&mut *self.0)
         .await?;
@@ -54,16 +76,50 @@ impl Conversations<'_> {
         .execute(&mut *self.0)
         .await?;
 
-        let conversation = History {
-            conversation: Conversation {
-                created,
-                id,
-                name: name.clone(),
-                deleted: None,
-            },
-        };
+        Ok(())
+    }
 
-        Ok(conversation)
+    async fn record_deleted(&mut self, deleted: &Deleted) -> Result<(), sqlx::Error> {
+        let Deleted { instant, id } = deleted;
+        sqlx::query!(
+            r#"
+                update conversation
+                set last_sequence = max(last_sequence, $1)
+                where id = $2
+            "#,
+            instant.sequence,
+            id,
+        )
+        .execute(&mut *self.0)
+        .await?;
+
+        sqlx::query!(
+            r#"
+                insert into conversation_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 conversation is deleted, its
+        // name is retconned to have been the empty string. Someone reading the event
+        // stream afterwards, or looking at conversations via the API, cannot retrieve
+        // the "deleted" conversation's information by ignoring the deletion event.
+        sqlx::query!(
+            r#"
+                delete from conversation_name
+                where id = $1
+            "#,
+            id,
+        )
+        .execute(&mut *self.0)
+        .await?;
+
+        Ok(())
     }
 
     pub async fn by_id(&mut self, conversation: &Id) -> Result<History, LoadError> {
@@ -179,56 +235,6 @@ impl Conversations<'_> {
         Ok(conversations)
     }
 
-    pub async fn delete(
-        &mut self,
-        conversation: &History,
-        deleted: &Instant,
-    ) -> Result<History, LoadError> {
-        let id = conversation.id();
-        sqlx::query!(
-            r#"
-                update conversation
-                set last_sequence = max(last_sequence, $1)
-                where id = $2
-                returning id as "id: Id"
-            "#,
-            deleted.sequence,
-            id,
-        )
-        .fetch_one(&mut *self.0)
-        .await?;
-
-        sqlx::query!(
-            r#"
-                insert into conversation_deleted (id, deleted_at, deleted_sequence)
-                values ($1, $2, $3)
-            "#,
-            id,
-            deleted.at,
-            deleted.sequence,
-        )
-        .execute(&mut *self.0)
-        .await?;
-
-        // Small social responsibility hack here: when a conversation is deleted, its
-        // name is retconned to have been the empty string. Someone reading the event
-        // stream afterwards, or looking at conversations via the API, cannot retrieve
-        // the "deleted" conversation's information by ignoring the deletion event.
-        sqlx::query!(
-            r#"
-                delete from conversation_name
-                where id = $1
-            "#,
-            id,
-        )
-        .execute(&mut *self.0)
-        .await?;
-
-        let conversation = self.by_id(id).await?;
-
-        Ok(conversation)
-    }
-
     pub async fn purge(&mut self, purge_at: &DateTime) -> Result<(), sqlx::Error> {
         let conversations = sqlx::query_scalar!(
             r#"