package callruntime

import (
	"context"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"net"
	"os"
	"path/filepath"
	"strings"
	"time"

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

const (
	defaultAnswerTimeout     = 45 * time.Second
	defaultTurnWindow        = 5 * time.Second
	defaultFirstSpeechWait   = 20 * time.Second
	defaultMaxTurnDuration   = 12 * time.Second
	defaultEndSilence        = 900 * time.Millisecond
	defaultVoiceThreshold    = 350
	defaultConversationTurns = 3
	defaultCallDuration      = 2 * time.Minute
)

type Config struct {
	ARIURL          string
	ARIWebsocketURL string
	ARIApplication  string
	ARIUsername     string
	ARIPassword     string

	Endpoint        string
	CallerID        string
	MediaBind       string
	MediaPort       int
	MediaFormat     string
	MediaSampleRate int
	PayloadType     uint8

	RecordingDirectory string
	AnswerTimeout      time.Duration
	TurnWindow         time.Duration
	FirstSpeechTimeout time.Duration
	MaxTurnDuration    time.Duration
	EndSilence         time.Duration
	VoiceThreshold     int
	MaxTurns           int
	MaxCallDuration    time.Duration
	OpeningPrompt      string

	Snapshot ai.Snapshot
	Pipeline ai.Pipeline
}

type RecordingFact struct {
	Segment    string
	Path       string
	SHA256     string
	Bytes      int // Complete file size, including the WAV header.
	DurationMS int64
}

type Result struct {
	ChannelID          string
	Transcript         string
	Reply              string
	InvalidCall        bool
	InvalidReason      string
	Turns              []ai.TurnResult
	InboundPath        string
	InboundSHA256      string
	InboundBytes       int
	OutboundPath       string
	OutboundSHA256     string
	OutboundBytes      int
	InboundRecordings  []RecordingFact
	OutboundRecordings []RecordingFact
	RTP                media.RTPStats
}

func (c Config) validate() error {
	if c.ARIApplication == "" || c.ARIURL == "" || c.ARIWebsocketURL == "" {
		return errors.New("ARI URL, websocket URL and application are required")
	}
	if c.ARIPassword == "" || c.ARIUsername == "" {
		return errors.New("ARI credentials are required")
	}
	if c.Endpoint == "" {
		return errors.New("endpoint is required")
	}
	if c.MediaBind == "" || c.MediaPort < 1024 || c.MediaPort > 65535 {
		return errors.New("valid media bind and port are required")
	}
	if err := media.ValidateFormat(media.Format(c.MediaFormat), c.PayloadType, c.MediaSampleRate); err != nil {
		return err
	}
	if c.Pipeline == nil {
		return errors.New("AI pipeline is required")
	}
	if c.AnswerTimeout <= 0 {
		c.AnswerTimeout = defaultAnswerTimeout
	}
	if c.TurnWindow <= 0 {
		c.TurnWindow = defaultTurnWindow
	}
	return nil
}

func (c Config) normalized() Config {
	if c.AnswerTimeout <= 0 {
		c.AnswerTimeout = defaultAnswerTimeout
	}
	if c.TurnWindow <= 0 {
		c.TurnWindow = defaultTurnWindow
	}
	if c.MediaFormat == "" {
		c.MediaFormat = string(media.FormatSLIN16)
	}
	if c.MediaSampleRate <= 0 {
		c.MediaSampleRate = 16000
	}
	if c.FirstSpeechTimeout <= 0 {
		c.FirstSpeechTimeout = defaultFirstSpeechWait
	}
	if c.MaxTurnDuration <= 0 {
		c.MaxTurnDuration = defaultMaxTurnDuration
	}
	if c.EndSilence <= 0 {
		c.EndSilence = defaultEndSilence
	}
	if c.VoiceThreshold <= 0 {
		c.VoiceThreshold = defaultVoiceThreshold
	}
	if c.MaxTurns <= 0 {
		c.MaxTurns = defaultConversationTurns
	}
	if c.MaxCallDuration <= 0 {
		c.MaxCallDuration = defaultCallDuration
	}
	if c.OpeningPrompt == "" {
		c.OpeningPrompt = "您好，请说出您想咨询的内容。"
	}
	return c
}

func Run(ctx context.Context, cfg Config) (Result, error) {
	cfg = cfg.normalized()
	if err := cfg.validate(); err != nil {
		return Result{}, err
	}
	runCtx, cancelRun := context.WithTimeout(ctx, cfg.MaxCallDuration)
	defer cancelRun()
	stream, err := media.ListenRTPWithFormat(fmt.Sprintf("%s:%d", cfg.MediaBind, cfg.MediaPort), cfg.PayloadType, media.Format(cfg.MediaFormat), cfg.MediaSampleRate)
	if err != nil {
		return Result{}, fmt.Errorf("bind external media: %w", err)
	}
	defer stream.Close()

	client, err := native.Connect(&native.Options{
		URL:          cfg.ARIURL,
		WebsocketURL: cfg.ARIWebsocketURL,
		Application:  cfg.ARIApplication,
		Username:     cfg.ARIUsername,
		Password:     cfg.ARIPassword,
		SubscribeAll: true,
	})
	if err != nil {
		return Result{}, fmt.Errorf("connect ARI: %w", err)
	}
	defer client.Close()

	starts := client.Bus().Subscribe(nil, ari.Events.All)
	defer starts.Cancel()
	originate, err := client.Channel().Originate(nil, ari.OriginateRequest{
		Endpoint: cfg.Endpoint,
		// ari.OriginateRequest.Timeout is specified in seconds.
		Timeout:  int(cfg.AnswerTimeout / time.Second),
		CallerID: cfg.CallerID,
		App:      cfg.ARIApplication,
		Formats:  cfg.MediaFormat,
	})
	if err != nil {
		return Result{}, fmt.Errorf("originate %s: %w", cfg.Endpoint, err)
	}
	channelID := originate.ID()
	channelKey := ari.NewKey(ari.ChannelKey, channelID)
	var bridgeKey *ari.Key
	var externalKey *ari.Key
	defer func() {
		if externalKey != nil {
			_ = client.Channel().Hangup(externalKey, "normal")
		}
		if channelKey != nil {
			_ = client.Channel().Hangup(channelKey, "normal")
		}
		if bridgeKey != nil {
			_ = client.Bridge().Delete(bridgeKey)
		}
	}()

	if err := waitForStasisStart(runCtx, starts, channelID, cfg.AnswerTimeout); err != nil {
		return Result{}, err
	}
	callCtx, cancelCall := context.WithCancel(runCtx)
	defer cancelCall()
	go watchChannelLifecycle(callCtx, starts, channelID, cancelCall)
	if err := client.Channel().Answer(channelKey); err != nil && !strings.Contains(strings.ToLower(err.Error()), "already") {
		return Result{}, fmt.Errorf("answer channel: %w", err)
	}

	bridgeID := "agent-call-" + channelID
	bridge, err := client.Bridge().Create(ari.NewKey(ari.BridgeKey, bridgeID), "mixing", bridgeID)
	if err != nil {
		return Result{}, fmt.Errorf("create media bridge: %w", err)
	}
	bridgeKey = ari.NewKey(ari.BridgeKey, bridge.ID())
	if err := client.Bridge().AddChannel(bridgeKey, channelID); err != nil {
		return Result{}, fmt.Errorf("add call channel to bridge: %w", err)
	}
	external, err := client.Channel().ExternalMedia(nil, ari.ExternalMediaOptions{
		App:            cfg.ARIApplication,
		ExternalHost:   fmt.Sprintf("%s:%d", cfg.MediaBind, cfg.MediaPort),
		Encapsulation:  "rtp",
		Transport:      "udp",
		ConnectionType: "client",
		Format:         cfg.MediaFormat,
		Direction:      "both",
	})
	if err != nil {
		return Result{}, fmt.Errorf("create external media channel: %w", err)
	}
	externalKey = ari.NewKey(ari.ChannelKey, external.ID())
	address, err := external.GetVariable("UNICASTRTP_LOCAL_ADDRESS")
	if err != nil {
		return Result{}, fmt.Errorf("read ExternalMedia RTP address: %w", err)
	}
	port, err := external.GetVariable("UNICASTRTP_LOCAL_PORT")
	if err != nil {
		return Result{}, fmt.Errorf("read ExternalMedia RTP port: %w", err)
	}
	if err := stream.SetPeer(net.JoinHostPort(strings.TrimSpace(address), strings.TrimSpace(port))); err != nil {
		return Result{}, fmt.Errorf("configure ExternalMedia RTP peer: %w", err)
	}
	if err := client.Bridge().AddChannel(bridgeKey, external.ID()); err != nil {
		return Result{}, fmt.Errorf("add external media to bridge: %w", err)
	}

	flowResult, flowErr := callflow.ExecuteWithCapture(callCtx, stream, cfg.Pipeline, cfg.Snapshot, cfg.OpeningPrompt, callflow.CaptureConfig{
		FirstSpeechTimeout: cfg.FirstSpeechTimeout,
		MaxDuration:        cfg.MaxTurnDuration,
		EndSilence:         cfg.EndSilence,
		VoiceThreshold:     cfg.VoiceThreshold,
		MaxTurns:           cfg.MaxTurns,
	})
	inboundRecordings, outboundRecordings, recordingErr := persistConversationRecordings(cfg.RecordingDirectory, channelID, flowResult)
	if recordingErr != nil {
		return Result{}, recordingErr
	}
	turn := flowResult.Turn
	result := Result{
		ChannelID:          channelID,
		Transcript:         turn.Transcript,
		Reply:              turn.Reply,
		InvalidCall:        turn.InvalidCall,
		InvalidReason:      turn.InvalidReason,
		Turns:              append([]ai.TurnResult(nil), flowResult.Turns...),
		InboundRecordings:  inboundRecordings,
		OutboundRecordings: outboundRecordings,
		RTP:                flowResult.RTP,
	}
	if len(inboundRecordings) > 0 {
		last := inboundRecordings[len(inboundRecordings)-1]
		result.InboundPath, result.InboundBytes, result.InboundSHA256 = last.Path, last.Bytes, last.SHA256
	}
	if len(outboundRecordings) > 0 {
		last := outboundRecordings[len(outboundRecordings)-1]
		result.OutboundPath, result.OutboundBytes, result.OutboundSHA256 = last.Path, last.Bytes, last.SHA256
	}
	if flowErr != nil {
		stats := stream.Stats()
		return result, fmt.Errorf("execute call flow: %w (rtp rx_packets=%d rx_bytes=%d tx_packets=%d tx_bytes=%d)", flowErr, stats.ReceivedPackets, stats.ReceivedBytes, stats.SentPackets, stats.SentBytes)
	}
	return result, nil
}

func watchChannelLifecycle(ctx context.Context, sub ari.Subscription, channelID string, cancel context.CancelFunc) {
	for {
		select {
		case <-ctx.Done():
			return
		case event, ok := <-sub.Events():
			if !ok || event == nil {
				return
			}
			matched := false
			for _, key := range event.Keys() {
				if key != nil && key.ID == channelID {
					matched = true
					break
				}
			}
			if !matched {
				continue
			}
			switch event.GetType() {
			case "ChannelHangupRequest", "ChannelDestroyed":
				cancel()
				return
			}
		}
	}
}

func waitForStasisStart(ctx context.Context, sub ari.Subscription, channelID string, timeout time.Duration) error {
	waitCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	var recent []string
	for {
		select {
		case <-waitCtx.Done():
			return fmt.Errorf("waiting for channel %s answer/StasisStart (recent_events=%s): %w", channelID, strings.Join(recent, ","), waitCtx.Err())
		case event, ok := <-sub.Events():
			if !ok {
				return errors.New("ARI event subscription closed")
			}
			if event == nil {
				continue
			}
			typ := event.GetType()
			recent = append(recent, typ)
			if len(recent) > 8 {
				recent = recent[len(recent)-8:]
			}
			matched := false
			for _, key := range event.Keys() {
				if key != nil && key.ID == channelID {
					matched = true
					break
				}
			}
			if !matched {
				continue
			}
			if typ == "StasisStart" {
				return nil
			}
			if typ == "ChannelStateChange" {
				if state, ok := event.(*ari.ChannelStateChange); ok && strings.EqualFold(state.Channel.State, "Up") {
					return nil
				}
			}
			if typ == "ChannelHangupRequest" {
				if hangup, ok := event.(*ari.ChannelHangupRequest); ok {
					return fmt.Errorf("channel %s ended before StasisStart: %s cause=%d soft=%t state=%s", channelID, typ, hangup.Cause, hangup.Soft, hangup.Channel.State)
				}
				return fmt.Errorf("channel %s ended before StasisStart: %s", channelID, typ)
			}
			if typ == "ChannelDestroyed" {
				return fmt.Errorf("channel %s ended before StasisStart: %s", channelID, typ)
			}
		}
	}
}

func persistConversationRecordings(directory, channelID string, flowResult callflow.Result) ([]RecordingFact, []RecordingFact, error) {
	if directory == "" || (len(flowResult.InboundTurns) == 0 && len(flowResult.OutboundTurns) == 0) {
		return nil, nil, nil
	}
	if err := os.MkdirAll(directory, 0o700); err != nil {
		return nil, nil, err
	}
	stamp := time.Now().UTC().Format("20060102T150405.000000000Z")
	name := strings.NewReplacer("/", "_", "\\", "_", ":", "_").Replace(channelID)
	inbound := make([]RecordingFact, 0, len(flowResult.InboundTurns))
	for index, pcm := range flowResult.InboundTurns {
		path := filepath.Join(directory, fmt.Sprintf("%s-%s-inbound-turn-%02d.wav", stamp, name, index+1))
		if err := writeWAV(path, pcm, 16000); err != nil {
			return inbound, nil, err
		}
		digest, size, err := fileDigestAndSize(path)
		if err != nil {
			return inbound, nil, fmt.Errorf("hash inbound recording: %w", err)
		}
		inbound = append(inbound, RecordingFact{Segment: fmt.Sprintf("inbound_turn_%02d", index+1), Path: path, SHA256: digest, Bytes: size, DurationMS: int64(len(pcm)) * 1000 / 32000})
	}
	outbound := make([]RecordingFact, 0, len(flowResult.OutboundTurns))
	for index, pcm := range flowResult.OutboundTurns {
		path := filepath.Join(directory, fmt.Sprintf("%s-%s-outbound-segment-%02d.wav", stamp, name, index))
		if err := writeWAV(path, pcm, 16000); err != nil {
			return inbound, outbound, err
		}
		digest, size, err := fileDigestAndSize(path)
		if err != nil {
			return inbound, outbound, fmt.Errorf("hash outbound recording: %w", err)
		}
		outbound = append(outbound, RecordingFact{Segment: fmt.Sprintf("outbound_segment_%02d", index), Path: path, SHA256: digest, Bytes: size, DurationMS: int64(len(pcm)) * 1000 / 32000})
	}
	return inbound, outbound, nil
}

func writeWAV(path string, pcm []byte, sampleRate int) error {
	if len(pcm)%2 != 0 {
		return errors.New("PCM16 has odd byte length")
	}
	dataSize := uint32(len(pcm))
	byteRate := uint32(sampleRate * 2)
	blockAlign := uint16(2)
	buf := make([]byte, 44+len(pcm))
	copy(buf[:4], "RIFF")
	binary.LittleEndian.PutUint32(buf[4:8], 36+dataSize)
	copy(buf[8:12], "WAVE")
	copy(buf[12:16], "fmt ")
	binary.LittleEndian.PutUint32(buf[16:20], 16)
	binary.LittleEndian.PutUint16(buf[20:22], 1)
	binary.LittleEndian.PutUint16(buf[22:24], 1)
	binary.LittleEndian.PutUint32(buf[24:28], uint32(sampleRate))
	binary.LittleEndian.PutUint32(buf[28:32], byteRate)
	binary.LittleEndian.PutUint16(buf[32:34], blockAlign)
	binary.LittleEndian.PutUint16(buf[34:36], 16)
	copy(buf[36:40], "data")
	binary.LittleEndian.PutUint32(buf[40:44], dataSize)
	copy(buf[44:], pcm)
	return os.WriteFile(path, buf, 0o600)
}

func fileDigestAndSize(path string) (string, int, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return "", 0, err
	}
	sum := sha256.Sum256(data)
	return hex.EncodeToString(sum[:]), len(data), nil
}