Merge pull request #5 from recurse-chat/finalized-messages-mvp

Finalized messages mvp

.gitignore

@@ -28,3 +28,4 @@ target
 private.key
 config.json
 messages
+users.db

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)
+                "INSERT INTO messages (id, author_pubkey, content, timestamp, signature, is_edited)
-                 VALUES (?1, ?2, ?3, ?4, ?5)",
+                 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<Option<StoredMessage>> {
+        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<StoredMessage> = rows.collect::<Result<_, _>>()?;
-            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(())
+        })
+    }
 }

src/data/mod.rs

@@ -1,2 +1,3 @@
 pub mod config;
 pub mod messages;
+pub mod users;

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<Connection>,
+}
+
+impl UserMetaStore {
+    pub fn new(path: PathBuf) -> anyhow::Result<Self> {
+        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<HashMap<String, ClientMeta>> {
+        let conn = self.conn.lock().await;
+        let placeholders = pubkeys.iter().map(|_| "?").collect::<Vec<_>>().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::<Result<HashMap<_, _>, _>>()
+            .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(())
+    }
+}

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<Server>,
+    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<Server>,
+    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(())
+}

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<String, Vec<StoredMessage>>,
     },
 
-    Error {
+    Users {
-        error: Cow<'static, str>,
+        users: HashMap<String, ClientMeta>,
+    },
+
+    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 {
+    GetUsers {
-        error: String,
+        pubkeys: Vec<String>,
+    },
+
+    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;

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<Server>,
+    _verifying_key: VerifyingKey,
+    socket: &Arc<Mutex<WebSocket>>,
+    pubkeys: Vec<String>,
+) -> anyhow::Result<()> {
+    let users = server.user_store.get_users(&pubkeys).await?;
+
+    send_socket(&mut *socket.lock().await, &ClientMethod::Users { users }).await?;
+
+    Ok(())
+}

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<HashMap<VerifyingKey, Arc<UserConnections>>>,
     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;