A call in a room, between everybody in it
#81 gave Matrix calls of two, which is what `m.call.*` was written for: one offer, one answer, one other person. A room with three people in it had no call at all. A group call has no invitation - it is a piece of room state saying people are in one, which is how anybody learns there is a call to join. So `m.call.member` is published on joining, refreshed while still here, and withdrawn on leaving; it carries an expiry because a client that dies mid-call never withdraws anything, and a membership past its stamp is somebody's crash rather than a participant. The mesh underneath is one connection per pair. Both ends see each other arrive, so the rule for who calls is a property of the pair rather than of who was first - "first" is not something two clients can agree on - and the smaller key calls the larger, computed the same way on both sides. The permission is the part that would otherwise fail silently: a membership is a state event, and a room's default is that only moderators may send those. So a call started in an ordinary room is one nobody else can join, and they learn it from a bare "not authorized". Starting a call now lowers that one event to what everybody has, where this account may say so - and where it may not, the refusal explains itself. Closes #122. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Claude Opus 5
@@ -111,3 +111,179 @@ pub async fn turn_servers(state: &AppState, account_id: &str) -> anyhow::Result<
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// How long a published membership stands before it is treated as stale.
|
||||
///
|
||||
/// A client that is killed mid-call never withdraws its membership, so the
|
||||
/// state event would say somebody is in a call for ever. Every client that
|
||||
/// implements MSC3401 stamps an expiry and refreshes it while it is still
|
||||
/// there; anything past its expiry is somebody's crash, not a participant.
|
||||
pub const MEMBERSHIP_TTL_MS: i64 = 90_000;
|
||||
|
||||
/// The event that says who is in a room's call.
|
||||
pub const EVENT_MEMBER: &str = "m.call.member";
|
||||
|
||||
/// Says this account is in the room's call, or is no longer.
|
||||
///
|
||||
/// One state event per user, keyed by user id, holding a list of memberships -
|
||||
/// one per device, since the same person may be in a call from a phone and a
|
||||
/// desktop and only one of those should stop when the other leaves.
|
||||
///
|
||||
/// Withdrawing sends the remaining memberships rather than empty content, for
|
||||
/// the same reason: another device of this account may still be in the call.
|
||||
pub async fn set_membership(
|
||||
state: &AppState,
|
||||
account_id: &str,
|
||||
buffer_id: &str,
|
||||
device_id: &str,
|
||||
joined: bool,
|
||||
) -> anyhow::Result<()> {
|
||||
let account = state.accounts.get_matrix(account_id).ok_or_else(|| anyhow::anyhow!("account not connected"))?;
|
||||
let room_id = state
|
||||
.runtime
|
||||
.get_matrix_room(buffer_id)
|
||||
.ok_or_else(|| anyhow::anyhow!("no known Matrix room for this buffer"))?;
|
||||
let now = chrono::Utc::now().timestamp_millis();
|
||||
let mut memberships: Vec<Value> = state
|
||||
.runtime
|
||||
.matrix_call_members(account_id, &room_id)
|
||||
.into_iter()
|
||||
.filter(|m| m["user_id"].as_str() == Some(account.user_id.as_str()))
|
||||
.filter_map(|m| m["membership"].clone().into())
|
||||
// This device's own entry is rewritten below either way.
|
||||
.filter(|m: &Value| m["device_id"].as_str() != Some(device_id))
|
||||
.filter(|m: &Value| m["expires_ts"].as_i64().unwrap_or(0) > now)
|
||||
.collect();
|
||||
if joined {
|
||||
memberships.push(json!({
|
||||
"application": "m.call",
|
||||
"call_id": "",
|
||||
"scope": "m.room",
|
||||
"device_id": device_id,
|
||||
"expires_ts": now + MEMBERSHIP_TTL_MS,
|
||||
// No focus: this is a mesh between the people in the room rather
|
||||
// than a conference on somebody's server. See the module note.
|
||||
"foci_active": [],
|
||||
}));
|
||||
}
|
||||
// Whether everybody in the room may join a call at all.
|
||||
//
|
||||
// A membership is a *state* event, and a room's default is that only
|
||||
// moderators may send those - so a call started in an ordinary room is
|
||||
// one nobody else can join, and they find out with a bare "not
|
||||
// authorized". Where this account can change the permissions, it lowers
|
||||
// that one event to what everybody has; where it cannot, the call still
|
||||
// works for whoever may already send it.
|
||||
if joined {
|
||||
allow_everybody_to_join(state, account_id, &room_id).await;
|
||||
}
|
||||
|
||||
let result = super::put_room_state(
|
||||
state,
|
||||
account_id,
|
||||
&room_id,
|
||||
EVENT_MEMBER,
|
||||
&account.user_id,
|
||||
json!({ "memberships": memberships }),
|
||||
)
|
||||
.await;
|
||||
if let Err(e) = &result {
|
||||
// The refusal that has an explanation worth giving.
|
||||
if e.to_string().contains("M_FORBIDDEN") {
|
||||
anyhow::bail!(
|
||||
"this room does not let its members join calls - somebody who can change the room's \
|
||||
permissions has to allow the m.call.member event"
|
||||
);
|
||||
}
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
/// Lets ordinary members join this room's calls, where we may say so.
|
||||
///
|
||||
/// One key in the power levels: `events["m.call.member"] = 0`. Without it the
|
||||
/// room's `state_default` applies, which is 50 - a moderator - and a call in
|
||||
/// an ordinary room is a call of one.
|
||||
///
|
||||
/// Best-effort and silent on refusal: somebody without the power to change
|
||||
/// permissions can still be in a call, and failing to start one because the
|
||||
/// room could not be reconfigured would be worse than a call only some people
|
||||
/// can join.
|
||||
async fn allow_everybody_to_join(state: &AppState, account_id: &str, room_id: &str) {
|
||||
let Some(levels) = state.runtime.matrix_power_levels(account_id, room_id) else { return };
|
||||
if levels["events"][EVENT_MEMBER].as_i64() == Some(0) {
|
||||
return;
|
||||
}
|
||||
let mut next = levels.clone();
|
||||
next["events"][EVENT_MEMBER] = json!(0);
|
||||
match super::put_room_state(state, account_id, room_id, "m.room.power_levels", "", next).await {
|
||||
Ok(()) => tracing::info!("matrix[{account_id}]: {room_id} now lets its members join calls"),
|
||||
Err(e) => tracing::debug!("matrix[{account_id}]: cannot open {room_id} to calls: {e:#}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Reads one `m.call.member` state event into the memberships it carries.
|
||||
///
|
||||
/// Tolerant, and expiry-aware: a membership past its stamp is somebody whose
|
||||
/// client died, and treating it as a participant would leave a tile in the
|
||||
/// grid for somebody who left hours ago.
|
||||
pub fn read_memberships(user_id: &str, content: &Value, now_ms: i64) -> Vec<Value> {
|
||||
content["memberships"]
|
||||
.as_array()
|
||||
.map(|list| {
|
||||
list.iter()
|
||||
.filter(|m| m["application"].as_str().unwrap_or("m.call") == "m.call")
|
||||
.filter(|m| m["expires_ts"].as_i64().unwrap_or(0) > now_ms)
|
||||
.map(|m| json!({ "user_id": user_id, "membership": m }))
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Who to call, and who to wait for.
|
||||
///
|
||||
/// Both ends of every pair see each other arrive, and if both call, both
|
||||
/// answer, and the call collides with itself. So the rule is one line and the
|
||||
/// same on both sides: the smaller id calls the larger. It has to be a
|
||||
/// property of the pair rather than of who arrived first, because "first" is
|
||||
/// not something two clients can agree on.
|
||||
pub fn should_offer(own_key: &str, their_key: &str) -> bool {
|
||||
own_key < their_key
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod member_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn a_membership_past_its_stamp_is_not_a_participant() {
|
||||
let content = json!({ "memberships": [
|
||||
{ "application": "m.call", "device_id": "HERE", "expires_ts": 2_000 },
|
||||
{ "application": "m.call", "device_id": "GONE", "expires_ts": 500 },
|
||||
]});
|
||||
let live = read_memberships("@a:example.org", &content, 1_000);
|
||||
assert_eq!(live.len(), 1);
|
||||
assert_eq!(live[0]["membership"]["device_id"], "HERE");
|
||||
assert_eq!(live[0]["user_id"], "@a:example.org");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn something_that_is_not_a_call_is_not_read_as_one() {
|
||||
let content = json!({ "memberships": [
|
||||
{ "application": "m.whiteboard", "device_id": "X", "expires_ts": 9_000 },
|
||||
]});
|
||||
assert!(read_memberships("@a:example.org", &content, 1_000).is_empty());
|
||||
assert!(read_memberships("@a:example.org", &json!({}), 1_000).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn exactly_one_end_of_every_pair_offers() {
|
||||
// Whichever way round the two clients ask, they must not agree.
|
||||
assert!(should_offer("@a:example.org|AAA", "@b:example.org|BBB"));
|
||||
assert!(!should_offer("@b:example.org|BBB", "@a:example.org|AAA"));
|
||||
// Including two devices of the same person, which is the case that
|
||||
// makes a user-id comparison alone wrong.
|
||||
assert!(should_offer("@a:example.org|AAA", "@a:example.org|ZZZ"));
|
||||
assert!(!should_offer("@a:example.org|ZZZ", "@a:example.org|AAA"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,6 +55,28 @@ pub async fn process_state_events(state: &AppState, account_id: &str, room_id: &
|
||||
);
|
||||
}
|
||||
}
|
||||
// Who is in this room's call. A state event per person, and the
|
||||
// one thing a client needs to know before it can join one: a
|
||||
// group call has no invitation, only people already in it.
|
||||
"m.call.member" => {
|
||||
let who = event["state_key"].as_str().unwrap_or_default();
|
||||
let live = crate::backend::matrix::calls::read_memberships(
|
||||
who,
|
||||
&event["content"],
|
||||
chrono::Utc::now().timestamp_millis(),
|
||||
);
|
||||
let everybody = state.runtime.set_matrix_call_members(account_id, room_id, who, live);
|
||||
if let Some(buffer_id) = state.runtime.matrix_buffer_for_room(account_id, room_id) {
|
||||
state.events.emit(
|
||||
"matrixCallMembers",
|
||||
serde_json::json!({
|
||||
"accountId": account_id,
|
||||
"bufferId": buffer_id,
|
||||
"members": everybody,
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
"m.room.avatar" => {
|
||||
if let Some(mxc) = event["content"]["url"].as_str() {
|
||||
if let Some(path) = cached_media_path(homeserver_url, access_token, mxc, "").await {
|
||||
|
||||
@@ -311,6 +311,28 @@ pub async fn dispatch(
|
||||
// read something new.
|
||||
backend::matrix::replay_read_receipts(state, buffer_id);
|
||||
|
||||
// A call already happening in this room. Replayed for the
|
||||
// same reason the roster is: it is state rather than news, so
|
||||
// a window opened after it started would show no sign of it
|
||||
// until somebody joined or left - which on a call of two
|
||||
// people already talking is never.
|
||||
if let Some(room_id) = state.runtime.get_matrix_room(buffer_id) {
|
||||
let account_id = state.runtime.get_buffer(buffer_id).map(|b| b.account_id).unwrap_or_default();
|
||||
let now = chrono::Utc::now().timestamp_millis();
|
||||
let members: Vec<serde_json::Value> = state
|
||||
.runtime
|
||||
.matrix_call_members(&account_id, &room_id)
|
||||
.into_iter()
|
||||
.filter(|m| m["membership"]["expires_ts"].as_i64().unwrap_or(0) > now)
|
||||
.collect();
|
||||
if !members.is_empty() {
|
||||
state.events.emit(
|
||||
"matrixCallMembers",
|
||||
serde_json::json!({ "accountId": account_id, "bufferId": buffer_id, "members": members }),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// And what a Kick channel is broadcasting, which changes while
|
||||
// nobody is looking: a viewer count from an hour ago is worse
|
||||
// than none, so opening the channel asks again.
|
||||
@@ -3623,6 +3645,59 @@ pub async fn dispatch(
|
||||
}
|
||||
}
|
||||
|
||||
// Joining a room's call, or leaving it. The membership is state in
|
||||
// the room, so everybody else in it learns there is a call to join.
|
||||
"setMatrixCallMembership" => {
|
||||
let Some(buffer_id) = p_str_opt(params, "bufferId") else {
|
||||
return (None, Some("setMatrixCallMembership requires \"bufferId\"".to_string()));
|
||||
};
|
||||
let joined = params.get("joined").and_then(|v| v.as_bool()).unwrap_or(true);
|
||||
let Some(buffer) = state.runtime.get_buffer(buffer_id) else {
|
||||
return (None, Some("no such buffer".to_string()));
|
||||
};
|
||||
let Some(account) = state.accounts.get_matrix(&buffer.account_id) else {
|
||||
return (None, Some("account not connected".to_string()));
|
||||
};
|
||||
match backend::matrix::calls::set_membership(state, &buffer.account_id, buffer_id, &account.device_id, joined).await {
|
||||
Ok(()) => (
|
||||
Some(serde_json::json!({
|
||||
"userId": account.user_id,
|
||||
"deviceId": account.device_id,
|
||||
// The room, which is what both ends of a call have in
|
||||
// common: each account names the same room by its own
|
||||
// buffer id, so a conference named after the buffer
|
||||
// would be two conferences that never meet.
|
||||
"roomId": state.runtime.get_matrix_room(buffer_id),
|
||||
})),
|
||||
None,
|
||||
),
|
||||
Err(e) => (None, Some(format!("{e:#}"))),
|
||||
}
|
||||
}
|
||||
|
||||
"listMatrixCallMembers" => {
|
||||
let Some(buffer_id) = p_str_opt(params, "bufferId") else {
|
||||
return (None, Some("listMatrixCallMembers requires \"bufferId\"".to_string()));
|
||||
};
|
||||
let Some(buffer) = state.runtime.get_buffer(buffer_id) else {
|
||||
return (None, Some("no such buffer".to_string()));
|
||||
};
|
||||
let Some(room_id) = state.runtime.get_matrix_room(buffer_id) else {
|
||||
return (Some(serde_json::json!([])), None);
|
||||
};
|
||||
let now = chrono::Utc::now().timestamp_millis();
|
||||
// Filtered on the way out as well as on the way in: a membership
|
||||
// read an hour ago may have expired since, and nobody refreshes
|
||||
// somebody else's.
|
||||
let live: Vec<serde_json::Value> = state
|
||||
.runtime
|
||||
.matrix_call_members(&buffer.account_id, &room_id)
|
||||
.into_iter()
|
||||
.filter(|m| m["membership"]["expires_ts"].as_i64().unwrap_or(0) > now)
|
||||
.collect();
|
||||
(Some(serde_json::json!(live)), None)
|
||||
}
|
||||
|
||||
// Not hearing from somebody, on whichever service they are on.
|
||||
//
|
||||
// One method rather than one per backend, because the person doing it
|
||||
|
||||
@@ -432,6 +432,12 @@ pub struct Runtime {
|
||||
/// refuses that line. Asking for history from a server that never granted
|
||||
/// it is a command it will answer with an error in the server tab.
|
||||
irc_caps: Mutex<HashMap<String, std::collections::HashSet<String>>>,
|
||||
/// Who is in a room's call, per account and room.
|
||||
///
|
||||
/// Read from `m.call.member` state rather than tracked here: the state is
|
||||
/// the truth, this is only the copy the client is told about without
|
||||
/// re-reading the room every time somebody asks.
|
||||
matrix_call_members: Mutex<HashMap<String, Vec<serde_json::Value>>>,
|
||||
/// Sticker packs, per account: pack key -> the images it offers.
|
||||
///
|
||||
/// One flat table rather than per-room, because a picker offers everything
|
||||
@@ -836,6 +842,7 @@ impl Runtime {
|
||||
irc_caps: Mutex::new(HashMap::new()),
|
||||
irc_monitor: Mutex::new(std::collections::HashSet::new()),
|
||||
matrix_stickers: Mutex::new(HashMap::new()),
|
||||
matrix_call_members: Mutex::new(HashMap::new()),
|
||||
matrix_rooms: Mutex::new(HashMap::new()),
|
||||
matrix_room_names: Mutex::new(HashMap::new()),
|
||||
matrix_space_parents: Mutex::new(HashMap::new()),
|
||||
@@ -2343,6 +2350,15 @@ impl Runtime {
|
||||
self.matrix_room_avatars.lock().unwrap().get(&(account_id.to_string(), room_id.to_string())).cloned()
|
||||
}
|
||||
|
||||
/// This room's power levels, as the state last said.
|
||||
pub fn matrix_power_levels(&self, account_id: &str, room_id: &str) -> Option<Value> {
|
||||
self.matrix_power_levels
|
||||
.lock()
|
||||
.unwrap()
|
||||
.get(&(account_id.to_string(), room_id.to_string()))
|
||||
.cloned()
|
||||
}
|
||||
|
||||
pub fn set_matrix_power_levels(&self, account_id: &str, room_id: &str, content: Value) {
|
||||
self.matrix_power_levels.lock().unwrap().insert((account_id.to_string(), room_id.to_string()), content);
|
||||
}
|
||||
@@ -2919,6 +2935,35 @@ impl Runtime {
|
||||
self.irc_caps.lock().unwrap().remove(account_id);
|
||||
}
|
||||
|
||||
/// Everybody currently in this room's call, as the state last said.
|
||||
pub fn matrix_call_members(&self, account_id: &str, room_id: &str) -> Vec<serde_json::Value> {
|
||||
self.matrix_call_members
|
||||
.lock()
|
||||
.unwrap()
|
||||
.get(&format!("{account_id}|{room_id}"))
|
||||
.cloned()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Replaces one person's memberships in a room, and answers with everybody
|
||||
/// who is in the call afterwards.
|
||||
///
|
||||
/// Per user because that is how the state is keyed: one event per person,
|
||||
/// listing the devices of theirs that are in the call.
|
||||
pub fn set_matrix_call_members(
|
||||
&self,
|
||||
account_id: &str,
|
||||
room_id: &str,
|
||||
user_id: &str,
|
||||
memberships: Vec<serde_json::Value>,
|
||||
) -> Vec<serde_json::Value> {
|
||||
let mut all = self.matrix_call_members.lock().unwrap();
|
||||
let room = all.entry(format!("{account_id}|{room_id}")).or_default();
|
||||
room.retain(|m| m["user_id"].as_str() != Some(user_id));
|
||||
room.extend(memberships);
|
||||
room.clone()
|
||||
}
|
||||
|
||||
/// Records one sticker pack, replacing whatever was under that key.
|
||||
///
|
||||
/// An empty pack removes the key rather than leaving an empty group in
|
||||
|
||||
Reference in New Issue
Block a user