guestlink/call.go

package main

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"log"
	"math"
	"net"
	"os"
	"os/exec"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pion/interceptor"
	"github.com/pion/rtp"
	"github.com/pion/stun/v4"
	"github.com/pion/webrtc/v4"
	"github.com/pion/webrtc/v4/pkg/media"
	"golang.org/x/sys/unix"
	"gopkg.in/hraban/opus.v2"
)

const (
	rate       = 48000
	channels   = 2
	frameSamps = 960 // 20 ms per channel
	opusFmtp   = "minptime=10;useinbandfec=1;stereo=1;sprop-stereo=1;maxaveragebitrate=128000"
)

type Call struct {
	pc     *webrtc.PeerConnection
	track  *webrtc.TrackLocalStaticSample
	ctx    context.Context
	cancel context.CancelFunc
	once   sync.Once

	started  time.Time
	pcState  atomic.Value // string
	iceState atomic.Value // string
	route    atomic.Value // string: selected candidate pair

	outPeak atomic.Uint32 // mix-minus to guest, peak*1e4 since last status
	inPeak  atomic.Uint32 // guest to show
	jb      *jitterBuffer
	jbMu    sync.Mutex
	sendOn  atomic.Bool
	playOn  atomic.Bool
}

var (
	pubIPMu    sync.Mutex
	pubIPCache string
	pubIPAt    time.Time
)

// publicIP returns the rig's public IPv4: -public-ip if set, else a STUN query (works no
// matter how DNS / Cloudflare proxying is set up), else the A record of -public-host.
func publicIP() (string, error) {
	if *publicIPFlag != "" {
		return *publicIPFlag, nil
	}
	pubIPMu.Lock()
	defer pubIPMu.Unlock()
	if pubIPCache != "" && time.Since(pubIPAt) < 10*time.Minute {
		return pubIPCache, nil
	}
	for _, srv := range []string{"stun.l.google.com:19302", "stun.cloudflare.com:3478"} {
		if ip, err := stunIP(srv); err == nil {
			pubIPCache, pubIPAt = ip, time.Now()
			return ip, nil
		} else {
			log.Printf("stun %s: %v", srv, err)
		}
	}
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	ips, err := net.DefaultResolver.LookupIP(ctx, "ip4", *publicHost)
	if err != nil || len(ips) == 0 {
		return "", fmt.Errorf("no public IP (stun failed; resolve %s: %v)", *publicHost, err)
	}
	return ips[0].String(), nil
}

func stunIP(server string) (string, error) {
	conn, err := net.DialTimeout("udp4", server, 2*time.Second)
	if err != nil {
		return "", err
	}
	defer conn.Close()
	conn.SetDeadline(time.Now().Add(2 * time.Second))
	msg := stun.MustBuild(stun.TransactionID, stun.BindingRequest)
	if _, err := conn.Write(msg.Raw); err != nil {
		return "", err
	}
	buf := make([]byte, 1500)
	n, err := conn.Read(buf)
	if err != nil {
		return "", err
	}
	res := &stun.Message{Raw: buf[:n]}
	if err := res.Decode(); err != nil {
		return "", err
	}
	var xor stun.XORMappedAddress
	if err := xor.GetFrom(res); err != nil {
		return "", err
	}
	if xor.IP.To4() == nil {
		return "", errors.New("not ipv4")
	}
	return xor.IP.String(), nil
}

func newAPI() (*webrtc.API, error) {
	m := &webrtc.MediaEngine{}
	if err := m.RegisterCodec(webrtc.RTPCodecParameters{
		RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: rate, Channels: 2, SDPFmtpLine: opusFmtp},
		PayloadType:        111,
	}, webrtc.RTPCodecTypeAudio); err != nil {
		return nil, err
	}
	ir := &interceptor.Registry{}
	if err := webrtc.RegisterDefaultInterceptors(m, ir); err != nil {
		return nil, err
	}

	_, lan, err := net.ParseCIDR(*lanCIDR)
	if err != nil {
		return nil, err
	}
	se := webrtc.SettingEngine{}
	se.SetICEUDPMux(hub.mux)
	se.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeUDP4})
	se.SetIPFilter(func(ip net.IP) bool { return lan.Contains(ip) })
	se.SetICETimeouts(8*time.Second, 25*time.Second, 2*time.Second)
	if pip, err := publicIP(); err == nil {
		// Keep the LAN host candidate and ADD one with the public IP on the same port.
		// Needs UDP <udp> forwarded to this machine on the router.
		if err := se.SetICEAddressRewriteRules(webrtc.ICEAddressRewriteRule{
			External: []string{pip}, AsCandidateType: webrtc.ICECandidateTypeHost, Mode: webrtc.ICEAddressRewriteAppend,
		}); err != nil {
			return nil, err
		}
	} else {
		log.Printf("warning: %v - offering LAN candidate only", err)
	}
	return webrtc.NewAPI(webrtc.WithMediaEngine(m), webrtc.WithInterceptorRegistry(ir), webrtc.WithSettingEngine(se)), nil
}

func newCall(offerSDP string) (*Call, string, error) {
	api, err := newAPI()
	if err != nil {
		return nil, "", err
	}
	pc, err := api.NewPeerConnection(webrtc.Configuration{})
	if err != nil {
		return nil, "", err
	}
	ctx, cancel := context.WithCancel(context.Background())
	c := &Call{pc: pc, ctx: ctx, cancel: cancel, started: time.Now()}
	c.pcState.Store("new")
	c.iceState.Store("new")
	c.route.Store("")

	fail := func(err error) (*Call, string, error) { c.Close(); return nil, "", err }

	if err := pc.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: offerSDP}); err != nil {
		return fail(err)
	}
	track, err := webrtc.NewTrackLocalStaticSample(
		webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: rate, Channels: 2, SDPFmtpLine: opusFmtp},
		"audio", "mmo-mixminus")
	if err != nil {
		return fail(err)
	}
	c.track = track
	sender, err := pc.AddTrack(track)
	if err != nil {
		return fail(err)
	}
	go func() { // drain RTCP so interceptors work
		buf := make([]byte, 1500)
		for {
			if _, _, err := sender.Read(buf); err != nil {
				return
			}
		}
	}()

	pc.OnICEConnectionStateChange(func(s webrtc.ICEConnectionState) {
		c.iceState.Store(s.String())
		log.Printf("ice: %s", s)
		if s == webrtc.ICEConnectionStateConnected {
			c.updateRoute()
		}
	})
	pc.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
		c.pcState.Store(s.String())
		log.Printf("peer connection: %s", s)
		switch s {
		case webrtc.PeerConnectionStateConnected:
			c.updateRoute()
			c.startSend()
		case webrtc.PeerConnectionStateFailed, webrtc.PeerConnectionStateClosed:
			hub.mu.Lock()
			mine := hub.call == c
			hub.mu.Unlock()
			if mine {
				go hub.endCall()
			}
		}
		go hub.pushState()
	})
	pc.OnTrack(func(t *webrtc.TrackRemote, _ *webrtc.RTPReceiver) {
		if t.Kind() != webrtc.RTPCodecTypeAudio {
			return
		}
		log.Printf("guest track: %s %dHz ch=%d", t.Codec().MimeType, t.Codec().ClockRate, t.Codec().Channels)
		c.startPlay(t)
	})

	answer, err := pc.CreateAnswer(nil)
	if err != nil {
		return fail(err)
	}
	gather := webrtc.GatheringCompletePromise(pc)
	if err := pc.SetLocalDescription(answer); err != nil {
		return fail(err)
	}
	select {
	case <-gather:
	case <-time.After(3 * time.Second):
	}
	return c, pc.LocalDescription().SDP, nil
}

func (c *Call) updateRoute() {
	for _, tr := range c.pc.GetTransceivers() {
		if s := tr.Sender(); s != nil && s.Transport() != nil {
			ice := s.Transport().ICETransport()
			if ice == nil {
				continue
			}
			if pair, err := ice.GetSelectedCandidatePair(); err == nil && pair != nil {
				r := fmt.Sprintf("%s:%d <-> %s:%d (%s)", pair.Local.Address, pair.Local.Port, pair.Remote.Address, pair.Remote.Port, pair.Remote.Typ)
				if old, _ := c.route.Swap(r).(string); old != r {
					log.Printf("route: %s", r)
				}
				return
			}
		}
	}
}

func (c *Call) Close() {
	c.once.Do(func() {
		c.cancel()
		c.pc.Close()
		log.Printf("call closed after %s", time.Since(c.started).Round(time.Second))
		c.jbMu.Lock()
		jb := c.jb
		c.jbMu.Unlock()
		if jb != nil {
			k, _, t := jb.snapshot()
			log.Printf("call totals: lost %d (fec %d) concealed %d late %d underruns %d drift-drops %d inserted %d, final target %d ms",
				k.Lost, k.FEC, k.PLC, k.Late, k.Underruns, k.Dropped, k.Inserted, t)
		}
	})
}

// logRx writes one line a minute about the guest's audio, so a bad night can be
// diagnosed afterwards (v1 only showed these numbers live on the host console).
func (c *Call) logRx(jb *jitterBuffer) {
	t := time.NewTicker(time.Minute)
	defer t.Stop()
	var prev jbCounters
	for {
		select {
		case <-c.ctx.Done():
			return
		case <-t.C:
		}
		k, d, tg := jb.snapshot()
		log.Printf("rx 1m: buffer %d/%d ms, lost %d (fec %d), concealed %d, late %d, underruns %d, drift-drops %d, inserted %d",
			d, tg, k.Lost-prev.Lost, k.FEC-prev.FEC, k.PLC-prev.PLC, k.Late-prev.Late, k.Underruns-prev.Underruns, k.Dropped-prev.Dropped, k.Inserted-prev.Inserted)
		prev = k
	}
}

func (c *Call) status() map[string]any {
	st := map[string]any{
		"pc":      c.pcState.Load(),
		"ice":     c.iceState.Load(),
		"route":   c.route.Load(),
		"secs":    int(time.Since(c.started).Seconds()),
		"sending": c.sendOn.Load(),
		"playing": c.playOn.Load(),
		"outPeak": float64(c.outPeak.Swap(0)) / 1e4,
		"inPeak":  float64(c.inPeak.Swap(0)) / 1e4,
	}
	c.jbMu.Lock()
	if c.jb != nil {
		st["jb"] = c.jb.stats()
	}
	c.jbMu.Unlock()
	return st
}

func storeMax(a *atomic.Uint32, v uint32) {
	for {
		o := a.Load()
		if v <= o || a.CompareAndSwap(o, v) {
			return
		}
	}
}

func peakOf(pcm []int16) uint32 {
	var p int32
	for _, s := range pcm {
		v := int32(s)
		if v < 0 {
			v = -v
		}
		if v > p {
			p = v
		}
	}
	return uint32(float64(p) / 32768 * 1e4)
}

// ---------------------------------------------------------------- mix-minus -> guest

func (c *Call) startSend() {
	if c.sendOn.Swap(true) {
		return
	}
	go func() {
		defer c.sendOn.Store(false)
		for c.ctx.Err() == nil {
			if err := c.sendLoop(); err != nil && c.ctx.Err() == nil {
				log.Printf("send: %v - retrying in 1s", err)
				select {
				case <-c.ctx.Done():
				case <-time.After(time.Second):
				}
			}
		}
	}()
}

func (c *Call) sendLoop() error {
	cmd := exec.CommandContext(c.ctx, "parec",
		"-d", *sourceName, "--raw", "--format=s16le", fmt.Sprintf("--rate=%d", rate), fmt.Sprintf("--channels=%d", channels),
		"--latency-msec=20", "--property=node.dont-fallback=true", "--property=node.dont-reconnect=true", "--property=node.dont-move=true", "--client-name=guestlink", "--stream-name=mix-minus to guest")
	out, err := cmd.StdoutPipe()
	if err != nil {
		return err
	}
	cmd.Stderr = os.Stderr
	if err := cmd.Start(); err != nil {
		return err
	}
	defer cmd.Wait()
	defer cmd.Process.Kill()

	enc, err := opus.NewEncoder(rate, channels, opus.AppAudio)
	if err != nil {
		return err
	}
	enc.SetBitrate(128000)
	enc.SetInBandFEC(true)
	enc.SetPacketLossPerc(5)
	enc.SetComplexity(9)

	raw := make([]byte, frameSamps*channels*2)
	pcm := make([]int16, frameSamps*channels)
	pkt := make([]byte, 4000)
	for {
		if _, err := io.ReadFull(out, raw); err != nil {
			if c.ctx.Err() != nil {
				return nil
			}
			return fmt.Errorf("parec read: %w", err)
		}
		for i := range pcm {
			pcm[i] = int16(binary.LittleEndian.Uint16(raw[i*2:]))
		}
		storeMax(&c.outPeak, peakOf(pcm))
		n, err := enc.Encode(pcm, pkt)
		if err != nil {
			return fmt.Errorf("opus encode: %w", err)
		}
		data := make([]byte, n)
		copy(data, pkt[:n])
		if err := c.track.WriteSample(media.Sample{Data: data, Duration: 20 * time.Millisecond}); err != nil && !errors.Is(err, io.ErrClosedPipe) {
			return fmt.Errorf("write sample: %w", err)
		}
	}
}

// ---------------------------------------------------------------- guest -> show

func (c *Call) startPlay(t *webrtc.TrackRemote) {
	dec, err := opus.NewDecoder(rate, channels)
	if err != nil {
		log.Printf("decoder: %v", err)
		return
	}
	jb := newJitterBuffer(dec)
	c.jbMu.Lock()
	c.jb = jb
	c.jbMu.Unlock()

	go c.logRx(jb)
	go func() { // network -> jitter buffer
		for {
			p, _, err := t.ReadRTP()
			if err != nil {
				return
			}
			jb.push(p)
		}
	}()
	go func() { // jitter buffer -> pacat, paced by the MOTU clock via a tiny pipe
		c.playOn.Store(true)
		defer c.playOn.Store(false)
		for c.ctx.Err() == nil {
			if err := c.playLoop(jb); err != nil && c.ctx.Err() == nil {
				log.Printf("play: %v - retrying in 1s", err)
				select {
				case <-c.ctx.Done():
				case <-time.After(time.Second):
				}
			}
		}
	}()
}

func (c *Call) playLoop(jb *jitterBuffer) error {
	if *testPlayoutFile != "" {
		return c.playToFile(jb)
	}
	r, w, err := os.Pipe()
	if err != nil {
		return err
	}
	// Small pipe => writes block on the sink's consumption => the MOTU clock paces us,
	// and the jitter buffer depth tells us about clock drift.
	unix.FcntlInt(w.Fd(), unix.F_SETPIPE_SZ, 4096)
	cmd := exec.CommandContext(c.ctx, "pacat", "--playback",
		"-d", *sinkName, "--raw", "--format=s16le", fmt.Sprintf("--rate=%d", rate), fmt.Sprintf("--channels=%d", channels),
		"--latency-msec=30", "--property=node.dont-fallback=true", "--property=node.dont-reconnect=true", "--property=node.dont-move=true", "--client-name=guestlink", "--stream-name=guest voice")
	cmd.Stdin = r
	cmd.Stderr = os.Stderr
	if err := cmd.Start(); err != nil {
		r.Close()
		w.Close()
		return err
	}
	r.Close()
	defer cmd.Wait()
	defer cmd.Process.Kill()
	defer w.Close()

	buf := make([]byte, 0, 5760*channels*2)
	for c.ctx.Err() == nil {
		pcm := jb.next()
		storeMax(&c.inPeak, peakOf(pcm))
		buf = buf[:0]
		for _, s := range pcm {
			buf = binary.LittleEndian.AppendUint16(buf, uint16(s))
		}
		if _, err := w.Write(buf); err != nil {
			if c.ctx.Err() != nil {
				return nil
			}
			return fmt.Errorf("pacat write: %w", err)
		}
	}
	return nil
}

// playToFile is a test path: playout to a raw PCM file paced by a 20 ms ticker instead of
// the sound card, so the receive chain can be verified without touching PipeWire.
func (c *Call) playToFile(jb *jitterBuffer) error {
	f, err := os.Create(*testPlayoutFile)
	if err != nil {
		return err
	}
	defer f.Close()
	t := time.NewTicker(20 * time.Millisecond)
	defer t.Stop()
	buf := make([]byte, 0, 5760*channels*2)
	for {
		select {
		case <-c.ctx.Done():
			return nil
		case <-t.C:
		}
		pcm := jb.next()
		storeMax(&c.inPeak, peakOf(pcm))
		buf = buf[:0]
		for _, s := range pcm {
			buf = binary.LittleEndian.AppendUint16(buf, uint16(s))
		}
		f.Write(buf)
	}
}

// ---------------------------------------------------------------- jitter buffer

// jitterBuffer reorders Opus packets, conceals losses (FEC, then PLC), adapts its depth to
// network jitter, and corrects clock drift between the guest's sound card and the MOTU.
//
// It is pulled once per 20 ms by the MOTU clock. Rules (v2, 2026-11-04):
//   - Underrun (nothing buffered): HOLD the sequence position and refill to the target
//     before playing again. The delayed packets are then played, not discarded as late.
//     (v1 skipped ahead and kept playing from an empty buffer, so every later wobble
//     became another glitch - ~12/min with a guest clock 30 ppm slow.)
//   - Too deep (fast guest clock / after a stall): discard quiet frames.
//   - Too shallow (slow guest clock): insert a concealment frame during a quiet moment.
//   - Underruns and late packets raise the target; it relaxes after 2 calm minutes.
type jitterBuffer struct {
	mu        sync.Mutex
	dec       *opus.Decoder
	pkts      map[uint16][]byte
	nextSeq   uint16
	haveSeq   bool // nextSeq is meaningful
	playing   bool // false = filling up to target
	target    int  // frames
	gap       int  // frames produced while refilling
	lastBad   time.Time
	lastBump  time.Time
	lateRun   int
	insertNow bool
	floor     float64 // running estimate of the guest's background-noise RMS
	pcm       []int16
	c         jbCounters
}

type jbCounters struct {
	Lost, FEC, PLC, Dropped, Inserted, Underruns, Late int
}

const (
	jbFloor  = 4     // 80 ms minimum target
	jbMax    = 15    // 300 ms maximum target
	quietRMS = 0.003 // always counts as quiet (-50 dBFS)
	quietCap = 0.02  // never counts as quiet (-34 dBFS)
)

// isQuiet: below -50 dBFS, or within ~6 dB of the guest's own noise floor (so drift
// correction still works for a guest in a noisy room), but never above -34 dBFS.
func (j *jitterBuffer) isQuiet(pcm []int16) bool {
	r := rms(pcm)
	if r < j.floor {
		j.floor = r
	} else {
		j.floor *= 1.0005 // creeps up ~2.5%/s so it follows a rising floor
	}
	lim := math.Min(quietCap, math.Max(quietRMS, 2*j.floor))
	return r < lim
}

func newJitterBuffer(dec *opus.Decoder) *jitterBuffer {
	return &jitterBuffer{dec: dec, pkts: map[uint16][]byte{}, target: jbFloor, pcm: make([]int16, 5760*channels), lastBad: time.Now(), floor: quietRMS}
}

func seqLess(a, b uint16) bool { return int16(a-b) < 0 }

func (j *jitterBuffer) reset() {
	j.pkts = map[uint16][]byte{}
	j.haveSeq, j.playing, j.lateRun, j.insertNow = false, false, 0, false
}

// bump raises the target after an underrun or a late packet (at most once a second, so a
// burst of late packets counts once).
func (j *jitterBuffer) bump() {
	j.lastBad = time.Now()
	if j.target < jbMax && time.Since(j.lastBump) > time.Second {
		j.target++
		j.lastBump = time.Now()
	}
}

func (j *jitterBuffer) push(p *rtp.Packet) {
	j.mu.Lock()
	defer j.mu.Unlock()
	if len(p.Payload) == 0 {
		return
	}
	if j.haveSeq {
		if seqLess(p.SequenceNumber, j.nextSeq) {
			j.c.Late++
			j.bump()
			if j.lateRun++; j.lateRun > 50 { // sender restarted its sequence numbers
				j.reset()
			}
			return
		}
		if d := uint16(p.SequenceNumber - j.nextSeq); d > 500 { // stream restarted
			j.reset()
		}
	}
	j.lateRun = 0
	if len(j.pkts) > 100 { // hard cap - should never happen
		j.reset()
	}
	j.pkts[p.SequenceNumber] = append([]byte(nil), p.Payload...)
}

func (j *jitterBuffer) oldest() uint16 {
	first := true
	var m uint16
	for s := range j.pkts {
		if first || seqLess(s, m) {
			m, first = s, false
		}
	}
	return m
}

func (j *jitterBuffer) silence() []int16 { return make([]int16, frameSamps*channels) }

func (j *jitterBuffer) decode(p []byte) []int16 {
	n, err := j.dec.Decode(p, j.pcm)
	if err != nil || n <= 0 {
		return j.conceal()
	}
	return append([]int16(nil), j.pcm[:n*channels]...)
}

func (j *jitterBuffer) plcFrame() []int16 {
	out := make([]int16, frameSamps*channels)
	if err := j.dec.DecodePLC(out); err != nil {
		return j.silence()
	}
	return out
}

func (j *jitterBuffer) conceal() []int16 { j.c.PLC++; return j.plcFrame() }

func rms(pcm []int16) float64 {
	if len(pcm) == 0 {
		return 0
	}
	var s float64
	for _, v := range pcm {
		f := float64(v) / 32768
		s += f * f
	}
	return math.Sqrt(s / float64(len(pcm)))
}

// fill is played while the buffer refills: a few concealment frames (Opus PLC fades out
// naturally), then silence.
func (j *jitterBuffer) fill() []int16 {
	j.gap++
	if j.haveSeq && j.gap <= 3 {
		return j.conceal()
	}
	return j.silence()
}

// next returns the next block of interleaved PCM to play (normally 20 ms).
func (j *jitterBuffer) next() []int16 {
	j.mu.Lock()
	defer j.mu.Unlock()

	// Relax the target after two calm minutes.
	if j.target > jbFloor && time.Since(j.lastBad) > 2*time.Minute {
		j.target--
		j.lastBad = time.Now()
	}

	if !j.playing {
		if len(j.pkts) < j.target {
			return j.fill()
		}
		if !j.haveSeq {
			j.nextSeq, j.haveSeq = j.oldest(), true
		}
		j.playing, j.gap = true, 0
	}

	if j.insertNow { // slow guest clock: stretch a quiet moment by one frame
		j.insertNow = false
		j.c.Inserted++
		return j.plcFrame()
	}

	for {
		if p, ok := j.pkts[j.nextSeq]; ok {
			delete(j.pkts, j.nextSeq)
			j.nextSeq++
			pcm := j.decode(p)
			depth := len(j.pkts)
			quiet := j.isQuiet(pcm)
			if (depth > j.target+2 && quiet) || depth > j.target+10 {
				j.c.Dropped++
				continue
			}
			if depth < j.target-1 && quiet {
				j.insertNow = true
			}
			return pcm
		}
		if len(j.pkts) == 0 {
			// Underrun: keep our place and refill to a larger target.
			j.c.Underruns++
			j.lastBump = time.Time{} // an underrun always counts
			j.bump()
			j.playing = false
			return j.fill()
		}
		// Gap with later packets present: this packet is lost (or too late to wait for).
		j.c.Lost++
		if fp, ok := j.pkts[j.nextSeq+1]; ok {
			out := make([]int16, frameSamps*channels)
			if err := j.dec.DecodeFEC(fp, out); err == nil {
				j.c.FEC++
				j.nextSeq++
				return out
			}
		}
		j.nextSeq++
		return j.conceal()
	}
}

func (j *jitterBuffer) snapshot() (c jbCounters, depthMs, targetMs int) {
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.c, len(j.pkts) * 20, j.target * 20
}

func (j *jitterBuffer) stats() map[string]any {
	c, d, t := j.snapshot()
	return map[string]any{
		"depthMs": d, "targetMs": t,
		"lost": c.Lost, "fec": c.FEC, "plc": c.PLC, "dropped": c.Dropped, "inserted": c.Inserted,
		"underruns": c.Underruns, "late": c.Late,
	}
}