1 files changed, 153 insertions, 0 deletions
diff --git a/src/user/repo.rs b/src/user/repo.rs
new file mode 100644
index 0000000..c02d50f
--- /dev/null
+++ b/src/user/repo.rs
@@ -0,0 +1,153 @@
+use futures::stream::{StreamExt as _, TryStreamExt as _};
+use sqlx::{SqliteConnection, Transaction, sqlite::Sqlite};
+
+use crate::{
+    clock::DateTime,
+    event::{Instant, Sequence},
+    name::{self, Name},
+    user::{History, Id, User, password::StoredHash},
+};
+
+pub trait Provider {
+    fn users(&mut self) -> Users;
+}
+
+impl Provider for Transaction<'_, Sqlite> {
+    fn users(&mut self) -> Users {
+        Users(self)
+    }
+}
+
+pub struct Users<'t>(&'t mut SqliteConnection);
+
+impl Users<'_> {
+    pub async fn create(
+        &mut self,
+        name: &Name,
+        password_hash: &StoredHash,
+        created: &Instant,
+    ) -> Result<History, sqlx::Error> {
+        let id = Id::generate();
+        let display_name = name.display();
+        let canonical_name = name.canonical();
+
+        sqlx::query!(
+            r#"
+                insert
+                into user (id, display_name, canonical_name, password_hash, created_sequence, created_at)
+                values ($1, $2, $3, $4, $5, $6)
+            "#,
+            id,
+            display_name,
+            canonical_name,
+            password_hash,
+            created.sequence,
+            created.at,
+        )
+        .execute(&mut *self.0)
+        .await?;
+
+        let user = History {
+            created: *created,
+            user: User {
+                id,
+                name: name.clone(),
+            },
+        };
+
+        Ok(user)
+    }
+
+    pub async fn set_password(
+        &mut self,
+        login: &History,
+        to: &StoredHash,
+    ) -> Result<(), sqlx::Error> {
+        let login = login.id();
+
+        sqlx::query_scalar!(
+            r#"
+                update user
+                set password_hash = $1
+                where id = $2
+                returning id as "id: Id"
+            "#,
+            to,
+            login,
+        )
+        .fetch_one(&mut *self.0)
+        .await?;
+
+        Ok(())
+    }
+
+    pub async fn all(&mut self, resume_at: Sequence) -> Result<Vec<History>, LoadError> {
+        let logins = sqlx::query!(
+            r#"
+                select
+                    id as "id: Id",
+                    display_name as "display_name: String",
+                    canonical_name as "canonical_name: String",
+                    created_sequence as "created_sequence: Sequence",
+                    created_at as "created_at: DateTime"
+                from user
+                    where created_sequence <= $1
+                    order by canonical_name
+            "#,
+            resume_at,
+        )
+        .map(|row| {
+            Ok::<_, LoadError>(History {
+                user: User {
+                    id: row.id,
+                    name: Name::new(row.display_name, row.canonical_name)?,
+                },
+                created: Instant::new(row.created_at, row.created_sequence),
+            })
+        })
+        .fetch(&mut *self.0)
+        .map(|res| res?)
+        .try_collect()
+        .await?;
+
+        Ok(logins)
+    }
+
+    pub async fn replay(&mut self, resume_at: Sequence) -> Result<Vec<History>, LoadError> {
+        let logins = sqlx::query!(
+            r#"
+                select
+                    id as "id: Id",
+                    display_name as "display_name: String",
+                    canonical_name as "canonical_name: String",
+                    created_sequence as "created_sequence: Sequence",
+                    created_at as "created_at: DateTime"
+                from user
+                where created_sequence > $1
+            "#,
+            resume_at,
+        )
+        .map(|row| {
+            Ok::<_, name::Error>(History {
+                user: User {
+                    id: row.id,
+                    name: Name::new(row.display_name, row.canonical_name)?,
+                },
+                created: Instant::new(row.created_at, row.created_sequence),
+            })
+        })
+        .fetch(&mut *self.0)
+        .map(|res| Ok::<_, LoadError>(res??))
+        .try_collect()
+        .await?;
+
+        Ok(logins)
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error(transparent)]
+pub enum LoadError {
+    Database(#[from] sqlx::Error),
+    Name(#[from] name::Error),
+}