From eff129bc1f29bcb1b2b9d10c6b49ab886edc83d6 Mon Sep 17 00:00:00 2001 From: Owen Jacobson Date: Fri, 27 Sep 2024 18:17:02 -0400 Subject: Make `/api/events` a firehose endpoint. It now includes events for all channels. Clients are responsible for filtering. The schema for channel events has changed; it now includes a channel name and ID, in the same format as the sender's name and ID. They also now include a `"type"` field, whose only valid value (as of this writing) is `"message"`. This is groundwork for delivering message deletion (expiry) events to clients, and notifying clients of channel lifecycle events. --- src/events/repo/broadcast.rs | 162 ------------------------------------------- src/events/repo/message.rs | 145 ++++++++++++++++++++++++++++++++++++++ src/events/repo/mod.rs | 2 +- 3 files changed, 146 insertions(+), 163 deletions(-) delete mode 100644 src/events/repo/broadcast.rs create mode 100644 src/events/repo/message.rs (limited to 'src/events/repo') diff --git a/src/events/repo/broadcast.rs b/src/events/repo/broadcast.rs deleted file mode 100644 index 6914573..0000000 --- a/src/events/repo/broadcast.rs +++ /dev/null @@ -1,162 +0,0 @@ -use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; - -use crate::{ - clock::DateTime, - repo::{ - channel::Channel, - login::{self, Login}, - message, - }, -}; - -pub trait Provider { - fn broadcast(&mut self) -> Broadcast; -} - -impl<'c> Provider for Transaction<'c, Sqlite> { - fn broadcast(&mut self) -> Broadcast { - Broadcast(self) - } -} - -pub struct Broadcast<'t>(&'t mut SqliteConnection); - -#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] -pub struct Message { - pub id: message::Id, - #[serde(skip)] - pub sequence: Sequence, - pub sender: Login, - pub body: String, - pub sent_at: DateTime, -} - -impl<'c> Broadcast<'c> { - pub async fn create( - &mut self, - sender: &Login, - channel: &Channel, - body: &str, - sent_at: &DateTime, - ) -> Result { - let sequence = self.next_sequence_for(channel).await?; - - let id = message::Id::generate(); - - let message = sqlx::query!( - r#" - insert into message - (id, channel, sequence, sender, body, sent_at) - values ($1, $2, $3, $4, $5, $6) - returning - id as "id: message::Id", - sequence as "sequence: Sequence", - sender as "sender: login::Id", - body, - sent_at as "sent_at: DateTime" - "#, - id, - channel.id, - sequence, - sender.id, - body, - sent_at, - ) - .map(|row| Message { - id: row.id, - sequence: row.sequence, - sender: sender.clone(), - body: row.body, - sent_at: row.sent_at, - }) - .fetch_one(&mut *self.0) - .await?; - - Ok(message) - } - - async fn next_sequence_for(&mut self, channel: &Channel) -> Result { - let Sequence(current) = sqlx::query_scalar!( - r#" - -- `max` never returns null, but sqlx can't detect that - select max(sequence) as "sequence!: Sequence" - from message - where channel = $1 - "#, - channel.id, - ) - .fetch_one(&mut *self.0) - .await?; - - Ok(Sequence(current + 1)) - } - - pub async fn expire(&mut self, expire_at: &DateTime) -> Result<(), sqlx::Error> { - sqlx::query!( - r#" - delete from message - where sent_at < $1 - "#, - expire_at, - ) - .execute(&mut *self.0) - .await?; - - Ok(()) - } - - pub async fn replay( - &mut self, - channel: &Channel, - resume_at: Option, - ) -> Result, sqlx::Error> { - let messages = sqlx::query!( - r#" - select - message.id as "id: message::Id", - sequence as "sequence: Sequence", - login.id as "sender_id: login::Id", - login.name as sender_name, - message.body, - message.sent_at as "sent_at: DateTime" - from message - join login on message.sender = login.id - where channel = $1 - and coalesce(sequence > $2, true) - order by sequence asc - "#, - channel.id, - resume_at, - ) - .map(|row| Message { - id: row.id, - sequence: row.sequence, - sender: Login { - id: row.sender_id, - name: row.sender_name, - }, - body: row.body, - sent_at: row.sent_at, - }) - .fetch_all(&mut *self.0) - .await?; - - Ok(messages) - } -} - -#[derive( - Debug, - Eq, - Ord, - PartialEq, - PartialOrd, - Clone, - Copy, - serde::Serialize, - serde::Deserialize, - sqlx::Type, -)] -#[serde(transparent)] -#[sqlx(transparent)] -pub struct Sequence(i64); diff --git a/src/events/repo/message.rs b/src/events/repo/message.rs new file mode 100644 index 0000000..b4724ea --- /dev/null +++ b/src/events/repo/message.rs @@ -0,0 +1,145 @@ +use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; + +use crate::{ + clock::DateTime, + events::types::{self, Sequence}, + repo::{ + channel::Channel, + login::{self, Login}, + 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, + body: &str, + sent_at: &DateTime, + ) -> Result { + let sequence = self.next_sequence_for(channel).await?; + + let id = message::Id::generate(); + + let message = sqlx::query!( + r#" + insert into message + (id, channel, sequence, sender, body, sent_at) + values ($1, $2, $3, $4, $5, $6) + returning + id as "id: message::Id", + sequence as "sequence: Sequence", + sender as "sender: login::Id", + body, + sent_at as "sent_at: DateTime" + "#, + id, + channel.id, + sequence, + sender.id, + body, + sent_at, + ) + .map(|row| types::ChannelEvent { + sequence: row.sequence, + at: row.sent_at, + channel: channel.clone(), + data: types::MessageEvent { + id: row.id, + sender: sender.clone(), + body: row.body, + } + .into(), + }) + .fetch_one(&mut *self.0) + .await?; + + Ok(message) + } + + async fn next_sequence_for(&mut self, channel: &Channel) -> Result { + let current = sqlx::query_scalar!( + r#" + -- `max` never returns null, but sqlx can't detect that + select max(sequence) as "sequence!: Sequence" + from message + where channel = $1 + "#, + channel.id, + ) + .fetch_one(&mut *self.0) + .await?; + + Ok(current.next()) + } + + pub async fn expire(&mut self, expire_at: &DateTime) -> Result<(), sqlx::Error> { + sqlx::query!( + r#" + delete from message + where sent_at < $1 + "#, + expire_at, + ) + .execute(&mut *self.0) + .await?; + + Ok(()) + } + + pub async fn replay( + &mut self, + channel: &Channel, + resume_at: Option, + ) -> Result, sqlx::Error> { + let events = sqlx::query!( + r#" + select + message.id as "id: message::Id", + sequence as "sequence: Sequence", + login.id as "sender_id: login::Id", + login.name as sender_name, + message.body, + message.sent_at as "sent_at: DateTime" + from message + join login on message.sender = login.id + where channel = $1 + and coalesce(sequence > $2, true) + order by sequence asc + "#, + channel.id, + resume_at, + ) + .map(|row| types::ChannelEvent { + sequence: row.sequence, + at: row.sent_at, + channel: channel.clone(), + data: types::MessageEvent { + id: row.id, + sender: login::Login { + id: row.sender_id, + name: row.sender_name, + }, + body: row.body, + } + .into(), + }) + .fetch_all(&mut *self.0) + .await?; + + Ok(events) + } +} diff --git a/src/events/repo/mod.rs b/src/events/repo/mod.rs index 2ed3062..e216a50 100644 --- a/src/events/repo/mod.rs +++ b/src/events/repo/mod.rs @@ -1 +1 @@ -pub mod broadcast; +pub mod message; -- cgit v1.2.3