Rename "channel" to "conversation" within the server.

I've split this from the schema and API changes because, frankly, it's huge. Annoyingly so. There are no semantic changes in this, it's all symbol changes, but there are a _lot_ of them because the term "channel" leaks all over everything in a service whose primary role is managing messages sent to channels (now, conversations).

I found a buggy test while working on this! It's not fixed in this commit, because it felt mean to hide a real change in the middle of this much chaff.

1 files changed, 236 insertions, 0 deletions

diff --git a/src/conversation/app.rs b/src/conversation/app.rs
new file mode 100644
index 0000000..81ccdcf
--- /dev/null
+++ b/src/conversation/app.rs
@@ -0,0 +1,236 @@
+use chrono::TimeDelta;
+use itertools::Itertools;
+use sqlx::sqlite::SqlitePool;
+
+use super::{
+    Conversation, Id,
+    repo::{LoadError, Provider as _},
+    validate,
+};
+use crate::{
+    clock::DateTime,
+    db::{Duplicate as _, NotFound as _},
+    event::{Broadcaster, Event, Sequence, repo::Provider as _},
+    message::{self, repo::Provider as _},
+    name::{self, Name},
+};
+
+pub struct Conversations<'a> {
+    db: &'a SqlitePool,
+    events: &'a Broadcaster,
+}
+
+impl<'a> Conversations<'a> {
+    pub const fn new(db: &'a SqlitePool, events: &'a Broadcaster) -> Self {
+        Self { db, events }
+    }
+
+    pub async fn create(
+        &self,
+        name: &Name,
+        created_at: &DateTime,
+    ) -> Result<Conversation, CreateError> {
+        if !validate::name(name) {
+            return Err(CreateError::InvalidName(name.clone()));
+        }
+
+        let mut tx = self.db.begin().await?;
+        let created = tx.sequence().next(created_at).await?;
+        let conversation = tx
+            .conversations()
+            .create(name, &created)
+            .await
+            .duplicate(|| CreateError::DuplicateName(name.clone()))?;
+        tx.commit().await?;
+
+        self.events
+            .broadcast(conversation.events().map(Event::from).collect::<Vec<_>>());
+
+        Ok(conversation.as_created())
+    }
+
+    // This function is careless with respect to time, and gets you the
+    // conversation as it exists in the specific moment when you call it.
+    pub async fn get(&self, conversation: &Id) -> Result<Conversation, Error> {
+        let to_not_found = || Error::NotFound(conversation.clone());
+        let to_deleted = || Error::Deleted(conversation.clone());
+
+        let mut tx = self.db.begin().await?;
+        let conversation = tx
+            .conversations()
+            .by_id(conversation)
+            .await
+            .not_found(to_not_found)?;
+        tx.commit().await?;
+
+        conversation.as_snapshot().ok_or_else(to_deleted)
+    }
+
+    pub async fn delete(
+        &self,
+        conversation: &Id,
+        deleted_at: &DateTime,
+    ) -> Result<(), DeleteError> {
+        let mut tx = self.db.begin().await?;
+
+        let conversation = tx
+            .conversations()
+            .by_id(conversation)
+            .await
+            .not_found(|| DeleteError::NotFound(conversation.clone()))?;
+        conversation
+            .as_snapshot()
+            .ok_or_else(|| DeleteError::Deleted(conversation.id().clone()))?;
+
+        let mut events = Vec::new();
+
+        let messages = tx.messages().live(&conversation).await?;
+        let has_messages = messages
+            .iter()
+            .map(message::History::as_snapshot)
+            .any(|message| message.is_some());
+        if has_messages {
+            return Err(DeleteError::NotEmpty(conversation.id().clone()));
+        }
+
+        let deleted = tx.sequence().next(deleted_at).await?;
+        let conversation = tx.conversations().delete(&conversation, &deleted).await?;
+        events.extend(
+            conversation
+                .events()
+                .filter(Sequence::start_from(deleted.sequence))
+                .map(Event::from),
+        );
+
+        tx.commit().await?;
+
+        self.events.broadcast(events);
+
+        Ok(())
+    }
+
+    pub async fn expire(&self, relative_to: &DateTime) -> Result<(), ExpireError> {
+        // Somewhat arbitrarily, expire after 7 days. Active conversation will not be
+        // expired until their messages expire.
+        let expire_at = relative_to.to_owned() - TimeDelta::days(7);
+
+        let mut tx = self.db.begin().await?;
+        let expired = tx.conversations().expired(&expire_at).await?;
+
+        let mut events = Vec::with_capacity(expired.len());
+        for conversation in expired {
+            let deleted = tx.sequence().next(relative_to).await?;
+            let conversation = tx.conversations().delete(&conversation, &deleted).await?;
+            events.push(
+                conversation
+                    .events()
+                    .filter(Sequence::start_from(deleted.sequence)),
+            );
+        }
+
+        tx.commit().await?;
+
+        self.events.broadcast(
+            events
+                .into_iter()
+                .kmerge_by(Sequence::merge)
+                .map(Event::from)
+                .collect::<Vec<_>>(),
+        );
+
+        Ok(())
+    }
+
+    pub async fn purge(&self, relative_to: &DateTime) -> Result<(), sqlx::Error> {
+        // Somewhat arbitrarily, purge after 6 hours.
+        let purge_at = relative_to.to_owned() - TimeDelta::hours(6);
+
+        let mut tx = self.db.begin().await?;
+        tx.conversations().purge(&purge_at).await?;
+        tx.commit().await?;
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum CreateError {
+    #[error("conversation named {0} already exists")]
+    DuplicateName(Name),
+    #[error("invalid conversation name: {0}")]
+    InvalidName(Name),
+    #[error(transparent)]
+    Database(#[from] sqlx::Error),
+    #[error(transparent)]
+    Name(#[from] name::Error),
+}
+
+impl From<LoadError> for CreateError {
+    fn from(error: LoadError) -> Self {
+        match error {
+            LoadError::Database(error) => error.into(),
+            LoadError::Name(error) => error.into(),
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("conversation {0} not found")]
+    NotFound(Id),
+    #[error("conversation {0} deleted")]
+    Deleted(Id),
+    #[error(transparent)]
+    Database(#[from] sqlx::Error),
+    #[error(transparent)]
+    Name(#[from] name::Error),
+}
+
+impl From<LoadError> for Error {
+    fn from(error: LoadError) -> Self {
+        match error {
+            LoadError::Database(error) => error.into(),
+            LoadError::Name(error) => error.into(),
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum DeleteError {
+    #[error("conversation {0} not found")]
+    NotFound(Id),
+    #[error("conversation {0} deleted")]
+    Deleted(Id),
+    #[error("conversation {0} not empty")]
+    NotEmpty(Id),
+    #[error(transparent)]
+    Database(#[from] sqlx::Error),
+    #[error(transparent)]
+    Name(#[from] name::Error),
+}
+
+impl From<LoadError> for DeleteError {
+    fn from(error: LoadError) -> Self {
+        match error {
+            LoadError::Database(error) => error.into(),
+            LoadError::Name(error) => error.into(),
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ExpireError {
+    #[error(transparent)]
+    Database(#[from] sqlx::Error),
+    #[error(transparent)]
+    Name(#[from] name::Error),
+}
+
+impl From<LoadError> for ExpireError {
+    fn from(error: LoadError) -> Self {
+        match error {
+            LoadError::Database(error) => error.into(),
+            LoadError::Name(error) => error.into(),
+        }
+    }
+}