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

	capture *captureWriter
}

// 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() || chatID != h.opts.ChatID {
		return
	}
	for _, f := range frames {
		h.capture.writePCM(f.SSRC, f.Data)
		h.feedFrame(f)
	}
}

// returns number of opus packets emitted (0 or 1)
func (h *TgToXmpp) feedFrame(f PCMFrame) int {
	if len(f.Data) != NtgFrameBytes {
		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 {
		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.log.WithError(err).Warn("WriteSample failed")
		return 0
	}
	return 1
}