CamTalk/backend/internal/session/tiered.go

package session

import (
	"context"
	"sync/atomic"
	"time"

	"github.com/hhs/camtalk/internal/logger"
	"github.com/hhs/camtalk/internal/models"
	"github.com/hhs/camtalk/internal/store"
)

// TieredManager 三级存储 SessionManager 实现。
//
// L1（内存）→ L2（Redis）→ L3（PostgreSQL）
//
// 读：L1 miss → L2 miss → L3，回填到 L1+L2
// 写：L1 → L2（同步）→ L3（异步）
// 降级：Redis 不可用时，回退到 L1+L3 模式
type TieredManager struct {
	l1      *MemoryManager  // L1: 内存缓存
	l2      *RedisManager   // L2: Redis（可选）
	sessRepo store.SessionRepository // L3: PostgreSQL 会话持久化（可选）
	msgRepo  store.MessageRepository // L3: PostgreSQL 消息持久化（可选）

	redisOK  atomic.Bool     // Redis 健康状态
	stopCh   chan struct{}   // 停止信号
}

// TieredOption TieredManager 的函数式选项。
type TieredOption func(*TieredManager)

// WithTieredSessionRepository 注入 L3 会话持久化仓库。
func WithTieredSessionRepository(repo store.SessionRepository) TieredOption {
	return func(m *TieredManager) {
		m.sessRepo = repo
	}
}

// WithTieredMessageRepository 注入 L3 消息持久化仓库。
func WithTieredMessageRepository(repo store.MessageRepository) TieredOption {
	return func(m *TieredManager) {
		m.msgRepo = repo
	}
}

// NewTieredManager 创建三级存储 SessionManager。
// l2 为 nil 时降级为 L1+L3 模式。
func NewTieredManager(
	ttl time.Duration,
	maxHistory int,
	l2 *RedisManager,
	opts ...TieredOption,
) *TieredManager {
	m := &TieredManager{
		l2:     l2,
		stopCh: make(chan struct{}),
	}

	for _, opt := range opts {
		opt(m)
	}

	// 初始化 L1（内存），注入 L3 仓库实现 Write-Through
	var l1Opts []Option
	if m.sessRepo != nil {
		l1Opts = append(l1Opts, WithSessionRepository(m.sessRepo))
	}
	if m.msgRepo != nil {
		l1Opts = append(l1Opts, WithMessageRepository(m.msgRepo))
	}
	m.l1 = NewMemoryManager(ttl, maxHistory, l1Opts...)

	// 初始化 Redis 健康状态
	if l2 != nil {
		m.redisOK.Store(true)
		go m.healthCheck()
	}

	return m
}

// healthCheck 定期检查 Redis 健康状态。
func (m *TieredManager) healthCheck() {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			err := m.l2.Ping(ctx)
			cancel()

			wasOK := m.redisOK.Load()
			isOK := err == nil
			m.redisOK.Store(isOK)

			if wasOK && !isOK {
				logger.Log.Warn("Redis connection lost, degrading to L1+L3 mode")
			} else if !wasOK && isOK {
				logger.Log.Info("Redis connection restored, resuming L1+L2+L3 mode")
			}
		case <-m.stopCh:
			return
		}
	}
}

// isRedisOK 检查 Redis 是否可用。
func (m *TieredManager) isRedisOK() bool {
	return m.l2 != nil && m.redisOK.Load()
}

// Stop 停止 TieredManager（清理后台 goroutine）。
func (m *TieredManager) Stop() {
	close(m.stopCh)
	m.l1.Stop()
}

// Create 创建新会话。
// 写入：L1 → L2（同步）→ L3（异步）
func (m *TieredManager) Create(ctx context.Context, userID string, config models.SessionConfig) (string, error) {
	// L1: 内存
	id, err := m.l1.Create(ctx, userID, config)
	if err != nil {
		return "", err
	}

	// L2: Redis（同步），使用 L1 生成的 ID 保证一致性
	if m.isRedisOK() {
		if _, err := m.l2.CreateWithID(ctx, id, userID, config); err != nil {
			logger.Log.Warnw("Redis Create failed, continuing without L2",
				"session", id, "error", err)
		}
	}

	// L3: PostgreSQL（异步，由 L1 的 Write-Through 处理）

	return id, nil
}

// Get 获取会话。
// 读取：L1 → L2（回填 L1）→ L3（回填 L1+L2）
func (m *TieredManager) Get(ctx context.Context, sessionID string) (*models.Session, error) {
	// L1: 内存
	sess, err := m.l1.Get(ctx, sessionID)
	if err == nil {
		return sess, nil
	}
	if err != ErrSessionNotFound {
		return nil, err
	}

	// L2: Redis
	if m.isRedisOK() {
		sess, err = m.l2.Get(ctx, sessionID)
		if err == nil {
			// 回填 L1
			history, _ := m.l2.GetHistory(ctx, sessionID, 0)
			m.l1.LoadSession(sess, history)
			return sess, nil
		}
		if err != ErrSessionNotFound {
			logger.Log.Warnw("Redis Get failed",
				"session", sessionID, "error", err)
		}
	}

	// L3: PostgreSQL（由 L1 的 Cache-Aside 处理）
	// L1.Get 已经实现了从 PostgreSQL 恢复的逻辑
	return nil, ErrSessionNotFound
}

// UpdateConfig 更新会话配置。
// 写入：L1 → L2（同步）→ L3（异步）
func (m *TieredManager) UpdateConfig(ctx context.Context, sessionID string, patch models.SessionConfigPatch) error {
	// L1: 内存
	if err := m.l1.UpdateConfig(ctx, sessionID, patch); err != nil {
		return err
	}

	// L2: Redis（同步）
	if m.isRedisOK() {
		if err := m.l2.UpdateConfig(ctx, sessionID, patch); err != nil {
			logger.Log.Warnw("Redis UpdateConfig failed",
				"session", sessionID, "error", err)
		}
	}

	// L3: PostgreSQL（异步，由 L1 的 Write-Through 处理）

	return nil
}

// UpdateTitle 更新会话标题。
// 写入：L1 → L2（同步）→ L3（异步）
func (m *TieredManager) UpdateTitle(ctx context.Context, sessionID string, title string) error {
	// L1: 内存
	if err := m.l1.UpdateTitle(ctx, sessionID, title); err != nil {
		return err
	}

	// L2: Redis（同步）
	if m.isRedisOK() {
		if err := m.l2.UpdateTitle(ctx, sessionID, title); err != nil {
			logger.Log.Warnw("Redis UpdateTitle failed",
				"session", sessionID, "error", err)
		}
	}

	// L3: PostgreSQL（异步，由 L1 的 Write-Through 处理）

	return nil
}

// ListByUser 查询用户的会话列表。
// 读取：L1 + L2 + L3 合并去重
func (m *TieredManager) ListByUser(ctx context.Context, userID string, page, size int) ([]ConversationSummary, int, error) {
	// 优先使用 L1（已集成 L3 回退逻辑）
	return m.l1.ListByUser(ctx, userID, page, size)
}

// GetHistory 获取对话历史。
// 读取：L1 → L2（回填 L1）→ L3（回填 L1+L2）
func (m *TieredManager) GetHistory(ctx context.Context, sessionID string, limit int) ([]models.Message, error) {
	// L1: 内存
	msgs, err := m.l1.GetHistory(ctx, sessionID, limit)
	if err == nil && len(msgs) > 0 {
		return msgs, nil
	}

	// L2: Redis
	if m.isRedisOK() {
		msgs, err = m.l2.GetHistory(ctx, sessionID, limit)
		if err == nil && len(msgs) > 0 {
			// 回填 L1（通过 Get 触发）
			m.l1.Get(ctx, sessionID)
			return msgs, nil
		}
	}

	// L3: PostgreSQL（由 L1 的 Cache-Aside 处理）
	return nil, ErrSessionNotFound
}

// AppendMessage 追加消息。
// 写入：L1 → L2（同步）→ L3（异步）
func (m *TieredManager) AppendMessage(ctx context.Context, sessionID string, msg models.Message) error {
	// L1: 内存（Write-Through 到 L3）
	if err := m.l1.AppendMessage(ctx, sessionID, msg); err != nil {
		return err
	}

	// L2: Redis（同步）
	if m.isRedisOK() {
		if err := m.l2.AppendMessage(ctx, sessionID, msg); err != nil {
			logger.Log.Warnw("Redis AppendMessage failed",
				"session", sessionID, "error", err)
		}
	}

	return nil
}

// SetActiveRequest 设置当前活跃请求。
func (m *TieredManager) SetActiveRequest(ctx context.Context, sessionID string, requestID string) error {
	// L1: 内存
	if err := m.l1.SetActiveRequest(ctx, sessionID, requestID); err != nil {
		return err
	}

	// L2: Redis（同步）
	if m.isRedisOK() {
		if err := m.l2.SetActiveRequest(ctx, sessionID, requestID); err != nil {
			logger.Log.Warnw("Redis SetActiveRequest failed",
				"session", sessionID, "error", err)
		}
	}

	return nil
}

// GetActiveRequestID 获取当前活跃请求 ID。
func (m *TieredManager) GetActiveRequestID(ctx context.Context, sessionID string) (string, error) {
	// L1: 内存
	id, err := m.l1.GetActiveRequestID(ctx, sessionID)
	if err == nil && id != "" {
		return id, nil
	}

	// L2: Redis
	if m.isRedisOK() {
		id, err = m.l2.GetActiveRequestID(ctx, sessionID)
		if err == nil && id != "" {
			return id, nil
		}
	}

	return "", nil
}

// ClearActiveRequest 清除当前活跃请求。
func (m *TieredManager) ClearActiveRequest(ctx context.Context, sessionID string) error {
	// L1: 内存
	if err := m.l1.ClearActiveRequest(ctx, sessionID); err != nil {
		return err
	}

	// L2: Redis（同步）
	if m.isRedisOK() {
		if err := m.l2.ClearActiveRequest(ctx, sessionID); err != nil {
			logger.Log.Warnw("Redis ClearActiveRequest failed",
				"session", sessionID, "error", err)
		}
	}

	return nil
}

// Touch 刷新会话活跃时间。
func (m *TieredManager) Touch(ctx context.Context, sessionID string) error {
	// L1: 内存
	if err := m.l1.Touch(ctx, sessionID); err != nil {
		return err
	}

	// L2: Redis（同步）
	if m.isRedisOK() {
		if err := m.l2.Touch(ctx, sessionID); err != nil {
			logger.Log.Warnw("Redis Touch failed",
				"session", sessionID, "error", err)
		}
	}

	return nil
}

// Destroy 销毁会话。
// 写入：L1 → L2 → L3
func (m *TieredManager) Destroy(ctx context.Context, sessionID string) error {
	// L1: 内存（Write-Through 到 L3）
	if err := m.l1.Destroy(ctx, sessionID); err != nil {
		return err
	}

	// L2: Redis（同步）
	if m.isRedisOK() {
		if err := m.l2.Destroy(ctx, sessionID); err != nil {
			logger.Log.Warnw("Redis Destroy failed",
				"session", sessionID, "error", err)
		}
	}

	return nil
}

// ActiveCount 返回活跃会话数量。
func (m *TieredManager) ActiveCount() int {
	return m.l1.ActiveCount()
}