{who}
- {needsFocus ? ' · through a media server moho cannot join yet' : ''}
+ {onServer ? ' · on a media server' : ''}
- {!needsFocus && (
- <>
-
-
-
- )}
+
+
)
}
diff --git a/src/renderer/src/lib/livekit.ts b/src/renderer/src/lib/livekit.ts
new file mode 100644
index 0000000..ba38a4d
--- /dev/null
+++ b/src/renderer/src/lib/livekit.ts
@@ -0,0 +1,175 @@
+/**
+ * A call held on a media server, which is what Element's calls are.
+ *
+ * The mesh beside this connects everybody to everybody: fine for a handful of
+ * people, and nothing any other Matrix client speaks. Element Call sends one
+ * stream up to a LiveKit SFU and gets everybody else's back down, which costs
+ * one upload however many people are in the room - and, more to the point, is
+ * where the people using Element already are.
+ *
+ * moho is a LiveKit client here and nothing more: the daemon has already
+ * turned this account's identity into a URL and a token (see calls.rs), and
+ * what happens after that is the SFU's own protocol rather than Matrix's.
+ *
+ * Deliberately not a second implementation of the tiles. It hands back the
+ * same shape the mesh does - a list of people and their streams - so the grid,
+ * the speaking ring and the corner view do not know which kind of call they
+ * are drawing.
+ */
+
+import {
+ ConnectionState,
+ Room,
+ RoomEvent,
+ Track,
+ type RemoteTrack,
+ type RemoteTrackPublication
+} from 'livekit-client'
+
+/** One person on the media server, and what they are sending. */
+export interface SfuParticipant {
+ /** LiveKit's own identity, which the JWT service sets to `user:device`. */
+ key: string
+ userId: string
+ stream: MediaStream
+ hasVideo: boolean
+}
+
+export interface SfuHandlers {
+ /** Somebody joined, left, or started sending something different. */
+ onParticipants: (people: SfuParticipant[]) => void
+ /** Whether this end is on the server, which is what "connected" means for
+ * a call held on one - being alone in it is still being in it. */
+ onConnected: (connected: boolean) => void
+ onError: (message: string) => void
+}
+
+/**
+ * The Matrix user behind a LiveKit identity.
+ *
+ * The JWT service sets the identity to `@user:server:DEVICEID`, so the user id
+ * is everything up to the last colon - which is not the same as splitting on
+ * the first, because a Matrix id has a colon in it already.
+ */
+export function userFromIdentity(identity: string): string {
+ const cut = identity.lastIndexOf(':')
+ return cut > 0 ? identity.slice(0, cut) : identity
+}
+
+export class SfuCall {
+ private room: Room
+ private handlers: SfuHandlers
+ /** This end's own capture, so the tile shows what is being sent. */
+ local: MediaStream | null = null
+
+ constructor(handlers: SfuHandlers) {
+ this.handlers = handlers
+ this.room = new Room({
+ // Let LiveKit decide what to send as the window and the connection
+ // change: it can see the packet loss and this cannot.
+ adaptiveStream: true,
+ dynacast: true
+ })
+ const announce = (): void => this.handlers.onParticipants(this.people())
+ this.room
+ .on(RoomEvent.TrackSubscribed, announce)
+ .on(RoomEvent.TrackUnsubscribed, announce)
+ .on(RoomEvent.ParticipantConnected, announce)
+ .on(RoomEvent.ParticipantDisconnected, announce)
+ .on(RoomEvent.Disconnected, () => {
+ this.handlers.onParticipants([])
+ this.handlers.onConnected(false)
+ })
+ .on(RoomEvent.ConnectionStateChanged, (state) => {
+ // Being on the server is what connected means here, whether or not
+ // anybody else has arrived - unlike the mesh, where there is nothing
+ // to be connected *to* until somebody answers.
+ this.handlers.onConnected(state === ConnectionState.Connected)
+ announce()
+ })
+ }
+
+ /** Joins the media server and starts sending. */
+ async join(url: string, token: string, video: boolean): Promise {
+ await this.room.connect(url, token)
+ try {
+ await this.room.localParticipant.setMicrophoneEnabled(true)
+ if (video) await this.room.localParticipant.setCameraEnabled(true)
+ } catch (e) {
+ // A machine with no camera still belongs in the call.
+ this.handlers.onError(`Couldn't open the camera: ${(e as Error).message}`)
+ await this.room.localParticipant.setMicrophoneEnabled(true).catch(() => {})
+ }
+ this.local = this.ownStream()
+ this.handlers.onConnected(this.room.state === ConnectionState.Connected)
+ this.handlers.onParticipants(this.people())
+ }
+
+ /** Everybody else, with whatever they are currently sending. */
+ people(): SfuParticipant[] {
+ const out: SfuParticipant[] = []
+ for (const participant of this.room.remoteParticipants.values()) {
+ const stream = new MediaStream()
+ let hasVideo = false
+ for (const publication of participant.trackPublications.values()) {
+ const track = (publication as RemoteTrackPublication).track as RemoteTrack | undefined
+ if (!track?.mediaStreamTrack) continue
+ stream.addTrack(track.mediaStreamTrack)
+ if (track.kind === Track.Kind.Video) hasVideo = true
+ }
+ if (stream.getTracks().length === 0) continue
+ out.push({
+ key: participant.identity,
+ userId: userFromIdentity(participant.identity),
+ stream,
+ hasVideo
+ })
+ }
+ return out
+ }
+
+ /** What this end is sending, for its own tile. */
+ private ownStream(): MediaStream | null {
+ const stream = new MediaStream()
+ for (const publication of this.room.localParticipant.trackPublications.values()) {
+ const track = publication.track?.mediaStreamTrack
+ if (track) stream.addTrack(track)
+ }
+ return stream.getTracks().length > 0 ? stream : null
+ }
+
+ /** Silences the microphone, or lets it speak again. Returns muted. */
+ async toggleMute(): Promise {
+ const on = this.room.localParticipant.isMicrophoneEnabled
+ await this.room.localParticipant.setMicrophoneEnabled(!on)
+ return on
+ }
+
+ /** Shares a screen, or stops. Returns whether one is being shared. */
+ async toggleScreen(): Promise {
+ const on = this.room.localParticipant.isScreenShareEnabled
+ await this.room.localParticipant.setScreenShareEnabled(!on)
+ this.local = this.ownStream()
+ return !on
+ }
+
+ /** Leaves, letting go of the camera and the microphone. */
+ async leave(): Promise {
+ await this.room.disconnect()
+ this.local = null
+ }
+
+ /**
+ * Whether the media server is telling us it cannot read what is arriving.
+ *
+ * Element encrypts a call's media per sender in an encrypted room, with the
+ * keys shared over Matrix. moho does not do that yet, so a call in such a
+ * room is one where everybody is connected and nobody can hear anybody -
+ * which is worth saying rather than leaving people to wonder.
+ */
+ get anyoneUndecodable(): boolean {
+ return [...this.room.remoteParticipants.values()].some((p) =>
+ [...p.trackPublications.values()].some((pub) => pub.isEncrypted && !pub.isSubscribed)
+ )
+ }
+}
diff --git a/src/renderer/src/lib/matrixcall.ts b/src/renderer/src/lib/matrixcall.ts
index 44e130b..b810ba4 100644
--- a/src/renderer/src/lib/matrixcall.ts
+++ b/src/renderer/src/lib/matrixcall.ts
@@ -13,6 +13,7 @@
*/
import { Call, type CallPhase } from './webrtc'
+import { SfuCall, type SfuParticipant } from './livekit'
/** What the daemon sends up when a call event arrives in a room. */
export interface MatrixCallEvent {
@@ -78,7 +79,15 @@ export interface CallSurface {
iceServers: (accountId: string) => Promise
/** Publishes this end's membership of the room's call, and says who we are
* in it - the user and the device, since one person may be in from two. */
- joinMembership: (bufferId: string) => Promise<{ userId: string; deviceId: string; roomId?: string } | null>
+ joinMembership: (bufferId: string) => Promise<{
+ userId: string
+ deviceId: string
+ roomId?: string
+ /** Where the media goes, when the call is held on a media server. */
+ media?: { url: string; jwt: string; identity: string } | null
+ /** Whether the room is encrypted, which decides whether the media is. */
+ encrypted?: boolean
+ } | null>
leaveMembership: (bufferId: string) => Promise
/** Who the room says is in the call. */
callMembers: (bufferId: string) => Promise
@@ -136,6 +145,15 @@ export interface GroupCall {
peers: Map
/** This end's camera and microphone, opened once and lent to every leg. */
local: MediaStream | null
+ /**
+ * The media server, where the call is held on one.
+ *
+ * Set means this is an SFU call - Element's kind - and `peers` stays empty
+ * because nobody is connected to anybody directly. Absent means the mesh.
+ */
+ sfu: SfuCall | null
+ /** Who the media server says is here, for the SFU kind. */
+ sfuPeople: SfuParticipant[]
/** The whole call's id, on every signal, so a second call in the same room
* is a second call rather than a crossed line. */
confId: string
@@ -235,8 +253,50 @@ export class MatrixCalls {
}
const own = await this.surface.joinMembership(bufferId)
if (!own) return
- this.iceForGroup = await this.surface.iceServers(accountId)
const ownKey = memberKey(own.userId, own.deviceId)
+
+ // A call held on a media server, which is what Element's are. Nobody
+ // connects to anybody directly, so none of the mesh below runs.
+ if (own.media) {
+ const sfu = new SfuCall({
+ onParticipants: (people) => {
+ if (this.group) this.group.sfuPeople = people
+ },
+ onConnected: (connected) => this.surface.onPhase(connected ? 'connected' : 'connecting', null),
+ onError: (message) => this.surface.onError(message)
+ })
+ this.group = {
+ accountId,
+ bufferId,
+ ownKey,
+ video,
+ local: null,
+ peers: new Map(),
+ sfu,
+ sfuPeople: [],
+ confId: `moho-conf-${own.roomId ?? bufferId}`
+ }
+ this.heartbeat = setInterval(() => void this.surface.joinMembership(bufferId), 30_000)
+ this.surface.onPhase('connecting', null)
+ try {
+ await sfu.join(own.media.url, own.media.jwt, video)
+ this.group.local = sfu.local
+ if (own.encrypted) {
+ // Element encrypts a call's media per sender in an encrypted room,
+ // with the keys shared over Matrix. moho does not do that yet, so
+ // this is a call everybody is in and nobody can hear.
+ this.surface.onError(
+ 'This room is encrypted, and moho cannot yet read the media in an encrypted call - the others will not hear you'
+ )
+ }
+ } catch (e) {
+ this.surface.onError(`Couldn't reach the media server: ${(e as Error).message}`)
+ this.leaveGroup()
+ }
+ return
+ }
+
+ this.iceForGroup = await this.surface.iceServers(accountId)
// Opened once here rather than per leg: five connections asking the
// machine for the same camera get one capture and four refusals.
let local: MediaStream | null = null
@@ -262,6 +322,8 @@ export class MatrixCalls {
video,
local,
peers: new Map(),
+ sfu: null,
+ sfuPeople: [],
// Named for the room rather than for the buffer: each account calls the
// same room by a different buffer id, and a conference named after one
// of those is a conference the other end never joins.
@@ -285,6 +347,9 @@ export class MatrixCalls {
async meet(members: CallMember[]): Promise {
const group = this.group
if (!group) return
+ // On a media server there is nobody to meet: the server is what everybody
+ // is connected to, and it says who is there.
+ if (group.sfu) return
const present = new Set()
for (const member of members) {
const key = memberKey(member.user_id, member.membership?.device_id ?? '')
@@ -329,8 +394,12 @@ export class MatrixCalls {
if (this.heartbeat) clearInterval(this.heartbeat)
this.heartbeat = null
for (const peer of group.peers.values()) void peer.call.hangUp('user_hangup', true)
- // The shared capture is nobody's leg to stop, so it is stopped here.
- for (const track of group.local?.getTracks() ?? []) track.stop()
+ if (group.sfu) {
+ void group.sfu.leave()
+ } else {
+ // The shared capture is nobody's leg to stop, so it is stopped here.
+ for (const track of group.local?.getTracks() ?? []) track.stop()
+ }
void this.surface.leaveMembership(group.bufferId)
this.surface.onPhase('ended', null)
}
@@ -402,7 +471,7 @@ export class MatrixCalls {
*/
private handleGroup(event: MatrixCallEvent): boolean {
const group = this.group
- if (!group) return false
+ if (!group || group.sfu) return false
const { content, kind, callId } = event
if (content.conf_id !== group.confId) return false
// Somebody else's leg of the mesh, seen because room events are seen by
diff --git a/src/renderer/src/state/store.ts b/src/renderer/src/state/store.ts
index 6a78f52..14e20a2 100644
--- a/src/renderer/src/state/store.ts
+++ b/src/renderer/src/state/store.ts
@@ -1837,9 +1837,12 @@ export class ChatStore {
async joinMatrixGroupCall(bufferId: string, video: boolean): Promise {
const account = this.accountFor(bufferId)
if (!account) return
- await this.matrixCalls.joinGroup(account.id, bufferId, video)
- const group = this.matrixCalls.currentGroup
- if (!group) return
+ // The stage goes up before the joining rather than after it: connecting
+ // to a media server takes a moment, and a window that shows nothing for
+ // that moment looks like a button that did nothing. Set first so the
+ // phases the call reports afterwards land on top of it rather than
+ // underneath - reporting "connected" and then overwriting it with
+ // "connecting" was exactly the bug this shape avoids.
this.set({
callMinimized: false,
activeCall: {
@@ -1851,6 +1854,8 @@ export class ChatStore {
sharingScreen: false
}
})
+ await this.matrixCalls.joinGroup(account.id, bufferId, video)
+ if (!this.matrixCalls.currentGroup) this.set({ activeCall: null })
}
/**
@@ -1914,6 +1919,18 @@ export class ChatStore {
speaking: !this.state.activeCall?.muted && this.levels.speaking('self')
}
]
+ // On a media server, everybody arrives from the server rather than from
+ // a connection of their own - the same tiles either way.
+ for (const person of group.sfuPeople) {
+ this.levels.watch(person.key, person.stream)
+ tiles.push({
+ id: person.key,
+ label: this.matrixDisplayName(group.bufferId, person.userId),
+ stream: person.stream,
+ hasVideo: person.hasVideo,
+ speaking: this.levels.speaking(person.key)
+ })
+ }
for (const peer of group.peers.values()) {
this.levels.watch(peer.key, peer.remote)
// Only once there is something to show: a tile for somebody who has
@@ -1973,8 +1990,26 @@ export class ChatStore {
}
toggleCallMute(): void {
+ if (!this.state.activeCall) return
+ // Three kinds of call reach this button and the button does not care:
+ // one to one, a mesh in a room, and a room call held on a media server.
+ const group = this.matrixCalls.currentGroup
+ if (group?.sfu) {
+ void group.sfu.toggleMute().then((muted) => {
+ if (this.state.activeCall) this.set({ activeCall: { ...this.state.activeCall, muted } })
+ })
+ return
+ }
+ if (group) {
+ // Every leg carries the same capture, so muting it mutes all of them.
+ const track = group.local?.getAudioTracks()[0]
+ if (!track) return
+ track.enabled = !track.enabled
+ this.set({ activeCall: { ...this.state.activeCall, muted: !track.enabled } })
+ return
+ }
const call = this.matrixCalls.current
- if (!call || !this.state.activeCall) return
+ if (!call) return
const muted = call.call.toggleMute()
this.set({ activeCall: { ...this.state.activeCall, muted } })
}
@@ -1987,8 +2022,20 @@ export class ChatStore {
* windows somebody actually has open is the only way to choose one.
*/
async toggleScreenShare(): Promise {
+ if (!this.state.activeCall) return
+ // On a media server the picker is LiveKit's own job, and Electron's
+ // source list is what answers it - see the permission handler in main.
+ const group = this.matrixCalls.currentGroup
+ if (group?.sfu) {
+ const sharing = await group.sfu.toggleScreen().catch((e: Error) => {
+ this.toast('error', e.message)
+ return false
+ })
+ this.set({ activeCall: { ...this.state.activeCall, sharingScreen: sharing } })
+ return
+ }
const call = this.matrixCalls.current
- if (!call || !this.state.activeCall) return
+ if (!call) return
try {
if (call.call.sharingScreen) {
await call.call.shareScreen()