// Package ws provides the LiveKit companion process manager. // // LiveKitProcess manages the lifecycle of a livekit-server binary running // alongside chatserver. It auto-generates a minimal livekit.yaml config, // starts the process, monitors health, and restarts on crash. package ws import ( "context" "fmt" "log/slog" "net/http" "os" "os/exec" "path/filepath" "strings" "time" "github.com/J3vb/OwnCord/Server/config" "github.com/J3vb/OwnCord/Server/syncutil" ) // LiveKitProcess manages the companion livekit-server binary. type LiveKitProcess struct { cfg *config.VoiceConfig tlsCfg *config.TLSConfig dataDir string httpClient *http.Client // for health checks — no redirect following mu syncutil.Mutex cmd *exec.Cmd cancel context.CancelFunc stopped bool runDone chan struct{} // closed by runLoop when cmd.Wait() returns loopDone chan struct{} // closed when runLoop exits entirely } // NewLiveKitProcess creates a new process manager. It does not start the // process — call Start() to launch the LiveKit server. // The tlsCfg is used to configure TLS on the LiveKit server using the same // certs as OwnCord (avoids mixed-content blocks in WebView2). func NewLiveKitProcess(cfg *config.VoiceConfig, tlsCfg *config.TLSConfig, dataDir string) *LiveKitProcess { return &LiveKitProcess{ cfg: cfg, tlsCfg: tlsCfg, dataDir: dataDir, httpClient: &http.Client{ // Own the transport rather than inheriting http.DefaultTransport: // its pool is shared process-wide, and httptest.Server.Close closes // its idle connections, which severs in-flight health checks. Transport: http.DefaultTransport.(*http.Transport).Clone(), CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, }, } } // autoGeneratedMarker identifies a livekit.yaml written by OwnCord. A file // without this marker is treated as user-managed and never overwritten. const autoGeneratedMarker = "Auto-generated by OwnCord" // generateConfig writes a minimal livekit.yaml for the companion process. // If the file exists and lacks the auto-generated marker, it is treated as // user-managed and left untouched so operators can set LiveKit options // OwnCord does not model (ips.includes, interfaces, stun_servers, ...). func (p *LiveKitProcess) generateConfig() (string, error) { cfgPath := filepath.Join(p.dataDir, "livekit.yaml") if existing, err := os.ReadFile(cfgPath); err == nil { // An empty/whitespace-only file is a truncated leftover, not a // user-managed config — regenerate it rather than wedging LiveKit. if len(strings.TrimSpace(string(existing))) > 0 && !strings.Contains(string(existing), autoGeneratedMarker) { slog.Info("livekit: livekit.yaml is user-managed (auto-generated marker absent), not overwriting", "path", cfgPath) slog.Warn("livekit: ensure the keys entry in your livekit.yaml matches voice.livekit_api_key / voice.livekit_api_secret") return cfgPath, nil } } else if !os.IsNotExist(err) { return "", fmt.Errorf("reading existing livekit config: %w", err) } // No TURN TLS config — LiveKit signaling is proxied through OwnCord's // HTTPS server at /livekit/*, so no separate TLS is needed on LiveKit. // Sanitize credentials for safe YAML interpolation: reject strings // containing characters that could break YAML structure. // // The error names the offending config field but never echoes the // offending byte: this error is wrapped by Start() and logged by the // caller, so quoting a character from the key or secret would write a // byte of a credential to the server log in clear text. const unsafeYAML = ":#{}\n\r\"\\" for _, cred := range []struct { field string value string }{ {"voice.livekit_api_key", p.cfg.LiveKitAPIKey}, {"voice.livekit_api_secret", p.cfg.LiveKitAPISecret}, } { if strings.ContainsAny(cred.value, unsafeYAML) { return "", fmt.Errorf(`%s contains a character that cannot be safely written to livekit.yaml (one of : # { } " \ CR LF)`, cred.field) } } // Build node_ip line only when configured (required for remote users behind NAT). // Validate: must be a plain IP address (no YAML-breaking chars). nodeIPLine := "" if p.cfg.NodeIP != "" { for _, ch := range p.cfg.NodeIP { if ch == '"' || ch == '\\' || ch == '\n' || ch == '\r' || ch == '#' || ch == '{' || ch == '}' { return "", fmt.Errorf("node_ip contains unsafe character %q", string(ch)) } } nodeIPLine = fmt.Sprintf("\n node_ip: %q", p.cfg.NodeIP) } // Advertise LAN host candidates alongside the external mapping so clients // on the local network can reach a dual-homed (LAN + public IP) server. advertiseInternalLine := "" if p.cfg.AdvertiseInternalIP { advertiseInternalLine = "\n advertise_internal_ip: true" } content := fmt.Sprintf(`# Auto-generated by OwnCord — regenerated on every server start. # To manage this file yourself (custom rtc options, multiple interfaces, etc.), # delete the first line above; OwnCord will then leave the file untouched. # Your keys entry must still match voice.livekit_api_key / # voice.livekit_api_secret in config.yaml. port: 7880 rtc: port_range_start: 50000 port_range_end: 60000 use_external_ip: true%s%s pli_throttle: low_quality: 500ms mid_quality: 1s high_quality: 1s keys: "%s": "%s" logging: level: info `, nodeIPLine, advertiseInternalLine, p.cfg.LiveKitAPIKey, p.cfg.LiveKitAPISecret) if err := os.MkdirAll(p.dataDir, 0o750); err != nil { return "", fmt.Errorf("creating data dir: %w", err) } if err := os.WriteFile(cfgPath, []byte(content), 0o600); err != nil { return "", fmt.Errorf("writing livekit config: %w", err) } return cfgPath, nil } // Start launches the livekit-server binary. If LiveKitBinaryPath is empty // and auto-download is disabled, this is a no-op (assumes LiveKit is managed // externally). With auto-download enabled, the binary is fetched (and // checksum-verified) in the background before the process starts, so server // startup is never blocked on the download. func (p *LiveKitProcess) Start() error { if p.cfg.LiveKitBinaryPath == "" && !p.cfg.AutoDownloadLiveKit { slog.Info("livekit: no binary path configured, assuming externally managed") return nil } p.mu.Lock() defer p.mu.Unlock() if p.cmd != nil { return fmt.Errorf("livekit process already running") } cfgPath, err := p.generateConfig() if err != nil { return fmt.Errorf("generating livekit config: %w", err) } ctx, cancel := context.WithCancel(context.Background()) p.cancel = cancel p.loopDone = make(chan struct{}) go func() { defer func() { p.mu.Lock() if p.loopDone != nil { close(p.loopDone) } p.mu.Unlock() }() binPath := p.cfg.LiveKitBinaryPath if binPath == "" { resolved, dlErr := p.resolveBinary(ctx) if dlErr != nil { if ctx.Err() == nil { slog.Error("livekit: auto-download failed — voice stays offline until livekit-server is available", "error", dlErr, "hint", "check the network, or set voice.livekit_binary in config.yaml to a binary you provide") } return } binPath = resolved } p.runLoop(ctx, cfgPath, binPath) }() return nil } // resolveBinary downloads (or reuses a cached copy of) the pinned // livekit-server release, retrying a few times so a transient network hiccup // at boot does not permanently disable voice until the next restart. func (p *LiveKitProcess) resolveBinary(ctx context.Context) (string, error) { const ( attempts = 3 retryDelay = 15 * time.Second ) var lastErr error for i := range attempts { if i > 0 { slog.Warn("livekit: retrying download", "attempt", i+1, "error", lastErr) select { case <-time.After(retryDelay): case <-ctx.Done(): return "", ctx.Err() } } attemptCtx, cancel := context.WithTimeout(ctx, 10*time.Minute) path, err := EnsureLiveKitBinary(attemptCtx, p.dataDir, p.cfg.LiveKitVersion) cancel() if err == nil { return path, nil } lastErr = err } return "", lastErr } // 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). loopDone is // closed by the Start goroutine that calls this. func (p *LiveKitProcess) runLoop(ctx context.Context, cfgPath, binPath string) { 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 { return } cmd := exec.CommandContext(ctx, binPath, "--config", cfgPath) //nolint:gosec // G204: binary path from trusted server config or verified download cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr cmd.WaitDelay = 6 * time.Second // bound Wait to prevent goroutine leak on Windows if attr := liveKitSysProcAttr(); attr != nil { // Linux: die with a parent that never ran its teardown (kill -9, // OOM, restart-backstop force-exit) instead of orphaning with // 7880/UDP still bound. cmd.SysProcAttr = attr } slog.Info("livekit: starting process", "binary", binPath, "config", cfgPath, "rapid_failures", rapidFailures) startTime := time.Now() p.mu.Lock() if p.stopped { p.mu.Unlock() return } // Start inside the critical section that publishes p.cmd: Start is // what writes cmd.Process, which IsRunning and Stop read under p.mu — // started after the unlock, that write races every such read. err := cmd.Start() if err == nil { p.cmd = cmd p.runDone = make(chan struct{}) } p.mu.Unlock() if err == nil { err = cmd.Wait() } p.mu.Lock() p.cmd = nil if p.runDone != nil { close(p.runDone) p.runDone = nil } stopped := p.stopped p.mu.Unlock() if stopped || ctx.Err() != nil { slog.Info("livekit: process stopped") 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, "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(delay): slog.Info("livekit: restarting process") case <-ctx.Done(): return } // Exponential backoff capped at maxDelay. delay *= 2 if delay > maxDelay { delay = maxDelay } } } // IsRunning returns true if the companion process is currently running. func (p *LiveKitProcess) IsRunning() bool { p.mu.Lock() defer p.mu.Unlock() 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(ctx context.Context) (bool, error) { httpURL := wsToHTTP(p.cfg.LiveKitURL) ctx, cancel := context.WithTimeout(ctx, 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. // It cancels the context (which signals runLoop) and waits up to 5 seconds // for the process to exit. The actual cmd.Wait() is done by runLoop — we only // monitor the process here to avoid calling exec.Cmd.Wait() twice (which has // undefined behavior). func (p *LiveKitProcess) Stop() { p.mu.Lock() p.stopped = true cancel := p.cancel cmd := p.cmd done := p.runDone loopDone := p.loopDone p.mu.Unlock() if cancel != nil { cancel() } // Wait for runLoop's cmd.Wait() to return (which closes runDone). // This avoids calling cmd.Wait() or cmd.Process.Wait() from a second // goroutine, which is unsafe on Windows. if done != nil { select { case <-done: slog.Info("livekit: process exited cleanly") case <-time.After(5 * time.Second): slog.Warn("livekit: process did not exit in time, killing") if cmd != nil && cmd.Process != nil { _ = cmd.Process.Kill() } } } // Wait for the entire runLoop goroutine to finish, ensuring no // new iteration can start after Stop returns. if loopDone != nil { select { case <-loopDone: case <-time.After(5 * time.Second): } } }