Merge branch 'unicode-normalization'

1 files changed, 166 insertions, 78 deletions

diff --git a/src/channel/repo.rs b/src/channel/repo.rs
index 27d35f0..e26ac2b 100644
--- a/src/channel/repo.rs
+++ b/src/channel/repo.rs
@@ -1,9 +1,12 @@
+use futures::stream::{StreamExt as _, TryStreamExt as _};
 use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction};
 
 use crate::{
     channel::{Channel, History, Id},
     clock::DateTime,
+    db::NotFound,
     event::{Instant, ResumePoint, Sequence},
+    name::{self, Name},
 };
 
 pub trait Provider {
@@ -19,134 +22,162 @@ impl<'c> Provider for Transaction<'c, Sqlite> {
 pub struct Channels<'t>(&'t mut SqliteConnection);
 
 impl<'c> Channels<'c> {
-    pub async fn create(&mut self, name: &str, created: &Instant) -> Result<History, sqlx::Error> {
+    pub async fn create(&mut self, name: &Name, created: &Instant) -> Result<History, sqlx::Error> {
         let id = Id::generate();
-        let channel = sqlx::query!(
+        let name = name.clone();
+        let display_name = name.display();
+        let canonical_name = name.canonical();
+        let created = *created;
+
+        sqlx::query!(
             r#"
                 insert
-                into channel (id, name, created_at, created_sequence)
-                values ($1, $2, $3, $4)
-                returning
-                    id as "id: Id",
-                    name as "name!", -- known non-null as we just set it
-                    created_at as "created_at: DateTime",
-                    created_sequence as "created_sequence: Sequence"
+                into channel (id, created_at, created_sequence)
+                values ($1, $2, $3)
             "#,
             id,
-            name,
             created.at,
             created.sequence,
         )
-        .map(|row| History {
+        .execute(&mut *self.0)
+        .await?;
+
+        sqlx::query!(
+            r#"
+                insert into channel_name (id, display_name, canonical_name)
+                values ($1, $2, $3)
+            "#,
+            id,
+            display_name,
+            canonical_name,
+        )
+        .execute(&mut *self.0)
+        .await?;
+
+        let channel = History {
             channel: Channel {
-                id: row.id,
-                name: row.name,
+                id,
+                name: name.clone(),
                 deleted_at: None,
             },
-            created: Instant::new(row.created_at, row.created_sequence),
+            created,
             deleted: None,
-        })
-        .fetch_one(&mut *self.0)
-        .await?;
+        };
 
         Ok(channel)
     }
 
-    pub async fn by_id(&mut self, channel: &Id) -> Result<History, sqlx::Error> {
+    pub async fn by_id(&mut self, channel: &Id) -> Result<History, LoadError> {
         let channel = sqlx::query!(
             r#"
                 select
                     id as "id: Id",
-                    channel.name,
+                    name.display_name as "display_name?: String",
+                    name.canonical_name as "canonical_name?: String",
                     channel.created_at as "created_at: DateTime",
                     channel.created_sequence as "created_sequence: Sequence",
                     deleted.deleted_at as "deleted_at?: DateTime",
                     deleted.deleted_sequence as "deleted_sequence?: Sequence"
                 from channel
+                left join channel_name as name
+                    using (id)
                 left join channel_deleted as deleted
                     using (id)
                 where id = $1
             "#,
             channel,
         )
-        .map(|row| History {
-            channel: Channel {
-                id: row.id,
-                name: row.name.unwrap_or_default(),
-                deleted_at: row.deleted_at,
-            },
-            created: Instant::new(row.created_at, row.created_sequence),
-            deleted: Instant::optional(row.deleted_at, row.deleted_sequence),
+        .map(|row| {
+            Ok::<_, name::Error>(History {
+                channel: Channel {
+                    id: row.id,
+                    name: Name::optional(row.display_name, row.canonical_name)?.unwrap_or_default(),
+                    deleted_at: row.deleted_at,
+                },
+                created: Instant::new(row.created_at, row.created_sequence),
+                deleted: Instant::optional(row.deleted_at, row.deleted_sequence),
+            })
         })
         .fetch_one(&mut *self.0)
-        .await?;
+        .await??;
 
         Ok(channel)
     }
 
-    pub async fn all(&mut self, resume_at: ResumePoint) -> Result<Vec<History>, sqlx::Error> {
+    pub async fn all(&mut self, resume_at: ResumePoint) -> Result<Vec<History>, LoadError> {
         let channels = sqlx::query!(
             r#"
                 select
                     id as "id: Id",
-                    channel.name,
+                    name.display_name as "display_name: String",
+                    name.canonical_name as "canonical_name: String",
                     channel.created_at as "created_at: DateTime",
                     channel.created_sequence as "created_sequence: Sequence",
-                    deleted.deleted_at as "deleted_at: DateTime",
-                    deleted.deleted_sequence as "deleted_sequence: Sequence"
+                    deleted.deleted_at as "deleted_at?: DateTime",
+                    deleted.deleted_sequence as "deleted_sequence?: Sequence"
                 from channel
+                left join channel_name as name
+                    using (id)
                 left join channel_deleted as deleted
                     using (id)
                 where coalesce(channel.created_sequence <= $1, true)
-                order by channel.name
+                order by name.canonical_name
             "#,
             resume_at,
         )
-        .map(|row| History {
-            channel: Channel {
-                id: row.id,
-                name: row.name.unwrap_or_default(),
-                deleted_at: row.deleted_at,
-            },
-            created: Instant::new(row.created_at, row.created_sequence),
-            deleted: Instant::optional(row.deleted_at, row.deleted_sequence),
+        .map(|row| {
+            Ok::<_, name::Error>(History {
+                channel: Channel {
+                    id: row.id,
+                    name: Name::optional(row.display_name, row.canonical_name)?.unwrap_or_default(),
+                    deleted_at: row.deleted_at,
+                },
+                created: Instant::new(row.created_at, row.created_sequence),
+                deleted: Instant::optional(row.deleted_at, row.deleted_sequence),
+            })
         })
-        .fetch_all(&mut *self.0)
+        .fetch(&mut *self.0)
+        .map(|res| Ok::<_, LoadError>(res??))
+        .try_collect()
         .await?;
 
         Ok(channels)
     }
 
-    pub async fn replay(
-        &mut self,
-        resume_at: Option<Sequence>,
-    ) -> Result<Vec<History>, sqlx::Error> {
+    pub async fn replay(&mut self, resume_at: Option<Sequence>) -> Result<Vec<History>, LoadError> {
         let channels = sqlx::query!(
             r#"
                 select
                     id as "id: Id",
-                    channel.name,
+                    name.display_name as "display_name: String",
+                    name.canonical_name as "canonical_name: String",
                     channel.created_at as "created_at: DateTime",
                     channel.created_sequence as "created_sequence: Sequence",
-                    deleted.deleted_at as "deleted_at: DateTime",
-                    deleted.deleted_sequence as "deleted_sequence: Sequence"
+                    deleted.deleted_at as "deleted_at?: DateTime",
+                    deleted.deleted_sequence as "deleted_sequence?: Sequence"
                 from channel
+                left join channel_name as name
+                    using (id)
                 left join channel_deleted as deleted
                     using (id)
                 where coalesce(channel.created_sequence > $1, true)
             "#,
             resume_at,
         )
-        .map(|row| History {
-            channel: Channel {
-                id: row.id,
-                name: row.name.unwrap_or_default(),
-                deleted_at: row.deleted_at,
-            },
-            created: Instant::new(row.created_at, row.created_sequence),
-            deleted: Instant::optional(row.deleted_at, row.deleted_sequence),
+        .map(|row| {
+            Ok::<_, name::Error>(History {
+                channel: Channel {
+                    id: row.id,
+                    name: Name::optional(row.display_name, row.canonical_name)?.unwrap_or_default(),
+                    deleted_at: row.deleted_at,
+                },
+                created: Instant::new(row.created_at, row.created_sequence),
+                deleted: Instant::optional(row.deleted_at, row.deleted_sequence),
+            })
         })
-        .fetch_all(&mut *self.0)
+        .fetch(&mut *self.0)
+        .map(|res| Ok::<_, LoadError>(res??))
+        .try_collect()
         .await?;
 
         Ok(channels)
@@ -156,19 +187,18 @@ impl<'c> Channels<'c> {
         &mut self,
         channel: &History,
         deleted: &Instant,
-    ) -> Result<History, sqlx::Error> {
+    ) -> Result<History, LoadError> {
         let id = channel.id();
-        sqlx::query_scalar!(
+        sqlx::query!(
             r#"
                 insert into channel_deleted (id, deleted_at, deleted_sequence)
                 values ($1, $2, $3)
-                returning 1 as "deleted: bool"
             "#,
             id,
             deleted.at,
             deleted.sequence,
         )
-        .fetch_one(&mut *self.0)
+        .execute(&mut *self.0)
         .await?;
 
         // Small social responsibility hack here: when a channel is deleted, its name is
@@ -179,16 +209,14 @@ impl<'c> Channels<'c> {
         // This also avoids the need for a separate name reservation table to ensure
         // that live channels have unique names, since the `channel` table's name field
         // is unique over non-null values.
-        sqlx::query_scalar!(
+        sqlx::query!(
             r#"
-                update channel
-                set name = null
+                delete from channel_name
                 where id = $1
-                returning 1 as "updated: bool"
             "#,
             id,
         )
-        .fetch_one(&mut *self.0)
+        .execute(&mut *self.0)
         .await?;
 
         let channel = self.by_id(id).await?;
@@ -230,38 +258,98 @@ impl<'c> Channels<'c> {
         Ok(())
     }
 
-    pub async fn expired(&mut self, expired_at: &DateTime) -> Result<Vec<History>, sqlx::Error> {
+    pub async fn expired(&mut self, expired_at: &DateTime) -> Result<Vec<History>, LoadError> {
         let channels = sqlx::query!(
             r#"
                 select
                     channel.id as "id: Id",
-                    channel.name,
+                    name.display_name as "display_name: String",
+                    name.canonical_name as "canonical_name: String",
                     channel.created_at as "created_at: DateTime",
                     channel.created_sequence as "created_sequence: Sequence",
                     deleted.deleted_at as "deleted_at?: DateTime",
                     deleted.deleted_sequence as "deleted_sequence?: Sequence"
                 from channel
+                left join channel_name as name
+                    using (id)
                 left join channel_deleted as deleted
                     using (id)
                 left join message
+                    on channel.id = message.channel
                 where channel.created_at < $1
                     and message.id is null
                     and deleted.id is null
             "#,
             expired_at,
         )
-        .map(|row| History {
-            channel: Channel {
-                id: row.id,
-                name: row.name.unwrap_or_default(),
-                deleted_at: row.deleted_at,
-            },
-            created: Instant::new(row.created_at, row.created_sequence),
-            deleted: Instant::optional(row.deleted_at, row.deleted_sequence),
+        .map(|row| {
+            Ok::<_, name::Error>(History {
+                channel: Channel {
+                    id: row.id,
+                    name: Name::optional(row.display_name, row.canonical_name)?.unwrap_or_default(),
+                    deleted_at: row.deleted_at,
+                },
+                created: Instant::new(row.created_at, row.created_sequence),
+                deleted: Instant::optional(row.deleted_at, row.deleted_sequence),
+            })
         })
-        .fetch_all(&mut *self.0)
+        .fetch(&mut *self.0)
+        .map(|res| Ok::<_, LoadError>(res??))
+        .try_collect()
         .await?;
 
         Ok(channels)
     }
+
+    pub async fn recanonicalize(&mut self) -> Result<(), sqlx::Error> {
+        let channels = sqlx::query!(
+            r#"
+                select
+                    id as "id: Id",
+                    display_name as "display_name: String"
+                from channel_name
+            "#,
+        )
+        .fetch_all(&mut *self.0)
+        .await?;
+
+        for channel in channels {
+            let name = Name::from(channel.display_name);
+            let canonical_name = name.canonical();
+
+            sqlx::query!(
+                r#"
+                    update channel_name
+                    set canonical_name = $1
+                    where id = $2
+                "#,
+                canonical_name,
+                channel.id,
+            )
+            .execute(&mut *self.0)
+            .await?;
+        }
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error(transparent)]
+pub enum LoadError {
+    Database(#[from] sqlx::Error),
+    Name(#[from] name::Error),
+}
+
+impl<T> NotFound for Result<T, LoadError> {
+    type Ok = T;
+    type Error = LoadError;
+
+    fn optional(self) -> Result<Option<T>, LoadError> {
+        match self {
+            Ok(value) => Ok(Some(value)),
+            Err(LoadError::Database(sqlx::Error::RowNotFound)) => Ok(None),
+            Err(other) => Err(other),
+        }
+    }
 }