Rename "channels" to "conversations."

The term "channel" for a conversational container has a long and storied history, but is mostly evocative of IRC and of other, ah, "nerd-centric" services. It does show up in more widespread contexts: Discord and Slack both refer to their primary conversational containers as "channels," for example. However, I think it's unnecessary jargon, and I'd like to do away with it.

To that end, this change pervasively changes one term to the other wherever it appears, with the following exceptions:

* A `channel` concept (unrelated to conversations) is also provided by an external library; we can't and shouldn't try to rename that.
* The code to deal with the `pilcrow:channelData` and `pilcrow:lastActiveChannel` local storage properties is still present, to migrate existing data to new keys. It will be removed in a later change.

This is a **breaking API change**. As we are not yet managing any API compatibility promises, this is formally not an issue, but it is something to be aware of practically. The major API changes are:

* Paths beginning with `/api/channels` are now under `/api/conversations`, without other modifications.
* Fields labelled with `channel…` terms are now labelled with `conversation…` terms. For example, a `message` `sent` event is now sent to a `conversation`, not a `channel`.

This is also a **breaking UI change**. Specifically, any saved paths for `/ch/CHANNELID` will now lead to a 404. The corresponding paths are `/c/CONVERSATIONID`. While I've made an effort to migrate the location of stored data, I have not tried to provide adapters to fix this specific issue, because the disruption is short-lived and very easily addressed by opening a channel in the client UI.

This change is obnoxiously large and difficult to review, for which I apologize. If this shows up in `git annotate`, please forgive me. These kinds of renamings are hard to carry out without a major disruption, especially when the concept ("channel" in this case) is used so pervasively throughout the system.

I think it's worth making this change that pervasively so that we don't have an indefinitely-long tail of "well, it's a conversation in the docs, but the table is called `channel` for historical reasons" type issues.

Merges conversations-not-channels into main.

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(),
+        }
+    }
+}