sundynix-agentix/sundynix-gateway/internal/handler/voice.go

package handler

import (
	"context"
	"encoding/json"
	"log"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"

	"github.com/sundynix/sundynix-gateway/internal/voice"
)

// 语音交互 WebSocket 端点（JARVIS）。一条连接承载上行音频 + 下行转写 + 下行 TTS 音频，
// 协议见 voice/protocol.go。鉴权走 AuthFromHeaderOrQuery（EventSource/WS 带不了 Bearer 头，
// 用 ?token=）。本文件是会话外壳 + 客户端↔网关协议循环；火山 ASR/TTS 客户端在下一步接入。

var voiceUpgrader = websocket.Upgrader{
	ReadBufferSize:  4096,
	WriteBufferSize: 4096,
	// CheckOrigin 放行：鉴权已由 token 把关（跨源 WS 无法读响应，且我们不依赖 cookie）。
	CheckOrigin: func(*http.Request) bool { return true },
}

const voiceWriteWait = 10 * time.Second

// VoiceStream: GET /api/v1/voice/stream —— 升级为 WebSocket 语音会话。
func (h *Handler) VoiceStream(c *gin.Context) {
	uid := userID(c)
	if uid == "" {
		c.JSON(http.StatusUnauthorized, gin.H{"error": "需要登录"})
		return
	}
	cfg := h.loadVoiceConfig(c.Request.Context())
	if !cfg.ASREnabled() {
		c.JSON(http.StatusServiceUnavailable, gin.H{"error": "语音服务未配置（缺 API Key / ASR resource-id）"})
		return
	}

	conn, err := voiceUpgrader.Upgrade(c.Writer, c.Request, nil)
	if err != nil {
		log.Printf("[voice] 升级 WS 失败 uid=%s: %v", uid, err)
		return
	}
	defer conn.Close()

	// 租户/会话在升级时（还握着 gin.Context）一并抓取，供 WS 读循环里提交任务复用共用关卡。
	sess := &voiceSession{
		conn: conn, uid: uid, cfg: cfg, h: h,
		tenantID: tenantID(c), sessionID: sessionID(c),
	}
	sess.send(voice.ServerMsg{Type: voice.ServerReady})
	sess.run()
	sess.stopASR() // 连接结束，收掉在跑的识别会话
}

// voiceSession 是一次语音会话的外壳：持 WS 连接，跑协议循环。
// 上行 = 音频→ASR→转写→提交任务；下行（token流→攒句→TTS→音频）将在 TTS 步接上。
type voiceSession struct {
	conn      *websocket.Conn
	h         *Handler // 复用 preflightCore/launchCore 提交任务
	uid       string
	tenantID  string // 升级时抓取（读循环里无 gin.Context）
	sessionID string
	cfg       voice.Config

	writeMu   sync.Mutex // gorilla WS 不允许并发写：读循环与 ASR 结果 goroutine 都会 send，须串行化
	asr       *voice.ASRSession
	asrCancel context.CancelFunc

	pendingGraph string // 客户端 start 时带的画布编排图（语音触发既有编排），空则按转写现组
	lastFinal    string // 最近一次已提交的最终转写，去重连发的重复 final
}

// send 下发一条控制/事件消息（文本帧，JSON）。并发安全。
func (s *voiceSession) send(m voice.ServerMsg) {
	b, _ := json.Marshal(m)
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	_ = s.conn.SetWriteDeadline(time.Now().Add(voiceWriteWait))
	if err := s.conn.WriteMessage(websocket.TextMessage, b); err != nil {
		log.Printf("[voice] 写控制消息失败 uid=%s: %v", s.uid, err)
	}
}

// sendAudio 下发一帧 TTS 音频（二进制帧）。并发安全。
func (s *voiceSession) sendAudio(pcm []byte) {
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	_ = s.conn.SetWriteDeadline(time.Now().Add(voiceWriteWait))
	if err := s.conn.WriteMessage(websocket.BinaryMessage, pcm); err != nil {
		log.Printf("[voice] 写音频失败 uid=%s: %v", s.uid, err)
	}
}

// run 是协议读循环：二进制帧=上行音频，文本帧=控制消息。
func (s *voiceSession) run() {
	for {
		mt, data, err := s.conn.ReadMessage()
		if err != nil {
			return // 客户端断开 / 读错误
		}
		switch mt {
		case websocket.BinaryMessage:
			s.onAudio(data)
		case websocket.TextMessage:
			var m voice.ClientMsg
			if json.Unmarshal(data, &m) != nil {
				continue
			}
			if s.onControl(m) {
				return // bye
			}
		}
	}
}

// onAudio 收到一帧上行音频 → 喂火山 ASR。
func (s *voiceSession) onAudio(pcm []byte) {
	if s.asr == nil {
		s.startASR() // 客户端没显式 start 就直接说话时，惰性开一路识别
	}
	if s.asr != nil {
		if err := s.asr.PushAudio(pcm); err != nil {
			log.Printf("[voice] 喂 ASR 音频失败 uid=%s: %v", s.uid, err)
		}
	}
}

// onControl 处理客户端控制消息，返回 true 表示会话应结束。
func (s *voiceSession) onControl(m voice.ClientMsg) (done bool) {
	switch m.Type {
	case voice.ClientBye:
		return true
	case voice.ClientStart:
		s.pendingGraph = m.Graph // 客户端画布图（可空）：本轮若有转写则语音触发它跑
		s.lastFinal = ""
		s.stopASR()
		s.startASR() // 新一轮：重开识别
	case voice.ClientEnd:
		if s.asr != nil {
			_ = s.asr.Finish() // 告知火山本轮说完，等最终转写（结果流里带 Final=true→提交任务）
		}
	case voice.ClientBargeIn:
		// 打断：停当前 TTS 播放（TTS 接入后处理）。
	}
	return false
}

// startASR 开一路火山流式识别，并起 goroutine 把转写实时回推客户端。
func (s *voiceSession) startASR() {
	ctx, cancel := context.WithCancel(context.Background())
	asr, err := voice.StartASR(ctx, s.cfg, s.uid)
	if err != nil {
		cancel()
		log.Printf("[voice] 启动 ASR 失败 uid=%s: %v", s.uid, err)
		s.send(voice.ServerMsg{Type: voice.ServerError, Msg: "语音识别启动失败"})
		return
	}
	s.asr = asr
	s.asrCancel = cancel
	go func() {
		for r := range asr.Results() {
			if r.Err != nil {
				return // 识别流结束/出错
			}
			s.send(voice.ServerMsg{Type: voice.ServerTranscript, Text: r.Text, Final: r.Final})
			if r.Final {
				s.onFinalTranscript(r.Text) // 最终转写 → 提交任务
			} else {
				s.lastFinal = "" // 新的部分结果=新一轮开口，放行下一次 final 提交
			}
		}
	}()
}

// onFinalTranscript 拿到一段最终转写就提交一次任务。去重连发的重复 final（同一句 SAUC 可能回多条）。
// 只在结果 goroutine 里调，lastFinal 无需加锁。
func (s *voiceSession) onFinalTranscript(text string) {
	txt := strings.TrimSpace(text)
	if txt == "" || txt == s.lastFinal {
		return
	}
	s.lastFinal = txt
	taskID, err := s.submitVoiceTask(txt, s.pendingGraph)
	if err != nil {
		s.send(voice.ServerMsg{Type: voice.ServerError, Msg: "任务提交失败：" + err.Error()})
		return
	}
	s.pendingGraph = "" // 画布图一次性消费，避免后续转写重复触发同图
	s.send(voice.ServerMsg{Type: voice.ServerTask, TaskID: taskID})
	// 下行（订阅 token 流→攒句→TTS→音频帧）在 TTS 步接上；此处先只回 task_id 供客户端切运行视图。
}

// stopASR 收掉当前识别会话（幂等）。
func (s *voiceSession) stopASR() {
	if s.asr != nil {
		s.asr.Close()
		s.asr = nil
	}
	if s.asrCancel != nil {
		s.asrCancel()
		s.asrCancel = nil
	}
}