feat(log): 解耦用户访问日志存储，支持切换日志主库

用户访问日志可在 ClickHouse、PostgreSQL、SQLite 之间切换。关闭 ClickHouse 时由主库承接写入与查询；切换任务会冻结写入、复制数据后翻转主库。启动时校验日志主库与运行配置一致，定期清理按各库保留天数删除过期记录。
2026-10-07 16:16:37 +08:00 · 2026-08-16 11:17:55 +08:00
parent 6a53619dd2
commit e52592b16d
34 changed files with 2325 additions and 145 deletions
@@ -14,10 +14,9 @@ import (
 	"time"

 	"github.com/Rain-kl/Wavelet/internal/apps/admin"
-	"github.com/Rain-kl/Wavelet/internal/infra/config"
-	"github.com/Rain-kl/Wavelet/internal/infra/persistence"
 	"github.com/Rain-kl/Wavelet/internal/repository"
 	analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
+	"github.com/Rain-kl/Wavelet/internal/repository/logstore"
 	"github.com/Rain-kl/Wavelet/pkg/logger"
 	"github.com/gin-gonic/gin"

@@ -227,7 +226,7 @@ func enrichAccessLogsWithUsers(ctx context.Context, list []accessLogItem) {

 // GetAccessLogs 获取 ClickHouse 异步采集的访问日志
 // @Summary 获取用户访问日志
-// @Description 分页并按照用户、接口路径、时间范围等维度检索 ClickHouse 用户访问日志列表（需要管理员权限，ClickHouse 未启用时报错）
+// @Description 分页并按照用户、接口路径、时间范围等维度检索用户访问日志列表（需要管理员权限）
 // @Tags admin
 // @Produce json
 // @Security SessionCookie
@@ -238,14 +237,16 @@ func enrichAccessLogsWithUsers(ctx context.Context, list []accessLogItem) {
 // @Param start_time query string false "起始时间（RFC3339 或 YYYY-MM-DD HH:MM:SS）"
 // @Param end_time query string false "结束时间（RFC3339 或 YYYY-MM-DD HH:MM:SS）"
 // @Success 200 {object} response.Any{data=logs.accessLogsResponse} "访问日志列表"
-// @Failure 400 {object} response.Any "ClickHouse 未启用或参数错误"
+// @Failure 400 {object} response.Any "参数错误"
 // @Failure 401 {object} response.Any "未登录"
 // @Failure 403 {object} response.Any "无管理员权限"
+// @Failure 500 {object} response.Any "内部错误"
 // @Router /api/v1/admin/logs/access [get]
 func GetAccessLogs(c *gin.Context) {
 	ctx := c.Request.Context()
-	if !config.Config.ClickHouse.Enabled || db.ChDB(ctx) == nil {
-		response.AbortWithError(c, http.StatusBadRequest, "ClickHouse 存储服务未启用，无法检索访问日志")
+	store, err := logstore.Active(ctx)
+	if err != nil {
+		response.AbortInternal(c, "日志存储初始化失败")
 		return
 	}

@@ -271,7 +272,7 @@ func GetAccessLogs(c *gin.Context) {
 		return
 	}

-	logs, total, err := analyticsrepo.ListAccessLogs(ctx, filter, page, pageSize)
+	logs, total, err := store.UserAccessLogs.List(ctx, filter, page, pageSize)
 	if err != nil {
 		response.AbortWithError(c, http.StatusInternalServerError, err.Error())
 		return
@@ -333,25 +334,26 @@ type logsAnalyticsResponse struct {

 // GetLogsAnalytics 获取 ClickHouse 访问日志图表聚合指标
 // @Summary 获取访问日志分析数据
-// @Description 聚合统计最近 7 天的每日访问趋势、浏览器分布以及前 10 名最活跃用户排行（需要管理员权限，ClickHouse 未启用时报错）
+// @Description 聚合统计最近 7 天的每日访问趋势、浏览器分布以及前 10 名最活跃用户排行（需要管理员权限）
 // @Tags admin
 // @Produce json
 // @Security SessionCookie
 // @Success 200 {object} response.Any{data=logs.logsAnalyticsResponse} "分析统计数据"
-// @Failure 400 {object} response.Any "ClickHouse 未启用"
+// @Failure 500 {object} response.Any "内部错误"
 // @Failure 401 {object} response.Any "未登录"
 // @Failure 403 {object} response.Any "无管理员权限"
 // @Router /api/v1/admin/logs/analytics [get]
 func GetLogsAnalytics(c *gin.Context) {
 	ctx := c.Request.Context()
-	if !config.Config.ClickHouse.Enabled || db.ChDB(ctx) == nil {
-		response.AbortWithError(c, http.StatusBadRequest, "ClickHouse 存储服务未启用，无法获取分析数据")
+	store, err := logstore.Active(ctx)
+	if err != nil {
+		response.AbortInternal(c, "日志存储初始化失败")
 		return
 	}

 	startTime := time.Now().AddDate(0, 0, -(analyticsDays - 1)).Truncate(hoursInDay * time.Hour)

-	trendPoints, err := analyticsrepo.GetDailyTrend(ctx, analyticsDays)
+	trendPoints, err := store.UserAccessLogs.GetDailyTrend(ctx, analyticsDays)
 	if err != nil {
 		response.AbortWithError(c, http.StatusInternalServerError, "查询访问趋势失败: "+err.Error())
 		return
@@ -364,7 +366,7 @@ func GetLogsAnalytics(c *gin.Context) {
 		}
 	}

-	browserPoints, err := analyticsrepo.GetBrowserDistribution(ctx, startTime)
+	browserPoints, err := store.UserAccessLogs.GetBrowserDistribution(ctx, startTime)
 	if err != nil {
 		response.AbortWithError(c, http.StatusInternalServerError, "查询浏览器分布失败: "+err.Error())
 		return
@@ -377,7 +379,7 @@ func GetLogsAnalytics(c *gin.Context) {
 		}
 	}

-	topUserPoints, err := analyticsrepo.GetTopActiveUsers(ctx, startTime, topActiveLimit)
+	topUserPoints, err := store.UserAccessLogs.GetTopActiveUsers(ctx, startTime, topActiveLimit)
 	if err != nil {
 		response.AbortWithError(c, http.StatusInternalServerError, "查询活跃用户失败: "+err.Error())
 		return
@@ -0,0 +1,219 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package logs
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/Rain-kl/Wavelet/internal/apps/risk_control"
+	"github.com/Rain-kl/Wavelet/internal/infra/config"
+	"github.com/Rain-kl/Wavelet/internal/infra/task"
+	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/internal/repository"
+	"github.com/Rain-kl/Wavelet/internal/repository/logstore"
+	"github.com/Rain-kl/Wavelet/pkg/logger"
+)
+
+const (
+	// LogDBSwitchTask 切换日志数据库任务标识。
+	LogDBSwitchTask = "logs:db_switch"
+	// TaskTypeLogDBSwitch 管理端任务类型。
+	TaskTypeLogDBSwitch = "logs_db_switch"
+
+	copyBatchSize    = 1000
+	targetPostgres   = "postgres"
+	targetSQLite     = "sqlite"
+	targetClickHouse = "clickhouse"
+)
+
+// LogDBSwitchMeta 描述切换日志数据库任务。
+var LogDBSwitchMeta = task.TaskMeta{
+	Type:         TaskTypeLogDBSwitch,
+	AsynqTask:    LogDBSwitchTask,
+	Name:         "切换日志数据库",
+	Description:  "复制迁移用户访问日志并在成功后切换日志主库（期间禁止日志写入）",
+	SupportsTime: false,
+	MaxRetry:     task.DefaultMaxRetry,
+	Queue:        task.QueueDefault,
+	Retryable:    true,
+	Params: []task.TaskParam{
+		{Name: "target", Label: "目标日志库", Type: "string", Required: true,
+			Placeholder: "postgres|sqlite|clickhouse", Description: "迁移目标：postgres（主库为 PG 时）、sqlite（主库为 SQLite 时）或 clickhouse"},
+	},
+}
+
+type logDBSwitchPayload struct {
+	Target string `json:"target"`
+}
+
+// LogDBSwitchHandler 切换日志数据库任务处理器。
+type LogDBSwitchHandler struct{}
+
+// ValidatePayload 校验并规范化参数。
+func (h *LogDBSwitchHandler) ValidatePayload(payload []byte) ([]byte, error) {
+	var p logDBSwitchPayload
+	if err := json.Unmarshal(payload, &p); err != nil {
+		return nil, fmt.Errorf("参数解析失败: %w", err)
+	}
+	p.Target = normalizeTarget(p.Target)
+	if !validTarget(p.Target) {
+		return nil, fmt.Errorf("目标日志库不合法: %s", p.Target)
+	}
+	out, err := json.Marshal(p)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func normalizeTarget(v string) string {
+	switch v {
+	case targetPostgres, "postgresql":
+		return targetPostgres
+	case targetSQLite, "sqlite3":
+		return targetSQLite
+	case targetClickHouse, "ch":
+		return targetClickHouse
+	}
+	return v
+}
+
+func validTarget(v string) bool {
+	return v == targetPostgres || v == targetSQLite || v == targetClickHouse
+}
+
+// Execute 执行迁移。
+func (h *LogDBSwitchHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) {
+	var p logDBSwitchPayload
+	if err := json.Unmarshal(payload, &p); err != nil {
+		return nil, fmt.Errorf("参数解析失败: %w", err)
+	}
+	p.Target = normalizeTarget(p.Target)
+	if err := validateSwitch(ctx, p.Target); err != nil {
+		return nil, err
+	}
+
+	source, err := currentLogDatabase(ctx)
+	if err != nil {
+		task.AppendLog(ctx, "读取日志主库失败: %v", err)
+		return nil, err
+	}
+	task.AppendLog(ctx, "开始切换日志数据库：%s -> %s", source, p.Target)
+
+	if err := setMigrationFlag(ctx, "migrating"); err != nil {
+		return nil, err
+	}
+	defer func() {
+		if err := setMigrationFlag(ctx, ""); err != nil {
+			logger.ErrorF(ctx, "清除日志迁移冻结标记失败: %v", err)
+		}
+	}()
+
+	if err := risk_control.Drain(ctx); err != nil {
+		return nil, fmt.Errorf("排空日志写入队列失败: %w", err)
+	}
+
+	src, err := logstore.Active(ctx)
+	if err != nil {
+		return nil, err
+	}
+	dst, err := logstore.BuildForMigration(ctx, p.Target)
+	if err != nil {
+		return nil, err
+	}
+
+	if _, err := dst.UserAccessLogs.DeleteAll(ctx); err != nil {
+		return nil, fmt.Errorf("清空目标用户访问日志失败: %w", err)
+	}
+	from, to, err := src.UserAccessLogs.MigrationRange(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("读取源库时间范围失败: %w", err)
+	}
+	if !from.IsZero() && !to.IsZero() {
+		if err := dst.UserAccessLogs.EnsurePartitions(ctx, from, to); err != nil {
+			return nil, fmt.Errorf("预建目标分区失败: %w", err)
+		}
+	}
+
+	if err := copyUserAccessLogs(ctx, src, dst); err != nil {
+		return nil, err
+	}
+	if err := flipLogDatabase(ctx, p.Target); err != nil {
+		return nil, err
+	}
+	logstore.InvalidateCache()
+	task.AppendLog(ctx, "日志数据库已切换为 %s，写入恢复", p.Target)
+	return &task.TaskResult{Message: fmt.Sprintf("日志数据库已从 %s 切换为 %s", source, p.Target)}, nil
+}
+
+func validateSwitch(ctx context.Context, target string) error {
+	source, err := currentLogDatabase(ctx)
+	if err != nil {
+		return err
+	}
+	if source == target {
+		return errors.New("目标日志库与当前日志库相同，无需迁移")
+	}
+	switch target {
+	case targetClickHouse:
+		if !config.Config.ClickHouse.Enabled {
+			return errors.New("ClickHouse 未启用，无法迁移到 ClickHouse")
+		}
+	case targetPostgres:
+		if !config.Config.Database.Enabled {
+			return errors.New("PostgreSQL 未启用（当前主库为 SQLite），无法迁移到 PostgreSQL")
+		}
+	case targetSQLite:
+		if config.Config.Database.Enabled {
+			return errors.New("当前主库为 PostgreSQL，日志库不能设置为 SQLite")
+		}
+	}
+	return nil
+}
+
+func currentLogDatabase(ctx context.Context) (string, error) {
+	cfg, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeyLogDatabase)
+	if err != nil {
+		return "", fmt.Errorf("读取日志主库失败: %w", err)
+	}
+	if cfg.Value == "" {
+		return "", errors.New("日志主库配置为空")
+	}
+	return cfg.Value, nil
+}
+
+func setMigrationFlag(ctx context.Context, v string) error {
+	return repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyLogDBMigration, v)
+}
+
+func flipLogDatabase(ctx context.Context, target string) error {
+	return repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyLogDatabase, target)
+}
+
+func copyUserAccessLogs(ctx context.Context, src, dst *logstore.Store) error {
+	var afterID uint64
+	var copied int
+	for {
+		rows, err := src.UserAccessLogs.ListForMigration(ctx, afterID, copyBatchSize)
+		if err != nil {
+			return fmt.Errorf("读取源用户访问日志失败: %w", err)
+		}
+		if len(rows) == 0 {
+			break
+		}
+		if err := dst.UserAccessLogs.BatchInsert(ctx, rows); err != nil {
+			return fmt.Errorf("写入目标用户访问日志失败: %w", err)
+		}
+		afterID = rows[len(rows)-1].ID
+		copied += len(rows)
+		task.AppendLog(ctx, "已复制用户访问日志 %d 条", copied)
+		if len(rows) < copyBatchSize {
+			break
+		}
+	}
+	return nil
+}