package eino

import (
	"context"
	"encoding/base64"
	"io"

	"github.com/cloudwego/eino/compose"
	"github.com/cloudwego/eino/schema"

	"github.com/hhs/camtalk/internal/ai/tts"
	"github.com/hhs/camtalk/internal/models"
	"github.com/hhs/camtalk/internal/trace"
)

// NewTTSLambda 创建 TTS Transform Lambda 节点。
// 输入: StreamReader[string]（句子流）→ 输出: StreamReader[struct{}]（结果流）
//
// 流式消费每个句子，调用 ttsService.SynthesizeStream() 合成，
// 逐 chunk 推送 tts_audio 到客户端。TTS 失败静默跳过。
func NewTTSLambda(ttsService tts.Service, ttsVoice string, ttsSpeed float64, ttsOutputFmt string, ttsSampleRate int) *compose.Lambda {
	return compose.TransformableLambda(func(ctx context.Context, input *schema.StreamReader[string]) (*schema.StreamReader[struct{}], error) {
		sr, sw := schema.Pipe[struct{}](8)

		go func() {
			defer sw.Close()
			defer input.Close()

			log := trace.FromContext(ctx)
			sender := senderFromCtx(ctx)
			requestID := requestIDFromCtx(ctx)

			if sender == nil || requestID == "" {
				// 消费并丢弃流
				for {
					_, err := input.Recv()
					if err != nil {
						return
					}
				}
			}

			// 收集句子，按批次合成 TTS
			var sentences []string
			for {
				sentence, err := input.Recv()
				if err != nil {
					if err == io.EOF {
						break
					}
					log.Errorw("TTS: stream recv error", "error", err)
					break
				}
				if sentence != "" {
					sentences = append(sentences, sentence)
				}
			}

			if len(sentences) == 0 {
				sw.Send(struct{}{}, nil)
				return
			}

			log.Infow("开始 TTS 合成", "sentence_count", len(sentences))

			// 将句子数组转为 channel
			sentenceCh := make(chan string, len(sentences))
			for _, s := range sentences {
				sentenceCh <- s
			}
			close(sentenceCh)

			// 调用 TTS 服务
			ttsStream, err := ttsService.SynthesizeStream(ctx, sentenceCh, tts.Options{
				Voice:      ttsVoice,
				Speed:      ttsSpeed,
				OutputFmt:  ttsOutputFmt,
				SampleRate: ttsSampleRate,
			})
			if err != nil {
				log.Errorw("TTS 合成启动失败（已跳过）", "error", err)
				sw.Send(struct{}{}, nil)
				return
			}

			// 消费 TTS 音频流，推送到客户端
			for chunk := range ttsStream {
				select {
				case <-ctx.Done():
					log.Debugw("tts stream interrupted")
					sw.Send(struct{}{}, ctx.Err())
					return
				default:
				}

				audioBase64 := base64.StdEncoding.EncodeToString(chunk.Audio)

				if err := sender.SendTTSAudio(models.WsTTSAudio{
					Type:      "tts_audio",
					RequestID: requestID,
					Audio:     audioBase64,
					MimeType:  "audio/mp3",
					IsLast:    chunk.IsLast,
					Final:     chunk.Final,
				}); err != nil {
					log.Errorw("发送 tts_audio 失败", "error", err)
				}
			}

			log.Infow("TTS 合成完成")
			sw.Send(struct{}{}, nil)
		}()

		return sr, nil
	})
}