guestlink/main.go

// guestlink - self-hosted replacement for Cleanfeed on the MMO rig.
//
// One remote guest joins from a browser at https://guest.mmo.show/j/<token>.
//
//	guest's voice  -> pacat -> PipeWire sink "motu_guest"  (From Computer 1-2, the guest strip)
//	mix-minus      <- parec <- "motu_mixminus.monitor"      (To Computer 3-4, show minus guest)
//
// Plus an anonymous text chat between the guest and the host console.
//
// Two HTTP listeners:
//
//	-public (8790): guest page + guest websocket. This is what the public reverse proxy forwards to.
//	-host   (8791): host console. LAN only, never proxied.
//
// Media: one fixed UDP port (-udp 50090) shared by all ICE traffic.
//
// Audio safety: the parec/pacat streams are ordinary PipeWire client streams (exactly what
// Chromium did for Cleanfeed). They never reopen the MOTU. node.dont-fallback/dont-reconnect stop PipeWire from
// falling back to the default sink if motu_guest disappears.
package main

import (
	"crypto/rand"
	"embed"
	"encoding/base64"
	"encoding/json"
	"flag"
	"io/fs"
	"log"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
	"unicode/utf8"

	"github.com/gorilla/websocket"
	"github.com/pion/ice/v4"
	"github.com/pion/webrtc/v4"
)

//go:embed web
var webFS embed.FS

var (
	publicAddr      = flag.String("public", "0.0.0.0:8790", "guest-facing listener (proxied by NPM)")
	hostAddr        = flag.String("host", "0.0.0.0:8791", "host console listener (LAN only)")
	udpPort         = flag.Int("udp", 50090, "single UDP port for all WebRTC media")
	publicHost      = flag.String("public-host", "guest.mmo.show", "hostname; its A record is advertised as our public IP")
	publicIPFlag    = flag.String("public-ip", "", "override the public IP instead of resolving -public-host")
	lanCIDR         = flag.String("lan", "192.168.50.0/24", "LAN whose interface address is offered as a host candidate")
	sinkName        = flag.String("sink", "motu_guest", "PipeWire sink the guest's voice is played into")
	sourceName      = flag.String("source", "motu_mixminus.monitor", "PipeWire source sent to the guest")
	hostLabel       = flag.String("host-label", "Host", "name the guest sees on host chat messages")
	testPlayoutFile = flag.String("test-playout-file", "", "TEST ONLY: write guest audio to this raw file instead of the sink")
	stateDir        = flag.String("state", filepath.Join(os.Getenv("HOME"), ".local/state/guestlink"), "state directory (invite token)")
)

// ---------------------------------------------------------------- hub

type chatMsg struct {
	From string `json:"from"`
	Who  string `json:"who"` // "host" | "guest" | "system"
	Text string `json:"text"`
	TS   int64  `json:"ts"`
}

type wsConn struct {
	c  *websocket.Conn
	mu sync.Mutex
}

func (w *wsConn) send(v any) {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.c.SetWriteDeadline(time.Now().Add(5 * time.Second))
	if err := w.c.WriteJSON(v); err != nil {
		w.c.Close()
	}
}

type Hub struct {
	mu        sync.Mutex
	token     string
	guest     *wsConn
	guestName string
	call      *Call
	hosts     map[*wsConn]bool
	chat      []chatMsg
	onAir     bool // tally shown on the guest page; set by the host, never automatic
	udp       *net.UDPConn
	mux       ice.UDPMux
}

var hub = &Hub{hosts: map[*wsConn]bool{}}

func randToken() string {
	b := make([]byte, 15)
	rand.Read(b)
	return base64.RawURLEncoding.EncodeToString(b)
}

func tokenPath() string { return filepath.Join(*stateDir, "token") }

func (h *Hub) loadToken() {
	b, err := os.ReadFile(tokenPath())
	if err == nil {
		h.token = strings.TrimSpace(string(b))
	}
}

func (h *Hub) saveToken() {
	os.MkdirAll(*stateDir, 0o700)
	os.WriteFile(tokenPath(), []byte(h.token), 0o600)
}

func (h *Hub) link() string {
	if h.token == "" {
		return ""
	}
	return "https://" + *publicHost + "/j/" + h.token
}

// stateLocked builds the host-console status message. Caller holds h.mu.
func (h *Hub) stateLocked() map[string]any {
	st := map[string]any{
		"t":           "state",
		"link":        h.link(),
		"guestOnline": h.guest != nil,
		"guestName":   h.guestName,
		"onAir":       h.onAir,
		"call":        nil,
	}
	if h.call != nil {
		st["call"] = h.call.status()
	}
	return st
}

func (h *Hub) broadcastHosts(v any) {
	h.mu.Lock()
	hs := make([]*wsConn, 0, len(h.hosts))
	for c := range h.hosts {
		hs = append(hs, c)
	}
	h.mu.Unlock()
	for _, c := range hs {
		c.send(v)
	}
}

func (h *Hub) pushState() {
	h.mu.Lock()
	st := h.stateLocked()
	h.mu.Unlock()
	h.broadcastHosts(st)
}

func (h *Hub) addChat(m chatMsg) {
	h.mu.Lock()
	h.chat = append(h.chat, m)
	if len(h.chat) > 300 {
		h.chat = h.chat[len(h.chat)-300:]
	}
	g := h.guest
	h.mu.Unlock()
	out := map[string]any{"t": "chat", "msg": m}
	h.broadcastHosts(out)
	if g != nil {
		g.send(out)
	}
}

func (h *Hub) system(text string) {
	h.addChat(chatMsg{From: "", Who: "system", Text: text, TS: time.Now().UnixMilli()})
}

// endCall tears down the current call, if any.
func (h *Hub) endCall() {
	h.mu.Lock()
	c := h.call
	h.call = nil
	h.mu.Unlock()
	if c != nil {
		c.Close()
	}
	h.pushState()
}

// kickGuest drops the guest's websocket and call.
func (h *Hub) kickGuest(reason string) {
	h.mu.Lock()
	g := h.guest
	h.guest = nil
	h.mu.Unlock()
	if g != nil {
		g.send(map[string]any{"t": "ended", "reason": reason})
		g.c.Close()
	}
	h.endCall()
}

// ---------------------------------------------------------------- helpers

func cleanText(s string, max int) string {
	s = strings.TrimSpace(strings.ToValidUTF8(s, ""))
	s = strings.Map(func(r rune) rune {
		if r == '\n' || r == '\t' {
			return r
		}
		if r < 0x20 || r == 0x7f {
			return -1
		}
		return r
	}, s)
	if utf8.RuneCountInString(s) > max {
		s = string([]rune(s)[:max])
	}
	return s
}

func isPrivate(addr string) bool {
	host, _, err := net.SplitHostPort(addr)
	if err != nil {
		return false
	}
	ip := net.ParseIP(host)
	if ip == nil {
		return false
	}
	_, cg, _ := net.ParseCIDR("100.64.0.0/10") // tailscale / CGNAT VPN
	return ip.IsLoopback() || ip.IsPrivate() || cg.Contains(ip)
}

func securityHeaders(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		hd := w.Header()
		hd.Set("Content-Security-Policy", "default-src 'self'; connect-src 'self' wss: ws:; img-src 'self' data:; style-src 'self'; script-src 'self'; frame-ancestors 'none'; base-uri 'none'")
		hd.Set("Permissions-Policy", "microphone=(self), camera=(), geolocation=()")
		hd.Set("Referrer-Policy", "no-referrer")
		hd.Set("X-Content-Type-Options", "nosniff")
		hd.Set("Cache-Control", "no-store")
		next.ServeHTTP(w, r)
	})
}

func serveFile(name string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		b, err := webFS.ReadFile("web/" + name)
		if err != nil {
			http.NotFound(w, r)
			return
		}
		w.Header().Set("Content-Type", "text/html; charset=utf-8")
		w.Write(b)
	}
}

var upgrader = websocket.Upgrader{ReadBufferSize: 8192, WriteBufferSize: 8192}

// ---------------------------------------------------------------- guest side

type rateLimiter struct {
	times []time.Time
	n     int
	per   time.Duration
}

func (r *rateLimiter) allow() bool {
	now := time.Now()
	keep := r.times[:0]
	for _, t := range r.times {
		if now.Sub(t) < r.per {
			keep = append(keep, t)
		}
	}
	r.times = keep
	if len(r.times) >= r.n {
		return false
	}
	r.times = append(r.times, now)
	return true
}

func guestWS(w http.ResponseWriter, r *http.Request) {
	tok := r.URL.Query().Get("t")
	hub.mu.Lock()
	valid := hub.token != "" && tok == hub.token
	hub.mu.Unlock()
	if !valid {
		http.Error(w, "invite not active", http.StatusForbidden)
		return
	}
	c, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		return
	}
	c.SetReadLimit(64 << 10)
	me := &wsConn{c: c}
	// Keepalive: ping every 25 s. Without it the proxy closed the idle socket every 90 s
	// and the guest saw "reconnecting..." all show. Signalling only - audio is unaffected.
	c.SetReadDeadline(time.Now().Add(70 * time.Second))
	c.SetPongHandler(func(string) error { c.SetReadDeadline(time.Now().Add(70 * time.Second)); return nil })
	pingDone := make(chan struct{})
	defer close(pingDone)
	go func() {
		t := time.NewTicker(25 * time.Second)
		defer t.Stop()
		for {
			select {
			case <-pingDone:
				return
			case <-t.C:
				if err := c.WriteControl(websocket.PingMessage, nil, time.Now().Add(5*time.Second)); err != nil {
					return
				}
			}
		}
	}()

	hub.mu.Lock()
	old := hub.guest
	hub.guest = me
	hist := append([]chatMsg(nil), hub.chat...)
	onAir := hub.onAir
	hub.mu.Unlock()
	if old != nil {
		old.send(map[string]any{"t": "ended", "reason": "This link was opened in another window."})
		old.c.Close()
	}
	log.Printf("guest connected from %s", r.Header.Get("X-Real-IP"))
	me.send(map[string]any{"t": "welcome", "hostLabel": *hostLabel, "history": hist, "onAir": onAir})
	hub.pushState()

	chatRL := &rateLimiter{n: 8, per: 10 * time.Second}
	sigRL := &rateLimiter{n: 200, per: 10 * time.Second}

	defer func() {
		hub.mu.Lock()
		still := hub.guest == me
		if still {
			hub.guest = nil
		}
		hub.mu.Unlock()
		c.Close()
		if still {
			log.Printf("guest websocket closed")
			hub.pushState()
		}
	}()

	for {
		var m struct {
			T         string                   `json:"t"`
			Text      string                   `json:"text"`
			Name      string                   `json:"name"`
			SDP       string                   `json:"sdp"`
			Candidate *webrtc.ICECandidateInit `json:"candidate"`
		}
		if err := c.ReadJSON(&m); err != nil {
			return
		}
		c.SetReadDeadline(time.Now().Add(70 * time.Second))
		hub.mu.Lock()
		current := hub.guest == me && hub.token == tok
		hub.mu.Unlock()
		if !current {
			return
		}
		if !sigRL.allow() {
			continue
		}
		switch m.T {
		case "name":
			n := cleanText(strings.ReplaceAll(m.Name, "\n", " "), 40)
			hub.mu.Lock()
			hub.guestName = n
			hub.mu.Unlock()
			hub.pushState()
		case "chat":
			t := cleanText(m.Text, 1000)
			if t == "" {
				continue
			}
			if !chatRL.allow() {
				me.send(map[string]any{"t": "notice", "text": "Slow down a little - message not sent."})
				continue
			}
			hub.mu.Lock()
			name := hub.guestName
			hub.mu.Unlock()
			if name == "" {
				name = "Guest"
			}
			hub.addChat(chatMsg{From: name, Who: "guest", Text: t, TS: time.Now().UnixMilli()})
		case "offer":
			hub.endCall() // one call at a time; a new offer replaces the old one
			call, answer, err := newCall(m.SDP)
			if err != nil {
				log.Printf("offer failed: %v", err)
				me.send(map[string]any{"t": "error", "text": "Could not start audio: " + err.Error()})
				continue
			}
			hub.mu.Lock()
			hub.call = call
			hub.mu.Unlock()
			me.send(map[string]any{"t": "answer", "sdp": answer})
			hub.pushState()
		case "candidate":
			hub.mu.Lock()
			call := hub.call
			hub.mu.Unlock()
			if call != nil && m.Candidate != nil {
				call.pc.AddICECandidate(*m.Candidate)
			}
		case "hangup":
			hub.endCall()
		}
	}
}

// ---------------------------------------------------------------- host side

func hostWS(w http.ResponseWriter, r *http.Request) {
	if !isPrivate(r.RemoteAddr) {
		http.Error(w, "LAN only", http.StatusForbidden)
		return
	}
	c, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		return
	}
	me := &wsConn{c: c}
	hub.mu.Lock()
	hub.hosts[me] = true
	hist := append([]chatMsg(nil), hub.chat...)
	st := hub.stateLocked()
	hub.mu.Unlock()
	me.send(map[string]any{"t": "history", "history": hist, "hostLabel": *hostLabel})
	me.send(st)
	defer func() {
		hub.mu.Lock()
		delete(hub.hosts, me)
		hub.mu.Unlock()
		c.Close()
	}()
	for {
		var m struct {
			T    string `json:"t"`
			Text string `json:"text"`
			On   bool   `json:"on"`
		}
		if err := c.ReadJSON(&m); err != nil {
			return
		}
		switch m.T {
		case "chat":
			if t := cleanText(m.Text, 2000); t != "" {
				hub.addChat(chatMsg{From: *hostLabel, Who: "host", Text: t, TS: time.Now().UnixMilli()})
			}
		case "newlink":
			hub.kickGuest("The host started a new session.")
			hub.mu.Lock()
			hub.token = randToken()
			hub.chat = nil
			hub.guestName = ""
			hub.onAir = false
			hub.saveToken()
			hub.mu.Unlock()
			log.Printf("new invite link created")
			hub.broadcastHosts(map[string]any{"t": "history", "history": []chatMsg{}, "hostLabel": *hostLabel})
			hub.pushState()
		case "revoke":
			hub.kickGuest("The host closed this session.")
			hub.mu.Lock()
			hub.token = ""
			hub.onAir = false
			hub.saveToken()
			hub.mu.Unlock()
			log.Printf("invite link revoked")
			hub.pushState()
		case "kick":
			hub.kickGuest("The host ended the call.")
		case "endcall":
			hub.mu.Lock()
			g := hub.guest
			hub.mu.Unlock()
			hub.endCall()
			if g != nil {
				g.send(map[string]any{"t": "callended"})
			}
		case "onair":
			hub.mu.Lock()
			hub.onAir = m.On
			g := hub.guest
			hub.mu.Unlock()
			log.Printf("tally: on air %v", m.On)
			if g != nil {
				g.send(map[string]any{"t": "onair", "on": m.On})
			}
			hub.pushState()
		case "clearchat":
			hub.mu.Lock()
			hub.chat = nil
			g := hub.guest
			hub.mu.Unlock()
			clr := map[string]any{"t": "history", "history": []chatMsg{}, "hostLabel": *hostLabel}
			hub.broadcastHosts(clr)
			if g != nil {
				g.send(clr)
			}
		}
	}
}

// ---------------------------------------------------------------- main

func main() {
	flag.Parse()
	log.SetFlags(log.LstdFlags)
	hub.loadToken()

	udp, err := net.ListenUDP("udp4", &net.UDPAddr{Port: *udpPort})
	if err != nil {
		log.Fatalf("udp listen: %v", err)
	}
	hub.udp = udp
	hub.mux = webrtc.NewICEUDPMux(nil, udp)

	static, _ := fs.Sub(webFS, "web/static")
	staticH := http.StripPrefix("/static/", http.FileServer(http.FS(static)))

	pub := http.NewServeMux()
	pub.Handle("/static/", staticH)
	pub.HandleFunc("/j/", serveFile("guest.html"))
	pub.HandleFunc("/ws", guestWS)
	pub.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok\n")) })
	pub.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/" {
			http.NotFound(w, r)
			return
		}
		serveFile("index.html")(w, r)
	})

	host := http.NewServeMux()
	host.Handle("/static/", staticH)
	host.HandleFunc("/ws", hostWS)
	host.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		if !isPrivate(r.RemoteAddr) {
			http.Error(w, "LAN only", http.StatusForbidden)
			return
		}
		if r.URL.Path != "/" {
			http.NotFound(w, r)
			return
		}
		serveFile("host.html")(w, r)
	})

	// Periodic status for the host console (levels, jitter buffer).
	go func() {
		for range time.Tick(150 * time.Millisecond) {
			hub.mu.Lock()
			active := hub.call != nil && len(hub.hosts) > 0
			hub.mu.Unlock()
			if active {
				hub.pushState()
			}
		}
	}()

	go func() {
		log.Printf("host console on http://%s", *hostAddr)
		log.Fatal(http.ListenAndServe(*hostAddr, securityHeaders(host)))
	}()
	log.Printf("guest listener on %s, media on udp/%d", *publicAddr, *udpPort)
	log.Fatal(http.ListenAndServe(*publicAddr, securityHeaders(pub)))
}

// jsonString is used by the call code for compact logging.
func jsonString(v any) string { b, _ := json.Marshal(v); return string(b) }