From 37c83d4c12f5196e37cdc4e98b3e74de3709eb3d Mon Sep 17 00:00:00 2001
From: Victor_Jay <ovoex.cn@gmail.com>
Date: Sun, 31 May 2026 11:46:50 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20FallbackStore=20=E6=81=A2=E5=A4=8D?=
 =?UTF-8?q?=E6=97=B6=E9=9D=99=E9=BB=98=E8=BF=81=E7=A7=BB=E5=86=85=E5=AD=98?=
 =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E5=88=B0=20Redis?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- MemoryStore 新增 Snapshot() 返回活跃会话副本
- Redis 恢复时自动将降级期间的内存会话 Pipeline 写入 Redis
- 迁移成功后再切换 healthy 标记，用户无需重新登录
- 迁移失败则保持内存模式，避免切到空的 Redis 再被迫降级
---
 internal/session/fallback_store.go | 65 +++++++++++++++++++++++++-----
 internal/session/memory_store.go   | 20 +++++++++
 2 files changed, 75 insertions(+), 10 deletions(-)

diff --git a/internal/session/fallback_store.go b/internal/session/fallback_store.go
index c770f4c..1614492 100644
--- a/internal/session/fallback_store.go
+++ b/internal/session/fallback_store.go
@@ -2,6 +2,8 @@ package session
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"log"
 	"sync/atomic"
 	"time"
@@ -11,19 +13,20 @@ import (
 
 // FallbackStore 会话存储降级保护层
 // Redis 正常时委托给 RedisStore，异常时自动降级到 MemoryStore
-// 后台健康检测 goroutine 周期性检测 Redis 状态并自动恢复
+// Redis 恢复时自动将降级期间的内存会话静默迁移回 Redis，用户无感知
 type FallbackStore struct {
-	redis   *RedisStore
-	memory  Store
-	client  *redis.Client
+	redis  *RedisStore
+	memory *MemoryStore
+	client *redis.Client
 
 	healthy  atomic.Bool // Redis 当前是否健康
 	degraded atomic.Bool // 是否处于降级模式（曾因异常而切换到 Memory）
+	migrating atomic.Bool // 是否正在执行迁移（避免并发迁移）
 }
 
 // NewFallbackStore 创建降级保护存储
 // checkInterval: 健康检测间隔，建议与 session cleanup_interval 一致
-func NewFallbackStore(redis *RedisStore, memory Store, checkInterval time.Duration) *FallbackStore {
+func NewFallbackStore(redis *RedisStore, memory *MemoryStore, checkInterval time.Duration) *FallbackStore {
 	fs := &FallbackStore{
 		redis:  redis,
 		memory: memory,
@@ -34,11 +37,11 @@ func NewFallbackStore(redis *RedisStore, memory Store, checkInterval time.Durati
 	// 后台健康检测 goroutine
 	go fs.healthLoop(checkInterval)
 
-	log.Println("[FallbackStore] 已启动 Redis 健康检测")
+	log.Println("[FallbackStore] 已启动 Redis 健康检测（含自动迁移）")
 	return fs
 }
 
-// healthLoop 定期 ping Redis，自动升降级
+// healthLoop 定期 ping Redis，自动升降级 + 静默迁移
 func (fs *FallbackStore) healthLoop(interval time.Duration) {
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
@@ -46,20 +49,62 @@ func (fs *FallbackStore) healthLoop(interval time.Duration) {
 	for range ticker.C {
 		prev := fs.healthy.Load()
 		ok := fs.ping()
-		fs.healthy.Store(ok)
 
 		if ok && !prev {
-			// Redis 恢复
+			// Redis 恢复：先迁移内存会话，再切回 Redis
+			fs.migrateSessions()
+			fs.healthy.Store(true)
 			fs.degraded.Store(false)
-			log.Println("[FallbackStore] Redis 已恢复，切回 Redis 存储")
 		} else if !ok && prev {
 			// Redis 掉线，标记降级
+			fs.healthy.Store(false)
 			fs.degraded.Store(true)
 			log.Println("[FallbackStore] Redis 不可用，已降级为内存存储")
+		} else {
+			fs.healthy.Store(ok)
 		}
 	}
 }
 
+// migrateSessions 将降级期间内存中的活跃会话同步到 Redis
+func (fs *FallbackStore) migrateSessions() {
+	if !fs.migrating.CompareAndSwap(false, true) {
+		return // 已有迁移在进行中
+	}
+	defer fs.migrating.Store(false)
+
+	sessions := fs.memory.Snapshot()
+	if len(sessions) == 0 {
+		log.Println("[FallbackStore] 无需迁移（降级期间无活跃会话）")
+		return
+	}
+
+	ctx := context.Background()
+	pipe := fs.client.Pipeline()
+
+	for _, s := range sessions {
+		data, err := json.Marshal(s)
+		if err != nil {
+			log.Printf("[FallbackStore] 迁移序列化失败 sid=%s err=%v", s.ID[:16]+"...", err)
+			continue
+		}
+		ttl := fs.redis.idleTimeout
+		if s.RememberMe {
+			ttl = fs.redis.rememberTimeout
+		}
+		pipe.Set(ctx, keyPrefix+s.ID, data, ttl)
+		pipe.SAdd(ctx, userKeyPrefix+fmt.Sprint(s.UserID), s.ID)
+	}
+
+	_, err := pipe.Exec(ctx)
+	if err != nil {
+		log.Printf("[FallbackStore] 迁移到 Redis 失败: %v，保持内存模式", err)
+		return
+	}
+
+	log.Printf("[FallbackStore] Redis 已恢复，静默迁移 %d 条会话", len(sessions))
+}
+
 // ping 检查 Redis 连接状态（带超时）
 func (fs *FallbackStore) ping() bool {
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
diff --git a/internal/session/memory_store.go b/internal/session/memory_store.go
index 7e0cff1..baf322b 100644
--- a/internal/session/memory_store.go
+++ b/internal/session/memory_store.go
@@ -211,3 +211,23 @@ func (ms *MemoryStore) Metrics() StoreMetrics {
 		ActiveCount: count,
 	}
 }
+
+// Snapshot 返回所有活跃会话的副本（用于 Redis 恢复时迁移）
+func (ms *MemoryStore) Snapshot() []*Session {
+	ms.mu.RLock()
+	defer ms.mu.RUnlock()
+
+	var active []*Session
+	for _, s := range ms.sessions {
+		timeout := ms.idleTimeout
+		if s.RememberMe {
+			timeout = ms.rememberTimeout
+		}
+		if !s.IsExpired(timeout) {
+			// 返回副本，避免外部修改内部 map
+			copyS := *s
+			active = append(active, &copyS)
+		}
+	}
+	return active
+}