adamlamers/zep
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

replace eager signal deletion with scheduled backend cleanup; add Firefox audio fix, SDP quality tuning, and negotiation hardening

Browse Source
This commit is contained in:
Adam Lamers
2026-05-05 15:31:55 -04:00
parent c320813cf1
commit 47243e52ec
15 changed files with 441 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
package.json
+2 -1
View File
@@ -25,7 +25,8 @@
"openpgp": "^6.3.0",
"spacetimedb": "^2.1.0",
"svelte": "^5.55.1",
"svelte-check": "^4.4.6"
"svelte-check": "^4.4.6",
"webrtc-adapter": "^9.0.5"
},
"devDependencies": {
"@cloudflare/vite-plugin": "^1.31.0",
pnpm-lock.yaml
Generated
+16
View File
@@ -32,6 +32,9 @@ importers:
svelte-check:
specifier: ^4.4.6
version: 4.4.6(picomatch@4.0.4)(svelte@5.55.1)(typescript@5.6.3)
webrtc-adapter:
specifier: ^9.0.5
version: 9.0.5
devDependencies:
'@cloudflare/vite-plugin':
specifier: ^1.31.0
@@ -1924,6 +1927,9 @@ packages:
resolution: {integrity: sha512-xAg7SOnEhrm5zI3puOOKyy1OMcMlIJZYNJY7xLBwSze0UjhPLnWfj2GF2EpT0jmzaJKIWKHLsaSSajf35bcYnA==}
engines: {node: '>=v12.22.7'}
sdp@3.2.2:
resolution: {integrity: sha512-xZocWwfyp4hkbN4hLWxMjmv2Q8aNa9MhmOZ7L9aCZPT+dZsgRr6wZRrSYE3HTdyk/2pZKPSgqI7ns7Een1xMSA==}
semver@7.7.4:
resolution: {integrity: sha512-vFKC2IEtQnVhpT78h1Yp8wzwrf8CM+MzKMHGJZfBtzhZNycRFnXsHk6E5TxIkkMsgNS7mdX3AGB7x2QM2di4lA==}
engines: {node: '>=10'}
@@ -2234,6 +2240,10 @@ packages:
resolution: {integrity: sha512-VwddBukDzu71offAQR975unBIGqfKZpM+8ZX6ySk8nYhVoo5CYaZyzt3YBvYtRtO+aoGlqxPg/B87NGVZ/fu6g==}
engines: {node: '>=12'}
webrtc-adapter@9.0.5:
resolution: {integrity: sha512-U9vjByy/sK2OMXu5mmfuZFKTMIUQe34c0JXRO+oDrxJTsntdYT2iIFwYMOV7HhMTuktcZLGf2W1N/OcSf9ssWg==}
engines: {node: '>=6.0.0', npm: '>=3.10.0'}
whatwg-encoding@3.1.1:
resolution: {integrity: sha512-6qN4hJdMwfYBtE3YBTTHhoeuUrDBPZmbQaxWAqSALV/MeEnR5z1xd8UKud2RAkFoPkmB+hli1TZSnyi84xz1vQ==}
engines: {node: '>=18'}
@@ -3841,6 +3851,8 @@ snapshots:
dependencies:
xmlchars: 2.2.0
sdp@3.2.2: {}
semver@7.7.4: {}
sharp@0.34.5:
@@ -4138,6 +4150,10 @@ snapshots:
webidl-conversions@7.0.0: {}
webrtc-adapter@9.0.5:
dependencies:
sdp: 3.2.2
whatwg-encoding@3.1.1:
dependencies:
iconv-lite: 0.6.3
spacetimedb/src/lib.rs
+9
View File
@@ -86,6 +86,15 @@ pub fn init(ctx: &ReducerContext) {
// Grant access to system user
sync_server_access(&ctx.db, system_identity, s.id);
}
// Seed the recurring WebRTC signal cleanup job if it doesn't exist yet
if ctx.db.webrtc_signal_cleanup().iter().next().is_none() {
let first_run = ctx.timestamp + spacetimedb::TimeDuration::from_micros(30_000_000);
ctx.db.webrtc_signal_cleanup().insert(WebRTCSignalCleanup {
scheduled_id: 0,
scheduled_at: spacetimedb::ScheduleAt::Time(first_run),
});
}
}
#[spacetimedb::reducer(client_connected)]
spacetimedb/src/reducers.rs
+45
View File
@@ -1052,6 +1052,51 @@ pub fn send_webrtc_signal(
media_type,
data,
channel_id,
sent: ctx.timestamp,
});
}
#[spacetimedb::reducer]
pub fn delete_webrtc_signal(ctx: &ReducerContext, signal_id: u64) {
if let Some(signal) = ctx.db.webrtc_signal().id().find(signal_id) {
// Only allow sender or receiver to delete the signal
if signal.sender == ctx.sender() || signal.receiver == ctx.sender() {
ctx.db.webrtc_signal().id().delete(signal_id);
}
}
}
#[spacetimedb::table(accessor = webrtc_signal_cleanup, scheduled(run_webrtc_signal_cleanup))]
#[derive(Clone)]
pub struct WebRTCSignalCleanup {
#[primary_key]
#[auto_inc]
pub scheduled_id: u64,
pub scheduled_at: spacetimedb::ScheduleAt,
}
#[spacetimedb::reducer]
pub fn run_webrtc_signal_cleanup(ctx: &ReducerContext, _cleanup: WebRTCSignalCleanup) {
const CLEANUP_INTERVAL_MICROS: i64 = 30_000_000; // 30 seconds
const SIGNAL_TTL_MICROS: i64 = 60_000_000; // 60 seconds
let cutoff = ctx.timestamp - spacetimedb::TimeDuration::from_micros(SIGNAL_TTL_MICROS);
let stale: Vec<u64> = ctx.db.webrtc_signal()
.iter()
.filter(|s| s.sent < cutoff)
.map(|s| s.id)
.collect();
for id in stale {
ctx.db.webrtc_signal().id().delete(id);
}
// Re-schedule next cleanup
let next_run = ctx.timestamp + spacetimedb::TimeDuration::from_micros(CLEANUP_INTERVAL_MICROS);
ctx.db.webrtc_signal_cleanup().insert(WebRTCSignalCleanup {
scheduled_id: 0,
scheduled_at: spacetimedb::ScheduleAt::Time(next_run),
});
}
spacetimedb/src/tables.rs
+3
View File
@@ -152,8 +152,11 @@ pub struct WebRTCSignal {
pub data: String,
#[index(btree)]
pub channel_id: u64,
pub sent: Timestamp,
}
#[spacetimedb::table(accessor = channel_subscription)]
#[derive(Clone)]
pub struct ChannelSubscription {
spacetimedb/src/utils.rs
+12 -2
View File
@@ -415,13 +415,23 @@ pub fn clear_user_presence(db: &Local, identity: Identity) {
}
pub fn clear_signaling_for_user(db: &Local, identity: Identity) {
let signals: Vec<_> = db
let signals_as_sender: Vec<_> = db
.webrtc_signal()
.sender()
.filter(identity)
.map(|s| s.id)
.collect();
for id in signals {
for id in signals_as_sender {
db.webrtc_signal().id().delete(id);
}
let signals_as_receiver: Vec<_> = db
.webrtc_signal()
.receiver()
.filter(identity)
.map(|s| s.id)
.collect();
for id in signals_as_receiver {
db.webrtc_signal().id().delete(id);
}
}
src/SpacetimeProvider.svelte
+94 -49
View File
@@ -17,53 +17,19 @@
let providerKey = $state(0);
$effect(() => {
// Hold off until OIDC is settled for the first time
// Hold off until OIDC is settled
if (auth.isLoading) return;
const currentToken = auth.user?.id_token;
// 1. Initial creation
if (!builder) {
console.log(`[SpacetimeProvider] Initializing connection builder. OIDC present: ${!!currentToken}`);
// Only (re)create the builder if we don't have one, or if the token changed.
if (!builder || currentToken !== lastUsedOidcToken) {
console.log(`[SpacetimeProvider] Initializing connection (Auth Settled). OIDC present: ${!!currentToken}`);
untrack(() => {
builder = connectionBuilder(currentToken);
lastUsedOidcToken = currentToken;
providerKey += 1;
});
return;
}
// 2. Identity transition (Logged out -> Logged in)
// If we were a guest (or null) and now have a token, we SHOULD remount
// to ensure the OIDC credentials take over completely.
if (currentToken && !lastUsedOidcToken) {
console.log("[SpacetimeProvider] Transitioning from Guest/None to OIDC session. Remounting...");
untrack(() => {
builder = connectionBuilder(currentToken);
lastUsedOidcToken = currentToken;
providerKey += 1;
});
return;
}
// 3. Background Refresh (Token -> New Token)
// If it's just a refresh, we DON'T remount. We let InnerSpacetimeDBProvider
// handle the in-place upgrade via withToken.
if (currentToken && currentToken !== lastUsedOidcToken) {
console.log("[SpacetimeProvider] Background token refresh detected. Upgrading in-place.");
untrack(() => {
lastUsedOidcToken = currentToken;
// Notice we DON'T increment providerKey here
});
}
// 4. Logout (Token -> Null)
if (!currentToken && lastUsedOidcToken) {
console.log("[SpacetimeProvider] User logged out. Remounting for Guest mode.");
untrack(() => {
builder = connectionBuilder(undefined);
lastUsedOidcToken = undefined;
providerKey += 1;
providerKey += 1; // Force re-mount of InnerSpacetimeDBProvider for a clean handshake
});
}
});
@@ -71,17 +37,36 @@
// Reactive labels for the loading screen
const host = getStdbHost();
const dbName = getStdbDbName();
// Unified status message logic
const statusInfo = $derived.by(() => {
if (auth.isLoading) {
return {
title: "Authenticating...",
icon: "fa-id-card",
message: "Verifying your identity with the provider."
};
}
if (!builder) {
return {
title: "Preparing Handshake...",
icon: "fa-key",
message: "Finalizing security credentials."
};
}
// Default connecting state
return {
title: "Connecting to SpacetimeDB...",
icon: "fa-circle-notch",
spin: true,
message: "Establishing a secure connection to the chat server."
};
});
</script>
{#if !builder || (auth.isLoading && !auth.user)}
<div class="login-screen">
<div class="login-card" style="text-align: center;">
<i class="fas fa-id-card fa-spin" style="font-size: 3rem; color: var(--brand); margin-bottom: 20px;"></i>
<h1>Authenticating...</h1>
<p style="color: var(--text-muted); margin-top: 8px; margin-bottom: 24px;">Synchronizing your session credentials.</p>
</div>
</div>
{:else}
{#if builder && !auth.isLoading}
{#key providerKey}
<InnerSpacetimeDBProvider
{builder}
@@ -93,6 +78,35 @@
{@render children()}
</InnerSpacetimeDBProvider>
{/key}
{:else}
<div class="login-screen">
<div class="login-card" style="text-align: center;">
<i class="fas {statusInfo.icon} {statusInfo.spin ? 'fa-spin' : 'fa-pulse'}" style="font-size: 3rem; color: var(--brand); margin-bottom: 20px;"></i>
<h1>{statusInfo.title}</h1>
<p style="color: var(--text-muted); margin-top: 8px; margin-bottom: 24px;">{statusInfo.message}</p>
<div class="connection-details-box">
<div class="connection-detail">
<div class="detail-label">Host</div>
<div class="detail-value">{host}</div>
</div>
<div class="connection-detail">
<div class="detail-label">Database</div>
<div class="detail-value">{dbName}</div>
</div>
</div>
{#if onCancel}
<button
onclick={onCancel}
class="btn-secondary"
style="width: 100%;"
>
Cancel
</button>
{/if}
</div>
</div>
{/if}
<style>
@@ -134,4 +148,35 @@
margin-bottom: 24px;
font-size: 16px;
}
.connection-details-box {
background-color: var(--background-tertiary);
padding: 16px;
border-radius: 8px;
margin-bottom: 24px;
text-align: left;
font-size: 0.8rem;
border: 1px solid var(--background-modifier-accent);
display: flex;
flex-direction: column;
gap: 12px;
width: 100%;
box-sizing: border-box;
}
.detail-label {
color: var(--text-muted);
font-weight: 800;
text-transform: uppercase;
font-size: 0.65rem;
margin-bottom: 4px;
letter-spacing: 0.05em;
}
.detail-value {
color: var(--text-normal);
font-family: var(--font-code);
word-break: break-all;
line-height: 1.4;
}
</style>
src/chat/ChatContainer.svelte
+1 -1
View File
@@ -1,7 +1,7 @@
<script lang="ts">
import { useSpacetimeDB } from "spacetimedb/svelte";
import { setContext, onMount, untrack } from "svelte";
import { ChatService } from "./services/chat.svelte";
import { ChatService, Permissions } from "./services/chat.svelte";
import { WebRTCService } from "./services/webrtc/webrtc.svelte";
import ServerList from "./components/ServerList.svelte";
import ChannelList from "./components/ChannelList.svelte";
src/chat/components/settings/RoleSettings.svelte
+2 -2
View File
@@ -4,7 +4,7 @@
import { Permissions } from "../../services/chat.svelte";
import Button from "../ui/Button.svelte";
import Input from "../ui/Input.svelte";
import type * as Types from "../../../module_bindings/types";
const chat = getContext<ChatService>("chat");
@@ -202,7 +202,7 @@
class:drag-over={dragOverRoleId === -1n}
ondragover={(e) => { e.preventDefault(); dragOverRoleId = -1n; }}
onondragleave={() => dragOverRoleId = null}
ondrop={(e) => {
ondrop={(_e) => {
if (draggedRoleId !== null) {
// Logic to move to the very bottom
const roleList = [...roles];
src/chat/services/webrtc/channel-audio-webrtc.svelte.ts
+72 -46
View File
@@ -24,17 +24,20 @@ export class ChannelAudioWebRTCService {
isDeafened = $state(false);
peerManager: PeerManagerService;
localMedia: any; // Type added in constructor
constructor(
identity: Identity | null,
connectedChannelId: bigint | undefined,
localStream: MediaStream | null,
isDeafened: boolean,
localMedia: any,
) {
this.identity = identity;
this.connectedChannelId = connectedChannelId;
this.localStream = localStream;
this.isDeafened = isDeafened;
this.localMedia = localMedia;
const [usStore] = useTable(tables.visible_user_states);
usStore.subscribe((v) => (this.userStates = v));
@@ -58,9 +61,9 @@ export class ChannelAudioWebRTCService {
// Track Syncing
$effect(() => {
const audioTrack = this.localStream?.getAudioTracks()[0] || null;
// Accessing peers and peerStatuses to trigger effect on changes
void this.peerManager.peerStatuses;
const transmittableStream = this.localMedia.localStream ? this.localMedia.getTransmittableStream() : null;
const audioTrack = transmittableStream?.getAudioTracks()[0] || null;
this.peerManager.peers.forEach(async (peer, peerIdHex) => {
const transceivers = peer.pc.getTransceivers();
const audioTransceiver = transceivers.find(t => t.receiver.track.kind === 'audio' || t.sender.track?.kind === 'audio');
@@ -90,14 +93,14 @@ export class ChannelAudioWebRTCService {
const currentIdentity = this.identity;
if (!currentChannelId || !currentIdentity) {
console.log(`[WebRTC][voice] Cleaning up channel state (Channel: ${currentChannelId}, Identity: ${currentIdentity?.toHexString().substring(0,8)})`);
this.peerManager.peers.forEach((_, id) =>
this.peerManager.closePeer(id),
);
this.processedSignals.clear();
this.makingOffer.clear();
this.ignoreOffer.clear();
this.signalingQueue.clear();
if (lastChannelId) {
console.log(`[WebRTC][voice] Cleaning up channel state (Channel: ${lastChannelId}, Identity: ${currentIdentity?.toHexString().substring(0,8)})`);
this.peerManager.peers.forEach((_, id) => this.peerManager.closePeer(id));
this.processedSignals.clear();
this.makingOffer.clear();
this.ignoreOffer.clear();
this.signalingQueue.clear();
}
lastChannelId = undefined;
return;
}
@@ -128,7 +131,8 @@ export class ChannelAudioWebRTCService {
console.log(`[WebRTC][voice] Initiating mesh connection to ${id}`);
const pc = this.peerManager.createPeerConnection(id);
if (pc) {
const audioTrack = this.localStream?.getAudioTracks()[0] || null;
const transmittableStream = this.localMedia.localStream ? this.localMedia.getTransmittableStream() : null;
const audioTrack = transmittableStream?.getAudioTracks()[0] || null;
const transceiver = pc.addTransceiver('audio', { direction: 'sendrecv' });
if (audioTrack) {
transceiver.sender.replaceTrack(audioTrack);
@@ -161,10 +165,6 @@ export class ChannelAudioWebRTCService {
s.mediaType.tag === "Voice"
);
if (mySignals.length > 0) {
console.log(`[WebRTC][voice] Found ${mySignals.length} signals for channel ${currentChannelId}`);
}
for (const signal of mySignals) {
if (this.processedSignals.has(signal.id)) continue;
this.processedSignals.add(signal.id);
@@ -181,6 +181,7 @@ export class ChannelAudioWebRTCService {
this.handleIceCandidate(signal);
break;
}
}
});
}
@@ -227,7 +228,10 @@ export class ChannelAudioWebRTCService {
return;
try {
this.makingOffer.set(peerIdHex, true);
await pc.setLocalDescription();
const offer = await pc.createOffer();
const mungedOffer = this.peerManager.mungeSDP(offer);
await pc.setLocalDescription(mungedOffer);
console.log(`[WebRTC][voice] Sending Offer to ${peerIdHex.substring(0,8)}`);
this.#sendSignal({
receiver: Identity.fromString(peerIdHex),
@@ -236,6 +240,8 @@ export class ChannelAudioWebRTCService {
data: JSON.stringify(pc.localDescription),
channelId: this.connectedChannelId,
});
} catch (e) {
console.error(`[WebRTC][voice] Error during negotiation for ${peerIdHex}`, e);
} finally {
this.makingOffer.set(peerIdHex, false);
}
@@ -249,7 +255,7 @@ export class ChannelAudioWebRTCService {
receiver: Identity.fromString(peerIdHex),
signalKind: { tag: "IceCandidate" },
mediaType: { tag: "Voice" },
data: JSON.stringify(candidate),
data: JSON.stringify(candidate.toJSON()),
channelId: this.connectedChannelId,
});
}
@@ -257,40 +263,61 @@ export class ChannelAudioWebRTCService {
handleOffer(signal: Types.WebRtcSignal) {
const peerIdHex = signal.sender.toHexString();
console.log(`[WebRTC][voice] Received Offer from ${peerIdHex.substring(0,8)}`);
this.enqueueSignalingTask(peerIdHex, async () => {
const pc = this.peerManager.createPeerConnection(peerIdHex);
if (!pc) return;
try {
const description = JSON.parse(signal.data);
const isPolite = this.identity!.toHexString() < peerIdHex;
const offerCollision =
pc.signalingState !== "stable" || !!this.makingOffer.get(peerIdHex);
description.type === "offer" &&
(pc.signalingState !== "stable" || !!this.makingOffer.get(peerIdHex));
const ignoreOffer = !isPolite && offerCollision;
this.ignoreOffer.set(peerIdHex, ignoreOffer);
if (ignoreOffer) return;
if (ignoreOffer) {
console.log(`[WebRTC][voice] Ignoring offer collision from ${peerIdHex.substring(0,8)} (Impolite)`);
return;
}
if (offerCollision) await pc.setLocalDescription({ type: "rollback" });
await pc.setRemoteDescription(
new RTCSessionDescription(JSON.parse(signal.data)),
);
if (offerCollision) {
console.log(`[WebRTC][voice] Rolling back for offer from ${peerIdHex.substring(0,8)} (Polite)`);
await Promise.all([
pc.setLocalDescription({ type: "rollback" }),
pc.setRemoteDescription(description)
]);
} else {
await pc.setRemoteDescription(description);
}
// Map local track to the transceiver created by the offer
const transceivers = pc.getTransceivers();
const audioTrack = this.localStream?.getAudioTracks()[0];
const audioTransceiver = transceivers.find((t: RTCRtpTransceiver) => t.receiver.track.kind === 'audio');
const transmittableStream = this.localMedia.localStream ? this.localMedia.getTransmittableStream() : null;
const audioTrack = transmittableStream?.getAudioTracks()[0];
const audioTransceiver = transceivers.find((t: RTCRtpTransceiver) =>
t.receiver.track.kind === 'audio' || t.sender.track?.kind === 'audio'
);
if (audioTransceiver && audioTrack) {
await audioTransceiver.sender.replaceTrack(audioTrack);
audioTransceiver.direction = 'sendrecv';
if (audioTransceiver) {
if (audioTrack) {
await audioTransceiver.sender.replaceTrack(audioTrack);
}
if (audioTransceiver.direction === 'recvonly') {
audioTransceiver.direction = 'sendrecv';
}
}
const answer = await pc.createAnswer();
await pc.setLocalDescription(answer);
const mungedAnswer = this.peerManager.mungeSDP(answer);
await pc.setLocalDescription(mungedAnswer);
console.log(`[WebRTC][voice] Sending Answer to ${peerIdHex.substring(0,8)}`);
this.#sendSignal({
receiver: signal.sender,
signalKind: { tag: "Answer" },
mediaType: { tag: "Voice" },
data: JSON.stringify(answer),
data: JSON.stringify(pc.localDescription),
channelId: this.connectedChannelId!,
});
await this.drainCandidateQueue(peerIdHex, pc);
@@ -310,9 +337,8 @@ export class ChannelAudioWebRTCService {
const peer = this.peerManager.getPeer(peerIdHex);
if (!peer) return;
try {
await peer.pc.setRemoteDescription(
new RTCSessionDescription(JSON.parse(signal.data)),
);
const description = JSON.parse(signal.data);
await peer.pc.setRemoteDescription(description);
await this.drainCandidateQueue(peerIdHex, peer.pc);
} catch (e) {
console.error(
@@ -325,24 +351,24 @@ export class ChannelAudioWebRTCService {
handleIceCandidate(signal: Types.WebRtcSignal) {
const peerIdHex = signal.sender.toHexString();
console.log(`[WebRTC][voice] Received ICE candidate from ${peerIdHex.substring(0,8)}`);
this.enqueueSignalingTask(peerIdHex, async () => {
const pc = this.peerManager.createPeerConnection(peerIdHex);
const pc = this.peerManager.getPeer(peerIdHex)?.pc;
if (!pc) return;
try {
const candidate = JSON.parse(signal.data);
if (pc.remoteDescription) {
await pc.addIceCandidate(new RTCIceCandidate(candidate));
} else if (!this.ignoreOffer.get(peerIdHex)) {
const queue = this.candidateQueue.get(peerIdHex) || [];
queue.push(candidate);
this.candidateQueue.set(peerIdHex, queue);
if (candidate && candidate.candidate) {
if (pc.remoteDescription) {
await pc.addIceCandidate(candidate);
} else {
const queue = this.candidateQueue.get(peerIdHex) || [];
queue.push(candidate);
this.candidateQueue.set(peerIdHex, queue);
}
}
} catch (e) {
console.error(
`[WebRTC][voice] Failed to handle ICE from ${peerIdHex}`,
e,
);
if (!this.ignoreOffer.get(peerIdHex)) {
console.warn(`[WebRTC][voice] Failed to handle ICE from ${peerIdHex}`, e);
}
}
});
}
src/chat/services/webrtc/local-media.svelte.ts
+53 -14
View File
@@ -44,6 +44,8 @@ export class LocalMediaService {
connectedChannelId: bigint | undefined = $state();
#audioContext: AudioContext | null = null;
#outboundGainNode: GainNode | null = null;
#outboundStream: MediaStream | null = null;
constructor(connectedChannelId: bigint | undefined) {
this.connectedChannelId = connectedChannelId;
@@ -52,14 +54,14 @@ export class LocalMediaService {
this.enumerateDevices();
navigator.mediaDevices.ondevicechange = () => this.enumerateDevices();
// Handle Mute/Deafen/VAD effect on tracks
// Handle Mute/Deafen/VAD effect on OUTBOUND stream
// Instead of disabling the track (which breaks SRTP in Firefox),
// we use a GainNode to silence the signal.
$effect(() => {
if (this.localStream) {
this.localStream.getAudioTracks().forEach((track) => {
// Track is enabled only if NOT muted, NOT deafened, and ACTIVELY talking
// This prevents background noise from leaking to other peers.
track.enabled = !this.isMuted && !this.isDeafened && this.isTalking;
});
if (this.#outboundGainNode) {
const isSilenced = this.isMuted || this.isDeafened || !this.isTalking;
const ctx = this.getAudioContext();
this.#outboundGainNode.gain.setTargetAtTime(isSilenced ? 0 : 1, ctx.currentTime, 0.01);
}
});
@@ -254,6 +256,30 @@ export class LocalMediaService {
return this.#audioContext;
};
/**
* Returns a stream suitable for WebRTC transmission.
* Uses Web Audio to provide clean gating without breaking the SRTP transport.
* Memoizes the stream to prevent leaking nodes and redundant processing.
*/
getTransmittableStream = () => {
if (!this.localStream) return null;
if (this.#outboundStream) return this.#outboundStream;
console.log("[local-media] Creating outbound transmittable stream...");
const ctx = this.getAudioContext();
const source = ctx.createMediaStreamSource(this.localStream);
const destination = ctx.createMediaStreamDestination();
const gainNode = ctx.createGain();
source.connect(gainNode);
gainNode.connect(destination);
this.#outboundGainNode = gainNode;
this.#outboundStream = destination.stream;
return this.#outboundStream;
};
get isSharingScreen() {
return !!this.localScreenStream;
}
@@ -291,14 +317,22 @@ export class LocalMediaService {
try {
console.log("[local-media] Requesting mic permission...");
this.getAudioContext(); // Resume/Init context on user gesture
const constraints: MediaStreamConstraints = {
audio:
this.selectedDeviceId && this.selectedDeviceId !== "default"
? { deviceId: { exact: this.selectedDeviceId } }
: true,
video: false,
const audioConstraints: MediaTrackConstraints = {
echoCancellation: true,
noiseSuppression: true,
autoGainControl: true,
channelCount: 1, // Mono is more efficient for voice mesh
};
const stream = await navigator.mediaDevices.getUserMedia(constraints);
if (this.selectedDeviceId && this.selectedDeviceId !== "default") {
audioConstraints.deviceId = { exact: this.selectedDeviceId };
}
const stream = await navigator.mediaDevices.getUserMedia({
audio: audioConstraints,
video: false,
});
this.localStream = stream;
console.log("[local-media] Mic stream acquired successfully.");
// Re-enumerate to get labels if they weren't available before
@@ -322,6 +356,11 @@ export class LocalMediaService {
this.localStream.getTracks().forEach((track) => track.stop());
this.localStream = null;
}
if (this.#outboundGainNode) {
this.#outboundGainNode.disconnect();
this.#outboundGainNode = null;
}
this.#outboundStream = null;
};
startScreenShare = async (
src/chat/services/webrtc/peer-manager.svelte.ts
+61 -4
View File
@@ -2,7 +2,13 @@ import { Identity } from "spacetimedb";
import type { Peer, WebRTCStats } from "./types";
const ICE_SERVERS: RTCConfiguration = {
iceServers: [{ urls: "stun:stun.services.mozilla.com:3478" }],
iceServers: [
{ urls: "stun:stun.l.google.com:19302" },
{ urls: "stun:stun1.l.google.com:19302" },
{ urls: "stun:stun2.l.google.com:19302" },
{ urls: "stun:stun.services.mozilla.com" },
],
iceCandidatePoolSize: 2,
};
export class PeerManagerService {
@@ -210,7 +216,11 @@ export class PeerManagerService {
console.log(
`[WebRTC][${this.mediaType}] Creating new PeerConnection for ${peerIdHex}`,
);
const pc = new RTCPeerConnection(ICE_SERVERS);
const pc = new RTCPeerConnection({
...ICE_SERVERS,
bundlePolicy: 'max-bundle',
rtcpMuxPolicy: 'require'
});
pc.onnegotiationneeded = () => {
console.log(
@@ -376,6 +386,48 @@ export class PeerManagerService {
getPeer = (peerIdHex: string) => this.peers.get(peerIdHex);
/**
* Munges the SDP to force high-quality Opus audio parameters.
* Forces stereo=1, sprop-stereo=1, and maxaveragebitrate=48000.
* 48kbps is a better balance of quality and resilience for mesh networking.
*/
mungeSDP = (description: RTCSessionDescriptionInit): RTCSessionDescriptionInit => {
if (!description.sdp) return description;
let sdp = description.sdp;
// 1. Force Opus to use high bitrate and stereo
// We look for the Opus rtpmap and then the corresponding fmtp line
const opusRtpMap = sdp.match(/a=rtpmap:(\d+) opus\/48000\/2/);
if (opusRtpMap) {
const payloadType = opusRtpMap[1];
const fmtpRegex = new RegExp(`a=fmtp:${payloadType} (.*)`);
const fmtpMatch = sdp.match(fmtpRegex);
if (fmtpMatch) {
let params = fmtpMatch[1];
if (!params.includes("stereo=1")) params += ";stereo=1";
if (!params.includes("sprop-stereo=1")) params += ";sprop-stereo=1";
// Lowered from 128000 to 48000 for better mesh resilience
if (params.includes("maxaveragebitrate=")) {
params = params.replace(/maxaveragebitrate=\d+/, "maxaveragebitrate=48000");
} else {
params += ";maxaveragebitrate=48000";
}
sdp = sdp.replace(fmtpMatch[0], `a=fmtp:${payloadType} ${params}`);
} else {
// Add fmtp line if missing
sdp = sdp.replace(opusRtpMap[0], `${opusRtpMap[0]}\na=fmtp:${payloadType} stereo=1;sprop-stereo=1;maxaveragebitrate=48000`);
}
}
return {
...description,
sdp
};
};
applyEncoderSettings = async (pc: RTCPeerConnection, peerIdHex: string) => {
if (this.mediaType !== "screen") return;
@@ -393,8 +445,13 @@ export class PeerManagerService {
params.encodings[0].priority = "high";
// Maintain resolution over framerate during congestion
// @ts-expect-error - Svelte 5 rune or non-standard property
sender.degradationPreference = "maintain-resolution";
if ("degradationPreference" in sender) {
try {
(sender as any).degradationPreference = "maintain-resolution";
} catch (e) {
console.warn("[WebRTC][screen] Failed to set degradationPreference", e);
}
}
await sender.setParameters(params);
console.log(
src/chat/services/webrtc/screen-sharing-webrtc.svelte.ts
+68 -45
View File
@@ -23,15 +23,18 @@ export class ScreenSharingWebRTCService {
localScreenStream = $state<MediaStream | null>(null);
peerManager: PeerManagerService;
localMedia: any;
constructor(
identity: Identity | null,
connectedChannelId: bigint | undefined,
localScreenStream: MediaStream | null,
localMedia: any,
) {
this.identity = identity;
this.connectedChannelId = connectedChannelId;
this.localScreenStream = localScreenStream;
this.localMedia = localMedia;
const [usStore] = useTable(tables.visible_user_states);
usStore.subscribe((v) => (this.userStates = v));
@@ -56,8 +59,7 @@ export class ScreenSharingWebRTCService {
$effect(() => {
const videoTrack = this.localScreenStream?.getVideoTracks()[0] || null;
const audioTrack = this.localScreenStream?.getAudioTracks()[0] || null;
// Accessing peers and peerStatuses to trigger effect on changes
void this.peerManager.peerStatuses;
this.peerManager.peers.forEach(async (peer, peerIdHex) => {
const transceivers = peer.pc.getTransceivers();
let changed = false;
@@ -89,14 +91,14 @@ export class ScreenSharingWebRTCService {
const currentIdentity = this.identity;
if (!currentChannelId || !currentIdentity) {
console.log(`[WebRTC][screen] Cleaning up screen state (Channel: ${currentChannelId}, Identity: ${currentIdentity?.toHexString().substring(0,8)})`);
this.peerManager.peers.forEach((_, id) =>
this.peerManager.closePeer(id),
);
this.processedSignals.clear();
this.makingOffer.clear();
this.ignoreOffer.clear();
this.signalingQueue.clear();
if (lastChannelId) {
console.log(`[WebRTC][screen] Cleaning up screen state (Channel: ${lastChannelId}, Identity: ${currentIdentity?.toHexString().substring(0,8)})`);
this.peerManager.peers.forEach((_, id) => this.peerManager.closePeer(id));
this.processedSignals.clear();
this.makingOffer.clear();
this.ignoreOffer.clear();
this.signalingQueue.clear();
}
lastChannelId = undefined;
return;
}
@@ -169,10 +171,6 @@ export class ScreenSharingWebRTCService {
s.mediaType.tag === "Screen"
);
if (mySignals.length > 0) {
console.log(`[WebRTC][screen] Found ${mySignals.length} signals for channel ${currentChannelId}`);
}
for (const signal of mySignals) {
if (this.processedSignals.has(signal.id)) continue;
this.processedSignals.add(signal.id);
@@ -189,6 +187,7 @@ export class ScreenSharingWebRTCService {
this.handleIceCandidate(signal);
break;
}
}
});
}
@@ -244,6 +243,8 @@ export class ScreenSharingWebRTCService {
data: JSON.stringify(pc.localDescription),
channelId: this.connectedChannelId,
});
} catch (e) {
console.error(`[WebRTC][screen] Error during negotiation for ${peerIdHex}`, e);
} finally {
this.makingOffer.set(peerIdHex, false);
}
@@ -257,7 +258,7 @@ export class ScreenSharingWebRTCService {
receiver: Identity.fromString(peerIdHex),
signalKind: { tag: "IceCandidate" },
mediaType: { tag: "Screen" },
data: JSON.stringify(candidate),
data: JSON.stringify(candidate.toJSON()),
channelId: this.connectedChannelId,
});
}
@@ -270,17 +271,28 @@ export class ScreenSharingWebRTCService {
const pc = this.peerManager.createPeerConnection(peerIdHex);
if (!pc) return;
try {
const description = JSON.parse(signal.data);
const isPolite = this.identity!.toHexString() < peerIdHex;
const offerCollision =
pc.signalingState !== "stable" || !!this.makingOffer.get(peerIdHex);
description.type === "offer" &&
(pc.signalingState !== "stable" || !!this.makingOffer.get(peerIdHex));
const ignoreOffer = !isPolite && offerCollision;
this.ignoreOffer.set(peerIdHex, ignoreOffer);
if (ignoreOffer) return;
if (ignoreOffer) {
console.log(`[WebRTC][screen] Ignoring offer collision from ${peerIdHex.substring(0,8)} (Impolite)`);
return;
}
if (offerCollision) await pc.setLocalDescription({ type: "rollback" });
await pc.setRemoteDescription(
new RTCSessionDescription(JSON.parse(signal.data)),
);
if (offerCollision) {
console.log(`[WebRTC][screen] Rolling back for offer from ${peerIdHex.substring(0,8)} (Polite)`);
await Promise.all([
pc.setLocalDescription({ type: "rollback" }),
pc.setRemoteDescription(description)
]);
} else {
await pc.setRemoteDescription(description);
}
// After setting remote offer, the transceivers exist.
// We should attach our local tracks if we have them.
@@ -288,16 +300,28 @@ export class ScreenSharingWebRTCService {
const videoTrack = this.localScreenStream?.getVideoTracks()[0];
const audioTrack = this.localScreenStream?.getAudioTracks()[0];
const videoTransceiver = transceivers.find((t: RTCRtpTransceiver) => t.receiver.track.kind === 'video');
const audioTransceiver = transceivers.find((t: RTCRtpTransceiver) => t.receiver.track.kind === 'audio');
const videoTransceiver = transceivers.find((t: RTCRtpTransceiver) =>
t.receiver.track.kind === 'video' || t.sender.track?.kind === 'video'
);
const audioTransceiver = transceivers.find((t: RTCRtpTransceiver) =>
t.receiver.track.kind === 'audio' || t.sender.track?.kind === 'audio'
);
if (videoTransceiver && videoTrack) {
await videoTransceiver.sender.replaceTrack(videoTrack);
videoTransceiver.direction = 'sendrecv';
if (videoTransceiver) {
if (videoTrack) {
await videoTransceiver.sender.replaceTrack(videoTrack);
}
if (videoTransceiver.direction === 'recvonly') {
videoTransceiver.direction = 'sendrecv';
}
}
if (audioTransceiver && audioTrack) {
await audioTransceiver.sender.replaceTrack(audioTrack);
audioTransceiver.direction = 'sendrecv';
if (audioTransceiver) {
if (audioTrack) {
await audioTransceiver.sender.replaceTrack(audioTrack);
}
if (audioTransceiver.direction === 'recvonly') {
audioTransceiver.direction = 'sendrecv';
}
}
const answer = await pc.createAnswer();
@@ -307,7 +331,7 @@ export class ScreenSharingWebRTCService {
receiver: signal.sender,
signalKind: { tag: "Answer" },
mediaType: { tag: "Screen" },
data: JSON.stringify(answer),
data: JSON.stringify(pc.localDescription),
channelId: this.connectedChannelId!,
});
await this.drainCandidateQueue(peerIdHex, pc);
@@ -327,9 +351,8 @@ export class ScreenSharingWebRTCService {
const peer = this.peerManager.getPeer(peerIdHex);
if (!peer) return;
try {
await peer.pc.setRemoteDescription(
new RTCSessionDescription(JSON.parse(signal.data)),
);
const description = JSON.parse(signal.data);
await peer.pc.setRemoteDescription(description);
await this.drainCandidateQueue(peerIdHex, peer.pc);
} catch (e) {
console.error(
@@ -342,24 +365,24 @@ export class ScreenSharingWebRTCService {
handleIceCandidate(signal: Types.WebRtcSignal) {
const peerIdHex = signal.sender.toHexString();
console.log(`[WebRTC][screen] Received ICE candidate from ${peerIdHex.substring(0,8)}`);
this.enqueueSignalingTask(peerIdHex, async () => {
const pc = this.peerManager.createPeerConnection(peerIdHex);
const pc = this.peerManager.getPeer(peerIdHex)?.pc;
if (!pc) return;
try {
const candidate = JSON.parse(signal.data);
if (pc.remoteDescription) {
await pc.addIceCandidate(new RTCIceCandidate(candidate));
} else if (!this.ignoreOffer.get(peerIdHex)) {
const queue = this.candidateQueue.get(peerIdHex) || [];
queue.push(candidate);
this.candidateQueue.set(peerIdHex, queue);
if (candidate && candidate.candidate) {
if (pc.remoteDescription) {
await pc.addIceCandidate(candidate);
} else {
const queue = this.candidateQueue.get(peerIdHex) || [];
queue.push(candidate);
this.candidateQueue.set(peerIdHex, queue);
}
}
} catch (e) {
console.error(
`[WebRTC][screen] Failed to handle ICE from ${peerIdHex}`,
e,
);
if (!this.ignoreOffer.get(peerIdHex)) {
console.warn(`[WebRTC][screen] Failed to handle ICE from ${peerIdHex}`, e);
}
}
});
}
src/chat/services/webrtc/webrtc.svelte.ts
+2
View File
@@ -77,12 +77,14 @@ export class WebRTCService {
connectedChannelId,
this.localMedia.localStream,
this.localMedia.isDeafened,
this.localMedia,
);
this.screen = new ScreenSharingWebRTCService(
identity,
connectedChannelId,
this.localMedia.localScreenStream,
this.localMedia,
);
$effect(() => {
src/main.ts
+1
View File
@@ -1,4 +1,5 @@
// src/main.ts
import "webrtc-adapter";
import { mount } from "svelte";
import "@fortawesome/fontawesome-free/css/all.min.css";
import "./index.css";