Christian Nennemann
a710037dde
Rename all crate directories, package names, binary names, proto package/module paths, ALPN strings, env var prefixes, config filenames, mDNS service names, and plugin ABI symbols from quicproquo/qpq to quicprochat/qpc.
1042 lines
36 KiB
Rust
1042 lines
36 KiB
Rust
//! Conversation management — create DMs, groups, send and receive messages.
|
|
//!
|
|
//! This is the SDK-side conversation store (migrated from v1 client).
|
|
|
|
use std::path::Path;
|
|
|
|
use anyhow::Context;
|
|
use rusqlite::{params, Connection, OptionalExtension};
|
|
use zeroize::Zeroizing;
|
|
|
|
// ── Types ────────────────────────────────────────────────────────────────────
|
|
|
|
/// 16-byte conversation identifier.
|
|
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
|
|
pub struct ConversationId(pub [u8; 16]);
|
|
|
|
impl ConversationId {
|
|
pub fn from_slice(s: &[u8]) -> Option<Self> {
|
|
if s.len() == 16 {
|
|
let mut buf = [0u8; 16];
|
|
buf.copy_from_slice(s);
|
|
Some(Self(buf))
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
/// Derive a conversation ID from a group name via SHA-256 truncation.
|
|
pub fn from_group_name(name: &str) -> Self {
|
|
use sha2::{Digest, Sha256};
|
|
let hash = Sha256::digest(name.as_bytes());
|
|
let mut buf = [0u8; 16];
|
|
buf.copy_from_slice(&hash[..16]);
|
|
Self(buf)
|
|
}
|
|
|
|
pub fn hex(&self) -> String {
|
|
hex::encode(self.0)
|
|
}
|
|
}
|
|
|
|
/// The kind of conversation.
|
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
|
pub enum ConversationKind {
|
|
/// 1:1 DM channel with a specific peer.
|
|
Dm {
|
|
peer_key: Vec<u8>,
|
|
peer_username: Option<String>,
|
|
},
|
|
/// Named group with N members.
|
|
Group { name: String },
|
|
}
|
|
|
|
/// A conversation with its metadata and MLS state.
|
|
#[derive(Clone, Debug)]
|
|
pub struct Conversation {
|
|
pub id: ConversationId,
|
|
pub kind: ConversationKind,
|
|
pub display_name: String,
|
|
/// Serialized MLS group (bincode).
|
|
pub mls_group_blob: Option<Vec<u8>>,
|
|
/// Serialized keystore (bincode HashMap).
|
|
pub keystore_blob: Option<Vec<u8>>,
|
|
/// Member identity keys.
|
|
pub member_keys: Vec<Vec<u8>>,
|
|
pub unread_count: u32,
|
|
pub last_activity_ms: u64,
|
|
pub created_at_ms: u64,
|
|
/// Whether this conversation uses hybrid (X25519 + ML-KEM-768) MLS keys.
|
|
pub is_hybrid: bool,
|
|
/// Highest server-side delivery sequence number seen.
|
|
pub last_seen_seq: u64,
|
|
}
|
|
|
|
/// A stored message.
|
|
#[derive(Clone, Debug)]
|
|
pub struct StoredMessage {
|
|
pub conversation_id: ConversationId,
|
|
pub message_id: Option<[u8; 16]>,
|
|
pub sender_key: Vec<u8>,
|
|
pub sender_name: Option<String>,
|
|
pub body: String,
|
|
pub msg_type: String,
|
|
pub ref_msg_id: Option<[u8; 16]>,
|
|
pub timestamp_ms: u64,
|
|
pub is_outgoing: bool,
|
|
}
|
|
|
|
/// An entry in the offline outbox queue.
|
|
#[derive(Clone, Debug)]
|
|
pub struct OutboxEntry {
|
|
pub id: i64,
|
|
pub conversation_id: ConversationId,
|
|
pub recipient_key: Vec<u8>,
|
|
pub payload: Vec<u8>,
|
|
pub retry_count: u32,
|
|
}
|
|
|
|
/// A blocked user entry.
|
|
#[derive(Clone, Debug)]
|
|
pub struct BlockedUser {
|
|
pub identity_key: Vec<u8>,
|
|
pub blocked_at_ms: u64,
|
|
pub reason: String,
|
|
}
|
|
|
|
// ── ConversationStore ────────────────────────────────────────────────────────
|
|
|
|
/// SQLCipher-backed conversation and message store.
|
|
pub struct ConversationStore {
|
|
conn: Connection,
|
|
}
|
|
|
|
impl ConversationStore {
|
|
/// Open or create the conversation database.
|
|
pub fn open(db_path: &Path, password: Option<&str>) -> anyhow::Result<Self> {
|
|
if let Some(parent) = db_path.parent() {
|
|
std::fs::create_dir_all(parent).ok();
|
|
}
|
|
|
|
let conn = Connection::open(db_path).context("open conversation db")?;
|
|
|
|
if let Some(pw) = password {
|
|
let key = derive_db_key(pw, db_path)?;
|
|
#[allow(clippy::needless_borrows_for_generic_args)]
|
|
let hex_key = Zeroizing::new(hex::encode(&*key));
|
|
conn.pragma_update(None, "key", format!("x'{}'", &*hex_key))
|
|
.context("set SQLCipher key")?;
|
|
}
|
|
|
|
conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")
|
|
.context("set pragmas")?;
|
|
Self::migrate(&conn)?;
|
|
Ok(Self { conn })
|
|
}
|
|
|
|
fn migrate(conn: &Connection) -> anyhow::Result<()> {
|
|
conn.execute_batch(
|
|
"CREATE TABLE IF NOT EXISTS conversations (
|
|
id BLOB PRIMARY KEY,
|
|
kind TEXT NOT NULL,
|
|
display_name TEXT NOT NULL,
|
|
peer_key BLOB,
|
|
peer_username TEXT,
|
|
group_name TEXT,
|
|
mls_group_blob BLOB,
|
|
keystore_blob BLOB,
|
|
member_keys BLOB,
|
|
unread_count INTEGER NOT NULL DEFAULT 0,
|
|
last_activity_ms INTEGER NOT NULL DEFAULT 0,
|
|
created_at_ms INTEGER NOT NULL DEFAULT 0,
|
|
is_hybrid INTEGER NOT NULL DEFAULT 0,
|
|
last_seen_seq INTEGER NOT NULL DEFAULT 0
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
conversation_id BLOB NOT NULL REFERENCES conversations(id),
|
|
message_id BLOB,
|
|
sender_key BLOB NOT NULL,
|
|
sender_name TEXT,
|
|
body TEXT NOT NULL,
|
|
msg_type TEXT NOT NULL,
|
|
ref_msg_id BLOB,
|
|
timestamp_ms INTEGER NOT NULL,
|
|
is_outgoing INTEGER NOT NULL DEFAULT 0
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_messages_conv
|
|
ON messages(conversation_id, timestamp_ms);
|
|
|
|
CREATE TABLE IF NOT EXISTS outbox (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
conversation_id BLOB NOT NULL,
|
|
recipient_key BLOB NOT NULL,
|
|
payload BLOB NOT NULL,
|
|
created_at_ms INTEGER NOT NULL,
|
|
retry_count INTEGER NOT NULL DEFAULT 0,
|
|
status TEXT NOT NULL DEFAULT 'pending'
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_outbox_status
|
|
ON outbox(status, created_at_ms);
|
|
|
|
CREATE TABLE IF NOT EXISTS blocked_users (
|
|
identity_key BLOB PRIMARY KEY,
|
|
blocked_at_ms INTEGER NOT NULL,
|
|
reason TEXT NOT NULL DEFAULT ''
|
|
);",
|
|
)
|
|
.context("migrate conversation db")
|
|
}
|
|
|
|
/// Save or upsert a conversation.
|
|
pub fn save_conversation(&self, conv: &Conversation) -> anyhow::Result<()> {
|
|
let (kind_str, peer_key, peer_username, group_name) = match &conv.kind {
|
|
ConversationKind::Dm { peer_key, peer_username } => {
|
|
("dm", Some(peer_key.as_slice()), peer_username.as_deref(), None)
|
|
}
|
|
ConversationKind::Group { name } => ("group", None, None, Some(name.as_str())),
|
|
};
|
|
let member_keys_blob =
|
|
bincode::serialize(&conv.member_keys).context("serialize member_keys")?;
|
|
|
|
self.conn.execute(
|
|
"INSERT INTO conversations
|
|
(id, kind, display_name, peer_key, peer_username, group_name,
|
|
mls_group_blob, keystore_blob, member_keys, unread_count,
|
|
last_activity_ms, created_at_ms, is_hybrid, last_seen_seq)
|
|
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
display_name = excluded.display_name,
|
|
mls_group_blob = excluded.mls_group_blob,
|
|
keystore_blob = excluded.keystore_blob,
|
|
member_keys = excluded.member_keys,
|
|
unread_count = excluded.unread_count,
|
|
last_activity_ms = excluded.last_activity_ms,
|
|
is_hybrid = excluded.is_hybrid,
|
|
last_seen_seq = excluded.last_seen_seq",
|
|
params![
|
|
conv.id.0.as_slice(),
|
|
kind_str,
|
|
conv.display_name,
|
|
peer_key,
|
|
peer_username,
|
|
group_name,
|
|
conv.mls_group_blob,
|
|
conv.keystore_blob,
|
|
member_keys_blob,
|
|
conv.unread_count,
|
|
conv.last_activity_ms,
|
|
conv.created_at_ms,
|
|
conv.is_hybrid as i32,
|
|
conv.last_seen_seq as i64,
|
|
],
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Load a conversation by ID.
|
|
pub fn load_conversation(&self, id: &ConversationId) -> anyhow::Result<Option<Conversation>> {
|
|
self.conn
|
|
.query_row(
|
|
"SELECT kind, display_name, peer_key, peer_username, group_name,
|
|
mls_group_blob, keystore_blob, member_keys, unread_count,
|
|
last_activity_ms, created_at_ms, is_hybrid, last_seen_seq
|
|
FROM conversations WHERE id = ?1",
|
|
params![id.0.as_slice()],
|
|
|row| row_to_conversation(id, row),
|
|
)
|
|
.optional()
|
|
.context("load conversation")
|
|
}
|
|
|
|
/// List all conversations, most recent first.
|
|
pub fn list_conversations(&self) -> anyhow::Result<Vec<Conversation>> {
|
|
let mut stmt = self.conn.prepare(
|
|
"SELECT id, kind, display_name, peer_key, peer_username, group_name,
|
|
mls_group_blob, keystore_blob, member_keys, unread_count,
|
|
last_activity_ms, created_at_ms, is_hybrid, last_seen_seq
|
|
FROM conversations ORDER BY last_activity_ms DESC",
|
|
)?;
|
|
let rows = stmt.query_map([], |row| {
|
|
let id_blob: Vec<u8> = row.get(0)?;
|
|
let id = ConversationId::from_slice(&id_blob).unwrap_or(ConversationId([0; 16]));
|
|
row_to_conversation_full(&id, row)
|
|
})?;
|
|
let mut convs = Vec::new();
|
|
for row in rows {
|
|
convs.push(row?);
|
|
}
|
|
Ok(convs)
|
|
}
|
|
|
|
/// Find a DM by peer identity key.
|
|
pub fn find_dm_by_peer(&self, peer_key: &[u8]) -> anyhow::Result<Option<Conversation>> {
|
|
let id_blob: Option<Vec<u8>> = self
|
|
.conn
|
|
.query_row(
|
|
"SELECT id FROM conversations WHERE kind = 'dm' AND peer_key = ?1",
|
|
params![peer_key],
|
|
|row| row.get(0),
|
|
)
|
|
.optional()?;
|
|
match id_blob {
|
|
Some(blob) => {
|
|
let id = ConversationId::from_slice(&blob)
|
|
.context("invalid conversation id")?;
|
|
self.load_conversation(&id)
|
|
}
|
|
None => Ok(None),
|
|
}
|
|
}
|
|
|
|
/// Save a message.
|
|
pub fn save_message(&self, msg: &StoredMessage) -> anyhow::Result<()> {
|
|
self.conn.execute(
|
|
"INSERT INTO messages
|
|
(conversation_id, message_id, sender_key, sender_name, body,
|
|
msg_type, ref_msg_id, timestamp_ms, is_outgoing)
|
|
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
|
|
params![
|
|
msg.conversation_id.0.as_slice(),
|
|
msg.message_id.as_ref().map(|id| id.as_slice()),
|
|
msg.sender_key,
|
|
msg.sender_name,
|
|
msg.body,
|
|
msg.msg_type,
|
|
msg.ref_msg_id.as_ref().map(|id| id.as_slice()),
|
|
msg.timestamp_ms,
|
|
msg.is_outgoing as i32,
|
|
],
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Enqueue an outbox entry for offline sending.
|
|
pub fn enqueue_outbox(
|
|
&self,
|
|
conv_id: &ConversationId,
|
|
recipient_key: &[u8],
|
|
payload: &[u8],
|
|
) -> anyhow::Result<()> {
|
|
self.conn.execute(
|
|
"INSERT INTO outbox (conversation_id, recipient_key, payload, created_at_ms)
|
|
VALUES (?1, ?2, ?3, ?4)",
|
|
params![conv_id.0.as_slice(), recipient_key, payload, now_ms() as i64],
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Load all pending outbox entries, oldest first.
|
|
pub fn load_pending_outbox(&self) -> anyhow::Result<Vec<OutboxEntry>> {
|
|
let mut stmt = self.conn.prepare(
|
|
"SELECT id, conversation_id, recipient_key, payload, retry_count
|
|
FROM outbox WHERE status = 'pending' ORDER BY created_at_ms",
|
|
)?;
|
|
let rows = stmt.query_map([], |row| {
|
|
let id: i64 = row.get(0)?;
|
|
let conv_blob: Vec<u8> = row.get(1)?;
|
|
let recipient_key: Vec<u8> = row.get(2)?;
|
|
let payload: Vec<u8> = row.get(3)?;
|
|
let retry_count: u32 = row.get(4)?;
|
|
Ok(OutboxEntry {
|
|
id,
|
|
conversation_id: ConversationId::from_slice(&conv_blob)
|
|
.unwrap_or(ConversationId([0; 16])),
|
|
recipient_key,
|
|
payload,
|
|
retry_count,
|
|
})
|
|
})?;
|
|
let mut entries = Vec::new();
|
|
for row in rows {
|
|
entries.push(row?);
|
|
}
|
|
Ok(entries)
|
|
}
|
|
|
|
/// Mark an outbox entry as sent.
|
|
pub fn mark_outbox_sent(&self, id: i64) -> anyhow::Result<()> {
|
|
self.conn.execute(
|
|
"UPDATE outbox SET status = 'sent' WHERE id = ?1",
|
|
params![id],
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Mark an outbox entry as failed (retryable up to 10 times).
|
|
pub fn mark_outbox_failed(&self, id: i64, retry_count: u32) -> anyhow::Result<()> {
|
|
let new_status = if retry_count > 10 { "failed" } else { "pending" };
|
|
self.conn.execute(
|
|
"UPDATE outbox SET retry_count = ?2, status = ?3 WHERE id = ?1",
|
|
params![id, retry_count, new_status],
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Count pending outbox entries.
|
|
pub fn count_pending_outbox(&self) -> anyhow::Result<usize> {
|
|
let count: i64 = self.conn.query_row(
|
|
"SELECT COUNT(*) FROM outbox WHERE status = 'pending'",
|
|
[],
|
|
|row| row.get(0),
|
|
)?;
|
|
Ok(count as usize)
|
|
}
|
|
|
|
/// Delete all permanently failed outbox entries (status = 'failed').
|
|
/// Returns the number of entries removed.
|
|
pub fn clear_failed_outbox(&self) -> anyhow::Result<usize> {
|
|
let count = self.conn.execute(
|
|
"DELETE FROM outbox WHERE status = 'failed'",
|
|
[],
|
|
)?;
|
|
Ok(count)
|
|
}
|
|
|
|
// ── Sequence tracking ──────────────────────────────────────────────────
|
|
|
|
/// Update the last seen server sequence number for a conversation.
|
|
/// Only increases; a lower seq is a no-op.
|
|
pub fn update_last_seen_seq(
|
|
&self,
|
|
conv_id: &ConversationId,
|
|
seq: u64,
|
|
) -> anyhow::Result<()> {
|
|
self.conn.execute(
|
|
"UPDATE conversations SET last_seen_seq = ?2 WHERE id = ?1 AND last_seen_seq < ?2",
|
|
params![conv_id.0.as_slice(), seq as i64],
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Get the last seen server sequence number for a conversation.
|
|
pub fn get_last_seen_seq(&self, conv_id: &ConversationId) -> anyhow::Result<u64> {
|
|
let seq: i64 = self
|
|
.conn
|
|
.query_row(
|
|
"SELECT last_seen_seq FROM conversations WHERE id = ?1",
|
|
params![conv_id.0.as_slice()],
|
|
|row| row.get(0),
|
|
)
|
|
.optional()?
|
|
.unwrap_or(0);
|
|
Ok(seq as u64)
|
|
}
|
|
|
|
// ── Blocked users ─────────────────────────────────────────────────────
|
|
|
|
/// Block a user by identity key.
|
|
pub fn block_user(&self, identity_key: &[u8], reason: &str) -> anyhow::Result<()> {
|
|
self.conn.execute(
|
|
"INSERT OR REPLACE INTO blocked_users (identity_key, blocked_at_ms, reason)
|
|
VALUES (?1, ?2, ?3)",
|
|
params![identity_key, now_ms() as i64, reason],
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Unblock a user. Returns true if the user was blocked.
|
|
pub fn unblock_user(&self, identity_key: &[u8]) -> anyhow::Result<bool> {
|
|
let rows = self.conn.execute(
|
|
"DELETE FROM blocked_users WHERE identity_key = ?1",
|
|
params![identity_key],
|
|
)?;
|
|
Ok(rows > 0)
|
|
}
|
|
|
|
/// Check if a user is blocked.
|
|
pub fn is_blocked(&self, identity_key: &[u8]) -> anyhow::Result<bool> {
|
|
let count: i64 = self.conn.query_row(
|
|
"SELECT COUNT(*) FROM blocked_users WHERE identity_key = ?1",
|
|
params![identity_key],
|
|
|row| row.get(0),
|
|
)?;
|
|
Ok(count > 0)
|
|
}
|
|
|
|
/// List all blocked users: (identity_key, blocked_at_ms, reason).
|
|
pub fn list_blocked(&self) -> anyhow::Result<Vec<BlockedUser>> {
|
|
let mut stmt = self.conn.prepare(
|
|
"SELECT identity_key, blocked_at_ms, reason FROM blocked_users
|
|
ORDER BY blocked_at_ms DESC",
|
|
)?;
|
|
let rows = stmt.query_map([], |row| {
|
|
let identity_key: Vec<u8> = row.get(0)?;
|
|
let blocked_at_ms: u64 = row.get(1)?;
|
|
let reason: String = row.get(2)?;
|
|
Ok(BlockedUser {
|
|
identity_key,
|
|
blocked_at_ms,
|
|
reason,
|
|
})
|
|
})?;
|
|
let mut users = Vec::new();
|
|
for row in rows {
|
|
users.push(row?);
|
|
}
|
|
Ok(users)
|
|
}
|
|
|
|
/// Load recent messages, filtering out messages from blocked users.
|
|
pub fn load_recent_messages_filtered(
|
|
&self,
|
|
conv_id: &ConversationId,
|
|
limit: usize,
|
|
) -> anyhow::Result<Vec<StoredMessage>> {
|
|
let mut stmt = self.conn.prepare(
|
|
"SELECT m.message_id, m.sender_key, m.sender_name, m.body, m.msg_type,
|
|
m.ref_msg_id, m.timestamp_ms, m.is_outgoing
|
|
FROM messages m
|
|
WHERE m.conversation_id = ?1
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM blocked_users b WHERE b.identity_key = m.sender_key
|
|
)
|
|
ORDER BY m.timestamp_ms DESC LIMIT ?2",
|
|
)?;
|
|
let rows = stmt.query_map(
|
|
params![conv_id.0.as_slice(), limit.min(u32::MAX as usize) as u32],
|
|
|row| row_to_message(conv_id, row),
|
|
)?;
|
|
let mut msgs: Vec<StoredMessage> = rows.collect::<Result<_, _>>()?;
|
|
msgs.reverse();
|
|
Ok(msgs)
|
|
}
|
|
|
|
/// Load recent messages (newest first, then reversed to chronological).
|
|
pub fn load_recent_messages(
|
|
&self,
|
|
conv_id: &ConversationId,
|
|
limit: usize,
|
|
) -> anyhow::Result<Vec<StoredMessage>> {
|
|
let mut stmt = self.conn.prepare(
|
|
"SELECT message_id, sender_key, sender_name, body, msg_type,
|
|
ref_msg_id, timestamp_ms, is_outgoing
|
|
FROM messages WHERE conversation_id = ?1
|
|
ORDER BY timestamp_ms DESC LIMIT ?2",
|
|
)?;
|
|
let rows = stmt.query_map(
|
|
params![conv_id.0.as_slice(), limit.min(u32::MAX as usize) as u32],
|
|
|row| row_to_message(conv_id, row),
|
|
)?;
|
|
let mut msgs: Vec<StoredMessage> = rows.collect::<Result<_, _>>()?;
|
|
msgs.reverse();
|
|
Ok(msgs)
|
|
}
|
|
}
|
|
|
|
// ── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
fn derive_db_key(password: &str, db_path: &Path) -> anyhow::Result<Zeroizing<[u8; 32]>> {
|
|
use argon2::{Algorithm, Argon2, Params, Version};
|
|
|
|
let salt_path = db_path.with_extension("db-salt");
|
|
let salt = if salt_path.exists() {
|
|
std::fs::read(&salt_path).context("read db salt")?
|
|
} else {
|
|
let mut salt = vec![0u8; 16];
|
|
rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut salt);
|
|
std::fs::write(&salt_path, &salt).context("write db salt")?;
|
|
#[cfg(unix)]
|
|
{
|
|
use std::os::unix::fs::PermissionsExt;
|
|
std::fs::set_permissions(&salt_path, std::fs::Permissions::from_mode(0o600)).ok();
|
|
}
|
|
salt
|
|
};
|
|
|
|
let params = Params::new(19 * 1024, 2, 1, Some(32))
|
|
.map_err(|e| anyhow::anyhow!("argon2 params: {e}"))?;
|
|
let argon2 = Argon2::new(Algorithm::Argon2id, Version::default(), params);
|
|
let mut key = Zeroizing::new([0u8; 32]);
|
|
argon2
|
|
.hash_password_into(password.as_bytes(), &salt, &mut *key)
|
|
.map_err(|e| anyhow::anyhow!("db key derivation: {e}"))?;
|
|
Ok(key)
|
|
}
|
|
|
|
fn row_to_conversation(
|
|
id: &ConversationId,
|
|
row: &rusqlite::Row<'_>,
|
|
) -> rusqlite::Result<Conversation> {
|
|
let kind_str: String = row.get(0)?;
|
|
let display_name: String = row.get(1)?;
|
|
let peer_key: Option<Vec<u8>> = row.get(2)?;
|
|
let peer_username: Option<String> = row.get(3)?;
|
|
let group_name: Option<String> = row.get(4)?;
|
|
let mls_group_blob: Option<Vec<u8>> = row.get(5)?;
|
|
let keystore_blob: Option<Vec<u8>> = row.get(6)?;
|
|
let member_keys_blob: Option<Vec<u8>> = row.get(7)?;
|
|
let unread_count: u32 = row.get(8)?;
|
|
let last_activity_ms: u64 = row.get(9)?;
|
|
let created_at_ms: u64 = row.get(10)?;
|
|
let is_hybrid_int: i32 = row.get(11)?;
|
|
let last_seen_seq: i64 = row.get(12)?;
|
|
|
|
let kind = if kind_str == "dm" {
|
|
ConversationKind::Dm {
|
|
peer_key: peer_key.unwrap_or_default(),
|
|
peer_username,
|
|
}
|
|
} else {
|
|
ConversationKind::Group {
|
|
name: group_name.unwrap_or_default(),
|
|
}
|
|
};
|
|
let member_keys: Vec<Vec<u8>> = member_keys_blob
|
|
.and_then(|b| bincode::deserialize(&b).ok())
|
|
.unwrap_or_default();
|
|
|
|
Ok(Conversation {
|
|
id: id.clone(),
|
|
kind,
|
|
display_name,
|
|
mls_group_blob,
|
|
keystore_blob,
|
|
member_keys,
|
|
unread_count,
|
|
last_activity_ms,
|
|
created_at_ms,
|
|
is_hybrid: is_hybrid_int != 0,
|
|
last_seen_seq: last_seen_seq as u64,
|
|
})
|
|
}
|
|
|
|
fn row_to_conversation_full(
|
|
id: &ConversationId,
|
|
row: &rusqlite::Row<'_>,
|
|
) -> rusqlite::Result<Conversation> {
|
|
let kind_str: String = row.get(1)?;
|
|
let display_name: String = row.get(2)?;
|
|
let peer_key: Option<Vec<u8>> = row.get(3)?;
|
|
let peer_username: Option<String> = row.get(4)?;
|
|
let group_name: Option<String> = row.get(5)?;
|
|
let mls_group_blob: Option<Vec<u8>> = row.get(6)?;
|
|
let keystore_blob: Option<Vec<u8>> = row.get(7)?;
|
|
let member_keys_blob: Option<Vec<u8>> = row.get(8)?;
|
|
let unread_count: u32 = row.get(9)?;
|
|
let last_activity_ms: u64 = row.get(10)?;
|
|
let created_at_ms: u64 = row.get(11)?;
|
|
let is_hybrid_int: i32 = row.get(12)?;
|
|
let last_seen_seq: i64 = row.get(13)?;
|
|
|
|
let kind = if kind_str == "dm" {
|
|
ConversationKind::Dm {
|
|
peer_key: peer_key.unwrap_or_default(),
|
|
peer_username,
|
|
}
|
|
} else {
|
|
ConversationKind::Group {
|
|
name: group_name.unwrap_or_default(),
|
|
}
|
|
};
|
|
let member_keys: Vec<Vec<u8>> = member_keys_blob
|
|
.and_then(|b| bincode::deserialize(&b).ok())
|
|
.unwrap_or_default();
|
|
|
|
Ok(Conversation {
|
|
id: id.clone(),
|
|
kind,
|
|
display_name,
|
|
mls_group_blob,
|
|
keystore_blob,
|
|
member_keys,
|
|
unread_count,
|
|
last_activity_ms,
|
|
created_at_ms,
|
|
is_hybrid: is_hybrid_int != 0,
|
|
last_seen_seq: last_seen_seq as u64,
|
|
})
|
|
}
|
|
|
|
fn to_16(v: &[u8]) -> Option<[u8; 16]> {
|
|
if v.len() == 16 {
|
|
let mut buf = [0u8; 16];
|
|
buf.copy_from_slice(v);
|
|
Some(buf)
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
/// Current timestamp in milliseconds since UNIX epoch.
|
|
pub fn now_ms() -> u64 {
|
|
std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.unwrap_or_default()
|
|
.as_millis() as u64
|
|
}
|
|
|
|
fn row_to_message(
|
|
conv_id: &ConversationId,
|
|
row: &rusqlite::Row<'_>,
|
|
) -> rusqlite::Result<StoredMessage> {
|
|
let message_id: Option<Vec<u8>> = row.get(0)?;
|
|
let sender_key: Vec<u8> = row.get(1)?;
|
|
let sender_name: Option<String> = row.get(2)?;
|
|
let body: String = row.get(3)?;
|
|
let msg_type: String = row.get(4)?;
|
|
let ref_msg_id: Option<Vec<u8>> = row.get(5)?;
|
|
let timestamp_ms: u64 = row.get(6)?;
|
|
let is_outgoing: i32 = row.get(7)?;
|
|
|
|
Ok(StoredMessage {
|
|
conversation_id: conv_id.clone(),
|
|
message_id: message_id.as_deref().and_then(to_16),
|
|
sender_key,
|
|
sender_name,
|
|
body,
|
|
msg_type,
|
|
ref_msg_id: ref_msg_id.as_deref().and_then(to_16),
|
|
timestamp_ms,
|
|
is_outgoing: is_outgoing != 0,
|
|
})
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[allow(clippy::unwrap_used)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn open_test_store() -> (tempfile::TempDir, ConversationStore) {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let db_path = dir.path().join("test.db");
|
|
let store = ConversationStore::open(&db_path, None).unwrap();
|
|
(dir, store)
|
|
}
|
|
|
|
fn make_group_conv(name: &str, activity_ms: u64) -> Conversation {
|
|
Conversation {
|
|
id: ConversationId::from_group_name(name),
|
|
kind: ConversationKind::Group { name: name.to_string() },
|
|
display_name: format!("#{name}"),
|
|
mls_group_blob: None,
|
|
keystore_blob: None,
|
|
member_keys: vec![vec![1, 2, 3]],
|
|
unread_count: 0,
|
|
last_activity_ms: activity_ms,
|
|
created_at_ms: 1000,
|
|
is_hybrid: false,
|
|
last_seen_seq: 0,
|
|
}
|
|
}
|
|
|
|
fn make_dm_conv(peer_key: &[u8], peer_name: Option<&str>) -> Conversation {
|
|
let mut id_bytes = [0u8; 16];
|
|
id_bytes[..peer_key.len().min(16)].copy_from_slice(&peer_key[..peer_key.len().min(16)]);
|
|
Conversation {
|
|
id: ConversationId(id_bytes),
|
|
kind: ConversationKind::Dm {
|
|
peer_key: peer_key.to_vec(),
|
|
peer_username: peer_name.map(|s| s.to_string()),
|
|
},
|
|
display_name: peer_name.unwrap_or("unknown").to_string(),
|
|
mls_group_blob: None,
|
|
keystore_blob: None,
|
|
member_keys: vec![peer_key.to_vec()],
|
|
unread_count: 0,
|
|
last_activity_ms: 2000,
|
|
created_at_ms: 1000,
|
|
is_hybrid: false,
|
|
last_seen_seq: 0,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn save_and_load_group() {
|
|
let (_dir, store) = open_test_store();
|
|
let conv = make_group_conv("engineering", 5000);
|
|
store.save_conversation(&conv).unwrap();
|
|
|
|
let loaded = store.load_conversation(&conv.id).unwrap().unwrap();
|
|
assert_eq!(loaded.display_name, "#engineering");
|
|
assert_eq!(loaded.last_activity_ms, 5000);
|
|
match &loaded.kind {
|
|
ConversationKind::Group { name } => assert_eq!(name, "engineering"),
|
|
_ => panic!("expected Group kind"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn save_and_load_dm() {
|
|
let (_dir, store) = open_test_store();
|
|
let peer_key = vec![10u8; 32];
|
|
let conv = make_dm_conv(&peer_key, Some("alice"));
|
|
store.save_conversation(&conv).unwrap();
|
|
|
|
let loaded = store.load_conversation(&conv.id).unwrap().unwrap();
|
|
assert_eq!(loaded.display_name, "alice");
|
|
match &loaded.kind {
|
|
ConversationKind::Dm { peer_key: pk, peer_username } => {
|
|
assert_eq!(pk, &peer_key);
|
|
assert_eq!(peer_username.as_deref(), Some("alice"));
|
|
}
|
|
_ => panic!("expected Dm kind"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn find_dm_by_peer() {
|
|
let (_dir, store) = open_test_store();
|
|
let peer_key = vec![20u8; 32];
|
|
let conv = make_dm_conv(&peer_key, Some("bob"));
|
|
store.save_conversation(&conv).unwrap();
|
|
|
|
let found = store.find_dm_by_peer(&peer_key).unwrap().unwrap();
|
|
assert_eq!(found.id, conv.id);
|
|
|
|
let missing = store.find_dm_by_peer(&[99u8; 32]).unwrap();
|
|
assert!(missing.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn list_conversations_ordering() {
|
|
let (_dir, store) = open_test_store();
|
|
let c1 = make_group_conv("old-group", 1000);
|
|
let c2 = make_group_conv("new-group", 3000);
|
|
let c3 = make_group_conv("mid-group", 2000);
|
|
store.save_conversation(&c1).unwrap();
|
|
store.save_conversation(&c2).unwrap();
|
|
store.save_conversation(&c3).unwrap();
|
|
|
|
let list = store.list_conversations().unwrap();
|
|
assert_eq!(list.len(), 3);
|
|
// Most recent activity first
|
|
assert_eq!(list[0].last_activity_ms, 3000);
|
|
assert_eq!(list[1].last_activity_ms, 2000);
|
|
assert_eq!(list[2].last_activity_ms, 1000);
|
|
}
|
|
|
|
#[test]
|
|
fn save_and_load_messages() {
|
|
let (_dir, store) = open_test_store();
|
|
let conv = make_group_conv("chat", 1000);
|
|
store.save_conversation(&conv).unwrap();
|
|
|
|
for i in 0..5 {
|
|
store.save_message(&StoredMessage {
|
|
conversation_id: conv.id.clone(),
|
|
message_id: None,
|
|
sender_key: vec![1, 2, 3],
|
|
sender_name: Some("alice".to_string()),
|
|
body: format!("message {i}"),
|
|
msg_type: "chat".to_string(),
|
|
ref_msg_id: None,
|
|
timestamp_ms: 1000 + i as u64,
|
|
is_outgoing: i % 2 == 0,
|
|
}).unwrap();
|
|
}
|
|
|
|
let msgs = store.load_recent_messages(&conv.id, 3).unwrap();
|
|
assert_eq!(msgs.len(), 3);
|
|
// Should be in chronological order (reversed from DESC)
|
|
assert_eq!(msgs[0].body, "message 2");
|
|
assert_eq!(msgs[1].body, "message 3");
|
|
assert_eq!(msgs[2].body, "message 4");
|
|
}
|
|
|
|
#[test]
|
|
fn outbox_enqueue_and_mark_sent() {
|
|
let (_dir, store) = open_test_store();
|
|
let conv_id = ConversationId([1u8; 16]);
|
|
let recipient = vec![5u8; 32];
|
|
let payload = b"encrypted-payload";
|
|
|
|
store.enqueue_outbox(&conv_id, &recipient, payload).unwrap();
|
|
store.enqueue_outbox(&conv_id, &recipient, b"second").unwrap();
|
|
|
|
let pending = store.load_pending_outbox().unwrap();
|
|
assert_eq!(pending.len(), 2);
|
|
assert_eq!(store.count_pending_outbox().unwrap(), 2);
|
|
|
|
store.mark_outbox_sent(pending[0].id).unwrap();
|
|
assert_eq!(store.count_pending_outbox().unwrap(), 1);
|
|
|
|
let remaining = store.load_pending_outbox().unwrap();
|
|
assert_eq!(remaining.len(), 1);
|
|
assert_eq!(remaining[0].payload, b"second");
|
|
}
|
|
|
|
#[test]
|
|
fn conversation_id_from_group_name_determinism() {
|
|
let a = ConversationId::from_group_name("test-group");
|
|
let b = ConversationId::from_group_name("test-group");
|
|
assert_eq!(a, b);
|
|
|
|
let c = ConversationId::from_group_name("other-group");
|
|
assert_ne!(a, c);
|
|
}
|
|
|
|
#[test]
|
|
fn conversation_id_from_slice_wrong_length() {
|
|
assert!(ConversationId::from_slice(&[0u8; 15]).is_none());
|
|
assert!(ConversationId::from_slice(&[0u8; 17]).is_none());
|
|
assert!(ConversationId::from_slice(&[]).is_none());
|
|
assert!(ConversationId::from_slice(&[0u8; 16]).is_some());
|
|
}
|
|
|
|
#[test]
|
|
fn block_unblock_user() {
|
|
let (_dir, store) = open_test_store();
|
|
let ik = vec![42u8; 32];
|
|
|
|
assert!(!store.is_blocked(&ik).unwrap());
|
|
|
|
store.block_user(&ik, "spam").unwrap();
|
|
assert!(store.is_blocked(&ik).unwrap());
|
|
|
|
let blocked = store.list_blocked().unwrap();
|
|
assert_eq!(blocked.len(), 1);
|
|
assert_eq!(blocked[0].identity_key, ik);
|
|
assert_eq!(blocked[0].reason, "spam");
|
|
|
|
let removed = store.unblock_user(&ik).unwrap();
|
|
assert!(removed);
|
|
assert!(!store.is_blocked(&ik).unwrap());
|
|
assert!(store.list_blocked().unwrap().is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn block_user_idempotent() {
|
|
let (_dir, store) = open_test_store();
|
|
let ik = vec![42u8; 32];
|
|
|
|
store.block_user(&ik, "first").unwrap();
|
|
store.block_user(&ik, "updated").unwrap();
|
|
|
|
let blocked = store.list_blocked().unwrap();
|
|
assert_eq!(blocked.len(), 1);
|
|
assert_eq!(blocked[0].reason, "updated");
|
|
}
|
|
|
|
#[test]
|
|
fn unblock_nonexistent_returns_false() {
|
|
let (_dir, store) = open_test_store();
|
|
let removed = store.unblock_user(&[99u8; 32]).unwrap();
|
|
assert!(!removed);
|
|
}
|
|
|
|
#[test]
|
|
fn filtered_messages_exclude_blocked() {
|
|
let (_dir, store) = open_test_store();
|
|
let conv = make_group_conv("modchat", 1000);
|
|
store.save_conversation(&conv).unwrap();
|
|
|
|
let alice_key = vec![1u8; 32];
|
|
let bob_key = vec![2u8; 32];
|
|
|
|
// Alice sends 2 messages, Bob sends 1.
|
|
store.save_message(&StoredMessage {
|
|
conversation_id: conv.id.clone(),
|
|
message_id: None,
|
|
sender_key: alice_key.clone(),
|
|
sender_name: Some("alice".to_string()),
|
|
body: "hello from alice".to_string(),
|
|
msg_type: "chat".to_string(),
|
|
ref_msg_id: None,
|
|
timestamp_ms: 1000,
|
|
is_outgoing: false,
|
|
}).unwrap();
|
|
|
|
store.save_message(&StoredMessage {
|
|
conversation_id: conv.id.clone(),
|
|
message_id: None,
|
|
sender_key: bob_key.clone(),
|
|
sender_name: Some("bob".to_string()),
|
|
body: "hello from bob".to_string(),
|
|
msg_type: "chat".to_string(),
|
|
ref_msg_id: None,
|
|
timestamp_ms: 2000,
|
|
is_outgoing: false,
|
|
}).unwrap();
|
|
|
|
store.save_message(&StoredMessage {
|
|
conversation_id: conv.id.clone(),
|
|
message_id: None,
|
|
sender_key: alice_key.clone(),
|
|
sender_name: Some("alice".to_string()),
|
|
body: "another from alice".to_string(),
|
|
msg_type: "chat".to_string(),
|
|
ref_msg_id: None,
|
|
timestamp_ms: 3000,
|
|
is_outgoing: false,
|
|
}).unwrap();
|
|
|
|
// All 3 messages unfiltered.
|
|
let all = store.load_recent_messages(&conv.id, 10).unwrap();
|
|
assert_eq!(all.len(), 3);
|
|
|
|
// Block alice.
|
|
store.block_user(&alice_key, "spam").unwrap();
|
|
|
|
// Filtered: only bob's message.
|
|
let filtered = store.load_recent_messages_filtered(&conv.id, 10).unwrap();
|
|
assert_eq!(filtered.len(), 1);
|
|
assert_eq!(filtered[0].body, "hello from bob");
|
|
|
|
// Unblock alice — all messages visible again.
|
|
store.unblock_user(&alice_key).unwrap();
|
|
let unblocked = store.load_recent_messages_filtered(&conv.id, 10).unwrap();
|
|
assert_eq!(unblocked.len(), 3);
|
|
}
|
|
|
|
#[test]
|
|
fn clear_failed_outbox_entries() {
|
|
let (_dir, store) = open_test_store();
|
|
let conv_id = ConversationId([2u8; 16]);
|
|
let recipient = vec![5u8; 32];
|
|
|
|
store.enqueue_outbox(&conv_id, &recipient, b"msg1").unwrap();
|
|
store.enqueue_outbox(&conv_id, &recipient, b"msg2").unwrap();
|
|
|
|
let entries = store.load_pending_outbox().unwrap();
|
|
assert_eq!(entries.len(), 2);
|
|
|
|
// Mark first as permanently failed (retry_count > 10).
|
|
store.mark_outbox_failed(entries[0].id, 11).unwrap();
|
|
// Mark second as retryable failure.
|
|
store.mark_outbox_failed(entries[1].id, 3).unwrap();
|
|
|
|
// Only 1 pending (the retryable one).
|
|
assert_eq!(store.count_pending_outbox().unwrap(), 1);
|
|
|
|
// Clear failed entries.
|
|
let cleared = store.clear_failed_outbox().unwrap();
|
|
assert_eq!(cleared, 1);
|
|
|
|
// Still 1 pending.
|
|
assert_eq!(store.count_pending_outbox().unwrap(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn update_and_get_last_seen_seq() {
|
|
let (_dir, store) = open_test_store();
|
|
let conv = make_group_conv("seq-test", 1000);
|
|
store.save_conversation(&conv).unwrap();
|
|
|
|
// Initially 0.
|
|
assert_eq!(store.get_last_seen_seq(&conv.id).unwrap(), 0);
|
|
|
|
// Update to 5.
|
|
store.update_last_seen_seq(&conv.id, 5).unwrap();
|
|
assert_eq!(store.get_last_seen_seq(&conv.id).unwrap(), 5);
|
|
|
|
// Update to 10 — should work.
|
|
store.update_last_seen_seq(&conv.id, 10).unwrap();
|
|
assert_eq!(store.get_last_seen_seq(&conv.id).unwrap(), 10);
|
|
|
|
// Update to 7 — should be a no-op (only increases).
|
|
store.update_last_seen_seq(&conv.id, 7).unwrap();
|
|
assert_eq!(store.get_last_seen_seq(&conv.id).unwrap(), 10);
|
|
}
|
|
|
|
#[test]
|
|
fn get_last_seen_seq_missing_conversation() {
|
|
let (_dir, store) = open_test_store();
|
|
let missing = ConversationId([99u8; 16]);
|
|
// Returns 0 for unknown conversations.
|
|
assert_eq!(store.get_last_seen_seq(&missing).unwrap(), 0);
|
|
}
|
|
}
|