telegabber/calls/audio/tg_to_xmpp.go

package audio

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

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

// pion's default MediaEngine PT for opus; production reads the actual
// negotiated PT from TrackRemote, this is for tests with synthetic RTP
const opusPayloadType = 111

type TgToXmppOptions struct {
	Ntg    NtgClient
	ChatID int64 // P2P: equals the Telegram peer's user_id
	// params half of the peer's opus a=fmtp; empty -> defaults
	RemoteFmtp string
	Logger     *log.Entry // nil -> default logrus
}

// tg->xmpp half: ntgcalls Playback -> opus encode -> pion LocalTrack;
// callback-driven, no goroutines
type TgToXmpp struct {
	opts TgToXmppOptions
	log  *log.Entry

	enc   *Encoder
	track *webrtc.TrackLocalStaticSample

	mu          sync.Mutex
	acc         []byte // accumulates 10ms ntg frames until 20ms
	primarySSRC uint32 // first SSRC wins; mixing is wrong for 1:1
	opusOut     []byte // reusable encode destination

	closed  atomic.Bool
	started atomic.Bool

	unregister func() // returned by Ntg.OnFrame; called from Close

	// telemetry counters (atomic; sampled by the periodic stats log)
	cFramesIn        atomic.Int64
	cFramesWrongChat atomic.Int64
	cFramesWrongSSRC atomic.Int64
	cFramesBadSize   atomic.Int64
	cOpusOut         atomic.Int64
	cWriteErrs       atomic.Int64

	statsMu   sync.Mutex
	lastLog   time.Time
	callStart time.Time // first frame seen; anchor for cumulative drift
	prevSnap  tgToXmppSnap

	capture *captureWriter
}

type tgToXmppSnap struct {
	framesIn, wrongChat, wrongSSRC, badSize, opusOut, writeErrs int64
}

func (h *TgToXmpp) snap() tgToXmppSnap {
	return tgToXmppSnap{
		framesIn:  h.cFramesIn.Load(),
		wrongChat: h.cFramesWrongChat.Load(),
		wrongSSRC: h.cFramesWrongSSRC.Load(),
		badSize:   h.cFramesBadSize.Load(),
		opusOut:   h.cOpusOut.Load(),
		writeErrs: h.cWriteErrs.Load(),
	}
}

// builds the encoder, local opus track, registers ntgcalls frame handler;
// Start() flips on the ntgcalls source
func NewTgToXmpp(opts TgToXmppOptions) (*TgToXmpp, 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": "tg->xmpp", "chat_id": opts.ChatID})

	enc, err := NewEncoder()
	if err != nil {
		return nil, err
	}
	if err := ParseFmtp(opts.RemoteFmtp).ApplyTo(enc); err != nil {
		return nil, fmt.Errorf("apply fmtp: %w", err)
	}

	track, err := webrtc.NewTrackLocalStaticSample(
		webrtc.RTPCodecCapability{
			MimeType:    webrtc.MimeTypeOpus,
			ClockRate:   SampleRate,
			Channels:    Channels,
			SDPFmtpLine: "minptime=10;useinbandfec=1",
		},
		"audio",
		fmt.Sprintf("telegabber-%d", opts.ChatID),
	)
	if err != nil {
		return nil, fmt.Errorf("new track: %w", err)
	}

	h := &TgToXmpp{
		opts:    opts,
		log:     logger,
		enc:     enc,
		track:   track,
		acc:     make([]byte, 0, OpusInputBytes),
		opusOut: make([]byte, MaxOpusPacketBytes),
	}
	if cap, err := newCaptureWriter(
		fmt.Sprintf("tg-to-xmpp-%d-%d.bin", opts.ChatID, time.Now().Unix()),
		magicPCM,
	); err != nil {
		logger.WithError(err).Warn("capture: tg->xmpp writer disabled")
	} else if cap != nil {
		logger.Info("capture: tg->xmpp PCM capture enabled")
		h.capture = cap
	}

	// adapter pre-filters to (Playback, Microphone); handler also filters
	// by chatID. Close calls the returned cancel to drop our dispatch slot.
	h.unregister = opts.Ntg.OnFrame(h.onNtgFrames)
	return h, nil
}

func (h *TgToXmpp) LocalTrack() *webrtc.TrackLocalStaticSample { return h.track }

func (h *TgToXmpp) Start() error {
	if !h.started.CompareAndSwap(false, true) {
		return nil
	}
	if err := h.opts.Ntg.SetExternalMicrophone(h.opts.ChatID, false, SampleRate, Channels); err != nil {
		return fmt.Errorf("SetExternalMicrophone(playback): %w", err)
	}
	h.log.Info("tg->xmpp half started")
	return nil
}

// flips the closed flag and releases the capture writer; ntg sources
// are released by the xmpp->tg half's Close
func (h *TgToXmpp) Close() error {
	if !h.closed.CompareAndSwap(false, true) {
		return nil
	}
	if h.unregister != nil {
		h.unregister()
	}
	h.capture.close()
	h.log.Info("tg->xmpp half closed")
	return nil
}

func (h *TgToXmpp) onNtgFrames(chatID int64, frames []PCMFrame) {
	if h.closed.Load() {
		return
	}
	if chatID != h.opts.ChatID {
		h.cFramesWrongChat.Add(int64(len(frames)))
		return
	}
	h.cFramesIn.Add(int64(len(frames)))
	for _, f := range frames {
		h.capture.writePCM(f.SSRC, f.Data)
		h.feedFrame(f)
	}
	h.maybeLogStats()
}

// maybeLogStats emits one aggregated line per statsInterval from the ntg
// callback goroutine (this direction has no ticker to hang it off).
func (h *TgToXmpp) maybeLogStats() {
	now := time.Now()
	h.statsMu.Lock()
	if h.lastLog.IsZero() {
		h.lastLog = now
		h.callStart = now
		h.prevSnap = h.snap()
		h.statsMu.Unlock()
		return
	}
	callStart := h.callStart
	elapsed := now.Sub(h.lastLog)
	if elapsed < statsInterval {
		h.statsMu.Unlock()
		return
	}
	prev := h.prevSnap
	cur := h.snap()
	h.lastLog = now
	h.prevSnap = cur
	h.statsMu.Unlock()

	secs := elapsed.Seconds()
	if secs <= 0 {
		secs = 1
	}
	rate := func(d int64) float64 { return float64(d) / secs }
	h.mu.Lock()
	accBytes := len(h.acc)
	ssrc := h.primarySSRC
	h.mu.Unlock()
	// deliver_drift_ms = frames ntgcalls delivered minus what real time allows.
	// Grows if ntgcalls over-delivers: that excess is audio we forward to pion
	// faster than real time, piling up in the XMPP client's jitter buffer.
	sinceStart := now.Sub(callStart)
	expected := sinceStart.Milliseconds() / NtgFrameMs
	driftMs := (cur.framesIn - expected) * NtgFrameMs
	h.log.WithFields(log.Fields{
		"interval_ms":      elapsed.Milliseconds(),
		"frames_in_per_s":  rate(cur.framesIn - prev.framesIn),
		"opus_out_per_s":   rate(cur.opusOut - prev.opusOut),
		"deliver_drift_ms": driftMs,
		"call_secs":        int64(sinceStart.Seconds()),
		"wrong_chat":       cur.wrongChat - prev.wrongChat,
		"wrong_ssrc":       cur.wrongSSRC - prev.wrongSSRC,
		"bad_size":         cur.badSize - prev.badSize,
		"write_errs":       cur.writeErrs - prev.writeErrs,
		"acc_bytes":        accBytes,
		"primary_ssrc":     ssrc,
	}).Debug("tg->xmpp stats")
}

// returns number of opus packets emitted (0 or 1)
func (h *TgToXmpp) feedFrame(f PCMFrame) int {
	if len(f.Data) != NtgFrameBytes {
		h.cFramesBadSize.Add(1)
		h.log.WithField("len", len(f.Data)).Warn("unexpected ntg frame size")
		return 0
	}
	h.mu.Lock()
	defer h.mu.Unlock()

	if h.primarySSRC == 0 {
		h.primarySSRC = f.SSRC
	} else if f.SSRC != h.primarySSRC {
		h.cFramesWrongSSRC.Add(1)
		return 0
	}

	h.acc = append(h.acc, f.Data...)
	if len(h.acc) < OpusInputBytes {
		return 0
	}

	pcm := make([]int16, SamplesPerOpusFrame*Channels)
	PCMBytesToInt16(h.acc[:OpusInputBytes], pcm)
	h.acc = h.acc[:0]

	n, err := h.enc.Encode(pcm, h.opusOut)
	if err != nil {
		h.log.WithError(err).Warn("opus encode failed")
		return 0
	}
	// copy because the next encode reuses h.opusOut
	payload := make([]byte, n)
	copy(payload, h.opusOut[:n])
	if err := h.track.WriteSample(media.Sample{
		Data:     payload,
		Duration: OpusFrameMs * time.Millisecond,
	}); err != nil {
		h.cWriteErrs.Add(1)
		h.log.WithError(err).Warn("WriteSample failed")
		return 0
	}
	h.cOpusOut.Add(1)
	return 1
}