buzzert/Kordophone
Private
Public Access
1
0
Fork 0
Code Packages Activity

Don't overwrite already resolved participants, better naming of keys

Browse Source
This commit is contained in:
James Magahern
2025-06-26 18:23:15 -07:00
parent bb19db17cd
commit f6bb1a9b57
25 changed files with 263 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
kordophone-db/src/models/conversation.rs
View File

@@ -75,7 +75,7 @@ impl From<kordophone::model::Conversation> for Conversation {
participants: value
.participant_display_names
.into_iter()
.map(|p| p.into())
.map(|p| Participant::Remote { handle: p, contact_id: None }) // todo: this is wrong
.collect(),
}
}

13
kordophone-db/src/models/db/message.rs
View File

@@ -7,7 +7,7 @@ use diesel::prelude::*;
#[diesel(check_for_backend(diesel::sqlite::Sqlite))]
pub struct Record {
pub id: String,
pub sender_participant_id: Option<i32>,
pub sender_participant_handle: Option<String>,
pub text: String,
pub date: NaiveDateTime,
pub file_transfer_guids: Option<String>, // JSON array
@@ -28,9 +28,9 @@ impl From<Message> for Record {
Self {
id: message.id,
sender_participant_id: match message.sender {
sender_participant_handle: match message.sender {
Participant::Me => None,
Participant::Remote { id, .. } => id,
Participant::Remote { handle, .. } => Some(handle),
},
text: message.text,
date: message.date,
@@ -51,10 +51,13 @@ impl From<Record> for Message {
.attachment_metadata
.and_then(|json| serde_json::from_str(&json).ok());
let message_sender = match record.sender_participant_handle {
Some(handle) => Participant::Remote { handle, contact_id: None },
None => Participant::Me,
};
Self {
id: record.id,
// We'll set the proper sender later when loading participant info
sender: Participant::Me,
sender: message_sender,
text: record.text,
date: record.date,
file_transfer_guids,

29
kordophone-db/src/models/db/participant.rs
View File

@@ -4,9 +4,9 @@ use diesel::prelude::*;
#[derive(Queryable, Selectable, AsChangeset, Identifiable)]
#[diesel(table_name = crate::schema::participants)]
#[diesel(primary_key(handle))]
pub struct Record {
pub id: i32,
pub display_name: Option<String>,
pub handle: String,
pub is_me: bool,
pub contact_id: Option<String>,
}
@@ -14,7 +14,7 @@ pub struct Record {
#[derive(Insertable)]
#[diesel(table_name = crate::schema::participants)]
pub struct InsertableRecord {
pub display_name: Option<String>,
pub handle: String,
pub is_me: bool,
pub contact_id: Option<String>,
}
@@ -23,12 +23,12 @@ impl From<Participant> for InsertableRecord {
fn from(participant: Participant) -> Self {
match participant {
Participant::Me => InsertableRecord {
display_name: None,
handle: "me".to_string(),
is_me: true,
contact_id: None,
},
Participant::Remote { display_name, contact_id, .. } => InsertableRecord {
display_name: Some(display_name),
Participant::Remote { handle, contact_id, .. } => InsertableRecord {
handle,
is_me: false,
contact_id,
},
@@ -38,12 +38,12 @@ impl From<Participant> for InsertableRecord {
#[derive(Identifiable, Selectable, Queryable, Associations, Debug)]
#[diesel(belongs_to(super::conversation::Record, foreign_key = conversation_id))]
#[diesel(belongs_to(Record, foreign_key = participant_id))]
#[diesel(belongs_to(Record, foreign_key = participant_handle))]
#[diesel(table_name = conversation_participants)]
#[diesel(primary_key(conversation_id, participant_id))]
#[diesel(primary_key(conversation_id, participant_handle))]
pub struct ConversationParticipant {
pub conversation_id: String,
pub participant_id: i32,
pub participant_handle: String,
}
impl From<Record> for Participant {
@@ -52,8 +52,7 @@ impl From<Record> for Participant {
Participant::Me
} else {
Participant::Remote {
id: Some(record.id),
display_name: record.display_name.unwrap_or_default(),
handle: record.handle.clone(),
contact_id: record.contact_id,
}
}
@@ -64,14 +63,12 @@ impl From<Participant> for Record {
fn from(participant: Participant) -> Self {
match participant {
Participant::Me => Record {
id: 0, // This will be set by the database
display_name: None,
handle: "me".to_string(),
is_me: true,
contact_id: None,
},
Participant::Remote { display_name, contact_id, .. } => Record {
id: 0, // This will be set by the database
display_name: Some(display_name),
Participant::Remote { handle, contact_id, .. } => Record {
handle,
is_me: false,
contact_id,
},

3
kordophone-db/src/models/message.rs
View File

@@ -27,8 +27,7 @@ impl From<kordophone::model::Message> for Message {
id: value.guid,
sender: match value.sender {
Some(sender) => Participant::Remote {
id: None,
display_name: sender,
handle: sender,
contact_id: None,
},
None => Participant::Me,

2
kordophone-db/src/models/mod.rs
View File

@@ -5,4 +5,4 @@ pub mod participant;
pub use conversation::Conversation;
pub use message::Message;
pub use participant::Participant;
pub use participant::Participant;

32
kordophone-db/src/models/participant.rs
View File

@@ -2,37 +2,21 @@
pub enum Participant {
Me,
Remote {
id: Option<i32>,
display_name: String,
handle: String,
contact_id: Option<String>,
},
}
impl From<String> for Participant {
fn from(display_name: String) -> Self {
Participant::Remote {
id: None,
display_name,
contact_id: None,
}
}
}
impl From<&str> for Participant {
fn from(display_name: &str) -> Self {
Participant::Remote {
id: None,
display_name: display_name.to_string(),
contact_id: None,
}
}
}
impl Participant {
pub fn display_name(&self) -> String {
pub fn handle(&self) -> String {
match self {
Participant::Me => "(Me)".to_string(),
Participant::Remote { display_name, .. } => display_name.clone(),
Participant::Remote { handle, .. } => handle.clone(),
}
}
// Temporary alias for backward compatibility
pub fn display_name(&self) -> String {
self.handle()
}
}

114
kordophone-db/src/repository.rs
View File

@@ -36,21 +36,19 @@ impl<'a> Repository<'a> {
.values(&db_conversation)
.execute(self.connection)?;
diesel::replace_into(participants)
.values(&db_participants)
.execute(self.connection)?;
for participant in &db_participants {
diesel::insert_into(participants)
.values(participant)
.on_conflict_do_nothing()
.execute(self.connection)?;
}
// Sqlite backend doesn't support batch insert, so we have to do this manually
for participant in db_participants {
let pid = participants
.select(schema::participants::id)
.filter(schema::participants::display_name.eq(&participant.display_name))
.first::<i32>(self.connection)?;
for participant in &db_participants {
diesel::replace_into(conversation_participants)
.values((
conversation_id.eq(&db_conversation.id),
participant_id.eq(pid),
participant_handle.eq(&participant.handle),
))
.execute(self.connection)?;
}
@@ -117,7 +115,7 @@ impl<'a> Repository<'a> {
// Handle participant if message has a remote sender
let sender = message.sender.clone();
let mut db_message: MessageRecord = message.into();
db_message.sender_participant_id = self.get_or_create_participant(&sender);
db_message.sender_participant_handle = self.get_or_create_participant(&sender);
diesel::replace_into(messages)
.values(&db_message)
@@ -161,11 +159,11 @@ impl<'a> Repository<'a> {
// individual queries.
self.connection
.transaction::<_, diesel::result::Error, _>(|conn| {
// Cache participant ids we have already looked up / created – in a
// Cache participant handles we have already looked up / created – in a
// typical conversation we only have a handful of participants, but we
// might be processing hundreds of messages. Avoiding an extra SELECT per
// message saves a lot of round-trips to SQLite.
let mut participant_cache: HashMap<String, i32> = HashMap::new();
let mut participant_cache: HashMap<String, String> = HashMap::new();
// Prepare collections for the batch inserts.
let mut db_messages: Vec<MessageRecord> = Vec::with_capacity(in_messages.len());
@@ -174,50 +172,40 @@ impl<'a> Repository<'a> {
for message in in_messages {
// Resolve/insert the sender participant only once per display name.
let sender_id = match &message.sender {
let sender_handle_opt = match &message.sender {
Participant::Me => None,
Participant::Remote { display_name, .. } => {
if let Some(cached_participant_id) = participant_cache.get(display_name)
{
Some(*cached_participant_id)
Participant::Remote { handle, contact_id } => {
if participant_cache.contains_key(handle) {
Some(handle.clone())
} else {
// Try to load from DB first
let existing: Option<i32> = participants_dsl::participants
.filter(participants_dsl::display_name.eq(display_name))
.select(participants_dsl::id)
.first::<i32>(conn)
// Ensure participant exists in DB
let exists: Option<String> = participants_dsl::participants
.filter(participants_dsl::handle.eq(handle))
.select(participants_dsl::handle)
.first::<String>(conn)
.optional()?;
let participant_id = if let Some(pid) = existing {
pid
} else {
if exists.is_none() {
let new_participant = InsertableParticipantRecord {
display_name: Some(display_name.clone()),
handle: handle.clone(),
is_me: false,
contact_id: None,
contact_id: contact_id.clone(),
};
diesel::insert_into(participants_dsl::participants)
.values(&new_participant)
.execute(conn)?;
}
// last_insert_rowid() is connection-wide, but we are the only
// writer inside this transaction.
diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
"last_insert_rowid()",
))
.get_result::<i32>(conn)?
};
participant_cache.insert(display_name.clone(), participant_id);
Some(participant_id)
participant_cache.insert(handle.clone(), handle.clone());
Some(handle.clone())
}
}
};
// Convert the message into its DB form.
let mut db_message: MessageRecord = message.into();
db_message.sender_participant_id = sender_id;
db_message.sender_participant_handle = sender_handle_opt.clone();
conv_msg_records.push(InsertableConversationMessage {
conversation_id: conversation_guid.to_string(),
@@ -280,10 +268,10 @@ impl<'a> Repository<'a> {
for message_record in message_records {
let mut message: Message = message_record.clone().into();
// If there's a sender_participant_id, load the participant info
if let Some(pid) = message_record.sender_participant_id {
// If the message references a sender participant, load the participant info
if let Some(sender_handle) = message_record.sender_participant_handle {
let participant = participants
.find(pid)
.find(sender_handle)
.first::<ParticipantRecord>(self.connection)?;
message.sender = participant.into();
}
@@ -374,50 +362,44 @@ impl<'a> Repository<'a> {
/// Update the contact_id for an existing participant record.
pub fn update_participant_contact(
&mut self,
participant_db_id: i32,
participant_handle: &str,
new_contact_id: &str,
) -> Result<()> {
use crate::schema::participants::dsl::*;
log::debug!(target: target::REPOSITORY, "Updating participant contact id {} => {}", participant_db_id, new_contact_id);
diesel::update(participants.filter(id.eq(participant_db_id)))
log::debug!(target: target::REPOSITORY, "Updating participant contact {} => {}", participant_handle, new_contact_id);
diesel::update(participants.filter(handle.eq(participant_handle)))
.set(contact_id.eq(Some(new_contact_id.to_string())))
.execute(self.connection)?;
Ok(())
}
fn get_or_create_participant(&mut self, participant: &Participant) -> Option<i32> {
fn get_or_create_participant(&mut self, participant: &Participant) -> Option<String> {
match participant {
Participant::Me => None,
Participant::Remote {
display_name: p_name,
contact_id: c_id,
..
} => {
Participant::Remote { handle: p_handle, contact_id: c_id, .. } => {
use crate::schema::participants::dsl::*;
let existing_participant = participants
.filter(display_name.eq(p_name))
.filter(handle.eq(p_handle))
.first::<ParticipantRecord>(self.connection)
.optional()
.unwrap();
if let Some(participant) = existing_participant {
return Some(participant.id);
if existing_participant.is_none() {
let participant_record = InsertableParticipantRecord {
handle: p_handle.clone(),
is_me: false,
contact_id: c_id.clone(),
};
diesel::insert_into(participants)
.values(&participant_record)
.execute(self.connection)
.unwrap();
}
let participant_record = InsertableParticipantRecord {
display_name: Some(participant.display_name()),
is_me: false,
contact_id: c_id.clone(),
};
diesel::insert_into(participants)
.values(&participant_record)
.execute(self.connection)
.unwrap();
self.last_insert_id().ok()
Some(p_handle.clone())
}
}
}

25
kordophone-db/src/schema.rs
View File

@@ -12,18 +12,17 @@ diesel::table! {
}
diesel::table! {
participants (id) {
id -> Integer,
display_name -> Nullable<Text>,
participants (handle) {
handle -> Text,
is_me -> Bool,
contact_id -> Nullable<Text>,
}
}
diesel::table! {
conversation_participants (conversation_id, participant_id) {
conversation_participants (conversation_id, participant_handle) {
conversation_id -> Text,
participant_id -> Integer,
participant_handle -> Text,
}
}
@@ -31,7 +30,7 @@ diesel::table! {
messages (id) {
id -> Text, // guid
text -> Text,
sender_participant_id -> Nullable<Integer>,
sender_participant_handle -> Nullable<Text>,
date -> Timestamp,
file_transfer_guids -> Nullable<Text>, // JSON array of file transfer GUIDs
attachment_metadata -> Nullable<Text>, // JSON string of attachment metadata
@@ -53,13 +52,15 @@ diesel::table! {
}
diesel::joinable!(conversation_participants -> conversations (conversation_id));
diesel::joinable!(conversation_participants -> participants (participant_id));
diesel::joinable!(conversation_participants -> participants (participant_handle));
diesel::joinable!(messages -> participants (sender_participant_handle));
diesel::joinable!(conversation_messages -> conversations (conversation_id));
diesel::joinable!(conversation_messages -> messages (message_id));
diesel::allow_tables_to_appear_in_same_query!(
conversations,
participants,
conversation_participants
conversation_participants,
messages,
conversation_messages,
settings,
);
diesel::joinable!(conversation_messages -> conversations (conversation_id));
diesel::joinable!(conversation_messages -> messages (message_id));
diesel::allow_tables_to_appear_in_same_query!(conversations, messages, conversation_messages);

35
kordophone-db/src/tests/mod.rs
View File

@@ -12,14 +12,8 @@ fn participants_equal_ignoring_id(a: &Participant, b: &Participant) -> bool {
match (a, b) {
(Participant::Me, Participant::Me) => true,
(
Participant::Remote {
display_name: name_a,
..
},
Participant::Remote {
display_name: name_b,
..
},
Participant::Remote { handle: name_a, .. },
Participant::Remote { handle: name_b, .. },
) => name_a == name_b,
_ => false,
}
@@ -29,9 +23,14 @@ fn participants_vec_equal_ignoring_id(a: &[Participant], b: &[Participant]) -> b
if a.len() != b.len() {
return false;
}
a.iter()
.zip(b.iter())
.all(|(a, b)| participants_equal_ignoring_id(a, b))
// For each participant in a, check if there is a matching participant in b
a.iter().all(|a_participant| {
b.iter().any(|b_participant| participants_equal_ignoring_id(a_participant, b_participant))
}) &&
// Also check the reverse to ensure no duplicates
b.iter().all(|b_participant| {
a.iter().any(|a_participant| participants_equal_ignoring_id(b_participant, a_participant))
})
}
#[tokio::test]
@@ -214,8 +213,8 @@ async fn test_messages() {
// Verify second message (from Alice)
let retrieved_message2 = messages.iter().find(|m| m.id == message2.id).unwrap();
assert_eq!(retrieved_message2.text, "Hi there!");
if let Participant::Remote { display_name, .. } = &retrieved_message2.sender {
assert_eq!(display_name, "Alice");
if let Participant::Remote { handle, .. } = &retrieved_message2.sender {
assert_eq!(handle, "Alice");
} else {
panic!(
"Expected Remote participant. Got: {:?}",
@@ -345,14 +344,8 @@ async fn test_insert_messages_batch() {
match (&original.sender, &retrieved.sender) {
(Participant::Me, Participant::Me) => {}
(
Participant::Remote {
display_name: o_name,
..
},
Participant::Remote {
display_name: r_name,
..
},
Participant::Remote { handle: o_name, .. },
Participant::Remote { handle: r_name, .. },
) => assert_eq!(o_name, r_name),
_ => panic!(
"Sender mismatch: original {:?}, retrieved {:?}",