feat: 扩展 Session Manager 接口,新增 ListByUser、UpdateTitle 方法

- 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 调用
This commit is contained in:
hhs
2026-06-14 17:35:20 +08:00
parent 9ff971fd89
commit 6487a8ecab
7 changed files with 293 additions and 38 deletions

View File

@@ -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
}