From f172e6c468a6cb82dd1947b1585fe85fc895b7ed Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sat, 13 Jun 2026 15:25:23 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E5=86=85=E5=AD=98?= =?UTF-8?q?=E7=89=88=20SessionManager=EF=BC=88TTL=2030=20=E5=88=86?= =?UTF-8?q?=E9=92=9F=EF=BC=8C=E5=8E=86=E5=8F=B2=E4=B8=8A=E9=99=90=2020=20?= =?UTF-8?q?=E6=9D=A1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/session/memory.go | 275 +++++++++++++++++++++++++++++ 1 file changed, 275 insertions(+) create mode 100644 backend/internal/session/memory.go diff --git a/backend/internal/session/memory.go b/backend/internal/session/memory.go new file mode 100644 index 0000000..edea8cb --- /dev/null +++ b/backend/internal/session/memory.go @@ -0,0 +1,275 @@ +package session + +import ( + "context" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/hhs/camtalk/internal/logger" + "github.com/hhs/camtalk/internal/models" +) + +const ( + defaultTTL = 30 * time.Minute + defaultHistorySize = 20 +) + +// sessionEntry 内部会话条目。 +type sessionEntry struct { + session models.Session + history []models.Message + activeReqID string + lastActive time.Time +} + +// MemoryManager 基于内存的 SessionManager 实现。 +// 适用于 MVP 和无 Redis 的开发环境。 +type MemoryManager struct { + mu sync.RWMutex + sessions map[string]*sessionEntry + ttl time.Duration + maxHistory int + stopCleaner chan struct{} +} + +// NewMemoryManager 创建内存版 SessionManager。 +// ttl 为会话过期时间，maxHistory 为对话历史上限（0 表示使用默认值 20）。 +func NewMemoryManager(ttl time.Duration, maxHistory int) *MemoryManager { + if ttl <= 0 { + ttl = defaultTTL + } + if maxHistory <= 0 { + maxHistory = defaultHistorySize + } + + m := &MemoryManager{ + sessions: make(map[string]*sessionEntry), + ttl: ttl, + maxHistory: maxHistory, + stopCleaner: make(chan struct{}), + } + + // 启动后台清理 goroutine，每分钟清除过期会话。 + go m.cleanLoop() + + return m +} + +// cleanLoop 后台定期清理过期会话。 +func (m *MemoryManager) cleanLoop() { + ticker := time.NewTicker(1 * time.Minute) + defer ticker.Stop() + for { + select { + case <-ticker.C: + m.cleanExpired() + case <-m.stopCleaner: + return + } + } +} + +// cleanExpired 清除所有过期会话。 +func (m *MemoryManager) cleanExpired() { + m.mu.Lock() + defer m.mu.Unlock() + + now := time.Now() + for id, entry := range m.sessions { + if now.Sub(entry.lastActive) > m.ttl { + delete(m.sessions, id) + logger.Log.Debugw("session expired (cleaner)", "session", id) + } + } +} + +// Stop 停止后台清理 goroutine。应用退出前调用。 +func (m *MemoryManager) Stop() { + close(m.stopCleaner) +} + +// isExpired 检查会话是否过期（调用方需持锁或在已知 entry 存在时调用）。 +func (m *MemoryManager) isExpired(entry *sessionEntry) bool { + return time.Since(entry.lastActive) > m.ttl +} + +// Create 创建新会话。 +func (m *MemoryManager) Create(_ context.Context, config models.SessionConfig) (string, error) { + m.mu.Lock() + defer m.mu.Unlock() + + id := uuid.New().String() + now := time.Now() + m.sessions[id] = &sessionEntry{ + session: models.Session{ + ID: id, + CreatedAt: now, + Config: config, + }, + history: make([]models.Message, 0), + lastActive: now, + } + + logger.Log.Debugw("session created", "session", id) + return id, nil +} + +// Get 获取会话。 +func (m *MemoryManager) Get(_ context.Context, sessionID string) (*models.Session, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return nil, ErrSessionNotFound + } + + sess := entry.session // 复制一份返回 + return &sess, nil +} + +// UpdateConfig 更新会话配置。 +func (m *MemoryManager) UpdateConfig(_ context.Context, sessionID string, patch models.SessionConfigPatch) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + patch.Apply(&entry.session.Config) + entry.lastActive = time.Now() + + logger.Log.Debugw("session config updated", "session", sessionID) + return nil +} + +// GetHistory 获取最近 N 轮对话历史。 +func (m *MemoryManager) GetHistory(_ context.Context, sessionID string, limit int) ([]models.Message, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return nil, ErrSessionNotFound + } + + if limit <= 0 || limit > len(entry.history) { + limit = len(entry.history) + } + + // 返回最近 limit 条的副本 + result := make([]models.Message, limit) + copy(result, entry.history[len(entry.history)-limit:]) + return result, nil +} + +// AppendMessage 追加一条对话消息，同时刷新 TTL。 +func (m *MemoryManager) AppendMessage(_ context.Context, sessionID string, msg models.Message) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + entry.history = append(entry.history, msg) + + // 超过上限时裁剪，保留最新的 maxHistory 条 + if len(entry.history) > m.maxHistory { + entry.history = entry.history[len(entry.history)-m.maxHistory:] + } + + entry.lastActive = time.Now() + return nil +} + +// SetActiveRequest 标记当前正在处理的请求 ID。 +func (m *MemoryManager) SetActiveRequest(_ context.Context, sessionID string, requestID string) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + entry.activeReqID = requestID + entry.lastActive = time.Now() + return nil +} + +// GetActiveRequestID 获取当前活跃请求 ID。 +func (m *MemoryManager) GetActiveRequestID(_ context.Context, sessionID string) (string, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return "", ErrSessionNotFound + } + + return entry.activeReqID, nil +} + +// ClearActiveRequest 清除活跃请求标记。 +func (m *MemoryManager) ClearActiveRequest(_ context.Context, sessionID string) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + entry.activeReqID = "" + entry.lastActive = time.Now() + return nil +} + +// Touch 刷新 TTL。 +func (m *MemoryManager) Touch(_ context.Context, sessionID string) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, ok := m.sessions[sessionID] + if !ok || m.isExpired(entry) { + return ErrSessionNotFound + } + + entry.lastActive = time.Now() + return nil +} + +// Destroy 显式销毁会话。 +func (m *MemoryManager) Destroy(_ context.Context, sessionID string) error { + m.mu.Lock() + defer m.mu.Unlock() + + if _, ok := m.sessions[sessionID]; !ok { + return ErrSessionNotFound + } + + delete(m.sessions, sessionID) + logger.Log.Debugw("session destroyed", "session", sessionID) + return nil +} + +// ActiveCount 返回当前活跃会话数。 +func (m *MemoryManager) ActiveCount() int { + m.mu.RLock() + defer m.mu.RUnlock() + + now := time.Now() + count := 0 + for _, entry := range m.sessions { + if now.Sub(entry.lastActive) <= m.ttl { + count++ + } + } + return count +}