telegabber/calls/audio/xmpp_to_tg.go

package audio

import (
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pion/webrtc/v4"
	log "github.com/sirupsen/logrus"
)

type XmppToTgOptions struct {
	Ntg          NtgClient
	ChatID       int64  // P2P: equals the Telegram peer's user_id
	PrimePackets uint16 // buffered before playout starts; default 2 (~40ms)
	Logger       *log.Entry
}

// xmpp->tg half: pion RemoteTrack RTP -> opus decode -> ntgcalls
// SendMicrophonePCM. One reader goroutine per HandleTrack, plus one
// playout ticker spawned by Start.
type XmppToTg struct {
	opts XmppToTgOptions
	log  *log.Entry

	dec     *Decoder
	playout *Playout

	stop    chan struct{}
	wg      sync.WaitGroup
	pcmBuf  []int16 // decoder output (worst-case 120ms stereo)
	silence []byte  // 10ms silence for underrun

	closed  atomic.Bool
	started atomic.Bool
	// false until Start runs. Audio between pion OnTrack (during
	// SetRemoteDescription) and bridge OnEstablished is drained from
	// pion's per-track queue but NOT enqueued - otherwise every packet
	// from the setup window would back up and the ticker would replay
	// it at real time, manifesting as call-long lag.
	livePush        atomic.Bool
	preLiveDiscards atomic.Int64

	// telemetry counters (atomic; sampled by the periodic stats log)
	cRtpIn        atomic.Int64 // RTP packets read from pion
	cRtpSkippedPT atomic.Int64 // dropped: payload type != negotiated opus PT
	cShipTotal    atomic.Int64 // 10ms frames handed to ntgcalls (== ntg feed rate)
	cShipSilence  atomic.Int64 // of those, silence frames (underrun/priming)
	cPlcFrames    atomic.Int64 // PLC 10ms frames synthesised over gaps
	cGapEvents    atomic.Int64 // ticks where a seq gap was skipped
	cDtxFrames    atomic.Int64 // empty (DTX) packets concealed as comfort noise
	cSendErrs     atomic.Int64 // SendMicrophonePCM errors

	capture *captureWriter
}

// counter snapshot for computing per-interval deltas in the stats log
type xmppToTgSnap struct {
	rtpIn, rtpSkippedPT, shipTotal, shipSilence, plcFrames, gapEvents, dtxFrames, sendErrs int64
}

func (h *XmppToTg) snap() xmppToTgSnap {
	return xmppToTgSnap{
		rtpIn:        h.cRtpIn.Load(),
		rtpSkippedPT: h.cRtpSkippedPT.Load(),
		shipTotal:    h.cShipTotal.Load(),
		shipSilence:  h.cShipSilence.Load(),
		plcFrames:    h.cPlcFrames.Load(),
		gapEvents:    h.cGapEvents.Load(),
		dtxFrames:    h.cDtxFrames.Load(),
		sendErrs:     h.cSendErrs.Load(),
	}
}

func NewXmppToTg(opts XmppToTgOptions) (*XmppToTg, error) {
	if opts.Ntg == nil {
		return nil, errors.New("audio: Ntg required")
	}
	if opts.ChatID == 0 {
		return nil, errors.New("audio: ChatID required")
	}
	logger := opts.Logger
	if logger == nil {
		logger = log.WithField("module", "audio")
	}
	logger = logger.WithFields(log.Fields{"dir": "xmpp->tg", "chat_id": opts.ChatID})

	dec, err := NewDecoder()
	if err != nil {
		return nil, err
	}

	h := &XmppToTg{
		opts:    opts,
		log:     logger,
		dec:     dec,
		playout: NewPlayout(opts.PrimePackets),
		pcmBuf:  make([]int16, MaxDecodeSamplesPerChannel*Channels),
		silence: make([]byte, NtgFrameBytes),
		stop:    make(chan struct{}),
	}
	if cap, err := newCaptureWriter(
		fmt.Sprintf("xmpp-to-tg-%d-%d.bin", opts.ChatID, time.Now().Unix()),
		magicRTP,
	); err != nil {
		logger.WithError(err).Warn("capture: xmpp->tg writer disabled")
	} else if cap != nil {
		logger.Info("capture: xmpp->tg RTP capture enabled")
		h.capture = cap
	}
	return h, nil
}

// flips on the ntgcalls Capture source, the playout ticker, and livePush
func (h *XmppToTg) Start() error {
	if !h.started.CompareAndSwap(false, true) {
		return nil
	}
	if err := h.opts.Ntg.SetExternalMicrophone(h.opts.ChatID, true, SampleRate, Channels); err != nil {
		return fmt.Errorf("SetExternalMicrophone(capture): %w", err)
	}
	h.livePush.Store(true)
	h.wg.Add(1)
	go h.tickerLoop()
	h.log.WithFields(log.Fields{
		"pre_live_discards": h.preLiveDiscards.Load(),
		"prime_packets":     h.playout.TargetDepth,
		"target_depth":      h.playout.TargetDepth,
		"max_depth":         h.playout.MaxDepth,
		"max_depth_ms":      h.playout.MaxDepth * OpusFrameMs,
	}).Info("xmpp->tg half started")
	return nil
}

// spawns a reader for each pc.OnTrack
func (h *XmppToTg) HandleTrack(t *webrtc.TrackRemote) {
	if t == nil || t.Kind() != webrtc.RTPCodecTypeAudio {
		return
	}
	h.log.WithFields(log.Fields{
		"codec": t.Codec().MimeType,
		"ssrc":  t.SSRC(),
	}).Info("xmpp->tg: remote audio track received from pion")
	h.wg.Add(1)
	go h.reader(t)
}

// stops both pumps; ntgcalls Capture+Playback are released by Call.Close
func (h *XmppToTg) Close() error {
	if !h.closed.CompareAndSwap(false, true) {
		return nil
	}
	close(h.stop)
	h.wg.Wait()
	h.capture.close()
	h.log.Info("xmpp->tg half closed")
	return nil
}

func (h *XmppToTg) reader(t *webrtc.TrackRemote) {
	defer h.wg.Done()
	// negotiated PT for this track; peer is free to pick anything in 96-127,
	// hardcoding 111 would silently drop opus from peers that picked another
	expectedPT := uint8(t.PayloadType())
	for {
		select {
		case <-h.stop:
			return
		default:
		}
		pkt, _, err := t.ReadRTP()
		if err != nil {
			return
		}
		h.cRtpIn.Add(1)
		if pkt.PayloadType != expectedPT {
			// CN/DTMF/RED can share an SSRC; libopus rejects them
			h.cRtpSkippedPT.Add(1)
			continue
		}
		if !h.livePush.Load() {
			// pre-Start: drain pion's per-track queue but don't enqueue;
			// see livePush docstring for the lag rationale
			h.preLiveDiscards.Add(1)
			continue
		}
		h.capture.writeRTP(pkt.SSRC, pkt.SequenceNumber, pkt.Timestamp, pkt.PayloadType, pkt.Payload)
		h.playout.Push(pkt)
	}
}

// cross-tick playout state; separated so tests can drive processTick
// step by step without a time.Ticker
type tickState struct {
	lastDecodedSamples int      // samples/ch from last Decode; DecodePLC output matches
	pending10ms        [][]byte // 10ms PCM chunks awaiting shipment, one per tick
}

func newTickState() *tickState {
	return &tickState{lastDecodedSamples: SamplesPerOpusFrame}
}

// how often the aggregated stats line is emitted
const statsInterval = 5 * time.Second

// a tick gap over this means the goroutine was descheduled (GC, blocking CGO,
// scheduler pressure) and missed ticks. 25ms = missed a 10ms tick with margin.
const tickStallThreshold = 25 * time.Millisecond

func (h *XmppToTg) tickerLoop() {
	defer h.wg.Done()
	ticker := time.NewTicker(NtgFrameMs * time.Millisecond)
	defer ticker.Stop()

	st := newTickState()
	loopStart := time.Now()
	last := loopStart
	lastLog := loopStart
	prev := h.snap()
	var ticks, stalls, maxGapMs int64
	for {
		select {
		case <-h.stop:
			return
		case <-ticker.C:
		}
		now := time.Now()
		gap := now.Sub(last)
		last = now
		ticks++
		if gap > tickStallThreshold {
			stalls++
			if ms := gap.Milliseconds(); ms > maxGapMs {
				maxGapMs = ms
			}
		}

		h.processTick(st)

		if now.Sub(lastLog) >= statsInterval {
			h.logStats(now.Sub(lastLog), now.Sub(loopStart), ticks, stalls, maxGapMs, &prev)
			ticks, stalls, maxGapMs = 0, 0, 0
			lastLog = now
		}
	}
}

// logStats emits one aggregated telemetry line per statsInterval, with
// per-second rates over the interval and the current playout depth.
func (h *XmppToTg) logStats(elapsed, sinceStart time.Duration, ticks, stalls, maxGapMs int64, prev *xmppToTgSnap) {
	cur := h.snap()
	secs := elapsed.Seconds()
	if secs <= 0 {
		secs = 1
	}
	rate := func(d int64) float64 { return float64(d) / secs }
	ps := h.playout.Stats()
	// feed_drift_ms = frames shipped to ntgcalls minus what real time allows.
	// Ticker-paced, so ~0 normally; a growing value means the ticker is starved
	// and we're under-feeding ntgcalls' capture.
	expected := sinceStart.Milliseconds() / NtgFrameMs
	feedDriftMs := (cur.shipTotal - expected) * NtgFrameMs
	h.log.WithFields(log.Fields{
		"interval_ms":     elapsed.Milliseconds(),
		"call_secs":       int64(sinceStart.Seconds()),
		"ticks":           ticks,
		"tick_stalls":     stalls,
		"max_tick_gap_ms": maxGapMs,
		"rtp_in_per_s":    rate(cur.rtpIn - prev.rtpIn),
		"rtp_skipped_pt":  cur.rtpSkippedPT - prev.rtpSkippedPT,
		"ship_per_s":      rate(cur.shipTotal - prev.shipTotal),
		"silence_per_s":   rate(cur.shipSilence - prev.shipSilence),
		"feed_drift_ms":   feedDriftMs,
		"plc_frames":      cur.plcFrames - prev.plcFrames,
		"gap_events":      cur.gapEvents - prev.gapEvents,
		"dtx_frames":      cur.dtxFrames - prev.dtxFrames,
		"send_errs":       cur.sendErrs - prev.sendErrs,
		"depth":           ps.Depth,
		"depth_ms":        ps.Depth * OpusFrameMs,
		"peak_depth_ms":   ps.MaxDepthSeen * OpusFrameMs,
		"jb_pushed":       ps.Pushed,
		"jb_popped":       ps.Popped,
		"jb_late_drop":    ps.LateDrop,
		"jb_trim_drop":    ps.TrimDrop,
		"jb_trim_events":  ps.TrimEvents,
		"jb_shed_drop":    ps.ShedDrop,
	}).Debug("xmpp->tg stats")
	*prev = cur
}

// one iteration of the playout loop
func (h *XmppToTg) processTick(st *tickState) {
	if len(st.pending10ms) == 0 {
		// walk any accumulated latency back toward the target depth, one
		// packet per pop, so recovery after a stall is gradual rather than a
		// single large skip
		h.playout.ShedOne(h.playout.TargetDepth)
		pkt, plcCount, err := h.playout.PopOrSkip()
		if err != nil {
			// empty or still priming - feed silence
			h.cShipSilence.Add(1)
			h.sendToNtg(h.silence)
			return
		}
		if plcCount > 0 {
			h.cGapEvents.Add(1)
		}
		for i := 0; i < plcCount; i++ {
			plcOut := h.pcmBuf[:st.lastDecodedSamples*Channels]
			if err := h.dec.DecodePLC(plcOut); err != nil {
				h.log.WithError(err).Warn("DecodePLC failed")
				break
			}
			before := len(st.pending10ms)
			st.pending10ms = appendChunks10ms(st.pending10ms, plcOut)
			h.cPlcFrames.Add(int64(len(st.pending10ms) - before))
		}
		if len(pkt.Payload) == 0 {
			// empty payload = opus DTX/comfort-noise frame during silence, not
			// a decode error. Conceal for the last frame's duration (libopus
			// synthesises comfort noise) rather than emitting hard silence.
			h.cDtxFrames.Add(1)
			plcOut := h.pcmBuf[:st.lastDecodedSamples*Channels]
			if err := h.dec.DecodePLC(plcOut); err != nil {
				h.cShipSilence.Add(1)
				h.sendToNtg(h.silence)
				return
			}
			st.pending10ms = appendChunks10ms(st.pending10ms, plcOut)
		} else {
			n, err := h.dec.Decode(pkt.Payload, h.pcmBuf)
			if err != nil {
				h.log.WithError(err).Warn("opus decode failed")
				h.cShipSilence.Add(1)
				h.sendToNtg(h.silence)
				return
			}
			st.lastDecodedSamples = n
			st.pending10ms = appendChunks10ms(st.pending10ms, h.pcmBuf[:n*Channels])
		}
	}
	chunk := st.pending10ms[0]
	st.pending10ms = st.pending10ms[1:]
	h.sendToNtg(chunk)
}

// slice interleaved stereo PCM into 10ms chunks; partial trailing chunks
// are dropped (libopus's 2.5/5ms frames aren't multiples of 10ms)
func appendChunks10ms(dst [][]byte, pcm []int16) [][]byte {
	samplesPerChunk := SamplesPerNtgFrame * Channels
	for off := 0; off+samplesPerChunk <= len(pcm); off += samplesPerChunk {
		buf := make([]byte, NtgFrameBytes)
		PCMInt16ToBytes(pcm[off:off+samplesPerChunk], buf)
		dst = append(dst, buf)
	}
	return dst
}

func (h *XmppToTg) sendToNtg(pcm []byte) {
	h.cShipTotal.Add(1)
	if err := h.opts.Ntg.SendMicrophonePCM(h.opts.ChatID, pcm); err != nil {
		h.cSendErrs.Add(1)
		h.log.WithError(err).Debug("SendMicrophonePCM failed")
	}
}