优化
This commit is contained in:
@@ -125,7 +125,7 @@ func (r *authorityAccessRepo) CopyAuthority(ctx context.Context, sourceID uint,
|
||||
}
|
||||
parentAllowed := value.ParentID != nil && (allowed[*value.ParentID] || *value.ParentID == actor.AuthorityID)
|
||||
if !parentAllowed {
|
||||
return errors.New("严格角色模式下只能在自己的角色树中复制角色")
|
||||
return errors.New("您提交的角色ID不合法")
|
||||
}
|
||||
}
|
||||
var count int64
|
||||
@@ -168,7 +168,7 @@ func (r *authorityAccessRepo) CopyAuthority(ctx context.Context, sourceID uint,
|
||||
return err
|
||||
}
|
||||
if !allowed {
|
||||
return errors.New("严格角色模式下不能复制当前角色未拥有的 API")
|
||||
return errors.New("存在api不在权限列表中")
|
||||
}
|
||||
}
|
||||
key := api.V1 + "\x00" + api.V2
|
||||
|
||||
@@ -34,7 +34,7 @@ func newPolicyRule(authorityID uint, path, method string) casbinRulePO {
|
||||
}
|
||||
|
||||
func policyScope(db *gorm.DB) *gorm.DB {
|
||||
return db.Where("ptype = ?", "p")
|
||||
return db.Session(&gorm.Session{NewDB: true}).Model(&casbinRulePO{}).Where("ptype = ?", "p")
|
||||
}
|
||||
|
||||
func deletePoliciesForAuthority(db *gorm.DB, authorityID uint) error {
|
||||
|
||||
@@ -260,7 +260,7 @@ func (d *Data) reloadConfig(ctx context.Context) error {
|
||||
return fmt.Errorf("data.database and admin configuration are required")
|
||||
}
|
||||
next.Admin.ConfigPath = configPath
|
||||
candidateDB, err := openDatabase(next.Data.Database, false, "")
|
||||
candidateDB, err := openDatabase(next.Data.Database, false, "", d.logger())
|
||||
if err != nil {
|
||||
return fmt.Errorf("reload database: %w", err)
|
||||
}
|
||||
@@ -287,21 +287,27 @@ func (d *Data) reloadConfig(ctx context.Context) error {
|
||||
return fmt.Errorf("reload storage: %w", err)
|
||||
}
|
||||
useRedis := next.Admin.System != nil && next.Admin.System.UseRedis
|
||||
candidateRedis := openRedis(next.Data.Redis, useRedis)
|
||||
candidateRedis := openRedis(next.Data.Redis, useRedis, d.logger())
|
||||
useMongo := next.Admin.System != nil && next.Admin.System.UseMongo
|
||||
candidateMongo, mongoErr := openMongo(next.Data.Mongo, useMongo)
|
||||
if mongoErr != nil {
|
||||
d.logger().Error("mongo unavailable during configuration reload", "mod", "mongo", "error", mongoErr)
|
||||
}
|
||||
mongoAccepted := false
|
||||
defer func() {
|
||||
if !mongoAccepted && candidateMongo != nil {
|
||||
_ = candidateMongo.Disconnect(context.Background())
|
||||
}
|
||||
}()
|
||||
candidateDBList, err := openDatabaseList(next.Data.DatabaseList)
|
||||
candidateDBList, err := openDatabaseList(next.Data.DatabaseList, d.logger())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
d.gormDB.replace(candidateDB)
|
||||
d.gormDB.replace(candidateDB, d.enqueueDataScopeAudit)
|
||||
for _, item := range candidateDBList {
|
||||
registerDataScopeCallbacks(item, d.enqueueDataScopeAudit)
|
||||
}
|
||||
d.replaceDatabaseList(candidateDBList)
|
||||
d.redis.replace(candidateRedis)
|
||||
if mongoErr == nil {
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package data
|
||||
|
||||
import (
|
||||
"log"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -13,22 +12,23 @@ import (
|
||||
// /system/reloadSystem remains responsible
|
||||
// for rebuilding database, Redis, storage, and scheduled tasks.
|
||||
func (d *Data) watchConfig() func() {
|
||||
logger := d.logger()
|
||||
configPath := d.runtime.ConfigPath()
|
||||
if configPath == "" {
|
||||
return func() {}
|
||||
}
|
||||
absolute, err := filepath.Abs(configPath)
|
||||
if err != nil {
|
||||
log.Printf("resolve config watch path: %v", err)
|
||||
logger.Error("resolve config watch path", "mod", "system", "error", err)
|
||||
return func() {}
|
||||
}
|
||||
watcher, err := fsnotify.NewWatcher()
|
||||
if err != nil {
|
||||
log.Printf("create config watcher: %v", err)
|
||||
logger.Error("create config watcher", "mod", "system", "error", err)
|
||||
return func() {}
|
||||
}
|
||||
if err = watcher.Add(filepath.Dir(absolute)); err != nil {
|
||||
log.Printf("watch config directory: %v", err)
|
||||
logger.Error("watch config directory", "mod", "system", "error", err)
|
||||
_ = watcher.Close()
|
||||
return func() {}
|
||||
}
|
||||
@@ -48,20 +48,20 @@ func (d *Data) watchConfig() func() {
|
||||
timer = time.AfterFunc(100*time.Millisecond, func() {
|
||||
next, loadErr := readBootstrap(absolute)
|
||||
if loadErr != nil {
|
||||
log.Printf("reload changed config: %v", loadErr)
|
||||
logger.Error("reload changed config", "mod", "system", "error", loadErr)
|
||||
return
|
||||
}
|
||||
if next.Data == nil || next.Admin == nil {
|
||||
log.Printf("reload changed config: data and admin configuration are required")
|
||||
logger.Error("reload changed config: data and admin configuration are required", "mod", "system")
|
||||
return
|
||||
}
|
||||
next.Admin.ConfigPath = absolute
|
||||
d.runtime.Replace(next.Data, next.Admin)
|
||||
log.Printf("config file changed: %s", absolute)
|
||||
logger.Info("config file changed", "mod", "system", "path", absolute)
|
||||
})
|
||||
case watchErr, ok := <-watcher.Errors:
|
||||
if ok {
|
||||
log.Printf("config watcher error: %v", watchErr)
|
||||
logger.Error("config watcher error", "mod", "system", "error", watchErr)
|
||||
}
|
||||
case <-done:
|
||||
if timer != nil {
|
||||
|
||||
+46
-25
@@ -3,7 +3,7 @@ package data
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -16,24 +16,33 @@ import (
|
||||
var ProviderSet = wire.NewSet(NewData, NewRuntimeSettings, NewTokenIssuer, NewUserRepo, NewInitializationRepo, NewAuthorityAccessRepo, NewAPIRepo, NewPermissionRepo, NewMenuRepo, NewDepartmentRepo, NewPositionRepo, NewDictionaryRepo, NewParameterRepo, NewAPITokenRepo, NewSecurityRepo, NewVersionRepo, NewExportRepo, NewAuditRepo, NewAuditRecorderRepo, NewLogFileRepo, NewTaskRepo, NewMediaRepo, NewAnnouncementRepo, NewEmailRepo, NewCache, NewFileStorage)
|
||||
|
||||
type Data struct {
|
||||
initMu sync.Mutex
|
||||
configMu sync.Mutex
|
||||
gormDB *reloadableDB
|
||||
redis *reloadableRedis
|
||||
mongo *reloadableMongo
|
||||
runtime *conf.Runtime
|
||||
storage *reloadableStorage
|
||||
dbListMu sync.RWMutex
|
||||
dbList map[string]*gorm.DB
|
||||
initMu sync.Mutex
|
||||
configMu sync.Mutex
|
||||
gormDB *reloadableDB
|
||||
redis *reloadableRedis
|
||||
mongo *reloadableMongo
|
||||
runtime *conf.Runtime
|
||||
storage *reloadableStorage
|
||||
dbListMu sync.RWMutex
|
||||
dbList map[string]*gorm.DB
|
||||
appLogger *slog.Logger
|
||||
auditLog *dataScopeAuditWriter
|
||||
}
|
||||
|
||||
func openDatabaseList(configs []*conf.Data_Database) (map[string]*gorm.DB, error) {
|
||||
func (d *Data) logger() *slog.Logger {
|
||||
if d != nil && d.appLogger != nil {
|
||||
return d.appLogger
|
||||
}
|
||||
return slog.Default()
|
||||
}
|
||||
|
||||
func openDatabaseList(configs []*conf.Data_Database, appLogger ...*slog.Logger) (map[string]*gorm.DB, error) {
|
||||
items := make(map[string]*gorm.DB)
|
||||
for _, config := range configs {
|
||||
if config == nil || config.Disable || config.AliasName == "" {
|
||||
continue
|
||||
}
|
||||
db, err := openDatabase(config, false, "")
|
||||
db, err := openDatabase(config, false, "", appLogger...)
|
||||
if err != nil {
|
||||
for _, opened := range items {
|
||||
if sqlDB, dbErr := opened.DB(); dbErr == nil {
|
||||
@@ -42,7 +51,6 @@ func openDatabaseList(configs []*conf.Data_Database) (map[string]*gorm.DB, error
|
||||
}
|
||||
return nil, fmt.Errorf("open database %q: %w", config.AliasName, err)
|
||||
}
|
||||
registerDataScopeCallbacks(db)
|
||||
items[config.AliasName] = db
|
||||
}
|
||||
return items, nil
|
||||
@@ -77,30 +85,38 @@ func (d *Data) database(name string) (*gorm.DB, error) {
|
||||
return db, nil
|
||||
}
|
||||
|
||||
func NewData(runtime *conf.Runtime) (*Data, func(), error) {
|
||||
func NewData(runtime *conf.Runtime, appLogger *slog.Logger) (*Data, func(), error) {
|
||||
if appLogger == nil {
|
||||
appLogger = slog.Default()
|
||||
}
|
||||
c := runtime.Data()
|
||||
if c == nil || c.Database == nil {
|
||||
return nil, nil, fmt.Errorf("database configuration is required")
|
||||
}
|
||||
d := &Data{runtime: runtime}
|
||||
db, err := openDatabase(c.Database, false, "")
|
||||
d := &Data{runtime: runtime, appLogger: appLogger}
|
||||
db, err := openDatabase(c.Database, false, "", appLogger)
|
||||
usingFallback := false
|
||||
if err != nil {
|
||||
// The initialization endpoint must remain available when the configured
|
||||
// target database has not been created yet.
|
||||
log.Printf("configured database unavailable before initialization: %v", err)
|
||||
db, err = openFallbackDatabase()
|
||||
appLogger.Warn("configured database unavailable before initialization", "mod", "system", "error", err)
|
||||
db, err = openFallbackDatabase(appLogger)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("open bootstrap database: %w", err)
|
||||
}
|
||||
usingFallback = true
|
||||
}
|
||||
d.gormDB = newReloadableDB(db)
|
||||
d.dbList, err = openDatabaseList(c.DatabaseList)
|
||||
d.gormDB = newReloadableDB(db, d.enqueueDataScopeAudit)
|
||||
d.auditLog = newDataScopeAuditWriter(d, appLogger)
|
||||
d.dbList, err = openDatabaseList(c.DatabaseList, appLogger)
|
||||
if err != nil {
|
||||
d.auditLog.Close()
|
||||
d.gormDB.close()
|
||||
return nil, nil, err
|
||||
}
|
||||
for _, item := range d.dbList {
|
||||
registerDataScopeCallbacks(item, d.enqueueDataScopeAudit)
|
||||
}
|
||||
admin := runtime.Admin()
|
||||
disableAutoMigrate := admin != nil && admin.System != nil && admin.System.DisableAutoMigrate
|
||||
if !usingFallback && !disableAutoMigrate {
|
||||
@@ -109,17 +125,18 @@ func NewData(runtime *conf.Runtime) (*Data, func(), error) {
|
||||
}
|
||||
}
|
||||
useRedis := admin != nil && admin.System != nil && admin.System.UseRedis
|
||||
d.redis = newReloadableRedis(openRedis(c.Redis, useRedis))
|
||||
d.redis = newReloadableRedis(openRedis(c.Redis, useRedis, appLogger))
|
||||
useMongo := admin != nil && admin.System != nil && admin.System.UseMongo
|
||||
mongoClient, err := openMongo(c.Mongo, useMongo)
|
||||
if err != nil {
|
||||
log.Printf("mongo unavailable: %v", err)
|
||||
appLogger.Error("mongo unavailable", "mod", "mongo", "error", err)
|
||||
mongoClient = nil
|
||||
}
|
||||
d.mongo = newReloadableMongo(mongoClient)
|
||||
stopConfigWatcher := d.watchConfig()
|
||||
cleanup := func() {
|
||||
stopConfigWatcher()
|
||||
d.auditLog.Close()
|
||||
d.gormDB.close()
|
||||
closeDatabaseList(d.dbList)
|
||||
d.redis.close()
|
||||
@@ -128,7 +145,7 @@ func NewData(runtime *conf.Runtime) (*Data, func(), error) {
|
||||
return d, cleanup, nil
|
||||
}
|
||||
|
||||
func openRedis(config *conf.Data_Redis, enabled bool) redis.UniversalClient {
|
||||
func openRedis(config *conf.Data_Redis, enabled bool, appLogger ...*slog.Logger) redis.UniversalClient {
|
||||
if !enabled || config == nil || (config.Addr == "" && len(config.ClusterAddrs) == 0) {
|
||||
return nil
|
||||
}
|
||||
@@ -152,7 +169,11 @@ func openRedis(config *conf.Data_Redis, enabled bool) redis.UniversalClient {
|
||||
pingCtx, cancel := context.WithTimeout(context.Background(), 800*time.Millisecond)
|
||||
defer cancel()
|
||||
if err := candidate.Ping(pingCtx).Err(); err != nil {
|
||||
log.Printf("redis unavailable, using in-memory cache: %v", err)
|
||||
log := slog.Default()
|
||||
if len(appLogger) > 0 && appLogger[0] != nil {
|
||||
log = appLogger[0]
|
||||
}
|
||||
log.Warn("redis unavailable, using in-memory cache", "mod", "redis", "error", err)
|
||||
_ = candidate.Close()
|
||||
return nil
|
||||
}
|
||||
@@ -160,6 +181,6 @@ func openRedis(config *conf.Data_Redis, enabled bool) redis.UniversalClient {
|
||||
}
|
||||
|
||||
func (d *Data) activateDatabase(db *gorm.DB, config *conf.Data_Database) {
|
||||
d.gormDB.replace(db)
|
||||
d.gormDB.replace(db, d.enqueueDataScopeAudit)
|
||||
d.runtime.UpdateDatabase(config)
|
||||
}
|
||||
|
||||
+19
-15
@@ -2,7 +2,7 @@ package data
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"log"
|
||||
"log/slog"
|
||||
"reflect"
|
||||
"strings"
|
||||
|
||||
@@ -16,25 +16,27 @@ import (
|
||||
// registerDataScopeCallbacks installs the global GORM data-scope engine.
|
||||
// System tables are deliberately excluded: their access is controlled by
|
||||
// Casbin, while ownership columns on business tables are row-level scope.
|
||||
func registerDataScopeCallbacks(db *gorm.DB) {
|
||||
type dataScopeAuditEnqueue func(dataAccessLogPO)
|
||||
|
||||
func registerDataScopeCallbacks(db *gorm.DB, enqueue dataScopeAuditEnqueue) {
|
||||
if db == nil {
|
||||
return
|
||||
}
|
||||
q := db.Callback().Query()
|
||||
if q.Get("data_scope:query") == nil {
|
||||
_ = q.Before("gorm:query").Register("data_scope:query", applyDataScope("query"))
|
||||
_ = q.Before("gorm:query").Register("data_scope:query", applyDataScope("query", enqueue))
|
||||
}
|
||||
u := db.Callback().Update()
|
||||
if u.Get("data_scope:update") == nil {
|
||||
_ = u.Before("gorm:update").Register("data_scope:update", applyDataScope("update"))
|
||||
_ = u.Before("gorm:update").Register("data_scope:update", applyDataScope("update", enqueue))
|
||||
_ = u.Before("gorm:update").Register("data_scope:stamp_update", stampUpdatedBy)
|
||||
_ = u.After("gorm:update").Register("data_scope:audit_update", auditBlockedWrite("update"))
|
||||
_ = u.After("gorm:update").Register("data_scope:audit_update", auditBlockedWrite("update", enqueue))
|
||||
}
|
||||
d := db.Callback().Delete()
|
||||
if d.Get("data_scope:delete") == nil {
|
||||
_ = d.Before("gorm:delete").Register("data_scope:delete", applyDataScope("delete"))
|
||||
_ = d.Before("gorm:delete").Register("data_scope:delete", applyDataScope("delete", enqueue))
|
||||
_ = d.Before("gorm:delete").After("data_scope:delete").Register("data_scope:stamp_delete", stampDeletedBy)
|
||||
_ = d.After("gorm:delete").Register("data_scope:audit_delete", auditBlockedWrite("delete"))
|
||||
_ = d.After("gorm:delete").Register("data_scope:audit_delete", auditBlockedWrite("delete", enqueue))
|
||||
}
|
||||
c := db.Callback().Create()
|
||||
if c.Get("data_scope:stamp") == nil {
|
||||
@@ -50,7 +52,7 @@ func isControlledTable(db *gorm.DB) bool {
|
||||
return db.Statement.Schema != nil && !strings.HasPrefix(db.Statement.Table, "sys_") && (hasScopeField(db, "dept_id") || hasScopeField(db, "created_by"))
|
||||
}
|
||||
|
||||
func applyDataScope(operation string) func(*gorm.DB) {
|
||||
func applyDataScope(operation string, enqueue dataScopeAuditEnqueue) func(*gorm.DB) {
|
||||
return func(db *gorm.DB) {
|
||||
if !isControlledTable(db) {
|
||||
return
|
||||
@@ -65,8 +67,8 @@ func applyDataScope(operation string) func(*gorm.DB) {
|
||||
}
|
||||
scope, ok := biz.DataScopeFromContext(db.Statement.Context)
|
||||
if !ok {
|
||||
log.Printf("数据权限: 业务表[%s]访问无身份上下文, 已放行(待补 ctx / 或使用系统上下文)", db.Statement.Table)
|
||||
recordDataScopeEvent(db, "no_identity", operation, "无身份上下文访问受控表, 已放行", biz.DataScope{})
|
||||
slog.WarnContext(db.Statement.Context, "数据权限: 业务表访问无身份上下文, 已放行(待补 ctx / 或使用系统上下文)", "mod", "data-scope", "table", db.Statement.Table)
|
||||
recordDataScopeEvent(db, enqueue, "no_identity", operation, "无身份上下文访问受控表, 已放行", biz.DataScope{})
|
||||
return
|
||||
}
|
||||
if (operation == "update" || operation == "delete") && !db.AllowGlobalUpdate && !hasWriteConditions(db) {
|
||||
@@ -91,18 +93,20 @@ func applyDataScope(operation string) func(*gorm.DB) {
|
||||
}
|
||||
}
|
||||
|
||||
func recordDataScopeEvent(db *gorm.DB, eventType, operation, detail string, scope biz.DataScope) {
|
||||
record := &dataAccessLogPO{EventType: eventType, TargetTable: db.Statement.Table, Operation: operation, UserID: scope.UserID, AuthorityID: scope.AuthorityID, Scope: scope.Scope, RequestID: scope.RequestID, Method: scope.Method, Path: scope.Path, Detail: detail}
|
||||
_ = db.Session(&gorm.Session{NewDB: true, SkipHooks: true}).Create(record).Error
|
||||
func recordDataScopeEvent(db *gorm.DB, enqueue dataScopeAuditEnqueue, eventType, operation, detail string, scope biz.DataScope) {
|
||||
if enqueue == nil {
|
||||
return
|
||||
}
|
||||
enqueue(dataAccessLogPO{EventType: eventType, TargetTable: db.Statement.Table, Operation: operation, UserID: scope.UserID, AuthorityID: scope.AuthorityID, Scope: scope.Scope, RequestID: scope.RequestID, Method: scope.Method, Path: scope.Path, Detail: detail})
|
||||
}
|
||||
|
||||
func auditBlockedWrite(operation string) func(*gorm.DB) {
|
||||
func auditBlockedWrite(operation string, enqueue dataScopeAuditEnqueue) func(*gorm.DB) {
|
||||
return func(db *gorm.DB) {
|
||||
if _, applied := db.Statement.Clauses["data_scope:applied"]; !applied || db.Error != nil || db.RowsAffected != 0 {
|
||||
return
|
||||
}
|
||||
if scope, ok := biz.DataScopeFromContext(db.Statement.Context); ok && !scope.All {
|
||||
recordDataScopeEvent(db, "blocked_write", operation, "数据范围过滤后写操作影响 0 行(疑似越权尝试)", scope)
|
||||
recordDataScopeEvent(db, enqueue, "blocked_write", operation, "数据范围过滤后写操作影响 0 行(疑似越权尝试)", scope)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
package data
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
dataScopeAuditQueueSize = 1024
|
||||
dataScopeAuditBatchSize = 100
|
||||
dataScopeAuditInterval = 2 * time.Second
|
||||
)
|
||||
|
||||
// dataScopeAuditWriter keeps row-scope auditing best-effort: callbacks only
|
||||
// enqueue, while a background worker writes batches to the active database.
|
||||
type dataScopeAuditWriter struct {
|
||||
data *Data
|
||||
logger *slog.Logger
|
||||
queue chan dataAccessLogPO
|
||||
batchSize int
|
||||
interval time.Duration
|
||||
stop chan struct{}
|
||||
done chan struct{}
|
||||
closeOnce sync.Once
|
||||
}
|
||||
|
||||
func newDataScopeAuditWriter(data *Data, logger *slog.Logger) *dataScopeAuditWriter {
|
||||
return newDataScopeAuditWriterWithOptions(data, logger, dataScopeAuditQueueSize, dataScopeAuditBatchSize, dataScopeAuditInterval)
|
||||
}
|
||||
|
||||
func newDataScopeAuditWriterWithOptions(data *Data, logger *slog.Logger, queueSize, batchSize int, interval time.Duration) *dataScopeAuditWriter {
|
||||
w := &dataScopeAuditWriter{
|
||||
data: data,
|
||||
logger: logger,
|
||||
queue: make(chan dataAccessLogPO, queueSize),
|
||||
batchSize: batchSize,
|
||||
interval: interval,
|
||||
stop: make(chan struct{}),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
go w.run()
|
||||
return w
|
||||
}
|
||||
|
||||
func (d *Data) enqueueDataScopeAudit(record dataAccessLogPO) {
|
||||
if d == nil || d.auditLog == nil {
|
||||
return
|
||||
}
|
||||
d.auditLog.Enqueue(record)
|
||||
}
|
||||
|
||||
func (w *dataScopeAuditWriter) Enqueue(record dataAccessLogPO) {
|
||||
if w == nil {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case w.queue <- record:
|
||||
default:
|
||||
w.log().Warn("数据权限审计缓冲已满, 事件被丢弃", "mod", "data-scope", "event_type", record.EventType, "table", record.TargetTable)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *dataScopeAuditWriter) Close() {
|
||||
if w == nil {
|
||||
return
|
||||
}
|
||||
w.closeOnce.Do(func() { close(w.stop) })
|
||||
<-w.done
|
||||
}
|
||||
|
||||
func (w *dataScopeAuditWriter) run() {
|
||||
defer close(w.done)
|
||||
ticker := time.NewTicker(w.interval)
|
||||
defer ticker.Stop()
|
||||
batch := make([]dataAccessLogPO, 0, w.batchSize)
|
||||
flush := func() {
|
||||
if len(batch) == 0 || w.data == nil || w.data.gormDB == nil {
|
||||
return
|
||||
}
|
||||
db := w.data.gormDB.DB()
|
||||
if db == nil {
|
||||
return
|
||||
}
|
||||
if err := db.WithContext(context.Background()).Session(&gorm.Session{NewDB: true, SkipHooks: true}).Create(&batch).Error; err != nil {
|
||||
w.log().Warn("数据权限审计批量写入失败", "mod", "data-scope", "error", err)
|
||||
}
|
||||
batch = batch[:0]
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case record := <-w.queue:
|
||||
batch = append(batch, record)
|
||||
if len(batch) >= w.batchSize {
|
||||
flush()
|
||||
}
|
||||
case <-ticker.C:
|
||||
flush()
|
||||
case <-w.stop:
|
||||
for {
|
||||
select {
|
||||
case record := <-w.queue:
|
||||
batch = append(batch, record)
|
||||
default:
|
||||
flush()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *dataScopeAuditWriter) log() *slog.Logger {
|
||||
if w != nil && w.logger != nil {
|
||||
return w.logger
|
||||
}
|
||||
return slog.Default()
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
package data
|
||||
|
||||
import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/glebarez/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func openDataScopeAuditTestDB(t *testing.T, name string) *gorm.DB {
|
||||
t.Helper()
|
||||
db, err := gorm.Open(sqlite.Open("file:"+name+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = db.AutoMigrate(&dataAccessLogPO{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return db
|
||||
}
|
||||
|
||||
func waitForDataScopeAuditCount(t *testing.T, db *gorm.DB, want int64) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
var count int64
|
||||
if err := db.Model(&dataAccessLogPO{}).Count(&count).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count == want {
|
||||
return
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("data access log count did not reach %d", want)
|
||||
}
|
||||
|
||||
func TestDataScopeAuditWriterFlushesBatch(t *testing.T) {
|
||||
db := openDataScopeAuditTestDB(t, "data-scope-audit-batch")
|
||||
d := &Data{gormDB: newReloadableDB(db, nil)}
|
||||
w := newDataScopeAuditWriterWithOptions(d, slog.New(slog.NewTextHandler(io.Discard, nil)), 8, 2, time.Hour)
|
||||
d.auditLog = w
|
||||
t.Cleanup(w.Close)
|
||||
|
||||
d.enqueueDataScopeAudit(dataAccessLogPO{EventType: "no_identity", TargetTable: "example"})
|
||||
d.enqueueDataScopeAudit(dataAccessLogPO{EventType: "blocked_write", TargetTable: "example"})
|
||||
waitForDataScopeAuditCount(t, db, 2)
|
||||
}
|
||||
|
||||
func TestDataScopeAuditWriterUsesReloadedDatabase(t *testing.T) {
|
||||
first := openDataScopeAuditTestDB(t, "data-scope-audit-first")
|
||||
second := openDataScopeAuditTestDB(t, "data-scope-audit-second")
|
||||
d := &Data{gormDB: newReloadableDB(first, nil)}
|
||||
w := newDataScopeAuditWriterWithOptions(d, slog.New(slog.NewTextHandler(io.Discard, nil)), 8, 100, 10*time.Millisecond)
|
||||
d.auditLog = w
|
||||
t.Cleanup(w.Close)
|
||||
|
||||
d.enqueueDataScopeAudit(dataAccessLogPO{EventType: "no_identity", TargetTable: "example"})
|
||||
d.gormDB.replace(second, nil)
|
||||
waitForDataScopeAuditCount(t, second, 1)
|
||||
|
||||
var firstCount int64
|
||||
if err := first.Model(&dataAccessLogPO{}).Count(&firstCount).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if firstCount != 0 {
|
||||
t.Fatalf("audit log was written to retired database: %d", firstCount)
|
||||
}
|
||||
}
|
||||
+17
-12
@@ -2,6 +2,7 @@ package data
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/url"
|
||||
"os"
|
||||
@@ -102,7 +103,7 @@ func databaseDSN(c *conf.Data_Database, name string) (string, error) {
|
||||
return "", fmt.Errorf("unsupported database driver %q", c.Driver)
|
||||
}
|
||||
|
||||
func gormConfig(config *conf.Data_Database) *gorm.Config {
|
||||
func gormConfig(config *conf.Data_Database, appLogger ...*slog.Logger) *gorm.Config {
|
||||
level := logger.Info
|
||||
switch strings.ToLower(config.LogMode) {
|
||||
case "silent":
|
||||
@@ -112,15 +113,19 @@ func gormConfig(config *conf.Data_Database) *gorm.Config {
|
||||
case "warn":
|
||||
level = logger.Warn
|
||||
}
|
||||
return &gorm.Config{Logger: logger.Default.LogMode(level), NamingStrategy: schema.NamingStrategy{TablePrefix: config.Prefix, SingularTable: config.Singular}}
|
||||
var log *slog.Logger
|
||||
if len(appLogger) > 0 {
|
||||
log = appLogger[0]
|
||||
}
|
||||
return &gorm.Config{Logger: newGormLogger(log, level), NamingStrategy: schema.NamingStrategy{TablePrefix: config.Prefix, SingularTable: config.Singular}}
|
||||
}
|
||||
|
||||
func openWithDriver(driver, dsn string) (*gorm.DB, error) {
|
||||
return openWithDriverConfig(driver, dsn, &conf.Data_Database{Driver: driver})
|
||||
func openWithDriver(driver, dsn string, appLogger ...*slog.Logger) (*gorm.DB, error) {
|
||||
return openWithDriverConfig(driver, dsn, &conf.Data_Database{Driver: driver}, appLogger...)
|
||||
}
|
||||
|
||||
func openWithDriverConfig(driver, dsn string, config *conf.Data_Database) (*gorm.DB, error) {
|
||||
gormConfig := gormConfig(config)
|
||||
func openWithDriverConfig(driver, dsn string, config *conf.Data_Database, appLogger ...*slog.Logger) (*gorm.DB, error) {
|
||||
gormConfig := gormConfig(config, appLogger...)
|
||||
var db *gorm.DB
|
||||
var err error
|
||||
switch normalizedDriver(driver) {
|
||||
@@ -159,7 +164,7 @@ func openWithDriverConfig(driver, dsn string, config *conf.Data_Database) (*gorm
|
||||
return db, nil
|
||||
}
|
||||
|
||||
func openDatabase(c *conf.Data_Database, create bool, template string) (*gorm.DB, error) {
|
||||
func openDatabase(c *conf.Data_Database, create bool, template string, appLogger ...*slog.Logger) (*gorm.DB, error) {
|
||||
driver := normalizedDriver(c.Driver)
|
||||
if driver == "" {
|
||||
return nil, fmt.Errorf("unsupported database driver %q", c.Driver)
|
||||
@@ -172,7 +177,7 @@ func openDatabase(c *conf.Data_Database, create bool, template string) (*gorm.DB
|
||||
if err = os.MkdirAll(filepath.Dir(dsn), 0o755); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return openWithDriverConfig(driver, dsn, c)
|
||||
return openWithDriverConfig(driver, dsn, c, appLogger...)
|
||||
}
|
||||
if create && driver != "oracle" {
|
||||
if !databaseNamePattern.MatchString(c.Name) {
|
||||
@@ -189,7 +194,7 @@ func openDatabase(c *conf.Data_Database, create bool, template string) (*gorm.DB
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
adminDB, err := openWithDriverConfig(driver, dsn, c)
|
||||
adminDB, err := openWithDriverConfig(driver, dsn, c, appLogger...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("connect database server: %w", err)
|
||||
}
|
||||
@@ -225,9 +230,9 @@ func openDatabase(c *conf.Data_Database, create bool, template string) (*gorm.DB
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return openWithDriverConfig(driver, dsn, c)
|
||||
return openWithDriverConfig(driver, dsn, c, appLogger...)
|
||||
}
|
||||
|
||||
func openFallbackDatabase() (*gorm.DB, error) {
|
||||
return openWithDriver("sqlite", "file:kra-bootstrap?mode=memory&cache=shared")
|
||||
func openFallbackDatabase(appLogger ...*slog.Logger) (*gorm.DB, error) {
|
||||
return openWithDriver("sqlite", "file:kra-bootstrap?mode=memory&cache=shared", appLogger...)
|
||||
}
|
||||
|
||||
@@ -389,9 +389,11 @@ func (r *dictionaryRepo) ListDictionaryDetails(ctx context.Context, page, size i
|
||||
}
|
||||
func (r *dictionaryRepo) DictionaryDetailTree(ctx context.Context, dictionaryID uint, typ string) ([]*biz.DictionaryDetail, error) {
|
||||
if dictionaryID == 0 {
|
||||
active := true
|
||||
dictionary, err := r.FindDictionary(ctx, 0, typ, &active, false)
|
||||
if err != nil {
|
||||
// The tree-by-type endpoint resolves only by dictionary type. Unlike the
|
||||
// public dictionary lookup, it does not require the dictionary itself to
|
||||
// be enabled.
|
||||
var dictionary dictionaryPO
|
||||
if err := r.data.gormDB.WithContext(ctx).Where("type = ?", typ).First(&dictionary).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dictionaryID = dictionary.ID
|
||||
@@ -409,7 +411,7 @@ func (r *dictionaryRepo) DictionaryDetailTree(ctx context.Context, dictionaryID
|
||||
for _, item := range items {
|
||||
if item.ParentID != nil && byID[*item.ParentID] != nil {
|
||||
byID[*item.ParentID].Children = append(byID[*item.ParentID].Children, item)
|
||||
} else {
|
||||
} else if item.ParentID == nil {
|
||||
roots = append(roots, item)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
package data
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm/logger"
|
||||
)
|
||||
|
||||
// gormLogger forwards GORM diagnostics through the application logger so SQL
|
||||
// entries participate in the same daily/category files and error sink as the
|
||||
// rest of the service. Slow queries use the administration contract's 200ms
|
||||
// threshold.
|
||||
type gormLogger struct {
|
||||
logger *slog.Logger
|
||||
slowThreshold time.Duration
|
||||
level logger.LogLevel
|
||||
}
|
||||
|
||||
func newGormLogger(log *slog.Logger, level logger.LogLevel) *gormLogger {
|
||||
if log == nil {
|
||||
log = slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
}
|
||||
return &gormLogger{logger: log, slowThreshold: 200 * time.Millisecond, level: level}
|
||||
}
|
||||
|
||||
func (g *gormLogger) LogMode(level logger.LogLevel) logger.Interface {
|
||||
next := *g
|
||||
next.level = level
|
||||
return &next
|
||||
}
|
||||
|
||||
func (g *gormLogger) Info(ctx context.Context, message string, args ...any) {
|
||||
g.logger.InfoContext(ctx, fmt.Sprintf(message, args...), "mod", "sql", "gorm_logger", true)
|
||||
}
|
||||
|
||||
func (g *gormLogger) Warn(ctx context.Context, message string, args ...any) {
|
||||
g.logger.WarnContext(ctx, fmt.Sprintf(message, args...), "mod", "sql", "gorm_logger", true)
|
||||
}
|
||||
|
||||
func (g *gormLogger) Error(ctx context.Context, message string, args ...any) {
|
||||
g.logger.ErrorContext(ctx, fmt.Sprintf(message, args...), "mod", "sql", "gorm_logger", true)
|
||||
}
|
||||
|
||||
func (g *gormLogger) Trace(ctx context.Context, begin time.Time, query func() (string, int64), queryErr error) {
|
||||
if g.level <= logger.Silent {
|
||||
return
|
||||
}
|
||||
elapsed := time.Since(begin)
|
||||
sql, rows := query()
|
||||
fields := []any{"mod", "sql", "gorm_logger", true, "sql", sql, "rows", rows, "elapsed_ms", elapsed.Milliseconds()}
|
||||
switch {
|
||||
case queryErr != nil && g.level >= logger.Error && !errors.Is(queryErr, logger.ErrRecordNotFound):
|
||||
fields = append(fields, "error", queryErr)
|
||||
g.logger.ErrorContext(ctx, "SQL 执行错误", fields...)
|
||||
case elapsed > g.slowThreshold && g.level >= logger.Warn:
|
||||
g.logger.WarnContext(ctx, "SQL 慢查询", fields...)
|
||||
case g.level >= logger.Info:
|
||||
g.logger.InfoContext(ctx, "SQL", fields...)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package data
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"kra/pkg/logging"
|
||||
|
||||
"gorm.io/gorm/logger"
|
||||
)
|
||||
|
||||
func TestGORMLoggerUsesApplicationCategories(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
appLogger, cleanup := logging.NewZapLogger(root, "application.log", logging.Options{Level: "info", Format: "json"})
|
||||
databaseLogger := newGormLogger(appLogger, logger.Info)
|
||||
databaseLogger.Trace(context.Background(), time.Now(), func() (string, int64) { return "SELECT 1", 1 }, nil)
|
||||
databaseLogger.Trace(context.Background(), time.Now(), func() (string, int64) { return "SELECT missing", 0 }, errors.New("database failure"))
|
||||
cleanup()
|
||||
|
||||
date := time.Now().Format("2006-01-02")
|
||||
for _, name := range []string{"application.log", filepath.Join("sql", "application.log"), filepath.Join("error", "error.log")} {
|
||||
info, err := os.Stat(filepath.Join(root, date, name))
|
||||
if err != nil || info.Size() == 0 {
|
||||
t.Fatalf("expected non-empty GORM log %s: info=%v err=%v", name, info, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
package data
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"kra/internal/biz"
|
||||
"kra/internal/conf"
|
||||
)
|
||||
|
||||
func TestLogViewerReadsNestedCategoryFiles(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
date := "2026-08-16"
|
||||
path := filepath.Join(root, date, "http", "access.log")
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(path, []byte("first\nsecond\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
runtime := conf.NewRuntime(nil, &conf.AdminBackend{Zap: &conf.AdminBackend_Zap{Director: root}})
|
||||
repo := &logFileRepo{data: &Data{runtime: runtime}}
|
||||
files, err := repo.LogFiles(context.Background(), date)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(files) != 1 || files[0].Path != filepath.Join("http", "access.log") {
|
||||
t.Fatalf("unexpected nested log files: %+v", files)
|
||||
}
|
||||
content, err := repo.LogContent(context.Background(), date, "http/access.log", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if content.Content != "first\nsecond\n" || content.LineCount != 2 || content.HasMore {
|
||||
t.Fatalf("unexpected log content: %+v", content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLogViewerRejectsPathTraversal(t *testing.T) {
|
||||
runtime := conf.NewRuntime(nil, &conf.AdminBackend{Zap: &conf.AdminBackend_Zap{Director: t.TempDir()}})
|
||||
repo := &logFileRepo{data: &Data{runtime: runtime}}
|
||||
_, err := repo.LogContent(context.Background(), "2026-08-16", "../application.log", nil)
|
||||
if !errors.Is(err, biz.ErrInvalidLogPath) {
|
||||
t.Fatalf("expected invalid path error, got %v", err)
|
||||
}
|
||||
}
|
||||
+32
-20
@@ -38,7 +38,7 @@ func migrateAll(db *gorm.DB) error {
|
||||
// Older builds used a status label outside the administration page's supported
|
||||
// state set, so normalize existing rows during migration.
|
||||
func normalizeErrorRecordStatuses(db *gorm.DB) error {
|
||||
return db.Model(&errorRecordPO{}).Where("status = ?", "未解决").Update("status", "未处理").Error
|
||||
return db.Session(&gorm.Session{NewDB: true}).Model(&errorRecordPO{}).Where("status = ?", "未解决").Update("status", "未处理").Error
|
||||
}
|
||||
|
||||
// migrateLegacyAuthorityAPIsToCasbinRules upgrades the early Kra join-table
|
||||
@@ -46,7 +46,8 @@ func normalizeErrorRecordStatuses(db *gorm.DB) error {
|
||||
// table in place for backwards compatibility, but make casbin_rule the sole
|
||||
// live policy source. Existing policy rows are not duplicated.
|
||||
func migrateLegacyAuthorityAPIsToCasbinRules(db *gorm.DB) error {
|
||||
if !db.Migrator().HasTable(&authorityAPIPO{}) || !db.Migrator().HasTable(&casbinRulePO{}) {
|
||||
clean := db.Session(&gorm.Session{NewDB: true})
|
||||
if !clean.Migrator().HasTable(&authorityAPIPO{}) || !clean.Migrator().HasTable(&casbinRulePO{}) {
|
||||
return nil
|
||||
}
|
||||
type legacyPolicy struct {
|
||||
@@ -54,22 +55,28 @@ func migrateLegacyAuthorityAPIsToCasbinRules(db *gorm.DB) error {
|
||||
Path string
|
||||
Method string
|
||||
}
|
||||
var rows []legacyPolicy
|
||||
if err := db.Table("sys_authority_apis sa").
|
||||
query := clean.Table("sys_authority_apis sa").
|
||||
Select("sa.authority_id, a.path, a.method").
|
||||
Joins("JOIN sys_apis a ON a.id = sa.api_id").
|
||||
Where("a.deleted_at IS NULL").Find(&rows).Error; err != nil {
|
||||
Joins("JOIN sys_apis a ON a.id = sa.api_id")
|
||||
// Early Kra schemas stored sys_apis without soft-delete timestamps. The
|
||||
// legacy-policy migration must run before assuming that column exists;
|
||||
// otherwise an upgrade from those schemas cannot start on MySQL.
|
||||
if clean.Migrator().HasColumn(&apiPO{}, "deleted_at") {
|
||||
query = query.Where("a.deleted_at IS NULL")
|
||||
}
|
||||
var rows []legacyPolicy
|
||||
if err := query.Find(&rows).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
for _, row := range rows {
|
||||
exists, err := policyExists(db, row.AuthorityID, row.Path, row.Method)
|
||||
exists, err := policyExists(clean, row.AuthorityID, row.Path, row.Method)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if exists {
|
||||
continue
|
||||
}
|
||||
if err := db.Create(&casbinRulePO{Ptype: "p", V0: fmt.Sprint(row.AuthorityID), V1: row.Path, V2: row.Method}).Error; err != nil {
|
||||
if err := clean.Create(&casbinRulePO{Ptype: "p", V0: fmt.Sprint(row.AuthorityID), V1: row.Path, V2: row.Method}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
@@ -82,7 +89,8 @@ func migrateLegacyAuthorityAPIsToCasbinRules(db *gorm.DB) error {
|
||||
// composite primary key portably, so rebuild the small table once while
|
||||
// preserving every existing ignore rule.
|
||||
func migrateLegacyIgnoreAPITable(db *gorm.DB) error {
|
||||
if !db.Migrator().HasTable(&ignoredAPIPO{}) || db.Migrator().HasColumn(&ignoredAPIPO{}, "id") {
|
||||
clean := db.Session(&gorm.Session{NewDB: true})
|
||||
if !clean.Migrator().HasTable(&ignoredAPIPO{}) || clean.Migrator().HasColumn(&ignoredAPIPO{}, "id") {
|
||||
return nil
|
||||
}
|
||||
legacyTable := fmt.Sprintf("sys_ignore_apis_legacy_%d", time.Now().UnixNano())
|
||||
@@ -90,7 +98,7 @@ func migrateLegacyIgnoreAPITable(db *gorm.DB) error {
|
||||
Path string
|
||||
Method string
|
||||
}
|
||||
return db.Transaction(func(tx *gorm.DB) error {
|
||||
return clean.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Migrator().RenameTable(ignoredAPIPO{}.TableName(), legacyTable); err != nil {
|
||||
return fmt.Errorf("rename legacy ignore API table: %w", err)
|
||||
}
|
||||
@@ -123,16 +131,17 @@ func migrateLegacyIgnoreAPITable(db *gorm.DB) error {
|
||||
// database has the root role but no stored API links, materialize the same
|
||||
// policy set and let normal authorization read it thereafter.
|
||||
func reconcileRootAuthorityAPIs(db *gorm.DB) error {
|
||||
clean := db.Session(&gorm.Session{NewDB: true})
|
||||
var authorityCount int64
|
||||
if err := db.Model(&authorityPO{}).Where("authority_id = ?", 888).Count(&authorityCount).Error; err != nil || authorityCount == 0 {
|
||||
if err := clean.Session(&gorm.Session{NewDB: true}).Model(&authorityPO{}).Where("authority_id = ?", 888).Count(&authorityCount).Error; err != nil || authorityCount == 0 {
|
||||
return err
|
||||
}
|
||||
var policyCount int64
|
||||
if err := policyScope(db).Where("v0 = ?", "888").Count(&policyCount).Error; err != nil || policyCount != 0 {
|
||||
if err := policyScope(clean).Where("v0 = ?", "888").Count(&policyCount).Error; err != nil || policyCount != 0 {
|
||||
return err
|
||||
}
|
||||
var ignored []ignoredAPIPO
|
||||
if err := db.Find(&ignored).Error; err != nil {
|
||||
if err := clean.Session(&gorm.Session{NewDB: true}).Find(&ignored).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
ignoreSet := make(map[string]struct{}, len(ignored))
|
||||
@@ -140,7 +149,7 @@ func reconcileRootAuthorityAPIs(db *gorm.DB) error {
|
||||
ignoreSet[item.Method+"\x00"+item.Path] = struct{}{}
|
||||
}
|
||||
var apis []apiPO
|
||||
if err := db.Find(&apis).Error; err != nil {
|
||||
if err := clean.Session(&gorm.Session{NewDB: true}).Find(&apis).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
rules := make([]casbinRulePO, 0, len(apis))
|
||||
@@ -153,13 +162,14 @@ func reconcileRootAuthorityAPIs(db *gorm.DB) error {
|
||||
if len(rules) == 0 {
|
||||
return nil
|
||||
}
|
||||
return db.Create(&rules).Error
|
||||
return clean.Session(&gorm.Session{NewDB: true}).Create(&rules).Error
|
||||
}
|
||||
|
||||
// reconcileReferenceIndexes removes constraints created by older Kra builds
|
||||
// that are not part of the administration data model. Business services own
|
||||
// duplicate checks and their user-facing error messages.
|
||||
func reconcileReferenceIndexes(db *gorm.DB) error {
|
||||
clean := db.Session(&gorm.Session{NewDB: true})
|
||||
obsolete := []struct {
|
||||
model any
|
||||
name string
|
||||
@@ -171,8 +181,9 @@ func reconcileReferenceIndexes(db *gorm.DB) error {
|
||||
{&exportTemplatePO{}, "idx_sys_export_templates_template_id"},
|
||||
}
|
||||
for _, item := range obsolete {
|
||||
if db.Migrator().HasIndex(item.model, item.name) {
|
||||
if err := db.Migrator().DropIndex(item.model, item.name); err != nil {
|
||||
migrator := clean.Session(&gorm.Session{NewDB: true}).Migrator()
|
||||
if migrator.HasIndex(item.model, item.name) {
|
||||
if err := migrator.DropIndex(item.model, item.name); err != nil {
|
||||
return fmt.Errorf("drop obsolete index %s: %w", item.name, err)
|
||||
}
|
||||
}
|
||||
@@ -181,17 +192,18 @@ func reconcileReferenceIndexes(db *gorm.DB) error {
|
||||
name string
|
||||
field string
|
||||
}{{"idx_sys_users_uuid", "UUID"}, {"idx_sys_users_username", "Username"}} {
|
||||
unique, err := indexIsUnique(db, &userPO{}, item.name)
|
||||
unique, err := indexIsUnique(clean.Session(&gorm.Session{NewDB: true}), &userPO{}, item.name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !unique {
|
||||
continue
|
||||
}
|
||||
if err = db.Migrator().DropIndex(&userPO{}, item.name); err != nil {
|
||||
migrator := clean.Session(&gorm.Session{NewDB: true}).Migrator()
|
||||
if err = migrator.DropIndex(&userPO{}, item.name); err != nil {
|
||||
return fmt.Errorf("drop legacy unique index %s: %w", item.name, err)
|
||||
}
|
||||
if err = db.Migrator().CreateIndex(&userPO{}, item.field); err != nil {
|
||||
if err = migrator.CreateIndex(&userPO{}, item.field); err != nil {
|
||||
return fmt.Errorf("create reference index %s: %w", item.name, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -49,9 +49,9 @@ func (r *reloadableMongo) close() {
|
||||
}
|
||||
}
|
||||
|
||||
func newReloadableDB(db *gorm.DB) *reloadableDB {
|
||||
func newReloadableDB(db *gorm.DB, enqueue dataScopeAuditEnqueue) *reloadableDB {
|
||||
r := &reloadableDB{}
|
||||
registerDataScopeCallbacks(db)
|
||||
registerDataScopeCallbacks(db, enqueue)
|
||||
r.current.Store(db)
|
||||
return r
|
||||
}
|
||||
@@ -62,8 +62,8 @@ func (r *reloadableDB) WithContext(ctx context.Context) *gorm.DB {
|
||||
|
||||
func (r *reloadableDB) DB() *gorm.DB { return r.current.Load() }
|
||||
|
||||
func (r *reloadableDB) replace(db *gorm.DB) {
|
||||
registerDataScopeCallbacks(db)
|
||||
func (r *reloadableDB) replace(db *gorm.DB, enqueue dataScopeAuditEnqueue) {
|
||||
registerDataScopeCallbacks(db, enqueue)
|
||||
old := r.current.Swap(db)
|
||||
if old != nil && old != db {
|
||||
r.mu.Lock()
|
||||
|
||||
@@ -95,7 +95,7 @@ func (r *initializationRepo) Initialize(ctx context.Context, input *biz.Database
|
||||
if initialized {
|
||||
return errors.New("数据库已初始化,无需重复初始化")
|
||||
}
|
||||
candidate, err := openDatabase(config, true, input.Template)
|
||||
candidate, err := openDatabase(config, true, input.Template, r.data.logger())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -251,6 +251,7 @@ func (r *initializationRepo) Initialize(ctx context.Context, input *biz.Database
|
||||
func defaultIgnoredAPIs() []ignoredAPIPO {
|
||||
return []ignoredAPIPO{
|
||||
{Method: "GET", Path: "/api/freshCasbin"}, {Method: "GET", Path: "/health"},
|
||||
{Method: "GET", Path: "/swagger/*any"},
|
||||
{Method: "POST", Path: "/system/reloadSystem"}, {Method: "POST", Path: "/base/login"},
|
||||
{Method: "POST", Path: "/base/captcha"}, {Method: "POST", Path: "/init/initdb"},
|
||||
{Method: "POST", Path: "/init/checkdb"}, {Method: "GET", Path: "/info/getInfoDataSource"},
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
package data
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestDefaultIgnoredAPIsIncludeSwagger(t *testing.T) {
|
||||
for _, api := range defaultIgnoredAPIs() {
|
||||
if api.Method == "GET" && api.Path == "/swagger/*any" {
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatal("default ignored APIs do not include the Swagger handler")
|
||||
}
|
||||
@@ -19,7 +19,7 @@ func newTransactionTestData(t *testing.T) *Data {
|
||||
if err = migrateAll(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data := &Data{gormDB: newReloadableDB(db), redis: newReloadableRedis(nil), runtime: conf.NewRuntime(&conf.Data{Database: &conf.Data_Database{Driver: "sqlite"}}, &conf.AdminBackend{})}
|
||||
data := &Data{gormDB: newReloadableDB(db, nil), redis: newReloadableRedis(nil), runtime: conf.NewRuntime(&conf.Data{Database: &conf.Data_Database{Driver: "sqlite"}}, &conf.AdminBackend{})}
|
||||
t.Cleanup(func() { data.gormDB.close() })
|
||||
return data
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user