mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-09-30 06:16:37 +08:00
7d71f1e4e1
- New internal/repository/logstore abstraction: exported domain interfaces (AccessLogStore/ObservabilityStore/UserAccessLogStore/StatusStore), config-driven provider (Active/Build/Migrating/SetConfigReader), GORM implementation for PostgreSQL/SQLite (incl. hourly rollups computed in real time, migration listers, PG partition maintenance), and a ClickHouse wrapper preserving the native batch path; repository facade delegates to logstore; import-lint test enforces apps never import analyticsrepo. - ClickHouse is now optional: the log DB is either the main DB (postgres when database.enabled, else sqlite) or clickhouse; boot validation + first-run seed; log_database / log_db_migration are protected keys. - New user task 切换日志数据库 (of_log_db_switch): freeze log writes, drain batch writers, copy all 6 raw log tables by id (preserving IDs) with target-partition pre-creation for PG, flip log_database on success, clear the freeze flag on failure. - Per-store retention (log_retention_days_*) with expiry cleanup folded into the daily system_cleanup task; legacy database_auto_cleanup_* and of_database_auto_cleanup decommissioned. - goose migrations: 6 log tables in PG (2 monthly-partitioned) + SQLite, retention config seeds, schedule cleanup; GET /api/v1/admin/status/log-database endpoint; frontend retention settings, switch-task UI and status badge; changelog and docs updated. docs(plan): log database decoupling implementation plan docs(design): log database decoupling design (ClickHouse optional)
132 lines
3.4 KiB
Go
132 lines
3.4 KiB
Go
// Copyright 2026 Arctel.net
|
||
// SPDX-License-Identifier: Apache-2.0
|
||
|
||
package system_config
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"time"
|
||
|
||
"github.com/Rain-kl/Wavelet/internal/infra/objectstore"
|
||
"github.com/Rain-kl/Wavelet/internal/model"
|
||
"github.com/Rain-kl/Wavelet/internal/repository"
|
||
"github.com/Rain-kl/Wavelet/pkg/logger"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
func createSystemConfig(ctx context.Context, req CreateSystemConfigRequest) error {
|
||
// 防御:受保护 key(log_database / log_db_migration)仅允许内部写入,Handler 已拦截。
|
||
if isProtectedConfigKey(req.Key) {
|
||
return errors.New(protectedConfigKeyMessage)
|
||
}
|
||
|
||
exists, err := repository.SystemConfigExists(ctx, req.Key)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if exists {
|
||
return errors.New(ConfigKeyExists)
|
||
}
|
||
|
||
config := model.SystemConfig{
|
||
Key: req.Key,
|
||
Value: req.Value,
|
||
Type: req.Type,
|
||
Visibility: req.Visibility,
|
||
Description: req.Description,
|
||
}
|
||
if err := repository.CreateSystemConfig(ctx, &config); err != nil {
|
||
return err
|
||
}
|
||
|
||
invalidateSystemConfigCaches(ctx, req.Key)
|
||
if err := repository.InvalidateVisibleSystemConfigsCache(ctx); err != nil {
|
||
logger.WarnF(ctx, "清理公共配置列表缓存失败: %v", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func listSystemConfigs(ctx context.Context, configType string) ([]model.SystemConfig, error) {
|
||
return repository.ListAdminSystemConfigs(ctx, configType)
|
||
}
|
||
|
||
func getSystemConfig(ctx context.Context, key string) (model.SystemConfig, error) {
|
||
return repository.GetAdminSystemConfigByKey(ctx, key)
|
||
}
|
||
|
||
func updateSystemConfig(ctx context.Context, key string, req UpdateSystemConfigRequest) error {
|
||
config, err := repository.GetAdminSystemConfigByKey(ctx, key)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
var originalDriver objectstore.Driver
|
||
if key == model.ConfigKeyStorageConfig {
|
||
var currentCfg objectstore.Config
|
||
if err := json.Unmarshal([]byte(config.Value), ¤tCfg); err == nil {
|
||
originalDriver = currentCfg.Driver
|
||
}
|
||
|
||
validatedVal, err := validateAndMergeStorageConfig(ctx, req.Value, config.Value)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
req.Value = validatedVal
|
||
}
|
||
|
||
if err := repository.RunInTransaction(ctx, func(tx *gorm.DB) error {
|
||
updates := map[string]any{
|
||
"description": req.Description,
|
||
}
|
||
if req.Visibility != nil {
|
||
updates["visibility"] = *req.Visibility
|
||
config.Visibility = *req.Visibility
|
||
}
|
||
if key != model.ConfigKeySMTPPassword || req.Value != maskedConfigValue {
|
||
updates["value"] = req.Value
|
||
config.Value = req.Value
|
||
}
|
||
if err := repository.UpdateSystemConfigFieldsTx(tx, &config, updates); err != nil {
|
||
return err
|
||
}
|
||
resolveStorageMigrationTasksOnDirectDriverUpdate(ctx, tx, key, originalDriver, req.Value)
|
||
return nil
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
|
||
invalidateCachesAfterConfigUpdate(ctx, key)
|
||
return nil
|
||
}
|
||
|
||
func resolveStorageMigrationTasksOnDirectDriverUpdate(
|
||
ctx context.Context,
|
||
tx *gorm.DB,
|
||
key string,
|
||
originalDriver objectstore.Driver,
|
||
newValue string,
|
||
) {
|
||
if key != model.ConfigKeyStorageConfig || originalDriver == "" {
|
||
return
|
||
}
|
||
|
||
var newCfg objectstore.Config
|
||
if err := json.Unmarshal([]byte(newValue), &newCfg); err != nil {
|
||
return
|
||
}
|
||
if newCfg.Driver != originalDriver {
|
||
return
|
||
}
|
||
|
||
if err := repository.MarkFailedTaskExecutionsSucceededTx(
|
||
tx,
|
||
"storage:migrate",
|
||
"存储配置直接更新,故障迁移任务自动标记为已解决",
|
||
time.Now(),
|
||
); err != nil {
|
||
logger.ErrorF(ctx, "自动更新迁移任务状态失败: %v", err)
|
||
}
|
||
}
|