magic-sonar/cmd/sip-demo/main.go

package main

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/emiago/sipgo"
	"github.com/emiago/sipgo/sip"
	"github.com/gobwas/ws"
	"github.com/gobwas/ws/wsutil"
	"github.com/zaf/g711"
)

const (
	defaultSIPServer = "61.132.228.221:5060"
	caller           = "BD93205882"
	calleePrefix     = "7089"
	defaultTTS       = "您好，这里是北京美珈科技有限公司的人工智能语音测试，本次通话将被录音。"
)

type config struct {
	server, number, audio, ttsText, ttsVoice, logPath, recordDir, localAddr, advertiseIP string
	ringTimeout, maxDuration                                                             time.Duration
	vadThreshold                                                                         int
}

func main() {
	cfg := config{number: os.Getenv("SIP_TEST_NUMBER")}
	flag.StringVar(&cfg.server, "sip-server", envOr("SIP_SERVER", defaultSIPServer), "SIP host:port")
	flag.StringVar(&cfg.audio, "audio", "", "optional mono PCM16 WAV to play after TTS")
	flag.StringVar(&cfg.ttsText, "tts-text", defaultTTS, "text synthesized before the call")
	flag.StringVar(&cfg.ttsVoice, "tts-voice", "longxiaochun", "Bailian voice ID")
	flag.StringVar(&cfg.logPath, "sip-log", "logs/sip.log", "redacted SIP log path")
	flag.StringVar(&cfg.recordDir, "record-dir", "recordings", "untracked RX/TX WAV directory")
	flag.StringVar(&cfg.localAddr, "sip-local", envOr("SIP_LOCAL_ADDR", "0.0.0.0:0"), "local SIP UDP address")
	flag.StringVar(&cfg.advertiseIP, "advertise-ip", os.Getenv("SIP_ADVERTISE_IP"), "SDP/Via IP; auto-detected when empty")
	flag.DurationVar(&cfg.ringTimeout, "ring-timeout", 45*time.Second, "maximum ringing time")
	flag.DurationVar(&cfg.maxDuration, "max-duration", 60*time.Second, "maximum answered call duration")
	flag.IntVar(&cfg.vadThreshold, "vad-threshold", 1200, "inbound PCM RMS needed to interrupt playback")
	flag.Parse()

	if err := run(context.Background(), cfg); err != nil {
		fmt.Fprintln(os.Stderr, "sip-demo:", err)
		os.Exit(1)
	}
}

func run(ctx context.Context, cfg config) error {
	return runWithACK(ctx, cfg, func(ctx context.Context, session *sipgo.DialogClientSession) error {
		return session.Ack(ctx)
	})
}

func runWithACK(ctx context.Context, cfg config, ack func(context.Context, *sipgo.DialogClientSession) error) error {
	if err := validateNumber(cfg.number); err != nil {
		return fmt.Errorf("SIP_TEST_NUMBER: %w", err)
	}
	if cfg.maxDuration <= 0 || cfg.ringTimeout <= 0 || cfg.vadThreshold <= 0 {
		return errors.New("timeouts and vad-threshold must be positive")
	}
	host, portText, err := net.SplitHostPort(cfg.server)
	if err != nil {
		return fmt.Errorf("invalid SIP server: %w", err)
	}
	port, err := strconv.Atoi(portText)
	if err != nil || port < 1 || port > 65535 {
		return errors.New("invalid SIP server port")
	}

	program, err := mediaProgram(ctx, cfg)
	if err != nil {
		return err
	}
	rtpConn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0})
	if err != nil {
		return fmt.Errorf("listen RTP: %w", err)
	}
	defer rtpConn.Close()
	advertiseIP := cfg.advertiseIP
	if advertiseIP == "" {
		advertiseIP, err = routeIP(cfg.server)
		if err != nil {
			return err
		}
	}
	if net.ParseIP(advertiseIP) == nil {
		return errors.New("advertise-ip must be an IP address")
	}

	log, err := newSignalLog(cfg.logPath, cfg.number, caller)
	if err != nil {
		return err
	}
	defer log.Close()
	sdp := offerSDP(advertiseIP, rtpConn.LocalAddr().(*net.UDPAddr).Port)

	ua, err := sipgo.NewUA(sipgo.WithUserAgent(caller), sipgo.WithUserAgentHostname(host))
	if err != nil {
		return fmt.Errorf("create SIP UA: %w", err)
	}
	defer ua.Close()
	client, err := sipgo.NewClient(ua,
		sipgo.WithClientConnectionAddr(cfg.localAddr),
		sipgo.WithClientHostname(advertiseIP),
		sipgo.WithClientNAT(),
	)
	if err != nil {
		return fmt.Errorf("create SIP client: %w", err)
	}
	server, err := sipgo.NewServer(ua)
	if err != nil {
		return fmt.Errorf("create SIP server: %w", err)
	}
	dialogs := sipgo.NewDialogClientCache(client, sip.ContactHeader{Address: sip.Uri{User: caller}})
	remoteHangup := make(chan struct{})
	var hangupOnce sync.Once
	server.OnBye(func(req *sip.Request, tx sip.ServerTransaction) {
		log.message("<<<", req.String())
		if err := dialogs.ReadBye(req, tx); err != nil {
			log.event("failed to process remote BYE: " + err.Error())
			_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusCallTransactionDoesNotExists, "Call/Transaction Does Not Exist", nil))
		}
		hangupOnce.Do(func() { close(remoteHangup) })
	})

	callee := calleePrefix + cfg.number
	recipient := sip.Uri{Scheme: "sip", User: callee, Host: host, Port: port}
	ringCtx, cancelRing := context.WithTimeout(ctx, cfg.ringTimeout)
	defer cancelRing()
	session, err := dialogs.Invite(ringCtx, recipient, sdp, sip.NewHeader("Content-Type", "application/sdp"))
	if err != nil {
		return fmt.Errorf("send INVITE: %w", err)
	}
	defer session.Close()
	log.message(">>>", session.InviteRequest.String())

	err = session.WaitAnswer(ringCtx, sipgo.AnswerOptions{OnResponse: func(res *sip.Response) error {
		log.message("<<<", res.String())
		return nil
	}})
	if err != nil {
		var responseErr *sipgo.ErrDialogResponse
		if errors.As(err, &responseErr) && (responseErr.Res.StatusCode == sip.StatusUnauthorized || responseErr.Res.StatusCode == sip.StatusProxyAuthRequired) {
			return errors.New("trunk requested authentication although this MVP is configured for IP whitelist only")
		}
		return fmt.Errorf("wait for answer: %w", err)
	}
	acked := false
	var callErr error
	if err := ack(ctx, session); err != nil {
		callErr = fmt.Errorf("send ACK: %w", err)
	} else {
		acked = true
		log.event(fmt.Sprintf(">>> ACK call-id=%s cseq=%d", session.InviteRequest.CallID().Value(), session.CSEQ()))
		remoteRTP, payloadType, err := answerMedia(session.InviteResponse.Body())
		if err != nil {
			callErr = err
		} else {
			callErr = runMedia(ctx, rtpConn, remoteRTP, payloadType, program, cfg.maxDuration, cfg.vadThreshold, remoteHangup, cfg.recordDir, log)
		}
	}
	if session.LoadState() != sip.DialogStateEnded {
		callErr = errors.Join(callErr, endAnsweredSession(session, log, acked))
	}
	return callErr
}

func endAnsweredSession(session *sipgo.DialogClientSession, log *signalLog, acked bool) error {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	log.event(fmt.Sprintf(">>> BYE call-id=%s cseq=%d", session.InviteRequest.CallID().Value(), session.CSEQ()+1))
	if acked {
		if err := session.Bye(ctx); err != nil {
			return fmt.Errorf("send BYE: %w", err)
		}
		return nil
	}

	recipient := session.InviteRequest.Recipient
	if contact := session.InviteResponse.Contact(); contact != nil {
		recipient = contact.Address
	}
	response, err := session.Do(ctx, sip.NewRequest(sip.BYE, recipient))
	if err != nil {
		return fmt.Errorf("send BYE after ACK failure: %w", err)
	}
	if response.StatusCode != sip.StatusOK {
		return fmt.Errorf("send BYE after ACK failure: %w", &sipgo.ErrDialogResponse{Res: response})
	}
	return nil
}

func mediaProgram(ctx context.Context, cfg config) ([]int16, error) {
	var out []int16
	if cfg.ttsText != "" {
		base, key := os.Getenv("BAILIAN_BASE_URL"), os.Getenv("BAILIAN_API_KEY")
		if base == "" || key == "" {
			return nil, errors.New("BAILIAN_BASE_URL and BAILIAN_API_KEY are required when tts-text is set")
		}
		ttsCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
		pcm, rate, err := synthesize(ttsCtx, base, key, cfg.ttsVoice, cfg.ttsText)
		cancel()
		if err != nil {
			return nil, fmt.Errorf("Bailian TTS: %w", err)
		}
		out = append(out, resample(pcm, rate, 8000)...)
	}
	if cfg.audio != "" {
		f, err := os.Open(cfg.audio)
		if err != nil {
			return nil, fmt.Errorf("open audio: %w", err)
		}
		pcm, rate, err := readWAV(io.LimitReader(f, 64<<20))
		f.Close()
		if err != nil {
			return nil, fmt.Errorf("read audio: %w", err)
		}
		out = append(out, resample(pcm, rate, 8000)...)
	} else {
		out = append(out, tone(440, 2*time.Second)...)
	}
	return out, nil
}

func synthesize(ctx context.Context, base, key, voice, text string) ([]int16, int, error) {
	u, err := url.Parse(base)
	if err != nil || u.Host == "" || u.User != nil || u.Path != "/api-ws/v1/inference" || u.RawPath != "" ||
		u.RawQuery != "" || u.Fragment != "" || (u.Scheme != "wss" && !(u.Scheme == "ws" && isLoopback(u.Hostname()))) {
		return nil, 0, errors.New("invalid BAILIAN_BASE_URL")
	}
	taskID, err := newTaskID()
	if err != nil {
		return nil, 0, fmt.Errorf("create TTS task ID: %w", err)
	}
	status := 0
	dialer := ws.Dialer{
		Header: ws.HandshakeHeaderHTTP(http.Header{"Authorization": []string{"Bearer " + key}}),
		OnStatusError: func(code int, _ []byte, _ io.Reader) {
			status = code
		},
	}
	conn, buffered, _, err := dialer.Dial(ctx, u.String())
	if err != nil {
		if status != 0 {
			return nil, 0, fmt.Errorf("WebSocket handshake HTTP %d", status)
		}
		return nil, 0, fmt.Errorf("connect TTS WebSocket: %w", err)
	}
	defer conn.Close()
	if deadline, ok := ctx.Deadline(); ok {
		_ = conn.SetDeadline(deadline)
	}
	var stream io.ReadWriter = conn
	if buffered != nil {
		defer ws.PutReader(buffered)
		stream = struct {
			io.Reader
			io.Writer
		}{buffered, conn}
	}
	runTask := map[string]any{
		"header": map[string]any{"action": "run-task", "task_id": taskID, "streaming": "duplex"},
		"payload": map[string]any{
			"task_group": "audio", "task": "tts", "function": "SpeechSynthesizer", "model": "cosyvoice-v3.5-plus",
			"parameters": map[string]any{"text_type": "PlainText", "voice": voice, "format": "pcm", "sample_rate": 24000},
			"input":      map[string]any{},
		},
	}
	if err := writeTTSCommand(stream, runTask); err != nil {
		return nil, 0, err
	}
	if err := waitTTSEvent(stream, taskID, "task-started"); err != nil {
		return nil, 0, err
	}
	for _, command := range []map[string]any{
		{"header": map[string]any{"action": "continue-task", "task_id": taskID, "streaming": "duplex"}, "payload": map[string]any{"input": map[string]any{"text": text}}},
		{"header": map[string]any{"action": "finish-task", "task_id": taskID, "streaming": "duplex"}, "payload": map[string]any{"input": map[string]any{}}},
	} {
		if err := writeTTSCommand(stream, command); err != nil {
			return nil, 0, err
		}
	}
	var audio bytes.Buffer
	for {
		data, op, err := wsutil.ReadServerData(stream)
		if err != nil {
			return nil, 0, fmt.Errorf("read TTS WebSocket: %w", err)
		}
		if op == ws.OpBinary {
			if audio.Len()+len(data) > 32<<20 {
				return nil, 0, errors.New("audio response exceeds 32 MiB")
			}
			_, _ = audio.Write(data)
			continue
		}
		event, err := parseTTSEvent(data, taskID)
		if err != nil {
			return nil, 0, err
		}
		switch event.Header.Event {
		case "task-finished":
			data = audio.Bytes()
			if len(data) == 0 {
				return nil, 0, errors.New("TTS response contains no audio")
			}
			if len(data)%2 != 0 {
				return nil, 0, errors.New("TTS response is not PCM16")
			}
			return bytesToPCM(data), 24000, nil
		case "task-failed":
			return nil, 0, ttsFailure(event)
		}
	}
}

type ttsEvent struct {
	Header struct {
		Event     string `json:"event"`
		TaskID    string `json:"task_id"`
		ErrorCode string `json:"error_code"`
	} `json:"header"`
}

func writeTTSCommand(w io.Writer, command any) error {
	data, err := json.Marshal(command)
	if err != nil {
		return fmt.Errorf("encode TTS command: %w", err)
	}
	if err := wsutil.WriteClientText(w, data); err != nil {
		return fmt.Errorf("write TTS WebSocket: %w", err)
	}
	return nil
}

func waitTTSEvent(stream io.ReadWriter, taskID, want string) error {
	for {
		data, op, err := wsutil.ReadServerData(stream)
		if err != nil {
			return fmt.Errorf("read TTS WebSocket: %w", err)
		}
		if op != ws.OpText {
			return fmt.Errorf("expected TTS event %q", want)
		}
		event, err := parseTTSEvent(data, taskID)
		if err != nil {
			return err
		}
		if event.Header.Event == "task-failed" {
			return ttsFailure(event)
		}
		if event.Header.Event == want {
			return nil
		}
	}
}

func parseTTSEvent(data []byte, taskID string) (ttsEvent, error) {
	var event ttsEvent
	if err := json.Unmarshal(data, &event); err != nil {
		return event, fmt.Errorf("decode TTS event: %w", err)
	}
	if event.Header.Event == "" || event.Header.TaskID != taskID {
		return event, errors.New("invalid TTS event")
	}
	return event, nil
}

func ttsFailure(event ttsEvent) error {
	if event.Header.ErrorCode == "" {
		return errors.New("TTS task failed")
	}
	return fmt.Errorf("TTS task failed: %s", event.Header.ErrorCode)
}

func newTaskID() (string, error) {
	var raw [16]byte
	if _, err := rand.Read(raw[:]); err != nil {
		return "", err
	}
	raw[6] = raw[6]&0x0f | 0x40
	raw[8] = raw[8]&0x3f | 0x80
	var id [36]byte
	hex.Encode(id[0:8], raw[0:4])
	id[8] = '-'
	hex.Encode(id[9:13], raw[4:6])
	id[13] = '-'
	hex.Encode(id[14:18], raw[6:8])
	id[18] = '-'
	hex.Encode(id[19:23], raw[8:10])
	id[23] = '-'
	hex.Encode(id[24:36], raw[10:16])
	return string(id[:]), nil
}

func isLoopback(host string) bool {
	ip := net.ParseIP(host)
	return strings.EqualFold(host, "localhost") || ip != nil && ip.IsLoopback()
}

func runMedia(parent context.Context, conn *net.UDPConn, remote *net.UDPAddr, payloadType uint8, audio []int16, duration time.Duration, threshold int, remoteHangup <-chan struct{}, recordDir string, log *signalLog) error {
	ctx, cancel := context.WithTimeout(parent, duration)
	defer cancel()
	interrupted := make(chan struct{})
	received := make(chan []int16, 1)
	go receiveRTP(ctx, conn, remote, threshold, int(duration.Seconds()*8000), interrupted, received)

	var sent []int16
	seq := randomUint16()
	timestamp, ssrc := randomUint32(), randomUint32()
	ticker := time.NewTicker(20 * time.Millisecond)
	defer ticker.Stop()
	playing := true
	for offset := 0; offset < len(audio) && playing; offset += 160 {
		end := min(offset+160, len(audio))
		frame := make([]int16, 160)
		copy(frame, audio[offset:end])
		payload := encodeG711(pcmToBytes(frame), payloadType)
		packet := marshalRTP(payloadType, seq, timestamp, ssrc, payload)
		if _, err := conn.WriteToUDP(packet, remote); err != nil {
			cancel()
			<-received
			return fmt.Errorf("send RTP: %w", err)
		}
		sent = append(sent, frame...)
		seq++
		timestamp += 160
		select {
		case <-ticker.C:
		case <-interrupted:
			log.event("playback interrupted by inbound voice")
			playing = false
		case <-remoteHangup:
			playing = false
			cancel()
		case <-ctx.Done():
			playing = false
		}
	}

	select {
	case <-remoteHangup:
	case <-ctx.Done():
	}
	cancel()
	rx := <-received
	if err := os.MkdirAll(recordDir, 0o700); err != nil {
		return fmt.Errorf("create recording directory: %w", err)
	}
	stamp := time.Now().UTC().Format("20060102T150405Z")
	if err := writeWAV(filepath.Join(recordDir, "call-"+stamp+"-rx.wav"), rx, 8000); err != nil {
		return err
	}
	if err := writeWAV(filepath.Join(recordDir, "call-"+stamp+"-tx.wav"), sent, 8000); err != nil {
		return err
	}
	log.event(fmt.Sprintf("media ended rx_samples=%d tx_samples=%d", len(rx), len(sent)))
	return nil
}

func receiveRTP(ctx context.Context, conn *net.UDPConn, remote *net.UDPAddr, threshold, maxSamples int, interrupted chan<- struct{}, done chan<- []int16) {
	var pcm []int16
	voiceFrames := 0
	interruptedSent := false
	buf := make([]byte, 2048)
	for {
		_ = conn.SetReadDeadline(time.Now().Add(50 * time.Millisecond))
		n, source, err := conn.ReadFromUDP(buf)
		if err != nil {
			if errors.Is(err, os.ErrDeadlineExceeded) {
				select {
				case <-ctx.Done():
					done <- pcm
					return
				default:
					continue
				}
			}
			done <- pcm
			return
		}
		if !source.IP.Equal(remote.IP) || source.Port != remote.Port {
			continue
		}
		payload, pt, ok := rtpPayload(buf[:n])
		if !ok || (pt != 8 && pt != 0) {
			continue
		}
		frame := bytesToPCM(decodeG711(payload, pt))
		if remaining := maxSamples - len(pcm); remaining > 0 {
			pcm = append(pcm, frame[:min(len(frame), remaining)]...)
		}
		if rms(frame) >= threshold {
			voiceFrames++
		} else {
			voiceFrames = 0
		}
		if voiceFrames >= 3 && !interruptedSent {
			close(interrupted)
			interruptedSent = true
		}
	}
}

func offerSDP(ip string, port int) []byte {
	return []byte(fmt.Sprintf("v=0\r\no=- 0 0 IN IP4 %s\r\ns=magic-sonar\r\nc=IN IP4 %s\r\nt=0 0\r\nm=audio %d RTP/AVP 8 0\r\na=rtpmap:8 PCMA/8000\r\na=rtpmap:0 PCMU/8000\r\na=ptime:20\r\na=sendrecv\r\n", ip, ip, port))
}

func answerMedia(body []byte) (*net.UDPAddr, uint8, error) {
	var host string
	port := 0
	var payload uint8 = 255
	for _, raw := range strings.Split(strings.ReplaceAll(string(body), "\r\n", "\n"), "\n") {
		fields := strings.Fields(raw)
		if len(fields) == 3 && fields[0] == "c=IN" && fields[1] == "IP4" {
			host = fields[2]
		}
		if len(fields) >= 4 && fields[0] == "m=audio" {
			port, _ = strconv.Atoi(fields[1])
			for _, offered := range fields[3:] {
				if offered == "8" {
					payload = 8
					break
				}
				if offered == "0" {
					payload = 0
				}
			}
		}
	}
	if net.ParseIP(host) == nil || port < 1 || port > 65535 || payload == 255 {
		return nil, 0, errors.New("answer SDP lacks usable IPv4 G.711 media")
	}
	return &net.UDPAddr{IP: net.ParseIP(host), Port: port}, payload, nil
}

func marshalRTP(pt uint8, seq uint16, timestamp, ssrc uint32, payload []byte) []byte {
	p := make([]byte, 12+len(payload))
	p[0], p[1] = 0x80, pt&0x7f
	binary.BigEndian.PutUint16(p[2:4], seq)
	binary.BigEndian.PutUint32(p[4:8], timestamp)
	binary.BigEndian.PutUint32(p[8:12], ssrc)
	copy(p[12:], payload)
	return p
}

func rtpPayload(packet []byte) ([]byte, uint8, bool) {
	if len(packet) < 12 || packet[0]>>6 != 2 {
		return nil, 0, false
	}
	cc := int(packet[0] & 0x0f)
	offset := 12 + cc*4
	if len(packet) < offset {
		return nil, 0, false
	}
	if packet[0]&0x10 != 0 {
		if len(packet) < offset+4 {
			return nil, 0, false
		}
		offset += 4 + int(binary.BigEndian.Uint16(packet[offset+2:offset+4]))*4
	}
	end := len(packet)
	if packet[0]&0x20 != 0 {
		padding := int(packet[len(packet)-1])
		if padding == 0 || padding > end-offset {
			return nil, 0, false
		}
		end -= padding
	}
	if offset > end {
		return nil, 0, false
	}
	return packet[offset:end], packet[1] & 0x7f, true
}

func encodeG711(pcm []byte, pt uint8) []byte {
	if pt == 0 {
		return g711.EncodeUlaw(pcm)
	}
	return g711.EncodeAlaw(pcm)
}

func decodeG711(data []byte, pt uint8) []byte {
	if pt == 0 {
		return g711.DecodeUlaw(data)
	}
	return g711.DecodeAlaw(data)
}

func readWAV(r io.Reader) ([]int16, int, error) {
	data, err := io.ReadAll(r)
	if err != nil || len(data) < 12 || string(data[:4]) != "RIFF" || string(data[8:12]) != "WAVE" {
		return nil, 0, errors.New("not a RIFF/WAVE file")
	}
	var rate int
	var pcm []byte
	for offset := 12; offset+8 <= len(data); {
		size := int(binary.LittleEndian.Uint32(data[offset+4 : offset+8]))
		start, end := offset+8, offset+8+size
		if end > len(data) {
			return nil, 0, errors.New("truncated WAV chunk")
		}
		switch string(data[offset : offset+4]) {
		case "fmt ":
			if size < 16 || binary.LittleEndian.Uint16(data[start:start+2]) != 1 || binary.LittleEndian.Uint16(data[start+2:start+4]) != 1 || binary.LittleEndian.Uint16(data[start+14:start+16]) != 16 {
				return nil, 0, errors.New("WAV must be mono PCM16")
			}
			rate = int(binary.LittleEndian.Uint32(data[start+4 : start+8]))
		case "data":
			pcm = data[start:end]
		}
		offset = end + size%2
	}
	if rate < 8000 || rate > 96000 || len(pcm) == 0 || len(pcm)%2 != 0 {
		return nil, 0, errors.New("WAV lacks valid fmt/data chunks")
	}
	return bytesToPCM(pcm), rate, nil
}

func writeWAV(path string, pcm []int16, rate int) error {
	data := pcmToBytes(pcm)
	var out bytes.Buffer
	out.WriteString("RIFF")
	_ = binary.Write(&out, binary.LittleEndian, uint32(36+len(data)))
	out.WriteString("WAVEfmt ")
	_ = binary.Write(&out, binary.LittleEndian, uint32(16))
	_ = binary.Write(&out, binary.LittleEndian, uint16(1))
	_ = binary.Write(&out, binary.LittleEndian, uint16(1))
	_ = binary.Write(&out, binary.LittleEndian, uint32(rate))
	_ = binary.Write(&out, binary.LittleEndian, uint32(rate*2))
	_ = binary.Write(&out, binary.LittleEndian, uint16(2))
	_ = binary.Write(&out, binary.LittleEndian, uint16(16))
	out.WriteString("data")
	_ = binary.Write(&out, binary.LittleEndian, uint32(len(data)))
	out.Write(data)
	if err := os.WriteFile(path, out.Bytes(), 0o600); err != nil {
		return fmt.Errorf("write recording: %w", err)
	}
	return nil
}

func resample(in []int16, from, to int) []int16 {
	if len(in) == 0 || from <= 0 || to <= 0 || from == to {
		return append([]int16(nil), in...)
	}
	out := make([]int16, int(math.Ceil(float64(len(in))*float64(to)/float64(from))))
	for i := range out {
		position := float64(i) * float64(from) / float64(to)
		left := min(int(position), len(in)-1)
		right := min(left+1, len(in)-1)
		fraction := position - float64(left)
		out[i] = int16(float64(in[left])*(1-fraction) + float64(in[right])*fraction)
	}
	return out
}

func tone(hz float64, duration time.Duration) []int16 {
	n := int(duration.Seconds() * 8000)
	out := make([]int16, n)
	for i := range out {
		out[i] = int16(5000 * math.Sin(2*math.Pi*hz*float64(i)/8000))
	}
	return out
}

func rms(pcm []int16) int {
	if len(pcm) == 0 {
		return 0
	}
	var sum float64
	for _, sample := range pcm {
		sum += float64(sample) * float64(sample)
	}
	return int(math.Sqrt(sum / float64(len(pcm))))
}

func pcmToBytes(pcm []int16) []byte {
	out := make([]byte, len(pcm)*2)
	for i, sample := range pcm {
		binary.LittleEndian.PutUint16(out[i*2:], uint16(sample))
	}
	return out
}

func bytesToPCM(data []byte) []int16 {
	out := make([]int16, len(data)/2)
	for i := range out {
		out[i] = int16(binary.LittleEndian.Uint16(data[i*2:]))
	}
	return out
}

type signalLog struct {
	mu             sync.Mutex
	file           *os.File
	number, callee string
	caller         string
}

func newSignalLog(path, number, callerID string) (*signalLog, error) {
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return nil, fmt.Errorf("create log directory: %w", err)
	}
	f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
	if err != nil {
		return nil, fmt.Errorf("open SIP log: %w", err)
	}
	return &signalLog{file: f, number: number, callee: calleePrefix + number, caller: callerID}, nil
}

func (l *signalLog) Close() error { return l.file.Close() }

func (l *signalLog) message(direction, message string) {
	l.write(direction + "\n" + l.redact(message))
}

func (l *signalLog) event(message string) { l.write(l.redact(message)) }

func (l *signalLog) write(message string) {
	l.mu.Lock()
	defer l.mu.Unlock()
	_, _ = fmt.Fprintf(l.file, "[%s] %s\n", time.Now().UTC().Format(time.RFC3339Nano), strings.TrimSpace(message))
}

func (l *signalLog) redact(message string) string {
	message = strings.ReplaceAll(message, l.callee, calleePrefix+"*******"+l.number[len(l.number)-4:])
	message = strings.ReplaceAll(message, l.number, "*******"+l.number[len(l.number)-4:])
	message = strings.ReplaceAll(message, l.caller, "BD****"+l.caller[len(l.caller)-4:])
	return mobileNumber.ReplaceAllString(message, "*******$1")
}

var mobileNumber = regexp.MustCompile("\\b1[3-9][0-9]{6}([0-9]{4})\\b")

func validateNumber(number string) error {
	if !regexp.MustCompile("^1[3-9][0-9]{9}$").MatchString(number) {
		return errors.New("must be an authorized 11-digit mainland mobile number supplied only through the environment")
	}
	return nil
}

func routeIP(server string) (string, error) {
	conn, err := net.Dial("udp4", server)
	if err != nil {
		return "", fmt.Errorf("detect route IP: %w", err)
	}
	defer conn.Close()
	return conn.LocalAddr().(*net.UDPAddr).IP.String(), nil
}

func envOr(name, fallback string) string {
	if value := os.Getenv(name); value != "" {
		return value
	}
	return fallback
}

func randomUint16() uint16 {
	var b [2]byte
	_, _ = rand.Read(b[:])
	return binary.BigEndian.Uint16(b[:])
}

func randomUint32() uint32 {
	var b [4]byte
	_, _ = rand.Read(b[:])
	return binary.BigEndian.Uint32(b[:])
}