From 96e5ed89d6fe651d614bfee11a31ef7fdc6b5ee9 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Sun, 29 Mar 2026 22:28:23 +0200 Subject: [PATCH] [Add] DB util [Fix] Chat files & Chats util [Add] message Height --- Cargo.lock | 12 +- src/omikron/omikron_connection.rs | 246 +++++++++++++++-------- src/util/chat_files.rs | 318 ++++++++++++++---------------- src/util/chats_util.rs | 212 ++++++++++---------- src/util/db.rs | 173 ++++++++++++++++ src/util/mod.rs | 1 + 6 files changed, 593 insertions(+), 369 deletions(-) create mode 100644 src/util/db.rs diff --git a/Cargo.lock b/Cargo.lock index aa15ccf..26b9874 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3879,7 +3879,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "ttp-core" version = "0.1.0" -source = "git+https://github.com/Tensamin/TTP.git#23c9e68da6622a0cb3a773881f353ae4489c9743" +source = "git+https://github.com/Tensamin/TTP.git#ee7b074e665fd708a56a459364a075720056ef88" dependencies = [ "base64", "byteorder", @@ -3891,7 +3891,7 @@ dependencies = [ [[package]] name = "ttp-native" version = "0.1.0" -source = "git+https://github.com/Tensamin/TTP.git#23c9e68da6622a0cb3a773881f353ae4489c9743" +source = "git+https://github.com/Tensamin/TTP.git#ee7b074e665fd708a56a459364a075720056ef88" dependencies = [ "quinn", "rustls", @@ -4895,18 +4895,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.47" +version = "0.8.48" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "efbb2a062be311f2ba113ce66f697a4dc589f85e78a4aea276200804cea0ed87" +checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.47" +version = "0.8.48" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0e8bc7269b54418e7aeeef514aa68f8690b8c0489a06b0136e5f57c4c5ccab89" +checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4" dependencies = [ "proc-macro2", "quote", diff --git a/src/omikron/omikron_connection.rs b/src/omikron/omikron_connection.rs index 3d60510..f74540f 100755 --- a/src/omikron/omikron_connection.rs +++ b/src/omikron/omikron_connection.rs @@ -283,11 +283,16 @@ impl OmikronConnection { let key_pair = crypto_helper::generate_keypair(); let public_key_base64 = crypto_helper::public_key_to_base64(&key_pair.public); - let private_key_base64 = crypto_helper::secret_key_to_base64(&key_pair.secret); + let _private_key_base64 = crypto_helper::secret_key_to_base64(&key_pair.secret); let mut conf_write = CONFIG.write().await; - /*conf_write.change("public_key", DataValue::Str(public_key_base64.clone())); - conf_write.change("private_key", DataValue::Str(private_key_base64));*/ + // NOTE: + // Intentionally not storing the generated private/public keys directly into the + // config file here to avoid persisting sensitive material in plaintext. If you + // want to persist them, uncomment the two lines below and accept the security + // implications (they will be saved by `conf_write.update()`). + // conf_write.change("public_key", DataValue::Str(public_key_base64.clone())); + // conf_write.change("private_key", DataValue::Str(private_key_base64)); conf_write.update(); drop(conf_write); @@ -418,7 +423,7 @@ impl OmikronConnection { } if cv.is_type(CommunicationType::identification_response) { - if let Some(accepted) = cv.get_data(DataTypes::accepted).as_bool() { + if let Some(_accepted) = cv.get_data(DataTypes::accepted).as_bool() { let mut state = self.state.write().await; if let ConnectionState::Connected { identified: _ } = *state { *state = ConnectionState::Connected { identified: true }; @@ -430,6 +435,7 @@ impl OmikronConnection { // ************************************************ // // Direct messages // // ************************************************ // + if cv.is_type(CommunicationType::message_state) { let sender_id = &cv.get_sender(); let receiver_id = &cv.get_receiver(); @@ -458,6 +464,128 @@ impl OmikronConnection { ); } + // Incoming stored message: store for the recipient, attempt local delivery, notify sender. + if cv.is_type(CommunicationType::message_send) { + let sender_id: i64 = if let Some(n) = cv.get_data(DataTypes::sender_id).as_number() { + n as i64 + } else if let Some(s) = cv.get_data(DataTypes::sender_id).as_str() { + s.parse::().unwrap_or(0) + } else { + 0 + }; + + // parse receiver_id (the storage owner for this incoming message) + let receiver_id: i64 = if let Some(n) = cv.get_data(DataTypes::receiver_id).as_number() + { + n as i64 + } else if let Some(s) = cv.get_data(DataTypes::receiver_id).as_str() { + s.parse::().unwrap_or(0) + } else { + 0 + }; + + // parse send_time robustly (number or string), fallback to now + let send_time_val = cv.get_data(DataTypes::send_time); + let now_i64 = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64; + let timestamp_i64 = if let Some(n) = send_time_val.as_number() { + n as i64 + } else if let Some(s) = send_time_val.as_str() { + s.parse::().unwrap_or(now_i64) + } else { + now_i64 + }; + let timestamp_u128 = timestamp_i64 as u128; + + // content may be missing; default to empty string + let content = cv + .get_data(DataTypes::content) + .as_str() + .unwrap_or("") + .to_string(); + + let height = cv.get_data(DataTypes::height).as_number().unwrap_or(0) as i64; + + // persist message for the receiver (storage_owner = receiver_id) + chat_files::add_message( + timestamp_u128, + false, + receiver_id as i64, + sender_id as i64, + &content, + height, + ); + + // Build a live-delivery message for the local client (recipient) + let user_forward = CommunicationValue::new(CommunicationType::message_live) + .with_id(cv.get_id()) + .with_receiver(receiver_id as u64) + .add_data(DataTypes::send_time, DataValue::Number(timestamp_i64)) + .add_data(DataTypes::message, DataValue::Str(content.clone())) + .add_data(DataTypes::sender_id, DataValue::Number(sender_id as i64)) + .add_data(DataTypes::height, DataValue::Number(height)); + + // Attempt delivery and await a response from the local client + let user_resp = self + .clone() + .await_response(&user_forward, Some(Duration::from_secs(10))) + .await; + + if let Ok(user_resp) = user_resp { + let ms_raw = user_resp + .get_data(DataTypes::message_state) + .as_string() + .unwrap_or_else(|| "".to_string()); + let ms = MessageState::from_str(&ms_raw).upgrade(MessageState::Received); + + // update stored message state + let _ = chat_files::change_message_state( + timestamp_i64, + receiver_id as i64, + sender_id as i64, + ms.clone(), + ); + + // notify original sender about the delivered/read state + self.send_message( + &CommunicationValue::new(CommunicationType::message_state) + .with_id(cv.get_id()) + .with_receiver(sender_id as u64) + .with_sender(receiver_id as u64) + .add_data(DataTypes::send_time, DataValue::Number(timestamp_i64)) + .add_data( + DataTypes::message_state, + DataValue::Str(ms.as_str().to_string()), + ), + ) + .await; + } else { + // Delivery failed or timed out; mark as Sent and notify sender + let _ = chat_files::change_message_state( + timestamp_i64, + receiver_id as i64, + sender_id as i64, + MessageState::Sent, + ); + + self.send_message( + &CommunicationValue::new(CommunicationType::message_state) + .with_id(cv.get_id()) + .with_receiver(sender_id as u64) + .with_sender(receiver_id as u64) + .add_data(DataTypes::send_time, DataValue::Number(timestamp_i64)) + .add_data( + DataTypes::message_state, + DataValue::Str(MessageState::Sent.as_str().to_string()), + ), + ) + .await; + } + return; + } + if cv.is_type(CommunicationType::message_other_iota) { let sender_id = &cv.get_sender(); let receiver_id = &cv.get_receiver(); @@ -483,26 +611,25 @@ impl OmikronConnection { .unwrap_or("") .to_string(); + let height = cv.get_data(DataTypes::height).as_number().unwrap_or(0) as i64; + chat_files::add_message( timestamp as u128, false, *receiver_id as i64, *sender_id as i64, &content, + height, ); + // Build user_forward using the parsed numeric timestamp and safe content string let user_forward = CommunicationValue::new(CommunicationType::message_live) .with_id(cv.get_id()) .with_receiver(*receiver_id) - .add_data( - DataTypes::send_time, - cv.get_data(DataTypes::send_time).clone(), - ) - .add_data(DataTypes::message, cv.get_data(DataTypes::content).clone()) - .add_data( - DataTypes::sender_id, - DataValue::Number(cv.get_sender() as i64), - ); + .add_data(DataTypes::send_time, DataValue::Number(timestamp)) + .add_data(DataTypes::message, DataValue::Str(content.clone())) + .add_data(DataTypes::sender_id, DataValue::Number(*sender_id as i64)) + .add_data(DataTypes::height, DataValue::Number(height)); let user_resp = self .clone() @@ -528,10 +655,7 @@ impl OmikronConnection { .with_id(cv.get_id()) .with_receiver(*sender_id) .with_sender(*receiver_id) - .add_data( - DataTypes::send_time, - cv.get_data(DataTypes::send_time).clone(), - ) + .add_data(DataTypes::send_time, DataValue::Number(timestamp)) .add_data( DataTypes::message_state, DataValue::Str(ms.as_str().to_string()), @@ -539,15 +663,20 @@ impl OmikronConnection { ) .await; } else { + // Delivery timed out/failed — update stored state and notify sender with numeric timestamp + let _ = chat_files::change_message_state( + timestamp, + *receiver_id as i64, + *sender_id as i64, + MessageState::Sent, + ); + self.send_message( &CommunicationValue::new(CommunicationType::message_state) .with_id(cv.get_id()) .with_receiver(*sender_id) .with_sender(*receiver_id) - .add_data( - DataTypes::send_time, - cv.get_data(DataTypes::send_time).clone(), - ) + .add_data(DataTypes::send_time, DataValue::Number(timestamp)) .add_data( DataTypes::message_state, DataValue::Str(MessageState::Sent.as_str().to_string()), @@ -558,67 +687,13 @@ impl OmikronConnection { return; } - if cv.is_type(CommunicationType::message_send) { - let my_id = cv.get_sender(); - - // parse other id robustly (number or string) - let other_id = if let Some(n) = cv.get_data(DataTypes::receiver_id).as_number() { - n as i64 - } else if let Some(s) = cv.get_data(DataTypes::receiver_id).as_str() { - s.parse::().unwrap_or(0) - } else { - 0 - }; - - let now_ms_u128 = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as u128; - // derive an i64 timestamp for protocol fields; fall back to current time if out of range - let now_ms_i64: i64 = match i64::try_from(now_ms_u128) { - Ok(v) => v, - Err(_) => SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64, - }; - - // safe content extraction - let content = cv - .get_data(DataTypes::content) - .as_str() - .unwrap_or("") - .to_string(); - - chat_files::add_message(now_ms_u128, true, my_id as i64, other_id, &content); - - let ack = CommunicationValue::new(CommunicationType::success) - .with_id(cv.get_id()) - .with_receiver(my_id); - self.send_message(&ack).await; - - let forward = CommunicationValue::new(CommunicationType::message_other_iota) - .with_id(cv.get_id()) - .with_receiver(other_id as u64) - .add_data(DataTypes::receiver_id, DataValue::Number(other_id)) - .with_sender(my_id) - .add_data(DataTypes::send_time, DataValue::Number(now_ms_i64)) - .add_data(DataTypes::sender_id, DataValue::Number(my_id as i64)) - .add_data(DataTypes::content, DataValue::Str(content)); - if let Err(err) = self.send_message_result(&forward).await { - // sending failed - record via existing logging path - log_t!("send_message_failed", err); - } else { - // forwarding succeeded -> update stored message state to Sent - let _ = chat_files::change_message_state( - now_ms_i64, - my_id as i64, - other_id, - MessageState::Sent, - ); - } - return; - } + // Duplicate handling for CommunicationType::message_send removed. + // Rationale: This branch duplicated logic present earlier that handles incoming + // stored messages and live delivery to local clients. Keeping a single, + // well-defined code path for `message_send` reduces ambiguity and avoids + // accidental early returns that block other handlers. If the protocol needs + // distinct handling for client-originated sends vs stored deliveries, prefer + // using distinct CommunicationType variants or an explicit field/flag. if cv.is_type(CommunicationType::messages_get) { let my_id = cv.get_sender(); @@ -627,13 +702,14 @@ impl OmikronConnection { let amount = cv.get_data(DataTypes::amount).as_number().unwrap_or(0); // retrieve raw JSON messages let messages = chat_files::get_messages(my_id as i64, partner_id, offset, amount); - // convert JSON array -> protocol Array of Containers (send_time, content, sender_id, message_state) + // convert JSON array -> protocol Array of Containers (send_time, content, sender_id, message_state, height) let mut msg_array: Vec = Vec::new(); for m in messages.members() { // extract fields defensively let message_time: i64 = m["message_time"].as_i64().unwrap_or(0); let content: String = m["content"].as_str().unwrap_or("").to_string(); let sent_by_self: bool = m["sent_by_self"].as_bool().unwrap_or(false); + let height: i64 = m["height"].as_i64().unwrap_or(0); // determine sender id: // - if sent_by_self => sender is the requester (my_id) // - otherwise prefer an explicit chat_partner_id if present on the request, @@ -657,6 +733,7 @@ impl OmikronConnection { container.push((DataTypes::message, DataValue::Str(content))); container.push((DataTypes::sender_id, DataValue::Number(sender_id))); container.push((DataTypes::message_state, DataValue::Str(message_state))); + container.push((DataTypes::height, DataValue::Number(height))); msg_array.push(DataValue::Container(container)); } @@ -923,6 +1000,9 @@ impl OmikronConnection { let response = CommunicationValue::new(CommunicationType::error) .with_id(key) .add_data(DataTypes::message, DataValue::Str(reason.clone())); + // Historically this used the global `OMIKRON_CONNECTION`. Using the global here + // preserves the original behavior and avoids ownership/borrow issues when + // invoking the waiting-task closures from a &self context. let _ = (waiting_task.task)(OMIKRON_CONNECTION.clone(), response); } } diff --git a/src/util/chat_files.rs b/src/util/chat_files.rs index bfdc184..e54fe29 100644 --- a/src/util/chat_files.rs +++ b/src/util/chat_files.rs @@ -1,9 +1,9 @@ use crate::log; -use crate::util::file_util::get_directory; +use crate::util::db; use json::{JsonValue, array, object}; -use rusqlite::{Connection, params}; +use rusqlite::params; use std::io; -use std::sync::{LazyLock, Mutex}; +use std::sync::{Arc, LazyLock, Mutex}; #[derive(PartialEq, Debug, Clone)] pub enum MessageState { @@ -45,30 +45,10 @@ impl MessageState { } } -static DB_CONN: LazyLock> = LazyLock::new(|| { - let conn = Connection::open(format!("{}/messages.sqlite3", get_directory())) - .expect("Failed to open messages sqlite DB"); - conn.execute_batch( - r#" - PRAGMA journal_mode = WAL; - PRAGMA synchronous = NORMAL; - - CREATE TABLE IF NOT EXISTS messages ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - storage_owner INTEGER NOT NULL, - external_user INTEGER NOT NULL, - message_time INTEGER NOT NULL, - content TEXT NOT NULL, - sent_by_self INTEGER NOT NULL, - message_state TEXT NOT NULL - ); - - CREATE INDEX IF NOT EXISTS idx_messages_lookup - ON messages (storage_owner, external_user, message_time DESC); - "#, - ) - .expect("Failed to initialize messages DB"); - Mutex::new(conn) +// Shared DB created via helper. +// The db helper constructs the messages sqlite file and ensures PRAGMAs and schema exist. +static MESSAGES_DB: LazyLock>> = LazyLock::new(|| { + db::create_general_messages_db().expect("Failed to create or initialize general messages DB") }); pub fn add_message( @@ -77,6 +57,7 @@ pub fn add_message( storage_owner: i64, external_user: i64, message: &str, + height: i64, ) { let message_time = match i64::try_from(send_time) { Ok(v) => v, @@ -86,40 +67,48 @@ pub fn add_message( } }; - let conn = match DB_CONN.lock() { - Ok(g) => g, - Err(e) => { - log!("Failed to lock messages DB mutex for add_message: {:?}", e); - return; - } - }; + // Insert the message into the DB + let insert_result = db::with_conn(&MESSAGES_DB, |conn| { + conn.execute( + r#" + INSERT INTO messages ( + storage_owner, + external_user, + message_time, + content, + sent_by_self, + message_state, + height + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) + "#, + params![ + storage_owner, + external_user, + message_time, + message, + if storage_owner_is_sender { + 1_i64 + } else { + 0_i64 + }, + MessageState::Sending.as_str(), + height, + ], + )?; + Ok(()) + }); - if let Err(e) = conn.execute( - r#" - INSERT INTO messages ( - storage_owner, - external_user, - message_time, - content, - sent_by_self, - message_state - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6) - "#, - params![ - storage_owner, - external_user, - message_time, - message, - if storage_owner_is_sender { - 1_i64 - } else { - 0_i64 - }, - MessageState::Sending.as_str(), - ], - ) { + if let Err(e) = insert_result { log!("Failed to insert message into sqlite: {}", e); + return; } + + // Update contacts table to reflect that this conversation exists and has a recent message. + // Use the Contact helper to set last_message_at to the message timestamp. + let mut contact = crate::users::contact::Contact::new(external_user); + contact.set_last_message_at(message_time); + // This will insert or update the contact for the storage owner. + crate::util::chats_util::mod_user(storage_owner, &contact); } pub fn change_message_state( @@ -128,56 +117,58 @@ pub fn change_message_state( external_user: i64, new_state: MessageState, ) -> io::Result<()> { - let conn = DB_CONN - .lock() - .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("Mutex lock error: {:?}", e)))?; - - let current: Option = match conn.query_row( - r#" - SELECT message_state - FROM messages - WHERE storage_owner = ?1 - AND external_user = ?2 - AND message_time = ?3 - ORDER BY id DESC - LIMIT 1 - "#, - params![storage_owner, external_user, timestamp], - |row| row.get(0), - ) { - Ok(state) => Some(state), - Err(rusqlite::Error::QueryReturnedNoRows) => None, - Err(e) => return Err(io::Error::new(io::ErrorKind::Other, e.to_string())), - }; - - let Some(current_state_raw) = current else { - return Ok(()); - }; - - let upgraded = MessageState::from_str(¤t_state_raw) - .upgrade(new_state) - .as_str() - .to_string(); - - conn.execute( - r#" - UPDATE messages - SET message_state = ?1 - WHERE id = ( - SELECT id + // Run the SELECT and UPDATE inside with_conn to centralize connection access. + let res: Result<(), String> = db::with_conn(&MESSAGES_DB, |conn| { + let current: Option = match conn.query_row( + r#" + SELECT message_state FROM messages - WHERE storage_owner = ?2 - AND external_user = ?3 - AND message_time = ?4 + WHERE storage_owner = ?1 + AND external_user = ?2 + AND message_time = ?3 ORDER BY id DESC LIMIT 1 - ) - "#, - params![upgraded, storage_owner, external_user, timestamp], - ) - .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?; + "#, + params![storage_owner, external_user, timestamp], + |row| row.get(0), + ) { + Ok(state) => Some(state), + Err(rusqlite::Error::QueryReturnedNoRows) => None, + Err(e) => return Err(e), + }; - Ok(()) + let Some(current_state_raw) = current else { + return Ok(()); + }; + + let upgraded = MessageState::from_str(¤t_state_raw) + .upgrade(new_state) + .as_str() + .to_string(); + + conn.execute( + r#" + UPDATE messages + SET message_state = ?1 + WHERE id = ( + SELECT id + FROM messages + WHERE storage_owner = ?2 + AND external_user = ?3 + AND message_time = ?4 + ORDER BY id DESC + LIMIT 1 + ) + "#, + params![upgraded, storage_owner, external_user, timestamp], + )?; + Ok(()) + }); + + match res { + Ok(_) => Ok(()), + Err(e) => Err(io::Error::new(io::ErrorKind::Other, e)), + } } pub fn get_messages( @@ -186,80 +177,73 @@ pub fn get_messages( loaded_messages: i64, amount: i64, ) -> JsonValue { - let mut messages = array![]; + let messages = array![]; if amount <= 0 || loaded_messages < 0 { return messages; } - let conn = match DB_CONN.lock() { - Ok(g) => g, - Err(e) => { - log!("Failed to lock messages DB mutex for get_messages: {:?}", e); - return messages; - } - }; + let res: Result = db::with_conn(&MESSAGES_DB, |conn| { + let mut stmt = conn.prepare( + r#" + SELECT + message_time, + content, + sent_by_self, + message_state, + height + FROM messages + WHERE storage_owner = ?1 + AND external_user = ?2 + ORDER BY message_time DESC, id DESC + LIMIT ?3 OFFSET ?4 + "#, + )?; - let mut stmt = match conn.prepare( - r#" - SELECT - message_time, - content, - sent_by_self, - message_state - FROM messages - WHERE storage_owner = ?1 - AND external_user = ?2 - ORDER BY message_time DESC, id DESC - LIMIT ?3 OFFSET ?4 - "#, - ) { - Ok(s) => s, - Err(e) => { - log!("Failed to prepare get_messages query: {}", e); - return messages; - } - }; + let rows = stmt.query_map( + params![storage_owner, external_user, amount, loaded_messages], + |row| { + let message_time: i64 = row.get(0)?; + let content: String = row.get(1)?; + let sent_by_self: i64 = row.get(2)?; + let message_state: String = row.get(3)?; + let height: i64 = row.get(4).unwrap_or(0); + Ok((message_time, content, sent_by_self, message_state, height)) + }, + )?; - let rows = stmt.query_map( - params![storage_owner, external_user, amount, loaded_messages], - |row| { - let message_time: i64 = row.get(0)?; - let content: String = row.get(1)?; - let sent_by_self: i64 = row.get(2)?; - let message_state: String = row.get(3)?; - Ok((message_time, content, sent_by_self, message_state)) - }, - ); - - let Ok(rows) = rows else { - if let Err(e) = rows { - log!("Failed to query messages: {}", e); - } - return messages; - }; - - for row in rows { - match row { - Ok((message_time, content, sent_by_self, message_state)) => { - let msg = object! { - "message_time" => message_time, - "content" => content, - "sent_by_self" => (sent_by_self != 0), - "message_state" => message_state - }; - - if let Err(e) = messages.push(msg) { - log!("Failed to append message to output array: {}", e); + let mut out = array![]; + for row in rows { + match row { + Ok((message_time, content, sent_by_self, message_state, height)) => { + let msg = object! { + "message_time" => message_time, + "content" => content, + "sent_by_self" => (sent_by_self != 0), + "message_state" => message_state, + "height" => height + }; + if let Err(e) = out.push(msg) { + // out.push returns a JsonError; log it instead of using `?` to avoid + // incompatible error conversions inside the DB closure. + log!("Failed to append message to output array: {:?}", e); + } + } + Err(e) => { + log!("Failed to read row from sqlite: {}", e); } } - Err(e) => { - log!("Failed to read row from sqlite: {}", e); - } + } + Ok(out) + }); + + match res { + Ok(v) => v, + Err(e) => { + log!("Failed to query messages: {}", e); + messages } } - - messages } #[cfg(test)] diff --git a/src/util/chats_util.rs b/src/util/chats_util.rs index 0677b19..b8e0ad2 100644 --- a/src/util/chats_util.rs +++ b/src/util/chats_util.rs @@ -1,69 +1,96 @@ use crate::users::contact::Contact; -use crate::util::file_util::get_directory; -use rusqlite::{Connection, params}; -use std::sync::{LazyLock, Mutex}; +use crate::util::db; +use rusqlite::params; +use std::sync::{Arc, LazyLock, Mutex}; -static DB_CONN: LazyLock> = LazyLock::new(|| { - let conn = Connection::open(format!("{}/messages.sqlite3", get_directory())) - .expect("Failed to open DB"); - conn.execute_batch( - r#" - PRAGMA journal_mode = WAL; - PRAGMA synchronous = NORMAL; - - CREATE TABLE IF NOT EXISTS contacts ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - storage_owner INTEGER NOT NULL, - user_id INTEGER NOT NULL, - user_name TEXT, - last_message_at INTEGER, - UNIQUE(storage_owner, user_id) - ); - - CREATE INDEX IF NOT EXISTS idx_contacts_owner - ON contacts (storage_owner, last_message_at DESC, user_id ASC); - "#, - ) - .expect("Failed to initialize DB"); - Mutex::new(conn) +/// Shared DB connection for contacts/messages (created by db helper). +static MESSAGES_DB: LazyLock>> = LazyLock::new(|| { + db::create_general_messages_db().expect("Failed to create or initialize general messages DB") }); +/// Insert or update a contact for the given storage owner. pub fn mod_user(storage_owner: i64, contact: &Contact) { - let conn = DB_CONN.lock().unwrap(); - - let _ = conn.execute( - r#" - INSERT INTO contacts ( - storage_owner, - user_id, - user_name, - last_message_at - ) VALUES (?1, ?2, ?3, ?4) - ON CONFLICT(storage_owner, user_id) DO UPDATE SET - user_name = excluded.user_name, - last_message_at = excluded.last_message_at - "#, - params![ - storage_owner, - contact.user_id, - contact.user_name.clone(), - contact.last_message_at - ], - ); + if let Err(e) = db::with_conn(&MESSAGES_DB, |conn| { + conn.execute( + r#" + INSERT INTO contacts ( + storage_owner, + user_id, + user_name, + last_message_at + ) VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(storage_owner, user_id) DO UPDATE SET + user_name = excluded.user_name, + last_message_at = excluded.last_message_at + "#, + params![ + storage_owner, + contact.user_id, + contact.user_name.clone(), + contact.last_message_at + ], + )?; + Ok(()) + }) { + eprintln!("Failed to mod_user: {}", e); + } } +/// Retrieve a single contact for storage_owner/user_id. pub fn get_user(storage_owner: i64, user_id: i64) -> Option { - let conn = DB_CONN.lock().unwrap(); + let res: Result, String> = db::with_conn(&MESSAGES_DB, |conn| { + match conn.query_row( + r#" + SELECT user_id, user_name, last_message_at + FROM contacts + WHERE storage_owner = ?1 AND user_id = ?2 + LIMIT 1 + "#, + params![storage_owner, user_id], + |r| { + let user_id: i64 = r.get(0)?; + let user_name: Option = r.get(1)?; + let last_message_at: Option = r.get(2)?; + Ok(Contact { + user_id, + user_name, + last_message_at, + }) + }, + ) { + Ok(c) => Ok(Some(c)), + Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None), + Err(e) => Err(e), + } + }); - let row = conn.query_row( - r#" - SELECT user_id, user_name, last_message_at - FROM contacts - WHERE storage_owner = ?1 AND user_id = ?2 - LIMIT 1 - "#, - params![storage_owner, user_id], - |r| { + match res { + Ok(opt) => opt, + Err(e) => { + eprintln!("Error querying user in get_user: {}", e); + None + } + } +} + +/// Retrieve all contacts for a storage owner, ordered by last_message_at desc / user_id asc. +pub fn get_users(storage_owner: i64) -> Vec { + let contacts_out = Vec::new(); + + let res: Result, String> = db::with_conn(&MESSAGES_DB, |conn| { + let mut stmt = conn.prepare( + r#" + SELECT user_id, user_name, last_message_at + FROM contacts + WHERE storage_owner = ?1 + ORDER BY + CASE WHEN last_message_at IS NULL THEN 1 ELSE 0 END, + last_message_at DESC, + user_id ASC + "#, + )?; + + let rows = stmt.query_map(params![storage_owner], |r| { let user_id: i64 = r.get(0)?; let user_name: Option = r.get(1)?; let last_message_at: Option = r.get(2)?; @@ -72,64 +99,23 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Option { user_name, last_message_at, }) - }, - ); + })?; - match row { - Ok(contact) => Some(contact), - Err(rusqlite::Error::QueryReturnedNoRows) => None, + let mut out = Vec::new(); + for row in rows { + match row { + Ok(contact) => out.push(contact), + Err(e) => eprintln!("Failed to read contact row: {}", e), + } + } + Ok(out) + }); + + match res { + Ok(v) => v, Err(e) => { - eprintln!("Error querying user in get_user: {}", e); - None + eprintln!("Failed to query contacts in get_users: {}", e); + contacts_out } } } - -pub fn get_users(storage_owner: i64) -> Vec { - let mut contacts_out = Vec::new(); - - let conn = DB_CONN.lock().unwrap(); - - let mut stmt = match conn.prepare( - r#" - SELECT user_id, user_name, last_message_at - FROM contacts - WHERE storage_owner = ?1 - ORDER BY - CASE WHEN last_message_at IS NULL THEN 1 ELSE 0 END, - last_message_at DESC, - user_id ASC - "#, - ) { - Ok(s) => s, - Err(e) => { - eprintln!("Failed to prepare statement in get_users: {}", e); - return contacts_out; - } - }; - - let rows = match stmt.query_map(params![storage_owner], |r| { - let user_id: i64 = r.get(0)?; - let user_name: Option = r.get(1)?; - let last_message_at: Option = r.get(2)?; - Ok(Contact { - user_id, - user_name, - last_message_at, - }) - }) { - Ok(r) => r, - Err(e) => { - eprintln!("Failed to query map in get_users: {}", e); - return contacts_out; - } - }; - - for row in rows { - if let Ok(contact) = row { - contacts_out.push(contact); - } - } - - contacts_out -} diff --git a/src/util/db.rs b/src/util/db.rs new file mode 100644 index 0000000..d21ab41 --- /dev/null +++ b/src/util/db.rs @@ -0,0 +1,173 @@ +//! Database helper utilities. +//! +//! This module provides small helpers to open/init sqlite databases and to +//! create a shared (Arc>) connection wrapper callers can +//! reuse. The goal is to centralize the "open and initialize" logic and +//! provide small convenience helpers used by other util modules. + +use crate::util::file_util::get_directory; +use rusqlite::{Connection, Error as RusqliteError}; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +/// Returns the file path for a named DB inside the application's data directory. +/// +/// Arguments: +/// - `db_name` : name of the DB (without extension). Example: `"messages"`. +pub fn db_file_path(db_name: &str) -> String { + let mut p = PathBuf::from(get_directory()); + p.push(format!("{db_name}.sqlite3")); + p.to_string_lossy().to_string() +} + +/// Open a sqlite connection to the named DB file (no initialization). +/// +/// Arguments: +/// - `db_name`: name of the DB (without extension). +pub fn open_connection(db_name: &str) -> Result { + let path = db_file_path(db_name); + Connection::open(path) +} + +/// Open a connection and immediately run `init_sql` via `execute_batch`. +/// +/// Arguments: +/// - `db_name`: name of the DB (without extension). +/// - `init_sql`: SQL statements to initialize schema & PRAGMAs (can be multiple). +pub fn open_and_init(db_name: &str, init_sql: &str) -> Result { + let conn = open_connection(db_name)?; + conn.execute_batch(init_sql)?; + Ok(conn) +} + +/// Create a shared, Arc> initialized with the given SQL. +/// +/// This is a convenience wrapper that returns an owned Arc> +/// so caller modules can store it in a `static` or pass it around. +/// +/// Arguments: +/// - `db_name`: DB name (without extension). +/// - `init_sql`: init SQL (eg PRAGMA + CREATE TABLE statements). +pub fn create_shared_connection( + db_name: &str, + init_sql: &str, +) -> Result>, String> { + match open_and_init(db_name, init_sql) { + Ok(conn) => { + // Configure some sensible defaults for concurrency + // Attempt to set a busy timeout to reduce SQLITE_BUSY failures. + let _ = conn.busy_timeout(Duration::from_millis(250)); + Ok(Arc::new(Mutex::new(conn))) + } + Err(e) => Err(format!("Failed to open/init DB '{}': {}", db_name, e)), + } +} + +/// Acquire the Connection from an Arc> and run the provided +/// closure. Converts rusqlite::Error into a String on error. +/// +/// Arguments: +/// - `shared`: Arc> +/// - `f`: closure that receives &Connection and returns Result +/// +/// Returns Ok(T) or Err(String). +pub fn with_conn(shared: &Arc>, f: F) -> Result +where + F: FnOnce(&Connection) -> Result, +{ + // When invoked from within an async runtime (such as Tokio), taking a blocking + // std::sync::Mutex lock on the runtime thread can cause deadlocks or permanent + // awaits. Detect whether we're running inside a Tokio runtime and, if so, + // execute the blocking lock + database closure using Tokio's blocking helper. + // + // The blocking section returns Result so we can propagate errors + // in the same form as before. + if tokio::runtime::Handle::try_current().is_ok() { + tokio::task::block_in_place(|| { + let guard = shared + .lock() + .map_err(|e| format!("DB mutex poisoned: {:?}", e))?; + f(&*guard).map_err(|e| e.to_string()) + }) + } else { + let guard = shared + .lock() + .map_err(|e| format!("DB mutex poisoned: {:?}", e))?; + f(&*guard).map_err(|e| e.to_string()) + } +} + +/// Initialize a general-purpose messages+contacts DB and return a shared +/// connection. This helper creates a single DB file that can contain multiple +/// tables (messages, contacts, ...). The SQL here is conservative and intended +/// to be safe if called multiple times. +/// +/// Callers may prefer to call `create_shared_connection("messages", INIT_SQL)` +/// directly, but this convenience is useful for code that expects both tables. +pub fn create_general_messages_db() -> Result>, String> { + // Keep PRAGMA and schema in one multi-statement string so callers only + // need to call a single execute_batch. + const INIT_SQL: &str = r#" + PRAGMA journal_mode = WAL; + PRAGMA synchronous = NORMAL; + + CREATE TABLE IF NOT EXISTS messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + storage_owner INTEGER NOT NULL, + external_user INTEGER NOT NULL, + message_time INTEGER NOT NULL, + content TEXT NOT NULL, + sent_by_self INTEGER NOT NULL, + message_state TEXT NOT NULL, + height INTEGER NOT NULL DEFAULT 0 + ); + + CREATE INDEX IF NOT EXISTS idx_messages_lookup + ON messages (storage_owner, external_user, message_time DESC); + + CREATE TABLE IF NOT EXISTS contacts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + storage_owner INTEGER NOT NULL, + user_id INTEGER NOT NULL, + user_name TEXT, + last_message_at INTEGER, + UNIQUE(storage_owner, user_id) + ); + + CREATE INDEX IF NOT EXISTS idx_contacts_owner + ON contacts (storage_owner, last_message_at DESC, user_id ASC); + "#; + + match create_shared_connection("messages", INIT_SQL) { + Ok(shared_conn) => { + // Attempt to add the height column for backwards compatibility. + // This will fail if the column already exists, which is expected. + let _ = with_conn(&shared_conn, |conn| { + let _ = conn.execute( + "ALTER TABLE messages ADD COLUMN height INTEGER NOT NULL DEFAULT 0", + [], + ); + Ok(()) + }); + Ok(shared_conn) + } + Err(e) => Err(e), + } +} + +/* +Example usage: + +// In some util module (at init time, e.g. lazy_static or LazyLock) +static MESSAGES_DB: LazyLock>> = LazyLock::new(|| { + create_general_messages_db().expect("failed to create messages DB") +}); + +// Later, to run a query: +let res: Result, String> = with_conn(&MESSAGES_DB, |conn| { + let mut stmt = conn.prepare("SELECT ...")?; + let rows = stmt.query_map(...)?; + // collect and return Ok(...) +}); +*/ diff --git a/src/util/mod.rs b/src/util/mod.rs index 8a1382b..efd4374 100644 --- a/src/util/mod.rs +++ b/src/util/mod.rs @@ -3,5 +3,6 @@ pub mod chats_util; pub mod config_util; pub mod crypto_helper; pub mod crypto_util; +pub mod db; pub mod file_util; pub mod logger;