From 6f07e6869bbf62903ac83c9bc061e7bde997e6a8 Mon Sep 17 00:00:00 2001 From: Owen Jacobson Date: Wed, 2 Oct 2024 01:10:09 -0400 Subject: Retire top-level `repo`. This helped me discover an organizational scheme I like more. --- src/channel/app.rs | 6 +- src/channel/mod.rs | 1 + src/channel/repo.rs | 165 +++++++++++++++++++++++++++++++++++++++++++++ src/cli.rs | 4 +- src/db.rs | 42 ++++++++++++ src/event/app.rs | 6 +- src/event/repo/mod.rs | 3 + src/event/repo/sequence.rs | 44 ++++++++++++ src/lib.rs | 2 +- src/login/app.rs | 2 +- src/repo/channel.rs | 165 --------------------------------------------- src/repo/error.rs | 23 ------- src/repo/mod.rs | 4 -- src/repo/pool.rs | 18 ----- src/repo/sequence.rs | 44 ------------ src/test/fixtures/mod.rs | 4 +- src/token/app.rs | 2 +- 17 files changed, 267 insertions(+), 268 deletions(-) create mode 100644 src/channel/repo.rs create mode 100644 src/db.rs create mode 100644 src/event/repo/sequence.rs delete mode 100644 src/repo/channel.rs delete mode 100644 src/repo/error.rs delete mode 100644 src/repo/mod.rs delete mode 100644 src/repo/pool.rs delete mode 100644 src/repo/sequence.rs diff --git a/src/channel/app.rs b/src/channel/app.rs index 1422651..ef0a63f 100644 --- a/src/channel/app.rs +++ b/src/channel/app.rs @@ -2,11 +2,9 @@ use chrono::TimeDelta; use sqlx::sqlite::SqlitePool; use crate::{ - channel::Channel, + channel::{repo::Provider as _, Channel}, clock::DateTime, - event::Sequence, - event::{broadcaster::Broadcaster, types::ChannelEvent}, - repo::{channel::Provider as _, sequence::Provider as _}, + event::{broadcaster::Broadcaster, repo::Provider as _, types::ChannelEvent, Sequence}, }; pub struct Channels<'a> { diff --git a/src/channel/mod.rs b/src/channel/mod.rs index 02d0ed4..2672084 100644 --- a/src/channel/mod.rs +++ b/src/channel/mod.rs @@ -2,6 +2,7 @@ use crate::{clock::DateTime, event::Sequence}; pub mod app; mod id; +pub mod repo; mod routes; pub use self::{id::Id, routes::router}; diff --git a/src/channel/repo.rs b/src/channel/repo.rs new file mode 100644 index 0000000..18cd81f --- /dev/null +++ b/src/channel/repo.rs @@ -0,0 +1,165 @@ +use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; + +use crate::{ + channel::{Channel, Id}, + clock::DateTime, + event::{types, Sequence}, +}; + +pub trait Provider { + fn channels(&mut self) -> Channels; +} + +impl<'c> Provider for Transaction<'c, Sqlite> { + fn channels(&mut self) -> Channels { + Channels(self) + } +} + +pub struct Channels<'t>(&'t mut SqliteConnection); + +impl<'c> Channels<'c> { + pub async fn create( + &mut self, + name: &str, + created_at: &DateTime, + created_sequence: Sequence, + ) -> Result { + let id = Id::generate(); + let channel = sqlx::query_as!( + Channel, + r#" + insert + into channel (id, name, created_at, created_sequence) + values ($1, $2, $3, $4) + returning + id as "id: Id", + name, + created_at as "created_at: DateTime", + created_sequence as "created_sequence: Sequence" + "#, + id, + name, + created_at, + created_sequence, + ) + .fetch_one(&mut *self.0) + .await?; + + Ok(channel) + } + + pub async fn by_id(&mut self, channel: &Id) -> Result { + let channel = sqlx::query_as!( + Channel, + r#" + select + id as "id: Id", + name, + created_at as "created_at: DateTime", + created_sequence as "created_sequence: Sequence" + from channel + where id = $1 + "#, + channel, + ) + .fetch_one(&mut *self.0) + .await?; + + Ok(channel) + } + + pub async fn all( + &mut self, + resume_point: Option, + ) -> Result, sqlx::Error> { + let channels = sqlx::query_as!( + Channel, + r#" + select + id as "id: Id", + name, + created_at as "created_at: DateTime", + created_sequence as "created_sequence: Sequence" + from channel + where coalesce(created_sequence <= $1, true) + order by channel.name + "#, + resume_point, + ) + .fetch_all(&mut *self.0) + .await?; + + Ok(channels) + } + + pub async fn replay( + &mut self, + resume_at: Option, + ) -> Result, sqlx::Error> { + let channels = sqlx::query_as!( + Channel, + r#" + select + id as "id: Id", + name, + created_at as "created_at: DateTime", + created_sequence as "created_sequence: Sequence" + from channel + where coalesce(created_sequence > $1, true) + "#, + resume_at, + ) + .fetch_all(&mut *self.0) + .await?; + + Ok(channels) + } + + pub async fn delete( + &mut self, + channel: &Channel, + deleted_at: &DateTime, + deleted_sequence: Sequence, + ) -> Result { + let channel = channel.id.clone(); + sqlx::query_scalar!( + r#" + delete from channel + where id = $1 + returning 1 as "row: i64" + "#, + channel, + ) + .fetch_one(&mut *self.0) + .await?; + + Ok(types::ChannelEvent { + sequence: deleted_sequence, + at: *deleted_at, + data: types::DeletedEvent { channel }.into(), + }) + } + + pub async fn expired(&mut self, expired_at: &DateTime) -> Result, sqlx::Error> { + let channels = sqlx::query_as!( + Channel, + r#" + select + channel.id as "id: Id", + channel.name, + channel.created_at as "created_at: DateTime", + channel.created_sequence as "created_sequence: Sequence" + from channel + left join message + where created_at < $1 + and message.id is null + "#, + expired_at, + ) + .fetch_all(&mut *self.0) + .await?; + + Ok(channels) + } +} diff --git a/src/cli.rs b/src/cli.rs index ee95ea6..893fae2 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -10,7 +10,7 @@ use clap::Parser; use sqlx::sqlite::SqlitePool; use tokio::net; -use crate::{app::App, channel, clock, event, expire, login, repo::pool}; +use crate::{app::App, channel, clock, db, event, expire, login}; /// Command-line entry point for running the `hi` server. /// @@ -100,7 +100,7 @@ impl Args { } async fn pool(&self) -> sqlx::Result { - pool::prepare(&self.database_url).await + db::prepare(&self.database_url).await } } diff --git a/src/db.rs b/src/db.rs new file mode 100644 index 0000000..93a1169 --- /dev/null +++ b/src/db.rs @@ -0,0 +1,42 @@ +use std::str::FromStr; + +use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions}; + +pub async fn prepare(url: &str) -> sqlx::Result { + let pool = create(url).await?; + sqlx::migrate!().run(&pool).await?; + Ok(pool) +} + +async fn create(database_url: &str) -> sqlx::Result { + let options = SqliteConnectOptions::from_str(database_url)? + .create_if_missing(true) + .optimize_on_close(true, /* analysis_limit */ None); + + let pool = SqlitePoolOptions::new().connect_with(options).await?; + Ok(pool) +} + +pub trait NotFound { + type Ok; + fn not_found(self, map: F) -> Result + where + E: From, + F: FnOnce() -> E; +} + +impl NotFound for Result { + type Ok = T; + + fn not_found(self, map: F) -> Result + where + E: From, + F: FnOnce() -> E, + { + match self { + Err(sqlx::Error::RowNotFound) => Err(map()), + Err(other) => Err(other.into()), + Ok(value) => Ok(value), + } + } +} diff --git a/src/event/app.rs b/src/event/app.rs index b5f2ecc..3d35f1a 100644 --- a/src/event/app.rs +++ b/src/event/app.rs @@ -12,11 +12,11 @@ use super::{ types::{self, ChannelEvent}, }; use crate::{ - channel, + channel::{self, repo::Provider as _}, clock::DateTime, - event::Sequence, + db::NotFound as _, + event::{repo::Provider as _, Sequence}, login::Login, - repo::{channel::Provider as _, error::NotFound as _, sequence::Provider as _}, }; pub struct Events<'a> { diff --git a/src/event/repo/mod.rs b/src/event/repo/mod.rs index e216a50..cee840c 100644 --- a/src/event/repo/mod.rs +++ b/src/event/repo/mod.rs @@ -1 +1,4 @@ pub mod message; +mod sequence; + +pub use self::sequence::Provider; diff --git a/src/event/repo/sequence.rs b/src/event/repo/sequence.rs new file mode 100644 index 0000000..c985869 --- /dev/null +++ b/src/event/repo/sequence.rs @@ -0,0 +1,44 @@ +use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; + +use crate::event::Sequence; + +pub trait Provider { + fn sequence(&mut self) -> Sequences; +} + +impl<'c> Provider for Transaction<'c, Sqlite> { + fn sequence(&mut self) -> Sequences { + Sequences(self) + } +} + +pub struct Sequences<'t>(&'t mut SqliteConnection); + +impl<'c> Sequences<'c> { + pub async fn next(&mut self) -> Result { + let next = sqlx::query_scalar!( + r#" + update event_sequence + set last_value = last_value + 1 + returning last_value as "next_value: Sequence" + "#, + ) + .fetch_one(&mut *self.0) + .await?; + + Ok(next) + } + + pub async fn current(&mut self) -> Result { + let next = sqlx::query_scalar!( + r#" + select last_value as "last_value: Sequence" + from event_sequence + "#, + ) + .fetch_one(&mut *self.0) + .await?; + + Ok(next) + } +} diff --git a/src/lib.rs b/src/lib.rs index bbcb314..8ec13da 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,13 +7,13 @@ mod broadcast; mod channel; pub mod cli; mod clock; +mod db; mod error; mod event; mod expire; mod id; mod login; mod message; -mod repo; #[cfg(test)] mod test; mod token; diff --git a/src/login/app.rs b/src/login/app.rs index 69c1055..15adb31 100644 --- a/src/login/app.rs +++ b/src/login/app.rs @@ -1,6 +1,6 @@ use sqlx::sqlite::SqlitePool; -use crate::{event::Sequence, repo::sequence::Provider as _}; +use crate::event::{repo::Provider as _, Sequence}; #[cfg(test)] use super::{repo::Provider as _, Login, Password}; diff --git a/src/repo/channel.rs b/src/repo/channel.rs deleted file mode 100644 index 18cd81f..0000000 --- a/src/repo/channel.rs +++ /dev/null @@ -1,165 +0,0 @@ -use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; - -use crate::{ - channel::{Channel, Id}, - clock::DateTime, - event::{types, Sequence}, -}; - -pub trait Provider { - fn channels(&mut self) -> Channels; -} - -impl<'c> Provider for Transaction<'c, Sqlite> { - fn channels(&mut self) -> Channels { - Channels(self) - } -} - -pub struct Channels<'t>(&'t mut SqliteConnection); - -impl<'c> Channels<'c> { - pub async fn create( - &mut self, - name: &str, - created_at: &DateTime, - created_sequence: Sequence, - ) -> Result { - let id = Id::generate(); - let channel = sqlx::query_as!( - Channel, - r#" - insert - into channel (id, name, created_at, created_sequence) - values ($1, $2, $3, $4) - returning - id as "id: Id", - name, - created_at as "created_at: DateTime", - created_sequence as "created_sequence: Sequence" - "#, - id, - name, - created_at, - created_sequence, - ) - .fetch_one(&mut *self.0) - .await?; - - Ok(channel) - } - - pub async fn by_id(&mut self, channel: &Id) -> Result { - let channel = sqlx::query_as!( - Channel, - r#" - select - id as "id: Id", - name, - created_at as "created_at: DateTime", - created_sequence as "created_sequence: Sequence" - from channel - where id = $1 - "#, - channel, - ) - .fetch_one(&mut *self.0) - .await?; - - Ok(channel) - } - - pub async fn all( - &mut self, - resume_point: Option, - ) -> Result, sqlx::Error> { - let channels = sqlx::query_as!( - Channel, - r#" - select - id as "id: Id", - name, - created_at as "created_at: DateTime", - created_sequence as "created_sequence: Sequence" - from channel - where coalesce(created_sequence <= $1, true) - order by channel.name - "#, - resume_point, - ) - .fetch_all(&mut *self.0) - .await?; - - Ok(channels) - } - - pub async fn replay( - &mut self, - resume_at: Option, - ) -> Result, sqlx::Error> { - let channels = sqlx::query_as!( - Channel, - r#" - select - id as "id: Id", - name, - created_at as "created_at: DateTime", - created_sequence as "created_sequence: Sequence" - from channel - where coalesce(created_sequence > $1, true) - "#, - resume_at, - ) - .fetch_all(&mut *self.0) - .await?; - - Ok(channels) - } - - pub async fn delete( - &mut self, - channel: &Channel, - deleted_at: &DateTime, - deleted_sequence: Sequence, - ) -> Result { - let channel = channel.id.clone(); - sqlx::query_scalar!( - r#" - delete from channel - where id = $1 - returning 1 as "row: i64" - "#, - channel, - ) - .fetch_one(&mut *self.0) - .await?; - - Ok(types::ChannelEvent { - sequence: deleted_sequence, - at: *deleted_at, - data: types::DeletedEvent { channel }.into(), - }) - } - - pub async fn expired(&mut self, expired_at: &DateTime) -> Result, sqlx::Error> { - let channels = sqlx::query_as!( - Channel, - r#" - select - channel.id as "id: Id", - channel.name, - channel.created_at as "created_at: DateTime", - channel.created_sequence as "created_sequence: Sequence" - from channel - left join message - where created_at < $1 - and message.id is null - "#, - expired_at, - ) - .fetch_all(&mut *self.0) - .await?; - - Ok(channels) - } -} diff --git a/src/repo/error.rs b/src/repo/error.rs deleted file mode 100644 index a5961e2..0000000 --- a/src/repo/error.rs +++ /dev/null @@ -1,23 +0,0 @@ -pub trait NotFound { - type Ok; - fn not_found(self, map: F) -> Result - where - E: From, - F: FnOnce() -> E; -} - -impl NotFound for Result { - type Ok = T; - - fn not_found(self, map: F) -> Result - where - E: From, - F: FnOnce() -> E, - { - match self { - Err(sqlx::Error::RowNotFound) => Err(map()), - Err(other) => Err(other.into()), - Ok(value) => Ok(value), - } - } -} diff --git a/src/repo/mod.rs b/src/repo/mod.rs deleted file mode 100644 index 7abd46b..0000000 --- a/src/repo/mod.rs +++ /dev/null @@ -1,4 +0,0 @@ -pub mod channel; -pub mod error; -pub mod pool; -pub mod sequence; diff --git a/src/repo/pool.rs b/src/repo/pool.rs deleted file mode 100644 index b4aa6fc..0000000 --- a/src/repo/pool.rs +++ /dev/null @@ -1,18 +0,0 @@ -use std::str::FromStr; - -use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions}; - -pub async fn prepare(url: &str) -> sqlx::Result { - let pool = create(url).await?; - sqlx::migrate!().run(&pool).await?; - Ok(pool) -} - -async fn create(database_url: &str) -> sqlx::Result { - let options = SqliteConnectOptions::from_str(database_url)? - .create_if_missing(true) - .optimize_on_close(true, /* analysis_limit */ None); - - let pool = SqlitePoolOptions::new().connect_with(options).await?; - Ok(pool) -} diff --git a/src/repo/sequence.rs b/src/repo/sequence.rs deleted file mode 100644 index c985869..0000000 --- a/src/repo/sequence.rs +++ /dev/null @@ -1,44 +0,0 @@ -use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; - -use crate::event::Sequence; - -pub trait Provider { - fn sequence(&mut self) -> Sequences; -} - -impl<'c> Provider for Transaction<'c, Sqlite> { - fn sequence(&mut self) -> Sequences { - Sequences(self) - } -} - -pub struct Sequences<'t>(&'t mut SqliteConnection); - -impl<'c> Sequences<'c> { - pub async fn next(&mut self) -> Result { - let next = sqlx::query_scalar!( - r#" - update event_sequence - set last_value = last_value + 1 - returning last_value as "next_value: Sequence" - "#, - ) - .fetch_one(&mut *self.0) - .await?; - - Ok(next) - } - - pub async fn current(&mut self) -> Result { - let next = sqlx::query_scalar!( - r#" - select last_value as "last_value: Sequence" - from event_sequence - "#, - ) - .fetch_one(&mut *self.0) - .await?; - - Ok(next) - } -} diff --git a/src/test/fixtures/mod.rs b/src/test/fixtures/mod.rs index d1dd0c3..76467ab 100644 --- a/src/test/fixtures/mod.rs +++ b/src/test/fixtures/mod.rs @@ -1,6 +1,6 @@ use chrono::{TimeDelta, Utc}; -use crate::{app::App, clock::RequestedAt, repo::pool}; +use crate::{app::App, clock::RequestedAt, db}; pub mod channel; pub mod filter; @@ -10,7 +10,7 @@ pub mod login; pub mod message; pub async fn scratch_app() -> App { - let pool = pool::prepare("sqlite::memory:") + let pool = db::prepare("sqlite::memory:") .await .expect("setting up in-memory sqlite database"); App::from(pool) diff --git a/src/token/app.rs b/src/token/app.rs index 1477a9f..030ec69 100644 --- a/src/token/app.rs +++ b/src/token/app.rs @@ -11,8 +11,8 @@ use super::{ }; use crate::{ clock::DateTime, + db::NotFound as _, login::{repo::Provider as _, Login, Password}, - repo::error::NotFound as _, }; pub struct Tokens<'a> { -- cgit v1.2.3