summaryrefslogtreecommitdiff
path: root/src/message
diff options
context:
space:
mode:
Diffstat (limited to 'src/message')
-rw-r--r--src/message/app.rs88
-rw-r--r--src/message/event.rs50
-rw-r--r--src/message/history.rs43
-rw-r--r--src/message/mod.rs13
-rw-r--r--src/message/repo.rs214
-rw-r--r--src/message/snapshot.rs74
6 files changed, 475 insertions, 7 deletions
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
+ }
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-01-02 14:19:40 +0000