From 6487a8ecab5d73658427cb4407314e68a44f1ec1 Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sun, 14 Jun 2026 17:35:20 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=89=A9=E5=B1=95=20Session=20Manager?= =?UTF-8?q?=20=E6=8E=A5=E5=8F=A3=EF=BC=8C=E6=96=B0=E5=A2=9E=20ListByUser?= =?UTF-8?q?=E3=80=81UpdateTitle=20=E6=96=B9=E6=B3=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Manager.Create 签名新增 userID 参数 - 新增 ConversationSummary 类型和 ListByUser 分页查询 - 新增 UpdateTitle 方法 - MemoryManager 实现：ListByUser 遍历+过滤+排序，UpdateTitle，自动标题生成 - RedisManager 实现：user:{id}:sessions 索引，ListByUser 通过 SMEMBERS 查询 - AppendMessage 自动更新标题（首条 user 消息时，取前 20 字符） - 更新 ws handler、api/session.go、orchestrator mock 的 Create 调用 --- backend/internal/api/session.go | 2 +- .../internal/orchestrator/pipeline_test.go | 15 +- backend/internal/session/manager.go | 20 +- backend/internal/session/memory.go | 98 +++++++++- backend/internal/session/memory_test.go | 22 +-- backend/internal/session/redis.go | 172 ++++++++++++++++-- backend/internal/ws/handler.go | 2 +- 7 files changed, 293 insertions(+), 38 deletions(-) diff --git a/backend/internal/api/session.go b/backend/internal/api/session.go index 135758b..78bae2d 100644 --- a/backend/internal/api/session.go +++ b/backend/internal/api/session.go @@ -36,7 +36,7 @@ func (h *SessionHandler) CreateSession(c *gin.Context) { cfg = *req.Config } - sessionID, err := h.sessionMgr.Create(c.Request.Context(), cfg) + sessionID, err := h.sessionMgr.Create(c.Request.Context(), "", cfg) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{ "code": "INTERNAL_ERROR", diff --git a/backend/internal/orchestrator/pipeline_test.go b/backend/internal/orchestrator/pipeline_test.go index ad9b847..9000e96 100644 --- a/backend/internal/orchestrator/pipeline_test.go +++ b/backend/internal/orchestrator/pipeline_test.go @@ -16,6 +16,7 @@ import ( "github.com/hhs/camtalk/internal/config" "github.com/hhs/camtalk/internal/logger" "github.com/hhs/camtalk/internal/models" + "github.com/hhs/camtalk/internal/session" ) func init() { @@ -63,11 +64,21 @@ type MockSessionManager struct { mock.Mock } -func (m *MockSessionManager) Create(ctx context.Context, config models.SessionConfig) (string, error) { - args := m.Called(ctx, config) +func (m *MockSessionManager) Create(ctx context.Context, userID string, config models.SessionConfig) (string, error) { + args := m.Called(ctx, userID, config) return args.String(0), args.Error(1) } +func (m *MockSessionManager) UpdateTitle(ctx context.Context, sessionID string, title string) error { + args := m.Called(ctx, sessionID, title) + return args.Error(0) +} + +func (m *MockSessionManager) ListByUser(ctx context.Context, userID string, page, size int) ([]session.ConversationSummary, int, error) { + args := m.Called(ctx, userID, page, size) + return args.Get(0).([]session.ConversationSummary), args.Int(1), args.Error(2) +} + func (m *MockSessionManager) Get(ctx context.Context, sessionID string) (*models.Session, error) { args := m.Called(ctx, sessionID) if args.Get(0) == nil { diff --git a/backend/internal/session/manager.go b/backend/internal/session/manager.go index 6e7f0eb..6a2527e 100644 --- a/backend/internal/session/manager.go +++ b/backend/internal/session/manager.go @@ -4,6 +4,7 @@ package session import ( "context" "errors" + "time" "github.com/hhs/camtalk/internal/models" ) @@ -11,11 +12,20 @@ import ( // ErrSessionNotFound 会话不存在或已过期。 var ErrSessionNotFound = errors.New("session not found") +// ConversationSummary 对话摘要（列表展示用）。 +type ConversationSummary struct { + ID string `json:"id"` + Title string `json:"title"` + LastMessage string `json:"last_message"` + MessageCount int `json:"message_count"` + UpdatedAt time.Time `json:"updated_at"` +} + // Manager 会话管理器接口。 // WebSocket Handler 通过此接口操作会话，不直接接触存储层。 type Manager interface { - // Create 创建新会话，返回 session ID。 - Create(ctx context.Context, config models.SessionConfig) (string, error) + // Create 创建新会话，返回 session ID。userID 为空表示匿名会话。 + Create(ctx context.Context, userID string, config models.SessionConfig) (string, error) // Get 获取会话（含 config）。不存在返回 ErrSessionNotFound。 Get(ctx context.Context, sessionID string) (*models.Session, error) @@ -23,6 +33,12 @@ type Manager interface { // UpdateConfig 更新会话配置（config 消息触发）。 UpdateConfig(ctx context.Context, sessionID string, patch models.SessionConfigPatch) error + // UpdateTitle 更新会话标题。 + UpdateTitle(ctx context.Context, sessionID string, title string) error + + // ListByUser 获取用户的对话列表（分页，按 UpdatedAt 降序）。 + ListByUser(ctx context.Context, userID string, page, size int) ([]ConversationSummary, int, error) + // GetHistory 获取最近 N 轮对话历史（供 Orchestrator 构建 LLM 上下文）。 GetHistory(ctx context.Context, sessionID string, limit int) ([]models.Message, error) diff --git a/backend/internal/session/memory.go b/backend/internal/session/memory.go index edea8cb..2df0be3 100644 --- a/backend/internal/session/memory.go +++ b/backend/internal/session/memory.go @@ -2,6 +2,7 @@ package session import ( "context" + "sort" "sync" "time" @@ -95,8 +96,8 @@ 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) { +// Create 创建新会话。userID 为空表示匿名会话。 +func (m *MemoryManager) Create(_ context.Context, userID string, config models.SessionConfig) (string, error) { m.mu.Lock() defer m.mu.Unlock() @@ -105,14 +106,17 @@ func (m *MemoryManager) Create(_ context.Context, config models.SessionConfig) ( m.sessions[id] = &sessionEntry{ session: models.Session{ ID: id, + UserID: userID, + Title: models.DefaultSessionTitle, CreatedAt: now, + UpdatedAt: now, Config: config, }, history: make([]models.Message, 0), lastActive: now, } - logger.Log.Debugw("session created", "session", id) + logger.Log.Debugw("session created", "session", id, "user_id", userID) return id, nil } @@ -147,6 +151,76 @@ func (m *MemoryManager) UpdateConfig(_ context.Context, sessionID string, patch return nil } +// UpdateTitle 更新会话标题。 +func (m *MemoryManager) UpdateTitle(_ context.Context, sessionID string, title string) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + entry.session.Title = title + entry.session.UpdatedAt = time.Now() + entry.lastActive = time.Now() + + logger.Log.Debugw("session title updated", "session", sessionID, "title", title) + return nil +} + +// ListByUser 获取用户的对话列表（分页，按 UpdatedAt 降序）。 +func (m *MemoryManager) ListByUser(_ context.Context, userID string, page, size int) ([]ConversationSummary, int, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + // 收集该用户的所有 session + var list []ConversationSummary + for _, entry := range m.sessions { + if entry.session.UserID != userID { + continue + } + if m.isExpired(entry) { + continue + } + summary := ConversationSummary{ + ID: entry.session.ID, + Title: entry.session.Title, + MessageCount: len(entry.history), + UpdatedAt: entry.lastActive, + } + if len(entry.history) > 0 { + summary.LastMessage = entry.history[len(entry.history)-1].Content + } + list = append(list, summary) + } + + // 按 UpdatedAt 降序排序 + sort.Slice(list, func(i, j int) bool { + return list[i].UpdatedAt.After(list[j].UpdatedAt) + }) + + total := len(list) + + // 分页 + if page <= 0 { + page = 1 + } + if size <= 0 { + size = 20 + } + start := (page - 1) * size + if start >= total { + return []ConversationSummary{}, total, nil + } + end := start + size + if end > total { + end = total + } + + return list[start:end], total, nil +} + // GetHistory 获取最近 N 轮对话历史。 func (m *MemoryManager) GetHistory(_ context.Context, sessionID string, limit int) ([]models.Message, error) { m.mu.RLock() @@ -179,15 +253,31 @@ func (m *MemoryManager) AppendMessage(_ context.Context, sessionID string, msg m entry.history = append(entry.history, msg) + // 自动更新标题：首条 user 消息时，如果标题为默认值，自动更新为消息前 20 字符 + if msg.Role == "user" && entry.session.Title == models.DefaultSessionTitle { + entry.session.Title = generateTitle(msg.Content) + } + // 超过上限时裁剪，保留最新的 maxHistory 条 if len(entry.history) > m.maxHistory { entry.history = entry.history[len(entry.history)-m.maxHistory:] } - entry.lastActive = time.Now() + now := time.Now() + entry.lastActive = now + entry.session.UpdatedAt = now return nil } +// generateTitle 从首条消息生成对话标题（取前 20 个字符）。 +func generateTitle(firstMessage string) string { + runes := []rune(firstMessage) + if len(runes) > 20 { + return string(runes[:20]) + "…" + } + return firstMessage +} + // SetActiveRequest 标记当前正在处理的请求 ID。 func (m *MemoryManager) SetActiveRequest(_ context.Context, sessionID string, requestID string) error { m.mu.Lock() diff --git a/backend/internal/session/memory_test.go b/backend/internal/session/memory_test.go index acfa433..eb37b73 100644 --- a/backend/internal/session/memory_test.go +++ b/backend/internal/session/memory_test.go @@ -19,7 +19,7 @@ func TestCreateAndGet(t *testing.T) { ctx := context.Background() config := models.DefaultConfig() - id, err := m.Create(ctx, config) + id, err := m.Create(ctx, "", config) if err != nil { t.Fatalf("Create: %v", err) } @@ -56,7 +56,7 @@ func TestExpire(t *testing.T) { defer m.Stop() ctx := context.Background() - id, _ := m.Create(ctx, models.DefaultConfig()) + id, _ := m.Create(ctx, "", models.DefaultConfig()) // 未过期时应能获取 _, err := m.Get(ctx, id) @@ -78,7 +78,7 @@ func TestDestroy(t *testing.T) { defer m.Stop() ctx := context.Background() - id, _ := m.Create(ctx, models.DefaultConfig()) + id, _ := m.Create(ctx, "", models.DefaultConfig()) if err := m.Destroy(ctx, id); err != nil { t.Fatalf("Destroy: %v", err) @@ -106,7 +106,7 @@ func TestAppendMessageAndGetHistory(t *testing.T) { defer m.Stop() ctx := context.Background() - id, _ := m.Create(ctx, models.DefaultConfig()) + id, _ := m.Create(ctx, "", models.DefaultConfig()) msgs := []models.Message{ {Role: "user", Content: "你好"}, @@ -138,7 +138,7 @@ func TestGetHistoryLimit(t *testing.T) { defer m.Stop() ctx := context.Background() - id, _ := m.Create(ctx, models.DefaultConfig()) + id, _ := m.Create(ctx, "", models.DefaultConfig()) for i := 0; i < 10; i++ { m.AppendMessage(ctx, id, models.Message{Role: "user", Content: "msg"}) @@ -159,7 +159,7 @@ func TestHistoryLimit(t *testing.T) { defer m.Stop() ctx := context.Background() - id, _ := m.Create(ctx, models.DefaultConfig()) + id, _ := m.Create(ctx, "", models.DefaultConfig()) // 插入超过上限的消息 for i := 0; i < 10; i++ { @@ -180,7 +180,7 @@ func TestUpdateConfig(t *testing.T) { defer m.Stop() ctx := context.Background() - id, _ := m.Create(ctx, models.DefaultConfig()) + id, _ := m.Create(ctx, "", models.DefaultConfig()) ttsEnabled := false detailLevel := "high" @@ -211,7 +211,7 @@ func TestActiveRequest(t *testing.T) { defer m.Stop() ctx := context.Background() - id, _ := m.Create(ctx, models.DefaultConfig()) + id, _ := m.Create(ctx, "", models.DefaultConfig()) // 初始应为空 reqID, err := m.GetActiveRequestID(ctx, id) @@ -246,7 +246,7 @@ func TestTouchRefreshesTTL(t *testing.T) { defer m.Stop() ctx := context.Background() - id, _ := m.Create(ctx, models.DefaultConfig()) + id, _ := m.Create(ctx, "", models.DefaultConfig()) // 50ms 后 Touch，应重置 TTL time.Sleep(50 * time.Millisecond) @@ -278,8 +278,8 @@ func TestActiveCount(t *testing.T) { t.Errorf("initial ActiveCount = %d, want 0", m.ActiveCount()) } - m.Create(ctx, models.DefaultConfig()) - m.Create(ctx, models.DefaultConfig()) + 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 index ae24f11..ec1dc6b 100644 --- a/backend/internal/session/redis.go +++ b/backend/internal/session/redis.go @@ -18,6 +18,7 @@ import ( // 数据结构： // - session:{id}:meta → Hash（会话元数据） // - session:{id}:history → List（对话历史） +// - user:{id}:sessions → Set（用户会话索引） type RedisManager struct { rdb *redis.Client ttl time.Duration @@ -35,37 +36,48 @@ func NewRedisManager(rdb *redis.Client, ttl time.Duration, maxHistory int) *Redi 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) } +func metaKey(id string) string { return fmt.Sprintf("session:%s:meta", id) } +func histKey(id string) string { return fmt.Sprintf("session:%s:history", id) } +func userSessKey(id string) string { return fmt.Sprintf("user:%s:sessions", id) } -// Create 创建新会话。 -func (m *RedisManager) Create(ctx context.Context, config models.SessionConfig) (string, error) { +// Create 创建新会话。userID 为空表示匿名会话。 +func (m *RedisManager) Create(ctx context.Context, userID string, 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), + meta := map[string]interface{}{ + "session_id": id, + "user_id": userID, + "title": models.DefaultSessionTitle, + "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": "", - }) + "config.language": config.Language, + "created_at": now.Format(time.RFC3339), + "updated_at": now.Format(time.RFC3339), + "last_active": now.Format(time.RFC3339), + "active_request_id": "", + } + pipe.HSet(ctx, metaKey(id), meta) pipe.Expire(ctx, metaKey(id), m.ttl) // 初始化空 history List pipe.RPush(ctx, histKey(id), placeholderHistoryMark) pipe.Expire(ctx, histKey(id), m.ttl) + // 如果有 userID，添加到用户会话索引 + if userID != "" { + pipe.SAdd(ctx, userSessKey(userID), id) + pipe.Expire(ctx, userSessKey(userID), 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) + logger.Log.Debugw("redis session created", "session", id, "user_id", userID) return id, nil } @@ -83,9 +95,12 @@ func (m *RedisManager) Get(ctx context.Context, sessionID string) (*models.Sessi } sess := &models.Session{ - ID: vals["session_id"], + ID: vals["session_id"], + UserID: vals["user_id"], + Title: vals["title"], } sess.CreatedAt, _ = time.Parse(time.RFC3339, vals["created_at"]) + sess.UpdatedAt, _ = time.Parse(time.RFC3339, vals["updated_at"]) sess.Config.TTSEnabled, _ = strconv.ParseBool(vals["config.tts_enabled"]) sess.Config.DetailLevel = vals["config.detail_level"] sess.Config.Language = vals["config.language"] @@ -104,8 +119,10 @@ func (m *RedisManager) UpdateConfig(ctx context.Context, sessionID string, patch return ErrSessionNotFound } + now := time.Now().UTC().Format(time.RFC3339) fields := map[string]interface{}{ - "last_active": time.Now().UTC().Format(time.RFC3339), + "last_active": now, + "updated_at": now, } if patch.TTSEnabled != nil { fields["config.tts_enabled"] = strconv.FormatBool(*patch.TTSEnabled) @@ -127,6 +144,109 @@ func (m *RedisManager) UpdateConfig(ctx context.Context, sessionID string, patch return nil } +// UpdateTitle 更新会话标题。 +func (m *RedisManager) UpdateTitle(ctx context.Context, sessionID string, title 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 + } + + now := time.Now().UTC().Format(time.RFC3339) + if err := m.rdb.HSet(ctx, metaKey(sessionID), "title", title, "updated_at", now, "last_active", now).Err(); err != nil { + return fmt.Errorf("redis update title: %w", err) + } + + m.rdb.Expire(ctx, metaKey(sessionID), m.ttl) + logger.Log.Debugw("redis session title updated", "session", sessionID, "title", title) + return nil +} + +// ListByUser 获取用户的对话列表（分页，按 UpdatedAt 降序）。 +func (m *RedisManager) ListByUser(ctx context.Context, userID string, page, size int) ([]ConversationSummary, int, error) { + if page <= 0 { + page = 1 + } + if size <= 0 { + size = 20 + } + + // 从用户会话索引获取所有 session ID + sessionIDs, err := m.rdb.SMembers(ctx, userSessKey(userID)).Result() + if err != nil { + return nil, 0, fmt.Errorf("redis list user sessions: %w", err) + } + + // 收集有效的会话摘要 + var list []ConversationSummary + for _, sid := range sessionIDs { + vals, err := m.rdb.HGetAll(ctx, metaKey(sid)).Result() + if err != nil || len(vals) == 0 { + continue + } + + updatedAt, _ := time.Parse(time.RFC3339, vals["updated_at"]) + lastActive, _ := time.Parse(time.RFC3339, vals["last_active"]) + + // 检查是否过期 + if time.Since(lastActive) > m.ttl { + continue + } + + // 获取最后一条消息 + lastMsg := "" + msgCount := 0 + raws, err := m.rdb.LRange(ctx, histKey(sid), 0, 0).Result() + if err == nil && len(raws) > 0 && raws[0] != placeholderHistoryMark { + var msg models.Message + if json.Unmarshal([]byte(raws[0]), &msg) == nil { + lastMsg = msg.Content + } + } + // 获取消息总数（减去占位符） + totalLen, err := m.rdb.LLen(ctx, histKey(sid)).Result() + if err == nil { + msgCount = int(totalLen) + if msgCount > 0 { + msgCount-- // 减去占位符 + } + } + + list = append(list, ConversationSummary{ + ID: vals["session_id"], + Title: vals["title"], + LastMessage: lastMsg, + MessageCount: msgCount, + UpdatedAt: updatedAt, + }) + } + + // 按 UpdatedAt 降序排序 + for i := 0; i < len(list); i++ { + for j := i + 1; j < len(list); j++ { + if list[j].UpdatedAt.After(list[i].UpdatedAt) { + list[i], list[j] = list[j], list[i] + } + } + } + + total := len(list) + + // 分页 + start := (page - 1) * size + if start >= total { + return []ConversationSummary{}, total, nil + } + end := start + size + if end > total { + end = total + } + + return list[start:end], total, nil +} + // GetHistory 获取最近 N 轮对话历史。 func (m *RedisManager) GetHistory(ctx context.Context, sessionID string, limit int) ([]models.Message, error) { // 检查会话是否存在 @@ -193,8 +313,18 @@ func (m *RedisManager) AppendMessage(ctx context.Context, sessionID string, msg // 刷新 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)) + + now := time.Now().UTC().Format(time.RFC3339) + // 更新 last_active 和 updated_at + pipe.HSet(ctx, metaKey(sessionID), "last_active", now, "updated_at", now) + + // 自动更新标题：首条 user 消息时，如果标题为默认值 + if msg.Role == "user" { + title, _ := m.rdb.HGet(ctx, metaKey(sessionID), "title").Result() + if title == models.DefaultSessionTitle { + pipe.HSet(ctx, metaKey(sessionID), "title", generateTitle(msg.Content)) + } + } if _, err := pipe.Exec(ctx); err != nil { return fmt.Errorf("redis append message: %w", err) @@ -280,6 +410,9 @@ func (m *RedisManager) Touch(ctx context.Context, sessionID string) error { // Destroy 显式销毁会话。 func (m *RedisManager) Destroy(ctx context.Context, sessionID string) error { + // 先获取 user_id 以便清理索引 + userID, _ := m.rdb.HGet(ctx, metaKey(sessionID), "user_id").Result() + deleted, err := m.rdb.Del(ctx, metaKey(sessionID), histKey(sessionID)).Result() if err != nil { return fmt.Errorf("redis destroy session: %w", err) @@ -288,6 +421,11 @@ func (m *RedisManager) Destroy(ctx context.Context, sessionID string) error { return ErrSessionNotFound } + // 清理用户会话索引 + if userID != "" { + m.rdb.SRem(ctx, userSessKey(userID), sessionID) + } + logger.Log.Debugw("redis session destroyed", "session", sessionID) return nil } diff --git a/backend/internal/ws/handler.go b/backend/internal/ws/handler.go index 5aa2acd..25d0f97 100644 --- a/backend/internal/ws/handler.go +++ b/backend/internal/ws/handler.go @@ -114,7 +114,7 @@ func serveWS(c *gin.Context, sessionMgr session.Manager, orch orchestrator.Orche defer conn.Close() // 创建会话 - sessionID, err := sessionMgr.Create(context.Background(), models.DefaultConfig()) + sessionID, err := sessionMgr.Create(context.Background(), "", models.DefaultConfig()) if err != nil { logger.Log.Errorw("create session failed", "error", err) return