优化结构

internal/biz/integration_config.go

@@ -10,7 +10,11 @@ import (
 	"strings"
 )
 
-const IntegrationKindPayment = "payment"
+const (
+	IntegrationKindPayment   = "payment"
+	IntegrationKindMQ        = "mq"
+	IntegrationKindWebSocket = "websocket"
+)
 
 type IntegrationConfig struct {
 	Kind     string
@@ -150,10 +154,79 @@ func mergeIntegrationDefaults(defaults, values map[string]any) map[string]any {
 }
 
 func ValidateIntegrationConfig(kind, provider string, values map[string]any) error {
-	if normalizeIntegrationPart(kind) != IntegrationKindPayment {
+	kind = normalizeIntegrationPart(kind)
+	provider = normalizeIntegrationPart(provider)
+	switch kind {
+	case IntegrationKindPayment:
+		return validatePaymentIntegrationConfig(provider, values)
+	case IntegrationKindMQ, IntegrationKindWebSocket:
+		return validateCommunicationIntegrationConfig(kind, provider, values)
+	default:
 		return nil
 	}
-	return validatePaymentIntegrationConfig(normalizeIntegrationPart(provider), values)
+}
+
+func validateCommunicationIntegrationConfig(kind, provider string, values map[string]any) error {
+	definition, ok := IntegrationDefinition(kind, provider)
+	if !ok {
+		return errors.New("不支持的通信集成")
+	}
+	for _, field := range definition.Fields {
+		if field.Required && integrationText(values, field.Key) == "" {
+			return fmt.Errorf("%s 缺少配置字段 %s", provider, field.Key)
+		}
+	}
+
+	switch kind + "/" + provider {
+	case IntegrationKindMQ + "/emqx":
+		broker := strings.ToLower(integrationText(values, "broker"))
+		if !strings.HasPrefix(broker, "tcp://") && !strings.HasPrefix(broker, "ssl://") && !strings.HasPrefix(broker, "ws://") && !strings.HasPrefix(broker, "wss://") && !strings.HasPrefix(broker, "mqtt://") {
+			return errors.New("emqx broker 必须使用 tcp、ssl、ws、wss 或 mqtt 协议")
+		}
+		if keepAlive := integrationInt64(values, "keep_alive", 0); keepAlive <= 0 {
+			return errors.New("emqx keep_alive 必须大于 0")
+		}
+		if timeout := integrationInt64(values, "connect_timeout", 0); timeout <= 0 {
+			return errors.New("emqx connect_timeout 必须大于 0")
+		}
+	case IntegrationKindMQ + "/rabbitmq":
+		port := integrationInt64(values, "port", 0)
+		if port < 1 || port > 65535 {
+			return errors.New("rabbitmq port 必须在 1-65535 之间")
+		}
+		exchangeType := strings.ToLower(integrationText(values, "exchange_type"))
+		if exchangeType != "direct" && exchangeType != "fanout" && exchangeType != "topic" && exchangeType != "headers" {
+			return errors.New("rabbitmq exchange_type 必须是 direct、fanout、topic 或 headers")
+		}
+		if integrationInt64(values, "prefetch_count", -1) < 0 {
+			return errors.New("rabbitmq prefetch_count 不能小于 0")
+		}
+		if integrationInt64(values, "heartbeat", -1) < 0 {
+			return errors.New("rabbitmq heartbeat 不能小于 0")
+		}
+		if integrationInt64(values, "connect_timeout", 0) <= 0 {
+			return errors.New("rabbitmq connect_timeout 必须大于 0")
+		}
+	case IntegrationKindWebSocket + "/melody":
+		path := integrationText(values, "path")
+		if !strings.HasPrefix(path, "/") {
+			return errors.New("websocket path 必须以 / 开头")
+		}
+		for _, key := range []string{"write_wait", "pong_wait", "ping_period"} {
+			value := integrationText(values, key)
+			if value == "" {
+				continue
+			}
+			duration, err := time.ParseDuration(value)
+			if err != nil || duration <= 0 {
+				return fmt.Errorf("websocket %s 必须是大于 0 的时长", key)
+			}
+		}
+		if integrationInt64(values, "max_message_size", -1) < 0 || integrationInt64(values, "message_buffer_size", -1) < 0 {
+			return errors.New("websocket 消息大小和缓冲区不能小于 0")
+		}
+	}
+	return nil
 }
 
 func validatePaymentIntegrationConfig(provider string, values map[string]any) error {

internal/biz/integration_config_definition.go

@@ -89,6 +89,58 @@ func genericPaymentDefinition(provider, name, description string) IntegrationCon
 }
 
 var integrationDefinitions = map[string][]IntegrationConfigDefinition{
+	IntegrationKindMQ: {
+		{
+			Kind: IntegrationKindMQ, Provider: "emqx", Name: "EMQX", Description: "EMQX MQTT 消息服务",
+			Defaults: map[string]any{"broker": "tcp://127.0.0.1:1883", "client_id": "kra", "username": "", "password": "", "keep_alive": 30, "clean_session": true, "connect_timeout": 10},
+			Fields: []IntegrationConfigField{
+				{Key: "broker", Label: "Broker 地址", Type: "text", Required: true, Placeholder: "tcp://127.0.0.1:1883"},
+				{Key: "client_id", Label: "客户端 ID", Type: "text", Required: true, Placeholder: "kra"},
+				{Key: "username", Label: "用户名", Type: "text"},
+				{Key: "password", Label: "密码", Type: "password", Secret: true},
+				{Key: "keep_alive", Label: "心跳间隔（秒）", Type: "number", Required: true},
+				{Key: "clean_session", Label: "清理会话", Type: "switch", Description: "连接时不恢复 Broker 端保存的旧会话。"},
+				{Key: "connect_timeout", Label: "连接超时（秒）", Type: "number", Required: true},
+			},
+		},
+		{
+			Kind: IntegrationKindMQ, Provider: "rabbitmq", Name: "RabbitMQ", Description: "RabbitMQ AMQP 消息队列",
+			Defaults: map[string]any{"host": "127.0.0.1", "port": 5672, "username": "guest", "password": "guest", "vhost": "/", "exchange": "kra", "exchange_type": "topic", "queue": "kra", "routing_key": "#", "durable": true, "auto_delete": false, "prefetch_count": 10, "heartbeat": 10, "connect_timeout": 10, "tls": false},
+			Fields: []IntegrationConfigField{
+				{Key: "host", Label: "主机", Type: "text", Required: true, Placeholder: "127.0.0.1"},
+				{Key: "port", Label: "端口", Type: "number", Required: true},
+				{Key: "username", Label: "用户名", Type: "text", Required: true},
+				{Key: "password", Label: "密码", Type: "password", Required: true, Secret: true},
+				{Key: "vhost", Label: "Virtual Host", Type: "text", Required: true, Placeholder: "/"},
+				{Key: "exchange", Label: "Exchange", Type: "text", Required: true, Placeholder: "kra"},
+				{Key: "exchange_type", Label: "Exchange 类型", Type: "select", Required: true, Options: []IntegrationConfigOption{{Label: "topic", Value: "topic"}, {Label: "direct", Value: "direct"}, {Label: "fanout", Value: "fanout"}, {Label: "headers", Value: "headers"}}},
+				{Key: "queue", Label: "Queue", Type: "text", Required: true, Placeholder: "kra"},
+				{Key: "routing_key", Label: "默认 Routing Key", Type: "text", Required: true, Placeholder: "#", Description: "业务未指定订阅键时使用；topic 类型支持 * 和 #。"},
+				{Key: "durable", Label: "持久化", Type: "switch"},
+				{Key: "auto_delete", Label: "自动删除", Type: "switch"},
+				{Key: "prefetch_count", Label: "预取数量", Type: "number"},
+				{Key: "heartbeat", Label: "心跳间隔（秒）", Type: "number"},
+				{Key: "connect_timeout", Label: "连接超时（秒）", Type: "number", Required: true},
+				{Key: "tls", Label: "启用 TLS", Type: "switch"},
+			},
+		},
+	},
+	IntegrationKindWebSocket: {
+		{
+			Kind: IntegrationKindWebSocket, Provider: "melody", Name: "WebSocket", Description: "WebSocket 实时连接服务",
+			Defaults: map[string]any{"path": "/ws", "allow_origins": []string{}, "max_message_size": 0, "write_wait": "10s", "pong_wait": "60s", "ping_period": "54s", "message_buffer_size": 0, "concurrent_message_handling": false},
+			Fields: []IntegrationConfigField{
+				{Key: "path", Label: "访问路径", Type: "text", Required: true, Placeholder: "/ws"},
+				{Key: "allow_origins", Label: "允许的来源", Type: "string-list", Placeholder: "https://admin.example.com", Description: "每行一个 Origin；留空时沿用 WebSocket 组件默认策略。"},
+				{Key: "max_message_size", Label: "最大消息字节数", Type: "number", Description: "0 表示使用组件默认值。"},
+				{Key: "write_wait", Label: "写入超时", Type: "text", Required: true, Placeholder: "10s"},
+				{Key: "pong_wait", Label: "Pong 等待时间", Type: "text", Required: true, Placeholder: "60s"},
+				{Key: "ping_period", Label: "Ping 间隔", Type: "text", Required: true, Placeholder: "54s"},
+				{Key: "message_buffer_size", Label: "消息缓冲区", Type: "number", Description: "0 表示不额外缓冲。"},
+				{Key: "concurrent_message_handling", Label: "并发处理消息", Type: "switch"},
+			},
+		},
+	},
 	IntegrationKindPayment: {
 		paymentDefinition(PaymentAlipay, "支付宝", "支付宝 OpenAPI RSA2 支付", map[string]any{"app_id": "", "private_key": "", "public_key": "", "environment": "production", "sign_type": "RSA2", "gateway_url": "https://openapi.alipay.com/gateway.do", "method": "alipay.trade.create"},
 			integrationField("app_id", "应用 ID", true, false, "text"), integrationField("private_key", "应用私钥", true, true, "textarea"), integrationField("public_key", "支付宝公钥", true, true, "textarea"),

internal/conf/conf.proto

@@ -93,8 +93,6 @@ message AdminBackend {
   Zap zap = 11;
   CORS cors = 12;
   App app = 13;
-  WebSocket websocket = 14;
-  MQ mq = 15;
 
   message JWT {
     string signing_key = 1;
@@ -181,31 +179,6 @@ message AdminBackend {
     string env = 3;
   }
 
-  message WebSocket {
-    bool enabled = 1;
-    string path = 2;
-    repeated string allow_origins = 3;
-    int64 max_message_size = 4;
-    google.protobuf.Duration write_wait = 5;
-    google.protobuf.Duration pong_wait = 6;
-    google.protobuf.Duration ping_period = 7;
-    int32 message_buffer_size = 8;
-    bool concurrent_message_handling = 9;
-  }
-
-  // MQ config is persisted in sys_integration_configs (kind=mq/provider=emqx).
-  // The bootstrap fields are retained as a one-time migration source.
-  message MQ {
-    bool enabled = 1;
-    string broker = 2;
-    string client_id = 3;
-    string username = 4;
-    string password = 5;
-    int32 keep_alive = 6;
-    bool clean_session = 7;
-    int32 connect_timeout = 8;
-  }
-
   message Storage {
     string type = 1;
     Qiniu qiniu = 2;

internal/data/config_store.go

@@ -182,8 +182,6 @@ func (d *Data) persistConfigValuesLocked(dataConfig *conf.Data, adminConfig *con
 	fileAdmin := cloneAdminConfig(adminConfig)
 	fileAdmin.Storage = nil
 	fileAdmin.Email = nil
-	fileAdmin.Websocket = nil
-	fileAdmin.Mq = nil
 	adminValue, err := protoMap(fileAdmin)
 	if err != nil {
 		return err
@@ -365,8 +363,6 @@ func (d *Data) reloadConfig(ctx context.Context) error {
 	}
 	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 {
@@ -376,12 +372,6 @@ func (d *Data) reloadConfig(ctx context.Context) error {
 	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)
@@ -392,16 +382,6 @@ func (d *Data) reloadConfig(ctx context.Context) error {
 		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)
@@ -436,6 +416,9 @@ func (d *Data) reloadConfig(ctx context.Context) error {
 		mongoAccepted = true
 	}
 	d.runtime.Replace(next.Data, next.Admin)
+	if err = d.loadIntegrationRuntime(candidateDB); err != nil {
+		return fmt.Errorf("reload integration runtime: %w", err)
+	}
 	if d.storage != nil {
 		d.storage.Replace(candidateStorage)
 	}

internal/data/config_watch.go

@@ -58,8 +58,6 @@ func (d *Data) watchConfig() func() {
 					if current := d.runtime.Admin(); current != nil {
 						next.Admin.Storage = current.Storage
 						next.Admin.Email = current.Email
-						next.Admin.Mq = current.Mq
-						next.Admin.Websocket = current.Websocket
 					}
 					next.Admin.ConfigPath = absolute
 					d.runtime.Replace(next.Data, next.Admin)

internal/data/data.go

@@ -15,6 +15,7 @@ import (
 	datapayment "kra/internal/data/payment"
 	datasystem "kra/internal/data/repository"
 	"kra/internal/integration/storage"
+	"kra/internal/integrationruntime"
 	"kra/pkg/module"
 )
 
@@ -41,6 +42,7 @@ type Data struct {
 	redis         *reloadableRedis
 	mongo         *reloadableMongo
 	runtime       *conf.Runtime
+	integrations  *integrationruntime.Store
 	storage       *storage.Reloadable
 	dbListMu      sync.RWMutex
 	dbList        map[string]*gorm.DB
@@ -73,6 +75,15 @@ func (d *Data) Runtime() *conf.Runtime {
 	return d.runtime
 }
 
+// IntegrationRuntime exposes database-backed integration configuration to
+// long-lived adapters without making config.yaml part of their lifecycle.
+func (d *Data) IntegrationRuntime() *integrationruntime.Store {
+	if d == nil {
+		return nil
+	}
+	return d.integrations
+}
+
 // Database resolves the primary or a named database for repositories such as
 // the system export module.
 func (d *Data) Database(name string) (*gorm.DB, error) {
@@ -156,7 +167,7 @@ func NewData(runtime *conf.Runtime, appLogger *slog.Logger, storageManager *stor
 		// and /init/initdb remain available.
 		c.Database = &conf.Data_Database{}
 	}
-	d := &Data{runtime: runtime, appLogger: appLogger, storage: storageManager, catalog: catalog}
+	d := &Data{runtime: runtime, integrations: integrationruntime.NewStore(), appLogger: appLogger, storage: storageManager, catalog: catalog}
 	usingFallback := !databaseConnectionConfigured(c.Database)
 	var db *gorm.DB
 	var err error
@@ -208,17 +219,10 @@ func NewData(runtime *conf.Runtime, appLogger *slog.Logger, storageManager *stor
 		}
 		admin.Storage = storageConfig
 		admin.Email = emailConfig
-		websocketConfig, websocketErr := resolveWebSocketIntegrationConfig(db, admin.Websocket)
-		if websocketErr != nil {
-			return nil, nil, fmt.Errorf("load websocket integration configuration: %w", websocketErr)
-		}
-		admin.Websocket = websocketConfig
-		mqConfig, mqErr := resolveMQIntegrationConfig(db, admin.Mq)
-		if mqErr != nil {
-			return nil, nil, fmt.Errorf("load mq integration configuration: %w", mqErr)
-		}
-		admin.Mq = mqConfig
 		runtime.Replace(c, admin)
+		if err = d.loadIntegrationRuntime(db); err != nil {
+			return nil, nil, fmt.Errorf("load integration runtime: %w", err)
+		}
 		activeStorage, storageErr := storage.New(admin)
 		if storageErr != nil {
 			return nil, nil, fmt.Errorf("initialize storage: %w", storageErr)

internal/data/initialization_backend.go

@@ -69,12 +69,6 @@ func (d *Data) PersistAdminConfig(ctx context.Context, raw []byte) error {
 	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 {
@@ -86,12 +80,6 @@ func (d *Data) PersistAdminConfig(ctx context.Context, raw []byte) error {
 	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
 	}
@@ -120,12 +108,6 @@ func (d *Data) PersistRuntimeConfig(ctx context.Context, dataRaw, adminRaw []byt
 	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 {
@@ -137,12 +119,6 @@ func (d *Data) PersistRuntimeConfig(ctx context.Context, dataRaw, adminRaw []byt
 	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
 	}
@@ -229,14 +205,6 @@ func (d *Data) InitializeDatabase(ctx context.Context, input *biz.DatabaseConfig
 	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)
@@ -251,14 +219,11 @@ func (d *Data) InitializeDatabase(ctx context.Context, input *biz.DatabaseConfig
 	}
 	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)
+	if err = d.loadIntegrationRuntime(candidate); err != nil {
+		return fmt.Errorf("initialize integration runtime: %w", err)
+	}
 	activated = true
 	return nil
 }

internal/data/integration_config.go

@@ -19,7 +19,6 @@ const (
 	integrationKindStorage = "storage"
 	integrationKindEmail   = "email"
 	integrationKindPayment = "payment"
-	integrationKindMQ      = "mq"
 )
 
 // integrationConfigPO stores credentials and provider-specific options for
@@ -35,88 +34,6 @@ type integrationConfigPO struct {
 	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{

internal/data/integration_runtime.go

@@ -0,0 +1,38 @@
+package data
+
+import (
+	"errors"
+
+	"kra/internal/integrationruntime"
+
+	"gorm.io/gorm"
+)
+
+func readIntegrationRuntime(db *gorm.DB) ([]integrationruntime.Config, error) {
+	if db == nil || !db.Migrator().HasTable(&integrationConfigPO{}) {
+		return nil, nil
+	}
+	var rows []integrationConfigPO
+	if err := db.Session(&gorm.Session{NewDB: true}).
+		Where("kind IN ?", []string{"mq", "websocket"}).
+		Order("kind ASC, provider ASC").
+		Find(&rows).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
+		return nil, err
+	}
+	configs := make([]integrationruntime.Config, 0, len(rows))
+	for _, row := range rows {
+		configs = append(configs, integrationruntime.Config{Kind: row.Kind, Provider: row.Provider, Enabled: row.Enabled, Values: []byte(row.Config)})
+	}
+	return configs, nil
+}
+
+func (d *Data) loadIntegrationRuntime(db *gorm.DB) error {
+	configs, err := readIntegrationRuntime(db)
+	if err != nil {
+		return err
+	}
+	if d.integrations != nil {
+		d.integrations.Replace(configs)
+	}
+	return nil
+}

internal/data/repository/integration_config.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"kra/internal/biz"
+	"kra/internal/integrationruntime"
 
 	"gorm.io/gorm"
 )
@@ -65,7 +66,11 @@ func (r *integrationConfigRepo) SaveIntegrationConfig(ctx context.Context, confi
 			}
 		}
 		encoded, _ := json.Marshal(values)
-		return db.Create(&integrationConfigPO{Kind: config.Kind, Provider: config.Provider, Enabled: config.Enabled, Config: string(encoded)}).Error
+		if err := db.Create(&integrationConfigPO{Kind: config.Kind, Provider: config.Provider, Enabled: config.Enabled, Config: string(encoded)}).Error; err != nil {
+			return err
+		}
+		r.publish(config.Kind, config.Provider, config.Enabled, encoded)
+		return nil
 	}
 	if err != nil {
 		return err
@@ -77,11 +82,27 @@ func (r *integrationConfigRepo) SaveIntegrationConfig(ctx context.Context, confi
 		}
 	}
 	encoded, _ := json.Marshal(values)
-	return db.Model(&row).Updates(map[string]any{"enabled": config.Enabled, "config": string(encoded)}).Error
+	if err := db.Model(&row).Updates(map[string]any{"enabled": config.Enabled, "config": string(encoded)}).Error; err != nil {
+		return err
+	}
+	r.publish(config.Kind, config.Provider, config.Enabled, encoded)
+	return nil
 }
 
 func (r *integrationConfigRepo) DeleteIntegrationConfig(ctx context.Context, kind, provider string) error {
-	return r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", kind, provider).Delete(&integrationConfigPO{}).Error
+	if err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", kind, provider).Delete(&integrationConfigPO{}).Error; err != nil {
+		return err
+	}
+	if runtime := r.data.IntegrationRuntime(); runtime != nil {
+		runtime.Delete(kind, provider)
+	}
+	return nil
+}
+
+func (r *integrationConfigRepo) publish(kind, provider string, enabled bool, values []byte) {
+	if runtime := r.data.IntegrationRuntime(); runtime != nil {
+		runtime.Set(integrationruntime.Config{Kind: kind, Provider: provider, Enabled: enabled, Values: values})
+	}
 }
 
 func integrationConfigFromPO(row integrationConfigPO) *biz.IntegrationConfig {

internal/data/repository/provider.go

@@ -2,6 +2,7 @@ package system
 
 import (
 	"kra/internal/conf"
+	"kra/internal/integrationruntime"
 
 	"gorm.io/gorm"
 )
@@ -13,4 +14,5 @@ type Provider interface {
 	Database(name string) (*gorm.DB, error)
 	DatabaseReady() bool
 	Runtime() *conf.Runtime
+	IntegrationRuntime() *integrationruntime.Store
 }

internal/data/websocket_config.go

@@ -1,96 +0,0 @@
-package data
-
-import (
-	"context"
-	"encoding/json"
-	"errors"
-	"fmt"
-
-	"kra/internal/conf"
-
-	"google.golang.org/protobuf/encoding/protojson"
-	"google.golang.org/protobuf/proto"
-	"gorm.io/gorm"
-)
-
-const integrationKindWebSocket = "websocket"
-
-// saveWebSocketIntegrationConfig persists the Melody settings in the shared
-// integration table. It intentionally lives separately from storage/payment
-// persistence so adding another transport does not expand their API surface.
-func saveWebSocketIntegrationConfig(db *gorm.DB, config *conf.AdminBackend_WebSocket) error {
-	if config == nil {
-		config = &conf.AdminBackend_WebSocket{}
-	}
-	raw, err := protojson.MarshalOptions{UseProtoNames: true, EmitDefaultValues: true}.Marshal(config)
-	if err != nil {
-		return fmt.Errorf("encode websocket integration configuration: %w", err)
-	}
-	clean := db.Session(&gorm.Session{NewDB: true})
-	var current integrationConfigPO
-	err = clean.Where("kind = ? AND provider = ?", integrationKindWebSocket, "melody").First(&current).Error
-	switch {
-	case errors.Is(err, gorm.ErrRecordNotFound):
-		return clean.Create(&integrationConfigPO{Kind: integrationKindWebSocket, Provider: "melody", Enabled: config.Enabled, Config: string(raw)}).Error
-	case err != nil:
-		return err
-	default:
-		return clean.Model(&current).Updates(map[string]any{"enabled": config.Enabled, "config": string(raw)}).Error
-	}
-}
-
-func loadWebSocketIntegrationConfig(db *gorm.DB) (*conf.AdminBackend_WebSocket, bool, error) {
-	var row integrationConfigPO
-	err := db.Session(&gorm.Session{NewDB: true}).Where("kind = ? AND provider = ?", integrationKindWebSocket, "melody").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 websocket integration configuration")
-	}
-	config := &conf.AdminBackend_WebSocket{}
-	if err = (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal([]byte(row.Config), config); err != nil {
-		return nil, false, fmt.Errorf("decode websocket integration configuration: %w", err)
-	}
-	config.Enabled = row.Enabled
-	return config, true, nil
-}
-
-func resolveWebSocketIntegrationConfig(db *gorm.DB, legacy *conf.AdminBackend_WebSocket) (*conf.AdminBackend_WebSocket, error) {
-	clean := db.Session(&gorm.Session{NewDB: true})
-	if !clean.Migrator().HasTable(&integrationConfigPO{}) {
-		if legacy == nil {
-			return &conf.AdminBackend_WebSocket{}, nil
-		}
-		return proto.Clone(legacy).(*conf.AdminBackend_WebSocket), nil
-	}
-	config, found, err := loadWebSocketIntegrationConfig(clean)
-	if err != nil {
-		return nil, err
-	}
-	if found {
-		return config, nil
-	}
-	if legacy == nil {
-		legacy = &conf.AdminBackend_WebSocket{}
-	}
-	if err = saveWebSocketIntegrationConfig(clean, legacy); err != nil {
-		return nil, err
-	}
-	config, _, err = loadWebSocketIntegrationConfig(clean)
-	return config, err
-}
-
-func (d *Data) PersistWebSocketConfig(ctx context.Context, config *conf.AdminBackend_WebSocket) 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 saveWebSocketIntegrationConfig(db, config)
-}

internal/data/websocket_config_test.go

@@ -1,50 +0,0 @@
-package data
-
-import (
-	"testing"
-	"time"
-
-	"kra/internal/conf"
-
-	"google.golang.org/protobuf/types/known/durationpb"
-)
-
-func TestWebSocketIntegrationConfigRoundTrip(t *testing.T) {
-	db := openIntegrationConfigTestDB(t)
-	want := &conf.AdminBackend_WebSocket{
-		Enabled: true, Path: "/events", AllowOrigins: []string{"https://admin.example.com"},
-		MaxMessageSize: 4096, WriteWait: durationpb.New(3 * time.Second),
-		PongWait: durationpb.New(20 * time.Second), PingPeriod: durationpb.New(15 * time.Second),
-		MessageBufferSize: 32, ConcurrentMessageHandling: true,
-	}
-	if err := saveWebSocketIntegrationConfig(db, want); err != nil {
-		t.Fatal(err)
-	}
-	got, found, err := loadWebSocketIntegrationConfig(db)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if !found {
-		t.Fatal("websocket integration configuration was not found")
-	}
-	if !got.Enabled || got.Path != want.Path || got.MaxMessageSize != want.MaxMessageSize || got.MessageBufferSize != want.MessageBufferSize {
-		t.Fatalf("loaded websocket config = %#v", got)
-	}
-	if len(got.AllowOrigins) != 1 || got.AllowOrigins[0] != want.AllowOrigins[0] {
-		t.Fatalf("allow origins = %v", got.AllowOrigins)
-	}
-}
-
-func TestResolveWebSocketIntegrationConfigPrefersDatabase(t *testing.T) {
-	db := openIntegrationConfigTestDB(t)
-	if err := saveWebSocketIntegrationConfig(db, &conf.AdminBackend_WebSocket{Enabled: true, Path: "/database"}); err != nil {
-		t.Fatal(err)
-	}
-	got, err := resolveWebSocketIntegrationConfig(db, &conf.AdminBackend_WebSocket{Path: "/legacy"})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if got.Path != "/database" || !got.Enabled {
-		t.Fatalf("resolved websocket config = %#v", got)
-	}
-}

internal/initialize/configuration.go

@@ -89,16 +89,6 @@ func (r *Repo) ConfigurationJSON() (json.RawMessage, error) {
 		if adminConfig.App != nil {
 			admin["app"] = adminConfig.App
 		}
-		if adminConfig.Websocket != nil {
-			admin["websocket"] = adminConfig.Websocket
-		}
-		if adminConfig.Mq != nil {
-			mqConfig := proto.Clone(adminConfig.Mq).(*conf.AdminBackend_MQ)
-			if mqConfig.Password != "" {
-				mqConfig.Password = "******"
-			}
-			admin["mq"] = mqConfig
-		}
 	}
 	// Never mask secrets on the live runtime object. ConfigurationJSON is a
 	// read-only operation; mutating dataConfig here would replace the actual
@@ -115,9 +105,6 @@ func (r *Repo) ConfigurationJSON() (json.RawMessage, error) {
 			safeAdmin.Email.Secret = "******"
 		}
 		maskStorageSecrets(safeAdmin.Storage)
-		if safeAdmin.Mq != nil && safeAdmin.Mq.Password != "" {
-			safeAdmin.Mq.Password = "******"
-		}
 	}
 	dataMap := map[string]any{}
 	if safeData != nil {
@@ -320,9 +307,6 @@ func preserveAdminSecrets(next, current *conf.AdminBackend) {
 		next.Email.Secret = current.Email.Secret
 	}
 	preserveStorageSecrets(next.Storage, current.Storage)
-	if next.Mq != nil && current.Mq != nil && maskedSecret(next.Mq.Password) {
-		next.Mq.Password = current.Mq.Password
-	}
 }
 
 func maskedSecret(value string) bool { return value == "" || value == "******" }

internal/integrationruntime/store.go

@@ -0,0 +1,145 @@
+// Package integrationruntime keeps the active database-backed integration
+// settings and notifies long-lived provider clients when they change.
+package integrationruntime
+
+import (
+	"encoding/json"
+	"strings"
+	"sync"
+)
+
+type Config struct {
+	Kind     string
+	Provider string
+	Enabled  bool
+	Values   json.RawMessage
+}
+
+type listener struct {
+	kind     string
+	provider string
+	callback func(Config)
+}
+
+type Store struct {
+	mu        sync.RWMutex
+	values    map[string]Config
+	listeners map[uint64]listener
+	nextID    uint64
+}
+
+func NewStore() *Store {
+	return &Store{values: make(map[string]Config), listeners: make(map[uint64]listener)}
+}
+
+func configKey(kind, provider string) string {
+	return strings.ToLower(strings.TrimSpace(kind)) + "/" + strings.ToLower(strings.TrimSpace(provider))
+}
+
+func cloneConfig(config Config) Config {
+	config.Values = append(json.RawMessage(nil), config.Values...)
+	return config
+}
+
+func (s *Store) Get(kind, provider string) (Config, bool) {
+	if s == nil {
+		return Config{}, false
+	}
+	s.mu.RLock()
+	config, ok := s.values[configKey(kind, provider)]
+	s.mu.RUnlock()
+	return cloneConfig(config), ok
+}
+
+func (s *Store) Set(config Config) {
+	if s == nil {
+		return
+	}
+	config.Kind = strings.ToLower(strings.TrimSpace(config.Kind))
+	config.Provider = strings.ToLower(strings.TrimSpace(config.Provider))
+	config = cloneConfig(config)
+	key := configKey(config.Kind, config.Provider)
+	s.mu.Lock()
+	s.values[key] = config
+	callbacks := s.matchingListenersLocked(config.Kind, config.Provider)
+	s.mu.Unlock()
+	for _, callback := range callbacks {
+		callback(cloneConfig(config))
+	}
+}
+
+func (s *Store) Delete(kind, provider string) {
+	if s == nil {
+		return
+	}
+	kind = strings.ToLower(strings.TrimSpace(kind))
+	provider = strings.ToLower(strings.TrimSpace(provider))
+	s.mu.Lock()
+	delete(s.values, configKey(kind, provider))
+	callbacks := s.matchingListenersLocked(kind, provider)
+	s.mu.Unlock()
+	config := Config{Kind: kind, Provider: provider}
+	for _, callback := range callbacks {
+		callback(config)
+	}
+}
+
+func (s *Store) Replace(configs []Config) {
+	if s == nil {
+		return
+	}
+	next := make(map[string]Config, len(configs))
+	for _, config := range configs {
+		config.Kind = strings.ToLower(strings.TrimSpace(config.Kind))
+		config.Provider = strings.ToLower(strings.TrimSpace(config.Provider))
+		config = cloneConfig(config)
+		next[configKey(config.Kind, config.Provider)] = config
+	}
+	s.mu.Lock()
+	previous := s.values
+	s.values = next
+	listeners := make([]listener, 0, len(s.listeners))
+	for _, item := range s.listeners {
+		listeners = append(listeners, item)
+	}
+	s.mu.Unlock()
+
+	changed := make(map[string]Config, len(previous)+len(next))
+	for key, config := range previous {
+		changed[key] = Config{Kind: config.Kind, Provider: config.Provider}
+	}
+	for key, config := range next {
+		changed[key] = config
+	}
+	for _, item := range listeners {
+		if config, ok := changed[configKey(item.kind, item.provider)]; ok {
+			item.callback(cloneConfig(config))
+		}
+	}
+}
+
+func (s *Store) Subscribe(kind, provider string, callback func(Config)) func() {
+	if s == nil || callback == nil {
+		return func() {}
+	}
+	s.mu.Lock()
+	s.nextID++
+	id := s.nextID
+	s.listeners[id] = listener{kind: strings.ToLower(strings.TrimSpace(kind)), provider: strings.ToLower(strings.TrimSpace(provider)), callback: callback}
+	s.mu.Unlock()
+	return func() {
+		s.mu.Lock()
+		delete(s.listeners, id)
+		s.mu.Unlock()
+	}
+}
+
+func (s *Store) matchingListenersLocked(kind, provider string) []func(Config) {
+	callbacks := make([]func(Config), 0)
+	for _, item := range s.listeners {
+		if item.kind == kind && item.provider == provider {
+			callbacks = append(callbacks, item.callback)
+		}
+	}
+	return callbacks
+}

internal/integrationruntime/store_test.go

@@ -0,0 +1,30 @@
+package integrationruntime
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestStoreSetDeleteAndSubscribe(t *testing.T) {
+	store := NewStore()
+	updates := make(chan Config, 2)
+	stop := store.Subscribe("mq", "rabbitmq", func(config Config) { updates <- config })
+	defer stop()
+
+	store.Set(Config{Kind: "MQ", Provider: "RabbitMQ", Enabled: true, Values: json.RawMessage(`{"host":"localhost"}`)})
+	loaded, ok := store.Get("mq", "rabbitmq")
+	if !ok || !loaded.Enabled || string(loaded.Values) != `{"host":"localhost"}` {
+		t.Fatalf("loaded config = %#v, ok=%v", loaded, ok)
+	}
+	if update := <-updates; !update.Enabled {
+		t.Fatalf("set update = %#v", update)
+	}
+
+	store.Delete("mq", "rabbitmq")
+	if _, ok = store.Get("mq", "rabbitmq"); ok {
+		t.Fatal("deleted config remained in store")
+	}
+	if update := <-updates; update.Enabled {
+		t.Fatalf("delete update = %#v", update)
+	}
+}