Compare commits

...

2 Commits

Author SHA1 Message Date
cel 🌸 b81f7f5bb4 feat: order chats by most recent message 2025-03-06 10:43:57 +00:00
cel 🌸 9baf682466 feat(luz): initial message timestamp support 2025-03-03 06:11:47 +00:00
10 changed files with 52 additions and 30 deletions

View File

@ -4,9 +4,9 @@
## TODO:
- [ ] how to know if stanza has been sent
- [x] how to know if stanza has been sent
- [ ] error states for all negotiation parts
- [ ] better errors
- [x] better errors
- [x] rename structs
- [x] remove commented code
- [ ] asynchronous connect (with take_mut?)
@ -16,7 +16,7 @@
### specs:
- [x] rfc 6120: core
- [ ] rfc 6121: im
- [x] rfc 6121: im
- [x] rfc 7590: tls
- [x] xep-0368: srv records for xmpp over tls
- [ ] server side downgrade protection for sasl

22
TODO.md
View File

@ -3,18 +3,7 @@
## next
feat(luz): everything in rfc6120 and rfc6121
feat(luz): handle_online
feat(luz): handle_offline
feat(luz): handle_stanza
feat(luz): database
feat(luz): error handling on stream according to rfc6120
feat(luz): send message
feat(luz): receive message
feat(luz): retreive messages stored in database
feat(luz): get roster (online and offline)
feat(luz): set roster
feat(luz): reconnect supervisorcommand
feat: thiserror everywhere
feat(luz): proper stanza ids
test: proper tests
ci: doc generation
@ -35,3 +24,14 @@ feature: sasl
feature: resource binding
feature: jabber client connection
feature: jid
feat: thiserror everywhere
feat(luz): handle_online
feat(luz): handle_offline
feat(luz): handle_stanza
feat(luz): database
feat(luz): send message
feat(luz): receive message
feat(luz): retreive messages stored in database
feat(luz): get roster (online and offline)
feat(luz): set roster
feat(luz): reconnect supervisorcommand

View File

@ -11,7 +11,7 @@ async-recursion = "1.0.4"
async-trait = "0.1.68"
lazy_static = "1.4.0"
nanoid = "0.4.0"
# TODO: remove unneeded features
# TODO: remove unneeded features and dependencies
rsasl = { version = "2.0.1", default_features = false, features = [
"provider_base64",
"plain",

View File

@ -8,7 +8,7 @@ futures = "0.3.31"
jabber = { version = "0.1.0", path = "../jabber" }
peanuts = { version = "0.1.0", path = "../../peanuts" }
jid = { version = "0.1.0", path = "../jid", features = ["sqlx"] }
sqlx = { version = "0.8.3", features = ["sqlite", "runtime-tokio", "uuid"] }
sqlx = { version = "0.8.3", features = ["sqlite", "runtime-tokio", "uuid", "chrono"] }
stanza = { version = "0.1.0", path = "../stanza" }
tokio = "1.42.0"
tokio-stream = "0.1.17"
@ -17,3 +17,4 @@ tracing = "0.1.41"
tracing-subscriber = "0.3.19"
uuid = { version = "1.13.1", features = ["v4"] }
thiserror = "2.0.11"
chrono = "0.4.40"

View File

@ -83,6 +83,7 @@ create table messages (
-- user is the current "owner" of the message
-- TODO: queued messages offline
-- TODO: timestamp
timestamp text not null,
-- TODO: icky
-- the user to show it coming from (not necessarily the original sender)

View File

@ -1,3 +1,4 @@
use chrono::{DateTime, Utc};
use jid::JID;
use uuid::Uuid;
@ -7,6 +8,7 @@ pub struct Message {
// does not contain full user information
#[sqlx(rename = "from_jid")]
pub from: JID,
pub timestamp: DateTime<Utc>,
// TODO: originally_from
// TODO: message edits
// TODO: message timestamp
@ -29,10 +31,15 @@ pub struct Body {
#[derive(sqlx::FromRow, Debug, Clone)]
pub struct Chat {
pub correspondent: JID,
// pub unread_messages: i32,
// pub latest_message: Message,
// when a new message is received, the chat should be updated, and the new message should be delivered too.
// message history is not stored in chat, retreived separately.
// pub message_history: Vec<Message>,
}
pub enum ChatUpdate {}
impl Chat {
pub fn new(correspondent: JID) -> Self {
Self { correspondent }

View File

@ -6,6 +6,7 @@ use std::{
time::Duration,
};
use chrono::{DateTime, Utc};
use jabber::{connection::Tls, jabber_stream::bound_stream::BoundJabberReader};
use stanza::client::Stanza;
use tokio::{
@ -188,6 +189,7 @@ async fn handle_stanza(
.map(|id| Uuid::from_str(&id).unwrap_or_else(|_| Uuid::new_v4()))
.unwrap_or_else(|| Uuid::new_v4()),
from: from.clone(),
timestamp: Utc::now(),
body: Body {
// TODO: should this be an option?
body: stanza_message

View File

@ -129,12 +129,11 @@ impl Db {
}
pub(crate) async fn read_contact_opt(&self, contact: &JID) -> Result<Option<Contact>, Error> {
let contact: Option<Contact> = sqlx::query_as(
"select * from roster full outer join users on jid = user_jid where jid = ?",
)
.bind(contact)
.fetch_optional(&self.db)
.await?;
let contact: Option<Contact> =
sqlx::query_as("select * from roster join users on jid = user_jid where jid = ?")
.bind(contact)
.fetch_optional(&self.db)
.await?;
if let Some(mut contact) = contact {
#[derive(sqlx::FromRow)]
struct Row {
@ -325,6 +324,15 @@ impl Db {
Ok(chats)
}
/// chats ordered by date of last message
// greatest-n-per-group
pub(crate) async fn read_chats_ordered(&self) -> Result<Vec<Chat>, Error> {
let chats = sqlx::query_as("select c.*, m.* from chats c join (select chat_id, max(timestamp) max_timestamp from messages group by chat_id) max_timestamps on c.id = max_timestamps.chat_id join messages m on max_timestamps.chat_id = m.chat_id and max_timestamps.max_timestamp = m.timestamp order by m.timestamp desc")
.fetch_all(&self.db)
.await?;
Ok(chats)
}
async fn read_chat_id(&self, chat: JID) -> Result<Uuid, Error> {
#[derive(sqlx::FromRow)]
struct Row {
@ -358,7 +366,7 @@ impl Db {
let bare_jid = message.from.as_bare();
let resource = message.from.resourcepart;
let chat_id = self.read_chat_id(chat).await?;
sqlx::query!("insert into messages (id, body, chat_id, from_jid, from_resource) values (?, ?, ?, ?, ?)", message.id, message.body.body, chat_id, bare_jid, resource).execute(&self.db).await?;
sqlx::query!("insert into messages (id, body, chat_id, from_jid, from_resource, timestamp) values (?, ?, ?, ?, ?, ?)", message.id, message.body.body, chat_id, bare_jid, resource, message.timestamp).execute(&self.db).await?;
Ok(())
}
@ -452,10 +460,11 @@ impl Db {
// TODO: paging
pub(crate) async fn read_message_history(&self, chat: JID) -> Result<Vec<Message>, Error> {
let chat_id = self.read_chat_id(chat).await?;
let messages: Vec<Message> = sqlx::query_as("select * from messages where chat_id = ?")
.bind(chat_id)
.fetch_all(&self.db)
.await?;
let messages: Vec<Message> =
sqlx::query_as("select * from messages where chat_id = ? order by timestamp asc")
.bind(chat_id)
.fetch_all(&self.db)
.await?;
Ok(messages)
}

View File

@ -6,6 +6,7 @@ use std::{
};
use chat::{Body, Chat, Message};
use chrono::Utc;
use connection::{write::WriteMessage, SupervisorSender};
use db::Db;
use error::{
@ -322,7 +323,7 @@ impl CommandMessage {
}
}
CommandMessage::GetChats(sender) => {
let chats = db.read_chats().await.map_err(|e| e.into());
let chats = db.read_chats_ordered().await.map_err(|e| e.into());
sender.send(chats);
}
CommandMessage::GetChat(jid, sender) => {
@ -508,7 +509,7 @@ impl CommandMessage {
}
}
CommandMessage::GetChats(sender) => {
let chats = db.read_chats().await.map_err(|e| e.into());
let chats = db.read_chats_ordered().await.map_err(|e| e.into());
sender.send(chats);
}
CommandMessage::GetChat(jid, sender) => {
@ -996,6 +997,7 @@ impl CommandMessage {
id,
from: owned_jid,
body,
timestamp: Utc::now(),
};
info!("send message {:?}", message);
if let Err(e) = db

View File

@ -16,7 +16,7 @@ async fn main() {
.await
.unwrap();
let (luz, mut recv) =
LuzHandle::new("test@blos.sm".try_into().unwrap(), "slayed".to_string(), db).unwrap();
LuzHandle::new("test@blos.sm".try_into().unwrap(), "slayed".to_string(), db);
tokio::spawn(async move {
while let Some(msg) = recv.recv().await {