From ba96974bdebd6d4ec345907d49944b5ee644ed47 Mon Sep 17 00:00:00 2001 From: Owen Jacobson Date: Wed, 9 Oct 2024 00:57:31 -0400 Subject: Provide a view of logins to clients. --- src/app.rs | 22 +++++---- src/boot/app.rs | 12 ++++- src/boot/mod.rs | 5 +- src/channel/app.rs | 2 +- src/channel/routes/test/on_create.rs | 4 +- src/channel/routes/test/on_send.rs | 6 +-- src/event/app.rs | 12 ++++- src/event/mod.rs | 13 ++++- src/event/routes/test.rs | 32 ++++++------- src/login/app.rs | 24 ++++++++-- src/login/event.rs | 36 ++++++++++++++ src/login/history.rs | 47 ++++++++++++++++++ src/login/mod.rs | 20 +++----- src/login/repo.rs | 92 ++++++++++++++++++++++++++++++++---- src/login/routes/test/login.rs | 6 +-- src/login/routes/test/logout.rs | 2 +- src/login/snapshot.rs | 49 +++++++++++++++++++ src/message/app.rs | 2 +- src/message/repo.rs | 66 ++++++++++++-------------- src/message/snapshot.rs | 4 +- src/test/fixtures/login.rs | 9 ++-- src/token/app.rs | 49 +++++++++++++------ src/token/broadcaster.rs | 3 +- src/token/event.rs | 10 +--- src/token/mod.rs | 4 +- src/token/repo/auth.rs | 28 +++++++---- src/token/repo/token.rs | 9 ++-- 27 files changed, 417 insertions(+), 151 deletions(-) create mode 100644 src/login/event.rs create mode 100644 src/login/history.rs create mode 100644 src/login/snapshot.rs (limited to 'src') diff --git a/src/app.rs b/src/app.rs index 6d007a9..177c134 100644 --- a/src/app.rs +++ b/src/app.rs @@ -3,9 +3,9 @@ use sqlx::sqlite::SqlitePool; use crate::{ boot::app::Boot, channel::app::Channels, - event::{app::Events, broadcaster::Broadcaster as EventBroadcaster}, + event::{self, app::Events}, message::app::Messages, - token::{app::Tokens, broadcaster::Broadcaster as TokenBroadcaster}, + token::{self, app::Tokens}, }; #[cfg(test)] @@ -14,15 +14,19 @@ use crate::login::app::Logins; #[derive(Clone)] pub struct App { db: SqlitePool, - events: EventBroadcaster, - tokens: TokenBroadcaster, + events: event::Broadcaster, + token_events: token::Broadcaster, } impl App { pub fn from(db: SqlitePool) -> Self { - let events = EventBroadcaster::default(); - let tokens = TokenBroadcaster::default(); - Self { db, events, tokens } + let events = event::Broadcaster::default(); + let token_events = token::Broadcaster::default(); + Self { + db, + events, + token_events, + } } } @@ -41,7 +45,7 @@ impl App { #[cfg(test)] pub const fn logins(&self) -> Logins { - Logins::new(&self.db) + Logins::new(&self.db, &self.events) } pub const fn messages(&self) -> Messages { @@ -49,6 +53,6 @@ impl App { } pub const fn tokens(&self) -> Tokens { - Tokens::new(&self.db, &self.tokens) + Tokens::new(&self.db, &self.events, &self.token_events) } } diff --git a/src/boot/app.rs b/src/boot/app.rs index fc84b3a..03e7230 100644 --- a/src/boot/app.rs +++ b/src/boot/app.rs @@ -2,7 +2,8 @@ use sqlx::sqlite::SqlitePool; use super::{Channel, Snapshot}; use crate::{ - channel::repo::Provider as _, event::repo::Provider as _, message::repo::Provider as _, + channel::repo::Provider as _, event::repo::Provider as _, login::repo::Provider as _, + message::repo::Provider as _, }; pub struct Boot<'a> { @@ -17,8 +18,14 @@ impl<'a> Boot<'a> { pub async fn snapshot(&self) -> Result { let mut tx = self.db.begin().await?; let resume_point = tx.sequence().current().await?; - let channels = tx.channels().all(resume_point.into()).await?; + let logins = tx.logins().all(resume_point.into()).await?; + let logins = logins + .into_iter() + .filter_map(|login| login.as_of(resume_point)) + .collect(); + + let channels = tx.channels().all(resume_point.into()).await?; let channels = { let mut snapshots = Vec::with_capacity(channels.len()); @@ -48,6 +55,7 @@ impl<'a> Boot<'a> { Ok(Snapshot { resume_point, + logins, channels, }) } diff --git a/src/boot/mod.rs b/src/boot/mod.rs index a68acac..1f94106 100644 --- a/src/boot/mod.rs +++ b/src/boot/mod.rs @@ -4,7 +4,7 @@ mod routes; use crate::{ channel, event::{Instant, Sequence}, - login::Login, + login::{self, Login}, message, }; @@ -13,6 +13,7 @@ pub use self::routes::router; #[derive(serde::Serialize)] pub struct Snapshot { pub resume_point: Sequence, + pub logins: Vec, pub channels: Vec, } @@ -43,7 +44,7 @@ impl Channel { pub struct Message { #[serde(flatten)] pub sent: Instant, - pub sender: Login, + pub sender: login::Id, pub id: message::Id, pub body: String, } diff --git a/src/channel/app.rs b/src/channel/app.rs index a9a9e84..cb7ad32 100644 --- a/src/channel/app.rs +++ b/src/channel/app.rs @@ -6,7 +6,7 @@ use super::{repo::Provider as _, Channel, Id}; use crate::{ clock::DateTime, db::NotFound, - event::{broadcaster::Broadcaster, repo::Provider as _, Event, Sequence}, + event::{repo::Provider as _, Broadcaster, Event, Sequence}, message::repo::Provider as _, }; diff --git a/src/channel/routes/test/on_create.rs b/src/channel/routes/test/on_create.rs index 8c3c62b..eeecc7f 100644 --- a/src/channel/routes/test/on_create.rs +++ b/src/channel/routes/test/on_create.rs @@ -12,7 +12,7 @@ async fn new_channel() { // Set up the environment let app = fixtures::scratch_app().await; - let creator = fixtures::login::create(&app).await; + let creator = fixtures::login::create(&app, &fixtures::now()).await; // Call the endpoint @@ -64,7 +64,7 @@ async fn duplicate_name() { // Set up the environment let app = fixtures::scratch_app().await; - let creator = fixtures::login::create(&app).await; + let creator = fixtures::login::create(&app, &fixtures::now()).await; let channel = fixtures::channel::create(&app, &fixtures::now()).await; // Call the endpoint diff --git a/src/channel/routes/test/on_send.rs b/src/channel/routes/test/on_send.rs index d2acc48..3fe3d1e 100644 --- a/src/channel/routes/test/on_send.rs +++ b/src/channel/routes/test/on_send.rs @@ -14,7 +14,7 @@ async fn messages_in_order() { // Set up the environment let app = fixtures::scratch_app().await; - let sender = fixtures::login::create(&app).await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; let channel = fixtures::channel::create(&app, &fixtures::now()).await; // Call the endpoint (twice) @@ -57,7 +57,7 @@ async fn messages_in_order() { assert!(matches!( event, event::Event::Message(message::Event::Sent(event)) - if event.message.sender == sender + if event.message.sender == sender.id && event.message.body == message )); } @@ -68,7 +68,7 @@ async fn nonexistent_channel() { // 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 diff --git a/src/event/app.rs b/src/event/app.rs index 141037d..951ce25 100644 --- a/src/event/app.rs +++ b/src/event/app.rs @@ -9,6 +9,7 @@ use sqlx::sqlite::SqlitePool; use super::{broadcaster::Broadcaster, Event, ResumePoint, Sequence, Sequenced}; use crate::{ channel::{self, repo::Provider as _}, + login::{self, repo::Provider as _}, message::{self, repo::Provider as _}, }; @@ -33,6 +34,14 @@ impl<'a> Events<'a> { let mut tx = self.db.begin().await?; + let logins = tx.logins().replay(resume_at).await?; + let login_events = logins + .iter() + .map(login::History::events) + .kmerge_by(Sequence::merge) + .filter(Sequence::after(resume_at)) + .map(Event::from); + let channels = tx.channels().replay(resume_at).await?; let channel_events = channels .iter() @@ -49,7 +58,8 @@ impl<'a> Events<'a> { .filter(Sequence::after(resume_at)) .map(Event::from); - let replay_events = channel_events + let replay_events = login_events + .merge_by(channel_events, Sequence::merge) .merge_by(message_events, Sequence::merge) .collect::>(); let resume_live_at = replay_events.last().map(Sequenced::sequence); diff --git a/src/event/mod.rs b/src/event/mod.rs index 698e55a..69c7a10 100644 --- a/src/event/mod.rs +++ b/src/event/mod.rs @@ -1,13 +1,14 @@ -use crate::{channel, message}; +use crate::{channel, login, message}; pub mod app; -pub mod broadcaster; +mod broadcaster; mod extract; pub mod repo; mod routes; mod sequence; pub use self::{ + broadcaster::Broadcaster, routes::router, sequence::{Instant, Sequence, Sequenced}, }; @@ -17,6 +18,7 @@ pub type ResumePoint = Option; #[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum Event { + Login(login::Event), Channel(channel::Event), Message(message::Event), } @@ -24,12 +26,19 @@ pub enum Event { impl Sequenced for Event { fn instant(&self) -> Instant { match self { + Self::Login(event) => event.instant(), Self::Channel(event) => event.instant(), Self::Message(event) => event.instant(), } } } +impl From for Event { + fn from(event: login::Event) -> Self { + Self::Login(event) + } +} + impl From for Event { fn from(event: channel::Event) -> Self { Self::Channel(event) diff --git a/src/event/routes/test.rs b/src/event/routes/test.rs index ba9953e..209a016 100644 --- a/src/event/routes/test.rs +++ b/src/event/routes/test.rs @@ -15,13 +15,13 @@ async fn includes_historical_message() { // Set up the environment let app = fixtures::scratch_app().await; - let sender = fixtures::login::create(&app).await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; let channel = fixtures::channel::create(&app, &fixtures::now()).await; let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await; // Call the endpoint - let subscriber_creds = fixtures::login::create_with_password(&app).await; + let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await; let routes::Events(events) = routes::events(State(app), subscriber, None, Query::default()) .await @@ -48,7 +48,7 @@ async fn includes_live_message() { // Call the endpoint - let subscriber_creds = fixtures::login::create_with_password(&app).await; + let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await; let routes::Events(events) = routes::events(State(app.clone()), subscriber, None, Query::default()) @@ -57,7 +57,7 @@ async fn includes_live_message() { // Verify the semantics - let sender = fixtures::login::create(&app).await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await; let event = events @@ -75,7 +75,7 @@ async fn includes_multiple_channels() { // Set up the environment let app = fixtures::scratch_app().await; - let sender = fixtures::login::create(&app).await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; let channels = [ fixtures::channel::create(&app, &fixtures::now()).await, @@ -94,7 +94,7 @@ async fn includes_multiple_channels() { // Call the endpoint - let subscriber_creds = fixtures::login::create_with_password(&app).await; + let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await; let routes::Events(events) = routes::events(State(app), subscriber, None, Query::default()) .await @@ -122,7 +122,7 @@ async fn sequential_messages() { let app = fixtures::scratch_app().await; let channel = fixtures::channel::create(&app, &fixtures::now()).await; - let sender = fixtures::login::create(&app).await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; let messages = vec![ fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, @@ -132,7 +132,7 @@ async fn sequential_messages() { // Call the endpoint - let subscriber_creds = fixtures::login::create_with_password(&app).await; + let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await; let routes::Events(events) = routes::events(State(app), subscriber, None, Query::default()) .await @@ -166,7 +166,7 @@ async fn resumes_from() { let app = fixtures::scratch_app().await; let channel = fixtures::channel::create(&app, &fixtures::now()).await; - let sender = fixtures::login::create(&app).await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; let initial_message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await; @@ -177,7 +177,7 @@ async fn resumes_from() { // Call the endpoint - let subscriber_creds = fixtures::login::create_with_password(&app).await; + let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await; let resume_at = { @@ -248,13 +248,13 @@ async fn serial_resume() { // Set up the environment let app = fixtures::scratch_app().await; - let sender = fixtures::login::create(&app).await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; let channel_a = fixtures::channel::create(&app, &fixtures::now()).await; let channel_b = fixtures::channel::create(&app, &fixtures::now()).await; // Call the endpoint - let subscriber_creds = fixtures::login::create_with_password(&app).await; + let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await; let resume_at = { @@ -372,11 +372,11 @@ async fn terminates_on_token_expiry() { let app = fixtures::scratch_app().await; let channel = fixtures::channel::create(&app, &fixtures::now()).await; - let sender = fixtures::login::create(&app).await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; // Subscribe via the endpoint - let subscriber_creds = fixtures::login::create_with_password(&app).await; + let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::ancient()).await; @@ -417,11 +417,11 @@ async fn terminates_on_logout() { let app = fixtures::scratch_app().await; let channel = fixtures::channel::create(&app, &fixtures::now()).await; - let sender = fixtures::login::create(&app).await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; // Subscribe via the endpoint - let subscriber_creds = fixtures::login::create_with_password(&app).await; + let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; let subscriber_token = fixtures::identity::logged_in(&app, &subscriber_creds, &fixtures::now()).await; let subscriber = 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 { + pub async fn create( + &self, + name: &str, + password: &Password, + created_at: &DateTime, + ) -> Result { 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::>()); + + 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 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) -> Option { + 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 { + [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 { + created: &Instant, + ) -> Result { 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, 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, 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, event: Event) -> Option { + match (state, event) { + (None, Event::Created(event)) => Some(event.into()), + (state, event) => panic!("invalid message event {event:#?} for state {state:#?}"), + } + } +} + +impl FromIterator for Option { + fn from_iter>(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 for Login { + fn from(event: Created) -> Self { + event.login + } +} diff --git a/src/message/app.rs b/src/message/app.rs index 385c92e..3385af2 100644 --- a/src/message/app.rs +++ b/src/message/app.rs @@ -7,7 +7,7 @@ use crate::{ channel::{self, repo::Provider as _}, clock::DateTime, db::NotFound as _, - event::{broadcaster::Broadcaster, repo::Provider as _, Event, Sequence}, + event::{repo::Provider as _, Broadcaster, Event, Sequence}, login::Login, }; diff --git a/src/message/repo.rs b/src/message/repo.rs index e098fb2..0560f4a 100644 --- a/src/message/repo.rs +++ b/src/message/repo.rs @@ -38,6 +38,10 @@ impl<'c> Messages<'c> { values ($1, $2, $3, $4, $5, $6) returning id as "id: Id", + channel as "channel: channel::Id", + sender as "sender: login::Id", + sent_at as "sent_at: DateTime", + sent_sequence as "sent_sequence: Sequence", body "#, id, @@ -49,9 +53,12 @@ impl<'c> Messages<'c> { ) .map(|row| History { message: Message { - sent: *sent, - channel: channel.id().clone(), - sender: sender.clone(), + sent: Instant { + at: row.sent_at, + sequence: row.sent_sequence, + }, + channel: row.channel, + sender: row.sender, id: row.id, body: row.body, }, @@ -72,18 +79,16 @@ impl<'c> Messages<'c> { let messages = sqlx::query!( r#" select - message.channel as "channel: channel::Id", - sender.id as "sender_id: login::Id", - sender.name as "sender_name", - message.id as "id: Id", - message.body, + channel as "channel: channel::Id", + sender as "sender: login::Id", + id as "id: Id", + body, sent_at as "sent_at: DateTime", sent_sequence as "sent_sequence: Sequence" from message - join login as sender on message.sender = sender.id - where message.channel = $1 - and coalesce(message.sent_sequence <= $2, true) - order by message.sent_sequence + where channel = $1 + and coalesce(sent_sequence <= $2, true) + order by sent_sequence "#, channel_id, resume_at, @@ -95,10 +100,7 @@ impl<'c> Messages<'c> { sequence: row.sent_sequence, }, channel: row.channel, - sender: Login { - id: row.sender_id, - name: row.sender_name, - }, + sender: row.sender, id: row.id, body: row.body, }, @@ -114,16 +116,14 @@ impl<'c> Messages<'c> { let message = sqlx::query!( r#" select - message.channel as "channel: channel::Id", - sender.id as "sender_id: login::Id", - sender.name as "sender_name", - message.id as "id: Id", - message.body, + channel as "channel: channel::Id", + sender as "sender: login::Id", + id as "id: Id", + body, sent_at as "sent_at: DateTime", sent_sequence as "sent_sequence: Sequence" from message - join login as sender on message.sender = sender.id - where message.id = $1 + where id = $1 "#, message, ) @@ -134,10 +134,7 @@ impl<'c> Messages<'c> { sequence: row.sent_sequence, }, channel: row.channel, - sender: Login { - id: row.sender_id, - name: row.sender_name, - }, + sender: row.sender, id: row.id, body: row.body, }, @@ -194,15 +191,13 @@ impl<'c> Messages<'c> { let messages = sqlx::query!( r#" select - message.channel as "channel: channel::Id", - sender.id as "sender_id: login::Id", - sender.name as "sender_name", - message.id as "id: Id", - message.body, + channel as "channel: channel::Id", + sender as "sender: login::Id", + id as "id: Id", + body, sent_at as "sent_at: DateTime", sent_sequence as "sent_sequence: Sequence" from message - join login as sender on message.sender = sender.id where coalesce(message.sent_sequence > $1, true) "#, resume_at, @@ -214,10 +209,7 @@ impl<'c> Messages<'c> { sequence: row.sent_sequence, }, channel: row.channel, - sender: Login { - id: row.sender_id, - name: row.sender_name, - }, + sender: row.sender, id: row.id, body: row.body, }, diff --git a/src/message/snapshot.rs b/src/message/snapshot.rs index 8c1e62a..0eb37bb 100644 --- a/src/message/snapshot.rs +++ b/src/message/snapshot.rs @@ -2,14 +2,14 @@ use super::{ event::{Event, Sent}, Id, }; -use crate::{channel, event::Instant, login::Login}; +use crate::{channel, event::Instant, login}; #[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] pub struct Message { #[serde(flatten)] pub sent: Instant, pub channel: channel::Id, - pub sender: Login, + pub sender: login::Id, pub id: Id, pub body: String, } diff --git a/src/test/fixtures/login.rs b/src/test/fixtures/login.rs index 00c2789..e5ac716 100644 --- a/src/test/fixtures/login.rs +++ b/src/test/fixtures/login.rs @@ -3,23 +3,24 @@ use uuid::Uuid; use crate::{ app::App, + clock::RequestedAt, login::{self, Login, Password}, }; -pub async fn create_with_password(app: &App) -> (String, Password) { +pub async fn create_with_password(app: &App, created_at: &RequestedAt) -> (String, Password) { let (name, password) = propose(); app.logins() - .create(&name, &password) + .create(&name, &password, created_at) .await .expect("should always succeed if the login is actually new"); (name, password) } -pub async fn create(app: &App) -> Login { +pub async fn create(app: &App, created_at: &RequestedAt) -> Login { let (name, password) = propose(); app.logins() - .create(&name, &password) + .create(&name, &password, created_at) .await .expect("should always succeed if the login is actually new") } diff --git a/src/token/app.rs b/src/token/app.rs index 5c4fcd5..b8af637 100644 --- a/src/token/app.rs +++ b/src/token/app.rs @@ -7,23 +7,34 @@ use futures::{ use sqlx::sqlite::SqlitePool; use super::{ - broadcaster::Broadcaster, event, repo::auth::Provider as _, repo::Provider as _, Id, Secret, + repo::auth::Provider as _, repo::Provider as _, Broadcaster, Event as TokenEvent, Id, Secret, }; use crate::{ clock::DateTime, db::NotFound as _, + event::{self, repo::Provider as _, Event as ServiceEvent}, login::{repo::Provider as _, Login, Password}, }; pub struct Tokens<'a> { db: &'a SqlitePool, - tokens: &'a Broadcaster, + events: &'a event::Broadcaster, + token_events: &'a Broadcaster, } impl<'a> Tokens<'a> { - pub const fn new(db: &'a SqlitePool, tokens: &'a Broadcaster) -> Self { - Self { db, tokens } + pub const fn new( + db: &'a SqlitePool, + events: &'a event::Broadcaster, + token_events: &'a Broadcaster, + ) -> Self { + Self { + db, + events, + token_events, + } } + pub async fn login( &self, name: &str, @@ -32,22 +43,30 @@ impl<'a> Tokens<'a> { ) -> Result { let mut tx = self.db.begin().await?; - let login = if let Some((login, stored_hash)) = tx.auth().for_name(name).await? { + let (login, created) = if let Some((login, stored_hash)) = tx.auth().for_name(name).await? { if stored_hash.verify(password)? { - // Password verified; use the login. - login + // Password verified, proceed with login + (login, false) } else { // Password NOT verified. return Err(LoginError::Rejected); } } else { let password_hash = password.hash()?; - tx.logins().create(name, &password_hash).await? + let created = tx.sequence().next(login_at).await?; + let login = tx.logins().create(name, &password_hash, &created).await?; + + (login, true) }; let token = tx.tokens().issue(&login, login_at).await?; tx.commit().await?; + if created { + self.events + .broadcast(login.events().map(ServiceEvent::from).collect::>()); + } + Ok(token) } @@ -76,7 +95,7 @@ impl<'a> Tokens<'a> { E: std::fmt::Debug, { // Subscribe, first. - let token_events = self.tokens.subscribe(); + let token_events = self.token_events.subscribe(); // Check that the token is valid at this point in time, second. If it is, then // any future revocations will appear in the subscription. If not, bail now. @@ -102,7 +121,9 @@ impl<'a> Tokens<'a> { // Then construct the guarded stream. First, project both streams into // `GuardedEvent`. let token_events = token_events - .filter(move |event| future::ready(event.token == token)) + .filter(move |event| { + future::ready(matches!(event, TokenEvent::Revoked(id) if id == &token)) + }) .map(|_| GuardedEvent::TokenRevoked); let events = events.map(|event| GuardedEvent::Event(event)); @@ -126,8 +147,8 @@ impl<'a> Tokens<'a> { let tokens = tx.tokens().expire(&expire_at).await?; tx.commit().await?; - for event in tokens.into_iter().map(event::TokenRevoked::from) { - self.tokens.broadcast(event); + for event in tokens.into_iter().map(TokenEvent::Revoked) { + self.token_events.broadcast(event); } Ok(()) @@ -138,8 +159,8 @@ impl<'a> Tokens<'a> { tx.tokens().revoke(token).await?; tx.commit().await?; - self.tokens - .broadcast(event::TokenRevoked::from(token.clone())); + self.token_events + .broadcast(TokenEvent::Revoked(token.clone())); Ok(()) } diff --git a/src/token/broadcaster.rs b/src/token/broadcaster.rs index 8e2e006..de2513a 100644 --- a/src/token/broadcaster.rs +++ b/src/token/broadcaster.rs @@ -1,4 +1,3 @@ -use super::event; use crate::broadcast; -pub type Broadcaster = broadcast::Broadcaster; +pub type Broadcaster = broadcast::Broadcaster; diff --git a/src/token/event.rs b/src/token/event.rs index d53d436..51b74d7 100644 --- a/src/token/event.rs +++ b/src/token/event.rs @@ -1,12 +1,6 @@ use crate::token; #[derive(Clone, Debug)] -pub struct TokenRevoked { - pub token: token::Id, -} - -impl From for TokenRevoked { - fn from(token: token::Id) -> Self { - Self { token } - } +pub enum Event { + Revoked(token::Id), } diff --git a/src/token/mod.rs b/src/token/mod.rs index d122611..eccb3cd 100644 --- a/src/token/mod.rs +++ b/src/token/mod.rs @@ -1,9 +1,9 @@ pub mod app; -pub mod broadcaster; +mod broadcaster; mod event; pub mod extract; mod id; mod repo; mod secret; -pub use self::{id::Id, secret::Secret}; +pub use self::{broadcaster::Broadcaster, event::Event, id::Id, secret::Secret}; diff --git a/src/token/repo/auth.rs b/src/token/repo/auth.rs index b299697..ddb5136 100644 --- a/src/token/repo/auth.rs +++ b/src/token/repo/auth.rs @@ -1,6 +1,10 @@ use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; -use crate::login::{self, password::StoredHash, Login}; +use crate::{ + clock::DateTime, + event::{Instant, Sequence}, + login::{self, password::StoredHash, History, Login}, +}; pub trait Provider { fn auth(&mut self) -> Auth; @@ -21,25 +25,33 @@ impl<'t> Auth<'t> { pub async fn for_name( &mut self, name: &str, - ) -> Result, sqlx::Error> { + ) -> Result, sqlx::Error> { let found = sqlx::query!( r#" select id as "id: login::Id", name, - password_hash as "password_hash: StoredHash" + password_hash as "password_hash: StoredHash", + created_sequence as "created_sequence: Sequence", + created_at as "created_at: DateTime" from login where name = $1 "#, name, ) - .map(|rec| { + .map(|row| { ( - Login { - id: rec.id, - name: rec.name, + History { + login: Login { + id: row.id, + name: row.name, + }, + created: Instant { + at: row.created_at, + sequence: row.created_sequence, + }, }, - rec.password_hash, + row.password_hash, ) }) .fetch_optional(&mut *self.0) diff --git a/src/token/repo/token.rs b/src/token/repo/token.rs index 5f64dac..c592dcd 100644 --- a/src/token/repo/token.rs +++ b/src/token/repo/token.rs @@ -3,7 +3,7 @@ use uuid::Uuid; use crate::{ clock::DateTime, - login::{self, Login}, + login::{self, History, Login}, token::{Id, Secret}, }; @@ -24,11 +24,12 @@ impl<'c> Tokens<'c> { // be used to control expiry, until the token is actually used. pub async fn issue( &mut self, - login: &Login, + login: &History, issued_at: &DateTime, ) -> Result { let id = Id::generate(); let secret = Uuid::new_v4().to_string(); + let login = login.id(); let secret = sqlx::query_scalar!( r#" @@ -39,7 +40,7 @@ impl<'c> Tokens<'c> { "#, id, secret, - login.id, + login, issued_at, ) .fetch_one(&mut *self.0) @@ -127,7 +128,7 @@ impl<'c> Tokens<'c> { select token.id as "token_id: Id", login.id as "login_id: login::Id", - name as "login_name" + login.name as "login_name" from login join token on login.id = token.login where token.secret = $1 -- cgit v1.2.3