go-sip/internal/asterisk/call.go

package asterisk

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/netip"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"git.ipao.vip/rogee/go-sip/internal/media"
	"github.com/CyCoreSystems/ari/v5"
)

// NativeDial contains only the Dispatcher-selected SIP identity and the
// Agent's explicitly configured SLIN16 RTP payload profile.
type NativeDial struct {
	ExecutionID, TrunkID, DialedCallee, CallerID string
	AnswerTimeout                                time.Duration
	MediaPayloadType                             uint8
}

// NativeCall owns one real ARI channel, bridge, ExternalMedia channel and RTP
// socket; closing it cannot originate or replay a second call.
type NativeCall struct {
	Media        *media.RTPStream
	Context      context.Context
	cancel       context.CancelCauseFunc
	monitorDone  chan struct{}
	destroyed    atomic.Bool
	client       ari.Client
	subscription ari.Subscription
	outbound     *ari.ChannelHandle
	external     *ari.ChannelHandle
	bridge       *ari.BridgeHandle
	closeOnce    sync.Once
	closeErr     error
}

// Originate uses the issued Cell-local ARI credential. The caller must have
// already persisted the signed instruction and passed the host capture gate.
func (l Loader) Originate(ctx context.Context, request NativeDial) (*NativeCall, error) {
	if _, err := approvedOriginateRequest(request.ExecutionID, request.TrunkID, request.DialedCallee, request.CallerID, request.AnswerTimeout); err != nil {
		return nil, err
	}
	client, err := l.OpenARI()
	if err != nil {
		return nil, err
	}
	return dialWithClient(ctx, client, request)
}

func dialWithClient(ctx context.Context, client ari.Client, request NativeDial) (_ *NativeCall, err error) {
	call := &NativeCall{client: client}
	defer func() {
		if err != nil {
			err = errors.Join(err, call.Close())
		}
	}()
	if ctx == nil || client == nil {
		return nil, errors.New("native ARI call context and client required")
	}
	call.Context, call.cancel = context.WithCancelCause(ctx)
	originate, err := approvedOriginateRequest(request.ExecutionID, request.TrunkID, request.DialedCallee, request.CallerID, request.AnswerTimeout)
	if err != nil {
		return nil, err
	}
	call.Media, err = media.ListenRTPWithFormat("127.0.0.1:0", request.MediaPayloadType, media.FormatSLIN16, 16000)
	if err != nil {
		return nil, fmt.Errorf("prepare Agent RTP capture: %w", err)
	}
	key := ari.NewKey(ari.ChannelKey, request.ExecutionID)
	call.subscription = client.Bus().Subscribe(key, "StasisStart", "StasisEnd", "ChannelHangupRequest", "ChannelDestroyed")
	if call.subscription == nil {
		return nil, errors.New("native ARI channel subscription unavailable before origination")
	}
	call.outbound, err = client.Channel().Originate(key, originate)
	if err != nil {
		// The HTTP response can be lost after Asterisk has created the
		// channel. Do not originate again: try to stop the same identity.
		call.outbound = client.Channel().Get(key)
		return nil, fmt.Errorf("native SIP origination outcome unknown: %T", err)
	}
	answerCtx, cancel := context.WithTimeout(ctx, request.AnswerTimeout)
	defer cancel()
	if err := awaitStasisStart(answerCtx, call.subscription, request.ExecutionID); err != nil {
		return nil, err
	}
	bridgeKey := ari.NewKey(ari.BridgeKey, request.ExecutionID+"-bridge")
	call.bridge, err = client.Bridge().Create(bridgeKey, "mixing", "go-sip-agent")
	if err != nil {
		// A lost reply does not prove that the bridge was not created.
		call.bridge = client.Bridge().Get(bridgeKey)
		return nil, fmt.Errorf("create Agent ARI mixing bridge outcome unknown: %T", err)
	}
	if err := call.bridge.AddChannel(call.outbound.ID()); err != nil {
		return nil, fmt.Errorf("add answered channel to ARI bridge: %T", err)
	}
	mediaKey := ari.NewKey(ari.ChannelKey, request.ExecutionID+"-media")
	call.external, err = client.Channel().ExternalMedia(mediaKey, ari.ExternalMediaOptions{
		App: "go-sip-agent", ExternalHost: net.JoinHostPort("127.0.0.1", strconv.Itoa(call.Media.LocalAddr().(*net.UDPAddr).Port)),
		Format: "slin16", Encapsulation: "rtp", Transport: "udp", ConnectionType: "client", Direction: "both",
	})
	if err != nil {
		// Use the same identity to stop an ExternalMedia channel created
		// before the HTTP outcome became unknown; never create another.
		call.external = client.Channel().Get(mediaKey)
		return nil, fmt.Errorf("create Agent ARI external media outcome unknown: %T", err)
	}
	address, err := call.external.GetVariable("UNICASTRTP_LOCAL_ADDRESS")
	if err != nil {
		return nil, fmt.Errorf("read Agent ARI RTP peer address: %T", err)
	}
	port, err := call.external.GetVariable("UNICASTRTP_LOCAL_PORT")
	if err != nil {
		return nil, fmt.Errorf("read Agent ARI RTP peer port: %T", err)
	}
	peer, err := netip.ParseAddr(address)
	if err != nil || !peer.IsLoopback() {
		return nil, errors.New("Agent ARI RTP peer is not a local Cell address")
	}
	if err := call.Media.SetPeer(net.JoinHostPort(address, port)); err != nil {
		return nil, fmt.Errorf("configure Agent RTP peer: %w", err)
	}
	if err := call.bridge.AddChannel(call.external.ID()); err != nil {
		return nil, fmt.Errorf("add Agent RTP channel to ARI bridge: %T", err)
	}
	call.monitorDone = make(chan struct{})
	go call.watch(request.ExecutionID)
	return call, nil
}

func (c *NativeCall) watch(channelID string) {
	defer close(c.monitorDone)
	for {
		select {
		case <-c.Context.Done():
			return
		case event, open := <-c.subscription.Events():
			if !open || event == nil {
				c.cancel(errors.New("native ARI event subscription lost during call"))
				return
			}
			if !eventBelongsToChannel(event, channelID) {
				continue
			}
			switch event.GetType() {
			case "ChannelDestroyed":
				if ended, ok := event.(*ari.ChannelDestroyed); ok {
					c.destroyed.Store(true)
					c.cancel(fmt.Errorf("native channel ended: cause=%d", ended.Cause))
				} else {
					c.cancel(errors.New("native channel ended: invalid destroy event"))
				}
				return
			case "ChannelHangupRequest", "StasisEnd":
				c.cancel(fmt.Errorf("native channel ended: event=%s", event.GetType()))
				return
			}
		}
	}
}

// Close releases each resource once. Errors remain visible to the caller; a
// failed Hangup never silently turns an uncertain call into a success.
func (c *NativeCall) Close() error {
	if c == nil {
		return nil
	}
	c.closeOnce.Do(func() {
		if c.cancel != nil {
			c.cancel(context.Canceled)
		}
		if c.subscription != nil {
			c.subscription.Cancel()
		}
		if c.external != nil {
			c.closeErr = errors.Join(c.closeErr, c.external.Hangup())
		}
		if c.outbound != nil && !c.destroyed.Load() {
			c.closeErr = errors.Join(c.closeErr, c.outbound.Hangup())
		}
		if c.bridge != nil {
			c.closeErr = errors.Join(c.closeErr, c.bridge.Delete())
		}
		if c.Media != nil {
			c.closeErr = errors.Join(c.closeErr, c.Media.Close())
		}
		if c.client != nil {
			c.client.Close()
		}
		if c.monitorDone != nil {
			<-c.monitorDone
		}
	})
	return c.closeErr
}