summaryrefslogtreecommitdiff
path: root/src/events
diff options
context:
space:
mode:
Diffstat (limited to 'src/events')
-rw-r--r--src/events/app.rs76
-rw-r--r--src/events/repo/message.rs79
-rw-r--r--src/events/routes.rs26
-rw-r--r--src/events/routes/test.rs47
-rw-r--r--src/events/types.rs79
5 files changed, 94 insertions, 213 deletions
diff --git a/src/events/app.rs b/src/events/app.rs
index db7f430..c15f11e 100644
--- a/src/events/app.rs
+++ b/src/events/app.rs
@@ -1,5 +1,3 @@
-use std::collections::BTreeMap;
-
use chrono::TimeDelta;
use futures::{
future,
@@ -11,7 +9,7 @@ use sqlx::sqlite::SqlitePool;
use super::{
broadcaster::Broadcaster,
repo::message::Provider as _,
- types::{self, ChannelEvent, ResumePoint},
+ types::{self, ChannelEvent},
};
use crate::{
clock::DateTime,
@@ -19,6 +17,7 @@ use crate::{
channel::{self, Provider as _},
error::NotFound as _,
login::Login,
+ sequence::{Provider as _, Sequence},
},
};
@@ -45,9 +44,10 @@ impl<'a> Events<'a> {
.by_id(channel)
.await
.not_found(|| EventsError::ChannelNotFound(channel.clone()))?;
+ let sent_sequence = tx.sequence().next().await?;
let event = tx
.message_events()
- .create(login, &channel, body, sent_at)
+ .create(login, &channel, sent_at, sent_sequence, body)
.await?;
tx.commit().await?;
@@ -64,10 +64,10 @@ impl<'a> Events<'a> {
let mut events = Vec::with_capacity(expired.len());
for (channel, message) in expired {
- let sequence = tx.message_events().assign_sequence(&channel).await?;
+ let deleted_sequence = tx.sequence().next().await?;
let event = tx
.message_events()
- .delete_expired(&channel, &message, sequence, relative_to)
+ .delete(&channel, &message, relative_to, deleted_sequence)
.await?;
events.push(event);
}
@@ -83,42 +83,30 @@ impl<'a> Events<'a> {
pub async fn subscribe(
&self,
- resume_at: ResumePoint,
- ) -> Result<impl Stream<Item = types::ResumableEvent> + std::fmt::Debug, sqlx::Error> {
- let mut tx = self.db.begin().await?;
- let channels = tx.channels().all().await?;
-
- let created_events = {
- let resume_at = resume_at.clone();
- let channels = channels.clone();
- stream::iter(
- channels
- .into_iter()
- .map(ChannelEvent::created)
- .filter(move |event| resume_at.not_after(event)),
- )
- };
-
+ resume_at: Option<Sequence>,
+ ) -> Result<impl Stream<Item = types::ChannelEvent> + 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 replays = BTreeMap::new();
- let mut resume_live_at = resume_at.clone();
- for channel in channels {
- let replay = tx
- .message_events()
- .replay(&channel, resume_at.get(&channel.id))
- .await?;
+ let mut tx = self.db.begin().await?;
+ let channels = tx.channels().replay(resume_at).await?;
- if let Some(last) = replay.last() {
- resume_live_at.advance(last);
- }
+ let channel_events = channels
+ .into_iter()
+ .map(ChannelEvent::created)
+ .filter(move |event| resume_at.map_or(true, |resume_at| event.sequence > resume_at));
- replays.insert(channel.id.clone(), replay);
- }
+ let message_events = tx.message_events().replay(resume_at).await?;
+
+ let mut replay_events = channel_events
+ .into_iter()
+ .chain(message_events.into_iter())
+ .collect::<Vec<_>>();
+ replay_events.sort_by_key(|event| event.sequence);
+ let resume_live_at = replay_events.last().map(|event| event.sequence);
- let replay = stream::select_all(replays.into_values().map(stream::iter));
+ 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.
@@ -132,25 +120,13 @@ impl<'a> Events<'a> {
// stored_messages.
.filter(Self::resume(resume_live_at));
- Ok(created_events.chain(replay).chain(live_messages).scan(
- resume_at,
- |resume_point, event| {
- match event.data {
- types::ChannelEventData::Deleted(_) => resume_point.forget(&event),
- _ => resume_point.advance(&event),
- }
-
- let event = types::ResumableEvent(resume_point.clone(), event);
-
- future::ready(Some(event))
- },
- ))
+ Ok(replay.chain(live_messages))
}
fn resume(
- resume_at: ResumePoint,
+ resume_at: Option<Sequence>,
) -> impl for<'m> FnMut(&'m types::ChannelEvent) -> future::Ready<bool> {
- move |event| future::ready(resume_at.not_after(event))
+ move |event| future::ready(resume_at < Some(event.sequence))
}
}
diff --git a/src/events/repo/message.rs b/src/events/repo/message.rs
index f8bae2b..3237553 100644
--- a/src/events/repo/message.rs
+++ b/src/events/repo/message.rs
@@ -2,11 +2,12 @@ use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction};
use crate::{
clock::DateTime,
- events::types::{self, Sequence},
+ events::types,
repo::{
channel::{self, Channel},
login::{self, Login},
message::{self, Message},
+ sequence::Sequence,
},
};
@@ -27,34 +28,33 @@ impl<'c> Events<'c> {
&mut self,
sender: &Login,
channel: &Channel,
- body: &str,
sent_at: &DateTime,
+ sent_sequence: Sequence,
+ body: &str,
) -> Result<types::ChannelEvent, sqlx::Error> {
- let sequence = self.assign_sequence(channel).await?;
-
let id = message::Id::generate();
let message = sqlx::query!(
r#"
insert into message
- (id, channel, sequence, sender, body, sent_at)
+ (id, channel, sender, sent_at, sent_sequence, body)
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"
+ sent_at as "sent_at: DateTime",
+ sent_sequence as "sent_sequence: Sequence",
+ body
"#,
id,
channel.id,
- sequence,
sender.id,
- body,
sent_at,
+ sent_sequence,
+ body,
)
.map(|row| types::ChannelEvent {
- sequence: row.sequence,
+ sequence: row.sent_sequence,
at: row.sent_at,
data: types::MessageEvent {
channel: channel.clone(),
@@ -72,28 +72,12 @@ impl<'c> Events<'c> {
Ok(message)
}
- pub async fn assign_sequence(&mut self, channel: &Channel) -> Result<Sequence, sqlx::Error> {
- let next = sqlx::query_scalar!(
- r#"
- update channel
- set last_sequence = last_sequence + 1
- where id = $1
- returning last_sequence as "next_sequence: Sequence"
- "#,
- channel.id,
- )
- .fetch_one(&mut *self.0)
- .await?;
-
- Ok(next)
- }
-
- pub async fn delete_expired(
+ pub async fn delete(
&mut self,
channel: &Channel,
message: &message::Id,
- sequence: Sequence,
deleted_at: &DateTime,
+ deleted_sequence: Sequence,
) -> Result<types::ChannelEvent, sqlx::Error> {
sqlx::query_scalar!(
r#"
@@ -107,7 +91,7 @@ impl<'c> Events<'c> {
.await?;
Ok(types::ChannelEvent {
- sequence,
+ sequence: deleted_sequence,
at: *deleted_at,
data: types::MessageDeletedEvent {
channel: channel.clone(),
@@ -127,6 +111,7 @@ impl<'c> Events<'c> {
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
@@ -141,6 +126,7 @@ impl<'c> Events<'c> {
id: row.channel_id,
name: row.channel_name,
created_at: row.channel_created_at,
+ created_sequence: row.channel_created_sequence,
},
row.message,
)
@@ -153,32 +139,39 @@ impl<'c> Events<'c> {
pub async fn replay(
&mut self,
- channel: &Channel,
resume_at: Option<Sequence>,
) -> Result<Vec<types::ChannelEvent>, 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"
+ 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 login on message.sender = login.id
- where channel = $1
- and coalesce(sequence > $2, true)
- order by sequence asc
+ 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
"#,
- channel.id,
resume_at,
)
.map(|row| types::ChannelEvent {
- sequence: row.sequence,
+ sequence: row.sent_sequence,
at: row.sent_at,
data: types::MessageEvent {
- channel: channel.clone(),
+ channel: Channel {
+ id: row.channel_id,
+ name: row.channel_name,
+ created_at: row.channel_created_at,
+ created_sequence: row.channel_created_sequence,
+ },
sender: login::Login {
id: row.sender_id,
name: row.sender_name,
diff --git a/src/events/routes.rs b/src/events/routes.rs
index f09474c..e3a959f 100644
--- a/src/events/routes.rs
+++ b/src/events/routes.rs
@@ -9,14 +9,12 @@ use axum::{
};
use futures::stream::{Stream, StreamExt as _};
-use super::{
- extract::LastEventId,
- types::{self, ResumePoint},
-};
+use super::{extract::LastEventId, types};
use crate::{
app::App,
error::{Internal, Unauthorized},
login::{app::ValidateError, extract::Identity},
+ repo::sequence::Sequence,
};
#[cfg(test)]
@@ -29,11 +27,9 @@ pub fn router() -> Router<App> {
async fn events(
State(app): State<App>,
identity: Identity,
- last_event_id: Option<LastEventId<ResumePoint>>,
-) -> Result<Events<impl Stream<Item = types::ResumableEvent> + std::fmt::Debug>, EventsError> {
- let resume_at = last_event_id
- .map(LastEventId::into_inner)
- .unwrap_or_default();
+ last_event_id: Option<LastEventId<Sequence>>,
+) -> Result<Events<impl Stream<Item = types::ChannelEvent> + std::fmt::Debug>, EventsError> {
+ let resume_at = last_event_id.map(LastEventId::into_inner);
let stream = app.events().subscribe(resume_at).await?;
let stream = app.logins().limit_stream(identity.token, stream).await?;
@@ -46,7 +42,7 @@ struct Events<S>(S);
impl<S> IntoResponse for Events<S>
where
- S: Stream<Item = types::ResumableEvent> + Send + 'static,
+ S: Stream<Item = types::ChannelEvent> + Send + 'static,
{
fn into_response(self) -> Response {
let Self(stream) = self;
@@ -57,14 +53,12 @@ where
}
}
-impl TryFrom<types::ResumableEvent> for sse::Event {
+impl TryFrom<types::ChannelEvent> for sse::Event {
type Error = serde_json::Error;
- fn try_from(value: types::ResumableEvent) -> Result<Self, Self::Error> {
- let types::ResumableEvent(resume_at, data) = value;
-
- let id = serde_json::to_string(&resume_at)?;
- let data = serde_json::to_string_pretty(&data)?;
+ fn try_from(event: types::ChannelEvent) -> 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/events/routes/test.rs b/src/events/routes/test.rs
index 820192d..1cfca4f 100644
--- a/src/events/routes/test.rs
+++ b/src/events/routes/test.rs
@@ -5,7 +5,7 @@ use futures::{
};
use crate::{
- events::{routes, types},
+ events::routes,
test::fixtures::{self, future::Immediately as _},
};
@@ -28,7 +28,7 @@ async fn includes_historical_message() {
// Verify the structure of the response.
- let types::ResumableEvent(_, event) = events
+ let event = events
.filter(fixtures::filter::messages())
.next()
.immediately()
@@ -58,7 +58,7 @@ async fn includes_live_message() {
let sender = fixtures::login::create(&app).await;
let message = fixtures::message::send(&app, &sender, &channel, &fixtures::now()).await;
- let types::ResumableEvent(_, event) = events
+ let event = events
.filter(fixtures::filter::messages())
.next()
.immediately()
@@ -108,9 +108,7 @@ async fn includes_multiple_channels() {
.await;
for message in &messages {
- assert!(events
- .iter()
- .any(|types::ResumableEvent(_, event)| { event == message }));
+ assert!(events.iter().any(|event| { event == message }));
}
}
@@ -138,12 +136,11 @@ async fn sequential_messages() {
// Verify the structure of the response.
- let mut events =
- events.filter(|types::ResumableEvent(_, event)| future::ready(messages.contains(event)));
+ let mut events = events.filter(|event| future::ready(messages.contains(event)));
// Verify delivery in order
for message in &messages {
- let types::ResumableEvent(_, event) = events
+ let event = events
.next()
.immediately()
.await
@@ -179,7 +176,7 @@ async fn resumes_from() {
.await
.expect("subscribe never fails");
- let types::ResumableEvent(last_event_id, event) = events
+ let event = events
.filter(fixtures::filter::messages())
.next()
.immediately()
@@ -188,7 +185,7 @@ async fn resumes_from() {
assert_eq!(initial_message, event);
- last_event_id
+ event.sequence
};
// Resume after disconnect
@@ -205,9 +202,7 @@ async fn resumes_from() {
.await;
for message in &later_messages {
- assert!(events
- .iter()
- .any(|types::ResumableEvent(_, event)| event == message));
+ assert!(events.iter().any(|event| event == message));
}
}
@@ -259,14 +254,12 @@ async fn serial_resume() {
.await;
for message in &initial_messages {
- assert!(events
- .iter()
- .any(|types::ResumableEvent(_, event)| event == message));
+ assert!(events.iter().any(|event| event == message));
}
- let types::ResumableEvent(id, _) = events.last().expect("this vec is non-empty");
+ let event = events.last().expect("this vec is non-empty");
- id.to_owned()
+ event.sequence
};
// Resume after disconnect
@@ -296,14 +289,12 @@ async fn serial_resume() {
.await;
for message in &resume_messages {
- assert!(events
- .iter()
- .any(|types::ResumableEvent(_, event)| event == message));
+ assert!(events.iter().any(|event| event == message));
}
- let types::ResumableEvent(id, _) = events.last().expect("this vec is non-empty");
+ let event = events.last().expect("this vec is non-empty");
- id.to_owned()
+ event.sequence
};
// Resume after disconnect a second time
@@ -335,9 +326,7 @@ 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(|types::ResumableEvent(_, event)| event == message));
+ assert!(events.iter().any(|event| event == message));
}
};
}
@@ -375,7 +364,7 @@ async fn terminates_on_token_expiry() {
];
assert!(events
- .filter(|types::ResumableEvent(_, event)| future::ready(messages.contains(event)))
+ .filter(|event| future::ready(messages.contains(event)))
.next()
.immediately()
.await
@@ -417,7 +406,7 @@ async fn terminates_on_logout() {
];
assert!(events
- .filter(|types::ResumableEvent(_, event)| future::ready(messages.contains(event)))
+ .filter(|event| future::ready(messages.contains(event)))
.next()
.immediately()
.await
diff --git a/src/events/types.rs b/src/events/types.rs
index d954512..aca3af4 100644
--- a/src/events/types.rs
+++ b/src/events/types.rs
@@ -1,84 +1,13 @@
-use std::collections::BTreeMap;
-
use crate::{
clock::DateTime,
repo::{
channel::{self, Channel},
login::Login,
message,
+ sequence::Sequence,
},
};
-#[derive(
- Debug,
- Default,
- Eq,
- Ord,
- PartialEq,
- PartialOrd,
- Clone,
- Copy,
- serde::Serialize,
- serde::Deserialize,
- sqlx::Type,
-)]
-#[serde(transparent)]
-#[sqlx(transparent)]
-pub struct Sequence(i64);
-
-impl Sequence {
- pub fn next(self) -> Self {
- let Self(current) = self;
- Self(current + 1)
- }
-}
-
-// For the purposes of event replay, a resume point is a vector of resume
-// elements. A resume element associates a channel (by ID) with the latest event
-// seen in that channel so far. Replaying the event stream can restart at a
-// predictable point - hence the name. These values can be serialized and sent
-// to the client as JSON dicts, then rehydrated to recover the resume point at a
-// later time.
-//
-// Using a sorted map ensures that there is a canonical representation for
-// each resume point.
-#[derive(Clone, Debug, Default, PartialEq, PartialOrd, serde::Deserialize, serde::Serialize)]
-#[serde(transparent)]
-pub struct ResumePoint(BTreeMap<channel::Id, Sequence>);
-
-impl ResumePoint {
- pub fn advance<'e>(&mut self, event: impl Into<ResumeElement<'e>>) {
- let Self(elements) = self;
- let ResumeElement(channel, sequence) = event.into();
- elements.insert(channel.clone(), sequence);
- }
-
- pub fn forget<'e>(&mut self, event: impl Into<ResumeElement<'e>>) {
- let Self(elements) = self;
- let ResumeElement(channel, _) = event.into();
- elements.remove(channel);
- }
-
- pub fn get(&self, channel: &channel::Id) -> Option<Sequence> {
- let Self(elements) = self;
- elements.get(channel).copied()
- }
-
- pub fn not_after<'e>(&self, event: impl Into<ResumeElement<'e>>) -> bool {
- let Self(elements) = self;
- let ResumeElement(channel, sequence) = event.into();
-
- elements
- .get(channel)
- .map_or(true, |resume_at| resume_at < &sequence)
- }
-}
-
-pub struct ResumeElement<'i>(&'i channel::Id, Sequence);
-
-#[derive(Clone, Debug)]
-pub struct ResumableEvent(pub ResumePoint, pub ChannelEvent);
-
#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
pub struct ChannelEvent {
#[serde(skip)]
@@ -92,7 +21,7 @@ impl ChannelEvent {
pub fn created(channel: Channel) -> Self {
Self {
at: channel.created_at,
- sequence: Sequence::default(),
+ sequence: channel.created_sequence,
data: CreatedEvent { channel }.into(),
}
}
@@ -107,9 +36,9 @@ impl ChannelEvent {
}
}
-impl<'c> From<&'c ChannelEvent> for ResumeElement<'c> {
+impl<'c> From<&'c ChannelEvent> for Sequence {
fn from(event: &'c ChannelEvent) -> Self {
- Self(event.channel_id(), event.sequence)
+ event.sequence
}
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-01-02 18:03:19 +0000