// turns protocol events into bridged calls package calls import ( "fmt" "sync" "time" "dev.narayana.im/narayana/telegabber/calls/audio" "dev.narayana.im/narayana/telegabber/calls/signaling" "dev.narayana.im/narayana/telegabber/calls/signaling/tgsig" "dev.narayana.im/narayana/telegabber/calls/signaling/xmppsig" "dev.narayana.im/narayana/telegabber/xmpp/jingle" "github.com/pion/webrtc/v4" log "github.com/sirupsen/logrus" "gosrc.io/xmpp/stanza" ) // look up per-session adapters by bare jid; ok=false drops the call type SessionLookup func(bareJID string) (tg *tgsig.Adapter, xmpp *xmppsig.Adapter, ok bool) type Config struct { JingleManager *jingle.Manager TgManager *tgsig.Manager Lookup SessionLookup // zero falls through to signaling defaults (60s/20s/15s) RingTimeout time.Duration ExchangeTimeout time.Duration ConnectTimeout time.Duration } type Orchestrator struct { cfg Config } func New(cfg Config) *Orchestrator { o := &Orchestrator{cfg: cfg} if cfg.JingleManager != nil { cfg.JingleManager.OnProposal = o.onProposal } return o } // per-side surfaces the audio layer needs type tgAudioSide interface { NtgClient() audio.NtgClient ChatID() int64 } type xmppAudioSide interface { PeerConnection() *webrtc.PeerConnection // swap in the orchestrator's track handler and drain buffered tracks SetTrackHandler(fn func(t *webrtc.TrackRemote)) } type Call struct { bridge *signaling.Bridge mu sync.Mutex tgSide tgAudioSide xmppSide xmppAudioSide audioTgToXmpp *audio.TgToXmpp audioXmppToTg *audio.XmppToTg closeOnce sync.Once } func (c *Call) Bridge() *signaling.Bridge { return c.bridge } // build the audio halves once the negotiated opus fmtp is in the SDP func (c *Call) setupAudio(remoteSDP string) { remoteFmtp := audio.ExtractOpusFmtp(remoteSDP) c.mu.Lock() if c.audioTgToXmpp != nil || c.audioXmppToTg != nil { c.mu.Unlock() return } tgSide, xmppSide := c.tgSide, c.xmppSide c.mu.Unlock() if tgSide == nil || xmppSide == nil { return } ntg := tgSide.NtgClient() pc := xmppSide.PeerConnection() if ntg == nil || pc == nil { // signaling-only test setup return } chatID := tgSide.ChatID() tgToXmpp, err := audio.NewTgToXmpp(audio.TgToXmppOptions{ Ntg: ntg, ChatID: chatID, RemoteFmtp: remoteFmtp, }) if err != nil { log.WithError(err).Error("setupAudio: NewTgToXmpp") return } xmppToTg, err := audio.NewXmppToTg(audio.XmppToTgOptions{ Ntg: ntg, ChatID: chatID, }) if err != nil { _ = tgToXmpp.Close() log.WithError(err).Error("setupAudio: NewXmppToTg") return } if err := attachOutgoingTrack(pc, tgToXmpp.LocalTrack()); err != nil { log.WithField("chat_id", chatID).WithError(err).Warn("setupAudio: attach outgoing track failed") } xmppSide.SetTrackHandler(xmppToTg.HandleTrack) log.WithFields(log.Fields{ "chat_id": chatID, "remote_fmtp": remoteFmtp, }).Info("setupAudio: halves wired; awaiting startAudio on bridge OnEstablished") c.mu.Lock() c.audioTgToXmpp = tgToXmpp c.audioXmppToTg = xmppToTg c.mu.Unlock() } // swap the real outgoing track into the placeholder sender from buildPC func attachOutgoingTrack(pc *webrtc.PeerConnection, track *webrtc.TrackLocalStaticSample) error { for _, s := range pc.GetSenders() { if s == nil { continue } if err := s.ReplaceTrack(track); err == nil { return nil } } return fmt.Errorf("no sender accepted track") } // fired by Bridge.OnEstablished func (c *Call) startAudio() { c.mu.Lock() tgHalf := c.audioTgToXmpp xmppHalf := c.audioXmppToTg c.mu.Unlock() if tgHalf == nil || xmppHalf == nil { return } if err := tgHalf.Start(); err != nil { log.WithError(err).Warn("tg->xmpp audio Start failed") } if err := xmppHalf.Start(); err != nil { log.WithError(err).Warn("xmpp->tg audio Start failed") } } func (c *Call) Close() { c.closeOnce.Do(func() { c.mu.Lock() tgHalf := c.audioTgToXmpp xmppHalf := c.audioXmppToTg tgSide := c.tgSide c.audioTgToXmpp = nil c.audioXmppToTg = nil c.mu.Unlock() if tgHalf != nil { _ = tgHalf.Close() } if xmppHalf != nil { _ = xmppHalf.Close() } // release ntg streams set in setupAudio; both halves are built // together, skipped in signaling-only test setups if (tgHalf != nil || xmppHalf != nil) && tgSide != nil { if ntg := tgSide.NtgClient(); ntg != nil { _ = ntg.ClearStreams(tgSide.ChatID()) } } }) } // inbound JMI ; postReady fires after the session is in StateRinging // so a fast tdlib Pending{IsReceived:true} doesn't drop the JMI func (o *Orchestrator) onProposal(p jingle.IncomingProposal) (*jingle.Session, func(), error) { j, err := stanza.NewJid(p.From) if err != nil { return nil, nil, nil } tg, xmpp, ok := o.cfg.Lookup(j.Bare()) if !ok || tg == nil || xmpp == nil { return nil, nil, nil } userID, ok := xmppsig.ParseTargetJID(p.To) if !ok { return nil, nil, nil } callee, sess, err := xmpp.NewCallee(p) if err != nil { return nil, nil, err } caller := tg.NewCaller(userID, false) // audio only call := &Call{tgSide: caller, xmppSide: callee} br := o.newBridge(caller, callee, call) call.bridge = br callee.Bind(br) caller.Bind(br) // defer audio half construction until SDP lands so a pre-SDP cancel // doesn't spin up ntgcalls callee.SetOnRemoteSDP(call.setupAudio) // off the dispatcher goroutine - Caller.Start hits tdlib CreateCall return sess, func() { go br.Start() }, nil } // IncomingCallHandler for tgsig.Manager func (o *Orchestrator) NewFromTelegram(jid string, callID int32, userID int64, isVideo bool) (ok bool) { entry := log.WithFields(log.Fields{ "jid": jid, "call_id": callID, "user_id": userID, "is_video": isVideo, }) defer func() { if r := recover(); r != nil { entry.WithField("panic", r).Error("NewFromTelegram: panic recovered") ok = false } }() tg, xmpp, lookupOk := o.cfg.Lookup(jid) if !lookupOk || tg == nil || xmpp == nil { entry.Warn("NewFromTelegram: no live session for jid") return false } // xmpp side first - it's the only fallible step, otherwise a failure here // would leak the tg side until tdlib timed the call out caller, err := xmpp.NewCaller(jid, userID) if err != nil { entry.WithError(err).Error("NewFromTelegram: NewCaller failed") return false } callee := tg.NewCallee(callID, userID, isVideo) call := &Call{tgSide: callee, xmppSide: caller} br := o.newBridge(caller, callee, call) call.bridge = br callee.Bind(br) caller.Bind(br) caller.SetOnRemoteSDP(call.setupAudio) br.Start() return true } func (o *Orchestrator) newBridge(caller signaling.Caller, callee signaling.Callee, call *Call) *signaling.Bridge { timers := signaling.NewStdTimers() br := signaling.New(signaling.Config{ Caller: caller, Callee: callee, Timers: timers, RingTimeout: o.cfg.RingTimeout, ExchangeTimeout: o.cfg.ExchangeTimeout, ConnectTimeout: o.cfg.ConnectTimeout, OnEstablished: call.startAudio, OnTerminated: call.Close, }) timers.SetBridge(br) return br }