kra-new/internal/data/initialization_backend.go

package data

import (
	"context"
	"errors"
	"fmt"
	"kra/internal/biz/system"

	"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
	}
	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.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
	}
	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.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 *system.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)
	}
	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.Email = emailConfig
	d.runtime.Replace(currentData, currentAdmin)
	if err = d.loadIntegrationRuntime(candidate); err != nil {
		return fmt.Errorf("initialize integration runtime: %w", err)
	}
	activated = true
	return nil
}