优化结构
This commit is contained in:
Yvan
@@ -0,0 +1,264 @@
|
||||
package data
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"kra/internal/biz"
|
||||
"kra/internal/conf"
|
||||
"kra/internal/integration/storage"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"google.golang.org/protobuf/encoding/protojson"
|
||||
"google.golang.org/protobuf/proto"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func (d *Data) RuntimeValues() (*conf.Data, *conf.AdminBackend) { return d.runtime.Values() }
|
||||
func (d *Data) RuntimeAdmin() *conf.AdminBackend { return d.runtime.Admin() }
|
||||
func (d *Data) RefreshDatabaseSources(value *conf.Data) error { return refreshDatabaseSources(value) }
|
||||
|
||||
func refreshDatabaseSources(value *conf.Data) error {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
if err := refreshDatabaseSource(value.Database); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, database := range value.DatabaseList {
|
||||
if database == nil || database.Disable {
|
||||
continue
|
||||
}
|
||||
if err := refreshDatabaseSource(database); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func refreshDatabaseSource(database *conf.Data_Database) error {
|
||||
if database == nil {
|
||||
return nil
|
||||
}
|
||||
hasStructuredConfig := database.Host != "" || database.Port != "" || database.User != "" || database.Password != "" || database.Name != "" || database.Config != "" || database.Path != ""
|
||||
if !hasStructuredConfig {
|
||||
return nil
|
||||
}
|
||||
previousSource := database.Source
|
||||
database.Source = ""
|
||||
source, err := databaseDSN(database, "")
|
||||
if err != nil {
|
||||
database.Source = previousSource
|
||||
return err
|
||||
}
|
||||
database.Source = source
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Data) PersistConfig(context.Context) error { return d.persistConfig() }
|
||||
func (d *Data) PersistAdminConfig(ctx context.Context, raw []byte) error {
|
||||
currentData, currentAdmin := d.runtime.Values()
|
||||
next := proto.Clone(currentAdmin).(*conf.AdminBackend)
|
||||
if err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(raw, next); err != nil {
|
||||
return err
|
||||
}
|
||||
if next.Storage == nil {
|
||||
next.Storage = currentAdmin.Storage
|
||||
}
|
||||
if next.Email == nil {
|
||||
next.Email = currentAdmin.Email
|
||||
}
|
||||
if next.Websocket == nil {
|
||||
next.Websocket = currentAdmin.Websocket
|
||||
}
|
||||
if next.Mq == nil {
|
||||
next.Mq = currentAdmin.Mq
|
||||
}
|
||||
next.ConfigPath = currentAdmin.ConfigPath
|
||||
candidateStorage, err := storage.New(next)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.persistStorageIntegrationConfig(ctx, next.Storage); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.persistEmailIntegrationConfig(ctx, next.Email); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.PersistWebSocketConfig(ctx, next.Websocket); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.persistMQIntegrationConfig(ctx, next.Mq); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.persistConfigValues(currentData, next); err != nil {
|
||||
return err
|
||||
}
|
||||
// Writing through the management API updates the same in-memory values
|
||||
// immediately; the file watcher remains the fallback for external edits.
|
||||
d.runtime.Replace(currentData, next)
|
||||
if d.storage != nil {
|
||||
d.storage.Replace(candidateStorage)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (d *Data) PersistRuntimeConfig(ctx context.Context, dataRaw, adminRaw []byte) error {
|
||||
currentData, currentAdmin := d.runtime.Values()
|
||||
nextData := proto.Clone(currentData).(*conf.Data)
|
||||
nextAdmin := proto.Clone(currentAdmin).(*conf.AdminBackend)
|
||||
options := protojson.UnmarshalOptions{DiscardUnknown: true}
|
||||
if err := options.Unmarshal(dataRaw, nextData); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := options.Unmarshal(adminRaw, nextAdmin); err != nil {
|
||||
return err
|
||||
}
|
||||
if nextAdmin.Storage == nil {
|
||||
nextAdmin.Storage = currentAdmin.Storage
|
||||
}
|
||||
if nextAdmin.Email == nil {
|
||||
nextAdmin.Email = currentAdmin.Email
|
||||
}
|
||||
if nextAdmin.Websocket == nil {
|
||||
nextAdmin.Websocket = currentAdmin.Websocket
|
||||
}
|
||||
if nextAdmin.Mq == nil {
|
||||
nextAdmin.Mq = currentAdmin.Mq
|
||||
}
|
||||
nextAdmin.ConfigPath = currentAdmin.ConfigPath
|
||||
candidateStorage, err := storage.New(nextAdmin)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.persistStorageIntegrationConfig(ctx, nextAdmin.Storage); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.persistEmailIntegrationConfig(ctx, nextAdmin.Email); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.PersistWebSocketConfig(ctx, nextAdmin.Websocket); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.persistMQIntegrationConfig(ctx, nextAdmin.Mq); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.persistConfigValues(nextData, nextAdmin); err != nil {
|
||||
return err
|
||||
}
|
||||
d.runtime.Replace(nextData, nextAdmin)
|
||||
if d.storage != nil {
|
||||
d.storage.Replace(candidateStorage)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (d *Data) ReloadConfig(ctx context.Context) error {
|
||||
return d.reloadConfig(ctx)
|
||||
}
|
||||
|
||||
func (d *Data) IsInitialized(context.Context) (bool, error) {
|
||||
return d.databaseReady.Load(), nil
|
||||
}
|
||||
|
||||
// InitializeDatabase opens and activates the configured database. The callback
|
||||
// is the application-level first-install hook; data owns only lifecycle and
|
||||
// schema migration, while initialize owns system seed orchestration.
|
||||
func (d *Data) InitializeDatabase(ctx context.Context, input *biz.DatabaseConfig, seed func(context.Context, *gorm.DB) error) error {
|
||||
config := &conf.Data_Database{}
|
||||
if current := d.runtime.Data(); current != nil && current.Database != nil {
|
||||
config = proto.Clone(current.Database).(*conf.Data_Database)
|
||||
}
|
||||
config.Driver = input.Driver
|
||||
config.Host = input.Host
|
||||
config.Port = input.Port
|
||||
config.User = input.User
|
||||
config.Password = input.Password
|
||||
config.Name = input.Name
|
||||
config.Path = input.Path
|
||||
config.Config = input.Config
|
||||
config.Source = ""
|
||||
source, err := databaseDSN(config, "")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
config.Source = source
|
||||
d.initMu.Lock()
|
||||
defer d.initMu.Unlock()
|
||||
initialized, err := d.IsInitialized(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if initialized {
|
||||
return errors.New("数据库已初始化，无需重复初始化")
|
||||
}
|
||||
candidate, err := openDatabase(config, true, input.Template, d.logger())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
activated := false
|
||||
defer func() {
|
||||
if !activated {
|
||||
if sqlDB, closeErr := candidate.DB(); closeErr == nil {
|
||||
_ = sqlDB.Close()
|
||||
}
|
||||
}
|
||||
}()
|
||||
db := candidate.WithContext(ctx)
|
||||
if err := migrateAll(db, d.catalog); err != nil {
|
||||
return err
|
||||
}
|
||||
if seed != nil {
|
||||
if err := seed(ctx, db); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
currentAdmin := d.runtime.Admin()
|
||||
var legacyStorage *conf.AdminBackend_Storage
|
||||
if currentAdmin != nil {
|
||||
legacyStorage = currentAdmin.Storage
|
||||
}
|
||||
storageConfig, err := resolveStorageIntegrationConfig(candidate.WithContext(ctx), legacyStorage)
|
||||
if err != nil {
|
||||
return fmt.Errorf("initialize storage integration configuration: %w", err)
|
||||
}
|
||||
var legacyEmail *conf.AdminBackend_Email
|
||||
if currentAdmin != nil {
|
||||
legacyEmail = currentAdmin.Email
|
||||
}
|
||||
emailConfig, err := resolveEmailIntegrationConfig(candidate.WithContext(ctx), legacyEmail)
|
||||
if err != nil {
|
||||
return fmt.Errorf("initialize email integration configuration: %w", err)
|
||||
}
|
||||
var legacyWebSocket *conf.AdminBackend_WebSocket
|
||||
if currentAdmin != nil {
|
||||
legacyWebSocket = currentAdmin.Websocket
|
||||
}
|
||||
websocketConfig, err := resolveWebSocketIntegrationConfig(candidate.WithContext(ctx), legacyWebSocket)
|
||||
if err != nil {
|
||||
return fmt.Errorf("initialize websocket integration configuration: %w", err)
|
||||
}
|
||||
signingKey := uuid.NewString()
|
||||
if err := d.persistDatabaseConfig(config, signingKey); err != nil {
|
||||
return fmt.Errorf("persist database configuration: %w", err)
|
||||
}
|
||||
d.activateDatabase(candidate, config)
|
||||
currentData, currentAdmin := d.runtime.Values()
|
||||
if currentAdmin == nil {
|
||||
currentAdmin = &conf.AdminBackend{}
|
||||
}
|
||||
if currentAdmin.Jwt == nil {
|
||||
currentAdmin.Jwt = &conf.AdminBackend_JWT{}
|
||||
}
|
||||
currentAdmin.Jwt.SigningKey = signingKey
|
||||
currentAdmin.Storage = storageConfig
|
||||
currentAdmin.Websocket = websocketConfig
|
||||
currentAdmin.Email = emailConfig
|
||||
mqConfig, err := resolveMQIntegrationConfig(candidate.WithContext(ctx), currentAdmin.Mq)
|
||||
if err != nil {
|
||||
return fmt.Errorf("initialize mq integration configuration: %w", err)
|
||||
}
|
||||
currentAdmin.Mq = mqConfig
|
||||
d.runtime.Replace(currentData, currentAdmin)
|
||||
activated = true
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user