kra-new/internal/data/integration/migrations.go

package integration

import (
	"encoding/json"
	"errors"
	"reflect"
	"strings"

	integrationbiz "kra/internal/biz/integration"
	"kra/pkg/database/migration"

	"gorm.io/gorm"
)

func Migrations() []migration.Step {
	return []migration.Step{
		{ID: "202608200001_data_infrastructure", Migrate: func(db *gorm.DB) error {
			return migration.CreateMissingTables(db, &ConfigPO{})
		}},
		{ID: "202608210001_communication_integration_defaults", Migrate: ensureCommunicationIntegrationConfigs},
		{ID: "202608200004_payment_defaults", Migrate: ensurePaymentIntegrationConfigs},
		{ID: "202608290004_kafka_integration_default", Migrate: ensureKafkaIntegrationConfig},
		{ID: "202608290005_communication_integration_default_upgrade", Migrate: upgradeCommunicationIntegrationDefaults},
	}
}

func ensureCommunicationIntegrationConfigs(db *gorm.DB) error {
	defaults := []struct{ kind, provider string }{
		{integrationbiz.IntegrationKindMQ, "emqx"},
		{integrationbiz.IntegrationKindMQ, "kafka"},
		{integrationbiz.IntegrationKindMQ, "rabbitmq"},
		{integrationbiz.IntegrationKindWebSocket, "melody"},
	}
	for _, item := range defaults {
		if err := ensureIntegrationConfig(db, item.kind, item.provider); err != nil {
			return err
		}
	}
	return nil
}

func ensureKafkaIntegrationConfig(db *gorm.DB) error {
	return ensureIntegrationConfig(db, integrationbiz.IntegrationKindMQ, "kafka")
}

func ensureIntegrationConfig(db *gorm.DB, kind, provider string) error {
	var row ConfigPO
	err := db.Where("kind = ? AND provider = ?", kind, provider).First(&row).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		values, marshalErr := json.Marshal(integrationbiz.DefaultIntegrationConfig(kind, provider))
		if marshalErr != nil {
			return marshalErr
		}
		return db.Create(&ConfigPO{Kind: kind, Provider: provider, Enabled: false, Config: string(values)}).Error
	}
	return err
}

func upgradeCommunicationIntegrationDefaults(db *gorm.DB) error {
	for _, kind := range []string{integrationbiz.IntegrationKindMQ, integrationbiz.IntegrationKindWebSocket} {
		for _, definition := range integrationbiz.IntegrationDefinitions(kind) {
			var row ConfigPO
			err := db.Where("kind = ? AND provider = ?", kind, definition.Provider).First(&row).Error
			if errors.Is(err, gorm.ErrRecordNotFound) {
				continue
			}
			if err != nil {
				return err
			}
			values := integrationObject(json.RawMessage(row.Config))
			changed := false
			for key, value := range definition.Defaults {
				if _, exists := values[key]; !exists {
					values[key] = value
					changed = true
				}
			}
			for _, field := range definition.Fields {
				if !field.Required || !emptyIntegrationConfigValue(values[field.Key]) || emptyIntegrationConfigValue(definition.Defaults[field.Key]) {
					continue
				}
				values[field.Key] = definition.Defaults[field.Key]
				changed = true
			}
			if _, exists := values["enabled"]; exists {
				delete(values, "enabled")
				changed = true
			}
			if !changed {
				continue
			}
			encoded, marshalErr := json.Marshal(values)
			if marshalErr != nil {
				return marshalErr
			}
			if err := db.Model(&row).Update("config", string(encoded)).Error; err != nil {
				return err
			}
		}
	}
	return nil
}

func emptyIntegrationConfigValue(value any) bool {
	switch typed := value.(type) {
	case nil:
		return true
	case string:
		return strings.TrimSpace(typed) == ""
	case []any:
		return len(typed) == 0
	case []string:
		return len(typed) == 0
	default:
		return false
	}
}

func ensurePaymentIntegrationConfigs(db *gorm.DB) error {
	for _, definition := range integrationbiz.IntegrationDefinitions(integrationbiz.IntegrationKindPayment) {
		provider := definition.Provider
		var row ConfigPO
		err := db.Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindPayment, provider).First(&row).Error
		defaults := integrationbiz.DefaultIntegrationConfig(integrationbiz.IntegrationKindPayment, provider)
		if errors.Is(err, gorm.ErrRecordNotFound) {
			encoded, marshalErr := json.Marshal(defaults)
			if marshalErr != nil {
				return marshalErr
			}
			if err := db.Create(&ConfigPO{Kind: integrationbiz.IntegrationKindPayment, Provider: provider, Config: string(encoded)}).Error; err != nil {
				return err
			}
			continue
		}
		if err != nil {
			return err
		}
		values := integrationObject(json.RawMessage(row.Config))
		merged := integrationbiz.MergeIntegrationDefaults(defaults, values)
		changed := !reflect.DeepEqual(values, merged)
		values = merged
		if changed {
			encoded, marshalErr := json.Marshal(values)
			if marshalErr != nil {
				return marshalErr
			}
			if err := db.Model(&row).Update("config", string(encoded)).Error; err != nil {
				return err
			}
		}
	}
	return nil
}