diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index 794ff53..a475fa4 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -12,6 +12,7 @@ import ( "github.com/hhs/camtalk/internal/config" "github.com/hhs/camtalk/internal/logger" + "github.com/hhs/camtalk/internal/session" "github.com/hhs/camtalk/internal/ws" ) @@ -33,6 +34,12 @@ func main() { "addr", cfg.Server.Addr(), ) + // 初始化 Session Manager（MVP 默认内存实现） + var sessionMgr session.Manager + // TODO: 当 Redis 配置非空时切换为 RedisManager + sessionMgr = session.NewMemoryManager(30*time.Minute, 20) + defer sessionMgr.(*session.MemoryManager).Stop() + // Gin 模式 if cfg.App.Env == "prod" { gin.SetMode(gin.ReleaseMode) @@ -44,11 +51,11 @@ func main() { // REST API api := r.Group("/api") { - api.GET("/health", healthHandler) + api.GET("/health", healthHandler(sessionMgr)) } // WebSocket - r.GET("/ws", ws.ServeWS) + r.GET("/ws", ws.ServeWS(sessionMgr)) // HTTP Server srv := &http.Server{ @@ -82,11 +89,13 @@ func main() { } // healthHandler 健康检查。 -func healthHandler(c *gin.Context) { - c.JSON(200, gin.H{ - "status": "ok", - "version": "0.1.0", - "uptime": time.Since(startTime).String(), - "active_sessions": 0, // TODO: 接入 Session Manager - }) +func healthHandler(sessionMgr session.Manager) gin.HandlerFunc { + return func(c *gin.Context) { + c.JSON(200, gin.H{ + "status": "ok", + "version": "0.1.0", + "uptime": time.Since(startTime).String(), + "active_sessions": sessionMgr.ActiveCount(), + }) + } } diff --git a/backend/go.mod b/backend/go.mod index 093d709..5ac21cb 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -1,6 +1,6 @@ module github.com/hhs/camtalk -go 1.23.0 +go 1.24 require ( github.com/gin-gonic/gin v1.10.0 @@ -13,6 +13,7 @@ require ( require ( github.com/bytedance/sonic v1.11.6 // indirect github.com/bytedance/sonic/loader v0.1.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.4 // indirect github.com/cloudwego/iasm v0.2.0 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect @@ -24,12 +25,13 @@ require ( github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/goccy/go-json v0.10.2 // indirect github.com/json-iterator/go v1.1.12 // indirect - github.com/klauspost/cpuid/v2 v2.2.7 // indirect + github.com/klauspost/cpuid/v2 v2.2.10 // indirect github.com/leodido/go-urn v1.4.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect + github.com/redis/go-redis/v9 v9.20.1 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect github.com/spf13/afero v1.15.0 // indirect @@ -38,12 +40,13 @@ require ( github.com/subosito/gotenv v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect + go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.10.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/arch v0.8.0 // indirect golang.org/x/crypto v0.23.0 // indirect golang.org/x/net v0.25.0 // indirect - golang.org/x/sys v0.29.0 // indirect + golang.org/x/sys v0.30.0 // indirect golang.org/x/text v0.28.0 // indirect google.golang.org/protobuf v1.34.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/backend/go.sum b/backend/go.sum index d3be072..eb42bfe 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -2,6 +2,8 @@ github.com/bytedance/sonic v1.11.6 h1:oUp34TzMlL+OY1OUWxHqsdkgC/Zfc85zGqw9siXjrc github.com/bytedance/sonic v1.11.6/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4= github.com/bytedance/sonic/loader v0.1.1 h1:c+e5Pt1k/cy5wMveRDyk2X4B9hF4g7an8N3zCYjJFNM= github.com/bytedance/sonic/loader v0.1.1/go.mod h1:ncP89zfokxS5LZrJxl5z0UJcsk4M4yY2JpfqGeCtNLU= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cloudwego/base64x v0.1.4 h1:jwCgWpFanWmN8xoIUHa2rtzmkd5J2plF/dnLS6Xd/0Y= github.com/cloudwego/base64x v0.1.4/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJgA0rcu/8w= github.com/cloudwego/iasm v0.2.0 h1:1KNIy1I1H9hNNFEEH3DVnI4UujN+1zjpuk6gwHLTssg= @@ -43,6 +45,8 @@ github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHm github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.2.7 h1:ZWSB3igEs+d0qvnxR/ZBzXVmxkgt8DdzP6m9pfuVLDM= github.com/klauspost/cpuid/v2 v2.2.7/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/knz/go-libedit v1.10.1/go.mod h1:MZTVkCWyz0oBc7JOWP3wNAzd002ZbM/5hgShxwh4x8M= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -61,6 +65,8 @@ github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.20.1 h1:sfCU6A8P3dXbKyWes02uxA2baehGux9dZHfEKtsTB1w= +github.com/redis/go-redis/v9 v9.20.1/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8= github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= @@ -91,6 +97,8 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE= github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= @@ -110,6 +118,8 @@ golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU= golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng= golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU= google.golang.org/protobuf v1.34.1 h1:9ddQBjfCyZPOHPUiPxpYESBLc+T8P3E+Vo4IbKZgFWg= diff --git a/backend/internal/models/models.go b/backend/internal/models/models.go index 2508195..73d017a 100644 --- a/backend/internal/models/models.go +++ b/backend/internal/models/models.go @@ -21,6 +21,26 @@ func DefaultConfig() SessionConfig { return SessionConfig{TTSEnabled: true, DetailLevel: "low", Language: "zh-CN"} } +// SessionConfigPatch 会话配置增量更新（指针字段表示"未传则不更新"）。 +type SessionConfigPatch struct { + TTSEnabled *bool `json:"tts_enabled,omitempty"` + DetailLevel *string `json:"detail_level,omitempty"` + Language *string `json:"language,omitempty"` +} + +// Apply 将 patch 中的非 nil 字段覆盖到 cfg。 +func (p SessionConfigPatch) Apply(cfg *SessionConfig) { + if p.TTSEnabled != nil { + cfg.TTSEnabled = *p.TTSEnabled + } + if p.DetailLevel != nil { + cfg.DetailLevel = *p.DetailLevel + } + if p.Language != nil { + cfg.Language = *p.Language + } +} + // Message 对话消息。 type Message struct { Role string `json:"role"` // "user" | "assistant" diff --git a/backend/internal/session/manager.go b/backend/internal/session/manager.go new file mode 100644 index 0000000..6e7f0eb --- /dev/null +++ b/backend/internal/session/manager.go @@ -0,0 +1,49 @@ +// Package session 提供会话生命周期管理能力。 +package session + +import ( + "context" + "errors" + + "github.com/hhs/camtalk/internal/models" +) + +// ErrSessionNotFound 会话不存在或已过期。 +var ErrSessionNotFound = errors.New("session not found") + +// Manager 会话管理器接口。 +// WebSocket Handler 通过此接口操作会话，不直接接触存储层。 +type Manager interface { + // Create 创建新会话，返回 session ID。 + Create(ctx context.Context, config models.SessionConfig) (string, error) + + // Get 获取会话（含 config）。不存在返回 ErrSessionNotFound。 + Get(ctx context.Context, sessionID string) (*models.Session, error) + + // UpdateConfig 更新会话配置（config 消息触发）。 + UpdateConfig(ctx context.Context, sessionID string, patch models.SessionConfigPatch) error + + // GetHistory 获取最近 N 轮对话历史（供 Orchestrator 构建 LLM 上下文）。 + GetHistory(ctx context.Context, sessionID string, limit int) ([]models.Message, error) + + // AppendMessage 追加一条对话消息，同时刷新 TTL。 + AppendMessage(ctx context.Context, sessionID string, msg models.Message) error + + // SetActiveRequest 标记当前正在处理的请求 ID（interrupt 用）。 + SetActiveRequest(ctx context.Context, sessionID string, requestID string) error + + // GetActiveRequestID 获取当前活跃请求 ID。 + GetActiveRequestID(ctx context.Context, sessionID string) (string, error) + + // ClearActiveRequest 清除活跃请求标记（请求完成或中断后）。 + ClearActiveRequest(ctx context.Context, sessionID string) error + + // Touch 刷新 TTL（心跳时调用）。 + Touch(ctx context.Context, sessionID string) error + + // Destroy 显式销毁会话（REST API DELETE 或连接断开清理）。 + Destroy(ctx context.Context, sessionID string) error + + // ActiveCount 返回当前活跃会话数（健康检查用）。 + ActiveCount() int +} diff --git a/backend/internal/session/memory.go b/backend/internal/session/memory.go new file mode 100644 index 0000000..edea8cb --- /dev/null +++ b/backend/internal/session/memory.go @@ -0,0 +1,275 @@ +package session + +import ( + "context" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/hhs/camtalk/internal/logger" + "github.com/hhs/camtalk/internal/models" +) + +const ( + defaultTTL = 30 * time.Minute + defaultHistorySize = 20 +) + +// sessionEntry 内部会话条目。 +type sessionEntry struct { + session models.Session + history []models.Message + activeReqID string + lastActive time.Time +} + +// MemoryManager 基于内存的 SessionManager 实现。 +// 适用于 MVP 和无 Redis 的开发环境。 +type MemoryManager struct { + mu sync.RWMutex + sessions map[string]*sessionEntry + ttl time.Duration + maxHistory int + stopCleaner chan struct{} +} + +// NewMemoryManager 创建内存版 SessionManager。 +// ttl 为会话过期时间，maxHistory 为对话历史上限（0 表示使用默认值 20）。 +func NewMemoryManager(ttl time.Duration, maxHistory int) *MemoryManager { + if ttl <= 0 { + ttl = defaultTTL + } + if maxHistory <= 0 { + maxHistory = defaultHistorySize + } + + m := &MemoryManager{ + sessions: make(map[string]*sessionEntry), + ttl: ttl, + maxHistory: maxHistory, + stopCleaner: make(chan struct{}), + } + + // 启动后台清理 goroutine，每分钟清除过期会话。 + go m.cleanLoop() + + return m +} + +// cleanLoop 后台定期清理过期会话。 +func (m *MemoryManager) cleanLoop() { + ticker := time.NewTicker(1 * time.Minute) + defer ticker.Stop() + for { + select { + case <-ticker.C: + m.cleanExpired() + case <-m.stopCleaner: + return + } + } +} + +// cleanExpired 清除所有过期会话。 +func (m *MemoryManager) cleanExpired() { + m.mu.Lock() + defer m.mu.Unlock() + + now := time.Now() + for id, entry := range m.sessions { + if now.Sub(entry.lastActive) > m.ttl { + delete(m.sessions, id) + logger.Log.Debugw("session expired (cleaner)", "session", id) + } + } +} + +// Stop 停止后台清理 goroutine。应用退出前调用。 +func (m *MemoryManager) Stop() { + close(m.stopCleaner) +} + +// isExpired 检查会话是否过期（调用方需持锁或在已知 entry 存在时调用）。 +func (m *MemoryManager) isExpired(entry *sessionEntry) bool { + return time.Since(entry.lastActive) > m.ttl +} + +// Create 创建新会话。 +func (m *MemoryManager) Create(_ context.Context, config models.SessionConfig) (string, error) { + m.mu.Lock() + defer m.mu.Unlock() + + id := uuid.New().String() + now := time.Now() + m.sessions[id] = &sessionEntry{ + session: models.Session{ + ID: id, + CreatedAt: now, + Config: config, + }, + history: make([]models.Message, 0), + lastActive: now, + } + + logger.Log.Debugw("session created", "session", id) + return id, nil +} + +// Get 获取会话。 +func (m *MemoryManager) Get(_ context.Context, sessionID string) (*models.Session, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return nil, ErrSessionNotFound + } + + sess := entry.session // 复制一份返回 + return &sess, nil +} + +// UpdateConfig 更新会话配置。 +func (m *MemoryManager) UpdateConfig(_ context.Context, sessionID string, patch models.SessionConfigPatch) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + patch.Apply(&entry.session.Config) + entry.lastActive = time.Now() + + logger.Log.Debugw("session config updated", "session", sessionID) + return nil +} + +// GetHistory 获取最近 N 轮对话历史。 +func (m *MemoryManager) GetHistory(_ context.Context, sessionID string, limit int) ([]models.Message, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return nil, ErrSessionNotFound + } + + if limit <= 0 || limit > len(entry.history) { + limit = len(entry.history) + } + + // 返回最近 limit 条的副本 + result := make([]models.Message, limit) + copy(result, entry.history[len(entry.history)-limit:]) + return result, nil +} + +// AppendMessage 追加一条对话消息，同时刷新 TTL。 +func (m *MemoryManager) AppendMessage(_ context.Context, sessionID string, msg models.Message) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + entry.history = append(entry.history, msg) + + // 超过上限时裁剪，保留最新的 maxHistory 条 + if len(entry.history) > m.maxHistory { + entry.history = entry.history[len(entry.history)-m.maxHistory:] + } + + entry.lastActive = time.Now() + return nil +} + +// SetActiveRequest 标记当前正在处理的请求 ID。 +func (m *MemoryManager) SetActiveRequest(_ context.Context, sessionID string, requestID string) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + entry.activeReqID = requestID + entry.lastActive = time.Now() + return nil +} + +// GetActiveRequestID 获取当前活跃请求 ID。 +func (m *MemoryManager) GetActiveRequestID(_ context.Context, sessionID string) (string, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return "", ErrSessionNotFound + } + + return entry.activeReqID, nil +} + +// ClearActiveRequest 清除活跃请求标记。 +func (m *MemoryManager) ClearActiveRequest(_ context.Context, sessionID string) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + entry.activeReqID = "" + entry.lastActive = time.Now() + return nil +} + +// Touch 刷新 TTL。 +func (m *MemoryManager) Touch(_ context.Context, sessionID string) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + entry.lastActive = time.Now() + return nil +} + +// Destroy 显式销毁会话。 +func (m *MemoryManager) Destroy(_ context.Context, sessionID string) error { + m.mu.Lock() + defer m.mu.Unlock() + + if _, ok := m.sessions[sessionID]; !ok { + return ErrSessionNotFound + } + + delete(m.sessions, sessionID) + logger.Log.Debugw("session destroyed", "session", sessionID) + return nil +} + +// ActiveCount 返回当前活跃会话数。 +func (m *MemoryManager) ActiveCount() int { + m.mu.RLock() + defer m.mu.RUnlock() + + now := time.Now() + count := 0 + for _, entry := range m.sessions { + if now.Sub(entry.lastActive) <= m.ttl { + count++ + } + } + return count +} diff --git a/backend/internal/session/memory_test.go b/backend/internal/session/memory_test.go new file mode 100644 index 0000000..acfa433 --- /dev/null +++ b/backend/internal/session/memory_test.go @@ -0,0 +1,286 @@ +package session + +import ( + "context" + "testing" + "time" + + "github.com/hhs/camtalk/internal/logger" + "github.com/hhs/camtalk/internal/models" +) + +func init() { + logger.Init("debug", "console") +} + +func TestCreateAndGet(t *testing.T) { + m := NewMemoryManager(30*time.Minute, 20) + defer m.Stop() + ctx := context.Background() + + config := models.DefaultConfig() + id, err := m.Create(ctx, config) + if err != nil { + t.Fatalf("Create: %v", err) + } + if id == "" { + t.Fatal("Create returned empty ID") + } + + sess, err := m.Get(ctx, id) + if err != nil { + t.Fatalf("Get: %v", err) + } + if sess.ID != id { + t.Errorf("ID = %q, want %q", sess.ID, id) + } + if sess.Config.Language != "zh-CN" { + t.Errorf("Language = %q, want %q", sess.Config.Language, "zh-CN") + } +} + +func TestGetNotFound(t *testing.T) { + m := NewMemoryManager(30*time.Minute, 20) + defer m.Stop() + ctx := context.Background() + + _, err := m.Get(ctx, "nonexistent") + if err != ErrSessionNotFound { + t.Errorf("Get nonexistent: err = %v, want ErrSessionNotFound", err) + } +} + +func TestExpire(t *testing.T) { + // 使用极短 TTL 测试过期 + m := NewMemoryManager(50*time.Millisecond, 20) + defer m.Stop() + ctx := context.Background() + + id, _ := m.Create(ctx, models.DefaultConfig()) + + // 未过期时应能获取 + _, err := m.Get(ctx, id) + if err != nil { + t.Fatalf("Get before expire: %v", err) + } + + // 等待过期 + time.Sleep(80 * time.Millisecond) + + _, err = m.Get(ctx, id) + if err != ErrSessionNotFound { + t.Errorf("Get after expire: err = %v, want ErrSessionNotFound", err) + } +} + +func TestDestroy(t *testing.T) { + m := NewMemoryManager(30*time.Minute, 20) + defer m.Stop() + ctx := context.Background() + + id, _ := m.Create(ctx, models.DefaultConfig()) + + if err := m.Destroy(ctx, id); err != nil { + t.Fatalf("Destroy: %v", err) + } + + _, err := m.Get(ctx, id) + if err != ErrSessionNotFound { + t.Errorf("Get after Destroy: err = %v, want ErrSessionNotFound", err) + } +} + +func TestDestroyNotFound(t *testing.T) { + m := NewMemoryManager(30*time.Minute, 20) + defer m.Stop() + ctx := context.Background() + + err := m.Destroy(ctx, "nonexistent") + if err != ErrSessionNotFound { + t.Errorf("Destroy nonexistent: err = %v, want ErrSessionNotFound", err) + } +} + +func TestAppendMessageAndGetHistory(t *testing.T) { + m := NewMemoryManager(30*time.Minute, 20) + defer m.Stop() + ctx := context.Background() + + id, _ := m.Create(ctx, models.DefaultConfig()) + + msgs := []models.Message{ + {Role: "user", Content: "你好"}, + {Role: "assistant", Content: "你好！有什么可以帮你的吗？"}, + {Role: "user", Content: "这是什么？"}, + {Role: "assistant", Content: "这是一朵花。"}, + } + + for _, msg := range msgs { + if err := m.AppendMessage(ctx, id, msg); err != nil { + t.Fatalf("AppendMessage: %v", err) + } + } + + history, err := m.GetHistory(ctx, id, 0) + if err != nil { + t.Fatalf("GetHistory: %v", err) + } + if len(history) != 4 { + t.Fatalf("GetHistory len = %d, want 4", len(history)) + } + if history[0].Content != "你好" { + t.Errorf("history[0] = %q, want %q", history[0].Content, "你好") + } +} + +func TestGetHistoryLimit(t *testing.T) { + m := NewMemoryManager(30*time.Minute, 20) + defer m.Stop() + ctx := context.Background() + + id, _ := m.Create(ctx, models.DefaultConfig()) + + for i := 0; i < 10; i++ { + m.AppendMessage(ctx, id, models.Message{Role: "user", Content: "msg"}) + } + + history, err := m.GetHistory(ctx, id, 3) + if err != nil { + t.Fatalf("GetHistory: %v", err) + } + if len(history) != 3 { + t.Fatalf("GetHistory limit=3: len = %d, want 3", len(history)) + } +} + +func TestHistoryLimit(t *testing.T) { + const maxHistory = 5 + m := NewMemoryManager(30*time.Minute, maxHistory) + defer m.Stop() + ctx := context.Background() + + id, _ := m.Create(ctx, models.DefaultConfig()) + + // 插入超过上限的消息 + for i := 0; i < 10; i++ { + m.AppendMessage(ctx, id, models.Message{Role: "user", Content: "msg"}) + } + + history, err := m.GetHistory(ctx, id, 0) + if err != nil { + t.Fatalf("GetHistory: %v", err) + } + if len(history) != maxHistory { + t.Fatalf("GetHistory after overflow: len = %d, want %d", len(history), maxHistory) + } +} + +func TestUpdateConfig(t *testing.T) { + m := NewMemoryManager(30*time.Minute, 20) + defer m.Stop() + ctx := context.Background() + + id, _ := m.Create(ctx, models.DefaultConfig()) + + ttsEnabled := false + detailLevel := "high" + patch := models.SessionConfigPatch{ + TTSEnabled: &ttsEnabled, + DetailLevel: &detailLevel, + } + + if err := m.UpdateConfig(ctx, id, patch); err != nil { + t.Fatalf("UpdateConfig: %v", err) + } + + sess, _ := m.Get(ctx, id) + if sess.Config.TTSEnabled != false { + t.Errorf("TTSEnabled = %v, want false", sess.Config.TTSEnabled) + } + if sess.Config.DetailLevel != "high" { + t.Errorf("DetailLevel = %q, want %q", sess.Config.DetailLevel, "high") + } + // Language 未传，应保持原值 + if sess.Config.Language != "zh-CN" { + t.Errorf("Language = %q, want %q", sess.Config.Language, "zh-CN") + } +} + +func TestActiveRequest(t *testing.T) { + m := NewMemoryManager(30*time.Minute, 20) + defer m.Stop() + ctx := context.Background() + + id, _ := m.Create(ctx, models.DefaultConfig()) + + // 初始应为空 + reqID, err := m.GetActiveRequestID(ctx, id) + if err != nil { + t.Fatalf("GetActiveRequestID: %v", err) + } + if reqID != "" { + t.Errorf("initial active request = %q, want empty", reqID) + } + + // 设置 + if err := m.SetActiveRequest(ctx, id, "req-123"); err != nil { + t.Fatalf("SetActiveRequest: %v", err) + } + reqID, _ = m.GetActiveRequestID(ctx, id) + if reqID != "req-123" { + t.Errorf("active request = %q, want %q", reqID, "req-123") + } + + // 清除 + if err := m.ClearActiveRequest(ctx, id); err != nil { + t.Fatalf("ClearActiveRequest: %v", err) + } + reqID, _ = m.GetActiveRequestID(ctx, id) + if reqID != "" { + t.Errorf("active request after clear = %q, want empty", reqID) + } +} + +func TestTouchRefreshesTTL(t *testing.T) { + m := NewMemoryManager(100*time.Millisecond, 20) + defer m.Stop() + ctx := context.Background() + + id, _ := m.Create(ctx, models.DefaultConfig()) + + // 50ms 后 Touch，应重置 TTL + time.Sleep(50 * time.Millisecond) + if err := m.Touch(ctx, id); err != nil { + t.Fatalf("Touch: %v", err) + } + + // 再等 70ms（距创建 120ms，但距 Touch 只有 70ms），不应过期 + time.Sleep(70 * time.Millisecond) + _, err := m.Get(ctx, id) + if err != nil { + t.Errorf("Get after Touch: %v, want nil (should not expire yet)", err) + } + + // 再等 50ms（距 Touch 120ms），应过期 + time.Sleep(50 * time.Millisecond) + _, err = m.Get(ctx, id) + if err != ErrSessionNotFound { + t.Errorf("Get after TTL: err = %v, want ErrSessionNotFound", err) + } +} + +func TestActiveCount(t *testing.T) { + m := NewMemoryManager(30*time.Minute, 20) + defer m.Stop() + ctx := context.Background() + + if m.ActiveCount() != 0 { + t.Errorf("initial ActiveCount = %d, want 0", m.ActiveCount()) + } + + m.Create(ctx, models.DefaultConfig()) + m.Create(ctx, models.DefaultConfig()) + if m.ActiveCount() != 2 { + t.Errorf("ActiveCount = %d, want 2", m.ActiveCount()) + } +} diff --git a/backend/internal/session/redis.go b/backend/internal/session/redis.go new file mode 100644 index 0000000..ae24f11 --- /dev/null +++ b/backend/internal/session/redis.go @@ -0,0 +1,324 @@ +package session + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "time" + + "github.com/google/uuid" + "github.com/redis/go-redis/v9" + + "github.com/hhs/camtalk/internal/logger" + "github.com/hhs/camtalk/internal/models" +) + +// RedisManager 基于 Redis 的 SessionManager 实现。 +// 数据结构： +// - session:{id}:meta → Hash（会话元数据） +// - session:{id}:history → List（对话历史） +type RedisManager struct { + rdb *redis.Client + ttl time.Duration + maxHistory int +} + +// NewRedisManager 创建 Redis 版 SessionManager。 +func NewRedisManager(rdb *redis.Client, ttl time.Duration, maxHistory int) *RedisManager { + if ttl <= 0 { + ttl = defaultTTL + } + if maxHistory <= 0 { + maxHistory = defaultHistorySize + } + return &RedisManager{rdb: rdb, ttl: ttl, maxHistory: maxHistory} +} + +func metaKey(id string) string { return fmt.Sprintf("session:%s:meta", id) } +func histKey(id string) string { return fmt.Sprintf("session:%s:history", id) } + +// Create 创建新会话。 +func (m *RedisManager) Create(ctx context.Context, config models.SessionConfig) (string, error) { + id := uuidNew() + now := time.Now().UTC() + + pipe := m.rdb.Pipeline() + + // 写入 meta Hash + pipe.HSet(ctx, metaKey(id), map[string]interface{}{ + "session_id": id, + "config.tts_enabled": strconv.FormatBool(config.TTSEnabled), + "config.detail_level": config.DetailLevel, + "config.language": config.Language, + "created_at": now.Format(time.RFC3339), + "last_active": now.Format(time.RFC3339), + "active_request_id": "", + }) + pipe.Expire(ctx, metaKey(id), m.ttl) + + // 初始化空 history List + pipe.RPush(ctx, histKey(id), placeholderHistoryMark) + pipe.Expire(ctx, histKey(id), m.ttl) + + if _, err := pipe.Exec(ctx); err != nil { + return "", fmt.Errorf("redis create session: %w", err) + } + + logger.Log.Debugw("redis session created", "session", id) + return id, nil +} + +// placeholderHistoryMark 占位符，避免 Redis 对空 key 的特殊行为。 +const placeholderHistoryMark = "__placeholder__" + +// Get 获取会话。 +func (m *RedisManager) Get(ctx context.Context, sessionID string) (*models.Session, error) { + vals, err := m.rdb.HGetAll(ctx, metaKey(sessionID)).Result() + if err != nil { + return nil, fmt.Errorf("redis get session: %w", err) + } + if len(vals) == 0 { + return nil, ErrSessionNotFound + } + + sess := &models.Session{ + ID: vals["session_id"], + } + sess.CreatedAt, _ = time.Parse(time.RFC3339, vals["created_at"]) + sess.Config.TTSEnabled, _ = strconv.ParseBool(vals["config.tts_enabled"]) + sess.Config.DetailLevel = vals["config.detail_level"] + sess.Config.Language = vals["config.language"] + + return sess, nil +} + +// UpdateConfig 更新会话配置。 +func (m *RedisManager) UpdateConfig(ctx context.Context, sessionID string, patch models.SessionConfigPatch) error { + // 先检查会话是否存在 + exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result() + if err != nil { + return fmt.Errorf("redis check session: %w", err) + } + if exists == 0 { + return ErrSessionNotFound + } + + fields := map[string]interface{}{ + "last_active": time.Now().UTC().Format(time.RFC3339), + } + if patch.TTSEnabled != nil { + fields["config.tts_enabled"] = strconv.FormatBool(*patch.TTSEnabled) + } + if patch.DetailLevel != nil { + fields["config.detail_level"] = *patch.DetailLevel + } + if patch.Language != nil { + fields["config.language"] = *patch.Language + } + + if err := m.rdb.HSet(ctx, metaKey(sessionID), fields).Err(); err != nil { + return fmt.Errorf("redis update config: %w", err) + } + + // 刷新 TTL + m.rdb.Expire(ctx, metaKey(sessionID), m.ttl) + logger.Log.Debugw("redis session config updated", "session", sessionID) + return nil +} + +// GetHistory 获取最近 N 轮对话历史。 +func (m *RedisManager) GetHistory(ctx context.Context, sessionID string, limit int) ([]models.Message, error) { + // 检查会话是否存在 + exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result() + if err != nil { + return nil, fmt.Errorf("redis check session: %w", err) + } + if exists == 0 { + return nil, ErrSessionNotFound + } + + if limit <= 0 { + limit = m.maxHistory + } + + // LRANGE 0 {limit-1}，最新在前（LPUSH），需要反转为时间顺序 + raws, err := m.rdb.LRange(ctx, histKey(sessionID), 0, int64(limit)).Result() + if err != nil { + return nil, fmt.Errorf("redis get history: %w", err) + } + + var msgs []models.Message + for _, raw := range raws { + if raw == placeholderHistoryMark { + continue + } + var msg models.Message + if err := json.Unmarshal([]byte(raw), &msg); err != nil { + logger.Log.Warnw("invalid history entry", "session", sessionID, "raw", raw) + continue + } + msgs = append(msgs, msg) + } + + // 反转为时间顺序（LPUSH 最新在前 → 需要最旧在前） + for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 { + msgs[i], msgs[j] = msgs[j], msgs[i] + } + + return msgs, nil +} + +// AppendMessage 追加一条对话消息，同时刷新 TTL。 +func (m *RedisManager) AppendMessage(ctx context.Context, sessionID string, msg models.Message) error { + // 检查会话是否存在 + exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result() + if err != nil { + return fmt.Errorf("redis check session: %w", err) + } + if exists == 0 { + return ErrSessionNotFound + } + + data, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("marshal message: %w", err) + } + + pipe := m.rdb.Pipeline() + // LPUSH 新消息到左头（最新在前） + pipe.LPush(ctx, histKey(sessionID), string(data)) + // LTRIM 保留最近 maxHistory 条（+1 是因为有占位符） + pipe.LTrim(ctx, histKey(sessionID), 0, int64(m.maxHistory)) + // 刷新 TTL + pipe.Expire(ctx, histKey(sessionID), m.ttl) + pipe.Expire(ctx, metaKey(sessionID), m.ttl) + // 更新 last_active + pipe.HSet(ctx, metaKey(sessionID), "last_active", time.Now().UTC().Format(time.RFC3339)) + + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("redis append message: %w", err) + } + + return nil +} + +// SetActiveRequest 标记当前正在处理的请求 ID。 +func (m *RedisManager) SetActiveRequest(ctx context.Context, sessionID string, requestID string) error { + exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result() + if err != nil { + return fmt.Errorf("redis check session: %w", err) + } + if exists == 0 { + return ErrSessionNotFound + } + + pipe := m.rdb.Pipeline() + pipe.HSet(ctx, metaKey(sessionID), "active_request_id", requestID) + pipe.HSet(ctx, metaKey(sessionID), "last_active", time.Now().UTC().Format(time.RFC3339)) + pipe.Expire(ctx, metaKey(sessionID), m.ttl) + + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("redis set active request: %w", err) + } + return nil +} + +// GetActiveRequestID 获取当前活跃请求 ID。 +func (m *RedisManager) GetActiveRequestID(ctx context.Context, sessionID string) (string, error) { + val, err := m.rdb.HGet(ctx, metaKey(sessionID), "active_request_id").Result() + if err == redis.Nil { + return "", ErrSessionNotFound + } + if err != nil { + return "", fmt.Errorf("redis get active request: %w", err) + } + return val, nil +} + +// ClearActiveRequest 清除活跃请求标记。 +func (m *RedisManager) ClearActiveRequest(ctx context.Context, sessionID string) error { + exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result() + if err != nil { + return fmt.Errorf("redis check session: %w", err) + } + if exists == 0 { + return ErrSessionNotFound + } + + pipe := m.rdb.Pipeline() + pipe.HSet(ctx, metaKey(sessionID), "active_request_id", "") + pipe.HSet(ctx, metaKey(sessionID), "last_active", time.Now().UTC().Format(time.RFC3339)) + pipe.Expire(ctx, metaKey(sessionID), m.ttl) + + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("redis clear active request: %w", err) + } + return nil +} + +// Touch 刷新 TTL。 +func (m *RedisManager) Touch(ctx context.Context, sessionID string) error { + exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result() + if err != nil { + return fmt.Errorf("redis check session: %w", err) + } + if exists == 0 { + return ErrSessionNotFound + } + + pipe := m.rdb.Pipeline() + pipe.Expire(ctx, metaKey(sessionID), m.ttl) + pipe.Expire(ctx, histKey(sessionID), m.ttl) + pipe.HSet(ctx, metaKey(sessionID), "last_active", time.Now().UTC().Format(time.RFC3339)) + + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("redis touch: %w", err) + } + return nil +} + +// Destroy 显式销毁会话。 +func (m *RedisManager) Destroy(ctx context.Context, sessionID string) error { + deleted, err := m.rdb.Del(ctx, metaKey(sessionID), histKey(sessionID)).Result() + if err != nil { + return fmt.Errorf("redis destroy session: %w", err) + } + if deleted == 0 { + return ErrSessionNotFound + } + + logger.Log.Debugw("redis session destroyed", "session", sessionID) + return nil +} + +// ActiveCount 返回当前活跃会话数。 +// Redis 实现通过 SCAN 遍历 meta key，适用于中等规模。 +// 大规模部署建议维护独立的活跃会话集合。 +func (m *RedisManager) ActiveCount() int { + ctx := context.Background() + count := 0 + var cursor uint64 + for { + keys, nextCursor, err := m.rdb.Scan(ctx, cursor, "session:*:meta", 100).Result() + if err != nil { + break + } + for _, key := range keys { + exists, _ := m.rdb.Exists(ctx, key).Result() + if exists > 0 { + count++ + } + } + cursor = nextCursor + if cursor == 0 { + break + } + } + return count +} + +// uuidNew 生成 UUID，便于测试时 mock。 +var uuidNew = func() string { + return uuid.New().String() +} diff --git a/backend/internal/ws/handler.go b/backend/internal/ws/handler.go index e27f4c0..83316b4 100644 --- a/backend/internal/ws/handler.go +++ b/backend/internal/ws/handler.go @@ -1,17 +1,19 @@ package ws import ( + "context" "encoding/json" "net/http" "sync" "time" "github.com/gin-gonic/gin" - "github.com/google/uuid" "github.com/gorilla/websocket" + "github.com/hhs/camtalk/internal/errors" "github.com/hhs/camtalk/internal/logger" "github.com/hhs/camtalk/internal/models" + "github.com/hhs/camtalk/internal/session" ) var upgrader = websocket.Upgrader{ @@ -25,14 +27,21 @@ type Client struct { mu sync.Mutex } -func (c *Client) sendJSON(v any) error { +// SendJSON 向客户端发送 JSON 消息（公开以便 errors 包调用）。 +func (c *Client) SendJSON(v any) error { c.mu.Lock() defer c.mu.Unlock() return c.conn.WriteJSON(v) } // ServeWS 处理 WebSocket 升级请求。 -func ServeWS(c *gin.Context) { +func ServeWS(sessionMgr session.Manager) gin.HandlerFunc { + return func(c *gin.Context) { + serveWS(c, sessionMgr) + } +} + +func serveWS(c *gin.Context, sessionMgr session.Manager) { conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) if err != nil { logger.Log.Errorw("websocket upgrade failed", "error", err) @@ -40,11 +49,17 @@ func ServeWS(c *gin.Context) { } defer conn.Close() - sessionID := uuid.New().String() + // 创建会话 + sessionID, err := sessionMgr.Create(context.Background(), models.DefaultConfig()) + if err != nil { + logger.Log.Errorw("create session failed", "error", err) + return + } + client := &Client{conn: conn, sessionID: sessionID} // 发送 connected 消息 - _ = client.sendJSON(models.WsConnected{ + _ = client.SendJSON(models.WsConnected{ Type: "connected", SessionID: sessionID, ServerVersion: "0.1.0", @@ -92,51 +107,67 @@ func ServeWS(c *gin.Context) { Type string `json:"type"` } if err := json.Unmarshal(message, &envelope); err != nil { - _ = client.sendJSON(models.WsError{ - Type: "error", - Code: "INVALID_MESSAGE", - Message: "invalid JSON", - }) + errors.SendWSError(client, errors.CodeInvalidMessage, "", err) continue } switch envelope.Type { case "ping": - _ = client.sendJSON(models.WsPong{Type: "pong"}) + _ = client.SendJSON(models.WsPong{Type: "pong"}) case "query": var msg models.WsQuery if err := json.Unmarshal(message, &msg); err != nil { - _ = client.sendJSON(models.WsError{ - Type: "error", - Code: "INVALID_MESSAGE", - Message: "invalid query message", - RequestID: msg.RequestID, - }) + errors.SendWSError(client, errors.CodeInvalidMessage, msg.RequestID, err) continue } logger.Log.Infow("query received", "session", sessionID, "request", msg.RequestID) - // TODO: 调用 AI 编排流程（STT → LLM → TTS） + + // 刷新会话 TTL + if err := sessionMgr.Touch(context.Background(), sessionID); err != nil { + logger.Log.Warnw("touch session failed", "session", sessionID, "error", err) + } + + // 标记活跃请求 + if err := sessionMgr.SetActiveRequest(context.Background(), sessionID, msg.RequestID); err != nil { + logger.Log.Warnw("set active request failed", "session", sessionID, "error", err) + } + + // 获取对话历史（供后续 Orchestrator 使用） + _, _ = sessionMgr.GetHistory(context.Background(), sessionID, 20) + + // TODO: 解码 audio Base64 → 启动 orchestrator.ProcessQuery goroutine case "config": var msg models.WsConfig if err := json.Unmarshal(message, &msg); err != nil { - _ = client.sendJSON(models.WsError{ - Type: "error", - Code: "INVALID_MESSAGE", - Message: "invalid config message", - }) + errors.SendWSError(client, errors.CodeInvalidMessage, "", err) continue } - logger.Log.Infow("config update", "session", sessionID) - // TODO: 更新会话配置 + + patch := models.SessionConfigPatch{ + TTSEnabled: msg.Payload.TTSEnabled, + DetailLevel: msg.Payload.DetailLevel, + Language: msg.Payload.Language, + } + if err := sessionMgr.UpdateConfig(context.Background(), sessionID, patch); err != nil { + errors.SendWSError(client, errors.CodeInternalError, "", err) + continue + } + logger.Log.Infow("config updated", "session", sessionID) case "interrupt": logger.Log.Infow("interrupt received", "session", sessionID) - // TODO: 中断当前 AI 响应 + + // 获取活跃请求 ID（实际 cancel 在 Phase 5 接入 orchestrator 后实现） + reqID, _ := sessionMgr.GetActiveRequestID(context.Background(), sessionID) + if reqID != "" { + _ = sessionMgr.ClearActiveRequest(context.Background(), sessionID) + // TODO: 取消对应 context cancel func + } default: - _ = client.sendJSON(models.WsError{ + _ = client.SendJSON(models.WsError{ Type: "error", Code: "INVALID_MESSAGE", Message: "unknown message type: " + envelope.Type, @@ -145,5 +176,7 @@ func ServeWS(c *gin.Context) { } close(done) + + // 断开连接时不销毁会话，让其自然过期（支持重连恢复） logger.Log.Infow("client disconnected", "session", sessionID) }