CamTalk/backend/internal/session/redis.go

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"
	"github.com/hhs/camtalk/internal/util"
)

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

// Ping 检查 Redis 连接是否正常。
func (m *RedisManager) Ping(ctx context.Context) error {
	return m.rdb.Ping(ctx).Err()
}

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) {
	return m.CreateWithID(ctx, uuidNew(), userID, config)
}

// CreateWithID 使用指定 ID 创建新会话。
// 供 TieredManager 调用，确保 L1/L2 使用相同的 session ID。
func (m *RedisManager) CreateWithID(ctx context.Context, id string, userID string, config models.SessionConfig) (string, error) {
	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_len", len(raw),
				"raw_preview", util.Truncate(raw, 100))
			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()
}