mirror of
https://github.com/J3vb/OwnCord.git
synced 2026-09-03 03:50:00 +03:00
fix: security hardening, LiveKit class refactor, and eng review fixes
Server: - Fix YAML injection in LiveKit config generation (quote values) - Revert token TTL to 4h (no server-side JWT revocation) - Derive LiveKit publish permissions from user role (prevent SFU bypass) - Add CAS guard for webhook/voice_leave race condition - Add voice_leave broadcast to rollbackVoiceJoin (prevent ghost state) - Limit webhook body to 64KB (prevent memory abuse) - Add rate limit to voice_token_refresh handler (1/60s) - Add LiveKit health check endpoint (GET /api/v1/livekit/health, 503 on degraded) - Add voice_token_refresh WS handler for client-initiated token refresh - Consolidate voice quality constants (single source of truth) - Fix video limit TOCTOU race (count from DB instead of LiveKit API) - Raise default voice_max_video from 10 to 25 (Discord parity) - Add CountActiveCameras DB query - Non-blocking broadcast send, circuit breaker, exponential backoff - Close send channel before context cancel in serve.go - Guard voice mute/deafen for active channel - Delete orphaned message on attachment link failure - Redact query string from proxy logs (prevent token leak) - Use instance-level HTTP client for health checks (no redirect following) - Set cmd.WaitDelay to prevent goroutine leak on Windows - Log buildJSON marshal errors Client: - Refactor livekitSession.ts from singleton module to LiveKitSession class - Share single AudioContext for all analysers (was 1 per participant) - Extract createRoom() helper (DRY) - Add token refresh timer (3.5h interval, re-arms on failure) - Skip setSpeakers if unchanged (sort in-place, no allocations) - Distinguish user-initiated leave from connection error in retry - Add YouTube videoId validation (prevent iframe src injection) - Add try/finally to disableCamera - Wrap store subscription callbacks in try/catch - Track and cancel initial scroll RAF on cleanup - Add 5s timeout + encodeURIComponent to YouTube oEmbed fetch - Clean raw mic stream on RNNoise suppressor failure - Full voice cleanup on logout via cleanupAll() Tests: - Add 7 new server tests (webhook parsing, voice guards, quality fallback) - Fix 2 pre-existing test failures (mute/deafen invalid payload)
This commit is contained in:
jevb
@@ -379,7 +379,8 @@ export function createMessageList(options: MessageListOptions): MessageListCompo
|
||||
// doesn't cause jumps from estimate→measured height corrections.
|
||||
premeasureAll();
|
||||
scrollToBottom();
|
||||
requestAnimationFrame(() => scrollToBottom());
|
||||
const initialScrollRaf = requestAnimationFrame(() => scrollToBottom());
|
||||
ac.signal.addEventListener("abort", () => cancelAnimationFrame(initialScrollRaf));
|
||||
|
||||
unsubscribers.push(messagesStore.subscribeSelector(
|
||||
(s) => s.messagesByChannel,
|
||||
|
||||
@@ -66,8 +66,19 @@ export function extractYouTubeId(url: string): string | null {
|
||||
/** Cache for YouTube video titles to avoid re-fetching on every re-render. */
|
||||
const ytTitleCache = new Map<string, string>();
|
||||
|
||||
/** Strict pattern for YouTube video IDs (alphanumeric, hyphens, underscores). */
|
||||
const YOUTUBE_ID_RE = /^[\w-]{1,20}$/;
|
||||
|
||||
/** Render a YouTube embed player with title header. */
|
||||
export function renderYouTubeEmbed(videoId: string, originalUrl: string): HTMLDivElement {
|
||||
// Validate videoId to prevent injection into iframe src / img src.
|
||||
if (!YOUTUBE_ID_RE.test(videoId)) {
|
||||
const fallback = createElement("div", { class: "msg-embed" });
|
||||
const link = createElement("a", { href: originalUrl, target: "_blank", rel: "noopener noreferrer" });
|
||||
setText(link, originalUrl);
|
||||
fallback.appendChild(link);
|
||||
return fallback;
|
||||
}
|
||||
const wrap = createElement("div", { class: "msg-embed msg-embed-youtube" });
|
||||
|
||||
// Header: channel name + video title
|
||||
@@ -85,10 +96,10 @@ export function renderYouTubeEmbed(videoId: string, originalUrl: string): HTMLDi
|
||||
setText(titleLink, cached);
|
||||
} else {
|
||||
setText(titleLink, "Loading...");
|
||||
const oembedUrl = `https://www.youtube.com/oembed?url=https://www.youtube.com/watch?v=${videoId}&format=json`;
|
||||
fetch(oembedUrl)
|
||||
.then((res) => (res.ok ? res.json() : null))
|
||||
.then((data: { title?: string } | null) => {
|
||||
const oembedUrl = `https://www.youtube.com/oembed?url=https://www.youtube.com/watch?v=${encodeURIComponent(videoId)}&format=json`;
|
||||
fetch(oembedUrl, { signal: AbortSignal.timeout(5000) })
|
||||
.then((res) => (res.ok ? (res.json() as Promise<{ title?: string } | null>) : null))
|
||||
.then((data) => {
|
||||
const title = data?.title ?? "YouTube Video";
|
||||
ytTitleCache.set(videoId, title);
|
||||
setText(titleLink, title);
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -430,7 +430,8 @@ export type ClientMessage =
|
||||
| (WsEnvelope<VoiceDeafenPayload> & { readonly type: "voice_deafen" })
|
||||
| (WsEnvelope<VoiceCameraPayload> & { readonly type: "voice_camera" })
|
||||
| (WsEnvelope<VoiceScreensharePayload> & { readonly type: "voice_screenshare" })
|
||||
| (WsEnvelope<SoundboardPlayPayload> & { readonly type: "soundboard_play" });
|
||||
| (WsEnvelope<SoundboardPlayPayload> & { readonly type: "soundboard_play" })
|
||||
| (WsEnvelope<Record<string, never>> & { readonly type: "voice_token_refresh" });
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// REST API Response Types
|
||||
|
||||
@@ -96,7 +96,11 @@ export function createWsClient() {
|
||||
if (state !== newState) {
|
||||
state = newState;
|
||||
for (const listener of stateListeners) {
|
||||
listener(state);
|
||||
try {
|
||||
listener(state);
|
||||
} catch (err) {
|
||||
log.error("State listener error", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ import { channelsStore, getActiveChannel } from "@stores/channels.store";
|
||||
import { voiceStore } from "@stores/voice.store";
|
||||
import {
|
||||
leaveVoice as voiceSessionLeave,
|
||||
cleanupAll as voiceCleanupAll,
|
||||
setOnRemoteVideo,
|
||||
setOnRemoteVideoRemoved,
|
||||
clearOnRemoteVideo,
|
||||
@@ -126,19 +127,27 @@ export function createMainPage(options: MainPageOptions): MountableComponent {
|
||||
|
||||
unsubscribers.push(
|
||||
ws.onStateChange((wsState) => {
|
||||
if (banner === null) return;
|
||||
if (wsState === "reconnecting") {
|
||||
banner.showReconnecting();
|
||||
} else if (wsState === "connected") {
|
||||
banner.hide();
|
||||
try {
|
||||
if (banner === null) return;
|
||||
if (wsState === "reconnecting") {
|
||||
banner.showReconnecting();
|
||||
} else if (wsState === "connected") {
|
||||
banner.hide();
|
||||
}
|
||||
} catch (err) {
|
||||
log.error("State change handler error", err);
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
unsubscribers.push(
|
||||
ws.on("server_restart", (payload) => {
|
||||
if (banner !== null) {
|
||||
banner.showRestart(payload.delay_seconds);
|
||||
try {
|
||||
if (banner !== null) {
|
||||
banner.showRestart(payload.delay_seconds);
|
||||
}
|
||||
} catch (err) {
|
||||
log.error("Server restart handler error", err);
|
||||
}
|
||||
}),
|
||||
);
|
||||
@@ -279,21 +288,25 @@ export function createMainPage(options: MainPageOptions): MountableComponent {
|
||||
let prevLocalCamera = voiceStore.getState().localCamera;
|
||||
let prevCameraSignature = "";
|
||||
unsubscribers.push(voiceStore.subscribe((state) => {
|
||||
// Build a lightweight signature of camera-relevant state
|
||||
let sig = state.localCamera ? "1" : "0";
|
||||
const channelId = state.currentChannelId;
|
||||
if (channelId !== null) {
|
||||
const users = state.voiceUsers.get(channelId);
|
||||
if (users) {
|
||||
for (const [uid, u] of users) {
|
||||
if (u.camera) sig += `:${uid}`;
|
||||
try {
|
||||
// Build a lightweight signature of camera-relevant state
|
||||
let sig = state.localCamera ? "1" : "0";
|
||||
const channelId = state.currentChannelId;
|
||||
if (channelId !== null) {
|
||||
const users = state.voiceUsers.get(channelId);
|
||||
if (users) {
|
||||
for (const [uid, u] of users) {
|
||||
if (u.camera) sig += `:${uid}`;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if (sig !== prevCameraSignature || state.localCamera !== prevLocalCamera) {
|
||||
prevCameraSignature = sig;
|
||||
prevLocalCamera = state.localCamera;
|
||||
videoModeCtrl?.checkVideoMode();
|
||||
if (sig !== prevCameraSignature || state.localCamera !== prevLocalCamera) {
|
||||
prevCameraSignature = sig;
|
||||
prevLocalCamera = state.localCamera;
|
||||
videoModeCtrl?.checkVideoMode();
|
||||
}
|
||||
} catch (err) {
|
||||
log.error("Voice store subscription error", err);
|
||||
}
|
||||
}));
|
||||
|
||||
@@ -327,42 +340,50 @@ export function createMainPage(options: MainPageOptions): MountableComponent {
|
||||
|
||||
function destroy(): void {
|
||||
log.info("MainPage destroying");
|
||||
// Clean up voice session before destroying UI — prevents stale
|
||||
// module-level state persisting across logout/reconnect cycles.
|
||||
voiceSessionLeave(false);
|
||||
clearVoiceOnError();
|
||||
clearOnRemoteVideo();
|
||||
channelCtrl?.destroyChannel();
|
||||
channelCtrl = null;
|
||||
try {
|
||||
// Full voice cleanup — tears down room, callbacks, ws ref, serverHost.
|
||||
// Prevents stale module-level state persisting across logout/reconnect cycles.
|
||||
voiceCleanupAll();
|
||||
channelCtrl?.destroyChannel();
|
||||
channelCtrl = null;
|
||||
|
||||
reactionCtrl?.destroy();
|
||||
reactionCtrl = null;
|
||||
msgCtrl = null;
|
||||
videoModeCtrl?.destroy();
|
||||
videoModeCtrl = null;
|
||||
reactionCtrl?.destroy();
|
||||
reactionCtrl = null;
|
||||
msgCtrl = null;
|
||||
videoModeCtrl?.destroy();
|
||||
videoModeCtrl = null;
|
||||
|
||||
videoGrid = null;
|
||||
videoGrid = null;
|
||||
|
||||
for (const child of children) {
|
||||
child.destroy?.();
|
||||
for (const child of children) {
|
||||
try {
|
||||
child.destroy?.();
|
||||
} catch (err) {
|
||||
log.error("Child destroy error", err);
|
||||
}
|
||||
}
|
||||
children = [];
|
||||
|
||||
for (const unsub of unsubscribers) {
|
||||
try {
|
||||
unsub();
|
||||
} catch (err) {
|
||||
log.error("Unsubscribe error", err);
|
||||
}
|
||||
}
|
||||
unsubscribers = [];
|
||||
|
||||
if (banner !== null) {
|
||||
banner.destroy();
|
||||
banner = null;
|
||||
}
|
||||
} finally {
|
||||
if (root !== null) {
|
||||
root.remove();
|
||||
root = null;
|
||||
}
|
||||
container = null;
|
||||
}
|
||||
children = [];
|
||||
|
||||
for (const unsub of unsubscribers) {
|
||||
unsub();
|
||||
}
|
||||
unsubscribers = [];
|
||||
|
||||
if (banner !== null) {
|
||||
banner.destroy();
|
||||
banner = null;
|
||||
}
|
||||
|
||||
if (root !== null) {
|
||||
root.remove();
|
||||
root = null;
|
||||
}
|
||||
container = null;
|
||||
}
|
||||
|
||||
return { mount, destroy };
|
||||
|
||||
@@ -74,7 +74,7 @@ export function createChatArea(opts: ChatAreaOptions): ChatAreaResult {
|
||||
getCurrentChannelId: () => getChannelCtrl()?.currentChannelId ?? null,
|
||||
onJumpToMessage: (msgId: number) => {
|
||||
const ctrl = getChannelCtrl();
|
||||
if (ctrl?.messageList === null || ctrl?.messageList === undefined) return false;
|
||||
if (ctrl == null || ctrl.messageList == null) return false;
|
||||
return ctrl.messageList.scrollToMessage(msgId);
|
||||
},
|
||||
});
|
||||
@@ -87,7 +87,7 @@ export function createChatArea(opts: ChatAreaOptions): ChatAreaResult {
|
||||
getCurrentChannelId: () => getChannelCtrl()?.currentChannelId ?? null,
|
||||
onJumpToMessage: (_channelId: number, msgId: number) => {
|
||||
const ctrl = getChannelCtrl();
|
||||
if (ctrl?.messageList === null || ctrl?.messageList === undefined) return false;
|
||||
if (ctrl == null || ctrl.messageList == null) return false;
|
||||
return ctrl.messageList.scrollToMessage(msgId);
|
||||
},
|
||||
});
|
||||
|
||||
@@ -19,7 +19,7 @@ import (
|
||||
//
|
||||
// The client connects to wss://server:8443/livekit/ which is proxied to
|
||||
// ws://localhost:7880/ on the LiveKit server.
|
||||
func NewLiveKitProxy(livekitURL string) http.Handler {
|
||||
func NewLiveKitProxy(livekitURL string, allowedOrigins []string) http.Handler {
|
||||
target, err := url.Parse(livekitURL)
|
||||
if err != nil {
|
||||
target, _ = url.Parse("http://localhost:7880")
|
||||
@@ -54,7 +54,7 @@ func NewLiveKitProxy(livekitURL string) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
// Detect WebSocket upgrade requests.
|
||||
if isWebSocketUpgrade(r) {
|
||||
proxyWebSocket(w, r, &wsTarget)
|
||||
proxyWebSocket(w, r, &wsTarget, allowedOrigins)
|
||||
return
|
||||
}
|
||||
httpProxy.ServeHTTP(w, r)
|
||||
@@ -72,7 +72,7 @@ func isWebSocketUpgrade(r *http.Request) bool {
|
||||
|
||||
// proxyWebSocket opens a backend WS connection and shovels data in both
|
||||
// directions until either side closes.
|
||||
func proxyWebSocket(w http.ResponseWriter, r *http.Request, target *url.URL) {
|
||||
func proxyWebSocket(w http.ResponseWriter, r *http.Request, target *url.URL, allowedOrigins []string) {
|
||||
// Build backend URL preserving the request path and query.
|
||||
backendURL := *target
|
||||
backendURL.Path = r.URL.Path
|
||||
@@ -83,7 +83,7 @@ func proxyWebSocket(w http.ResponseWriter, r *http.Request, target *url.URL) {
|
||||
Subprotocols: r.Header.Values("Sec-WebSocket-Protocol"),
|
||||
})
|
||||
if err != nil {
|
||||
slog.Warn("livekit proxy: backend dial failed", "url", backendURL.String(), "err", err)
|
||||
slog.Warn("livekit proxy: backend dial failed", "host", backendURL.Host, "path", backendURL.Path, "err", err)
|
||||
http.Error(w, "backend unavailable", http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
@@ -92,7 +92,7 @@ func proxyWebSocket(w http.ResponseWriter, r *http.Request, target *url.URL) {
|
||||
// Accept the frontend WebSocket.
|
||||
frontConn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
|
||||
Subprotocols: []string{backConn.Subprotocol()},
|
||||
OriginPatterns: []string{"*"},
|
||||
OriginPatterns: allowedOrigins,
|
||||
})
|
||||
if err != nil {
|
||||
slog.Warn("livekit proxy: frontend accept failed", "err", err)
|
||||
@@ -100,7 +100,11 @@ func proxyWebSocket(w http.ResponseWriter, r *http.Request, target *url.URL) {
|
||||
}
|
||||
defer frontConn.Close(websocket.StatusNormalClosure, "")
|
||||
|
||||
ctx := r.Context()
|
||||
// Use a cancellable context so when one direction finishes, the other
|
||||
// goroutine's copyWS read/write is unblocked and can drain cleanly.
|
||||
ctx, cancel := context.WithCancel(r.Context())
|
||||
defer cancel()
|
||||
|
||||
errc := make(chan error, 2)
|
||||
|
||||
// Frontend → Backend
|
||||
@@ -113,7 +117,9 @@ func proxyWebSocket(w http.ResponseWriter, r *http.Request, target *url.URL) {
|
||||
errc <- copyWS(ctx, frontConn, backConn)
|
||||
}()
|
||||
|
||||
// Wait for either direction to finish.
|
||||
// Wait for either direction to finish, then cancel+drain both.
|
||||
<-errc
|
||||
cancel()
|
||||
<-errc
|
||||
}
|
||||
|
||||
|
||||
+35
-1
@@ -89,10 +89,14 @@ func NewRouter(cfg *config.Config, database *db.DB, ver string, logBuf *admin.Ri
|
||||
r.Post("/api/v1/livekit/webhook",
|
||||
ws.MountWebhookRoute(hub, cfg.Voice.LiveKitAPIKey, cfg.Voice.LiveKitAPISecret))
|
||||
|
||||
// LiveKit health check — admin-IP-restricted.
|
||||
r.With(AdminIPRestrict(cfg.Server.AdminAllowedCIDRs)).
|
||||
Get("/api/v1/livekit/health", handleLiveKitHealth(hub))
|
||||
|
||||
// Reverse proxy LiveKit signaling through OwnCord's HTTPS server.
|
||||
// This avoids mixed-content blocks (secure page → insecure WS).
|
||||
// Client connects to wss://server:8443/livekit/* → ws://localhost:7880/*
|
||||
r.Handle("/livekit/*", http.StripPrefix("/livekit", NewLiveKitProxy(cfg.Voice.LiveKitURL)))
|
||||
r.Handle("/livekit/*", http.StripPrefix("/livekit", NewLiveKitProxy(cfg.Voice.LiveKitURL, cfg.Server.AllowedOrigins)))
|
||||
}
|
||||
|
||||
go hub.Run()
|
||||
@@ -152,6 +156,36 @@ func handleInfo(cfg *config.Config, ver string) http.HandlerFunc {
|
||||
}
|
||||
}
|
||||
|
||||
// livekitHealthResponse is the JSON shape returned by GET /api/v1/livekit/health.
|
||||
type livekitHealthResponse struct {
|
||||
Status string `json:"status"`
|
||||
LiveKitReachable bool `json:"livekit_reachable"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
func handleLiveKitHealth(hub *ws.Hub) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
ok, err := hub.LiveKitHealthCheck()
|
||||
if ok {
|
||||
writeJSON(w, http.StatusOK, livekitHealthResponse{
|
||||
Status: "ok",
|
||||
LiveKitReachable: true,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
errMsg := "unknown"
|
||||
if err != nil {
|
||||
errMsg = err.Error()
|
||||
}
|
||||
writeJSON(w, http.StatusServiceUnavailable, livekitHealthResponse{
|
||||
Status: "degraded",
|
||||
LiveKitReachable: false,
|
||||
Error: errMsg,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// setRequestIDHeader copies the request ID from context into the response header.
|
||||
func setRequestIDHeader(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
@@ -170,6 +170,21 @@ func (d *DB) ClearAllVoiceStates() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// CountActiveCameras returns the number of users with camera enabled in the
|
||||
// given voice channel. Uses the DB as source of truth (race-free via SQLite
|
||||
// serialization) rather than querying LiveKit.
|
||||
func (d *DB) CountActiveCameras(channelID int64) (int, error) {
|
||||
var count int
|
||||
err := d.sqlDB.QueryRow(
|
||||
`SELECT COUNT(*) FROM voice_states WHERE channel_id = ? AND camera = 1`,
|
||||
channelID,
|
||||
).Scan(&count)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("CountActiveCameras: %w", err)
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
// UpdateVoiceCamera sets the camera field for the given user's voice state.
|
||||
func (d *DB) UpdateVoiceCamera(userID int64, camera bool) error {
|
||||
_, err := d.sqlDB.Exec(
|
||||
|
||||
+23
-1
@@ -48,7 +48,10 @@ func main() {
|
||||
// run is the real entrypoint — separated for testability.
|
||||
func run(log *slog.Logger, logBuf *admin.RingBuffer) error {
|
||||
// Clean up old binary from a previous update.
|
||||
if exePath, err := os.Executable(); err == nil {
|
||||
exePath, exeErr := os.Executable()
|
||||
if exeErr != nil {
|
||||
log.Warn("failed to determine executable path", "error", exeErr)
|
||||
} else {
|
||||
oldPath := exePath + ".old"
|
||||
if _, statErr := os.Stat(oldPath); statErr == nil {
|
||||
if rmErr := os.Remove(oldPath); rmErr != nil {
|
||||
@@ -148,11 +151,23 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer) error {
|
||||
go func() {
|
||||
ticker := time.NewTicker(15 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
consecutiveFailures := 0
|
||||
const maxConsecutiveFailures = 5
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
if consecutiveFailures >= maxConsecutiveFailures {
|
||||
log.Error("maintenance loop: circuit breaker open, skipping tick",
|
||||
"consecutive_failures", consecutiveFailures)
|
||||
// Reset after one skip to allow retry next tick.
|
||||
consecutiveFailures = maxConsecutiveFailures - 1
|
||||
continue
|
||||
}
|
||||
|
||||
tickFailed := false
|
||||
if err := database.DeleteExpiredSessions(); err != nil {
|
||||
log.Warn("failed to delete expired sessions", "error", err)
|
||||
tickFailed = true
|
||||
}
|
||||
|
||||
// Clean up orphaned attachments (uploaded but never linked to a message).
|
||||
@@ -160,6 +175,7 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer) error {
|
||||
orphanFiles, orphanErr := database.DeleteOrphanedAttachments(cutoff)
|
||||
if orphanErr != nil {
|
||||
log.Warn("failed to delete orphaned attachments", "error", orphanErr)
|
||||
tickFailed = true
|
||||
} else if len(orphanFiles) > 0 {
|
||||
// Best-effort file cleanup.
|
||||
if fileStorage != nil {
|
||||
@@ -171,6 +187,12 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer) error {
|
||||
}
|
||||
log.Info("cleaned up orphaned attachments", "count", len(orphanFiles))
|
||||
}
|
||||
|
||||
if tickFailed {
|
||||
consecutiveFailures++
|
||||
} else {
|
||||
consecutiveFailures = 0
|
||||
}
|
||||
case <-stopMaintenance:
|
||||
return
|
||||
}
|
||||
|
||||
@@ -6,4 +6,4 @@ ALTER TABLE voice_states ADD COLUMN screenshare INTEGER NOT NULL DEFAULT 0;
|
||||
ALTER TABLE channels ADD COLUMN voice_max_users INTEGER NOT NULL DEFAULT 0;
|
||||
ALTER TABLE channels ADD COLUMN voice_quality TEXT;
|
||||
ALTER TABLE channels ADD COLUMN mixing_threshold INTEGER;
|
||||
ALTER TABLE channels ADD COLUMN voice_max_video INTEGER NOT NULL DEFAULT 10;
|
||||
ALTER TABLE channels ADD COLUMN voice_max_video INTEGER NOT NULL DEFAULT 25;
|
||||
|
||||
@@ -592,6 +592,9 @@ func TestHandleVoiceMute_InvalidPayload(t *testing.T) {
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
// Put client in voice so the "not in voice" guard doesn't fire first.
|
||||
ws.SetClientVoiceChID(c, 999)
|
||||
|
||||
raw, _ := json.Marshal(map[string]any{
|
||||
"type": "voice_mute",
|
||||
"payload": "not-an-object",
|
||||
@@ -613,6 +616,9 @@ func TestHandleVoiceDeafen_InvalidPayload(t *testing.T) {
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
// Put client in voice so the "not in voice" guard doesn't fire first.
|
||||
ws.SetClientVoiceChID(c, 999)
|
||||
|
||||
raw, _ := json.Marshal(map[string]any{
|
||||
"type": "voice_deafen",
|
||||
"payload": "not-an-object",
|
||||
@@ -2055,3 +2061,130 @@ func TestHandleChatSend_WithNonNilAvatar(t *testing.T) {
|
||||
t.Error("expected chat_message with non-nil avatar")
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Webhook parse helpers ──────────────────────────────────────────────────
|
||||
|
||||
func TestWebhookParseIdentity_Valid(t *testing.T) {
|
||||
id, err := ws.ParseIdentityForTest("user-42")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if id != 42 {
|
||||
t.Errorf("id = %d, want 42", id)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhookParseIdentity_Invalid(t *testing.T) {
|
||||
_, err := ws.ParseIdentityForTest("invalid")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for invalid identity, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhookParseRoomChannelID_Valid(t *testing.T) {
|
||||
id, err := ws.ParseRoomChannelIDForTest("channel-5")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if id != 5 {
|
||||
t.Errorf("id = %d, want 5", id)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhookParseRoomChannelID_Invalid(t *testing.T) {
|
||||
_, err := ws.ParseRoomChannelIDForTest("bad")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for invalid room name, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Voice control "not in voice" guards ────────────────────────────────────
|
||||
|
||||
func TestHandleVoiceMute_NotInVoice(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "vm-not-in-voice")
|
||||
send := make(chan []byte, 16)
|
||||
c := ws.NewTestClientWithUser(hub, user, 0, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
raw, _ := json.Marshal(map[string]any{
|
||||
"type": "voice_mute",
|
||||
"payload": map[string]any{
|
||||
"muted": true,
|
||||
},
|
||||
})
|
||||
hub.HandleMessageForTest(c, raw)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
code := drainForErrorCode(send, 200*time.Millisecond)
|
||||
if code != "VOICE_ERROR" {
|
||||
t.Errorf("error code = %q, want VOICE_ERROR", code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleVoiceDeafen_NotInVoice(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "vd-not-in-voice")
|
||||
send := make(chan []byte, 16)
|
||||
c := ws.NewTestClientWithUser(hub, user, 0, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
raw, _ := json.Marshal(map[string]any{
|
||||
"type": "voice_deafen",
|
||||
"payload": map[string]any{
|
||||
"deafened": true,
|
||||
},
|
||||
})
|
||||
hub.HandleMessageForTest(c, raw)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
code := drainForErrorCode(send, 200*time.Millisecond)
|
||||
if code != "VOICE_ERROR" {
|
||||
t.Errorf("error code = %q, want VOICE_ERROR", code)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Voice join with invalid quality fallback ───────────────────────────────
|
||||
|
||||
func TestHandleVoiceJoin_InvalidQualityFallsBackToMedium(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "vj-badquality-user")
|
||||
|
||||
vcID, err := database.CreateChannel("badquality-vc", "voice", "", "", 0)
|
||||
if err != nil {
|
||||
t.Fatalf("CreateChannel: %v", err)
|
||||
}
|
||||
_, err = database.Exec("UPDATE channels SET voice_quality = 'garbage' WHERE id = ?", vcID)
|
||||
if err != nil {
|
||||
t.Fatalf("UPDATE: %v", err)
|
||||
}
|
||||
|
||||
send := make(chan []byte, 64)
|
||||
c := ws.NewTestClientWithUser(hub, user, 0, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
raw, _ := json.Marshal(map[string]any{
|
||||
"type": "voice_join",
|
||||
"payload": map[string]any{
|
||||
"channel_id": vcID,
|
||||
},
|
||||
})
|
||||
hub.HandleMessageForTest(c, raw)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
msgs := drainChanTimeout(send, 300*time.Millisecond)
|
||||
for _, msg := range msgs {
|
||||
var env map[string]any
|
||||
if json.Unmarshal(msg, &env) == nil && env["type"] == "voice_config" {
|
||||
p := env["payload"].(map[string]any)
|
||||
if p["quality"] != "medium" {
|
||||
t.Errorf("voice_config quality = %v, want medium", p["quality"])
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Error("expected voice_config with medium quality fallback")
|
||||
}
|
||||
|
||||
@@ -41,3 +41,13 @@ func ParseChannelIDForTest(payload json.RawMessage) (int64, error) {
|
||||
func BuildJSONForTest(v any) []byte {
|
||||
return buildJSON(v)
|
||||
}
|
||||
|
||||
// ParseIdentityForTest exposes parseIdentity for external tests.
|
||||
func ParseIdentityForTest(identity string) (int64, error) {
|
||||
return parseIdentity(identity)
|
||||
}
|
||||
|
||||
// ParseRoomChannelIDForTest exposes parseRoomChannelID for external tests.
|
||||
func ParseRoomChannelIDForTest(roomName string) (int64, error) {
|
||||
return parseRoomChannelID(roomName)
|
||||
}
|
||||
|
||||
@@ -121,6 +121,8 @@ func (h *Hub) handleMessage(c *Client, raw []byte) {
|
||||
h.handleVoiceJoin(c, env.Payload)
|
||||
case "voice_leave":
|
||||
h.handleVoiceLeave(c)
|
||||
case "voice_token_refresh":
|
||||
h.handleVoiceTokenRefresh(c)
|
||||
case "voice_mute":
|
||||
h.handleVoiceMute(c, env.Payload)
|
||||
case "voice_deafen":
|
||||
@@ -214,7 +216,13 @@ func (h *Hub) handleChatSend(c *Client, reqID string, payload json.RawMessage) {
|
||||
if len(p.Attachments) > 0 {
|
||||
linked, linkErr := h.db.LinkAttachmentsToMessage(msgID, p.Attachments)
|
||||
if linkErr != nil {
|
||||
slog.Error("ws handleChatSend LinkAttachments", "err", linkErr)
|
||||
slog.Error("ws handleChatSend LinkAttachments", "err", linkErr, "msg_id", msgID)
|
||||
// Delete the orphaned message so it doesn't persist without its attachments.
|
||||
if delErr := h.db.DeleteMessage(msgID, c.userID, true); delErr != nil {
|
||||
slog.Error("ws handleChatSend DeleteMessage (cleanup)", "err", delErr, "msg_id", msgID)
|
||||
}
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to send message with attachments"))
|
||||
return
|
||||
}
|
||||
if linked > 0 {
|
||||
attMap, attErr := h.db.GetAttachmentsByMessageIDs([]int64{msgID})
|
||||
|
||||
+28
-5
@@ -104,6 +104,17 @@ func (h *Hub) SetLiveKit(lk *LiveKitClient) {
|
||||
h.livekit = lk
|
||||
}
|
||||
|
||||
// LiveKitHealthCheck probes the LiveKit server for connectivity.
|
||||
// It tries the SDK client first (ListRooms), and falls back to an HTTP probe
|
||||
// if a managed process is configured. Returns false with a reason if LiveKit
|
||||
// is not configured or unreachable.
|
||||
func (h *Hub) LiveKitHealthCheck() (bool, error) {
|
||||
if h.livekit == nil {
|
||||
return false, fmt.Errorf("not configured")
|
||||
}
|
||||
return h.livekit.HealthCheck()
|
||||
}
|
||||
|
||||
// SetLiveKitProcess sets the LiveKit process manager on the hub.
|
||||
func (h *Hub) SetLiveKitProcess(p *LiveKitProcess) {
|
||||
h.lkProcess = p
|
||||
@@ -113,7 +124,7 @@ func (h *Hub) SetLiveKitProcess(p *LiveKitProcess) {
|
||||
// Must be called in its own goroutine.
|
||||
//
|
||||
// A panic recovery wrapper restarts the select loop automatically. If the hub
|
||||
// panics more than 5 times within a 60-second window it stops permanently to
|
||||
// panics more than 3 times within a 60-second window it stops permanently to
|
||||
// avoid a tight crash loop.
|
||||
func (h *Hub) Run() {
|
||||
var panicCount int
|
||||
@@ -140,7 +151,7 @@ func (h *Hub) Run() {
|
||||
"panic_count", panicCount,
|
||||
"stack", string(buf[:n]))
|
||||
|
||||
if panicCount >= 5 {
|
||||
if panicCount >= 3 {
|
||||
slog.Error("hub: too many panics in 60s, stopping")
|
||||
return
|
||||
}
|
||||
@@ -172,7 +183,7 @@ func (h *Hub) Run() {
|
||||
}()
|
||||
|
||||
// If we reach here without a panic recovery continuing, stop.
|
||||
if panicCount >= 5 {
|
||||
if panicCount >= 3 {
|
||||
return
|
||||
}
|
||||
// If stop was signaled, exit.
|
||||
@@ -278,13 +289,25 @@ func (h *Hub) Unregister(c *Client) {
|
||||
|
||||
// BroadcastToChannel enqueues msg for delivery to all clients subscribed to
|
||||
// channelID. When channelID is 0 the message is sent to every connected client.
|
||||
// Non-blocking: if the broadcast channel is full the message is dropped with a warning.
|
||||
func (h *Hub) BroadcastToChannel(channelID int64, msg []byte) {
|
||||
h.broadcast <- broadcastMsg{channelID: channelID, msg: msg}
|
||||
select {
|
||||
case h.broadcast <- broadcastMsg{channelID: channelID, msg: msg}:
|
||||
default:
|
||||
slog.Warn("hub: broadcast channel full, dropping message",
|
||||
"channel_id", channelID, "msg_len", len(msg))
|
||||
}
|
||||
}
|
||||
|
||||
// BroadcastToAll enqueues msg for delivery to every connected client.
|
||||
// Non-blocking: if the broadcast channel is full the message is dropped with a warning.
|
||||
func (h *Hub) BroadcastToAll(msg []byte) {
|
||||
h.broadcast <- broadcastMsg{channelID: 0, msg: msg}
|
||||
select {
|
||||
case h.broadcast <- broadcastMsg{channelID: 0, msg: msg}:
|
||||
default:
|
||||
slog.Warn("hub: broadcast channel full, dropping global message",
|
||||
"msg_len", len(msg))
|
||||
}
|
||||
}
|
||||
|
||||
// BroadcastServerRestart sends a server_restart message to all connected clients.
|
||||
|
||||
@@ -19,6 +19,9 @@ import (
|
||||
)
|
||||
|
||||
// tokenTTL is the validity duration for generated LiveKit access tokens.
|
||||
// Kept at 4h to limit exposure if a token is leaked — there is no server-side
|
||||
// revocation for LiveKit JWTs. The client can request a refresh via
|
||||
// voice_token_refresh before expiry.
|
||||
const tokenTTL = 4 * time.Hour
|
||||
|
||||
// LiveKitClient provides token generation and room management via
|
||||
@@ -157,6 +160,20 @@ func (c *LiveKitClient) CountVideoTracks(channelID int64) (int, error) {
|
||||
return count, nil
|
||||
}
|
||||
|
||||
// HealthCheck verifies connectivity to the LiveKit server by listing rooms.
|
||||
// Returns true if the server responds successfully.
|
||||
func (c *LiveKitClient) HealthCheck() (bool, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
|
||||
_, err := c.roomSvc.ListRooms(ctx, &livekit.ListRoomsRequest{})
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("livekit health check failed: %w", err)
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// wsToHTTP converts a WebSocket URL to an HTTP URL.
|
||||
func wsToHTTP(wsURL string) string {
|
||||
switch {
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
@@ -20,9 +21,10 @@ import (
|
||||
|
||||
// LiveKitProcess manages the companion livekit-server binary.
|
||||
type LiveKitProcess struct {
|
||||
cfg *config.VoiceConfig
|
||||
tlsCfg *config.TLSConfig
|
||||
dataDir string
|
||||
cfg *config.VoiceConfig
|
||||
tlsCfg *config.TLSConfig
|
||||
dataDir string
|
||||
httpClient *http.Client // for health checks — no redirect following
|
||||
|
||||
mu sync.Mutex
|
||||
cmd *exec.Cmd
|
||||
@@ -39,6 +41,11 @@ func NewLiveKitProcess(cfg *config.VoiceConfig, tlsCfg *config.TLSConfig, dataDi
|
||||
cfg: cfg,
|
||||
tlsCfg: tlsCfg,
|
||||
dataDir: dataDir,
|
||||
httpClient: &http.Client{
|
||||
CheckRedirect: func(*http.Request, []*http.Request) error {
|
||||
return http.ErrUseLastResponse
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -56,7 +63,7 @@ rtc:
|
||||
port_range_end: 60000
|
||||
use_external_ip: true
|
||||
keys:
|
||||
%s: %s
|
||||
"%s": "%s"
|
||||
logging:
|
||||
level: info
|
||||
`, p.cfg.LiveKitAPIKey, p.cfg.LiveKitAPISecret)
|
||||
@@ -100,8 +107,18 @@ func (p *LiveKitProcess) Start() error {
|
||||
}
|
||||
|
||||
// runLoop starts and restarts the process until stopped or context cancelled.
|
||||
// Uses exponential backoff (3s → 6s → 12s … up to 60s) and stops after 10
|
||||
// consecutive rapid failures (process exits within 30 seconds).
|
||||
func (p *LiveKitProcess) runLoop(ctx context.Context, cfgPath string) {
|
||||
const restartDelay = 3 * time.Second
|
||||
const (
|
||||
baseDelay = 3 * time.Second
|
||||
maxDelay = 60 * time.Second
|
||||
maxRetries = 10
|
||||
stableAfter = 30 * time.Second // reset counter if process runs longer than this
|
||||
)
|
||||
|
||||
rapidFailures := 0
|
||||
delay := baseDelay
|
||||
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
@@ -111,6 +128,7 @@ func (p *LiveKitProcess) runLoop(ctx context.Context, cfgPath string) {
|
||||
cmd := exec.CommandContext(ctx, p.cfg.LiveKitBinaryPath, "--config", cfgPath)
|
||||
cmd.Stdout = os.Stdout
|
||||
cmd.Stderr = os.Stderr
|
||||
cmd.WaitDelay = 6 * time.Second // bound Wait to prevent goroutine leak on Windows
|
||||
|
||||
p.mu.Lock()
|
||||
if p.stopped {
|
||||
@@ -122,8 +140,10 @@ func (p *LiveKitProcess) runLoop(ctx context.Context, cfgPath string) {
|
||||
|
||||
slog.Info("livekit: starting process",
|
||||
"binary", p.cfg.LiveKitBinaryPath,
|
||||
"config", cfgPath)
|
||||
"config", cfgPath,
|
||||
"rapid_failures", rapidFailures)
|
||||
|
||||
startTime := time.Now()
|
||||
err := cmd.Run()
|
||||
|
||||
p.mu.Lock()
|
||||
@@ -136,18 +156,39 @@ func (p *LiveKitProcess) runLoop(ctx context.Context, cfgPath string) {
|
||||
return
|
||||
}
|
||||
|
||||
// If the process ran for a while, it was stable — reset backoff.
|
||||
if time.Since(startTime) > stableAfter {
|
||||
rapidFailures = 0
|
||||
delay = baseDelay
|
||||
} else {
|
||||
rapidFailures++
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
slog.Error("livekit: process exited unexpectedly",
|
||||
"error", err,
|
||||
"restart_delay", restartDelay)
|
||||
"rapid_failures", rapidFailures,
|
||||
"restart_delay", delay)
|
||||
}
|
||||
|
||||
if rapidFailures >= maxRetries {
|
||||
slog.Error("livekit: too many rapid failures, giving up",
|
||||
"rapid_failures", rapidFailures)
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case <-time.After(restartDelay):
|
||||
case <-time.After(delay):
|
||||
slog.Info("livekit: restarting process")
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
// Exponential backoff capped at maxDelay.
|
||||
delay *= 2
|
||||
if delay > maxDelay {
|
||||
delay = maxDelay
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -158,6 +199,28 @@ func (p *LiveKitProcess) IsRunning() bool {
|
||||
return p.cmd != nil && p.cmd.Process != nil
|
||||
}
|
||||
|
||||
// HealthCheck probes the LiveKit HTTP endpoint to verify it is accepting
|
||||
// connections. Returns true if the server responds (any status code).
|
||||
func (p *LiveKitProcess) HealthCheck() (bool, error) {
|
||||
httpURL := wsToHTTP(p.cfg.LiveKitURL)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, httpURL, nil)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("creating health check request: %w", err)
|
||||
}
|
||||
|
||||
resp, err := p.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("livekit health check failed: %w", err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Stop gracefully stops the companion process.
|
||||
func (p *LiveKitProcess) Stop() {
|
||||
p.mu.Lock()
|
||||
|
||||
@@ -22,7 +22,7 @@ import (
|
||||
// RoomEvent.ActiveSpeakersChanged (lower latency than webhooks).
|
||||
func (h *Hub) NewLiveKitWebhookHandler(apiKey, apiSecret string) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
body, err := io.ReadAll(r.Body)
|
||||
body, err := io.ReadAll(io.LimitReader(r.Body, 64*1024))
|
||||
if err != nil {
|
||||
slog.Error("livekit webhook: read body failed", "error", err)
|
||||
http.Error(w, "bad request", http.StatusBadRequest)
|
||||
@@ -151,9 +151,11 @@ func (h *Hub) handleWebhookParticipantLeft(event *livekit.WebhookEvent) {
|
||||
h.mu.RUnlock()
|
||||
|
||||
if exists {
|
||||
// Only clean up if the user is still in the channel that fired the
|
||||
// webhook. If they've already moved or left, don't touch their state.
|
||||
currentChID := c.getVoiceChID()
|
||||
if currentChID == channelID {
|
||||
c.setVoiceChID(0)
|
||||
c.clearVoiceChID()
|
||||
|
||||
if h.db != nil {
|
||||
_ = h.db.LeaveVoiceChannel(userID)
|
||||
|
||||
@@ -3,6 +3,7 @@ package ws
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
"github.com/owncord/server/db"
|
||||
)
|
||||
@@ -155,6 +156,7 @@ type serverRestartPayload struct {
|
||||
func buildJSON(v any) []byte {
|
||||
b, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
slog.Error("buildJSON marshal failed", "error", err, "type", fmt.Sprintf("%T", v))
|
||||
// Fallback: send a generic error rather than panicking.
|
||||
b, _ = json.Marshal(map[string]string{"type": "error", "message": "internal marshal error"})
|
||||
}
|
||||
|
||||
@@ -92,6 +92,7 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun
|
||||
writeCtx, writeCancel := context.WithCancel(ctx)
|
||||
go writePump(writeCtx, conn, c)
|
||||
readPump(ctx, conn, hub, c)
|
||||
c.closeSend()
|
||||
writeCancel()
|
||||
return
|
||||
}
|
||||
@@ -120,9 +121,12 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun
|
||||
hub.BroadcastToAll(buildPresenceMsg(user.ID, "online"))
|
||||
|
||||
// writePump runs in background; readPump blocks.
|
||||
// When readPump returns (disconnect), close the send channel first
|
||||
// so writePump drains any remaining messages, then cancel its context.
|
||||
writeCtx, writeCancel := context.WithCancel(ctx)
|
||||
go writePump(writeCtx, conn, c)
|
||||
readPump(ctx, conn, hub, c)
|
||||
c.closeSend()
|
||||
writeCancel()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,16 +13,21 @@ const (
|
||||
voiceScreenshareWindow = time.Second
|
||||
)
|
||||
|
||||
// voiceQualities maps accepted voice quality presets to their target bitrate
|
||||
// in bits/s. This is the single source of truth — voice_join.go validates
|
||||
// against these keys, qualityBitrate looks up the value.
|
||||
var voiceQualities = map[string]int{
|
||||
"low": 32000,
|
||||
"medium": 64000,
|
||||
"high": 128000,
|
||||
}
|
||||
|
||||
// qualityBitrate returns the target audio bitrate in bits/s based on a quality preset.
|
||||
func qualityBitrate(quality string) int {
|
||||
switch quality {
|
||||
case "low":
|
||||
return 32000
|
||||
case "high":
|
||||
return 128000
|
||||
default:
|
||||
return 64000
|
||||
if bitrate, ok := voiceQualities[quality]; ok {
|
||||
return bitrate
|
||||
}
|
||||
return voiceQualities["medium"]
|
||||
}
|
||||
|
||||
// broadcastVoiceStateUpdate fetches the current voice state for the client
|
||||
|
||||
@@ -13,6 +13,11 @@ import (
|
||||
// 2. Updates DB.
|
||||
// 3. Broadcasts voice_state update to channel.
|
||||
func (h *Hub) handleVoiceMute(c *Client, payload json.RawMessage) {
|
||||
if c.getVoiceChID() == 0 {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeVoiceError, "not in a voice channel"))
|
||||
return
|
||||
}
|
||||
|
||||
var p struct {
|
||||
Muted bool `json:"muted"`
|
||||
}
|
||||
@@ -36,6 +41,11 @@ func (h *Hub) handleVoiceMute(c *Client, payload json.RawMessage) {
|
||||
// 2. Updates DB.
|
||||
// 3. Broadcasts voice_state update to channel.
|
||||
func (h *Hub) handleVoiceDeafen(c *Client, payload json.RawMessage) {
|
||||
if c.getVoiceChID() == 0 {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeVoiceError, "not in a voice channel"))
|
||||
return
|
||||
}
|
||||
|
||||
var p struct {
|
||||
Deafened bool `json:"deafened"`
|
||||
}
|
||||
@@ -58,7 +68,7 @@ func (h *Hub) handleVoiceDeafen(c *Client, payload json.RawMessage) {
|
||||
// 1. Rate limits at 2/sec per user.
|
||||
// 2. Checks USE_VIDEO permission.
|
||||
// 3. Parses enabled bool.
|
||||
// 4. Enforces MaxVideo limit via LiveKit.
|
||||
// 4. Enforces MaxVideo limit via DB count (race-free).
|
||||
// 5. Updates DB.
|
||||
// 6. Broadcasts voice_state update to channel.
|
||||
func (h *Hub) handleVoiceCamera(c *Client, payload json.RawMessage) {
|
||||
@@ -87,12 +97,13 @@ func (h *Hub) handleVoiceCamera(c *Client, payload json.RawMessage) {
|
||||
}
|
||||
|
||||
// Enforce MaxVideo limit when enabling camera.
|
||||
// Count from DB (race-free via SQLite serialization) instead of LiveKit API.
|
||||
if p.Enabled {
|
||||
ch, chErr := h.db.GetChannel(voiceChID)
|
||||
if chErr == nil && ch != nil && ch.VoiceMaxVideo > 0 && h.livekit != nil {
|
||||
videoCount, countErr := h.livekit.CountVideoTracks(voiceChID)
|
||||
if chErr == nil && ch != nil && ch.VoiceMaxVideo > 0 {
|
||||
videoCount, countErr := h.db.CountActiveCameras(voiceChID)
|
||||
if countErr != nil {
|
||||
slog.Error("handleVoiceCamera CountVideoTracks", "err", countErr, "channel_id", voiceChID)
|
||||
slog.Error("handleVoiceCamera CountActiveCameras", "err", countErr, "channel_id", voiceChID)
|
||||
} else if videoCount >= ch.VoiceMaxVideo {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeVideoLimit,
|
||||
fmt.Sprintf("maximum %d video streams reached", ch.VoiceMaxVideo)))
|
||||
|
||||
+85
-9
@@ -2,11 +2,20 @@ package ws
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/owncord/server/permissions"
|
||||
)
|
||||
|
||||
// validVoiceQuality returns true if q is an accepted voice quality preset.
|
||||
// Uses voiceQualities (defined in voice_broadcast.go) as the single source of truth.
|
||||
func validVoiceQuality(q string) bool {
|
||||
_, ok := voiceQualities[q]
|
||||
return ok
|
||||
}
|
||||
|
||||
// handleVoiceJoin processes a voice_join message.
|
||||
// 1. Parses channel_id.
|
||||
// 2. Checks CONNECT_VOICE permission.
|
||||
@@ -73,30 +82,39 @@ func (h *Hub) handleVoiceJoin(c *Client, payload json.RawMessage) {
|
||||
c.setVoiceChID(channelID)
|
||||
|
||||
// Generate LiveKit token if LiveKit client is available.
|
||||
// Token generation failure is fatal — without a token the client cannot
|
||||
// connect to the SFU, so we must roll back the DB join.
|
||||
if h.livekit != nil {
|
||||
if c.user == nil {
|
||||
slog.Error("handleVoiceJoin: nil user on client", "user_id", c.userID)
|
||||
h.rollbackVoiceJoin(c, channelID)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "not authenticated"))
|
||||
return
|
||||
}
|
||||
canPublish := true
|
||||
// Derive publish permissions from role — prevents SFU-level bypass
|
||||
// when client connects directly via direct_url.
|
||||
canPublish := h.hasChannelPerm(c, channelID, permissions.SpeakVoice)
|
||||
canSubscribe := true
|
||||
token, tokenErr := h.livekit.GenerateToken(c.userID, c.user.Username, channelID, canPublish, canSubscribe)
|
||||
if tokenErr != nil {
|
||||
slog.Error("ws handleVoiceJoin GenerateToken", "err", tokenErr, "user_id", c.userID)
|
||||
// Non-fatal: voice join still succeeds at the DB/state level.
|
||||
} else {
|
||||
// Send both proxy path and direct URL. The client uses direct_url
|
||||
// when on localhost (avoids self-signed TLS issues with WebView
|
||||
// fetch) and falls back to the /livekit proxy for remote clients.
|
||||
c.sendMsg(buildVoiceToken(channelID, token, "/livekit", h.livekit.URL()))
|
||||
h.rollbackVoiceJoin(c, channelID)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to generate voice token"))
|
||||
return
|
||||
}
|
||||
// Send both proxy path and direct URL. The client uses direct_url
|
||||
// when on localhost (avoids self-signed TLS issues with WebView
|
||||
// fetch) and falls back to the /livekit proxy for remote clients.
|
||||
c.sendMsg(buildVoiceToken(channelID, token, "/livekit", h.livekit.URL()))
|
||||
}
|
||||
|
||||
// Get and broadcast the joiner's state.
|
||||
// Get and broadcast the joiner's state. Failure here means other users
|
||||
// won't see the join (ghost state), so roll back to avoid inconsistency.
|
||||
state, err := h.db.GetVoiceState(c.userID)
|
||||
if err != nil || state == nil {
|
||||
slog.Error("ws handleVoiceJoin GetVoiceState", "err", err, "user_id", c.userID)
|
||||
h.rollbackVoiceJoin(c, channelID)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to join voice channel"))
|
||||
return
|
||||
}
|
||||
|
||||
@@ -119,10 +137,68 @@ func (h *Hub) handleVoiceJoin(c *Client, payload json.RawMessage) {
|
||||
// Send voice_config to the joiner.
|
||||
quality := "medium"
|
||||
if ch.VoiceQuality != nil && *ch.VoiceQuality != "" {
|
||||
quality = *ch.VoiceQuality
|
||||
q := *ch.VoiceQuality
|
||||
if validVoiceQuality(q) {
|
||||
quality = q
|
||||
} else {
|
||||
slog.Warn("ws handleVoiceJoin invalid voice quality, using default",
|
||||
"quality", q, "channel_id", channelID)
|
||||
}
|
||||
}
|
||||
bitrate := qualityBitrate(quality)
|
||||
c.sendMsg(buildVoiceConfig(channelID, quality, bitrate, maxUsers))
|
||||
|
||||
slog.Info("voice join", "user_id", c.userID, "channel_id", channelID)
|
||||
}
|
||||
|
||||
// handleVoiceTokenRefresh generates a fresh LiveKit token for a client
|
||||
// that is already in a voice channel. This lets clients request a new token
|
||||
// (e.g. before a manual reconnect) without leaving and rejoining voice.
|
||||
func (h *Hub) handleVoiceTokenRefresh(c *Client) {
|
||||
ratKey := fmt.Sprintf("voice_token_refresh:%d", c.userID)
|
||||
if !h.limiter.Allow(ratKey, 1, 60*time.Second) {
|
||||
c.sendMsg(buildRateLimitError("token refresh rate limited", 60))
|
||||
return
|
||||
}
|
||||
|
||||
channelID := c.getVoiceChID()
|
||||
if channelID == 0 {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "not in voice"))
|
||||
return
|
||||
}
|
||||
|
||||
if h.livekit == nil {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "voice not configured"))
|
||||
return
|
||||
}
|
||||
|
||||
if c.user == nil {
|
||||
slog.Error("handleVoiceTokenRefresh: nil user on client", "user_id", c.userID)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "not authenticated"))
|
||||
return
|
||||
}
|
||||
|
||||
canPublish := h.hasChannelPerm(c, channelID, permissions.SpeakVoice)
|
||||
canSubscribe := true
|
||||
token, err := h.livekit.GenerateToken(c.userID, c.user.Username, channelID, canPublish, canSubscribe)
|
||||
if err != nil {
|
||||
slog.Error("ws handleVoiceTokenRefresh GenerateToken", "err", err, "user_id", c.userID)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to generate voice token"))
|
||||
return
|
||||
}
|
||||
|
||||
c.sendMsg(buildVoiceToken(channelID, token, "/livekit", h.livekit.URL()))
|
||||
slog.Info("voice token refreshed", "user_id", c.userID, "channel_id", channelID)
|
||||
}
|
||||
|
||||
// rollbackVoiceJoin undoes a partially-completed voice join: clears the
|
||||
// client's voice channel ID, removes the DB voice state row, and broadcasts
|
||||
// voice_leave so other clients don't see a ghost participant.
|
||||
func (h *Hub) rollbackVoiceJoin(c *Client, channelID int64) {
|
||||
c.clearVoiceChID()
|
||||
if err := h.db.LeaveVoiceChannel(c.userID); err != nil {
|
||||
slog.Error("ws rollbackVoiceJoin LeaveVoiceChannel", "err", err,
|
||||
"user_id", c.userID, "channel_id", channelID)
|
||||
}
|
||||
h.BroadcastToAll(buildVoiceLeave(channelID, c.userID))
|
||||
}
|
||||
|
||||
@@ -25,7 +25,7 @@ func (h *Hub) handleVoiceLeave(c *Client) {
|
||||
// Remove from LiveKit (best-effort).
|
||||
if h.livekit != nil {
|
||||
if err := h.livekit.RemoveParticipant(oldChID, c.userID); err != nil {
|
||||
slog.Debug("handleVoiceLeave RemoveParticipant (may already be gone)",
|
||||
slog.Warn("handleVoiceLeave RemoveParticipant failed (may already be gone)",
|
||||
"err", err, "user_id", c.userID, "channel_id", oldChID)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
# TODOS
|
||||
|
||||
Deferred work items from engineering review (2026-03-21).
|
||||
|
||||
## Voice E2E Test Infrastructure
|
||||
|
||||
**What:** Add E2E test infrastructure for voice flows (voice_join → LiveKit connect → audio → voice_leave).
|
||||
**Why:** The voice path is critical UX with zero automated E2E coverage. Unit tests cover handlers and controllers, but nothing tests the full integration.
|
||||
**Pros:** Catches integration bugs between server + LiveKit + client.
|
||||
**Cons:** Requires LiveKit binary in CI, WebRTC support in test browser, ~200 lines of test infra.
|
||||
**Context:** The existing native E2E infrastructure (WebView2 CDP) could be extended. Needs CI setup first.
|
||||
**Depends on:** LiveKit binary available in CI environment.
|
||||
**Added:** 2026-03-21 (eng review of feature/livekit-migration)
|
||||
Reference in New Issue
Block a user