feat: 实现 Redis 版 SessionManager（Hash + List，TTL 自动刷新）

2026-06-13 15:27:41 +08:00
parent f172e6c468
commit ba6d565c0f
3 changed files with 340 additions and 3 deletions
--- a/backend/internal/session/redis.go
+++ 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()
+}