Merge branch 'wip/event-vocabulary'

8 files changed, 245 insertions, 31 deletions

diff --git a/src/login/app.rs b/src/login/app.rs
index 4f60b89..bb1419b 100644
--- a/src/login/app.rs
+++ b/src/login/app.rs
@@ -1,24 +1,38 @@
 use sqlx::sqlite::SqlitePool;
 
 use super::{repo::Provider as _, Login, Password};
+use crate::{
+    clock::DateTime,
+    event::{repo::Provider as _, Broadcaster, Event},
+};
 
 pub struct Logins<'a> {
     db: &'a SqlitePool,
+    events: &'a Broadcaster,
 }
 
 impl<'a> Logins<'a> {
-    pub const fn new(db: &'a SqlitePool) -> Self {
-        Self { db }
+    pub const fn new(db: &'a SqlitePool, events: &'a Broadcaster) -> Self {
+        Self { db, events }
     }
 
-    pub async fn create(&self, name: &str, password: &Password) -> Result<Login, CreateError> {
+    pub async fn create(
+        &self,
+        name: &str,
+        password: &Password,
+        created_at: &DateTime,
+    ) -> Result<Login, CreateError> {
         let password_hash = password.hash()?;
 
         let mut tx = self.db.begin().await?;
-        let login = tx.logins().create(name, &password_hash).await?;
+        let created = tx.sequence().next(created_at).await?;
+        let login = tx.logins().create(name, &password_hash, &created).await?;
         tx.commit().await?;
 
-        Ok(login)
+        self.events
+            .broadcast(login.events().map(Event::from).collect::<Vec<_>>());
+
+        Ok(login.as_created())
     }
 }
 
diff --git a/src/login/event.rs b/src/login/event.rs
new file mode 100644
index 0000000..b03451a
--- /dev/null
+++ b/src/login/event.rs
@@ -0,0 +1,36 @@
+use super::snapshot::Login;
+use crate::event::{Instant, Sequenced};
+
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
+#[serde(tag = "event", rename_all = "snake_case")]
+pub enum Event {
+    Created(Created),
+}
+
+impl Sequenced for Event {
+    fn instant(&self) -> Instant {
+        match self {
+            Self::Created(created) => created.instant(),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
+pub struct Created {
+    #[serde(flatten)]
+    pub instant: Instant,
+    #[serde(flatten)]
+    pub login: Login,
+}
+
+impl Sequenced for Created {
+    fn instant(&self) -> Instant {
+        self.instant
+    }
+}
+
+impl From<Created> for Event {
+    fn from(event: Created) -> Self {
+        Self::Created(event)
+    }
+}
diff --git a/src/login/history.rs b/src/login/history.rs
new file mode 100644
index 0000000..add7d1e
--- /dev/null
+++ b/src/login/history.rs
@@ -0,0 +1,47 @@
+use super::{
+    event::{Created, Event},
+    Id, Login,
+};
+use crate::event::{Instant, ResumePoint, Sequence};
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct History {
+    pub login: Login,
+    pub created: Instant,
+}
+
+// State interface
+impl History {
+    pub fn id(&self) -> &Id {
+        &self.login.id
+    }
+
+    // Snapshot of this login as it was when created. (Note to the future: it's okay
+    // if this returns a redacted or modified version of the login. If we  implement
+    // renames by redacting the original name, then this should return the edited login, not the original, even if that's not how it was "as created.")
+    #[cfg(test)]
+    pub fn as_created(&self) -> Login {
+        self.login.clone()
+    }
+
+    pub fn as_of(&self, resume_point: impl Into<ResumePoint>) -> Option<Login> {
+        self.events()
+            .filter(Sequence::up_to(resume_point.into()))
+            .collect()
+    }
+}
+
+// Events interface
+impl History {
+    fn created(&self) -> Event {
+        Created {
+            instant: self.created,
+            login: self.login.clone(),
+        }
+        .into()
+    }
+
+    pub fn events(&self) -> impl Iterator<Item = Event> {
+        [self.created()].into_iter()
+    }
+}
diff --git a/src/login/mod.rs b/src/login/mod.rs
index f272f80..98cc3d7 100644
--- a/src/login/mod.rs
+++ b/src/login/mod.rs
@@ -1,22 +1,14 @@
 #[cfg(test)]
 pub mod app;
+pub mod event;
 pub mod extract;
+mod history;
 mod id;
 pub mod password;
 pub mod repo;
 mod routes;
+mod snapshot;
 
-pub use self::{id::Id, password::Password, routes::router};
-
-// This also implements FromRequestParts (see `./extract.rs`). As a result, it
-// can be used as an extractor for endpoints that want to require login, or for
-// endpoints that need to behave differently depending on whether the client is
-// or is not logged in.
-#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
-pub struct Login {
-    pub id: Id,
-    pub name: String,
-    // The omission of the hashed password is deliberate, to minimize the
-    // chance that it ends up tangled up in debug output or in some other chunk
-    // of logic elsewhere.
-}
+pub use self::{
+    event::Event, history::History, id::Id, password::Password, routes::router, snapshot::Login,
+};
diff --git a/src/login/repo.rs b/src/login/repo.rs
index d1a02c4..6d6510c 100644
--- a/src/login/repo.rs
+++ b/src/login/repo.rs
@@ -1,6 +1,10 @@
 use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction};
 
-use crate::login::{password::StoredHash, Id, Login};
+use crate::{
+    clock::DateTime,
+    event::{Instant, ResumePoint, Sequence},
+    login::{password::StoredHash, History, Id, Login},
+};
 
 pub trait Provider {
     fn logins(&mut self) -> Logins;
@@ -19,28 +23,100 @@ impl<'c> Logins<'c> {
         &mut self,
         name: &str,
         password_hash: &StoredHash,
-    ) -> Result<Login, sqlx::Error> {
+        created: &Instant,
+    ) -> Result<History, sqlx::Error> {
         let id = Id::generate();
 
-        let login = sqlx::query_as!(
-            Login,
+        let login = sqlx::query!(
             r#"
-                insert or fail
-                into login (id, name, password_hash)
-                values ($1, $2, $3)
+                insert
+                into login (id, name, password_hash, created_sequence, created_at)
+                values ($1, $2, $3, $4, $5)
                 returning
                     id as "id: Id",
-                    name
+                    name,
+                    created_sequence as "created_sequence: Sequence",
+                    created_at as "created_at: DateTime"
             "#,
             id,
             name,
             password_hash,
+            created.sequence,
+            created.at,
         )
+        .map(|row| History {
+            login: Login {
+                id: row.id,
+                name: row.name,
+            },
+            created: Instant {
+                at: row.created_at,
+                sequence: row.created_sequence,
+            },
+        })
         .fetch_one(&mut *self.0)
         .await?;
 
         Ok(login)
     }
+
+    pub async fn all(&mut self, resume_at: ResumePoint) -> Result<Vec<History>, sqlx::Error> {
+        let channels = sqlx::query!(
+            r#"
+                select
+                    id as "id: Id",
+                    name,
+                    created_sequence as "created_sequence: Sequence",
+                    created_at as "created_at: DateTime"
+                from login
+                    where coalesce(created_sequence <= $1, true)
+                    order by created_sequence
+            "#,
+            resume_at,
+        )
+        .map(|row| History {
+            login: Login {
+                id: row.id,
+                name: row.name,
+            },
+            created: Instant {
+                at: row.created_at,
+                sequence: row.created_sequence,
+            },
+        })
+        .fetch_all(&mut *self.0)
+        .await?;
+
+        Ok(channels)
+    }
+    pub async fn replay(&mut self, resume_at: ResumePoint) -> Result<Vec<History>, sqlx::Error> {
+        let messages = sqlx::query!(
+            r#"
+                select
+                    id as "id: Id",
+                    name,
+                    created_sequence as "created_sequence: Sequence",
+                    created_at as "created_at: DateTime"
+                from login
+                where coalesce(login.created_sequence > $1, true)
+            "#,
+            resume_at,
+        )
+        .map(|row| History {
+            login: Login {
+                id: row.id,
+                name: row.name,
+            },
+            created: Instant {
+                at: row.created_at,
+                sequence: row.created_sequence,
+            },
+        })
+        .fetch_all(&mut *self.0)
+        .await?;
+
+        Ok(messages)
+    }
 }
 
 impl<'t> From<&'t mut SqliteConnection> for Logins<'t> {
diff --git a/src/login/routes/test/login.rs b/src/login/routes/test/login.rs
index 3c82738..6a3b79c 100644
--- a/src/login/routes/test/login.rs
+++ b/src/login/routes/test/login.rs
@@ -47,7 +47,7 @@ async fn existing_identity() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let (name, password) = fixtures::login::create_with_password(&app).await;
+    let (name, password) = fixtures::login::create_with_password(&app, &fixtures::now()).await;
 
     // Call the endpoint
 
@@ -84,7 +84,7 @@ async fn authentication_failed() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let login = fixtures::login::create(&app).await;
+    let login = fixtures::login::create(&app, &fixtures::now()).await;
 
     // Call the endpoint
 
@@ -109,7 +109,7 @@ async fn token_expires() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let (name, password) = fixtures::login::create_with_password(&app).await;
+    let (name, password) = fixtures::login::create_with_password(&app, &fixtures::now()).await;
 
     // Call the endpoint
 
diff --git a/src/login/routes/test/logout.rs b/src/login/routes/test/logout.rs
index 42b2534..611829e 100644
--- a/src/login/routes/test/logout.rs
+++ b/src/login/routes/test/logout.rs
@@ -11,7 +11,7 @@ async fn successful() {
 
     let app = fixtures::scratch_app().await;
     let now = fixtures::now();
-    let login = fixtures::login::create_with_password(&app).await;
+    let login = fixtures::login::create_with_password(&app, &fixtures::now()).await;
     let identity = fixtures::identity::logged_in(&app, &login, &now).await;
     let secret = fixtures::identity::secret(&identity);
 
diff --git a/src/login/snapshot.rs b/src/login/snapshot.rs
new file mode 100644
index 0000000..1379005
--- /dev/null
+++ b/src/login/snapshot.rs
@@ -0,0 +1,49 @@
+use super::{
+    event::{Created, Event},
+    Id,
+};
+
+// This also implements FromRequestParts (see `./extract.rs`). As a result, it
+// can be used as an extractor for endpoints that want to require login, or for
+// endpoints that need to behave differently depending on whether the client is
+// or is not logged in.
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
+pub struct Login {
+    pub id: Id,
+    pub name: String,
+    // The omission of the hashed password is deliberate, to minimize the
+    // chance that it ends up tangled up in debug output or in some other chunk
+    // of logic elsewhere.
+}
+
+impl Login {
+    // Two reasons for this allow:
+    //
+    // 1. This is used to collect streams using a fold, below, which requires a type consistent with the fold, and
+    // 2. It's also consistent with the other history state machine types.
+    #[allow(clippy::unnecessary_wraps)]
+    fn apply(state: Option<Self>, event: Event) -> Option<Self> {
+        match (state, event) {
+            (None, Event::Created(event)) => Some(event.into()),
+            (state, event) => panic!("invalid message event {event:#?} for state {state:#?}"),
+        }
+    }
+}
+
+impl FromIterator<Event> for Option<Login> {
+    fn from_iter<I: IntoIterator<Item = Event>>(events: I) -> Self {
+        events.into_iter().fold(None, Login::apply)
+    }
+}
+
+impl From<&Created> for Login {
+    fn from(event: &Created) -> Self {
+        event.login.clone()
+    }
+}
+
+impl From<Created> for Login {
+    fn from(event: Created) -> Self {
+        event.login
+    }
+}