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(对话历史) // - user:{id}:sessions → Set(用户会话索引) 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) } func userSessKey(id string) string { return fmt.Sprintf("user:%s:sessions", id) } // 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 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), "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, "user_id", userID) 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"], 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"] 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 } now := time.Now().UTC().Format(time.RFC3339) fields := map[string]interface{}{ "last_active": now, "updated_at": now, } 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 } // 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) { // 检查会话是否存在 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) 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) } 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 { // 先获取 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) } if deleted == 0 { return ErrSessionNotFound } // 清理用户会话索引 if userID != "" { m.rdb.SRem(ctx, userSessKey(userID), sessionID) } 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() }