Move to SQLite (#13)

youmubot-db-sql/src/models/mod.rs

@@ -0,0 +1,22 @@
+use crate::*;
+use futures_util::stream::{Stream, StreamExt};
+use sqlx::{query, query_as, Executor};
+
+/// The DateTime used in the package.
+pub type DateTime = chrono::DateTime<chrono::Utc>;
+
+pub mod osu;
+pub mod osu_user;
+
+/// Map a `fetch_many` result to a normal result.
+pub(crate) async fn map_many_result<T, E, W>(
+    item: Result<either::Either<W, T>, E>,
+) -> Option<Result<T>>
+where
+    E: Into<Error>,
+{
+    match item {
+        Ok(v) => v.right().map(Ok),
+        Err(e) => Some(Err(e.into())),
+    }
+}

youmubot-db-sql/src/models/osu.rs

@@ -0,0 +1,319 @@
+use crate::models::*;
+
+pub struct LastBeatmap {
+    pub channel_id: i64,
+    pub beatmap: Vec<u8>,
+    pub mode: u8,
+}
+
+impl LastBeatmap {
+    /// Get a [`LastBeatmap`] by the channel id.
+    pub async fn by_channel_id(
+        id: i64,
+        conn: impl Executor<'_, Database = Database>,
+    ) -> Result<Option<LastBeatmap>> {
+        let m = query_as!(
+            LastBeatmap,
+            r#"SELECT
+                    channel_id as "channel_id: i64",
+                    beatmap,
+                    mode as "mode: u8"
+                FROM osu_last_beatmaps
+                WHERE channel_id = ?"#,
+            id
+        )
+        .fetch_optional(conn)
+        .await?;
+        Ok(m)
+    }
+}
+
+impl LastBeatmap {
+    /// Store the value.
+    pub async fn store(&self, conn: impl Executor<'_, Database = Database>) -> Result<()> {
+        query!(
+            r#"INSERT INTO
+                  osu_last_beatmaps (channel_id, beatmap, mode)
+               VALUES
+                  (?, ?, ?)
+               ON CONFLICT (channel_id) DO UPDATE
+                  SET
+                    beatmap = excluded.beatmap,
+                    mode = excluded.mode"#,
+            self.channel_id,
+            self.beatmap,
+            self.mode,
+        )
+        .execute(conn)
+        .await?;
+        Ok(())
+    }
+}
+
+pub struct UserBestScore {
+    pub beatmap_id: i64,
+    pub mode: u8,
+    pub user_id: i64,
+    pub mods: i64,
+
+    pub cached_at: DateTime,
+    /// To be deserialized by `bincode`
+    pub score: Vec<u8>,
+}
+
+impl UserBestScore {
+    /// Get a list of scores by the given map and user.
+    pub async fn by_map_and_user(
+        beatmap: i64,
+        mode: u8,
+        user: i64,
+        conn: impl Executor<'_, Database = Database>,
+    ) -> Result<Vec<Self>> {
+        query_as!(
+            UserBestScore,
+            r#"SELECT
+                beatmap_id as "beatmap_id: i64",
+                mode as "mode: u8",
+                user_id as "user_id: i64",
+                mods as "mods: i64",
+                cached_at as "cached_at: DateTime",
+                score as "score: Vec<u8>"
+            FROM osu_user_best_scores
+            WHERE
+                beatmap_id = ?
+                AND mode = ?
+                AND user_id = ?"#,
+            beatmap,
+            mode,
+            user
+        )
+        .fetch_all(conn)
+        .await
+        .map_err(Error::from)
+    }
+    /// Get a list of scores by the given map.
+    pub async fn by_map(
+        beatmap: i64,
+        mode: u8,
+        conn: impl Executor<'_, Database = Database>,
+    ) -> Result<Vec<Self>> {
+        query_as!(
+            UserBestScore,
+            r#"SELECT
+                beatmap_id as "beatmap_id: i64",
+                mode as "mode: u8",
+                user_id as "user_id: i64",
+                mods as "mods: i64",
+                cached_at as "cached_at: DateTime",
+                score as "score: Vec<u8>"
+            FROM osu_user_best_scores
+            WHERE
+                beatmap_id = ?
+                AND mode = ?"#,
+            beatmap,
+            mode
+        )
+        .fetch_all(conn)
+        .await
+        .map_err(Error::from)
+    }
+}
+
+impl UserBestScore {
+    pub async fn store(&mut self, conn: impl Executor<'_, Database = Database>) -> Result<()> {
+        self.cached_at = chrono::Utc::now();
+        query!(
+            r#"
+                INSERT INTO
+                    osu_user_best_scores (beatmap_id, mode, user_id, mods, cached_at, score)
+                VALUES
+                    (?, ?, ?, ?, ?, ?)
+                ON CONFLICT (beatmap_id, mode, user_id, mods)
+                DO UPDATE
+                    SET
+                        cached_at = excluded.cached_at,
+                        score = excluded.score
+            "#,
+            self.beatmap_id,
+            self.mode,
+            self.user_id,
+            self.mods,
+            self.cached_at,
+            self.score
+        )
+        .execute(conn)
+        .await?;
+        Ok(())
+    }
+
+    pub async fn clear_user(
+        user_id: i64,
+        conn: impl Executor<'_, Database = Database>,
+    ) -> Result<()> {
+        query!(
+            "DELETE FROM osu_user_best_scores WHERE user_id = ?",
+            user_id
+        )
+        .execute(conn)
+        .await?;
+        Ok(())
+    }
+}
+
+pub struct CachedBeatmap {
+    pub beatmap_id: i64,
+    pub mode: u8,
+    pub cached_at: DateTime,
+    pub beatmap: Vec<u8>,
+}
+
+impl CachedBeatmap {
+    /// Get a cached beatmap by its id.
+    pub async fn by_id(
+        id: i64,
+        mode: u8,
+        conn: impl Executor<'_, Database = Database>,
+    ) -> Result<Option<Self>> {
+        query_as!(
+            Self,
+            r#"SELECT
+                beatmap_id as "beatmap_id: i64",
+                mode as "mode: u8",
+                cached_at as "cached_at: DateTime",
+                beatmap as "beatmap: Vec<u8>"
+            FROM osu_cached_beatmaps
+            WHERE
+                beatmap_id = ?
+                AND mode = ?
+                "#,
+            id,
+            mode
+        )
+        .fetch_optional(conn)
+        .await
+        .map_err(Error::from)
+    }
+
+    pub async fn by_beatmapset(
+        beatmapset: i64,
+        conn: impl Executor<'_, Database = Database>,
+    ) -> Result<Vec<Self>> {
+        query_as!(
+            Self,
+            r#"SELECT
+                beatmap.beatmap_id as "beatmap_id: i64",
+                beatmap.mode as "mode: u8",
+                beatmap.cached_at as "cached_at: DateTime",
+                beatmap.beatmap as "beatmap: Vec<u8>"
+            FROM osu_cached_beatmapsets
+            INNER JOIN osu_cached_beatmaps AS beatmap
+                ON osu_cached_beatmapsets.beatmap_id = beatmap.beatmap_id
+                AND osu_cached_beatmapsets.mode = beatmap.mode
+            WHERE
+                beatmapset_id = ?
+                "#,
+            beatmapset
+        )
+        .fetch_all(conn)
+        .await
+        .map_err(Error::from)
+    }
+}
+
+impl CachedBeatmap {
+    pub async fn store(&mut self, conn: impl Executor<'_, Database = Database>) -> Result<()> {
+        self.cached_at = chrono::Utc::now();
+        query!(
+            r#"
+                INSERT INTO
+                    osu_cached_beatmaps (beatmap_id, mode, cached_at, beatmap)
+                VALUES
+                    (?, ?, ?, ?)
+                ON CONFLICT (beatmap_id, mode)
+                DO UPDATE
+                    SET
+                        cached_at = excluded.cached_at,
+                        beatmap = excluded.beatmap
+            "#,
+            self.beatmap_id,
+            self.mode,
+            self.cached_at,
+            self.beatmap
+        )
+        .execute(conn)
+        .await?;
+        Ok(())
+    }
+
+    pub async fn link_beatmapset(
+        &self,
+        beatmapset_id: i64,
+        conn: impl Executor<'_, Database = Database>,
+    ) -> Result<()> {
+        query!(
+            r#"INSERT INTO osu_cached_beatmapsets(beatmapset_id, beatmap_id, mode)
+                VALUES (?, ?, ?)
+                ON CONFLICT DO NOTHING"#,
+            beatmapset_id,
+            self.beatmap_id,
+            self.mode,
+        )
+        .execute(conn)
+        .await?;
+        Ok(())
+    }
+}
+
+pub struct CachedBeatmapContent {
+    pub beatmap_id: i64,
+    pub cached_at: DateTime,
+    pub content: Vec<u8>,
+}
+
+impl CachedBeatmapContent {
+    /// Get a cached beatmap by its id.
+    pub async fn by_id(
+        id: i64,
+        conn: impl Executor<'_, Database = Database>,
+    ) -> Result<Option<Self>> {
+        query_as!(
+            Self,
+            r#"SELECT
+                beatmap_id as "beatmap_id: i64",
+                cached_at as "cached_at: DateTime",
+                content as "content: Vec<u8>"
+            FROM osu_cached_beatmap_contents
+            WHERE
+                beatmap_id = ? "#,
+            id,
+        )
+        .fetch_optional(conn)
+        .await
+        .map_err(Error::from)
+    }
+}
+
+impl CachedBeatmapContent {
+    pub async fn store(&mut self, conn: impl Executor<'_, Database = Database>) -> Result<()> {
+        self.cached_at = chrono::Utc::now();
+        query!(
+            r#"
+                INSERT INTO
+                    osu_cached_beatmap_contents (beatmap_id, cached_at, content)
+                VALUES
+                    (?, ?, ?)
+                ON CONFLICT (beatmap_id)
+                DO UPDATE
+                    SET
+                        cached_at = excluded.cached_at,
+                        content = excluded.content
+            "#,
+            self.beatmap_id,
+            self.cached_at,
+            self.content
+        )
+        .execute(conn)
+        .await?;
+        Ok(())
+    }
+}

youmubot-db-sql/src/models/osu_user.rs

@@ -0,0 +1,118 @@
+use super::*;
+use sqlx::{query, query_as, Executor};
+
+/// An osu user, as represented in the SQL.
+#[derive(Debug, Clone)]
+pub struct OsuUser {
+    pub user_id: i64,
+    pub id: i64,
+    pub last_update: DateTime,
+    pub pp_std: Option<f32>,
+    pub pp_taiko: Option<f32>,
+    pub pp_mania: Option<f32>,
+    pub pp_catch: Option<f32>,
+    /// Number of consecutive update failures
+    pub failures: u8,
+}
+
+impl OsuUser {
+    /// Query an user by their user id.
+    pub async fn by_user_id<'a, E>(user_id: i64, conn: &'a mut E) -> Result<Option<Self>>
+    where
+        &'a mut E: Executor<'a, Database = Database>,
+    {
+        let u = query_as!(
+            Self,
+            r#"SELECT
+                user_id as "user_id: i64",
+                id as "id: i64",
+                last_update as "last_update: DateTime",
+                pp_std, pp_taiko, pp_mania, pp_catch,
+                failures as "failures: u8"
+            FROM osu_users WHERE user_id = ?"#,
+            user_id
+        )
+        .fetch_optional(conn)
+        .await?;
+        Ok(u)
+    }
+
+    /// Query an user by their osu id.
+    pub async fn by_osu_id<'a, E>(osu_id: i64, conn: &'a mut E) -> Result<Option<Self>>
+    where
+        &'a mut E: Executor<'a, Database = Database>,
+    {
+        let u = query_as!(
+            Self,
+            r#"SELECT
+                user_id as "user_id: i64",
+                id as "id: i64",
+                last_update as "last_update: DateTime",
+                pp_std, pp_taiko, pp_mania, pp_catch,
+                failures as "failures: u8"
+            FROM osu_users WHERE id = ?"#,
+            osu_id
+        )
+        .fetch_optional(conn)
+        .await?;
+        Ok(u)
+    }
+
+    /// Query all users.
+    pub fn all<'a, E>(conn: &'a mut E) -> impl Stream<Item = Result<Self>> + 'a
+    where
+        &'a mut E: Executor<'a, Database = Database>,
+    {
+        query_as!(
+            Self,
+            r#"SELECT
+                user_id as "user_id: i64",
+                id as "id: i64",
+                last_update as "last_update: DateTime",
+                pp_std, pp_taiko, pp_mania, pp_catch,
+                failures as "failures: u8"
+            FROM osu_users"#,
+        )
+        .fetch_many(conn)
+        .filter_map(map_many_result)
+    }
+}
+
+impl OsuUser {
+    /// Stores the user.
+    pub async fn store<'a, E>(&self, conn: &'a mut E) -> Result<()>
+    where
+        &'a mut E: Executor<'a, Database = Database>,
+    {
+        query!(
+            r#"INSERT
+               INTO osu_users(user_id, id, last_update, pp_std, pp_taiko, pp_mania, pp_catch, failures)
+               VALUES(?, ?, ?, ?, ?, ?, ?, ?)
+               ON CONFLICT (user_id) WHERE id = ? DO UPDATE
+               SET
+                last_update = excluded.last_update,
+                pp_std = excluded.pp_std,
+                pp_taiko = excluded.pp_taiko,
+                pp_mania = excluded.pp_mania,
+                pp_catch = excluded.pp_catch,
+                failures = excluded.failures
+            "#,
+            self.user_id,
+            self.id,
+            self.last_update,
+            self.pp_std,
+            self.pp_taiko,
+            self.pp_mania,
+            self.pp_catch,
+            self.failures,
+            self.user_id).execute(conn).await?;
+        Ok(())
+    }
+
+    pub async fn delete(user_id: i64, conn: impl Executor<'_, Database = Database>) -> Result<()> {
+        query!("DELETE FROM osu_users WHERE user_id = ?", user_id)
+            .execute(conn)
+            .await?;
+        Ok(())
+    }
+}