database work

1 files changed, 318 insertions, 0 deletions

diff --git a/luz/src/db/mod.rs b/luz/src/db/mod.rs
new file mode 100644
index 0000000..f598fbb
--- /dev/null
+++ b/luz/src/db/mod.rs
@@ -0,0 +1,318 @@
+use std::collections::HashSet;
+
+use jid::JID;
+use sqlx::{Error, SqlitePool};
+use uuid::Uuid;
+
+use crate::{
+    chat::{Chat, Message},
+    roster::Contact,
+    user::User,
+};
+
+#[derive(Clone)]
+pub struct Db {
+    db: SqlitePool,
+}
+
+impl Db {
+    pub fn new(db: SqlitePool) -> Self {
+        Self { db }
+    }
+
+    pub async fn create_user(&self, user: User) -> Result<(), Error> {
+        sqlx::query!(
+            "insert into users ( jid, cached_status_message ) values ( ?, ? )",
+            user.jid,
+            user.cached_status_message
+        )
+        .execute(&self.db)
+        .await?;
+        Ok(())
+    }
+
+    pub async fn read_user(&self, user: JID) -> Result<User, Error> {
+        let user: User = sqlx::query_as("select * from users where jid = ?")
+            .bind(user)
+            .fetch_one(&self.db)
+            .await?;
+        Ok(user)
+    }
+
+    pub async fn update_user(&self, user: User) -> Result<(), Error> {
+        sqlx::query!(
+            "update users set cached_status_message = ? where jid = ?",
+            user.cached_status_message,
+            user.jid
+        )
+        .execute(&self.db)
+        .await?;
+        Ok(())
+    }
+
+    // TODO: should this be allowed? messages need to reference users. should probably only allow delete if every other thing referencing it has been deleted, or if you make clear to the user deleting a user will delete all messages associated with them.
+    // pub async fn delete_user(&self, user: JID) -> Result<(), Error> {}
+
+    /// does not create the underlying user, if underlying user does not exist, create_user() must be called separately
+    pub async fn create_contact(&self, contact: Contact) -> Result<(), Error> {
+        sqlx::query!(
+            "insert into roster ( user_jid, name, subscription ) values ( ?, ?, ? )",
+            contact.user_jid,
+            contact.name,
+            contact.subscription
+        )
+        .execute(&self.db)
+        .await?;
+        // TODO: abstract this out in to add_to_group() function ?
+        for group in contact.groups {
+            sqlx::query!(
+                "insert into groups (group_name) values (?) on conflict do nothing",
+                group
+            )
+            .execute(&self.db)
+            .await?;
+            sqlx::query!(
+                "insert into groups_roster (group_name, contact_jid) values (?, ?)",
+                group,
+                contact.user_jid
+            )
+            .execute(&self.db)
+            .await?;
+        }
+        Ok(())
+    }
+
+    pub async fn read_contact(&self, contact: JID) -> Result<Contact, Error> {
+        let mut contact: Contact = sqlx::query_as("select * from roster where user_jid = ?")
+            .bind(contact)
+            .fetch_one(&self.db)
+            .await?;
+        #[derive(sqlx::FromRow)]
+        struct Row {
+            group_name: String,
+        }
+        let groups: Vec<Row> =
+            sqlx::query_as("select group_name from groups_roster where contact_jid = ?")
+                .bind(&contact.user_jid)
+                .fetch_all(&self.db)
+                .await?;
+        contact.groups = HashSet::from_iter(groups.into_iter().map(|row| row.group_name));
+        Ok(contact)
+    }
+
+    pub async fn read_contact_opt(&self, contact: &JID) -> Result<Option<Contact>, Error> {
+        let contact: Option<Contact> = sqlx::query_as(
+            "select * from roster full outer join users on jid = user_jid where jid = ?",
+        )
+        .bind(contact)
+        .fetch_optional(&self.db)
+        .await?;
+        if let Some(mut contact) = contact {
+            #[derive(sqlx::FromRow)]
+            struct Row {
+                group_name: String,
+            }
+            let groups: Vec<Row> =
+                sqlx::query_as("select group_name from groups_roster where contact_jid = ?")
+                    .bind(&contact.user_jid)
+                    .fetch_all(&self.db)
+                    .await?;
+            contact.groups = HashSet::from_iter(groups.into_iter().map(|row| row.group_name));
+            Ok(Some(contact))
+        } else {
+            Ok(None)
+        }
+    }
+
+    /// does not update the underlying user, to update user, update_user() must be called separately
+    pub async fn update_contact(&self, contact: Contact) -> Result<(), Error> {
+        sqlx::query!(
+            "update roster set name = ?, subscription = ? where user_jid = ?",
+            contact.name,
+            contact.subscription,
+            contact.user_jid
+        )
+        .execute(&self.db)
+        .await?;
+        sqlx::query!(
+            "delete from groups_roster where contact_jid = ?",
+            contact.user_jid
+        )
+        .execute(&self.db)
+        .await?;
+        // TODO: delete orphaned groups from groups table
+        for group in contact.groups {
+            sqlx::query!(
+                "insert into groups (group_name) values (?) on conflict do nothing",
+                group
+            )
+            .execute(&self.db)
+            .await?;
+            sqlx::query!(
+                "insert into groups_roster (group_name, contact_jid) values (?, ?)",
+                group,
+                contact.user_jid
+            )
+            .execute(&self.db)
+            .await?;
+        }
+        Ok(())
+    }
+
+    pub async fn delete_contact(&self, contact: JID) -> Result<(), Error> {
+        sqlx::query!("delete from roster where user_jid = ?", contact)
+            .execute(&self.db)
+            .await?;
+        // TODO: delete orphaned groups from groups table
+        Ok(())
+    }
+
+    pub async fn replace_cached_roster(&self, roster: Vec<Contact>) -> Result<(), Error> {
+        sqlx::query!("delete from roster").execute(&self.db).await?;
+        for contact in roster {
+            self.create_contact(contact).await?;
+        }
+        Ok(())
+    }
+
+    pub async fn read_cached_roster(&self) -> Result<Vec<Contact>, Error> {
+        let mut roster: Vec<Contact> =
+            sqlx::query_as("select * from roster full outer join users on jid = user_jid")
+                .fetch_all(&self.db)
+                .await?;
+        for contact in &mut roster {
+            #[derive(sqlx::FromRow)]
+            struct Row {
+                group_name: String,
+            }
+            let groups: Vec<Row> =
+                sqlx::query_as("select group_name from groups_roster where contact_jid = ?")
+                    .bind(&contact.user_jid)
+                    .fetch_all(&self.db)
+                    .await?;
+            contact.groups = HashSet::from_iter(groups.into_iter().map(|row| row.group_name));
+        }
+        Ok(roster)
+    }
+
+    pub async fn create_chat(&self, chat: Chat) -> Result<(), Error> {
+        let id = Uuid::new_v4();
+        let jid = chat.correspondent();
+        sqlx::query!(
+            "insert into chats (id, correspondent) values (?, ?)",
+            id,
+            jid
+        )
+        .execute(&self.db)
+        .await?;
+        Ok(())
+    }
+
+    // TODO: what happens if a correspondent changes from a user to a contact? maybe just have correspondent be a user, then have the client make the user show up as a contact in ui if they are in the loaded roster.
+
+    pub async fn read_chat(&self, chat: JID) -> Result<Chat, Error> {
+        // check if the chat correponding with the jid exists
+        let chat: Chat = sqlx::query_as("select correspondent from chats where correspondent = ?")
+            .bind(chat)
+            .fetch_one(&self.db)
+            .await?;
+        Ok(chat)
+    }
+
+    pub async fn update_chat_correspondent(
+        &self,
+        old_chat: Chat,
+        new_correspondent: JID,
+    ) -> Result<Chat, Error> {
+        // TODO: update other chat data if it differs (for now there is only correspondent so doesn't matter)
+        let new_jid = &new_correspondent;
+        let old_jid = old_chat.correspondent();
+        sqlx::query!(
+            "update chats set correspondent = ? where correspondent = ?",
+            new_jid,
+            old_jid,
+        )
+        .execute(&self.db)
+        .await?;
+        let chat = self.read_chat(new_correspondent).await?;
+        Ok(chat)
+    }
+
+    // pub async fn update_chat
+
+    pub async fn delete_chat(&self, chat: JID) -> Result<(), Error> {
+        sqlx::query!("delete from chats where correspondent = ?", chat)
+            .execute(&self.db)
+            .await?;
+        Ok(())
+    }
+
+    /// TODO: sorting and filtering (for now there is no sorting)
+    pub async fn read_chats(&self) -> Result<Vec<Chat>, Error> {
+        let chats: Vec<Chat> = sqlx::query_as("select * from chats")
+            .fetch_all(&self.db)
+            .await?;
+        Ok(chats)
+    }
+
+    async fn read_chat_id(&self, chat: JID) -> Result<Uuid, Error> {
+        #[derive(sqlx::FromRow)]
+        struct Row {
+            id: Uuid,
+        }
+        let chat_id: Row = sqlx::query_as("select id from chats where correspondent = ?")
+            .bind(chat)
+            .fetch_one(&self.db)
+            .await?;
+        let chat_id = chat_id.id;
+        Ok(chat_id)
+    }
+
+    async fn read_chat_id_opt(&self, chat: JID) -> Result<Option<Uuid>, Error> {
+        #[derive(sqlx::FromRow)]
+        struct Row {
+            id: Uuid,
+        }
+        let chat_id: Option<Row> = sqlx::query_as("select id from chats where correspondent = ?")
+            .bind(chat)
+            .fetch_optional(&self.db)
+            .await?;
+        let chat_id = chat_id.map(|row| row.id);
+        Ok(chat_id)
+    }
+
+    /// if the chat doesn't already exist, it must be created by calling create_chat() before running this function.
+    pub async fn create_message(&self, message: Message, chat: JID) -> Result<(), Error> {
+        // TODO: one query
+        let chat_id = self.read_chat_id(chat).await?;
+        sqlx::query!("insert into messages (id, body, chat_id, from_jid, originally_from) values (?, ?, ?, ?, ?)", message.id, message.body.body, chat_id, message.from, message.from).execute(&self.db).await?;
+        Ok(())
+    }
+
+    pub async fn read_message(&self, message: Uuid) -> Result<Message, Error> {
+        let message: Message = sqlx::query_as("select * from messages where id = ?")
+            .bind(message)
+            .fetch_one(&self.db)
+            .await?;
+        Ok(message)
+    }
+
+    // TODO: message updates/edits pub async fn update_message(&self, message: Message) -> Result<(), Error> {}
+
+    pub async fn delete_message(&self, message: Uuid) -> Result<(), Error> {
+        sqlx::query!("delete from messages where id = ?", message)
+            .execute(&self.db)
+            .await?;
+        Ok(())
+    }
+
+    // TODO: paging
+    pub async fn read_message_history(&self, chat: JID) -> Result<Vec<Message>, Error> {
+        let chat_id = self.read_chat_id(chat).await?;
+        let messages: Vec<Message> = sqlx::query_as("select * from messages where chat_id = ?")
+            .bind(chat_id)
+            .fetch_all(&self.db)
+            .await?;
+        Ok(messages)
+    }
+}