tensamin/iota
Watch
6
0
Fork
You've already forked iota
 0
Code Issues Pull requests 3 Projects Releases 1 Packages Wiki Activity Actions

[Add] DB util

Browse source 
[Fix] Chat files & Chats util
[Add] message Height
This commit is contained in:
Alex Emmet 2026-03-29 22:28:23 +02:00
commit 96e5ed89d6
6 changed files with 587 additions and 363 deletions
Download patch file Download diff file Expand all files Collapse all files

12
Cargo.lock generated
View file

@ -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",

246
src/omikron/omikron_connection.rs
View file

@ -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::<i64>().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::<i64>().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::<i64>().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::<i64>().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<DataValue> = 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);
}
}

318
src/util/chat_files.rs
View file

@ -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<Mutex<Connection>> = 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<Arc<Mutex<rusqlite::Connection>>> = 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<String> = 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(&current_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<String> = 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(&current_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<JsonValue, String> = 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)]

212
src/util/chats_util.rs
View file

@ -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<Mutex<Connection>> = 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<Arc<Mutex<rusqlite::Connection>>> = 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<Contact> {
let conn = DB_CONN.lock().unwrap();
let res: Result<Option<Contact>, 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<String> = r.get(1)?;
let last_message_at: Option<i64> = 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<Contact> {
let contacts_out = Vec::new();
let res: Result<Vec<Contact>, 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<String> = r.get(1)?;
let last_message_at: Option<i64> = r.get(2)?;
@ -72,64 +99,23 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Option<Contact> {
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<Contact> {
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<String> = r.get(1)?;
let last_message_at: Option<i64> = 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
}

173
src/util/db.rs Normal file
View file

@ -0,0 +1,173 @@
//! Database helper utilities.
//!
//! This module provides small helpers to open/init sqlite databases and to
//! create a shared (Arc<Mutex<Connection>>) 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<Connection, RusqliteError> {
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<Connection, RusqliteError> {
let conn = open_connection(db_name)?;
conn.execute_batch(init_sql)?;
Ok(conn)
}
/// Create a shared, Arc<Mutex<Connection>> initialized with the given SQL.
///
/// This is a convenience wrapper that returns an owned Arc<Mutex<Connection>>
/// 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<Arc<Mutex<Connection>>, 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<Mutex<Connection>> and run the provided
/// closure. Converts rusqlite::Error into a String on error.
///
/// Arguments:
/// - `shared`: Arc<Mutex<Connection>>
/// - `f`: closure that receives &Connection and returns Result<T, RusqliteError>
///
/// Returns Ok(T) or Err(String).
pub fn with_conn<T, F>(shared: &Arc<Mutex<Connection>>, f: F) -> Result<T, String>
where
F: FnOnce(&Connection) -> Result<T, RusqliteError>,
{
// 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<T, String> 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<Arc<Mutex<Connection>>, 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<Arc<Mutex<Connection>>> = LazyLock::new(|| {
create_general_messages_db().expect("failed to create messages DB")
});
// Later, to run a query:
let res: Result<Vec<MyRow>, String> = with_conn(&MESSAGES_DB, |conn| {
let mut stmt = conn.prepare("SELECT ...")?;
let rows = stmt.query_map(...)?;
// collect and return Ok(...)
});
*/

1
src/util/mod.rs
View file

@ -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;