465 lines
13 KiB
Go
465 lines
13 KiB
Go
package data
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
|
|
"kra/internal/conf"
|
|
"kra/internal/integration/storage"
|
|
|
|
"google.golang.org/protobuf/encoding/protojson"
|
|
"google.golang.org/protobuf/proto"
|
|
"gopkg.in/yaml.v3"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
func protoMap(message proto.Message) (map[string]any, error) {
|
|
raw, err := (protojson.MarshalOptions{
|
|
UseProtoNames: true,
|
|
EmitDefaultValues: true,
|
|
}).Marshal(message)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var value map[string]any
|
|
if err = json.Unmarshal(raw, &value); err != nil {
|
|
return nil, err
|
|
}
|
|
delete(value, "config_path")
|
|
return value, nil
|
|
}
|
|
|
|
func setYAMLMapping(node *yaml.Node, key string, value any) error {
|
|
if node.Kind == yaml.DocumentNode {
|
|
node = node.Content[0]
|
|
}
|
|
if node.Kind != yaml.MappingNode {
|
|
return fmt.Errorf("configuration root is not a mapping")
|
|
}
|
|
raw, err := yaml.Marshal(value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var replacement yaml.Node
|
|
if err = yaml.Unmarshal(raw, &replacement); err != nil {
|
|
return err
|
|
}
|
|
for i := 0; i < len(node.Content); i += 2 {
|
|
if node.Content[i].Value == key {
|
|
mergeYAMLNode(node.Content[i+1], replacement.Content[0])
|
|
return nil
|
|
}
|
|
}
|
|
node.Content = append(node.Content, &yaml.Node{Kind: yaml.ScalarNode, Value: key}, replacement.Content[0])
|
|
return nil
|
|
}
|
|
|
|
// mergeYAMLNode updates values produced from the runtime configuration while
|
|
// retaining keys that are not represented by the protobuf and comments that
|
|
// were already present in the user's config file. Replacing an entire mapping
|
|
// with protojson output would discard configuration hints and extension keys.
|
|
func mergeYAMLNode(dst, src *yaml.Node) {
|
|
if dst.Kind == yaml.MappingNode && src.Kind == yaml.MappingNode {
|
|
for i := 0; i+1 < len(src.Content); i += 2 {
|
|
key := src.Content[i].Value
|
|
found := false
|
|
for j := 0; j+1 < len(dst.Content); j += 2 {
|
|
if dst.Content[j].Value == key {
|
|
mergeYAMLNode(dst.Content[j+1], src.Content[i+1])
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
if !found {
|
|
dst.Content = append(dst.Content, src.Content[i], src.Content[i+1])
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
// Keep comments attached to a scalar/sequence node when its value changes.
|
|
head, line, foot := dst.HeadComment, dst.LineComment, dst.FootComment
|
|
*dst = *src
|
|
dst.HeadComment, dst.LineComment, dst.FootComment = head, line, foot
|
|
}
|
|
|
|
func yamlMappingValue(node *yaml.Node, keys ...string) *yaml.Node {
|
|
if node == nil {
|
|
return nil
|
|
}
|
|
if node.Kind == yaml.DocumentNode && len(node.Content) > 0 {
|
|
node = node.Content[0]
|
|
}
|
|
for _, key := range keys {
|
|
if node.Kind != yaml.MappingNode {
|
|
return nil
|
|
}
|
|
var next *yaml.Node
|
|
for i := 0; i+1 < len(node.Content); i += 2 {
|
|
if node.Content[i].Value == key {
|
|
next = node.Content[i+1]
|
|
break
|
|
}
|
|
}
|
|
if next == nil {
|
|
return nil
|
|
}
|
|
node = next
|
|
}
|
|
return node
|
|
}
|
|
|
|
func deleteYAMLMapping(node *yaml.Node, keys ...string) {
|
|
if len(keys) == 0 || node == nil {
|
|
return
|
|
}
|
|
if node.Kind == yaml.DocumentNode && len(node.Content) > 0 {
|
|
node = node.Content[0]
|
|
}
|
|
if node.Kind != yaml.MappingNode {
|
|
return
|
|
}
|
|
key := keys[0]
|
|
for i := 0; i+1 < len(node.Content); i += 2 {
|
|
if node.Content[i].Value != key {
|
|
continue
|
|
}
|
|
if len(keys) == 1 {
|
|
node.Content = append(node.Content[:i], node.Content[i+2:]...)
|
|
return
|
|
}
|
|
deleteYAMLMapping(node.Content[i+1], keys[1:]...)
|
|
return
|
|
}
|
|
}
|
|
|
|
func setServerHTTPPort(document *yaml.Node, port int32) error {
|
|
if port <= 0 {
|
|
return nil
|
|
}
|
|
host := "0.0.0.0"
|
|
if addr := yamlMappingValue(document, "server", "http", "addr"); addr != nil {
|
|
if currentHost, _, err := net.SplitHostPort(addr.Value); err == nil && currentHost != "" {
|
|
host = currentHost
|
|
}
|
|
}
|
|
return setYAMLMapping(document, "server", map[string]any{"http": map[string]any{"addr": net.JoinHostPort(host, strconv.Itoa(int(port)))}})
|
|
}
|
|
|
|
func (d *Data) persistConfig() error {
|
|
dataConfig, adminConfig := d.runtime.Values()
|
|
return d.persistConfigValues(dataConfig, adminConfig)
|
|
}
|
|
|
|
func (d *Data) persistConfigValues(dataConfig *conf.Data, adminConfig *conf.AdminBackend) error {
|
|
d.configMu.Lock()
|
|
defer d.configMu.Unlock()
|
|
return d.persistConfigValuesLocked(dataConfig, adminConfig)
|
|
}
|
|
|
|
func (d *Data) persistConfigValuesLocked(dataConfig *conf.Data, adminConfig *conf.AdminBackend) error {
|
|
if adminConfig == nil || adminConfig.ConfigPath == "" {
|
|
return nil
|
|
}
|
|
configPath := adminConfig.ConfigPath
|
|
raw, err := os.ReadFile(configPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var document yaml.Node
|
|
if err = yaml.Unmarshal(raw, &document); err != nil {
|
|
return err
|
|
}
|
|
dataValue, err := protoMap(dataConfig)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
fileAdmin := cloneAdminConfig(adminConfig)
|
|
fileAdmin.Storage = nil
|
|
fileAdmin.Email = nil
|
|
fileAdmin.Websocket = nil
|
|
fileAdmin.Mq = nil
|
|
adminValue, err := protoMap(fileAdmin)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = setYAMLMapping(&document, "data", dataValue); err != nil {
|
|
return err
|
|
}
|
|
if err = setYAMLMapping(&document, "admin", adminValue); err != nil {
|
|
return err
|
|
}
|
|
deleteYAMLMapping(&document, "admin", "storage")
|
|
deleteYAMLMapping(&document, "admin", "email")
|
|
deleteYAMLMapping(&document, "admin", "websocket")
|
|
deleteYAMLMapping(&document, "admin", "mq")
|
|
if adminConfig.System != nil {
|
|
if err = setServerHTTPPort(&document, adminConfig.System.Addr); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return writeConfigDocument(configPath, &document)
|
|
}
|
|
|
|
// persistDatabaseConfig writes only the database selected on the init page and
|
|
// the freshly generated JWT signing key. Initialization must not serialize the
|
|
// partially populated runtime config back over the template: doing so removes
|
|
// all omitted/default settings from data and admin and leaves the next startup
|
|
// with no visible configuration.
|
|
func (d *Data) persistDatabaseConfig(database *conf.Data_Database, signingKey string) error {
|
|
d.configMu.Lock()
|
|
defer d.configMu.Unlock()
|
|
|
|
configPath := d.runtime.ConfigPath()
|
|
if configPath == "" {
|
|
return nil
|
|
}
|
|
raw, err := os.ReadFile(configPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var document yaml.Node
|
|
if err = yaml.Unmarshal(raw, &document); err != nil {
|
|
return err
|
|
}
|
|
source, err := databaseDSN(database, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Do not marshal the init request through protojson here. Proto3 omits
|
|
// empty and zero values, which makes a merge retain stale values from the
|
|
// template (an empty database password is the common example). Update the
|
|
// fields owned by the init page explicitly and leave advanced database,
|
|
// Redis, Mongo, admin and server settings untouched.
|
|
databaseValue := map[string]any{
|
|
"driver": database.Driver,
|
|
"source": source,
|
|
"host": database.Host,
|
|
"port": database.Port,
|
|
"user": database.User,
|
|
"password": database.Password,
|
|
"name": database.Name,
|
|
"config": database.Config,
|
|
"path": database.Path,
|
|
}
|
|
if err = setYAMLMapping(&document, "data", map[string]any{"database": databaseValue}); err != nil {
|
|
return err
|
|
}
|
|
if signingKey != "" {
|
|
if err = setYAMLMapping(&document, "admin", map[string]any{"jwt": map[string]any{"signing_key": signingKey}}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
deleteYAMLMapping(&document, "admin", "storage")
|
|
deleteYAMLMapping(&document, "admin", "email")
|
|
deleteYAMLMapping(&document, "admin", "websocket")
|
|
deleteYAMLMapping(&document, "admin", "mq")
|
|
return writeConfigDocument(configPath, &document)
|
|
}
|
|
|
|
func (d *Data) removeIntegrationConfigFromFile() error {
|
|
d.configMu.Lock()
|
|
defer d.configMu.Unlock()
|
|
configPath := d.runtime.ConfigPath()
|
|
if configPath == "" {
|
|
return nil
|
|
}
|
|
raw, err := os.ReadFile(configPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var document yaml.Node
|
|
if err = yaml.Unmarshal(raw, &document); err != nil {
|
|
return err
|
|
}
|
|
if yamlMappingValue(&document, "admin", "storage") == nil && yamlMappingValue(&document, "admin", "email") == nil && yamlMappingValue(&document, "admin", "websocket") == nil && yamlMappingValue(&document, "admin", "mq") == nil {
|
|
return nil
|
|
}
|
|
deleteYAMLMapping(&document, "admin", "storage")
|
|
deleteYAMLMapping(&document, "admin", "email")
|
|
deleteYAMLMapping(&document, "admin", "websocket")
|
|
deleteYAMLMapping(&document, "admin", "mq")
|
|
return writeConfigDocument(configPath, &document)
|
|
}
|
|
|
|
func writeConfigDocument(configPath string, document *yaml.Node) error {
|
|
output, err := yaml.Marshal(document)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = os.MkdirAll(filepath.Dir(configPath), 0o755); err != nil {
|
|
return err
|
|
}
|
|
temporary, err := os.CreateTemp(filepath.Dir(configPath), ".kra-config-*.yaml")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
tempName := temporary.Name()
|
|
defer os.Remove(tempName)
|
|
if _, err = temporary.Write(output); err != nil {
|
|
_ = temporary.Close()
|
|
return err
|
|
}
|
|
if err = temporary.Chmod(0o600); err != nil {
|
|
_ = temporary.Close()
|
|
return err
|
|
}
|
|
if err = temporary.Close(); err != nil {
|
|
return err
|
|
}
|
|
return os.Rename(tempName, configPath)
|
|
}
|
|
|
|
func (d *Data) reloadConfig(ctx context.Context) error {
|
|
d.configMu.Lock()
|
|
defer d.configMu.Unlock()
|
|
configPath := d.runtime.ConfigPath()
|
|
if configPath == "" {
|
|
return fmt.Errorf("configuration path is not set")
|
|
}
|
|
next, err := readBootstrap(configPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if next.Data == nil || next.Data.Database == nil || next.Admin == nil {
|
|
return fmt.Errorf("data.database and admin configuration are required")
|
|
}
|
|
next.Admin.ConfigPath = configPath
|
|
databaseReady := databaseConnectionConfigured(next.Data.Database)
|
|
var candidateDB *gorm.DB
|
|
if databaseReady {
|
|
candidateDB, err = openDatabase(next.Data.Database, false, "", d.logger())
|
|
if err != nil {
|
|
return fmt.Errorf("reload database: %w", err)
|
|
}
|
|
} else {
|
|
candidateDB, err = openFallbackDatabase(d.logger())
|
|
if err != nil {
|
|
return fmt.Errorf("reload bootstrap database: %w", err)
|
|
}
|
|
}
|
|
closeCandidate := true
|
|
defer func() {
|
|
if closeCandidate {
|
|
if sqlDB, closeErr := candidateDB.DB(); closeErr == nil {
|
|
_ = sqlDB.Close()
|
|
}
|
|
}
|
|
}()
|
|
if databaseReady {
|
|
if sqlDB, dbErr := candidateDB.DB(); dbErr != nil {
|
|
return dbErr
|
|
} else if err = sqlDB.PingContext(ctx); err != nil {
|
|
return fmt.Errorf("reload database: %w", err)
|
|
}
|
|
}
|
|
if databaseReady && (next.Admin.System == nil || !next.Admin.System.DisableAutoMigrate) {
|
|
if err = migrateAll(candidateDB.WithContext(ctx), d.catalog); err != nil {
|
|
return fmt.Errorf("reload database migrations: %w", err)
|
|
}
|
|
}
|
|
legacyStorage := next.Admin.Storage
|
|
legacyEmail := next.Admin.Email
|
|
legacyWebSocket := next.Admin.Websocket
|
|
legacyMQ := next.Admin.Mq
|
|
currentAdmin := d.runtime.Admin()
|
|
if legacyStorage == nil {
|
|
if currentAdmin != nil {
|
|
legacyStorage = currentAdmin.Storage
|
|
}
|
|
}
|
|
if legacyEmail == nil && currentAdmin != nil {
|
|
legacyEmail = currentAdmin.Email
|
|
}
|
|
if legacyWebSocket == nil && currentAdmin != nil {
|
|
legacyWebSocket = currentAdmin.Websocket
|
|
}
|
|
if legacyMQ == nil && currentAdmin != nil {
|
|
legacyMQ = currentAdmin.Mq
|
|
}
|
|
storageConfig, err := resolveStorageIntegrationConfig(candidateDB.WithContext(ctx), legacyStorage)
|
|
if err != nil {
|
|
return fmt.Errorf("reload storage configuration: %w", err)
|
|
}
|
|
next.Admin.Storage = storageConfig
|
|
emailConfig, err := resolveEmailIntegrationConfig(candidateDB.WithContext(ctx), legacyEmail)
|
|
if err != nil {
|
|
return fmt.Errorf("reload email configuration: %w", err)
|
|
}
|
|
next.Admin.Email = emailConfig
|
|
websocketConfig, err := resolveWebSocketIntegrationConfig(candidateDB.WithContext(ctx), legacyWebSocket)
|
|
if err != nil {
|
|
return fmt.Errorf("reload websocket configuration: %w", err)
|
|
}
|
|
next.Admin.Websocket = websocketConfig
|
|
mqConfig, err := resolveMQIntegrationConfig(candidateDB.WithContext(ctx), legacyMQ)
|
|
if err != nil {
|
|
return fmt.Errorf("reload mq configuration: %w", err)
|
|
}
|
|
next.Admin.Mq = mqConfig
|
|
candidateStorage, err := storage.New(next.Admin)
|
|
if err != nil {
|
|
return fmt.Errorf("reload storage: %w", err)
|
|
}
|
|
useRedis := next.Admin.System != nil && next.Admin.System.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, d.logger())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
d.gormDB.replace(candidateDB, d.enqueueDataScopeAudit)
|
|
d.databaseReady.Store(databaseReady)
|
|
for _, item := range candidateDBList {
|
|
registerDataScopeCallbacks(item, d.enqueueDataScopeAudit)
|
|
}
|
|
d.replaceDatabaseList(candidateDBList)
|
|
d.redis.replace(candidateRedis)
|
|
if mongoErr == nil {
|
|
d.mongo.replace(candidateMongo)
|
|
mongoAccepted = true
|
|
}
|
|
d.runtime.Replace(next.Data, next.Admin)
|
|
if d.storage != nil {
|
|
d.storage.Replace(candidateStorage)
|
|
}
|
|
closeCandidate = false
|
|
return nil
|
|
}
|
|
|
|
func readBootstrap(configPath string) (*conf.Bootstrap, error) {
|
|
raw, err := os.ReadFile(configPath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var plain map[string]any
|
|
if err = yaml.Unmarshal(raw, &plain); err != nil {
|
|
return nil, err
|
|
}
|
|
encoded, err := json.Marshal(plain)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var next conf.Bootstrap
|
|
if err = (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(encoded, &next); err != nil {
|
|
return nil, err
|
|
}
|
|
return &next, nil
|
|
}
|