diff options
Diffstat (limited to 'src')
33 files changed, 956 insertions, 533 deletions
@@ -4,6 +4,7 @@ use crate::{ channel::app::Channels, event::{app::Events, broadcaster::Broadcaster as EventBroadcaster}, login::app::Logins, + message::app::Messages, token::{app::Tokens, broadcaster::Broadcaster as TokenBroadcaster}, }; @@ -35,6 +36,10 @@ impl App { Logins::new(&self.db) } + pub const fn messages(&self) -> Messages { + Messages::new(&self.db, &self.events) + } + pub const fn tokens(&self) -> Tokens { Tokens::new(&self.db, &self.tokens) } diff --git a/src/broadcast.rs b/src/broadcast.rs index 083a301..bedc263 100644 --- a/src/broadcast.rs +++ b/src/broadcast.rs @@ -32,7 +32,7 @@ where { // panic: if ``message.channel.id`` has not been previously registered, // and was not part of the initial set of channels. - pub fn broadcast(&self, message: &M) { + pub fn broadcast(&self, message: impl Into<M>) { let tx = self.sender(); // Per the Tokio docs, the returned error is only used to indicate that @@ -42,7 +42,7 @@ where // // The successful return value, which includes the number of active // receivers, also isn't that interesting to us. - let _ = tx.send(message.clone()); + let _ = tx.send(message.into()); } // panic: if ``channel`` has not been previously registered, and was not diff --git a/src/channel/app.rs b/src/channel/app.rs index b7e3a10..6ce826b 100644 --- a/src/channel/app.rs +++ b/src/channel/app.rs @@ -1,10 +1,11 @@ use chrono::TimeDelta; +use itertools::Itertools; use sqlx::sqlite::SqlitePool; use crate::{ channel::{repo::Provider as _, Channel}, clock::DateTime, - event::{broadcaster::Broadcaster, repo::Provider as _, types::ChannelEvent, Sequence}, + event::{broadcaster::Broadcaster, repo::Provider as _, Sequence}, }; pub struct Channels<'a> { @@ -27,10 +28,11 @@ impl<'a> Channels<'a> { .map_err(|err| CreateError::from_duplicate_name(err, name))?; tx.commit().await?; - self.events - .broadcast(&ChannelEvent::created(channel.clone())); + for event in channel.events() { + self.events.broadcast(event); + } - Ok(channel) + Ok(channel.snapshot()) } pub async fn all(&self, resume_point: Option<Sequence>) -> Result<Vec<Channel>, InternalError> { @@ -38,6 +40,16 @@ impl<'a> Channels<'a> { let channels = tx.channels().all(resume_point).await?; tx.commit().await?; + let channels = channels + .into_iter() + .filter_map(|channel| { + channel + .events() + .filter(Sequence::up_to(resume_point)) + .collect() + }) + .collect(); + Ok(channels) } @@ -51,14 +63,21 @@ impl<'a> Channels<'a> { let mut events = Vec::with_capacity(expired.len()); for channel in expired { let deleted = tx.sequence().next(relative_to).await?; - let event = tx.channels().delete(&channel, &deleted).await?; - events.push(event); + let channel = tx.channels().delete(&channel, &deleted).await?; + events.push( + channel + .events() + .filter(Sequence::start_from(deleted.sequence)), + ); } tx.commit().await?; - for event in events { - self.events.broadcast(&event); + for event in events + .into_iter() + .kmerge_by(|a, b| a.instant.sequence < b.instant.sequence) + { + self.events.broadcast(event); } Ok(()) diff --git a/src/channel/event.rs b/src/channel/event.rs new file mode 100644 index 0000000..9c54174 --- /dev/null +++ b/src/channel/event.rs @@ -0,0 +1,48 @@ +use super::Channel; +use crate::{ + channel, + event::{Instant, Sequenced}, +}; + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +pub struct Event { + #[serde(flatten)] + pub instant: Instant, + #[serde(flatten)] + pub kind: Kind, +} + +impl Sequenced for Event { + fn instant(&self) -> Instant { + self.instant + } +} + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum Kind { + Created(Created), + Deleted(Deleted), +} + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +pub struct Created { + pub channel: Channel, +} + +impl From<Created> for Kind { + fn from(event: Created) -> Self { + Self::Created(event) + } +} + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +pub struct Deleted { + pub channel: channel::Id, +} + +impl From<Deleted> for Kind { + fn from(event: Deleted) -> Self { + Self::Deleted(event) + } +} diff --git a/src/channel/history.rs b/src/channel/history.rs new file mode 100644 index 0000000..3cc7d9d --- /dev/null +++ b/src/channel/history.rs @@ -0,0 +1,42 @@ +use super::{ + event::{Created, Deleted, Event}, + Channel, +}; +use crate::event::Instant; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct History { + pub channel: Channel, + pub created: Instant, + pub deleted: Option<Instant>, +} + +impl History { + fn created(&self) -> Event { + Event { + instant: self.created, + kind: Created { + channel: self.channel.clone(), + } + .into(), + } + } + + fn deleted(&self) -> Option<Event> { + self.deleted.map(|instant| Event { + instant, + kind: Deleted { + channel: self.channel.id.clone(), + } + .into(), + }) + } + + pub fn events(&self) -> impl Iterator<Item = Event> { + [self.created()].into_iter().chain(self.deleted()) + } + + pub fn snapshot(&self) -> Channel { + self.channel.clone() + } +} diff --git a/src/channel/mod.rs b/src/channel/mod.rs index 4baa7e3..eb8200b 100644 --- a/src/channel/mod.rs +++ b/src/channel/mod.rs @@ -1,16 +1,9 @@ -use crate::event::Instant; - pub mod app; +pub mod event; +mod history; mod id; pub mod repo; mod routes; +mod snapshot; -pub use self::{id::Id, routes::router}; - -#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] -pub struct Channel { - pub id: Id, - pub name: String, - #[serde(skip)] - pub created: Instant, -} +pub use self::{event::Event, history::History, id::Id, routes::router, snapshot::Channel}; diff --git a/src/channel/repo.rs b/src/channel/repo.rs index c000b56..8bb761b 100644 --- a/src/channel/repo.rs +++ b/src/channel/repo.rs @@ -1,9 +1,9 @@ use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; use crate::{ - channel::{Channel, Id}, + channel::{Channel, History, Id}, clock::DateTime, - event::{types, Instant, Sequence}, + event::{Instant, Sequence}, }; pub trait Provider { @@ -19,7 +19,7 @@ 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<Channel, sqlx::Error> { + pub async fn create(&mut self, name: &str, created: &Instant) -> Result<History, sqlx::Error> { let id = Id::generate(); let channel = sqlx::query!( r#" @@ -37,13 +37,16 @@ impl<'c> Channels<'c> { created.at, created.sequence, ) - .map(|row| Channel { - id: row.id, - name: row.name, + .map(|row| History { + channel: Channel { + id: row.id, + name: row.name, + }, created: Instant { at: row.created_at, sequence: row.created_sequence, }, + deleted: None, }) .fetch_one(&mut *self.0) .await?; @@ -51,7 +54,7 @@ impl<'c> Channels<'c> { Ok(channel) } - pub async fn by_id(&mut self, channel: &Id) -> Result<Channel, sqlx::Error> { + pub async fn by_id(&mut self, channel: &Id) -> Result<History, sqlx::Error> { let channel = sqlx::query!( r#" select @@ -64,13 +67,16 @@ impl<'c> Channels<'c> { "#, channel, ) - .map(|row| Channel { - id: row.id, - name: row.name, + .map(|row| History { + channel: Channel { + id: row.id, + name: row.name, + }, created: Instant { at: row.created_at, sequence: row.created_sequence, }, + deleted: None, }) .fetch_one(&mut *self.0) .await?; @@ -81,7 +87,7 @@ impl<'c> Channels<'c> { pub async fn all( &mut self, resume_point: Option<Sequence>, - ) -> Result<Vec<Channel>, sqlx::Error> { + ) -> Result<Vec<History>, sqlx::Error> { let channels = sqlx::query!( r#" select @@ -95,13 +101,16 @@ impl<'c> Channels<'c> { "#, resume_point, ) - .map(|row| Channel { - id: row.id, - name: row.name, + .map(|row| History { + channel: Channel { + id: row.id, + name: row.name, + }, created: Instant { at: row.created_at, sequence: row.created_sequence, }, + deleted: None, }) .fetch_all(&mut *self.0) .await?; @@ -112,7 +121,7 @@ impl<'c> Channels<'c> { pub async fn replay( &mut self, resume_at: Option<Sequence>, - ) -> Result<Vec<Channel>, sqlx::Error> { + ) -> Result<Vec<History>, sqlx::Error> { let channels = sqlx::query!( r#" select @@ -125,13 +134,16 @@ impl<'c> Channels<'c> { "#, resume_at, ) - .map(|row| Channel { - id: row.id, - name: row.name, + .map(|row| History { + channel: Channel { + id: row.id, + name: row.name, + }, created: Instant { at: row.created_at, sequence: row.created_sequence, }, + deleted: None, }) .fetch_all(&mut *self.0) .await?; @@ -141,35 +153,43 @@ impl<'c> Channels<'c> { pub async fn delete( &mut self, - channel: &Channel, + channel: &Id, deleted: &Instant, - ) -> Result<types::ChannelEvent, sqlx::Error> { - let channel = channel.id.clone(); - sqlx::query_scalar!( + ) -> Result<History, sqlx::Error> { + let channel = sqlx::query!( r#" delete from channel where id = $1 - returning 1 as "row: i64" + returning + id as "id: Id", + name, + created_at as "created_at: DateTime", + created_sequence as "created_sequence: Sequence" "#, channel, ) + .map(|row| History { + channel: Channel { + id: row.id, + name: row.name, + }, + created: Instant { + at: row.created_at, + sequence: row.created_sequence, + }, + deleted: Some(*deleted), + }) .fetch_one(&mut *self.0) .await?; - Ok(types::ChannelEvent { - instant: *deleted, - data: types::DeletedEvent { channel }.into(), - }) + Ok(channel) } - pub async fn expired(&mut self, expired_at: &DateTime) -> Result<Vec<Channel>, sqlx::Error> { - let channels = sqlx::query!( + pub async fn expired(&mut self, expired_at: &DateTime) -> Result<Vec<Id>, sqlx::Error> { + let channels = sqlx::query_scalar!( r#" select - channel.id as "id: Id", - channel.name, - channel.created_at as "created_at: DateTime", - channel.created_sequence as "created_sequence: Sequence" + channel.id as "id: Id" from channel left join message where created_at < $1 @@ -177,14 +197,6 @@ impl<'c> Channels<'c> { "#, expired_at, ) - .map(|row| Channel { - id: row.id, - name: row.name, - created: Instant { - at: row.created_at, - sequence: row.created_sequence, - }, - }) .fetch_all(&mut *self.0) .await?; diff --git a/src/channel/routes.rs b/src/channel/routes.rs index 5d8b61e..5bb1ee9 100644 --- a/src/channel/routes.rs +++ b/src/channel/routes.rs @@ -13,8 +13,9 @@ use crate::{ channel::{self, Channel}, clock::RequestedAt, error::Internal, - event::{app::EventsError, Sequence}, + event::Sequence, login::Login, + message::app::Error as MessageError, }; #[cfg(test)] @@ -99,8 +100,8 @@ async fn on_send( login: Login, Json(request): Json<SendRequest>, ) -> Result<StatusCode, ErrorResponse> { - app.events() - .send(&login, &channel, &request.message, &sent_at) + app.messages() + .send(&channel, &login, &sent_at, &request.message) .await // Could impl `From` here, but it's more code and this is used once. .map_err(ErrorResponse)?; @@ -109,13 +110,13 @@ async fn on_send( } #[derive(Debug)] -struct ErrorResponse(EventsError); +struct ErrorResponse(MessageError); impl IntoResponse for ErrorResponse { fn into_response(self) -> Response { let Self(error) = self; match error { - not_found @ EventsError::ChannelNotFound(_) => { + not_found @ MessageError::ChannelNotFound(_) => { (StatusCode::NOT_FOUND, not_found.to_string()).into_response() } other => Internal::from(other).into_response(), diff --git a/src/channel/routes/test/on_create.rs b/src/channel/routes/test/on_create.rs index 9988932..5733c9e 100644 --- a/src/channel/routes/test/on_create.rs +++ b/src/channel/routes/test/on_create.rs @@ -3,7 +3,7 @@ use futures::stream::StreamExt as _; use crate::{ channel::{app, routes}, - event::types, + event, test::fixtures::{self, future::Immediately as _}, }; @@ -50,8 +50,8 @@ async fn new_channel() { .expect("creation event published"); assert!(matches!( - event.data, - types::ChannelEventData::Created(event) + event.kind, + event::Kind::ChannelCreated(event) if event.channel == response_channel )); } diff --git a/src/channel/routes/test/on_send.rs b/src/channel/routes/test/on_send.rs index 33ec3b7..1027b29 100644 --- a/src/channel/routes/test/on_send.rs +++ b/src/channel/routes/test/on_send.rs @@ -4,7 +4,8 @@ use futures::stream::StreamExt; use crate::{ channel, channel::routes, - event::{app, types}, + event, + message::app, test::fixtures::{self, future::Immediately as _}, }; @@ -54,10 +55,10 @@ async fn messages_in_order() { for ((sent_at, message), event) in requests.into_iter().zip(events) { assert_eq!(*sent_at, event.instant.at); assert!(matches!( - event.data, - types::ChannelEventData::Message(event_message) - if event_message.sender == sender - && event_message.message.body == message + event.kind, + event::Kind::MessageSent(event) + if event.message.sender == sender + && event.message.body == message )); } } @@ -90,6 +91,6 @@ async fn nonexistent_channel() { assert!(matches!( error, - app::EventsError::ChannelNotFound(error_channel) if channel == error_channel + app::Error::ChannelNotFound(error_channel) if channel == error_channel )); } diff --git a/src/channel/snapshot.rs b/src/channel/snapshot.rs new file mode 100644 index 0000000..6462f25 --- /dev/null +++ b/src/channel/snapshot.rs @@ -0,0 +1,38 @@ +use super::{ + event::{Created, Event, Kind}, + Id, +}; + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +pub struct Channel { + pub id: Id, + pub name: String, +} + +impl Channel { + fn apply(state: Option<Self>, event: Event) -> Option<Self> { + match (state, event.kind) { + (None, Kind::Created(event)) => Some(event.into()), + (Some(channel), Kind::Deleted(event)) if channel.id == event.channel => None, + (state, event) => panic!("invalid channel event {event:#?} for state {state:#?}"), + } + } +} + +impl FromIterator<Event> for Option<Channel> { + fn from_iter<I: IntoIterator<Item = Event>>(events: I) -> Self { + events.into_iter().fold(None, Channel::apply) + } +} + +impl From<&Created> for Channel { + fn from(event: &Created) -> Self { + event.channel.clone() + } +} + +impl From<Created> for Channel { + fn from(event: Created) -> Self { + event.channel + } +} diff --git a/src/event/app.rs b/src/event/app.rs index 5e9e79a..e58bea9 100644 --- a/src/event/app.rs +++ b/src/event/app.rs @@ -1,22 +1,15 @@ -use chrono::TimeDelta; use futures::{ future, stream::{self, StreamExt as _}, Stream, }; +use itertools::Itertools as _; use sqlx::sqlite::SqlitePool; -use super::{ - broadcaster::Broadcaster, - repo::message::Provider as _, - types::{self, ChannelEvent}, -}; +use super::{broadcaster::Broadcaster, Event, Sequence, Sequenced}; use crate::{ channel::{self, repo::Provider as _}, - clock::DateTime, - db::NotFound as _, - event::{repo::Provider as _, Sequence}, - login::Login, + message::{self, repo::Provider as _}, }; pub struct Events<'a> { @@ -29,111 +22,52 @@ impl<'a> Events<'a> { Self { db, events } } - pub async fn send( - &self, - login: &Login, - channel: &channel::Id, - body: &str, - sent_at: &DateTime, - ) -> Result<types::ChannelEvent, EventsError> { - let mut tx = self.db.begin().await?; - let channel = tx - .channels() - .by_id(channel) - .await - .not_found(|| EventsError::ChannelNotFound(channel.clone()))?; - let sent = tx.sequence().next(sent_at).await?; - let event = tx - .message_events() - .create(login, &channel, &sent, body) - .await?; - tx.commit().await?; - - self.events.broadcast(&event); - Ok(event) - } - - pub async fn expire(&self, relative_to: &DateTime) -> Result<(), sqlx::Error> { - // Somewhat arbitrarily, expire after 90 days. - let expire_at = relative_to.to_owned() - TimeDelta::days(90); - - let mut tx = self.db.begin().await?; - let expired = tx.message_events().expired(&expire_at).await?; - - let mut events = Vec::with_capacity(expired.len()); - for (channel, message) in expired { - let deleted = tx.sequence().next(relative_to).await?; - let event = tx - .message_events() - .delete(&channel, &message, &deleted) - .await?; - events.push(event); - } - - tx.commit().await?; - - for event in events { - self.events.broadcast(&event); - } - - Ok(()) - } - pub async fn subscribe( &self, resume_at: Option<Sequence>, - ) -> Result<impl Stream<Item = types::ChannelEvent> + std::fmt::Debug, sqlx::Error> { + ) -> Result<impl Stream<Item = Event> + std::fmt::Debug, sqlx::Error> { // Subscribe before retrieving, to catch messages broadcast while we're // querying the DB. We'll prune out duplicates later. let live_messages = self.events.subscribe(); let mut tx = self.db.begin().await?; - let channels = tx.channels().replay(resume_at).await?; + let channels = tx.channels().replay(resume_at).await?; let channel_events = channels - .into_iter() - .map(ChannelEvent::created) - .filter(move |event| { - resume_at.map_or(true, |resume_at| Sequence::from(event) > resume_at) - }); - - let message_events = tx.message_events().replay(resume_at).await?; - - let mut replay_events = channel_events - .into_iter() - .chain(message_events.into_iter()) + .iter() + .map(channel::History::events) + .kmerge_by(|a, b| a.instant.sequence < b.instant.sequence) + .filter(Sequence::after(resume_at)) + .map(Event::from); + + let messages = tx.messages().replay(resume_at).await?; + let message_events = messages + .iter() + .map(message::History::events) + .kmerge_by(|a, b| a.instant.sequence < b.instant.sequence) + .filter(Sequence::after(resume_at)) + .map(Event::from); + + let replay_events = channel_events + .merge_by(message_events, |a, b| { + a.instant.sequence < b.instant.sequence + }) .collect::<Vec<_>>(); - replay_events.sort_by_key(|event| Sequence::from(event)); - let resume_live_at = replay_events.last().map(Sequence::from); + let resume_live_at = replay_events.last().map(Sequenced::sequence); let replay = stream::iter(replay_events); - // no skip_expired or resume transforms for stored_messages, as it's - // constructed not to contain messages meeting either criterion. - // - // * skip_expired is redundant with the `tx.broadcasts().expire(…)` call; - // * resume is redundant with the resume_at argument to - // `tx.broadcasts().replay(…)`. let live_messages = live_messages // Filtering on the broadcast resume point filters out messages // before resume_at, and filters out messages duplicated from - // stored_messages. + // `replay_events`. .filter(Self::resume(resume_live_at)); Ok(replay.chain(live_messages)) } - fn resume( - resume_at: Option<Sequence>, - ) -> impl for<'m> FnMut(&'m types::ChannelEvent) -> future::Ready<bool> { - move |event| future::ready(resume_at < Some(Sequence::from(event))) + fn resume(resume_at: Option<Sequence>) -> impl for<'m> FnMut(&'m Event) -> future::Ready<bool> { + let filter = Sequence::after(resume_at); + move |event| future::ready(filter(event)) } } - -#[derive(Debug, thiserror::Error)] -pub enum EventsError { - #[error("channel {0} not found")] - ChannelNotFound(channel::Id), - #[error(transparent)] - DatabaseError(#[from] sqlx::Error), -} diff --git a/src/event/broadcaster.rs b/src/event/broadcaster.rs index 92f631f..de2513a 100644 --- a/src/event/broadcaster.rs +++ b/src/event/broadcaster.rs @@ -1,3 +1,3 @@ -use crate::{broadcast, event::types}; +use crate::broadcast; -pub type Broadcaster = broadcast::Broadcaster<types::ChannelEvent>; +pub type Broadcaster = broadcast::Broadcaster<super::Event>; diff --git a/src/event/mod.rs b/src/event/mod.rs index c982d3a..1503b77 100644 --- a/src/event/mod.rs +++ b/src/event/mod.rs @@ -1,18 +1,75 @@ +use crate::{channel, message}; + pub mod app; pub mod broadcaster; mod extract; pub mod repo; mod routes; mod sequence; -pub mod types; -use crate::clock::DateTime; +pub use self::{ + routes::router, + sequence::{Instant, Sequence, Sequenced}, +}; + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +pub struct Event { + #[serde(flatten)] + pub instant: Instant, + #[serde(flatten)] + pub kind: Kind, +} + +impl Sequenced for Event { + fn instant(&self) -> Instant { + self.instant + } +} + +impl From<channel::Event> for Event { + fn from(event: channel::Event) -> Self { + Self { + instant: event.instant, + kind: event.kind.into(), + } + } +} + +impl From<message::Event> for Event { + fn from(event: message::Event) -> Self { + Self { + instant: event.instant, + kind: event.kind.into(), + } + } +} -pub use self::{routes::router, sequence::Sequence}; +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum Kind { + #[serde(rename = "created")] + ChannelCreated(channel::event::Created), + #[serde(rename = "message")] + MessageSent(message::event::Sent), + MessageDeleted(message::event::Deleted), + #[serde(rename = "deleted")] + ChannelDeleted(channel::event::Deleted), +} + +impl From<channel::event::Kind> for Kind { + fn from(kind: channel::event::Kind) -> Self { + match kind { + channel::event::Kind::Created(created) => Self::ChannelCreated(created), + channel::event::Kind::Deleted(deleted) => Self::ChannelDeleted(deleted), + } + } +} -#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Serialize)] -pub struct Instant { - pub at: DateTime, - #[serde(skip)] - pub sequence: Sequence, +impl From<message::event::Kind> for Kind { + fn from(kind: message::event::Kind) -> Self { + match kind { + message::event::Kind::Sent(created) => Self::MessageSent(created), + message::event::Kind::Deleted(deleted) => Self::MessageDeleted(deleted), + } + } } diff --git a/src/event/repo/sequence.rs b/src/event/repo.rs index 40d6a53..40d6a53 100644 --- a/src/event/repo/sequence.rs +++ b/src/event/repo.rs diff --git a/src/event/repo/message.rs b/src/event/repo/message.rs deleted file mode 100644 index f29c8a4..0000000 --- a/src/event/repo/message.rs +++ /dev/null @@ -1,196 +0,0 @@ -use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; - -use crate::{ - channel::{self, Channel}, - clock::DateTime, - event::{types, Instant, Sequence}, - login::{self, Login}, - message::{self, Message}, -}; - -pub trait Provider { - fn message_events(&mut self) -> Events; -} - -impl<'c> Provider for Transaction<'c, Sqlite> { - fn message_events(&mut self) -> Events { - Events(self) - } -} - -pub struct Events<'t>(&'t mut SqliteConnection); - -impl<'c> Events<'c> { - pub async fn create( - &mut self, - sender: &Login, - channel: &Channel, - sent: &Instant, - body: &str, - ) -> Result<types::ChannelEvent, sqlx::Error> { - let id = message::Id::generate(); - - let message = sqlx::query!( - r#" - insert into message - (id, channel, sender, sent_at, sent_sequence, body) - values ($1, $2, $3, $4, $5, $6) - returning - id as "id: message::Id", - sender as "sender: login::Id", - sent_at as "sent_at: DateTime", - sent_sequence as "sent_sequence: Sequence", - body - "#, - id, - channel.id, - sender.id, - sent.at, - sent.sequence, - body, - ) - .map(|row| types::ChannelEvent { - instant: Instant { - at: row.sent_at, - sequence: row.sent_sequence, - }, - data: types::MessageEvent { - channel: channel.clone(), - sender: sender.clone(), - message: Message { - id: row.id, - body: row.body, - }, - } - .into(), - }) - .fetch_one(&mut *self.0) - .await?; - - Ok(message) - } - - pub async fn delete( - &mut self, - channel: &Channel, - message: &message::Id, - deleted: &Instant, - ) -> Result<types::ChannelEvent, sqlx::Error> { - sqlx::query_scalar!( - r#" - delete from message - where id = $1 - returning 1 as "row: i64" - "#, - message, - ) - .fetch_one(&mut *self.0) - .await?; - - Ok(types::ChannelEvent { - instant: Instant { - at: deleted.at, - sequence: deleted.sequence, - }, - data: types::MessageDeletedEvent { - channel: channel.clone(), - message: message.clone(), - } - .into(), - }) - } - - pub async fn expired( - &mut self, - expire_at: &DateTime, - ) -> Result<Vec<(Channel, message::Id)>, sqlx::Error> { - let messages = sqlx::query!( - r#" - select - channel.id as "channel_id: channel::Id", - channel.name as "channel_name", - channel.created_at as "channel_created_at: DateTime", - channel.created_sequence as "channel_created_sequence: Sequence", - message.id as "message: message::Id" - from message - join channel on message.channel = channel.id - join login as sender on message.sender = sender.id - where sent_at < $1 - "#, - expire_at, - ) - .map(|row| { - ( - Channel { - id: row.channel_id, - name: row.channel_name, - created: Instant { - at: row.channel_created_at, - sequence: row.channel_created_sequence, - }, - }, - row.message, - ) - }) - .fetch_all(&mut *self.0) - .await?; - - Ok(messages) - } - - pub async fn replay( - &mut self, - resume_at: Option<Sequence>, - ) -> Result<Vec<types::ChannelEvent>, sqlx::Error> { - let events = sqlx::query!( - r#" - select - message.id as "id: message::Id", - channel.id as "channel_id: channel::Id", - channel.name as "channel_name", - channel.created_at as "channel_created_at: DateTime", - channel.created_sequence as "channel_created_sequence: Sequence", - sender.id as "sender_id: login::Id", - sender.name as sender_name, - message.sent_at as "sent_at: DateTime", - message.sent_sequence as "sent_sequence: Sequence", - message.body - from message - join channel on message.channel = channel.id - join login as sender on message.sender = sender.id - where coalesce(message.sent_sequence > $1, true) - order by sent_sequence asc - "#, - resume_at, - ) - .map(|row| types::ChannelEvent { - instant: Instant { - at: row.sent_at, - sequence: row.sent_sequence, - }, - data: types::MessageEvent { - channel: Channel { - id: row.channel_id, - name: row.channel_name, - created: Instant { - at: row.channel_created_at, - sequence: row.channel_created_sequence, - }, - }, - sender: Login { - id: row.sender_id, - name: row.sender_name, - }, - message: Message { - id: row.id, - body: row.body, - }, - } - .into(), - }) - .fetch_all(&mut *self.0) - .await?; - - Ok(events) - } -} diff --git a/src/event/repo/mod.rs b/src/event/repo/mod.rs deleted file mode 100644 index cee840c..0000000 --- a/src/event/repo/mod.rs +++ /dev/null @@ -1,4 +0,0 @@ -pub mod message; -mod sequence; - -pub use self::sequence::Provider; diff --git a/src/event/routes.rs b/src/event/routes.rs index c87bfb2..5b9c7e3 100644 --- a/src/event/routes.rs +++ b/src/event/routes.rs @@ -10,11 +10,11 @@ use axum::{ use axum_extra::extract::Query; use futures::stream::{Stream, StreamExt as _}; -use super::{extract::LastEventId, types}; +use super::{extract::LastEventId, Event}; use crate::{ app::App, error::{Internal, Unauthorized}, - event::Sequence, + event::{Sequence, Sequenced as _}, token::{app::ValidateError, extract::Identity}, }; @@ -35,7 +35,7 @@ async fn events( identity: Identity, last_event_id: Option<LastEventId<Sequence>>, Query(query): Query<EventsQuery>, -) -> Result<Events<impl Stream<Item = types::ChannelEvent> + std::fmt::Debug>, EventsError> { +) -> Result<Events<impl Stream<Item = Event> + std::fmt::Debug>, EventsError> { let resume_at = last_event_id .map(LastEventId::into_inner) .or(query.resume_point); @@ -51,7 +51,7 @@ struct Events<S>(S); impl<S> IntoResponse for Events<S> where - S: Stream<Item = types::ChannelEvent> + Send + 'static, + S: Stream<Item = Event> + Send + 'static, { fn into_response(self) -> Response { let Self(stream) = self; @@ -62,11 +62,11 @@ where } } -impl TryFrom<types::ChannelEvent> for sse::Event { +impl TryFrom<Event> for sse::Event { type Error = serde_json::Error; - fn try_from(event: types::ChannelEvent) -> Result<Self, Self::Error> { - let id = serde_json::to_string(&Sequence::from(&event))?; + fn try_from(event: Event) -> Result<Self, Self::Error> { + let id = serde_json::to_string(&event.sequence())?; let data = serde_json::to_string_pretty(&event)?; let event = Self::default().id(id).data(data); diff --git a/src/event/routes/test.rs b/src/event/routes/test.rs index 68b55cc..ba9953e 100644 --- a/src/event/routes/test.rs +++ b/src/event/routes/test.rs @@ -6,7 +6,7 @@ use futures::{ }; use crate::{ - event::{routes, Sequence}, + event::{routes, Sequenced as _}, test::fixtures::{self, future::Immediately as _}, }; @@ -17,7 +17,7 @@ async fn includes_historical_message() { let app = fixtures::scratch_app().await; let sender = fixtures::login::create(&app).await; let channel = fixtures::channel::create(&app, &fixtures::now()).await; - let message = fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await; + let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await; // Call the endpoint @@ -36,7 +36,7 @@ async fn includes_historical_message() { .await .expect("delivered stored message"); - assert_eq!(message, event); + assert!(fixtures::event::message_sent(&event, &message)); } #[tokio::test] @@ -58,7 +58,7 @@ async fn includes_live_message() { // Verify the semantics let sender = fixtures::login::create(&app).await; - let message = fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await; + let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await; let event = events .filter(fixtures::filter::messages()) @@ -67,7 +67,7 @@ async fn includes_live_message() { .await .expect("delivered live message"); - assert_eq!(message, event); + assert!(fixtures::event::message_sent(&event, &message)); } #[tokio::test] @@ -87,7 +87,7 @@ async fn includes_multiple_channels() { let app = app.clone(); let sender = sender.clone(); let channel = channel.clone(); - async move { fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await } + async move { fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await } }) .collect::<Vec<_>>() .await; @@ -110,7 +110,9 @@ async fn includes_multiple_channels() { .await; for message in &messages { - assert!(events.iter().any(|event| { event == message })); + assert!(events + .iter() + .any(|event| fixtures::event::message_sent(event, message))); } } @@ -123,9 +125,9 @@ async fn sequential_messages() { let sender = fixtures::login::create(&app).await; let messages = vec![ - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, ]; // Call the endpoint @@ -138,7 +140,13 @@ async fn sequential_messages() { // Verify the structure of the response. - let mut events = events.filter(|event| future::ready(messages.contains(event))); + let mut events = events.filter(|event| { + future::ready( + messages + .iter() + .any(|message| fixtures::event::message_sent(event, message)), + ) + }); // Verify delivery in order for message in &messages { @@ -148,7 +156,7 @@ async fn sequential_messages() { .await .expect("undelivered messages remaining"); - assert_eq!(message, &event); + assert!(fixtures::event::message_sent(&event, message)); } } @@ -160,11 +168,11 @@ async fn resumes_from() { let channel = fixtures::channel::create(&app, &fixtures::now()).await; let sender = fixtures::login::create(&app).await; - let initial_message = fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await; + let initial_message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await; let later_messages = vec![ - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, ]; // Call the endpoint @@ -190,9 +198,9 @@ async fn resumes_from() { .await .expect("delivered events"); - assert_eq!(initial_message, event); + assert!(fixtures::event::message_sent(&event, &initial_message)); - Sequence::from(&event) + event.sequence() }; // Resume after disconnect @@ -214,7 +222,9 @@ async fn resumes_from() { .await; for message in &later_messages { - assert!(events.iter().any(|event| event == message)); + assert!(events + .iter() + .any(|event| fixtures::event::message_sent(event, message))); } } @@ -249,8 +259,8 @@ async fn serial_resume() { let resume_at = { let initial_messages = [ - fixtures::message::send(&app, &sender, &channel_a, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel_b, &fixtures::now()).await, + fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await, ]; // First subscription @@ -271,12 +281,14 @@ async fn serial_resume() { .await; for message in &initial_messages { - assert!(events.iter().any(|event| event == message)); + assert!(events + .iter() + .any(|event| fixtures::event::message_sent(event, message))); } let event = events.last().expect("this vec is non-empty"); - Sequence::from(event) + event.sequence() }; // Resume after disconnect @@ -285,8 +297,8 @@ async fn serial_resume() { // Note that channel_b does not appear here. The buggy behaviour // would be masked if channel_b happened to send a new message // into the resumed event stream. - fixtures::message::send(&app, &sender, &channel_a, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel_a, &fixtures::now()).await, + fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, ]; // Second subscription @@ -307,12 +319,14 @@ async fn serial_resume() { .await; for message in &resume_messages { - assert!(events.iter().any(|event| event == message)); + assert!(events + .iter() + .any(|event| fixtures::event::message_sent(event, message))); } let event = events.last().expect("this vec is non-empty"); - Sequence::from(event) + event.sequence() }; // Resume after disconnect a second time @@ -321,8 +335,8 @@ async fn serial_resume() { // problem. The resume point should before both of these messages, but // after _all_ prior messages. let final_messages = [ - fixtures::message::send(&app, &sender, &channel_a, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel_b, &fixtures::now()).await, + fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await, ]; // Third subscription @@ -345,7 +359,9 @@ async fn serial_resume() { // This set of messages, in particular, _should not_ include any prior // messages from `initial_messages` or `resume_messages`. for message in &final_messages { - assert!(events.iter().any(|event| event == message)); + assert!(events + .iter() + .any(|event| fixtures::event::message_sent(event, message))); } }; } @@ -378,13 +394,17 @@ async fn terminates_on_token_expiry() { // These should not be delivered. let messages = [ - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, ]; assert!(events - .filter(|event| future::ready(messages.contains(event))) + .filter(|event| future::ready( + messages + .iter() + .any(|message| fixtures::event::message_sent(event, message)) + )) .next() .immediately() .await @@ -425,13 +445,17 @@ async fn terminates_on_logout() { // These should not be delivered. let messages = [ - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, - fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, ]; assert!(events - .filter(|event| future::ready(messages.contains(event))) + .filter(|event| future::ready( + messages + .iter() + .any(|message| fixtures::event::message_sent(event, message)) + )) .next() .immediately() .await diff --git a/src/event/sequence.rs b/src/event/sequence.rs index 9ebddd7..c566156 100644 --- a/src/event/sequence.rs +++ b/src/event/sequence.rs @@ -1,5 +1,20 @@ use std::fmt; +use crate::clock::DateTime; + +#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Serialize)] +pub struct Instant { + pub at: DateTime, + #[serde(skip)] + pub sequence: Sequence, +} + +impl From<Instant> for Sequence { + fn from(instant: Instant) -> Self { + instant.sequence + } +} + #[derive( Clone, Copy, @@ -22,3 +37,47 @@ impl fmt::Display for Sequence { value.fmt(f) } } + +impl Sequence { + pub fn up_to<E>(resume_point: Option<Self>) -> impl for<'e> Fn(&'e E) -> bool + where + E: Sequenced, + { + move |event| resume_point.map_or(true, |resume_point| event.sequence() <= resume_point) + } + + pub fn after<E>(resume_point: Option<Self>) -> impl for<'e> Fn(&'e E) -> bool + where + E: Sequenced, + { + move |event| resume_point < Some(event.sequence()) + } + + pub fn start_from<E>(resume_point: Self) -> impl for<'e> Fn(&'e E) -> bool + where + E: Sequenced, + { + move |event| resume_point <= event.sequence() + } +} + +pub trait Sequenced { + fn instant(&self) -> Instant; + + fn sequence(&self) -> Sequence { + self.instant().into() + } +} + +impl<E> Sequenced for &E +where + E: Sequenced, +{ + fn instant(&self) -> Instant { + (*self).instant() + } + + fn sequence(&self) -> Sequence { + (*self).sequence() + } +} diff --git a/src/event/types.rs b/src/event/types.rs deleted file mode 100644 index 2324dc1..0000000 --- a/src/event/types.rs +++ /dev/null @@ -1,85 +0,0 @@ -use crate::{ - channel::{self, Channel}, - event::{Instant, Sequence}, - login::Login, - message::{self, Message}, -}; - -#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] -pub struct ChannelEvent { - #[serde(flatten)] - pub instant: Instant, - #[serde(flatten)] - pub data: ChannelEventData, -} - -impl ChannelEvent { - pub fn created(channel: Channel) -> Self { - Self { - instant: channel.created, - data: CreatedEvent { channel }.into(), - } - } -} - -impl<'c> From<&'c ChannelEvent> for Sequence { - fn from(event: &'c ChannelEvent) -> Self { - event.instant.sequence - } -} - -#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] -#[serde(tag = "type", rename_all = "snake_case")] -pub enum ChannelEventData { - Created(CreatedEvent), - Message(MessageEvent), - MessageDeleted(MessageDeletedEvent), - Deleted(DeletedEvent), -} - -#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] -pub struct CreatedEvent { - pub channel: Channel, -} - -impl From<CreatedEvent> for ChannelEventData { - fn from(event: CreatedEvent) -> Self { - Self::Created(event) - } -} - -#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] -pub struct MessageEvent { - pub channel: Channel, - pub sender: Login, - pub message: Message, -} - -impl From<MessageEvent> for ChannelEventData { - fn from(event: MessageEvent) -> Self { - Self::Message(event) - } -} - -#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] -pub struct MessageDeletedEvent { - pub channel: Channel, - pub message: message::Id, -} - -impl From<MessageDeletedEvent> for ChannelEventData { - fn from(event: MessageDeletedEvent) -> Self { - Self::MessageDeleted(event) - } -} - -#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] -pub struct DeletedEvent { - pub channel: channel::Id, -} - -impl From<DeletedEvent> for ChannelEventData { - fn from(event: DeletedEvent) -> Self { - Self::Deleted(event) - } -} diff --git a/src/expire.rs b/src/expire.rs index a8eb8ad..e50bcb4 100644 --- a/src/expire.rs +++ b/src/expire.rs @@ -14,7 +14,7 @@ pub async fn middleware( next: Next, ) -> Result<Response, Internal> { app.tokens().expire(&expired_at).await?; - app.events().expire(&expired_at).await?; + app.messages().expire(&expired_at).await?; app.channels().expire(&expired_at).await?; Ok(next.run(req).await) } diff --git a/src/message/app.rs b/src/message/app.rs new file mode 100644 index 0000000..51f772e --- /dev/null +++ b/src/message/app.rs @@ -0,0 +1,88 @@ +use chrono::TimeDelta; +use itertools::Itertools; +use sqlx::sqlite::SqlitePool; + +use super::{repo::Provider as _, Message}; +use crate::{ + channel::{self, repo::Provider as _}, + clock::DateTime, + db::NotFound as _, + event::{broadcaster::Broadcaster, repo::Provider as _, Sequence}, + login::Login, +}; + +pub struct Messages<'a> { + db: &'a SqlitePool, + events: &'a Broadcaster, +} + +impl<'a> Messages<'a> { + pub const fn new(db: &'a SqlitePool, events: &'a Broadcaster) -> Self { + Self { db, events } + } + + pub async fn send( + &self, + channel: &channel::Id, + sender: &Login, + sent_at: &DateTime, + body: &str, + ) -> Result<Message, Error> { + let mut tx = self.db.begin().await?; + let channel = tx + .channels() + .by_id(channel) + .await + .not_found(|| Error::ChannelNotFound(channel.clone()))?; + let sent = tx.sequence().next(sent_at).await?; + let message = tx + .messages() + .create(&channel.snapshot(), sender, &sent, body) + .await?; + tx.commit().await?; + + for event in message.events() { + self.events.broadcast(event); + } + + Ok(message.snapshot()) + } + + pub async fn expire(&self, relative_to: &DateTime) -> Result<(), sqlx::Error> { + // Somewhat arbitrarily, expire after 90 days. + let expire_at = relative_to.to_owned() - TimeDelta::days(90); + + let mut tx = self.db.begin().await?; + let expired = tx.messages().expired(&expire_at).await?; + + let mut events = Vec::with_capacity(expired.len()); + for (channel, message) in expired { + let deleted = tx.sequence().next(relative_to).await?; + let message = tx.messages().delete(&channel, &message, &deleted).await?; + events.push( + message + .events() + .filter(Sequence::start_from(deleted.sequence)), + ); + } + + tx.commit().await?; + + for event in events + .into_iter() + .kmerge_by(|a, b| a.instant.sequence < b.instant.sequence) + { + self.events.broadcast(event); + } + + Ok(()) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("channel {0} not found")] + ChannelNotFound(channel::Id), + #[error(transparent)] + DatabaseError(#[from] sqlx::Error), +} diff --git a/src/message/event.rs b/src/message/event.rs new file mode 100644 index 0000000..bcc2238 --- /dev/null +++ b/src/message/event.rs @@ -0,0 +1,50 @@ +use super::{snapshot::Message, Id}; +use crate::{ + channel::Channel, + event::{Instant, Sequenced}, +}; + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +pub struct Event { + #[serde(flatten)] + pub instant: Instant, + #[serde(flatten)] + pub kind: Kind, +} + +impl Sequenced for Event { + fn instant(&self) -> Instant { + self.instant + } +} + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum Kind { + Sent(Sent), + Deleted(Deleted), +} + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +pub struct Sent { + #[serde(flatten)] + pub message: Message, +} + +impl From<Sent> for Kind { + fn from(event: Sent) -> Self { + Self::Sent(event) + } +} + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +pub struct Deleted { + pub channel: Channel, + pub message: Id, +} + +impl From<Deleted> for Kind { + fn from(event: Deleted) -> Self { + Self::Deleted(event) + } +} diff --git a/src/message/history.rs b/src/message/history.rs new file mode 100644 index 0000000..5aca47e --- /dev/null +++ b/src/message/history.rs @@ -0,0 +1,43 @@ +use super::{ + event::{Deleted, Event, Sent}, + Message, +}; +use crate::event::Instant; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct History { + pub message: Message, + pub sent: Instant, + pub deleted: Option<Instant>, +} + +impl History { + fn sent(&self) -> Event { + Event { + instant: self.sent, + kind: Sent { + message: self.message.clone(), + } + .into(), + } + } + + fn deleted(&self) -> Option<Event> { + self.deleted.map(|instant| Event { + instant, + kind: Deleted { + channel: self.message.channel.clone(), + message: self.message.id.clone(), + } + .into(), + }) + } + + pub fn events(&self) -> impl Iterator<Item = Event> { + [self.sent()].into_iter().chain(self.deleted()) + } + + pub fn snapshot(&self) -> Message { + self.message.clone() + } +} diff --git a/src/message/mod.rs b/src/message/mod.rs index 9a9bf14..52d56c1 100644 --- a/src/message/mod.rs +++ b/src/message/mod.rs @@ -1,9 +1,8 @@ +pub mod app; +pub mod event; +mod history; mod id; +pub mod repo; +mod snapshot; -pub use self::id::Id; - -#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] -pub struct Message { - pub id: Id, - pub body: String, -} +pub use self::{event::Event, history::History, id::Id, snapshot::Message}; diff --git a/src/message/repo.rs b/src/message/repo.rs new file mode 100644 index 0000000..3b2b8f7 --- /dev/null +++ b/src/message/repo.rs @@ -0,0 +1,214 @@ +use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; + +use super::{snapshot::Message, History, Id}; +use crate::{ + channel::{self, Channel}, + clock::DateTime, + event::{Instant, Sequence}, + login::{self, Login}, +}; + +pub trait Provider { + fn messages(&mut self) -> Messages; +} + +impl<'c> Provider for Transaction<'c, Sqlite> { + fn messages(&mut self) -> Messages { + Messages(self) + } +} + +pub struct Messages<'t>(&'t mut SqliteConnection); + +impl<'c> Messages<'c> { + pub async fn create( + &mut self, + channel: &Channel, + sender: &Login, + sent: &Instant, + body: &str, + ) -> Result<History, sqlx::Error> { + let id = Id::generate(); + + let message = sqlx::query!( + r#" + insert into message + (id, channel, sender, sent_at, sent_sequence, body) + values ($1, $2, $3, $4, $5, $6) + returning + id as "id: Id", + body + "#, + id, + channel.id, + sender.id, + sent.at, + sent.sequence, + body, + ) + .map(|row| History { + message: Message { + channel: channel.clone(), + sender: sender.clone(), + id: row.id, + body: row.body, + }, + sent: *sent, + deleted: None, + }) + .fetch_one(&mut *self.0) + .await?; + + Ok(message) + } + + async fn by_id(&mut self, channel: &Channel, message: &Id) -> Result<History, sqlx::Error> { + let message = sqlx::query!( + r#" + select + channel.id as "channel_id: channel::Id", + channel.name as "channel_name", + sender.id as "sender_id: login::Id", + sender.name as "sender_name", + message.id as "id: Id", + message.body, + sent_at as "sent_at: DateTime", + sent_sequence as "sent_sequence: Sequence" + from message + join channel on message.channel = channel.id + join login as sender on message.sender = sender.id + where message.id = $1 + and message.channel = $2 + "#, + message, + channel.id, + ) + .map(|row| History { + message: Message { + channel: Channel { + id: row.channel_id, + name: row.channel_name, + }, + sender: Login { + id: row.sender_id, + name: row.sender_name, + }, + id: row.id, + body: row.body, + }, + sent: Instant { + at: row.sent_at, + sequence: row.sent_sequence, + }, + deleted: None, + }) + .fetch_one(&mut *self.0) + .await?; + + Ok(message) + } + + pub async fn delete( + &mut self, + channel: &Channel, + message: &Id, + deleted: &Instant, + ) -> Result<History, sqlx::Error> { + let history = self.by_id(channel, message).await?; + + sqlx::query_scalar!( + r#" + delete from message + where + id = $1 + returning 1 as "deleted: i64" + "#, + history.message.id, + ) + .fetch_one(&mut *self.0) + .await?; + + Ok(History { + deleted: Some(*deleted), + ..history + }) + } + + pub async fn expired( + &mut self, + expire_at: &DateTime, + ) -> Result<Vec<(Channel, Id)>, sqlx::Error> { + let messages = sqlx::query!( + r#" + select + channel.id as "channel_id: channel::Id", + channel.name as "channel_name", + message.id as "message: Id" + from message + join channel on message.channel = channel.id + where sent_at < $1 + "#, + expire_at, + ) + .map(|row| { + ( + Channel { + id: row.channel_id, + name: row.channel_name, + }, + row.message, + ) + }) + .fetch_all(&mut *self.0) + .await?; + + Ok(messages) + } + + pub async fn replay( + &mut self, + resume_at: Option<Sequence>, + ) -> Result<Vec<History>, sqlx::Error> { + let messages = sqlx::query!( + r#" + select + channel.id as "channel_id: channel::Id", + channel.name as "channel_name", + sender.id as "sender_id: login::Id", + sender.name as "sender_name", + message.id as "id: Id", + message.body, + sent_at as "sent_at: DateTime", + sent_sequence as "sent_sequence: Sequence" + from message + join channel on message.channel = channel.id + join login as sender on message.sender = sender.id + where coalesce(message.sent_sequence > $1, true) + "#, + resume_at, + ) + .map(|row| History { + message: Message { + channel: Channel { + id: row.channel_id, + name: row.channel_name, + }, + sender: Login { + id: row.sender_id, + name: row.sender_name, + }, + id: row.id, + body: row.body, + }, + sent: Instant { + at: row.sent_at, + sequence: row.sent_sequence, + }, + deleted: None, + }) + .fetch_all(&mut *self.0) + .await?; + + Ok(messages) + } +} diff --git a/src/message/snapshot.rs b/src/message/snapshot.rs new file mode 100644 index 0000000..3adccbe --- /dev/null +++ b/src/message/snapshot.rs @@ -0,0 +1,74 @@ +use super::{ + event::{Event, Kind, Sent}, + Id, +}; +use crate::{channel::Channel, login::Login}; + +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +#[serde(into = "self::serialize::Message")] +pub struct Message { + pub channel: Channel, + pub sender: Login, + pub id: Id, + pub body: String, +} + +mod serialize { + use crate::{channel::Channel, login::Login, message::Id}; + + #[derive(serde::Serialize)] + pub struct Message { + channel: Channel, + sender: Login, + #[allow(clippy::struct_field_names)] + // Deliberately redundant with the module path; this produces a specific serialization. + message: MessageData, + } + + #[derive(serde::Serialize)] + pub struct MessageData { + id: Id, + body: String, + } + + impl From<super::Message> for Message { + fn from(message: super::Message) -> Self { + Self { + channel: message.channel, + sender: message.sender, + message: MessageData { + id: message.id, + body: message.body, + }, + } + } + } +} + +impl Message { + fn apply(state: Option<Self>, event: Event) -> Option<Self> { + match (state, event.kind) { + (None, Kind::Sent(event)) => Some(event.into()), + (Some(message), Kind::Deleted(event)) if message.id == event.message => None, + (state, event) => panic!("invalid message event {event:#?} for state {state:#?}"), + } + } +} + +impl FromIterator<Event> for Option<Message> { + fn from_iter<I: IntoIterator<Item = Event>>(events: I) -> Self { + events.into_iter().fold(None, Message::apply) + } +} + +impl From<&Sent> for Message { + fn from(event: &Sent) -> Self { + event.message.clone() + } +} + +impl From<Sent> for Message { + fn from(event: Sent) -> Self { + event.message + } +} diff --git a/src/test/fixtures/event.rs b/src/test/fixtures/event.rs new file mode 100644 index 0000000..09f0490 --- /dev/null +++ b/src/test/fixtures/event.rs @@ -0,0 +1,11 @@ +use crate::{ + event::{Event, Kind}, + message::Message, +}; + +pub fn message_sent(event: &Event, message: &Message) -> bool { + matches!( + &event.kind, + Kind::MessageSent(event) if message == &event.into() + ) +} diff --git a/src/test/fixtures/filter.rs b/src/test/fixtures/filter.rs index d1939a5..6e62aea 100644 --- a/src/test/fixtures/filter.rs +++ b/src/test/fixtures/filter.rs @@ -1,11 +1,11 @@ use futures::future; -use crate::event::types; +use crate::event::{Event, Kind}; -pub fn messages() -> impl FnMut(&types::ChannelEvent) -> future::Ready<bool> { - |event| future::ready(matches!(event.data, types::ChannelEventData::Message(_))) +pub fn messages() -> impl FnMut(&Event) -> future::Ready<bool> { + |event| future::ready(matches!(event.kind, Kind::MessageSent(_))) } -pub fn created() -> impl FnMut(&types::ChannelEvent) -> future::Ready<bool> { - |event| future::ready(matches!(event.data, types::ChannelEventData::Created(_))) +pub fn created() -> impl FnMut(&Event) -> future::Ready<bool> { + |event| future::ready(matches!(event.kind, Kind::ChannelCreated(_))) } diff --git a/src/test/fixtures/message.rs b/src/test/fixtures/message.rs index fd50887..381b10b 100644 --- a/src/test/fixtures/message.rs +++ b/src/test/fixtures/message.rs @@ -1,17 +1,12 @@ use faker_rand::lorem::Paragraphs; -use crate::{app::App, channel::Channel, clock::RequestedAt, event::types, login::Login}; +use crate::{app::App, channel::Channel, clock::RequestedAt, login::Login, message::Message}; -pub async fn send( - app: &App, - login: &Login, - channel: &Channel, - sent_at: &RequestedAt, -) -> types::ChannelEvent { +pub async fn send(app: &App, channel: &Channel, login: &Login, sent_at: &RequestedAt) -> Message { let body = propose(); - app.events() - .send(login, &channel.id, &body, sent_at) + app.messages() + .send(&channel.id, login, sent_at, &body) .await .expect("should succeed if the channel exists") } diff --git a/src/test/fixtures/mod.rs b/src/test/fixtures/mod.rs index 76467ab..c5efa9b 100644 --- a/src/test/fixtures/mod.rs +++ b/src/test/fixtures/mod.rs @@ -3,6 +3,7 @@ use chrono::{TimeDelta, Utc}; use crate::{app::App, clock::RequestedAt, db}; pub mod channel; +pub mod event; pub mod filter; pub mod future; pub mod identity; diff --git a/src/token/app.rs b/src/token/app.rs index 030ec69..5c4fcd5 100644 --- a/src/token/app.rs +++ b/src/token/app.rs @@ -127,7 +127,7 @@ impl<'a> Tokens<'a> { tx.commit().await?; for event in tokens.into_iter().map(event::TokenRevoked::from) { - self.tokens.broadcast(&event); + self.tokens.broadcast(event); } Ok(()) @@ -139,7 +139,7 @@ impl<'a> Tokens<'a> { tx.commit().await?; self.tokens - .broadcast(&event::TokenRevoked::from(token.clone())); + .broadcast(event::TokenRevoked::from(token.clone())); Ok(()) } |