go-sip/internal/callflow/flow.go

package callflow

import (
	"context"
	"errors"
	"fmt"
	"time"

	"git.ipao.vip/rogee/go-sip/internal/ai"
	"git.ipao.vip/rogee/go-sip/internal/media"
)

// MediaSession is the transport boundary of the approved call flow.
// The current isolated Mock uses MemorySession; ASR-only omits opening/reply TTS.
type MediaSession interface {
	ReadPayload(context.Context) ([]byte, error)
	SendPCM16(context.Context, []byte, int) error
	Stats() media.RTPStats
}

// CaptureWindow bounds the actual receipt of one user-side audio turn.
type CaptureWindow struct {
	StartedAt time.Time
	EndedAt   time.Time
}

type Result struct {
	Turn           ai.TurnResult
	Turns          []ai.TurnResult
	Inbound        []byte
	InboundTurns   [][]byte
	CaptureWindows []CaptureWindow
	OutboundTurns  [][]byte
	RTP            media.RTPStats
}

func executeFlow(ctx context.Context, session MediaSession, mode ai.Mode, open func(context.Context) ([]byte, error), runTurn func(context.Context, []byte) (ai.TurnResult, error), capture CaptureConfig) (Result, error) {
	if session == nil || open == nil || runTurn == nil {
		return Result{}, errors.New("media session and AI pipeline are required")
	}
	if mode != ai.ModeFullAI && mode != ai.ModeASROnly {
		return Result{}, fmt.Errorf("unsupported callflow AI mode %q", mode)
	}
	maxTurns := capture.MaxTurns
	if maxTurns <= 0 {
		maxTurns = 1
	}
	result := Result{}
	if mode == ai.ModeFullAI {
		openingPCM, err := open(ctx)
		if err != nil {
			return Result{}, err
		}
		if len(openingPCM) > 0 {
			if err := session.SendPCM16(ctx, openingPCM, 16000); err != nil {
				return result, err
			}
			result.OutboundTurns = append(result.OutboundTurns, clonePCM(openingPCM))
		}
	}
	for turnIndex := 0; turnIndex < maxTurns; turnIndex++ {
		captured, err := captureTurn(ctx, session, capture)
		if err != nil {
			return result, fmt.Errorf("capturing RTP turn %d after opening/reply prompt: %w", turnIndex+1, err)
		}
		inbound := captured.PCM
		result.Inbound = inbound
		result.InboundTurns = append(result.InboundTurns, clonePCM(inbound))
		result.CaptureWindows = append(result.CaptureWindows, captured.Window)
		if len(inbound) < 3200 {
			return result, fmt.Errorf("captured audio turn %d is too short", turnIndex+1)
		}
		turn, err := runTurn(ctx, inbound)
		if err != nil {
			return result, fmt.Errorf("run AI turn %d: %w", turnIndex+1, err)
		}
		result.Turn = turn
		result.Turns = append(result.Turns, turn)
		if turn.InvalidCall || turn.EndedByKeyword {
			result.RTP = session.Stats()
			return result, nil
		}
		if mode == ai.ModeASROnly {
			result.RTP = session.Stats()
			continue
		}
		if len(turn.AudioPCM16) == 0 {
			return result, fmt.Errorf("AI reply turn %d has no audio", turnIndex+1)
		}
		if err := session.SendPCM16(ctx, turn.AudioPCM16, 16000); err != nil {
			return result, fmt.Errorf("send AI reply turn %d: %w", turnIndex+1, err)
		}
		result.OutboundTurns = append(result.OutboundTurns, clonePCM(turn.AudioPCM16))
		result.RTP = session.Stats()
	}
	result.RTP = session.Stats()
	return result, nil
}

func clonePCM(pcm []byte) []byte {
	return append([]byte(nil), pcm...)
}

// MemorySession is a bounded transport adapter for mock/mixed-flow tests. It
// has no SIP semantics and never bypasses the shared call flow.
type MemorySession struct {
	inbound  [][]byte
	index    int
	outbound []byte
	stats    media.RTPStats
}

func NewMemorySession(pcm []byte) *MemorySession {
	const frameBytes = 640
	frames := make([][]byte, 0, (len(pcm)+frameBytes-1)/frameBytes)
	for offset := 0; offset < len(pcm); offset += frameBytes {
		end := offset + frameBytes
		if end > len(pcm) {
			end = len(pcm)
		}
		frames = append(frames, append([]byte(nil), pcm[offset:end]...))
	}
	return &MemorySession{inbound: frames}
}

func (m *MemorySession) ReadPayload(ctx context.Context) ([]byte, error) {
	if m.index < len(m.inbound) {
		payload := m.inbound[m.index]
		m.index++
		m.stats.ReceivedPackets++
		m.stats.ReceivedBytes += uint64(len(payload))
		return append([]byte(nil), payload...), nil
	}
	<-ctx.Done()
	return nil, ctx.Err()
}

func (m *MemorySession) SendPCM16(ctx context.Context, pcm []byte, _ int) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	m.outbound = append(m.outbound, pcm...)
	m.stats.SentBytes += uint64(len(pcm))
	m.stats.SentPackets += uint64((len(pcm) + 639) / 640)
	return nil
}

func (m *MemorySession) Stats() media.RTPStats { return m.stats }
func (m *MemorySession) OutboundPCM() []byte   { return append([]byte(nil), m.outbound...) }