framework for pinned messages after refactor of DB

src/handlers.rs

@@ -89,7 +89,7 @@ pub fn get_room(room_id: &str) -> Result<Response, Rejection> {
     // Get the room info if possible
     let raw_query = "SELECT id, name FROM main where id = (?1)";
     let room = match conn.query_row(&raw_query, params![room_id], |row| {
-        Ok(models::Room { id: row.get(0)?, name: row.get(1)?, pinned_message_id: row.get(2)? })
+        Ok(models::Room { id: row.get(0)?, name: row.get(1)? })
     }) {
         Ok(info) => info,
         Err(_) => return Err(warp::reject::custom(Error::NoSuchRoom)),
@@ -112,7 +112,7 @@ pub fn get_all_rooms() -> Result<Response, Rejection> {
     let raw_query = "SELECT id, name FROM main";
     let mut query = conn.prepare(&raw_query).map_err(|_| Error::DatabaseFailedInternally)?;
     let rows = match query
-        .query_map(params![], |row| Ok(models::Room { id: row.get(0)?, name: row.get(1)?, pinned_message_id: row.get(2)? }))
+        .query_map(params![], |row| Ok(models::Room { id: row.get(0)?, name: row.get(1)? }))
     {
         Ok(rows) => rows,
         Err(e) => {
@@ -964,6 +964,9 @@ pub fn get_banned_public_keys(
 pub fn pin_message(
     id: i64, auth_token: &str, pool: &storage::DatabaseConnectionPool,
 ) -> Result<Response, Rejection> {
+    if let Err(err) = delete_pinned_message(auth_token, pool) {
+        return Err(err);
+    };
     // Check authorization level
     let (has_authorization_level, requesting_public_key) =
         has_authorization_level(auth_token, AuthorizationLevel::Basic, pool)?;
@@ -976,9 +979,11 @@ pub fn pin_message(
     // Get a connection and open a transaction
     let mut conn = pool.get().map_err(|_| Error::DatabaseFailedInternally)?;
     let tx = conn.transaction().map_err(|_| Error::DatabaseFailedInternally)?;
+    //TODO sean this needs to map to an actual room_id when its created
+    let room = 1;
     // Pin the message if it's present
-    let stmt = format!("UPDATE main SET pinned_message_id = (?1)");
-    if let Err(err) = tx.execute(&stmt, params![id]) {
+    let stmt = format!("INSERT into pinned_messages (room, message) VALUES (?1, ?2)");
+    if let Err(err) = tx.execute(&stmt, params![room, id]) {
         error!("Couldn't pin message due to error: {}.", err);
         return Err(warp::reject::custom(Error::DatabaseFailedInternally));
     };
@@ -989,7 +994,7 @@ pub fn pin_message(
     return Ok(warp::reply::json(&json).into_response());
 }
 
-pub fn get_pinned_message(auth_token: &str, pool: &storage::DatabaseConnectionPool) -> Result<i64, Rejection> {
+pub fn get_pinned_message(auth_token: &str, pool: &storage::DatabaseConnectionPool) -> Result<Option<i64>, Rejection> {
     // Check authorization level
     let (has_authorization_level, _) = has_authorization_level(auth_token, AuthorizationLevel::Basic, pool)?;
     if !has_authorization_level {
@@ -998,7 +1003,7 @@ pub fn get_pinned_message(auth_token: &str, pool: &storage::DatabaseConnectionPo
     // Get a database connection
     let conn = pool.get().map_err(|_| Error::DatabaseFailedInternally)?;
     // Query the database
-    let raw_query = format!("SELECT pinned_message_id FROM main");
+    let raw_query = format!("SELECT message FROM pinned_messages ORDER BY pinned LIMIT 1");
     let mut query = conn.prepare(&raw_query).map_err(|_| Error::DatabaseFailedInternally)?;
     let rows = match query.query_map(params![], |row| row.get(0)) {
         Ok(rows) => rows,
@@ -1007,10 +1012,9 @@ pub fn get_pinned_message(auth_token: &str, pool: &storage::DatabaseConnectionPo
             return Err(warp::reject::custom(Error::DatabaseFailedInternally));
         }
     };
-    let pinned_message: i64 = rows
+    let pinned_message = rows
         .filter_map(|result| result.ok())
-        .next()
-        .ok_or_else(|| warp::reject::custom(Error::DatabaseFailedInternally))?;
+        .next();
     return Ok(pinned_message);
 }
 
@@ -1027,9 +1031,11 @@ pub fn delete_pinned_message(auth_token: &str, pool: &storage::DatabaseConnectio
     // Get a connection and open a transaction
     let mut conn = pool.get().map_err(|_| Error::DatabaseFailedInternally)?;
     let tx = conn.transaction().map_err(|_| Error::DatabaseFailedInternally)?;
+    //TODO sean this needs to map to an actual room_id when its created
+    let room = 1;
     // Pin the message if it's present
-    let stmt = format!("UPDATE main SET pinned_message_id = (?1)");
-    if let Err(err) = tx.execute(&stmt, [0]) {
+    let stmt = format!("DELETE from pinned_messages WHERE room = (?1)");
+    if let Err(err) = tx.execute(&stmt, [room]) {
         error!("Couldn't delete pinned message due to error: {}.", err);
         return Err(warp::reject::custom(Error::DatabaseFailedInternally));
     };
@@ -1093,9 +1099,9 @@ pub fn compact_poll(
             from_deletion_server_id,
         } = request_body;
         // Check that the room hasn't been deleted
-        let raw_query = "SELECT id, name, pinned_message FROM main where id = (?1)";
+        let raw_query = "SELECT id, name FROM main where id = (?1)";
         match main_conn.query_row(&raw_query, params![room_id], |row| {
-            Ok(models::Room { id: row.get(0)?, name: row.get(1)?, pinned_message_id: row.get(2)? })
+            Ok(models::Room { id: row.get(0)?, name: row.get(1)?})
         }) {
             Ok(_) => (),
             Err(_) => {

src/main.rs

@@ -176,7 +176,7 @@ async fn create_default_rooms() {
     for info in info {
         let id = info.0.to_string();
         let name = info.1.to_string();
-        let room = models::Room { id, name, pinned_message_id: None };
+        let room = models::Room { id, name };
         handlers::create_room(room).await.unwrap();
     }
 }

src/models.rs

@@ -21,12 +21,17 @@ pub struct DeletedMessage {
     pub deleted_message_id: i64,
 }
 
+#[derive(Debug, Deserialize, Serialize)]
+pub struct PinnedMessage {
+    pub room_id: i64,
+    pub message : i64,
+    pub timestamp: i64,
+}
+
 #[derive(Debug, Deserialize, Serialize)]
 pub struct Room {
     pub id: String,
     pub name: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub pinned_message_id: Option<i64>,
 }
 
 #[derive(Debug, Deserialize, Serialize)]

src/rpc.rs

@@ -155,7 +155,8 @@ async fn handle_get_request(
             #[derive(Debug, Deserialize, Serialize)]
             struct Response {
                 status_code: u16,
-                pinned_message: i64,
+                #[serde(skip_serializing_if = "Option::is_none")]
+                pinned_message: Option<i64>,
             }
             let response = Response { status_code: StatusCode::OK.as_u16(), pinned_message: pinned_message_id};
             return Ok(warp::reply::json(&response).into_response());

src/storage.rs

@@ -52,18 +52,28 @@ pub fn create_main_database_if_needed() {
     let pool = &MAIN_POOL;
     let conn = pool.get().unwrap();
     create_main_tables_if_needed(&conn);
+    create_pinned_messages_table_if_needed(&conn);
 }
 
 fn create_main_tables_if_needed(conn: &DatabaseConnection) {
     let main_table_cmd = "CREATE TABLE IF NOT EXISTS main (
         id TEXT PRIMARY KEY,
         name TEXT,
-        pinned_message_id INTEGER,
         image_id TEXT
     )";
     conn.execute(&main_table_cmd, params![]).expect("Couldn't create main table.");
 }
 
+fn create_pinned_messages_table_if_needed(conn: &DatabaseConnection) {
+    let pinned_messages_table_cmd = "CREATE TABLE IF NOT EXISTS pinned_messages (
+        room INTEGER NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
+        message INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
+        pinned FLOAT NOT NULL DEFAULT ((julianday('now') - 2440587.5)*86400.0), /* unix epoch */
+        PRIMARY KEY(room, message)
+    )";
+    conn.execute(&pinned_messages_table_cmd, params![]).expect("Couldn't create pinned_messages table.");
+}
+
 // Rooms
 
 pub const PENDING_TOKEN_EXPIRATION: i64 = 10 * 60;