mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-04 23:16:37 +08:00
refactor(plugins): restructure admin and message_gateway into standard layered sub-packages
This commit is contained in:
@@ -0,0 +1,144 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"Wavelet/pkg/cache/ram"
|
||||
"Wavelet/plugins/domain/admin/model"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
// SystemConfigBroadcastChannel broadcasts system config cache updates across nodes.
|
||||
SystemConfigBroadcastChannel = "system:config_broadcast"
|
||||
|
||||
// SystemConfigInvalidationChannel is kept as an alias for backward compatibility.
|
||||
SystemConfigInvalidationChannel = SystemConfigBroadcastChannel
|
||||
|
||||
// SystemConfigRedisHashKey is kept for backward compatibility in tests.
|
||||
SystemConfigRedisHashKey = "system:system_configs"
|
||||
// SystemConfigVisibleListRedisKey is kept for backward compatibility in tests.
|
||||
SystemConfigVisibleListRedisKey = "system:visible_configs"
|
||||
|
||||
// ConfigCacheType is the cache type for all system configs.
|
||||
ConfigCacheType = "config"
|
||||
)
|
||||
|
||||
// ConfigLoader loads configuration data from the database.
|
||||
type ConfigLoader struct{}
|
||||
|
||||
// LoadAll loads all system configs from database as CacheItems.
|
||||
func (ConfigLoader) LoadAll(ctx context.Context, configType string) ([]ram.CacheItem, error) {
|
||||
configs, err := PreheatSystemConfigs(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
items := make([]ram.CacheItem, len(configs))
|
||||
for i, cfg := range configs {
|
||||
valBytes, err := json.Marshal(cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items[i] = ram.CacheItem{
|
||||
Key: cfg.Key,
|
||||
Value: string(valBytes),
|
||||
Type: configType,
|
||||
TTL: determineTTL(cfg.Key),
|
||||
}
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
// LoadOne loads a single system config from database as CacheItem.
|
||||
func (ConfigLoader) LoadOne(ctx context.Context, configType, key string) (ram.CacheItem, error) {
|
||||
cfg, err := GetSystemConfigByKey(ctx, key)
|
||||
if err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return ram.CacheItem{}, ram.ErrNotFound
|
||||
}
|
||||
return ram.CacheItem{}, err
|
||||
}
|
||||
valBytes, err := json.Marshal(cfg)
|
||||
if err != nil {
|
||||
return ram.CacheItem{}, err
|
||||
}
|
||||
return ram.CacheItem{
|
||||
Key: cfg.Key,
|
||||
Value: string(valBytes),
|
||||
Type: configType,
|
||||
TTL: determineTTL(cfg.Key),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// GetCachedSystemConfig retrieves a single system config with RAM L1 fallback to DB.
|
||||
func GetCachedSystemConfig(ctx context.Context, key string) (*model.SystemConfig, error) {
|
||||
if item, ok := ram.Get(ConfigCacheType, key); ok {
|
||||
var cfg model.SystemConfig
|
||||
if err := json.Unmarshal([]byte(item.Value), &cfg); err == nil {
|
||||
return &cfg, nil
|
||||
}
|
||||
}
|
||||
|
||||
cfg, err := GetSystemConfigByKey(ctx, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
valBytes, err := json.Marshal(cfg)
|
||||
if err == nil {
|
||||
ram.Set(ram.CacheItem{
|
||||
Key: cfg.Key,
|
||||
Value: string(valBytes),
|
||||
Type: ConfigCacheType,
|
||||
TTL: determineTTL(key),
|
||||
})
|
||||
}
|
||||
return &cfg, nil
|
||||
}
|
||||
|
||||
// StopSystemConfigCacheListener stops the cache invalidation listener (kept for backward compatibility).
|
||||
func StopSystemConfigCacheListener() {
|
||||
}
|
||||
|
||||
// StartSystemConfigCacheListener starts the cache listener (kept for backward compatibility).
|
||||
func StartSystemConfigCacheListener() {
|
||||
}
|
||||
|
||||
func ensureSystemConfigCacheListener() {
|
||||
}
|
||||
|
||||
func determineTTL(_ string) time.Duration {
|
||||
return -1
|
||||
}
|
||||
|
||||
// InvalidateSystemConfigCache triggers a broadcast to refresh the cache for key.
|
||||
func InvalidateSystemConfigCache(ctx context.Context, key string) error {
|
||||
ram.Delete(ConfigCacheType, key)
|
||||
if cacheSvc := GetCache(ctx); cacheSvc != nil {
|
||||
_ = cacheSvc.Delete(ctx, "system:config:"+key)
|
||||
_ = cacheSvc.Delete(ctx, SystemConfigVisibleListRedisKey)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// InvalidateAllSystemConfigCaches triggers a broadcast to refresh the entire config cache.
|
||||
func InvalidateAllSystemConfigCaches(ctx context.Context) error {
|
||||
ram.UpdateTypeItems(ConfigCacheType, nil)
|
||||
if cacheSvc := GetCache(ctx); cacheSvc != nil {
|
||||
_ = cacheSvc.Delete(ctx, SystemConfigRedisHashKey)
|
||||
_ = cacheSvc.Delete(ctx, SystemConfigVisibleListRedisKey)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ResetSystemConfigRAMCacheForTest clears only the process-local RAM cache.
|
||||
func ResetSystemConfigRAMCacheForTest() {
|
||||
ram.ResetForTest()
|
||||
}
|
||||
@@ -0,0 +1,371 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"Wavelet/pkg/config"
|
||||
"Wavelet/plugins/domain/admin/errs"
|
||||
"Wavelet/plugins/domain/admin/model"
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultSQLiteDBPath = "./data/wavelet.db"
|
||||
logDBNameSQLite = "sqlite"
|
||||
)
|
||||
|
||||
// sqliteDatabasePath resolves the effective SQLite file path from configuration.
|
||||
func sqliteDatabasePath() string {
|
||||
name := config.Config.Database.SQLitePath
|
||||
if name == "" {
|
||||
name = defaultSQLiteDBPath
|
||||
}
|
||||
return name
|
||||
}
|
||||
|
||||
// QuoteTableName escapes a raw identifier for use inside a quoted SQL fragment.
|
||||
func QuoteTableName(table string) string {
|
||||
return `"` + strings.ReplaceAll(table, `"`, `""`) + `"`
|
||||
}
|
||||
|
||||
// GetSQLiteOverview collects the SQLite runtime overview.
|
||||
func GetSQLiteOverview(ctx context.Context) (model.DBOverviewResponse, error) {
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return model.DBOverviewResponse{}, errs.ErrDatabaseUninitialized
|
||||
}
|
||||
|
||||
name := sqliteDatabasePath()
|
||||
|
||||
var version string
|
||||
var ver string
|
||||
if err := gormDB.Raw("SELECT sqlite_version()").Scan(&ver).Error; err == nil {
|
||||
version = "SQLite " + ver
|
||||
} else {
|
||||
version = "SQLite"
|
||||
}
|
||||
|
||||
var sizeStr string
|
||||
if fi, err := os.Stat(name); err == nil {
|
||||
size := fi.Size()
|
||||
if size < 0 {
|
||||
size = 0
|
||||
}
|
||||
sizeStr = model.FormatBytes(uint64(size))
|
||||
} else {
|
||||
sizeStr = "0 B"
|
||||
}
|
||||
|
||||
var tableCount int64
|
||||
if err := gormDB.Raw("SELECT count(*) FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").Scan(&tableCount).Error; err != nil {
|
||||
tableCount = 0
|
||||
}
|
||||
|
||||
var connCount int64
|
||||
if sqlDB, err := gormDB.DB(); err == nil {
|
||||
connCount = int64(sqlDB.Stats().OpenConnections)
|
||||
} else {
|
||||
connCount = 1
|
||||
}
|
||||
|
||||
return model.DBOverviewResponse{
|
||||
Type: logDBNameSQLite,
|
||||
Version: version,
|
||||
Name: name,
|
||||
Size: sizeStr,
|
||||
TableCount: tableCount,
|
||||
Connections: connCount,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// GetPostgresOverview collects the PostgreSQL runtime overview.
|
||||
func GetPostgresOverview(ctx context.Context) (model.DBOverviewResponse, error) {
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return model.DBOverviewResponse{}, errs.ErrDatabaseUninitialized
|
||||
}
|
||||
|
||||
name := config.Config.Database.Database
|
||||
|
||||
var version string
|
||||
var ver string
|
||||
if err := gormDB.Raw("SELECT version()").Scan(&ver).Error; err == nil {
|
||||
version = ver
|
||||
} else {
|
||||
version = "PostgreSQL"
|
||||
}
|
||||
|
||||
var sizeStr string
|
||||
var sizeBytes sql.NullInt64
|
||||
if err := gormDB.Raw("SELECT pg_database_size(current_database())").Scan(&sizeBytes).Error; err == nil && sizeBytes.Valid {
|
||||
size := sizeBytes.Int64
|
||||
if size < 0 {
|
||||
size = 0
|
||||
}
|
||||
sizeStr = model.FormatBytes(uint64(size))
|
||||
} else {
|
||||
sizeStr = "0 B"
|
||||
}
|
||||
|
||||
var tableCount int64
|
||||
if err := gormDB.Raw("SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema()").Scan(&tableCount).Error; err != nil {
|
||||
tableCount = 0
|
||||
}
|
||||
|
||||
var connCount int64
|
||||
var pgc sql.NullInt64
|
||||
if err := gormDB.Raw("SELECT count(*) FROM pg_stat_activity WHERE datname = current_database()").Scan(&pgc).Error; err == nil && pgc.Valid {
|
||||
connCount = pgc.Int64
|
||||
} else {
|
||||
if sqlDB, err := gormDB.DB(); err == nil {
|
||||
connCount = int64(sqlDB.Stats().OpenConnections)
|
||||
} else {
|
||||
connCount = 1
|
||||
}
|
||||
}
|
||||
|
||||
return model.DBOverviewResponse{
|
||||
Type: "postgres",
|
||||
Version: version,
|
||||
Name: name,
|
||||
Size: sizeStr,
|
||||
TableCount: tableCount,
|
||||
Connections: connCount,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ListDatabaseTableNames returns every user table of the active database.
|
||||
func ListDatabaseTableNames(ctx context.Context) ([]string, error) {
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return nil, errs.ErrDatabaseUninitialized
|
||||
}
|
||||
|
||||
var tables []string
|
||||
var err error
|
||||
|
||||
if !config.Config.Database.Enabled {
|
||||
err = gormDB.Raw("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name").Scan(&tables).Error
|
||||
} else {
|
||||
err = gormDB.Raw("SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema() ORDER BY table_name").Scan(&tables).Error
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return tables, nil
|
||||
}
|
||||
|
||||
// CountDatabaseTableRows counts the rows of the quoted table.
|
||||
func CountDatabaseTableRows(ctx context.Context, quotedTable string) (int64, error) {
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return 0, errs.ErrDatabaseUninitialized
|
||||
}
|
||||
|
||||
var total int64
|
||||
if err := gormDB.Raw("SELECT count(*) FROM " + quotedTable).Scan(&total).Error; err != nil {
|
||||
return 0, errs.NewInvalidInputError(err.Error())
|
||||
}
|
||||
return total, nil
|
||||
}
|
||||
|
||||
// QueryDatabaseTableRows loads one page of raw rows from the quoted table.
|
||||
func QueryDatabaseTableRows(
|
||||
ctx context.Context,
|
||||
quotedTable string,
|
||||
limit int,
|
||||
offset int,
|
||||
) ([]string, []map[string]any, error) {
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return nil, nil, errs.ErrDatabaseUninitialized
|
||||
}
|
||||
|
||||
rows, err := gormDB.Raw("SELECT * FROM "+quotedTable+" LIMIT ? OFFSET ?", limit, offset).Rows()
|
||||
if err != nil {
|
||||
return nil, nil, errs.NewInvalidInputError(err.Error())
|
||||
}
|
||||
defer func() {
|
||||
_ = rows.Close()
|
||||
}()
|
||||
|
||||
cols, err := rows.Columns()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
results, err := scanTableRows(rows, cols)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return cols, results, nil
|
||||
}
|
||||
|
||||
// RunSelectSQL executes an arbitrary select-like statement.
|
||||
func RunSelectSQL(ctx context.Context, sqlStr string) ([]string, []map[string]any, error) {
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return nil, nil, errs.ErrDatabaseUninitialized
|
||||
}
|
||||
|
||||
rows, err := gormDB.Raw(sqlStr).Rows()
|
||||
if err != nil {
|
||||
return nil, nil, errs.NewInvalidInputError(err.Error())
|
||||
}
|
||||
defer func() {
|
||||
_ = rows.Close()
|
||||
}()
|
||||
|
||||
cols, err := rows.Columns()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
results, err := scanTableRows(rows, cols)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return cols, results, nil
|
||||
}
|
||||
|
||||
// RunMutationSQL executes a non-query statement and reports affected rows.
|
||||
func RunMutationSQL(ctx context.Context, sqlStr string) (int64, error) {
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return 0, errs.ErrDatabaseUninitialized
|
||||
}
|
||||
tx := gormDB.Exec(sqlStr)
|
||||
if tx.Error != nil {
|
||||
return 0, errs.NewInvalidInputError(tx.Error.Error())
|
||||
}
|
||||
return tx.RowsAffected, nil
|
||||
}
|
||||
|
||||
// scanTableRows decodes every row of the result set into a column keyed map.
|
||||
func scanTableRows(rows *sql.Rows, cols []string) ([]map[string]any, error) {
|
||||
results := make([]map[string]any, 0)
|
||||
for rows.Next() {
|
||||
row, err := scanRowAsMap(rows, cols)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results = append(results, row)
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// scanRowAsMap decodes a single row, normalising driver byte slices to strings.
|
||||
func scanRowAsMap(rows *sql.Rows, cols []string) (map[string]any, error) {
|
||||
columns := make([]any, len(cols))
|
||||
columnPointers := make([]any, len(cols))
|
||||
for i := range columns {
|
||||
columnPointers[i] = &columns[i]
|
||||
}
|
||||
|
||||
if err := rows.Scan(columnPointers...); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
rowMap := make(map[string]any)
|
||||
for i, colName := range cols {
|
||||
val := columns[i]
|
||||
if b, ok := val.([]byte); ok {
|
||||
rowMap[colName] = string(b)
|
||||
continue
|
||||
}
|
||||
rowMap[colName] = val
|
||||
}
|
||||
return rowMap, nil
|
||||
}
|
||||
|
||||
// GetSQLiteInfo collects the SQLite type/name/version triple.
|
||||
func GetSQLiteInfo(ctx context.Context) model.DatabaseInfoResponse {
|
||||
info := model.DatabaseInfoResponse{
|
||||
Type: logDBNameSQLite,
|
||||
Name: config.Config.Database.SQLitePath,
|
||||
Version: "SQLite",
|
||||
}
|
||||
if info.Name == "" {
|
||||
info.Name = defaultSQLiteDBPath
|
||||
}
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return info
|
||||
}
|
||||
var ver string
|
||||
if err := gormDB.Raw("SELECT sqlite_version()").Scan(&ver).Error; err == nil && ver != "" {
|
||||
info.Version = "SQLite " + ver
|
||||
}
|
||||
return info
|
||||
}
|
||||
|
||||
// GetPostgresInfo collects the PostgreSQL type/name/version triple.
|
||||
func GetPostgresInfo(ctx context.Context) model.DatabaseInfoResponse {
|
||||
info := model.DatabaseInfoResponse{
|
||||
Type: "postgres",
|
||||
Name: config.Config.Database.Database,
|
||||
Version: "PostgreSQL",
|
||||
}
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return info
|
||||
}
|
||||
var ver string
|
||||
if err := gormDB.Raw("SELECT version()").Scan(&ver).Error; err == nil && ver != "" {
|
||||
info.Version = ver
|
||||
}
|
||||
return info
|
||||
}
|
||||
|
||||
// OpenSQLiteExportFile opens the active SQLite database file together with its stat info.
|
||||
func OpenSQLiteExportFile() (*os.File, os.FileInfo, error) {
|
||||
//nolint:gosec // export db file path is trusted
|
||||
f, err := os.Open(sqliteDatabasePath())
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("%s: %w", errs.ErrOpenDatabaseFileFailed, err)
|
||||
}
|
||||
|
||||
fi, err := f.Stat()
|
||||
if err != nil {
|
||||
_ = f.Close()
|
||||
return nil, nil, fmt.Errorf("%s: %w", errs.ErrReadDatabaseFileInfoFailed, err)
|
||||
}
|
||||
return f, fi, nil
|
||||
}
|
||||
|
||||
// NewPgDumpCommand builds the streaming pg_dump command for the active database.
|
||||
func NewPgDumpCommand(ctx context.Context) (*exec.Cmd, string, error) {
|
||||
dbCfg := config.Config.Database
|
||||
|
||||
pgDumpPath, err := exec.LookPath("pg_dump")
|
||||
if err != nil {
|
||||
return nil, "", errors.New(errs.ErrPgDumpUnavailable)
|
||||
}
|
||||
|
||||
args := []string{
|
||||
"--no-password",
|
||||
"-h", dbCfg.Host,
|
||||
"-p", fmt.Sprintf("%d", dbCfg.Port),
|
||||
"-U", dbCfg.Username,
|
||||
dbCfg.Database,
|
||||
}
|
||||
|
||||
//nolint:gosec // pg_dump args are constructed from validated db config
|
||||
cmd := exec.CommandContext(ctx, pgDumpPath, args...)
|
||||
if dbCfg.Password != "" {
|
||||
cmd.Env = append(os.Environ(), "PGPASSWORD="+dbCfg.Password)
|
||||
} else {
|
||||
cmd.Env = os.Environ()
|
||||
}
|
||||
|
||||
fileName := fmt.Sprintf("wavelet_%s.sql", time.Now().Format("20060102_150405"))
|
||||
return cmd, fileName, nil
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package repository_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/redis/go-redis/v9/maintnotifications"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/glebarez/sqlite"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"Wavelet/core"
|
||||
"Wavelet/core/contracts"
|
||||
"Wavelet/plugins/domain/admin/model"
|
||||
"Wavelet/plugins/domain/admin/repository"
|
||||
cacheplugin "Wavelet/plugins/infra/cache"
|
||||
)
|
||||
|
||||
// stubDBService 用内存 SQLite 满足 DBService 契约,隔离外部依赖。
|
||||
type stubDBService struct{ db *gorm.DB }
|
||||
|
||||
func (s stubDBService) GORM() *gorm.DB { return s.db }
|
||||
|
||||
func (s stubDBService) DB(context.Context) *gorm.DB { return s.db }
|
||||
|
||||
func (s stubDBService) Named(string) *gorm.DB { return s.db }
|
||||
|
||||
// newFlushLogTestCache 构建真实多层缓存服务并注入 admin 插件上下文。
|
||||
func newFlushLogTestCache(t *testing.T) (contracts.CacheService, *miniredis.Miniredis, func()) {
|
||||
t.Helper()
|
||||
mr, err := miniredis.Run()
|
||||
require.NoError(t, err)
|
||||
rdb := redis.NewClient(&redis.Options{Addr: mr.Addr(), MaintNotificationsConfig: &maintnotifications.Config{Mode: maintnotifications.ModeDisabled}})
|
||||
|
||||
p := cacheplugin.New(cacheplugin.WithRedis(rdb), cacheplugin.WithRAMCapacity(64))
|
||||
ctx := core.NewContext(context.Background())
|
||||
require.NoError(t, p.Apply(ctx))
|
||||
svc, err := core.Inject[contracts.CacheService](ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
repository.SetCacheService(svc)
|
||||
cleanup := func() {
|
||||
repository.SetCacheService(nil)
|
||||
_ = rdb.Close()
|
||||
mr.Close()
|
||||
}
|
||||
return svc, mr, cleanup
|
||||
}
|
||||
|
||||
// TestFlushTaskExecutionLogPropagatesCacheError 回归:缓存读取失败(非未命中)时,
|
||||
// FlushTaskExecutionLog 必须返回错误而不是静默吞掉日志并误报成功(nilerr 修复)。
|
||||
func TestFlushTaskExecutionLogPropagatesCacheError(t *testing.T) {
|
||||
_, mr, cleanup := newFlushLogTestCache(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := context.Background()
|
||||
const taskID = "flush-err-task"
|
||||
|
||||
// 先缓冲一行日志
|
||||
require.NoError(t, repository.AppendTaskExecutionLog(ctx, taskID, "step-1 ok"))
|
||||
|
||||
// 关闭 miniredis 模拟缓存基础设施故障(读取出错而非未命中)
|
||||
mr.Close()
|
||||
|
||||
err := repository.FlushTaskExecutionLog(ctx, taskID)
|
||||
assert.Error(t, err, "缓存故障时必须返回错误,防止缓冲日志被静默丢弃")
|
||||
}
|
||||
|
||||
// TestFlushTaskExecutionLogCacheMissIsNoop 回归:任务无缓冲日志(未命中)时应为空操作成功。
|
||||
func TestFlushTaskExecutionLogCacheMissIsNoop(t *testing.T) {
|
||||
_, _, cleanup := newFlushLogTestCache(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := context.Background()
|
||||
assert.NoError(t, repository.FlushTaskExecutionLog(ctx, "missing-task"))
|
||||
}
|
||||
|
||||
// TestFlushTaskExecutionLogPersistsAndClears 验证正常路径:缓冲日志写入执行记录后清理缓存。
|
||||
func TestFlushTaskExecutionLogPersistsAndClears(t *testing.T) {
|
||||
svc, _, cleanup := newFlushLogTestCache(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := context.Background()
|
||||
const taskID = "flush-ok-task"
|
||||
require.NoError(t, repository.AppendTaskExecutionLog(ctx, taskID, "done"))
|
||||
|
||||
sqliteDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, sqliteDB.AutoMigrate(&model.TaskExecution{}))
|
||||
repository.SetDBService(stubDBService{db: sqliteDB})
|
||||
defer repository.SetDBService(nil)
|
||||
gormDB := sqliteDB
|
||||
exec := &model.TaskExecution{TaskID: taskID, TaskType: "upload:test", TaskName: "t", Status: model.TaskExecutionStatusSucceeded}
|
||||
require.NoError(t, gormDB.Create(exec).Error)
|
||||
|
||||
require.NoError(t, repository.FlushTaskExecutionLog(ctx, taskID))
|
||||
|
||||
var got model.TaskExecution
|
||||
require.NoError(t, gormDB.First(&got, exec.ID).Error)
|
||||
assert.Contains(t, got.Log, "done")
|
||||
|
||||
// 缓存中的缓冲日志应已被清理
|
||||
var buf string
|
||||
err = svc.Get(ctx, repository.TaskExecutionLogRedisKey(taskID), &buf)
|
||||
assert.True(t, errors.Is(err, contracts.ErrCacheMiss), "flush 后缓存应清空, got %v", err)
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"Wavelet/pkg/util"
|
||||
"context"
|
||||
)
|
||||
|
||||
// UserDisplayName is the minimal user projection needed to decorate access log rows.
|
||||
type UserDisplayName struct {
|
||||
Username string
|
||||
Nickname string
|
||||
}
|
||||
|
||||
// SearchUserIDsByUsername is the database fallback used when the user contract is absent.
|
||||
func SearchUserIDsByUsername(ctx context.Context, username string) ([]uint64, error) {
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
var ids []uint64
|
||||
if err := gormDB.Table("w_users").
|
||||
Where("username LIKE ? ESCAPE '\\'", "%"+util.EscapeLike(username)+"%").
|
||||
Pluck("id", &ids).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
// LoadUserDisplayNames resolves usernames and nicknames for the given ids.
|
||||
func LoadUserDisplayNames(ctx context.Context, userIDs []uint64) (map[uint64]UserDisplayName, error) {
|
||||
result := make(map[uint64]UserDisplayName, len(userIDs))
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil || len(userIDs) == 0 {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
var users []struct {
|
||||
ID uint64
|
||||
Username string
|
||||
Nickname string
|
||||
}
|
||||
if err := gormDB.Table("w_users").Where("id IN ?", userIDs).Find(&users).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, u := range users {
|
||||
result[u.ID] = UserDisplayName{Username: u.Username, Nickname: u.Nickname}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -0,0 +1,421 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
// Package repository provides persistence operations for the admin domain.
|
||||
package repository
|
||||
|
||||
import (
|
||||
"Wavelet/core/contracts"
|
||||
"Wavelet/pkg/cache/ram"
|
||||
"Wavelet/pkg/logger"
|
||||
"Wavelet/plugins/domain/admin/errs"
|
||||
"Wavelet/plugins/domain/admin/model"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/shopspring/decimal"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
configTypeSystem = "system"
|
||||
)
|
||||
|
||||
var (
|
||||
repoMu sync.RWMutex
|
||||
dbService contracts.DBService
|
||||
cacheService contracts.CacheService
|
||||
)
|
||||
|
||||
// SetDBService injects the DBService contract.
|
||||
func SetDBService(s contracts.DBService) {
|
||||
repoMu.Lock()
|
||||
defer repoMu.Unlock()
|
||||
dbService = s
|
||||
}
|
||||
|
||||
// SetCacheService injects the CacheService contract.
|
||||
func SetCacheService(s contracts.CacheService) {
|
||||
repoMu.Lock()
|
||||
defer repoMu.Unlock()
|
||||
cacheService = s
|
||||
}
|
||||
|
||||
// ResetServices clears injected persistence services.
|
||||
func ResetServices() {
|
||||
repoMu.Lock()
|
||||
defer repoMu.Unlock()
|
||||
dbService = nil
|
||||
cacheService = nil
|
||||
}
|
||||
|
||||
// GetDB returns the GORM DB instance bound to the context if available.
|
||||
func GetDB(ctx context.Context) *gorm.DB {
|
||||
repoMu.RLock()
|
||||
defer repoMu.RUnlock()
|
||||
if dbService == nil {
|
||||
return nil
|
||||
}
|
||||
return dbService.DB(ctx)
|
||||
}
|
||||
|
||||
// GetCache returns the unified CacheService instance.
|
||||
func GetCache(_ context.Context) contracts.CacheService {
|
||||
repoMu.RLock()
|
||||
defer repoMu.RUnlock()
|
||||
return cacheService
|
||||
}
|
||||
|
||||
// PreheatSystemConfigs loads all system configs from database.
|
||||
func PreheatSystemConfigs(ctx context.Context) ([]model.SystemConfig, error) {
|
||||
database := GetDB(ctx)
|
||||
if database == nil {
|
||||
return nil, errors.New(errs.ErrDatabaseNotInitialized)
|
||||
}
|
||||
|
||||
var configs []model.SystemConfig
|
||||
if err := database.Find(&configs).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return configs, nil
|
||||
}
|
||||
|
||||
// PreheatSystemConfigByKey loads a single config key from database.
|
||||
func PreheatSystemConfigByKey(ctx context.Context, key string) (model.SystemConfig, error) {
|
||||
database := GetDB(ctx)
|
||||
if database == nil {
|
||||
return model.SystemConfig{}, errors.New(errs.ErrDatabaseNotInitialized)
|
||||
}
|
||||
|
||||
var sc model.SystemConfig
|
||||
if err := database.Where("key = ?", key).First(&sc).Error; err != nil {
|
||||
return model.SystemConfig{}, err
|
||||
}
|
||||
return sc, nil
|
||||
}
|
||||
|
||||
// GetSystemConfigByGroup queries a configuration by Type and Key.
|
||||
func GetSystemConfigByGroup(ctx context.Context, configType, key string) (model.SystemConfig, error) {
|
||||
ensureSystemConfigCacheListener()
|
||||
|
||||
if item, ok := ram.Get(configType, key); ok {
|
||||
var sc model.SystemConfig
|
||||
if err := json.Unmarshal([]byte(item.Value), &sc); err == nil {
|
||||
return sc, nil
|
||||
}
|
||||
}
|
||||
|
||||
database := GetDB(ctx)
|
||||
if database == nil {
|
||||
return model.SystemConfig{}, errors.New(errs.ErrDatabaseNotInitialized)
|
||||
}
|
||||
|
||||
var sc model.SystemConfig
|
||||
if err := database.Where("key = ?", key).First(&sc).Error; err != nil {
|
||||
return model.SystemConfig{}, err
|
||||
}
|
||||
|
||||
valBytes, err := json.Marshal(sc)
|
||||
if err == nil {
|
||||
ram.Set(ram.CacheItem{
|
||||
Key: sc.Key,
|
||||
Value: string(valBytes),
|
||||
Type: configType,
|
||||
TTL: determineTTL(sc.Key),
|
||||
})
|
||||
}
|
||||
|
||||
return sc, nil
|
||||
}
|
||||
|
||||
// GetSystemConfigByKey queries config by key.
|
||||
func GetSystemConfigByKey(ctx context.Context, key string) (model.SystemConfig, error) {
|
||||
return GetSystemConfigByGroup(ctx, ConfigCacheType, key)
|
||||
}
|
||||
|
||||
// ListSystemConfigsByKeys loads multiple config keys.
|
||||
func ListSystemConfigsByKeys(ctx context.Context, keys []string) (map[string]model.SystemConfig, error) {
|
||||
if len(keys) == 0 {
|
||||
return map[string]model.SystemConfig{}, nil
|
||||
}
|
||||
|
||||
ensureSystemConfigCacheListener()
|
||||
|
||||
result := make(map[string]model.SystemConfig, len(keys))
|
||||
missing := make([]string, 0, len(keys))
|
||||
|
||||
for _, key := range keys {
|
||||
if item, ok := ram.Get(ConfigCacheType, key); ok {
|
||||
var sc model.SystemConfig
|
||||
if err := json.Unmarshal([]byte(item.Value), &sc); err == nil {
|
||||
result[key] = sc
|
||||
continue
|
||||
}
|
||||
}
|
||||
missing = append(missing, key)
|
||||
}
|
||||
|
||||
if len(missing) == 0 {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
database := GetDB(ctx)
|
||||
if database == nil {
|
||||
return nil, errors.New(errs.ErrDatabaseNotInitialized)
|
||||
}
|
||||
|
||||
var configs []model.SystemConfig
|
||||
if err := database.Where("key IN ?", missing).Find(&configs).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for i := range configs {
|
||||
valBytes, err := json.Marshal(configs[i])
|
||||
if err == nil {
|
||||
ram.Set(ram.CacheItem{
|
||||
Key: configs[i].Key,
|
||||
Value: string(valBytes),
|
||||
Type: ConfigCacheType,
|
||||
TTL: determineTTL(configs[i].Key),
|
||||
})
|
||||
}
|
||||
result[configs[i].Key] = configs[i]
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// InvalidateVisibleSystemConfigsCache clears the cached public config list.
|
||||
func InvalidateVisibleSystemConfigsCache(ctx context.Context) error {
|
||||
return InvalidateAllSystemConfigCaches(ctx)
|
||||
}
|
||||
|
||||
// ListVisibleSystemConfigs queries visible configs using local cache store.
|
||||
func ListVisibleSystemConfigs(ctx context.Context) ([]model.SystemConfig, error) {
|
||||
ensureSystemConfigCacheListener()
|
||||
|
||||
items := ram.GetTypeItems(ConfigCacheType)
|
||||
if len(items) > 0 {
|
||||
var list []model.SystemConfig
|
||||
for _, item := range items {
|
||||
var sc model.SystemConfig
|
||||
if err := json.Unmarshal([]byte(item.Value), &sc); err == nil {
|
||||
if sc.Visibility == model.ConfigVisibilityVisible {
|
||||
list = append(list, sc)
|
||||
}
|
||||
}
|
||||
}
|
||||
return list, nil
|
||||
}
|
||||
|
||||
database := GetDB(ctx)
|
||||
if database == nil {
|
||||
return nil, errors.New(errs.ErrDatabaseNotInitialized)
|
||||
}
|
||||
|
||||
var configs []model.SystemConfig
|
||||
if err := database.Where("visibility = ?", model.ConfigVisibilityVisible).Find(&configs).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, cfg := range configs {
|
||||
valBytes, err := json.Marshal(cfg)
|
||||
if err == nil {
|
||||
ram.Set(ram.CacheItem{
|
||||
Key: cfg.Key,
|
||||
Value: string(valBytes),
|
||||
Type: ConfigCacheType,
|
||||
TTL: determineTTL(cfg.Key),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return configs, nil
|
||||
}
|
||||
|
||||
// GetIntByKey queries config and converts to int.
|
||||
func GetIntByKey(ctx context.Context, key string) (int, error) {
|
||||
sc, err := GetSystemConfigByKey(ctx, key)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
value, err := strconv.Atoi(sc.Value)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf(errs.ErrConfigIntParseFailed, key, sc.Value, err)
|
||||
}
|
||||
|
||||
return value, nil
|
||||
}
|
||||
|
||||
// GetDecimalByKey queries config and converts to decimal.Decimal.
|
||||
func GetDecimalByKey(ctx context.Context, key string, precision int32) (decimal.Decimal, error) {
|
||||
sc, err := GetSystemConfigByKey(ctx, key)
|
||||
if err != nil {
|
||||
return decimal.Zero, err
|
||||
}
|
||||
|
||||
value, err := decimal.NewFromString(sc.Value)
|
||||
if err != nil {
|
||||
return decimal.Zero, fmt.Errorf(errs.ErrConfigDecimalParseFailed, key, sc.Value, err)
|
||||
}
|
||||
|
||||
return value.Truncate(precision), nil
|
||||
}
|
||||
|
||||
// GetBoolByKey queries config and converts to bool.
|
||||
func GetBoolByKey(ctx context.Context, key string) (bool, error) {
|
||||
sc, err := GetSystemConfigByKey(ctx, key)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
value, err := strconv.ParseBool(sc.Value)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf(errs.ErrConfigBoolParseFailed, key, sc.Value, err)
|
||||
}
|
||||
|
||||
return value, nil
|
||||
}
|
||||
|
||||
// GetMenuDisplayConfig queries and parses menu config.
|
||||
func GetMenuDisplayConfig(ctx context.Context) (map[string]bool, error) {
|
||||
sc, err := GetSystemConfigByKey(ctx, model.ConfigKeyMenuDisplayConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
config := make(map[string]bool)
|
||||
if sc.Value == "" || sc.Value == "{}" {
|
||||
return config, nil
|
||||
}
|
||||
|
||||
if err := json.Unmarshal([]byte(sc.Value), &config); err != nil {
|
||||
return nil, fmt.Errorf(errs.ErrParseMenuDisplayConfigFailed, err)
|
||||
}
|
||||
|
||||
return config, nil
|
||||
}
|
||||
|
||||
// ListAdminSystemConfigs returns all configs, optionally filtered by type.
|
||||
func ListAdminSystemConfigs(ctx context.Context, configType string) ([]model.SystemConfig, error) {
|
||||
query := GetDB(ctx).Order("created_at DESC")
|
||||
if configType != "" {
|
||||
query = query.Where("type = ?", configType)
|
||||
}
|
||||
var configs []model.SystemConfig
|
||||
if err := query.Find(&configs).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return configs, nil
|
||||
}
|
||||
|
||||
// GetAdminSystemConfigByKey loads a config directly from DB.
|
||||
func GetAdminSystemConfigByKey(ctx context.Context, key string) (model.SystemConfig, error) {
|
||||
var config model.SystemConfig
|
||||
if err := GetDB(ctx).Where("key = ?", key).First(&config).Error; err != nil {
|
||||
return model.SystemConfig{}, err
|
||||
}
|
||||
return config, nil
|
||||
}
|
||||
|
||||
// SystemConfigExists reports whether a config key already exists.
|
||||
func SystemConfigExists(ctx context.Context, key string) (bool, error) {
|
||||
var existing model.SystemConfig
|
||||
err := GetDB(ctx).Where("key = ?", key).First(&existing).Error
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// CreateSystemConfigRecord persists a new system config row.
|
||||
func CreateSystemConfigRecord(ctx context.Context, config *model.SystemConfig) error {
|
||||
return GetDB(ctx).Create(config).Error
|
||||
}
|
||||
|
||||
// UpdateSystemConfigFields applies partial updates to a system config row.
|
||||
func UpdateSystemConfigFields(ctx context.Context, config *model.SystemConfig, updates map[string]any) error {
|
||||
return GetDB(ctx).Model(config).Updates(updates).Error
|
||||
}
|
||||
|
||||
// UpdateSystemConfigTx applies the config row updates inside a transaction and, when
|
||||
// resolveTaskType is not empty, marks that task type's failed executions as succeeded
|
||||
// within the same transaction.
|
||||
func UpdateSystemConfigTx(
|
||||
ctx context.Context,
|
||||
config *model.SystemConfig,
|
||||
updates map[string]any,
|
||||
resolveTaskType string,
|
||||
resolveResult string,
|
||||
) error {
|
||||
database := GetDB(ctx)
|
||||
if database == nil {
|
||||
return errors.New(errs.ErrDatabaseServiceNotAvailable)
|
||||
}
|
||||
|
||||
return database.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Model(config).Updates(updates).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if resolveTaskType == "" {
|
||||
return nil
|
||||
}
|
||||
if err := MarkFailedTaskExecutionsSucceededTx(tx, resolveTaskType, resolveResult, time.Now()); err != nil {
|
||||
logger.ErrorF(ctx, errs.ErrAutoResolveMigrationTaskFailed, err)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// SaveOrUpdateSystemConfig creates or updates a config row and invalidates cache.
|
||||
func SaveOrUpdateSystemConfig(ctx context.Context, key, value string) error {
|
||||
var sc model.SystemConfig
|
||||
err := GetDB(ctx).Where("key = ?", key).First(&sc).Error
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return err
|
||||
}
|
||||
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
sc = model.SystemConfig{
|
||||
Key: key,
|
||||
Value: value,
|
||||
Type: configTypeSystem,
|
||||
Visibility: model.ConfigVisibilityHidden,
|
||||
}
|
||||
if err := GetDB(ctx).Create(&sc).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
sc.Value = value
|
||||
if err := GetDB(ctx).Save(&sc).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return InvalidateSystemConfigCache(ctx, key)
|
||||
}
|
||||
|
||||
// CountActiveUploads counts non-deleted rows of the storage upload table. A missing
|
||||
// database handle yields zero, matching the pre-refactor guard behaviour.
|
||||
func CountActiveUploads(ctx context.Context) (int64, error) {
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
var uploadCount int64
|
||||
if err := gormDB.Table("w_uploads").
|
||||
Where("status != ?", "deleted").
|
||||
Count(&uploadCount).Error; err != nil {
|
||||
return 0, fmt.Errorf(errs.ErrCheckExistingUploadsFailed, err)
|
||||
}
|
||||
return uploadCount, nil
|
||||
}
|
||||
@@ -0,0 +1,331 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"Wavelet/core/contracts"
|
||||
"Wavelet/pkg/idgen"
|
||||
"Wavelet/pkg/util"
|
||||
"Wavelet/plugins/domain/admin/errs"
|
||||
"Wavelet/plugins/domain/admin/model"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
taskExecutionLogRedisKeyPrefix = "task:execution:log:"
|
||||
taskExecutionLogExpiration = 24 * time.Hour
|
||||
)
|
||||
|
||||
// CreateScheduleRecord 创建定时任务
|
||||
func CreateScheduleRecord(ctx context.Context, schedule *model.Schedule) error {
|
||||
return GetDB(ctx).Create(schedule).Error
|
||||
}
|
||||
|
||||
// UpdateScheduleRecord 更新定时任务
|
||||
func UpdateScheduleRecord(ctx context.Context, schedule *model.Schedule) error {
|
||||
return GetDB(ctx).Save(schedule).Error
|
||||
}
|
||||
|
||||
// DeleteScheduleRecord 删除定时任务
|
||||
func DeleteScheduleRecord(ctx context.Context, id uint64) error {
|
||||
return GetDB(ctx).Delete(&model.Schedule{}, id).Error
|
||||
}
|
||||
|
||||
// GetScheduleByID 根据 ID 获取定时任务
|
||||
func GetScheduleByID(ctx context.Context, id uint64) (*model.Schedule, error) {
|
||||
var schedule model.Schedule
|
||||
if err := GetDB(ctx).Where("id = ?", id).First(&schedule).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &schedule, nil
|
||||
}
|
||||
|
||||
// ListSchedulesRecord 获取所有定时任务
|
||||
func ListSchedulesRecord(ctx context.Context) ([]model.Schedule, error) {
|
||||
var schedules []model.Schedule
|
||||
if err := GetDB(ctx).Order("id DESC").Find(&schedules).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return schedules, nil
|
||||
}
|
||||
|
||||
// ListActiveSchedules 获取所有启用的定时任务
|
||||
func ListActiveSchedules(ctx context.Context) ([]model.Schedule, error) {
|
||||
var schedules []model.Schedule
|
||||
if err := GetDB(ctx).Where("is_active = ?", true).Find(&schedules).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return schedules, nil
|
||||
}
|
||||
|
||||
// CreateTaskExecutionRecord 创建任务执行记录
|
||||
func CreateTaskExecutionRecord(ctx context.Context, execution *model.TaskExecution) error {
|
||||
execution.ID = idgen.NextUint64ID()
|
||||
return GetDB(ctx).Create(execution).Error
|
||||
}
|
||||
|
||||
// UpdateTaskExecutionRecord 更新任务执行记录,忽略由 Redis 缓冲和归档流程管理的 log 字段。
|
||||
func UpdateTaskExecutionRecord(ctx context.Context, execution *model.TaskExecution) error {
|
||||
return GetDB(ctx).Omit("log").Save(execution).Error
|
||||
}
|
||||
|
||||
// GetTaskExecutionByTaskID 根据 TaskID 获取执行记录
|
||||
func GetTaskExecutionByTaskID(ctx context.Context, taskID string) (*model.TaskExecution, error) {
|
||||
var execution model.TaskExecution
|
||||
if err := GetDB(ctx).Where("task_id = ?", taskID).First(&execution).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := loadTaskExecutionLog(ctx, &execution); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &execution, nil
|
||||
}
|
||||
|
||||
// GetTaskExecutionByID 根据 ID 获取执行记录
|
||||
func GetTaskExecutionByID(ctx context.Context, id uint64) (*model.TaskExecution, error) {
|
||||
var execution model.TaskExecution
|
||||
if err := GetDB(ctx).Where("id = ?", id).First(&execution).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := loadTaskExecutionLog(ctx, &execution); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &execution, nil
|
||||
}
|
||||
|
||||
// GetLatestTaskExecutionByTaskType returns the most recent execution for a task type.
|
||||
func GetLatestTaskExecutionByTaskType(ctx context.Context, taskType string) (*model.TaskExecution, bool, error) {
|
||||
var execution model.TaskExecution
|
||||
err := GetDB(ctx).
|
||||
Where("task_type = ?", taskType).
|
||||
Order("id DESC").
|
||||
First(&execution).Error
|
||||
if err == nil {
|
||||
if loadErr := loadTaskExecutionLog(ctx, &execution); loadErr != nil {
|
||||
return nil, false, loadErr
|
||||
}
|
||||
return &execution, true, nil
|
||||
}
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, false, nil
|
||||
}
|
||||
return nil, false, err
|
||||
}
|
||||
|
||||
// AppendTaskExecutionLog 将日志追加到缓冲,任务完成后再持久化到数据库。
|
||||
func AppendTaskExecutionLog(ctx context.Context, taskID, logLine string) error {
|
||||
cacheSvc := GetCache(ctx)
|
||||
if cacheSvc == nil {
|
||||
return errors.New(errs.ErrCacheServiceNotInitialized)
|
||||
}
|
||||
|
||||
now := time.Now().Format("15:04:05")
|
||||
line := fmt.Sprintf("[%s] %s\n", now, logLine)
|
||||
key := TaskExecutionLogRedisKey(taskID)
|
||||
|
||||
var existing string
|
||||
_ = cacheSvc.Get(ctx, key, &existing)
|
||||
return cacheSvc.Set(ctx, key, existing+line, taskExecutionLogExpiration)
|
||||
}
|
||||
|
||||
// FlushTaskExecutionLog 将缓冲中的完整任务日志写入数据库,并在成功后清理缓存。
|
||||
func FlushTaskExecutionLog(ctx context.Context, taskID string) error {
|
||||
cacheSvc := GetCache(ctx)
|
||||
if cacheSvc == nil {
|
||||
return errors.New(errs.ErrCacheServiceNotInitialized)
|
||||
}
|
||||
|
||||
key := TaskExecutionLogRedisKey(taskID)
|
||||
var logText string
|
||||
if err := cacheSvc.Get(ctx, key, &logText); err != nil {
|
||||
// 缓存未命中属于正常情况(任务无输出),其余错误必须上抛,
|
||||
// 否则缓冲日志会被静默丢弃并误报持久化成功。
|
||||
if !errors.Is(err, contracts.ErrCacheMiss) {
|
||||
return fmt.Errorf("load buffered task execution log: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if logText == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
gormDB := GetDB(ctx)
|
||||
if gormDB == nil {
|
||||
return errors.New(errs.ErrDatabaseNotInitialized)
|
||||
}
|
||||
result := gormDB.Model(&model.TaskExecution{}).
|
||||
Where("task_id = ?", taskID).
|
||||
Update("log", logText)
|
||||
if result.Error != nil {
|
||||
return fmt.Errorf("persist task execution log: %w", result.Error)
|
||||
}
|
||||
if result.RowsAffected == 0 {
|
||||
return fmt.Errorf("persist task execution log: task %q not found", taskID)
|
||||
}
|
||||
|
||||
_ = cacheSvc.Delete(ctx, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ListTaskExecutionRecords 分页查询任务执行记录
|
||||
func ListTaskExecutionRecords(ctx context.Context, req model.ListTaskExecutionsRequest) ([]model.TaskExecution, int64, error) {
|
||||
if req.Page <= 0 {
|
||||
req.Page = 1
|
||||
}
|
||||
if req.PageSize <= 0 {
|
||||
req.PageSize = 20
|
||||
}
|
||||
|
||||
query := GetDB(ctx).Model(&model.TaskExecution{})
|
||||
|
||||
if req.Status != "" {
|
||||
query = query.Where("status = ?", req.Status)
|
||||
}
|
||||
if req.TaskType != "" {
|
||||
query = query.Where("task_type = ?", req.TaskType)
|
||||
} else if types := parseTaskTypesFilter(req.TaskTypes); len(types) > 0 {
|
||||
query = query.Where("task_type IN ?", types)
|
||||
} else if req.TaskTypePrefix != "" {
|
||||
query = query.Where("task_type LIKE ? ESCAPE '\\'", util.EscapeLike(req.TaskTypePrefix)+"%")
|
||||
}
|
||||
|
||||
var total int64
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
var executions []model.TaskExecution
|
||||
offset := (req.Page - 1) * req.PageSize
|
||||
if err := query.Order("id DESC").Offset(offset).Limit(req.PageSize).Find(&executions).Error; err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
if err := loadTaskExecutionLogs(ctx, executions); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
return executions, total, nil
|
||||
}
|
||||
|
||||
func parseTaskTypesFilter(raw string) []string {
|
||||
if strings.TrimSpace(raw) == "" {
|
||||
return nil
|
||||
}
|
||||
parts := strings.Split(raw, ",")
|
||||
out := make([]string, 0, len(parts))
|
||||
for _, part := range parts {
|
||||
part = strings.TrimSpace(part)
|
||||
if part != "" {
|
||||
out = append(out, part)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// MarkFailedTaskExecutionsSucceededTx marks failed executions of a task type as succeeded within a transaction.
|
||||
func MarkFailedTaskExecutionsSucceededTx(
|
||||
tx *gorm.DB,
|
||||
taskType string,
|
||||
result string,
|
||||
finishedAt time.Time,
|
||||
) error {
|
||||
return tx.Model(&model.TaskExecution{}).
|
||||
Where("task_type = ? AND status = ?", taskType, model.TaskExecutionStatusFailed).
|
||||
Updates(map[string]any{
|
||||
"status": model.TaskExecutionStatusSucceeded,
|
||||
"result": result,
|
||||
"finished_at": finishedAt,
|
||||
}).Error
|
||||
}
|
||||
|
||||
// CleanupTaskExecutionLogs removes finished task execution logs according to frequency-based retention.
|
||||
func CleanupTaskExecutionLogs(ctx context.Context, now time.Time) (model.TaskExecutionCleanupStats, error) {
|
||||
const (
|
||||
frequencyWindowDays = 30
|
||||
highFrequencyThreshold = frequencyWindowDays
|
||||
)
|
||||
|
||||
frequencyWindowStart := now.AddDate(0, 0, -frequencyWindowDays)
|
||||
highFrequencyCutoff := now.AddDate(0, 0, -3)
|
||||
lowFrequencyCutoff := now.AddDate(0, 0, -30)
|
||||
terminalStatuses := []model.TaskExecutionStatus{model.TaskExecutionStatusSucceeded, model.TaskExecutionStatusFailed}
|
||||
|
||||
var highFrequencyTaskTypes []string
|
||||
if err := GetDB(ctx).
|
||||
Model(&model.TaskExecution{}).
|
||||
Select("task_type").
|
||||
Where("created_at >= ?", frequencyWindowStart).
|
||||
Group("task_type").
|
||||
Having("COUNT(*) > ?", highFrequencyThreshold).
|
||||
Pluck("task_type", &highFrequencyTaskTypes).Error; err != nil {
|
||||
return model.TaskExecutionCleanupStats{}, fmt.Errorf("query high-frequency task types: %w", err)
|
||||
}
|
||||
|
||||
var highFrequencyDeleted int64
|
||||
if len(highFrequencyTaskTypes) > 0 {
|
||||
highFrequencyResult := GetDB(ctx).
|
||||
Where("status IN ?", terminalStatuses).
|
||||
Where("created_at < ?", highFrequencyCutoff).
|
||||
Where("task_type IN ?", highFrequencyTaskTypes).
|
||||
Delete(&model.TaskExecution{})
|
||||
if highFrequencyResult.Error != nil {
|
||||
return model.TaskExecutionCleanupStats{}, fmt.Errorf("delete high-frequency task execution logs: %w", highFrequencyResult.Error)
|
||||
}
|
||||
highFrequencyDeleted = highFrequencyResult.RowsAffected
|
||||
}
|
||||
|
||||
lowFrequencyQuery := GetDB(ctx).
|
||||
Where("status IN ?", terminalStatuses).
|
||||
Where("created_at < ?", lowFrequencyCutoff)
|
||||
if len(highFrequencyTaskTypes) > 0 {
|
||||
lowFrequencyQuery = lowFrequencyQuery.Where("task_type NOT IN ?", highFrequencyTaskTypes)
|
||||
}
|
||||
lowFrequencyResult := lowFrequencyQuery.Delete(&model.TaskExecution{})
|
||||
if lowFrequencyResult.Error != nil {
|
||||
return model.TaskExecutionCleanupStats{}, fmt.Errorf("delete low-frequency task execution logs: %w", lowFrequencyResult.Error)
|
||||
}
|
||||
|
||||
return model.TaskExecutionCleanupStats{
|
||||
HighFrequencyDeleted: highFrequencyDeleted,
|
||||
LowFrequencyDeleted: lowFrequencyResult.RowsAffected,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// TaskExecutionLogRedisKey builds the Redis key for task execution logs.
|
||||
func TaskExecutionLogRedisKey(taskID string) string {
|
||||
return taskExecutionLogRedisKeyPrefix + taskID
|
||||
}
|
||||
|
||||
func loadTaskExecutionLog(ctx context.Context, execution *model.TaskExecution) error {
|
||||
cacheSvc := GetCache(ctx)
|
||||
if cacheSvc == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
var logText string
|
||||
if err := cacheSvc.Get(ctx, TaskExecutionLogRedisKey(execution.TaskID), &logText); err == nil && logText != "" {
|
||||
execution.Log = logText
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadTaskExecutionLogs(ctx context.Context, executions []model.TaskExecution) error {
|
||||
cacheSvc := GetCache(ctx)
|
||||
if cacheSvc == nil || len(executions) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
for i := range executions {
|
||||
var logText string
|
||||
if err := cacheSvc.Get(ctx, TaskExecutionLogRedisKey(executions[i].TaskID), &logText); err == nil && logText != "" {
|
||||
executions[i].Log = logText
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"Wavelet/plugins/domain/admin/model"
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// ListTemplatesRecord returns all templates ordered by system flag and creation time.
|
||||
func ListTemplatesRecord(ctx context.Context) ([]model.Template, error) {
|
||||
var templates []model.Template
|
||||
if err := GetDB(ctx).Order("is_system DESC, created_at DESC").Find(&templates).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return templates, nil
|
||||
}
|
||||
|
||||
// GetTemplateByKey loads a template by its key.
|
||||
func GetTemplateByKey(ctx context.Context, key string) (model.Template, error) {
|
||||
var tmpl model.Template
|
||||
if err := GetDB(ctx).Where("key = ?", key).First(&tmpl).Error; err != nil {
|
||||
return model.Template{}, err
|
||||
}
|
||||
return tmpl, nil
|
||||
}
|
||||
|
||||
// TemplateExistsByKey reports whether a template key is already taken.
|
||||
func TemplateExistsByKey(ctx context.Context, key string) (bool, error) {
|
||||
var existing model.Template
|
||||
err := GetDB(ctx).Where("key = ?", key).First(&existing).Error
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// CreateTemplateRecord persists a new template.
|
||||
func CreateTemplateRecord(ctx context.Context, tmpl *model.Template) error {
|
||||
return GetDB(ctx).Create(tmpl).Error
|
||||
}
|
||||
|
||||
// SaveTemplateRecord updates an existing template.
|
||||
func SaveTemplateRecord(ctx context.Context, tmpl *model.Template) error {
|
||||
return GetDB(ctx).Save(tmpl).Error
|
||||
}
|
||||
|
||||
// DeleteTemplateRecord removes a template record.
|
||||
func DeleteTemplateRecord(ctx context.Context, tmpl *model.Template) error {
|
||||
return GetDB(ctx).Delete(tmpl).Error
|
||||
}
|
||||
Reference in New Issue
Block a user