2026-06-13 15:27:41 +08:00
|
|
|
|
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(对话历史)
|
2026-06-14 17:35:20 +08:00
|
|
|
|
// - user:{id}:sessions → Set(用户会话索引)
|
2026-06-13 15:27:41 +08:00
|
|
|
|
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}
|
|
|
|
|
|
}
|
|
|
|
|
|
|
2026-06-14 17:35:20 +08:00
|
|
|
|
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) }
|
2026-06-13 15:27:41 +08:00
|
|
|
|
|
2026-06-14 17:35:20 +08:00
|
|
|
|
// Create 创建新会话。userID 为空表示匿名会话。
|
|
|
|
|
|
func (m *RedisManager) Create(ctx context.Context, userID string, config models.SessionConfig) (string, error) {
|
2026-06-13 15:27:41 +08:00
|
|
|
|
id := uuidNew()
|
|
|
|
|
|
now := time.Now().UTC()
|
|
|
|
|
|
|
|
|
|
|
|
pipe := m.rdb.Pipeline()
|
|
|
|
|
|
|
|
|
|
|
|
// 写入 meta Hash
|
2026-06-14 17:35:20 +08:00
|
|
|
|
meta := map[string]interface{}{
|
|
|
|
|
|
"session_id": id,
|
|
|
|
|
|
"user_id": userID,
|
|
|
|
|
|
"title": models.DefaultSessionTitle,
|
|
|
|
|
|
"config.tts_enabled": strconv.FormatBool(config.TTSEnabled),
|
2026-06-13 15:27:41 +08:00
|
|
|
|
"config.detail_level": config.DetailLevel,
|
2026-06-14 17:35:20 +08:00
|
|
|
|
"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)
|
2026-06-13 15:27:41 +08:00
|
|
|
|
pipe.Expire(ctx, metaKey(id), m.ttl)
|
|
|
|
|
|
|
|
|
|
|
|
// 初始化空 history List
|
|
|
|
|
|
pipe.RPush(ctx, histKey(id), placeholderHistoryMark)
|
|
|
|
|
|
pipe.Expire(ctx, histKey(id), m.ttl)
|
|
|
|
|
|
|
2026-06-14 17:35:20 +08:00
|
|
|
|
// 如果有 userID,添加到用户会话索引
|
|
|
|
|
|
if userID != "" {
|
|
|
|
|
|
pipe.SAdd(ctx, userSessKey(userID), id)
|
|
|
|
|
|
pipe.Expire(ctx, userSessKey(userID), m.ttl)
|
|
|
|
|
|
}
|
|
|
|
|
|
|
2026-06-13 15:27:41 +08:00
|
|
|
|
if _, err := pipe.Exec(ctx); err != nil {
|
|
|
|
|
|
return "", fmt.Errorf("redis create session: %w", err)
|
|
|
|
|
|
}
|
|
|
|
|
|
|
2026-06-14 17:35:20 +08:00
|
|
|
|
logger.Log.Debugw("redis session created", "session", id, "user_id", userID)
|
2026-06-13 15:27:41 +08:00
|
|
|
|
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{
|
2026-06-14 17:35:20 +08:00
|
|
|
|
ID: vals["session_id"],
|
|
|
|
|
|
UserID: vals["user_id"],
|
|
|
|
|
|
Title: vals["title"],
|
2026-06-13 15:27:41 +08:00
|
|
|
|
}
|
|
|
|
|
|
sess.CreatedAt, _ = time.Parse(time.RFC3339, vals["created_at"])
|
2026-06-14 17:35:20 +08:00
|
|
|
|
sess.UpdatedAt, _ = time.Parse(time.RFC3339, vals["updated_at"])
|
2026-06-13 15:27:41 +08:00
|
|
|
|
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
|
|
|
|
|
|
}
|
|
|
|
|
|
|
2026-06-14 17:35:20 +08:00
|
|
|
|
now := time.Now().UTC().Format(time.RFC3339)
|
2026-06-13 15:27:41 +08:00
|
|
|
|
fields := map[string]interface{}{
|
2026-06-14 17:35:20 +08:00
|
|
|
|
"last_active": now,
|
|
|
|
|
|
"updated_at": now,
|
2026-06-13 15:27:41 +08:00
|
|
|
|
}
|
|
|
|
|
|
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
|
|
|
|
|
|
}
|
|
|
|
|
|
|
2026-06-14 17:35:20 +08:00
|
|
|
|
// 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
|
|
|
|
|
|
}
|
|
|
|
|
|
|
2026-06-13 15:27:41 +08:00
|
|
|
|
// 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)
|
2026-06-14 17:35:20 +08:00
|
|
|
|
|
|
|
|
|
|
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))
|
|
|
|
|
|
}
|
|
|
|
|
|
}
|
2026-06-13 15:27:41 +08:00
|
|
|
|
|
|
|
|
|
|
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 {
|
2026-06-14 17:35:20 +08:00
|
|
|
|
// 先获取 user_id 以便清理索引
|
|
|
|
|
|
userID, _ := m.rdb.HGet(ctx, metaKey(sessionID), "user_id").Result()
|
|
|
|
|
|
|
2026-06-13 15:27:41 +08:00
|
|
|
|
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
|
|
|
|
|
|
}
|
|
|
|
|
|
|
2026-06-14 17:35:20 +08:00
|
|
|
|
// 清理用户会话索引
|
|
|
|
|
|
if userID != "" {
|
|
|
|
|
|
m.rdb.SRem(ctx, userSessKey(userID), sessionID)
|
|
|
|
|
|
}
|
|
|
|
|
|
|
2026-06-13 15:27:41 +08:00
|
|
|
|
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()
|
|
|
|
|
|
}
|