diff --git a/.gitignore b/.gitignore index 4fc816a..20f3419 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,4 @@ target private.key config.json messages +users.db diff --git a/src/data/messages.rs b/src/data/messages.rs index 8434097..96b81f4 100644 --- a/src/data/messages.rs +++ b/src/data/messages.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use rusqlite::{Connection, params}; +use rusqlite::{Connection, OptionalExtension, params}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::path::PathBuf; @@ -21,6 +21,7 @@ pub struct MessageData { pub struct StoredMessage { pub id: String, pub author: String, + pub is_edited: bool, #[serde(flatten)] pub data: MessageData, } @@ -53,7 +54,8 @@ impl MessageStore { author_pubkey TEXT NOT NULL, content TEXT NOT NULL, timestamp INTEGER NOT NULL, - signature TEXT NOT NULL + signature TEXT NOT NULL, + is_edited INTEGER NOT NULL DEFAULT 0 )", [], )?; @@ -73,20 +75,46 @@ impl MessageStore { pub fn insert_message(&self, channel_id: &str, msg: &StoredMessage) -> Result<()> { self.with_channel(channel_id, |conn| { conn.execute( - "INSERT INTO messages (id, author_pubkey, content, timestamp, signature) - VALUES (?1, ?2, ?3, ?4, ?5)", + "INSERT INTO messages (id, author_pubkey, content, timestamp, signature, is_edited) + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", params![ msg.id, msg.author, msg.data.content, msg.data.timestamp, msg.data.signature, + msg.is_edited, ], )?; Ok(()) }) } + /// Fetches a single message by id, or `None` if it doesn't exist. + pub fn get_message(&self, channel_id: &str, id: &str) -> Result> { + self.with_channel(channel_id, |conn| { + conn.query_row( + "SELECT id, author_pubkey, content, timestamp, signature, is_edited + FROM messages WHERE id = ?1", + params![id], + |row| { + Ok(StoredMessage { + id: row.get(0)?, + author: row.get(1)?, + data: MessageData { + content: row.get(2)?, + timestamp: row.get(3)?, + signature: row.get(4)?, + }, + is_edited: row.get(5)?, + }) + }, + ) + .optional() + .map_err(Into::into) + }) + } + /// Fetches the most recent `limit` messages by the `offset`. pub fn get_recent_messages( &self, @@ -98,7 +126,7 @@ impl MessageStore { let offset = chunk * limit; let mut stmt = conn.prepare( - "SELECT id, author_pubkey, content, timestamp, signature + "SELECT id, author_pubkey, content, timestamp, signature, is_edited FROM messages ORDER BY timestamp DESC LIMIT ?1 OFFSET ?2", @@ -113,12 +141,51 @@ impl MessageStore { timestamp: row.get(3)?, signature: row.get(4)?, }, + is_edited: row.get(5)?, }) })?; let mut messages: Vec = rows.collect::>()?; - messages.reverse(); // DESC query, then flip to oldest-first for display + messages.reverse(); Ok(messages) }) } + + /// Updates a message's content and signature, and marks it as edited. + /// Returns an error if no row with that id exists. + pub fn update_message( + &self, + channel_id: &str, + id: &str, + new_content: &str, + new_signature: &str, + ) -> Result<()> { + self.with_channel(channel_id, |conn| { + let updated = conn.execute( + "UPDATE messages + SET content = ?1, signature = ?2, is_edited = 1 + WHERE id = ?3", + params![new_content, new_signature, id], + )?; + + if updated == 0 { + anyhow::bail!("Message not found"); + } + + Ok(()) + }) + } + + /// Deletes a message. Returns an error if no row with that id existed. + pub fn delete_message(&self, channel_id: &str, id: &str) -> Result<()> { + self.with_channel(channel_id, |conn| { + let deleted = conn.execute("DELETE FROM messages WHERE id = ?1", params![id])?; + + if deleted == 0 { + anyhow::bail!("Message not found"); + } + + Ok(()) + }) + } } diff --git a/src/data/mod.rs b/src/data/mod.rs index ca69f9c..994db7e 100644 --- a/src/data/mod.rs +++ b/src/data/mod.rs @@ -1,2 +1,3 @@ pub mod config; pub mod messages; +pub mod users; diff --git a/src/data/users.rs b/src/data/users.rs new file mode 100644 index 0000000..f9bc526 --- /dev/null +++ b/src/data/users.rs @@ -0,0 +1,62 @@ +use std::{collections::HashMap, path::PathBuf}; + +use rusqlite::{Connection, params}; +use tokio::sync::Mutex; + +use crate::types::ClientMeta; + +pub struct UserMetaStore { + conn: Mutex, +} + +impl UserMetaStore { + pub fn new(path: PathBuf) -> anyhow::Result { + let conn = Connection::open(&path)?; + conn.execute( + "CREATE TABLE IF NOT EXISTS users ( + pubkey TEXT PRIMARY KEY, + display_name TEXT NOT NULL, + avatar TEXT + )", + [], + )?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + pub async fn get_users( + &self, + pubkeys: &[String], + ) -> anyhow::Result> { + let conn = self.conn.lock().await; + let placeholders = pubkeys.iter().map(|_| "?").collect::>().join(","); + let query = format!( + "SELECT pubkey, display_name, avatar FROM users WHERE pubkey IN ({placeholders})" + ); + + let mut stmt = conn.prepare(&query)?; + let rows = stmt.query_map(rusqlite::params_from_iter(pubkeys), |row| { + Ok(( + row.get::<_, String>(0)?, + ClientMeta { + display_name: row.get(1)?, + avatar: row.get(2)?, + }, + )) + })?; + + rows.collect::, _>>() + .map_err(Into::into) + } + + pub async fn upsert_user(&self, pubkey: &str, meta: &ClientMeta) -> anyhow::Result<()> { + let conn = self.conn.lock().await; + conn.execute( + "INSERT INTO users (pubkey, display_name, avatar) VALUES (?1, ?2, ?3) + ON CONFLICT(pubkey) DO UPDATE SET display_name = ?2, avatar = ?3", + params![pubkey, meta.display_name, meta.avatar], + )?; + Ok(()) + } +} diff --git a/src/protocol/message.rs b/src/protocol/message.rs index b19542c..3c4b2c8 100644 --- a/src/protocol/message.rs +++ b/src/protocol/message.rs @@ -40,6 +40,7 @@ pub async fn send_message( let stored = StoredMessage { id: uuid::Uuid::new_v4().to_string(), author: crate::signature::to_string(&verifying_key), + is_edited: false, data: message, }; @@ -77,3 +78,89 @@ pub async fn get_messages( Ok(()) } + +pub async fn edit_message( + server: &Arc, + verifying_key: VerifyingKey, + message_id: String, + channel_id: String, + new_content: String, + new_signature: String, +) -> anyhow::Result<()> { + let existing = server + .message_store + .get_message(&channel_id, &message_id)? + .ok_or_else(|| anyhow::anyhow!("Message not found"))?; + + let author_pubkey = crate::signature::to_string(&verifying_key); + if existing.author != author_pubkey { + anyhow::bail!("Not authorized to edit this message"); + } + + let server_pubkey_string = crate::signature::to_string(&server.key.verifying_key()); + let signed_string = format!( + "{}@{}@{}", + existing.data.timestamp, server_pubkey_string, new_content + ); + + let signature = crate::signature::from_string_sig(&new_signature) + .map_err(|_| anyhow::anyhow!("Invalid signature encoding"))?; + + verifying_key + .verify(signed_string.as_bytes(), &signature) + .map_err(|_| anyhow::anyhow!("Signature verification failed"))?; + + server + .message_store + .update_message(&channel_id, &message_id, &new_content, &new_signature)?; + + let updated = StoredMessage { + id: message_id, + author: author_pubkey, + is_edited: true, + data: MessageData { + content: new_content, + timestamp: existing.data.timestamp, + signature: new_signature, + }, + }; + + server + .broadcast(&ClientMethod::MessageEdited { + channel_id: channel_id.clone(), + message: updated, + }) + .await?; + + Ok(()) +} + +pub async fn delete_message( + server: &Arc, + verifying_key: VerifyingKey, + message_id: String, + channel_id: String, +) -> anyhow::Result<()> { + let existing = server + .message_store + .get_message(&channel_id, &message_id)? + .ok_or_else(|| anyhow::anyhow!("Message not found"))?; + + let author_pubkey = crate::signature::to_string(&verifying_key); + if existing.author != author_pubkey { + anyhow::bail!("Not authorized to delete this message"); + } + + server + .message_store + .delete_message(&channel_id, &message_id)?; + + server + .broadcast(&ClientMethod::MessageDeleted { + channel_id: channel_id.clone(), + message_id, + }) + .await?; + + Ok(()) +} diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index d15c575..d526c29 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -9,6 +9,7 @@ use crate::{data::messages::StoredMessage, server::Server, types::ClientMeta}; pub mod initialize; pub mod message; +pub mod user; #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "method")] @@ -21,12 +22,26 @@ pub enum ClientMethod { hostname: String, }, + Error { + error: Cow<'static, str>, + }, + Messages { messages: HashMap>, }, - Error { - error: Cow<'static, str>, + Users { + users: HashMap, + }, + + MessageEdited { + channel_id: String, + message: StoredMessage, + }, + + MessageDeleted { + channel_id: String, + message_id: String, }, } @@ -41,6 +56,10 @@ pub enum ServerMethod { hostname: String, }, + Error { + error: String, + }, + SendMessage { channel_id: String, data: crate::data::messages::MessageData, @@ -53,8 +72,20 @@ pub enum ServerMethod { Meta(ClientMeta), - Error { - error: String, + GetUsers { + pubkeys: Vec, + }, + + EditMessage { + message_id: String, + channel_id: String, + content: String, + signature: String, + }, + + DeleteMessage { + message_id: String, + channel_id: String, }, } @@ -93,6 +124,34 @@ pub async fn read_loop( ServerMethod::GetMessages { channel_id, chunk } => { message::get_messages(server, verifying_key, socket, channel_id, chunk).await?; } + + ServerMethod::DeleteMessage { + message_id, + channel_id, + } => { + message::delete_message(server, verifying_key, message_id, channel_id).await?; + } + + ServerMethod::EditMessage { + message_id, + channel_id, + content, + signature, + } => { + message::edit_message( + server, + verifying_key, + message_id, + channel_id, + content, + signature, + ) + .await?; + } + + ServerMethod::GetUsers { pubkeys } => { + user::get_users(server, verifying_key, socket, pubkeys).await?; + } } socket_lock = socket.lock().await; diff --git a/src/protocol/user.rs b/src/protocol/user.rs new file mode 100644 index 0000000..8bc0ac1 --- /dev/null +++ b/src/protocol/user.rs @@ -0,0 +1,23 @@ +use std::sync::Arc; + +use axum::extract::ws::WebSocket; +use ed25519_dalek::VerifyingKey; +use tokio::sync::Mutex; + +use crate::{ + protocol::{ClientMethod, send_socket}, + server::Server, +}; + +pub async fn get_users( + server: &Arc, + _verifying_key: VerifyingKey, + socket: &Arc>, + pubkeys: Vec, +) -> anyhow::Result<()> { + let users = server.user_store.get_users(&pubkeys).await?; + + send_socket(&mut *socket.lock().await, &ClientMethod::Users { users }).await?; + + Ok(()) +} diff --git a/src/server.rs b/src/server.rs index 9440f7e..c6ca1f3 100644 --- a/src/server.rs +++ b/src/server.rs @@ -12,7 +12,7 @@ use ed25519_dalek::{SigningKey, VerifyingKey}; use tokio::{sync::Mutex, task::JoinSet}; use crate::{ - data::{config::Config, messages::MessageStore}, + data::{config::Config, messages::MessageStore, users::UserMetaStore}, protocol::{ClientMethod, read_loop, send_socket}, types::ClientMeta, }; @@ -29,6 +29,7 @@ pub struct Server { pub config: Config, pub clients: Mutex>>, pub message_store: MessageStore, + pub user_store: UserMetaStore, } impl Server { @@ -38,6 +39,7 @@ impl Server { config: Config::get().await?, clients: Mutex::new(HashMap::new()), message_store: MessageStore::new(PathBuf::from("messages"))?, + user_store: UserMetaStore::new(PathBuf::from("users.db"))?, })) } } @@ -49,6 +51,14 @@ impl Server { ws.on_upgrade(move |socket: WebSocket| async move { match UserConnections::initialize(&s, socket).await { Ok((client, public_key, meta)) => { + if let Err(e) = s + .user_store + .upsert_user(&crate::signature::to_string(&public_key), &meta) + .await + { + eprintln!("Failed to upsert client: {e}"); + } + let client = Arc::new(Mutex::new(client)); let mut clients_meta = s.clients.lock().await;