package data import ( "context" "encoding/json" "errors" "fmt" "strings" "time" "kra/internal/conf" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "gorm.io/gorm" ) const ( integrationKindStorage = "storage" integrationKindEmail = "email" integrationKindPayment = "payment" integrationKindMQ = "mq" ) // integrationConfigPO stores credentials and provider-specific options for // external services. Payment integrations use the same table with kind // "payment", keeping secrets out of the bootstrap configuration file. type integrationConfigPO struct { ID uint `gorm:"primaryKey"` CreatedAt time.Time UpdatedAt time.Time Kind string `gorm:"size:32;not null;uniqueIndex:idx_integration_kind_provider"` Provider string `gorm:"size:64;not null;uniqueIndex:idx_integration_kind_provider"` Enabled bool `gorm:"not null;default:false;index"` Config string `gorm:"type:text;not null"` } func defaultMQIntegrationConfig() *conf.AdminBackend_MQ { return &conf.AdminBackend_MQ{CleanSession: true, KeepAlive: 30, ConnectTimeout: 10} } func saveMQIntegrationConfig(db *gorm.DB, config *conf.AdminBackend_MQ) error { if config == nil { config = defaultMQIntegrationConfig() } raw, err := protojson.MarshalOptions{UseProtoNames: true, EmitDefaultValues: true}.Marshal(config) if err != nil { return fmt.Errorf("encode emqx integration configuration: %w", err) } enabled := config.Enabled && strings.TrimSpace(config.Broker) != "" clean := db.Session(&gorm.Session{NewDB: true}) var row integrationConfigPO err = clean.Where("kind = ? AND provider = ?", integrationKindMQ, "emqx").First(&row).Error switch { case errors.Is(err, gorm.ErrRecordNotFound): return clean.Create(&integrationConfigPO{Kind: integrationKindMQ, Provider: "emqx", Enabled: enabled, Config: string(raw)}).Error case err != nil: return err default: return clean.Model(&row).Updates(map[string]any{"enabled": enabled, "config": string(raw)}).Error } } func loadMQIntegrationConfig(db *gorm.DB) (*conf.AdminBackend_MQ, bool, error) { var row integrationConfigPO err := db.Session(&gorm.Session{NewDB: true}).Where("kind = ? AND provider = ?", integrationKindMQ, "emqx").First(&row).Error if errors.Is(err, gorm.ErrRecordNotFound) { return nil, false, nil } if err != nil { return nil, false, err } if !json.Valid([]byte(row.Config)) { return nil, false, errors.New("invalid emqx integration configuration") } config := defaultMQIntegrationConfig() if err = (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal([]byte(row.Config), config); err != nil { return nil, false, fmt.Errorf("decode emqx integration configuration: %w", err) } config.Enabled = row.Enabled return config, true, nil } func resolveMQIntegrationConfig(db *gorm.DB, legacy *conf.AdminBackend_MQ) (*conf.AdminBackend_MQ, error) { clean := db.Session(&gorm.Session{NewDB: true}) if !clean.Migrator().HasTable(&integrationConfigPO{}) { if legacy == nil { return defaultMQIntegrationConfig(), nil } return proto.Clone(legacy).(*conf.AdminBackend_MQ), nil } loaded, found, err := loadMQIntegrationConfig(clean) if err != nil { return nil, err } if found { return loaded, nil } if legacy == nil { legacy = defaultMQIntegrationConfig() } if err = saveMQIntegrationConfig(clean, legacy); err != nil { return nil, err } loaded, _, err = loadMQIntegrationConfig(clean) return loaded, err } func (d *Data) persistMQIntegrationConfig(ctx context.Context, config *conf.AdminBackend_MQ) error { if !d.databaseReady.Load() { return errors.New("database is not initialized") } db := d.gormDB.WithContext(ctx) if !db.Migrator().HasTable(&integrationConfigPO{}) { return errors.New("integration configuration table does not exist") } return saveMQIntegrationConfig(db, config) } func (integrationConfigPO) TableName() string { return "sys_integration_configs" } var storageProviderNames = []string{ "local", "qiniu", "aliyun-oss", "huawei-obs", "tencent-cos", "aws-s3", "cloudflare-r2", "minio", } func normalizeStorageType(value string) string { value = strings.ToLower(strings.TrimSpace(value)) if value == "" { return "local" } return value } func storageProviderMessage(storage *conf.AdminBackend_Storage, provider string) proto.Message { if storage == nil { storage = &conf.AdminBackend_Storage{} } switch provider { case "qiniu": if storage.Qiniu == nil { storage.Qiniu = &conf.AdminBackend_Qiniu{} } return storage.Qiniu case "aliyun-oss": return ensureObjectStore(&storage.AliyunOss) case "huawei-obs": return ensureObjectStore(&storage.HuaweiObs) case "tencent-cos": return ensureObjectStore(&storage.TencentCos) case "aws-s3": return ensureObjectStore(&storage.AwsS3) case "cloudflare-r2": return ensureObjectStore(&storage.CloudflareR2) case "minio": return ensureObjectStore(&storage.Minio) default: return nil } } func ensureObjectStore(value **conf.AdminBackend_ObjectStore) proto.Message { if *value == nil { *value = &conf.AdminBackend_ObjectStore{} } return *value } func marshalStorageProvider(storage *conf.AdminBackend_Storage, provider string) (string, error) { message := storageProviderMessage(storage, provider) if message == nil { return "{}", nil } raw, err := protojson.MarshalOptions{UseProtoNames: true, EmitDefaultValues: true}.Marshal(message) if err != nil { return "", err } return string(raw), nil } func unmarshalStorageProvider(storage *conf.AdminBackend_Storage, provider, value string) error { if provider == "local" || strings.TrimSpace(value) == "" { return nil } message := storageProviderMessage(storage, provider) if message == nil { return nil } if !json.Valid([]byte(value)) { return fmt.Errorf("invalid %s integration configuration", provider) } if err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal([]byte(value), message); err != nil { return fmt.Errorf("decode %s integration configuration: %w", provider, err) } return nil } func saveStorageIntegrationConfig(db *gorm.DB, storage *conf.AdminBackend_Storage) error { if storage == nil { storage = &conf.AdminBackend_Storage{} } active := normalizeStorageType(storage.Type) known := false for _, provider := range storageProviderNames { if provider == active { known = true break } } if !known { return fmt.Errorf("unsupported storage type %q", active) } return db.Session(&gorm.Session{NewDB: true}).Transaction(func(tx *gorm.DB) error { for _, provider := range storageProviderNames { value, err := marshalStorageProvider(storage, provider) if err != nil { return fmt.Errorf("encode %s integration configuration: %w", provider, err) } var current integrationConfigPO err = tx.Where("kind = ? AND provider = ?", integrationKindStorage, provider).First(¤t).Error switch { case errors.Is(err, gorm.ErrRecordNotFound): current = integrationConfigPO{Kind: integrationKindStorage, Provider: provider} current.Enabled, current.Config = provider == active, value if err = tx.Create(¤t).Error; err != nil { return err } case err != nil: return err default: if err = tx.Model(¤t).Updates(map[string]any{"enabled": provider == active, "config": value}).Error; err != nil { return err } } } return nil }) } func loadStorageIntegrationConfig(db *gorm.DB) (*conf.AdminBackend_Storage, bool, error) { var rows []integrationConfigPO err := db.Session(&gorm.Session{NewDB: true}). Where("kind = ?", integrationKindStorage). Order("id ASC"). Find(&rows).Error if err != nil { return nil, false, err } if len(rows) == 0 { return nil, false, nil } storage := &conf.AdminBackend_Storage{Type: "local"} for _, row := range rows { if err = unmarshalStorageProvider(storage, row.Provider, row.Config); err != nil { return nil, false, err } if row.Enabled { storage.Type = row.Provider } } return storage, true, nil } // resolveStorageIntegrationConfig upgrades a legacy YAML configuration only // when the database has no storage rows yet. From then on the database is the // sole source of truth. func resolveStorageIntegrationConfig(db *gorm.DB, legacy *conf.AdminBackend_Storage) (*conf.AdminBackend_Storage, error) { clean := db.Session(&gorm.Session{NewDB: true}) if !clean.Migrator().HasTable(&integrationConfigPO{}) { if legacy == nil { return &conf.AdminBackend_Storage{Type: "local"}, nil } return proto.Clone(legacy).(*conf.AdminBackend_Storage), nil } storage, found, err := loadStorageIntegrationConfig(clean) if err != nil { return nil, err } if found { return storage, nil } if legacy == nil { legacy = &conf.AdminBackend_Storage{Type: "local"} } if err = saveStorageIntegrationConfig(clean, legacy); err != nil { return nil, err } storage, _, err = loadStorageIntegrationConfig(clean) return storage, err } func (d *Data) persistStorageIntegrationConfig(ctx context.Context, storage *conf.AdminBackend_Storage) error { if !d.databaseReady.Load() { return errors.New("database is not initialized") } db := d.gormDB.WithContext(ctx) if !db.Migrator().HasTable(&integrationConfigPO{}) { return errors.New("integration configuration table does not exist") } return saveStorageIntegrationConfig(db, storage) } func defaultEmailIntegrationConfig() *conf.AdminBackend_Email { return &conf.AdminBackend_Email{Port: 465, IsSsl: true} } func saveEmailIntegrationConfig(db *gorm.DB, email *conf.AdminBackend_Email) error { if email == nil { email = defaultEmailIntegrationConfig() } raw, err := protojson.MarshalOptions{UseProtoNames: true, EmitDefaultValues: true}.Marshal(email) if err != nil { return fmt.Errorf("encode smtp integration configuration: %w", err) } enabled := email.Host != "" && email.From != "" && email.Secret != "" && email.Port > 0 clean := db.Session(&gorm.Session{NewDB: true}) var current integrationConfigPO err = clean.Where("kind = ? AND provider = ?", integrationKindEmail, "smtp").First(¤t).Error switch { case errors.Is(err, gorm.ErrRecordNotFound): return clean.Create(&integrationConfigPO{ Kind: integrationKindEmail, Provider: "smtp", Enabled: enabled, Config: string(raw), }).Error case err != nil: return err default: return clean.Model(¤t).Updates(map[string]any{"enabled": enabled, "config": string(raw)}).Error } } func loadEmailIntegrationConfig(db *gorm.DB) (*conf.AdminBackend_Email, bool, error) { var row integrationConfigPO err := db.Session(&gorm.Session{NewDB: true}). Where("kind = ? AND provider = ?", integrationKindEmail, "smtp"). First(&row).Error if errors.Is(err, gorm.ErrRecordNotFound) { return nil, false, nil } if err != nil { return nil, false, err } if !json.Valid([]byte(row.Config)) { return nil, false, errors.New("invalid smtp integration configuration") } email := defaultEmailIntegrationConfig() if err = (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal([]byte(row.Config), email); err != nil { return nil, false, fmt.Errorf("decode smtp integration configuration: %w", err) } return email, true, nil } func resolveEmailIntegrationConfig(db *gorm.DB, legacy *conf.AdminBackend_Email) (*conf.AdminBackend_Email, error) { clean := db.Session(&gorm.Session{NewDB: true}) if !clean.Migrator().HasTable(&integrationConfigPO{}) { if legacy == nil { return defaultEmailIntegrationConfig(), nil } return proto.Clone(legacy).(*conf.AdminBackend_Email), nil } email, found, err := loadEmailIntegrationConfig(clean) if err != nil { return nil, err } if found { return email, nil } if legacy == nil { legacy = defaultEmailIntegrationConfig() } if err = saveEmailIntegrationConfig(clean, legacy); err != nil { return nil, err } email, _, err = loadEmailIntegrationConfig(clean) return email, err } func (d *Data) persistEmailIntegrationConfig(ctx context.Context, email *conf.AdminBackend_Email) error { if !d.databaseReady.Load() { return errors.New("database is not initialized") } db := d.gormDB.WithContext(ctx) if !db.Migrator().HasTable(&integrationConfigPO{}) { return errors.New("integration configuration table does not exist") } return saveEmailIntegrationConfig(db, email) }