Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-04-17 21:46:35 +03:00
Code Issues Packages Projects Releases Wiki Activity

Add calls table to store SDPs separately from message params

Browse Source 
Co-authored-by: link2xt <18373967+link2xt@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2025-11-06 20:06:47 +00:00
parent abece73db1
commit c556b07380
4 changed files with 142 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
src/calls.rs
View File

@@ -195,9 +195,24 @@ impl Context {
text: "Outgoing call".into(),
..Default::default()
};
// Set a placeholder parameter so the message is recognized as a call
// This will be used by mimefactory until the DB entry is available
call.param.set(Param::WebrtcRoom, &place_call_info);
call.id = send_msg(self, chat_id, &mut call).await?;
// Store SDP offer in calls table and remove from params
self.sql
.execute(
"INSERT OR REPLACE INTO calls (msg_id, offer_sdp) VALUES (?, ?)",
(call.id, &place_call_info),
)
.await?;
// Remove SDP from message params for privacy
call.param.remove(Param::WebrtcRoom);
call.update_param(self).await?;
let wait = RINGING_SECONDS;
task::spawn(Context::emit_end_call_if_unaccepted(
self.clone(),
@@ -229,6 +244,14 @@ impl Context {
chat.id.accept(self).await?;
}
// Store SDP answer in calls table
self.sql
.execute(
"UPDATE calls SET answer_sdp=? WHERE msg_id=?",
(&accept_call_info, call_id),
)
.await?;
// send an acceptance message around: to the caller as well as to the other devices of the callee
let mut msg = Message {
viewtype: Viewtype::Text,
@@ -237,8 +260,6 @@ impl Context {
};
msg.param.set_cmd(SystemMessage::CallAccepted);
msg.hidden = true;
msg.param
.set(Param::WebrtcAccepted, accept_call_info.to_string());
msg.set_quote(self, Some(&call.msg)).await?;
msg.id = send_msg(self, call.msg.chat_id, &mut msg).await?;
self.emit_event(EventType::IncomingCallAccepted {
@@ -327,6 +348,16 @@ impl Context {
from_id: ContactId,
) -> Result<()> {
if mime_message.is_call() {
// Extract SDP from message headers and store in calls table
if let Some(offer_sdp) = mime_message.get_header(HeaderDef::ChatWebrtcRoom) {
self.sql
.execute(
"INSERT OR IGNORE INTO calls (msg_id, offer_sdp) VALUES (?, ?)",
(call_id, offer_sdp),
)
.await?;
}
let Some(call) = self.load_call_by_id(call_id).await? else {
warn!(self, "{call_id} does not refer to a call message");
return Ok(());
@@ -391,6 +422,16 @@ impl Context {
return Ok(());
}
// Store SDP answer in calls table
if let Some(answer_sdp) = mime_message.get_header(HeaderDef::ChatWebrtcAccepted) {
self.sql
.execute(
"UPDATE calls SET answer_sdp=? WHERE msg_id=?",
(answer_sdp, call.msg.id),
)
.await?;
}
call.mark_as_accepted(self).await?;
self.emit_msgs_changed(call.msg.chat_id, call_id);
if call.is_incoming() {
@@ -463,33 +504,40 @@ impl Context {
/// not a call message, returns `None`.
pub async fn load_call_by_id(&self, call_id: MsgId) -> Result<Option<CallInfo>> {
let call = Message::load_from_db(self, call_id).await?;
Ok(self.load_call_by_message(call))
self.load_call_by_message(call).await
}
// Loads information about the call given the `Message`.
//
// If the `Message` is not a call message, returns `None`
fn load_call_by_message(&self, call: Message) -> Option<CallInfo> {
async fn load_call_by_message(&self, call: Message) -> Result<Option<CallInfo>> {
if call.viewtype != Viewtype::Call {
// This can happen e.g. if a "call accepted"
// or "call ended" message is received
// with `In-Reply-To` referring to non-call message.
return None;
return Ok(None);
}
Some(CallInfo {
place_call_info: call
.param
.get(Param::WebrtcRoom)
.unwrap_or_default()
.to_string(),
accept_call_info: call
.param
.get(Param::WebrtcAccepted)
.unwrap_or_default()
.to_string(),
// Load SDP from calls table
let (place_call_info, accept_call_info) = self
.sql
.query_row_optional(
"SELECT offer_sdp, answer_sdp FROM calls WHERE msg_id=?",
(call.id,),
|row| {
let offer: Option<String> = row.get(0)?;
let answer: Option<String> = row.get(1)?;
Ok((offer.unwrap_or_default(), answer.unwrap_or_default()))
},
)
.await?
.unwrap_or_default();
Ok(Some(CallInfo {
place_call_info,
accept_call_info,
msg: call,
})
}))
}
}

50
src/mimefactory.rs
View File

@@ -1680,6 +1680,27 @@ impl MimeFactory {
"Chat-Content",
mail_builder::headers::raw::Raw::new("call-accepted").into(),
));
// Get SDP answer from the referenced call message in calls table,
// or fall back to params if not yet migrated
if let Some(ref quoted_msg_id) = msg.in_reply_to {
let answer_sdp = context
.sql
.query_row_optional(
"SELECT answer_sdp FROM calls WHERE msg_id=?",
(quoted_msg_id,),
|row| row.get::<_, Option<String>>(0),
)
.await?
.flatten()
.or_else(|| msg.param.get(Param::WebrtcAccepted).map(|s| s.to_string()));
if let Some(answer_sdp) = answer_sdp {
headers.push((
"Chat-Webrtc-Accepted",
mail_builder::headers::raw::Raw::new(b_encode(&answer_sdp)).into(),
));
}
}
}
SystemMessage::CallEnded => {
headers.push((
@@ -1716,16 +1737,25 @@ impl MimeFactory {
);
}
if let Some(offer) = msg.param.get(Param::WebrtcRoom) {
headers.push((
"Chat-Webrtc-Room",
mail_builder::headers::raw::Raw::new(b_encode(offer)).into(),
));
} else if let Some(answer) = msg.param.get(Param::WebrtcAccepted) {
headers.push((
"Chat-Webrtc-Accepted",
mail_builder::headers::raw::Raw::new(b_encode(answer)).into(),
));
if msg.viewtype == Viewtype::Call {
// Get SDP offer from calls table, or fall back to params if not yet migrated
let offer_sdp = context
.sql
.query_row_optional(
"SELECT offer_sdp FROM calls WHERE msg_id=?",
(msg.id,),
|row| row.get::<_, Option<String>>(0),
)
.await?
.flatten()
.or_else(|| msg.param.get(Param::WebrtcRoom).map(|s| s.to_string()));
if let Some(offer_sdp) = offer_sdp {
headers.push((
"Chat-Webrtc-Room",
mail_builder::headers::raw::Raw::new(b_encode(&offer_sdp)).into(),
));
}
}
if msg.viewtype == Viewtype::Voice

22
src/sql.rs
View File

@@ -921,6 +921,28 @@ pub async fn housekeeping(context: &Context) -> Result<()> {
.log_err(context)
.ok();
// Delete call SDPs for ended calls (older than 24 hours) or orphaned calls.
// Ended calls have Param::Arg4 (H=timestamp) set in their params.
// We clean up calls that ended more than 24 hours ago to protect privacy
// as SDPs contain IP addresses.
context
.sql
.execute(
"DELETE FROM calls WHERE msg_id IN (
SELECT calls.msg_id FROM calls
LEFT JOIN msgs ON calls.msg_id = msgs.id
WHERE msgs.id IS NULL
OR msgs.chat_id = ?
OR (msgs.param LIKE '%H=%'
AND msgs.timestamp_sent < ?)
)",
(DC_CHAT_ID_TRASH, time() - 86400),
)
.await
.context("Failed to delete old call SDPs")
.log_err(context)
.ok();
info!(context, "Housekeeping done.");
Ok(())
}

15
src/sql/migrations.rs
View File

@@ -1339,6 +1339,21 @@ CREATE INDEX gossip_timestamp_index ON gossip_timestamp (chat_id, fingerprint);
.await?;
}
inc_and_check(&mut migration_version, 139)?;
if dbversion < migration_version {
sql.execute_migration(
"CREATE TABLE calls(
id INTEGER PRIMARY KEY AUTOINCREMENT,
msg_id INTEGER NOT NULL UNIQUE,
offer_sdp TEXT,
answer_sdp TEXT
);
CREATE INDEX calls_msg_id_index ON calls (msg_id);",
migration_version,
)
.await?;
}
let new_version = sql
.get_raw_config_int(VERSION_CFG)
.await?