Join the calls Element is actually in

moho could see an Element call and not be in it: Element's media goes
through a LiveKit media server and moho only knew how to mesh. Now it
does both, and which one a call uses is the call's own answer rather than
a setting.

Finding the server, in the order that answers best: the call itself,
since everybody has to be where the media already is and that is what
makes joining somebody else's call possible; then this account's own
setting, for a homeserver that publishes nothing - Element carries the
same fallback for the same reason; then the homeserver's `.well-known`,
which is where a server set up for calls says so.

Getting in is an OpenID token, not an access token. The service beside
the SFU has no way to check a Matrix access token, so the homeserver
vouches for the account in a form a third party can verify by asking the
homeserver back, and that is exchanged for a URL and a JWT good for one
room. Asked for before joining, so a refusal fails the join rather than
leaving this account listed in a call it never reached.

The membership then says `livekit` rather than `moho.mesh`, which is what
makes it legible to Element: its clients read the transports and connect
to the one they can use.

The mesh stays for rooms with no server at all, which is most self-hosted
homeservers, and is what MSC4143's pluggable transports are for.

Co-Authored-By: Claude Opus 5 <[email protected]>

src/accounts.rs

@@ -199,6 +199,15 @@ pub struct MatrixAccountConfig {
     pub next_batch: Option<String>,
     #[serde(default)]
     pub display_name: Option<String>,
+    /// A media server to hold calls on, where the homeserver names none.
+    ///
+    /// A Matrix room call runs either between the people in it or through an
+    /// SFU, and which one is a property of the homeserver: a server set up for
+    /// calls publishes `org.matrix.msc4143.rtc_foci` in its `.well-known`.
+    /// Plenty do not, so this is the same manual fallback Element carries -
+    /// name a LiveKit JWT service here and calls go through it.
+    #[serde(default)]
+    pub rtc_focus_url: Option<String>,
 }
 
 impl MatrixAccountConfig {
@@ -605,6 +614,23 @@ impl AccountStore {
         self.mutate(account_id, |a| a.notify = nicks_csv.to_string())
     }
 
+    /// Names the media server this account's room calls go through.
+    ///
+    /// Empty clears it, which puts calls back on the mesh between the people
+    /// in the room - that being what works with no infrastructure at all.
+    pub fn set_matrix_rtc_focus(&self, account_id: &str, url: &str) -> Result<bool> {
+        let mut matrix = self.matrix.lock().unwrap();
+        match matrix.get_mut(account_id) {
+            None => Ok(false),
+            Some(a) => {
+                let url = url.trim();
+                a.rtc_focus_url = if url.is_empty() { None } else { Some(url.to_string()) };
+                self.persist(&self.irc.lock().unwrap(), &self.discord.lock().unwrap(), &self.sneedchat.lock().unwrap(), &matrix, &self.kick.lock().unwrap())?;
+                Ok(true)
+            }
+        }
+    }
+
     pub fn set_nickserv_password(&self, account_id: &str, password: &str) -> Result<bool> {
         self.mutate(account_id, |a| {
             a.nickserv_password = if password.is_empty() { None } else { Some(password.to_string()) }

src/backend/matrix/backup.rs

@@ -265,6 +265,7 @@ mod tests {
             device_id: login.device_id.clone(),
             next_batch: None,
             display_name: None,
+            rtc_focus_url: None,
         };
         let saved = state.accounts.add_matrix(config).expect("add_matrix failed");
         let account_id = saved.account_id();

src/backend/matrix/calls.rs

@@ -23,6 +23,7 @@
 
 use serde_json::{json, Value};
 
+use anyhow::Context;
 use crate::state::AppState;
 
 /// Events a client has to see to be one end of a call.
@@ -170,6 +171,8 @@ pub async fn set_membership(
     buffer_id: &str,
     device_id: &str,
     joined: bool,
+    // The media server this end is on, where it is on one.
+    focus: Option<Value>,
 ) -> anyhow::Result<()> {
     let account = state.accounts.get_matrix(account_id).ok_or_else(|| anyhow::anyhow!("account not connected"))?;
     let room_id = state
@@ -198,12 +201,13 @@ pub async fn set_membership(
             "membershipID": format!("{}:{}", account.user_id, device_id),
             "created_ts": now,
             "expires": MEMBERSHIP_TTL_MS,
-            // What this end can actually be reached on. Not a LiveKit focus,
-            // because moho has no SFU to offer and claiming one would send
-            // everybody else to dial nothing; a client that does not know
-            // this transport ignores it, which is the honest outcome.
-            "focus_active": { "type": TRANSPORT_MESH },
-            "foci_preferred": [{ "type": TRANSPORT_MESH }],
+            // What this end is actually on. A media server where the call
+            // has one - which is what makes this membership legible to
+            // Element, since that is the only transport its clients speak -
+            // and the mesh where there is none, which a client that does not
+            // know it ignores rather than dialling nothing.
+            "focus_active": focus.clone().unwrap_or_else(|| json!({ "type": TRANSPORT_MESH })),
+            "foci_preferred": [focus.clone().unwrap_or_else(|| json!({ "type": TRANSPORT_MESH }))],
         })
     } else {
         // Leaving is an empty membership rather than a deleted event, because
@@ -293,10 +297,8 @@ pub fn read_membership(user_id: &str, content: &Value, now_ms: i64) -> Option<Va
     if expires_ts <= now_ms {
         return None;
     }
-    let transports: Vec<String> = content["foci_preferred"]
-        .as_array()
-        .map(|list| list.iter().filter_map(|t| t["type"].as_str().map(String::from)).collect())
-        .unwrap_or_default();
+    let foci = content["foci_preferred"].as_array().cloned().unwrap_or_default();
+    let transports: Vec<String> = foci.iter().filter_map(|t| t["type"].as_str().map(String::from)).collect();
     Some(json!({
         "user_id": user_id,
         "membership": {
@@ -304,20 +306,90 @@ pub fn read_membership(user_id: &str, content: &Value, now_ms: i64) -> Option<Va
             "expires_ts": expires_ts,
         },
         "transports": transports,
+        // The media server this participant is on, where they are on one:
+        // everybody in a call has to be where the media already is, so this
+        // is what a client joining afterwards has to use.
+        "focus": foci.iter().find(|f| f["type"].as_str() == Some("livekit")).cloned().unwrap_or(Value::Null),
     }))
 }
 
-/// What a call needs from whoever wants to join it.
+/// Where the media server is, if there is one to use.
 ///
-/// Read from the participants rather than decided here: a call where anybody
-/// offers a LiveKit focus is a call held on that server, and one where
-/// everybody is on the mesh is a call held between the people in it. moho can
-/// be in the second kind and not yet the first, and the difference has to be
-/// said rather than discovered by joining and hearing nothing.
-pub fn needs_a_focus(members: &[Value]) -> bool {
-    members
+/// Three places, in the order that answers the question best:
+///
+/// 1. The call itself. A call already running names its focus in every
+///    membership, and everybody has to be where the media already is - this
+///    is what makes joining *Element's* call possible rather than starting a
+///    second one beside it.
+/// 2. This account's own setting, for a homeserver that publishes nothing.
+///    Element has the same fallback for the same reason.
+/// 3. The homeserver's `.well-known`, which is where a server that has been
+///    set up for calls says so.
+pub async fn find_focus(state: &AppState, account_id: &str, room_id: &str) -> Option<Value> {
+    let members = state.runtime.matrix_call_members(account_id, room_id);
+    if let Some(focus) = members.iter().find_map(|m| m["focus"].as_object()) {
+        return Some(Value::Object(focus.clone()));
+    }
+    let account = state.accounts.get_matrix(account_id)?;
+    if let Some(url) = account.rtc_focus_url.as_deref().filter(|u| !u.trim().is_empty()) {
+        return Some(json!({ "type": "livekit", "livekit_service_url": url.trim() }));
+    }
+    // The homeserver's own answer. Fetched rather than remembered: it changes
+    // when somebody sets a server up for calls, which is exactly when a
+    // client that cached "none" would be wrong.
+    let base = account.homeserver_url.trim_end_matches('/');
+    let host = base.split("://").nth(1).unwrap_or(base).split('/').next().unwrap_or(base);
+    let well_known = super::http::get_json(&format!("https://{host}/.well-known/matrix/client"), "").await.ok()?;
+    well_known["org.matrix.msc4143.rtc_foci"]
+        .as_array()?
         .iter()
-        .any(|m| m["transports"].as_array().is_some_and(|t| t.iter().any(|t| t.as_str() == Some("livekit"))))
+        .find(|f| f["type"].as_str() == Some("livekit"))
+        .cloned()
+}
+
+/// Trades this account's identity for a way into the media server.
+///
+/// The server beside a LiveKit SFU does not take a Matrix access token - it
+/// would have no way to check one. It takes an OpenID token, which is the
+/// homeserver saying "this really is who they claim to be" in a form a third
+/// party can verify by asking the homeserver back. That is exchanged for a
+/// LiveKit URL and a JWT good for one room.
+pub async fn rtc_token(state: &AppState, account_id: &str, room_id: &str, focus: &Value) -> anyhow::Result<Value> {
+    let account = state.accounts.get_matrix(account_id).ok_or_else(|| anyhow::anyhow!("account not connected"))?;
+    let base = account.homeserver_url.trim_end_matches('/');
+    let openid = super::http::post_json(
+        &format!(
+            "{base}/_matrix/client/v3/user/{}/openid/request_token",
+            url::form_urlencoded::byte_serialize(account.user_id.as_bytes()).collect::<String>()
+        ),
+        Some(&account.access_token),
+        json!({}),
+    )
+    .await
+    .context("asking the homeserver to vouch for this account")?;
+
+    let service = focus["livekit_service_url"]
+        .as_str()
+        .ok_or_else(|| anyhow::anyhow!("that call's media server has no address"))?
+        .trim_end_matches('/');
+    // The room as the media server knows it. The call names its own alias
+    // where it has one, since everybody has to land in the same LiveKit room.
+    let alias = focus["livekit_alias"].as_str().unwrap_or(room_id);
+    let answer = super::http::post_json(
+        &format!("{service}/sfu/get"),
+        None,
+        json!({ "room": alias, "openid_token": openid, "device_id": account.device_id }),
+    )
+    .await
+    .context("asking the media server for a way in")?;
+    if answer["jwt"].as_str().unwrap_or_default().is_empty() {
+        anyhow::bail!("the media server would not let this account in");
+    }
+    Ok(json!({
+        "url": answer["url"],
+        "jwt": answer["jwt"],
+        "identity": format!("{}:{}", account.user_id, account.device_id),
+    }))
 }
 
 /// Who to call, and who to wait for./// Who to call, and who to wait for.
@@ -375,6 +447,17 @@ mod member_tests {
         assert!(read_membership("@them:example.org", &json!({}), 1_000).is_none());
     }
 
+    #[test]
+    fn a_membership_carries_the_media_server_it_is_on() {
+        let read = read_membership("@them:example.org", &element_membership(1_000, 9_000), 2_000).unwrap();
+        // Everybody in a call has to be where the media already is, so this
+        // is what a client joining afterwards connects to.
+        assert_eq!(read["focus"]["livekit_service_url"], "https://livekit-jwt.call.matrix.org");
+        let mesh = json!({ "application": "m.call", "device_id": "D", "created_ts": 1_000, "expires": 9_000,
+                           "foci_preferred": [{ "type": "moho.mesh" }] });
+        assert!(read_membership("@a:example.org", &mesh, 2_000).unwrap()["focus"].is_null());
+    }
+
     #[test]
     fn the_newer_shape_is_read_too() {
         let content = json!({
@@ -394,17 +477,6 @@ mod member_tests {
         assert!(read_membership("@a:example.org", &content, 1_000).is_none());
     }
 
-    #[test]
-    fn a_call_with_a_focus_in_it_is_not_a_mesh() {
-        let livekit = read_membership("@them:example.org", &element_membership(1_000, 9_000), 2_000).unwrap();
-        let mesh = json!({ "transports": ["moho.mesh"] });
-        assert!(needs_a_focus(&[livekit.clone()]));
-        assert!(!needs_a_focus(&[mesh.clone()]));
-        // One person on an SFU makes it an SFU call: everybody has to be
-        // where the media is.
-        assert!(needs_a_focus(&[mesh, livekit]));
-    }
-
     #[test]
     fn the_state_key_is_the_one_element_writes() {
         // Prefixed on an ordinary room, because a key starting with `@` is

src/backend/matrix/mod.rs

@@ -3724,7 +3724,8 @@ async fn try_sso_login(state: &AppState, login_id: &str, homeserver_url: &str, p
         device_id: login.device_id,
         next_batch: None,
         display_name: None,
-    };
+            rtc_focus_url: None,
+        };
     let saved = state.accounts.add_matrix(config)?;
     let account = crate::accounts::matrix_account_to_json(&saved, "connecting", false);
     spawn(state.clone(), saved);
@@ -3813,7 +3814,8 @@ async fn try_login(state: &AppState, login_id: &str, homeserver_url: &str, usern
         device_id: String::new(),
         next_batch: None,
         display_name: None,
-    };
+            rtc_focus_url: None,
+        };
     state.accounts.add_matrix(pending.clone())?;
 
     let login = auth::login(homeserver_url, username, password, None).await.context("logging in")?;
@@ -3826,7 +3828,8 @@ async fn try_login(state: &AppState, login_id: &str, homeserver_url: &str, usern
         device_id: login.device_id,
         next_batch: None,
         display_name: None,
-    };
+            rtc_focus_url: None,
+        };
     let saved = state.accounts.add_matrix(config)?;
     // The placeholder above was keyed on whatever was typed, and the server
     // answers with the canonical user id - "salastil" against "@salastil:
@@ -4035,6 +4038,7 @@ mod tests {
             device_id: login.device_id.clone(),
             next_batch: None,
             display_name: None,
+            rtc_focus_url: None,
         };
         let saved = state.accounts.add_matrix(config).expect("add_matrix failed");
         let account_id = saved.account_id();
@@ -4152,6 +4156,7 @@ mod tests {
             device_id: login.device_id.clone(),
             next_batch: None,
             display_name: None,
+            rtc_focus_url: None,
         };
         let saved = state.accounts.add_matrix(config).expect("add_matrix failed");
         let account_id = saved.account_id();
@@ -4263,6 +4268,7 @@ mod tests {
             device_id: login_a.device_id.clone(),
             next_batch: None,
             display_name: None,
+            rtc_focus_url: None,
         };
         let saved_a = state.accounts.add_matrix(config_a).expect("add_matrix failed");
         let account_id_a = saved_a.account_id();
@@ -4419,6 +4425,7 @@ mod tests {
             device_id: login.device_id.clone(),
             next_batch: None,
             display_name: None,
+            rtc_focus_url: None,
         };
         let saved = state.accounts.add_matrix(config).expect("add_matrix failed");
         let account_id = saved.account_id();
@@ -4574,6 +4581,7 @@ mod tests {
             device_id: login.device_id.clone(),
             next_batch: None,
             display_name: None,
+            rtc_focus_url: None,
         };
         let saved = state.accounts.add_matrix(config).expect("add_matrix failed");
         let account_id = saved.account_id();

src/backend/matrix/verification.rs

@@ -547,6 +547,7 @@ mod tests {
             device_id: login.device_id.clone(),
             next_batch: None,
             display_name: None,
+            rtc_focus_url: None,
         };
         let saved = state.accounts.add_matrix(config).expect("add_matrix failed");
         let account_id = saved.account_id();

src/rpc/methods.rs

@@ -3658,30 +3658,42 @@ pub async fn dispatch(
             let Some(account) = state.accounts.get_matrix(&buffer.account_id) else {
                 return (None, Some("account not connected".to_string()));
             };
-            // A call held on somebody's SFU is not one moho can be in yet:
-            // its media never goes peer to peer, and joining it would put
-            // this account in the participant list and send nothing. Said
-            // before joining rather than discovered by silence afterwards.
-            if joined {
-                if let Some(room_id) = state.runtime.get_matrix_room(buffer_id) {
-                    let members = state.runtime.matrix_call_members(&buffer.account_id, &room_id);
-                    if backend::matrix::calls::needs_a_focus(&members) {
-                        return (
-                            None,
-                            Some(
-                                concat!(
-                                    "this call runs through a media server, which is what Element's own ",
-                                    "calls use - moho cannot join one yet. It can hold a call between the ",
-                                    "people in the room, which is what it offers where no media server is ",
-                                    "in use."
-                                )
-                                .to_string(),
-                            ),
-                        );
+            let Some(room_id) = state.runtime.get_matrix_room(buffer_id) else {
+                return (None, Some("that conversation is not a Matrix room".to_string()));
+            };
+
+            // Which kind of call this is. A media server where the call
+            // already runs on one, or where this account has one to offer;
+            // otherwise a mesh between the people in the room. Decided before
+            // joining, because it decides what the membership says.
+            let focus = if joined {
+                backend::matrix::calls::find_focus(state, &buffer.account_id, &room_id).await
+            } else {
+                None
+            };
+            // And the way in, which is a token rather than a promise: asked
+            // for here so that failing to get one fails the join rather than
+            // leaving this account listed in a call it never reached.
+            let media = match (&focus, joined) {
+                (Some(focus), true) => {
+                    match backend::matrix::calls::rtc_token(state, &buffer.account_id, &room_id, focus).await {
+                        Ok(token) => Some(token),
+                        Err(e) => return (None, Some(format!("{e:#}"))),
                     }
                 }
-            }
-            match backend::matrix::calls::set_membership(state, &buffer.account_id, buffer_id, &account.device_id, joined).await {
+                _ => None,
+            };
+
+            match backend::matrix::calls::set_membership(
+                state,
+                &buffer.account_id,
+                buffer_id,
+                &account.device_id,
+                joined,
+                focus.clone(),
+            )
+            .await
+            {
                 Ok(()) => (
                     Some(serde_json::json!({
                         "userId": account.user_id,
@@ -3690,7 +3702,10 @@ pub async fn dispatch(
                         // 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),
+                        "roomId": room_id,
+                        // Where the media goes. Absent means the mesh.
+                        "media": media,
+                        "encrypted": state.runtime.is_matrix_room_encrypted(buffer_id),
                     })),
                     None,
                 ),
@@ -3698,6 +3713,13 @@ pub async fn dispatch(
             }
         }
 
+        // Where this account's room calls go, for a homeserver that names no
+        // media server of its own.
+        "setMatrixRtcFocus" => match p_str_opt(params, "accountId") {
+            None => (None, Some("setMatrixRtcFocus requires \"accountId\"".to_string())),
+            Some(id) => account_mutation_result(state.accounts.set_matrix_rtc_focus(id, p_str(params, "url", ""))),
+        },
+
         "listMatrixCallMembers" => {
             let Some(buffer_id) = p_str_opt(params, "bufferId") else {
                 return (None, Some("listMatrixCallMembers requires \"bufferId\"".to_string()));