From a2ce3ae218145510a65a3931e54a2807becc281a Mon Sep 17 00:00:00 2001 From: Yvan <8574526@qq,com> Date: Fri, 21 Aug 2026 19:11:14 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go.mod | 1 + go.sum | 2 + internal/biz/integration_config.go | 79 +- internal/biz/integration_config_definition.go | 52 ++ internal/conf/conf.pb.go | 513 +++--------- internal/conf/conf.proto | 27 - internal/data/config_store.go | 23 +- internal/data/config_watch.go | 2 - internal/data/data.go | 26 +- internal/data/initialization_backend.go | 41 +- internal/data/integration_config.go | 83 -- internal/data/integration_runtime.go | 38 + .../data/repository/integration_config.go | 27 +- internal/data/repository/provider.go | 2 + internal/data/websocket_config.go | 96 --- internal/data/websocket_config_test.go | 50 -- internal/initialize/configuration.go | 16 - internal/integrationruntime/store.go | 145 ++++ internal/integrationruntime/store_test.go | 30 + pkg/mq/mq.go | 10 + web/src/pathInfo.json | 1 + .../view/systemTools/integration/config.vue | 762 ++++++++++++++++++ 22 files changed, 1292 insertions(+), 734 deletions(-) create mode 100644 internal/data/integration_runtime.go delete mode 100644 internal/data/websocket_config.go delete mode 100644 internal/data/websocket_config_test.go create mode 100644 internal/integrationruntime/store.go create mode 100644 internal/integrationruntime/store_test.go create mode 100644 web/src/view/systemTools/integration/config.vue diff --git a/go.mod b/go.mod index 46b694f..e8ccad2 100644 --- a/go.mod +++ b/go.mod @@ -143,6 +143,7 @@ require ( github.com/pelletier/go-toml/v2 v2.2.2 // indirect github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c // indirect github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect + github.com/rabbitmq/amqp091-go v1.14.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/richardlehane/mscfb v1.0.4 // indirect github.com/richardlehane/msoleps v1.0.4 // indirect diff --git a/go.sum b/go.sum index 0d1d6e0..5eae956 100644 --- a/go.sum +++ b/go.sum @@ -331,6 +331,8 @@ github.com/qiniu/dyn v1.3.0/go.mod h1:E8oERcm8TtwJiZvkQPbcAh0RL8jO1G0VXJMW3FAWdk github.com/qiniu/go-sdk/v7 v7.25.2 h1:URwgZpxySdiwu2yQpHk93X4LXWHyFRp1x3Vmlk/YWvo= github.com/qiniu/go-sdk/v7 v7.25.2/go.mod h1:dmKtJ2ahhPWFVi9o1D5GemmWoh/ctuB9peqTowyTO8o= github.com/qiniu/x v1.10.5/go.mod h1:03Ni9tj+N2h2aKnAz+6N0Xfl8FwMEDRC2PAlxekASDs= +github.com/rabbitmq/amqp091-go v1.14.0 h1:RSaT7aOKt/OrkVUyswPDW29lnRz9psuGmfZFBmLqLek= +github.com/rabbitmq/amqp091-go v1.14.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E= github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw= github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= diff --git a/internal/biz/integration_config.go b/internal/biz/integration_config.go index c9d1a87..cc237e6 100644 --- a/internal/biz/integration_config.go +++ b/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 { diff --git a/internal/biz/integration_config_definition.go b/internal/biz/integration_config_definition.go index 6420772..0689809 100644 --- a/internal/biz/integration_config_definition.go +++ b/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"), diff --git a/internal/conf/conf.pb.go b/internal/conf/conf.pb.go index a168e3e..8cdf000 100644 --- a/internal/conf/conf.pb.go +++ b/internal/conf/conf.pb.go @@ -2,7 +2,7 @@ // versions: // protoc-gen-go v1.36.11 // protoc v5.28.3 -// source: internal/conf/conf.proto +// source: conf/conf.proto package conf @@ -33,7 +33,7 @@ type Bootstrap struct { func (x *Bootstrap) Reset() { *x = Bootstrap{} - mi := &file_internal_conf_conf_proto_msgTypes[0] + mi := &file_conf_conf_proto_msgTypes[0] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -45,7 +45,7 @@ func (x *Bootstrap) String() string { func (*Bootstrap) ProtoMessage() {} func (x *Bootstrap) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[0] + mi := &file_conf_conf_proto_msgTypes[0] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -58,7 +58,7 @@ func (x *Bootstrap) ProtoReflect() protoreflect.Message { // Deprecated: Use Bootstrap.ProtoReflect.Descriptor instead. func (*Bootstrap) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{0} + return file_conf_conf_proto_rawDescGZIP(), []int{0} } func (x *Bootstrap) GetServer() *Server { @@ -91,7 +91,7 @@ type Server struct { func (x *Server) Reset() { *x = Server{} - mi := &file_internal_conf_conf_proto_msgTypes[1] + mi := &file_conf_conf_proto_msgTypes[1] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -103,7 +103,7 @@ func (x *Server) String() string { func (*Server) ProtoMessage() {} func (x *Server) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[1] + mi := &file_conf_conf_proto_msgTypes[1] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -116,7 +116,7 @@ func (x *Server) ProtoReflect() protoreflect.Message { // Deprecated: Use Server.ProtoReflect.Descriptor instead. func (*Server) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{1} + return file_conf_conf_proto_rawDescGZIP(), []int{1} } func (x *Server) GetHttp() *Server_HTTP { @@ -139,7 +139,7 @@ type Data struct { func (x *Data) Reset() { *x = Data{} - mi := &file_internal_conf_conf_proto_msgTypes[2] + mi := &file_conf_conf_proto_msgTypes[2] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -151,7 +151,7 @@ func (x *Data) String() string { func (*Data) ProtoMessage() {} func (x *Data) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[2] + mi := &file_conf_conf_proto_msgTypes[2] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -164,7 +164,7 @@ func (x *Data) ProtoReflect() protoreflect.Message { // Deprecated: Use Data.ProtoReflect.Descriptor instead. func (*Data) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{2} + return file_conf_conf_proto_rawDescGZIP(), []int{2} } func (x *Data) GetDatabase() *Data_Database { @@ -212,22 +212,20 @@ type AdminBackend struct { Email *AdminBackend_Email `protobuf:"bytes,5,opt,name=email,proto3" json:"email,omitempty"` Storage *AdminBackend_Storage `protobuf:"bytes,6,opt,name=storage,proto3" json:"storage,omitempty"` // ConfigPath is populated by the entrypoint and is not required in YAML. - ConfigPath string `protobuf:"bytes,7,opt,name=config_path,json=configPath,proto3" json:"config_path,omitempty"` - Media *AdminBackend_Media `protobuf:"bytes,8,opt,name=media,proto3" json:"media,omitempty"` - DiskList []*AdminBackend_Disk `protobuf:"bytes,9,rep,name=disk_list,json=diskList,proto3" json:"disk_list,omitempty"` - System *AdminBackend_System `protobuf:"bytes,10,opt,name=system,proto3" json:"system,omitempty"` - Zap *AdminBackend_Zap `protobuf:"bytes,11,opt,name=zap,proto3" json:"zap,omitempty"` - Cors *AdminBackend_CORS `protobuf:"bytes,12,opt,name=cors,proto3" json:"cors,omitempty"` - App *AdminBackend_App `protobuf:"bytes,13,opt,name=app,proto3" json:"app,omitempty"` - Websocket *AdminBackend_WebSocket `protobuf:"bytes,14,opt,name=websocket,proto3" json:"websocket,omitempty"` - Mq *AdminBackend_MQ `protobuf:"bytes,15,opt,name=mq,proto3" json:"mq,omitempty"` + ConfigPath string `protobuf:"bytes,7,opt,name=config_path,json=configPath,proto3" json:"config_path,omitempty"` + Media *AdminBackend_Media `protobuf:"bytes,8,opt,name=media,proto3" json:"media,omitempty"` + DiskList []*AdminBackend_Disk `protobuf:"bytes,9,rep,name=disk_list,json=diskList,proto3" json:"disk_list,omitempty"` + System *AdminBackend_System `protobuf:"bytes,10,opt,name=system,proto3" json:"system,omitempty"` + Zap *AdminBackend_Zap `protobuf:"bytes,11,opt,name=zap,proto3" json:"zap,omitempty"` + Cors *AdminBackend_CORS `protobuf:"bytes,12,opt,name=cors,proto3" json:"cors,omitempty"` + App *AdminBackend_App `protobuf:"bytes,13,opt,name=app,proto3" json:"app,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } func (x *AdminBackend) Reset() { *x = AdminBackend{} - mi := &file_internal_conf_conf_proto_msgTypes[3] + mi := &file_conf_conf_proto_msgTypes[3] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -239,7 +237,7 @@ func (x *AdminBackend) String() string { func (*AdminBackend) ProtoMessage() {} func (x *AdminBackend) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[3] + mi := &file_conf_conf_proto_msgTypes[3] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -252,7 +250,7 @@ func (x *AdminBackend) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend.ProtoReflect.Descriptor instead. func (*AdminBackend) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3} + return file_conf_conf_proto_rawDescGZIP(), []int{3} } func (x *AdminBackend) GetRouterPrefix() string { @@ -346,20 +344,6 @@ func (x *AdminBackend) GetApp() *AdminBackend_App { return nil } -func (x *AdminBackend) GetWebsocket() *AdminBackend_WebSocket { - if x != nil { - return x.Websocket - } - return nil -} - -func (x *AdminBackend) GetMq() *AdminBackend_MQ { - if x != nil { - return x.Mq - } - return nil -} - type Server_HTTP struct { state protoimpl.MessageState `protogen:"open.v1"` Network string `protobuf:"bytes,1,opt,name=network,proto3" json:"network,omitempty"` @@ -371,7 +355,7 @@ type Server_HTTP struct { func (x *Server_HTTP) Reset() { *x = Server_HTTP{} - mi := &file_internal_conf_conf_proto_msgTypes[4] + mi := &file_conf_conf_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -383,7 +367,7 @@ func (x *Server_HTTP) String() string { func (*Server_HTTP) ProtoMessage() {} func (x *Server_HTTP) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[4] + mi := &file_conf_conf_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -396,7 +380,7 @@ func (x *Server_HTTP) ProtoReflect() protoreflect.Message { // Deprecated: Use Server_HTTP.ProtoReflect.Descriptor instead. func (*Server_HTTP) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{1, 0} + return file_conf_conf_proto_rawDescGZIP(), []int{1, 0} } func (x *Server_HTTP) GetNetwork() string { @@ -446,7 +430,7 @@ type Data_Database struct { func (x *Data_Database) Reset() { *x = Data_Database{} - mi := &file_internal_conf_conf_proto_msgTypes[5] + mi := &file_conf_conf_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -458,7 +442,7 @@ func (x *Data_Database) String() string { func (*Data_Database) ProtoMessage() {} func (x *Data_Database) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[5] + mi := &file_conf_conf_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -471,7 +455,7 @@ func (x *Data_Database) ProtoReflect() protoreflect.Message { // Deprecated: Use Data_Database.ProtoReflect.Descriptor instead. func (*Data_Database) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{2, 0} + return file_conf_conf_proto_rawDescGZIP(), []int{2, 0} } func (x *Data_Database) GetDriver() string { @@ -617,7 +601,7 @@ type Data_Redis struct { func (x *Data_Redis) Reset() { *x = Data_Redis{} - mi := &file_internal_conf_conf_proto_msgTypes[6] + mi := &file_conf_conf_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -629,7 +613,7 @@ func (x *Data_Redis) String() string { func (*Data_Redis) ProtoMessage() {} func (x *Data_Redis) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[6] + mi := &file_conf_conf_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -642,7 +626,7 @@ func (x *Data_Redis) ProtoReflect() protoreflect.Message { // Deprecated: Use Data_Redis.ProtoReflect.Descriptor instead. func (*Data_Redis) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{2, 1} + return file_conf_conf_proto_rawDescGZIP(), []int{2, 1} } func (x *Data_Redis) GetNetwork() string { @@ -718,7 +702,7 @@ type Data_MongoHost struct { func (x *Data_MongoHost) Reset() { *x = Data_MongoHost{} - mi := &file_internal_conf_conf_proto_msgTypes[7] + mi := &file_conf_conf_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -730,7 +714,7 @@ func (x *Data_MongoHost) String() string { func (*Data_MongoHost) ProtoMessage() {} func (x *Data_MongoHost) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[7] + mi := &file_conf_conf_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -743,7 +727,7 @@ func (x *Data_MongoHost) ProtoReflect() protoreflect.Message { // Deprecated: Use Data_MongoHost.ProtoReflect.Descriptor instead. func (*Data_MongoHost) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{2, 2} + return file_conf_conf_proto_rawDescGZIP(), []int{2, 2} } func (x *Data_MongoHost) GetHost() string { @@ -780,7 +764,7 @@ type Data_Mongo struct { func (x *Data_Mongo) Reset() { *x = Data_Mongo{} - mi := &file_internal_conf_conf_proto_msgTypes[8] + mi := &file_conf_conf_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -792,7 +776,7 @@ func (x *Data_Mongo) String() string { func (*Data_Mongo) ProtoMessage() {} func (x *Data_Mongo) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[8] + mi := &file_conf_conf_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -805,7 +789,7 @@ func (x *Data_Mongo) ProtoReflect() protoreflect.Message { // Deprecated: Use Data_Mongo.ProtoReflect.Descriptor instead. func (*Data_Mongo) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{2, 3} + return file_conf_conf_proto_rawDescGZIP(), []int{2, 3} } func (x *Data_Mongo) GetColl() string { @@ -904,7 +888,7 @@ type AdminBackend_JWT struct { func (x *AdminBackend_JWT) Reset() { *x = AdminBackend_JWT{} - mi := &file_internal_conf_conf_proto_msgTypes[9] + mi := &file_conf_conf_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -916,7 +900,7 @@ func (x *AdminBackend_JWT) String() string { func (*AdminBackend_JWT) ProtoMessage() {} func (x *AdminBackend_JWT) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[9] + mi := &file_conf_conf_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -929,7 +913,7 @@ func (x *AdminBackend_JWT) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_JWT.ProtoReflect.Descriptor instead. func (*AdminBackend_JWT) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 0} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 0} } func (x *AdminBackend_JWT) GetSigningKey() string { @@ -972,7 +956,7 @@ type AdminBackend_Captcha struct { func (x *AdminBackend_Captcha) Reset() { *x = AdminBackend_Captcha{} - mi := &file_internal_conf_conf_proto_msgTypes[10] + mi := &file_conf_conf_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -984,7 +968,7 @@ func (x *AdminBackend_Captcha) String() string { func (*AdminBackend_Captcha) ProtoMessage() {} func (x *AdminBackend_Captcha) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[10] + mi := &file_conf_conf_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -997,7 +981,7 @@ func (x *AdminBackend_Captcha) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_Captcha.ProtoReflect.Descriptor instead. func (*AdminBackend_Captcha) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 1} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 1} } func (x *AdminBackend_Captcha) GetKeyLong() int32 { @@ -1038,7 +1022,7 @@ type AdminBackend_Local struct { func (x *AdminBackend_Local) Reset() { *x = AdminBackend_Local{} - mi := &file_internal_conf_conf_proto_msgTypes[11] + mi := &file_conf_conf_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1050,7 +1034,7 @@ func (x *AdminBackend_Local) String() string { func (*AdminBackend_Local) ProtoMessage() {} func (x *AdminBackend_Local) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[11] + mi := &file_conf_conf_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1063,7 +1047,7 @@ func (x *AdminBackend_Local) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_Local.ProtoReflect.Descriptor instead. func (*AdminBackend_Local) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 2} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 2} } func (x *AdminBackend_Local) GetStorePath() string { @@ -1096,7 +1080,7 @@ type AdminBackend_Email struct { func (x *AdminBackend_Email) Reset() { *x = AdminBackend_Email{} - mi := &file_internal_conf_conf_proto_msgTypes[12] + mi := &file_conf_conf_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1108,7 +1092,7 @@ func (x *AdminBackend_Email) String() string { func (*AdminBackend_Email) ProtoMessage() {} func (x *AdminBackend_Email) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[12] + mi := &file_conf_conf_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1121,7 +1105,7 @@ func (x *AdminBackend_Email) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_Email.ProtoReflect.Descriptor instead. func (*AdminBackend_Email) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 3} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 3} } func (x *AdminBackend_Email) GetTo() string { @@ -1191,7 +1175,7 @@ type AdminBackend_Media struct { func (x *AdminBackend_Media) Reset() { *x = AdminBackend_Media{} - mi := &file_internal_conf_conf_proto_msgTypes[13] + mi := &file_conf_conf_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1203,7 +1187,7 @@ func (x *AdminBackend_Media) String() string { func (*AdminBackend_Media) ProtoMessage() {} func (x *AdminBackend_Media) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[13] + mi := &file_conf_conf_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1216,7 +1200,7 @@ func (x *AdminBackend_Media) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_Media.ProtoReflect.Descriptor instead. func (*AdminBackend_Media) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 4} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 4} } func (x *AdminBackend_Media) GetSessionTtl() int32 { @@ -1249,7 +1233,7 @@ type AdminBackend_Disk struct { func (x *AdminBackend_Disk) Reset() { *x = AdminBackend_Disk{} - mi := &file_internal_conf_conf_proto_msgTypes[14] + mi := &file_conf_conf_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1261,7 +1245,7 @@ func (x *AdminBackend_Disk) String() string { func (*AdminBackend_Disk) ProtoMessage() {} func (x *AdminBackend_Disk) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[14] + mi := &file_conf_conf_proto_msgTypes[14] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1274,7 +1258,7 @@ func (x *AdminBackend_Disk) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_Disk.ProtoReflect.Descriptor instead. func (*AdminBackend_Disk) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 5} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 5} } func (x *AdminBackend_Disk) GetMountPoint() string { @@ -1300,7 +1284,7 @@ type AdminBackend_System struct { func (x *AdminBackend_System) Reset() { *x = AdminBackend_System{} - mi := &file_internal_conf_conf_proto_msgTypes[15] + mi := &file_conf_conf_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1312,7 +1296,7 @@ func (x *AdminBackend_System) String() string { func (*AdminBackend_System) ProtoMessage() {} func (x *AdminBackend_System) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[15] + mi := &file_conf_conf_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1325,7 +1309,7 @@ func (x *AdminBackend_System) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_System.ProtoReflect.Descriptor instead. func (*AdminBackend_System) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 6} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 6} } func (x *AdminBackend_System) GetUseRedis() bool { @@ -1406,7 +1390,7 @@ type AdminBackend_Zap struct { func (x *AdminBackend_Zap) Reset() { *x = AdminBackend_Zap{} - mi := &file_internal_conf_conf_proto_msgTypes[16] + mi := &file_conf_conf_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1418,7 +1402,7 @@ func (x *AdminBackend_Zap) String() string { func (*AdminBackend_Zap) ProtoMessage() {} func (x *AdminBackend_Zap) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[16] + mi := &file_conf_conf_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1431,7 +1415,7 @@ func (x *AdminBackend_Zap) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_Zap.ProtoReflect.Descriptor instead. func (*AdminBackend_Zap) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 7} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 7} } func (x *AdminBackend_Zap) GetLevel() string { @@ -1542,7 +1526,7 @@ type AdminBackend_CORS struct { func (x *AdminBackend_CORS) Reset() { *x = AdminBackend_CORS{} - mi := &file_internal_conf_conf_proto_msgTypes[17] + mi := &file_conf_conf_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1554,7 +1538,7 @@ func (x *AdminBackend_CORS) String() string { func (*AdminBackend_CORS) ProtoMessage() {} func (x *AdminBackend_CORS) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[17] + mi := &file_conf_conf_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1567,7 +1551,7 @@ func (x *AdminBackend_CORS) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_CORS.ProtoReflect.Descriptor instead. func (*AdminBackend_CORS) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 8} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 8} } func (x *AdminBackend_CORS) GetMode() string { @@ -1597,7 +1581,7 @@ type AdminBackend_CORSRule struct { func (x *AdminBackend_CORSRule) Reset() { *x = AdminBackend_CORSRule{} - mi := &file_internal_conf_conf_proto_msgTypes[18] + mi := &file_conf_conf_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1609,7 +1593,7 @@ func (x *AdminBackend_CORSRule) String() string { func (*AdminBackend_CORSRule) ProtoMessage() {} func (x *AdminBackend_CORSRule) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[18] + mi := &file_conf_conf_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1622,7 +1606,7 @@ func (x *AdminBackend_CORSRule) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_CORSRule.ProtoReflect.Descriptor instead. func (*AdminBackend_CORSRule) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 9} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 9} } func (x *AdminBackend_CORSRule) GetAllowOrigin() string { @@ -1671,7 +1655,7 @@ type AdminBackend_App struct { func (x *AdminBackend_App) Reset() { *x = AdminBackend_App{} - mi := &file_internal_conf_conf_proto_msgTypes[19] + mi := &file_conf_conf_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1683,7 +1667,7 @@ func (x *AdminBackend_App) String() string { func (*AdminBackend_App) ProtoMessage() {} func (x *AdminBackend_App) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[19] + mi := &file_conf_conf_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1696,7 +1680,7 @@ func (x *AdminBackend_App) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_App.ProtoReflect.Descriptor instead. func (*AdminBackend_App) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 10} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 10} } func (x *AdminBackend_App) GetNode() string { @@ -1720,216 +1704,6 @@ func (x *AdminBackend_App) GetEnv() string { return "" } -type AdminBackend_WebSocket struct { - state protoimpl.MessageState `protogen:"open.v1"` - Enabled bool `protobuf:"varint,1,opt,name=enabled,proto3" json:"enabled,omitempty"` - Path string `protobuf:"bytes,2,opt,name=path,proto3" json:"path,omitempty"` - AllowOrigins []string `protobuf:"bytes,3,rep,name=allow_origins,json=allowOrigins,proto3" json:"allow_origins,omitempty"` - MaxMessageSize int64 `protobuf:"varint,4,opt,name=max_message_size,json=maxMessageSize,proto3" json:"max_message_size,omitempty"` - WriteWait *durationpb.Duration `protobuf:"bytes,5,opt,name=write_wait,json=writeWait,proto3" json:"write_wait,omitempty"` - PongWait *durationpb.Duration `protobuf:"bytes,6,opt,name=pong_wait,json=pongWait,proto3" json:"pong_wait,omitempty"` - PingPeriod *durationpb.Duration `protobuf:"bytes,7,opt,name=ping_period,json=pingPeriod,proto3" json:"ping_period,omitempty"` - MessageBufferSize int32 `protobuf:"varint,8,opt,name=message_buffer_size,json=messageBufferSize,proto3" json:"message_buffer_size,omitempty"` - ConcurrentMessageHandling bool `protobuf:"varint,9,opt,name=concurrent_message_handling,json=concurrentMessageHandling,proto3" json:"concurrent_message_handling,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache -} - -func (x *AdminBackend_WebSocket) Reset() { - *x = AdminBackend_WebSocket{} - mi := &file_internal_conf_conf_proto_msgTypes[20] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) -} - -func (x *AdminBackend_WebSocket) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*AdminBackend_WebSocket) ProtoMessage() {} - -func (x *AdminBackend_WebSocket) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[20] - if x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -// Deprecated: Use AdminBackend_WebSocket.ProtoReflect.Descriptor instead. -func (*AdminBackend_WebSocket) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 11} -} - -func (x *AdminBackend_WebSocket) GetEnabled() bool { - if x != nil { - return x.Enabled - } - return false -} - -func (x *AdminBackend_WebSocket) GetPath() string { - if x != nil { - return x.Path - } - return "" -} - -func (x *AdminBackend_WebSocket) GetAllowOrigins() []string { - if x != nil { - return x.AllowOrigins - } - return nil -} - -func (x *AdminBackend_WebSocket) GetMaxMessageSize() int64 { - if x != nil { - return x.MaxMessageSize - } - return 0 -} - -func (x *AdminBackend_WebSocket) GetWriteWait() *durationpb.Duration { - if x != nil { - return x.WriteWait - } - return nil -} - -func (x *AdminBackend_WebSocket) GetPongWait() *durationpb.Duration { - if x != nil { - return x.PongWait - } - return nil -} - -func (x *AdminBackend_WebSocket) GetPingPeriod() *durationpb.Duration { - if x != nil { - return x.PingPeriod - } - return nil -} - -func (x *AdminBackend_WebSocket) GetMessageBufferSize() int32 { - if x != nil { - return x.MessageBufferSize - } - return 0 -} - -func (x *AdminBackend_WebSocket) GetConcurrentMessageHandling() bool { - if x != nil { - return x.ConcurrentMessageHandling - } - return false -} - -// MQ config is persisted in sys_integration_configs (kind=mq/provider=emqx). -// The bootstrap fields are retained as a one-time migration source. -type AdminBackend_MQ struct { - state protoimpl.MessageState `protogen:"open.v1"` - Enabled bool `protobuf:"varint,1,opt,name=enabled,proto3" json:"enabled,omitempty"` - Broker string `protobuf:"bytes,2,opt,name=broker,proto3" json:"broker,omitempty"` - ClientId string `protobuf:"bytes,3,opt,name=client_id,json=clientId,proto3" json:"client_id,omitempty"` - Username string `protobuf:"bytes,4,opt,name=username,proto3" json:"username,omitempty"` - Password string `protobuf:"bytes,5,opt,name=password,proto3" json:"password,omitempty"` - KeepAlive int32 `protobuf:"varint,6,opt,name=keep_alive,json=keepAlive,proto3" json:"keep_alive,omitempty"` - CleanSession bool `protobuf:"varint,7,opt,name=clean_session,json=cleanSession,proto3" json:"clean_session,omitempty"` - ConnectTimeout int32 `protobuf:"varint,8,opt,name=connect_timeout,json=connectTimeout,proto3" json:"connect_timeout,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache -} - -func (x *AdminBackend_MQ) Reset() { - *x = AdminBackend_MQ{} - mi := &file_internal_conf_conf_proto_msgTypes[21] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) -} - -func (x *AdminBackend_MQ) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*AdminBackend_MQ) ProtoMessage() {} - -func (x *AdminBackend_MQ) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[21] - if x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -// Deprecated: Use AdminBackend_MQ.ProtoReflect.Descriptor instead. -func (*AdminBackend_MQ) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 12} -} - -func (x *AdminBackend_MQ) GetEnabled() bool { - if x != nil { - return x.Enabled - } - return false -} - -func (x *AdminBackend_MQ) GetBroker() string { - if x != nil { - return x.Broker - } - return "" -} - -func (x *AdminBackend_MQ) GetClientId() string { - if x != nil { - return x.ClientId - } - return "" -} - -func (x *AdminBackend_MQ) GetUsername() string { - if x != nil { - return x.Username - } - return "" -} - -func (x *AdminBackend_MQ) GetPassword() string { - if x != nil { - return x.Password - } - return "" -} - -func (x *AdminBackend_MQ) GetKeepAlive() int32 { - if x != nil { - return x.KeepAlive - } - return 0 -} - -func (x *AdminBackend_MQ) GetCleanSession() bool { - if x != nil { - return x.CleanSession - } - return false -} - -func (x *AdminBackend_MQ) GetConnectTimeout() int32 { - if x != nil { - return x.ConnectTimeout - } - return 0 -} - type AdminBackend_Storage struct { state protoimpl.MessageState `protogen:"open.v1"` Type string `protobuf:"bytes,1,opt,name=type,proto3" json:"type,omitempty"` @@ -1946,7 +1720,7 @@ type AdminBackend_Storage struct { func (x *AdminBackend_Storage) Reset() { *x = AdminBackend_Storage{} - mi := &file_internal_conf_conf_proto_msgTypes[22] + mi := &file_conf_conf_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1958,7 +1732,7 @@ func (x *AdminBackend_Storage) String() string { func (*AdminBackend_Storage) ProtoMessage() {} func (x *AdminBackend_Storage) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[22] + mi := &file_conf_conf_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1971,7 +1745,7 @@ func (x *AdminBackend_Storage) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_Storage.ProtoReflect.Descriptor instead. func (*AdminBackend_Storage) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 13} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 11} } func (x *AdminBackend_Storage) GetType() string { @@ -2045,7 +1819,7 @@ type AdminBackend_Qiniu struct { func (x *AdminBackend_Qiniu) Reset() { *x = AdminBackend_Qiniu{} - mi := &file_internal_conf_conf_proto_msgTypes[23] + mi := &file_conf_conf_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2057,7 +1831,7 @@ func (x *AdminBackend_Qiniu) String() string { func (*AdminBackend_Qiniu) ProtoMessage() {} func (x *AdminBackend_Qiniu) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[23] + mi := &file_conf_conf_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2070,7 +1844,7 @@ func (x *AdminBackend_Qiniu) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_Qiniu.ProtoReflect.Descriptor instead. func (*AdminBackend_Qiniu) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 14} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 12} } func (x *AdminBackend_Qiniu) GetZone() string { @@ -2142,7 +1916,7 @@ type AdminBackend_ObjectStore struct { func (x *AdminBackend_ObjectStore) Reset() { *x = AdminBackend_ObjectStore{} - mi := &file_internal_conf_conf_proto_msgTypes[24] + mi := &file_conf_conf_proto_msgTypes[22] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2154,7 +1928,7 @@ func (x *AdminBackend_ObjectStore) String() string { func (*AdminBackend_ObjectStore) ProtoMessage() {} func (x *AdminBackend_ObjectStore) ProtoReflect() protoreflect.Message { - mi := &file_internal_conf_conf_proto_msgTypes[24] + mi := &file_conf_conf_proto_msgTypes[22] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2167,7 +1941,7 @@ func (x *AdminBackend_ObjectStore) ProtoReflect() protoreflect.Message { // Deprecated: Use AdminBackend_ObjectStore.ProtoReflect.Descriptor instead. func (*AdminBackend_ObjectStore) Descriptor() ([]byte, []int) { - return file_internal_conf_conf_proto_rawDescGZIP(), []int{3, 15} + return file_conf_conf_proto_rawDescGZIP(), []int{3, 13} } func (x *AdminBackend_ObjectStore) GetEndpoint() string { @@ -2240,11 +2014,11 @@ func (x *AdminBackend_ObjectStore) GetAccountId() string { return "" } -var File_internal_conf_conf_proto protoreflect.FileDescriptor +var File_conf_conf_proto protoreflect.FileDescriptor -const file_internal_conf_conf_proto_rawDesc = "" + +const file_conf_conf_proto_rawDesc = "" + "\n" + - "\x18internal/conf/conf.proto\x12\n" + + "\x0fconf/conf.proto\x12\n" + "kratos.api\x1a\x1egoogle/protobuf/duration.proto\"\x8d\x01\n" + "\tBootstrap\x12*\n" + "\x06server\x18\x01 \x01(\v2\x12.kratos.api.ServerR\x06server\x12$\n" + @@ -2312,7 +2086,7 @@ const file_internal_conf_conf_proto_rawDesc = "" + "\x12connect_timeout_ms\x18\n" + " \x01(\x03R\x10connectTimeoutMs\x12\x15\n" + "\x06is_zap\x18\v \x01(\bR\x05isZap\x120\n" + - "\x05hosts\x18\f \x03(\v2\x1a.kratos.api.Data.MongoHostR\x05hosts\"\xc0\"\n" + + "\x05hosts\x18\f \x03(\v2\x1a.kratos.api.Data.MongoHostR\x05hosts\"\xad\x1c\n" + "\fAdminBackend\x12#\n" + "\rrouter_prefix\x18\x01 \x01(\tR\frouterPrefix\x12.\n" + "\x03jwt\x18\x02 \x01(\v2\x1c.kratos.api.AdminBackend.JWTR\x03jwt\x12:\n" + @@ -2328,9 +2102,7 @@ const file_internal_conf_conf_proto_rawDesc = "" + " \x01(\v2\x1f.kratos.api.AdminBackend.SystemR\x06system\x12.\n" + "\x03zap\x18\v \x01(\v2\x1c.kratos.api.AdminBackend.ZapR\x03zap\x121\n" + "\x04cors\x18\f \x01(\v2\x1d.kratos.api.AdminBackend.CORSR\x04cors\x12.\n" + - "\x03app\x18\r \x01(\v2\x1c.kratos.api.AdminBackend.AppR\x03app\x12@\n" + - "\twebsocket\x18\x0e \x01(\v2\".kratos.api.AdminBackend.WebSocketR\twebsocket\x12+\n" + - "\x02mq\x18\x0f \x01(\v2\x1b.kratos.api.AdminBackend.MQR\x02mq\x1a\xb8\x01\n" + + "\x03app\x18\r \x01(\v2\x1c.kratos.api.AdminBackend.AppR\x03app\x1a\xb8\x01\n" + "\x03JWT\x12\x1f\n" + "\vsigning_key\x18\x01 \x01(\tR\n" + "signingKey\x12<\n" + @@ -2403,29 +2175,7 @@ const file_internal_conf_conf_proto_rawDesc = "" + "\x03App\x12\x12\n" + "\x04node\x18\x01 \x01(\tR\x04node\x12\x15\n" + "\x06app_id\x18\x02 \x01(\tR\x05appId\x12\x10\n" + - "\x03env\x18\x03 \x01(\tR\x03env\x1a\xa6\x03\n" + - "\tWebSocket\x12\x18\n" + - "\aenabled\x18\x01 \x01(\bR\aenabled\x12\x12\n" + - "\x04path\x18\x02 \x01(\tR\x04path\x12#\n" + - "\rallow_origins\x18\x03 \x03(\tR\fallowOrigins\x12(\n" + - "\x10max_message_size\x18\x04 \x01(\x03R\x0emaxMessageSize\x128\n" + - "\n" + - "write_wait\x18\x05 \x01(\v2\x19.google.protobuf.DurationR\twriteWait\x126\n" + - "\tpong_wait\x18\x06 \x01(\v2\x19.google.protobuf.DurationR\bpongWait\x12:\n" + - "\vping_period\x18\a \x01(\v2\x19.google.protobuf.DurationR\n" + - "pingPeriod\x12.\n" + - "\x13message_buffer_size\x18\b \x01(\x05R\x11messageBufferSize\x12>\n" + - "\x1bconcurrent_message_handling\x18\t \x01(\bR\x19concurrentMessageHandling\x1a\xf8\x01\n" + - "\x02MQ\x12\x18\n" + - "\aenabled\x18\x01 \x01(\bR\aenabled\x12\x16\n" + - "\x06broker\x18\x02 \x01(\tR\x06broker\x12\x1b\n" + - "\tclient_id\x18\x03 \x01(\tR\bclientId\x12\x1a\n" + - "\busername\x18\x04 \x01(\tR\busername\x12\x1a\n" + - "\bpassword\x18\x05 \x01(\tR\bpassword\x12\x1d\n" + - "\n" + - "keep_alive\x18\x06 \x01(\x05R\tkeepAlive\x12#\n" + - "\rclean_session\x18\a \x01(\bR\fcleanSession\x12'\n" + - "\x0fconnect_timeout\x18\b \x01(\x05R\x0econnectTimeout\x1a\xe8\x03\n" + + "\x03env\x18\x03 \x01(\tR\x03env\x1a\xe8\x03\n" + "\aStorage\x12\x12\n" + "\x04type\x18\x01 \x01(\tR\x04type\x124\n" + "\x05qiniu\x18\x02 \x01(\v2\x1e.kratos.api.AdminBackend.QiniuR\x05qiniu\x12C\n" + @@ -2466,19 +2216,19 @@ const file_internal_conf_conf_proto_rawDesc = "" + " \x01(\tR\taccountIdB\x18Z\x16kra/internal/conf;confb\x06proto3" var ( - file_internal_conf_conf_proto_rawDescOnce sync.Once - file_internal_conf_conf_proto_rawDescData []byte + file_conf_conf_proto_rawDescOnce sync.Once + file_conf_conf_proto_rawDescData []byte ) -func file_internal_conf_conf_proto_rawDescGZIP() []byte { - file_internal_conf_conf_proto_rawDescOnce.Do(func() { - file_internal_conf_conf_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_internal_conf_conf_proto_rawDesc), len(file_internal_conf_conf_proto_rawDesc))) +func file_conf_conf_proto_rawDescGZIP() []byte { + file_conf_conf_proto_rawDescOnce.Do(func() { + file_conf_conf_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_conf_conf_proto_rawDesc), len(file_conf_conf_proto_rawDesc))) }) - return file_internal_conf_conf_proto_rawDescData + return file_conf_conf_proto_rawDescData } -var file_internal_conf_conf_proto_msgTypes = make([]protoimpl.MessageInfo, 25) -var file_internal_conf_conf_proto_goTypes = []any{ +var file_conf_conf_proto_msgTypes = make([]protoimpl.MessageInfo, 23) +var file_conf_conf_proto_goTypes = []any{ (*Bootstrap)(nil), // 0: kratos.api.Bootstrap (*Server)(nil), // 1: kratos.api.Server (*Data)(nil), // 2: kratos.api.Data @@ -2499,14 +2249,12 @@ var file_internal_conf_conf_proto_goTypes = []any{ (*AdminBackend_CORS)(nil), // 17: kratos.api.AdminBackend.CORS (*AdminBackend_CORSRule)(nil), // 18: kratos.api.AdminBackend.CORSRule (*AdminBackend_App)(nil), // 19: kratos.api.AdminBackend.App - (*AdminBackend_WebSocket)(nil), // 20: kratos.api.AdminBackend.WebSocket - (*AdminBackend_MQ)(nil), // 21: kratos.api.AdminBackend.MQ - (*AdminBackend_Storage)(nil), // 22: kratos.api.AdminBackend.Storage - (*AdminBackend_Qiniu)(nil), // 23: kratos.api.AdminBackend.Qiniu - (*AdminBackend_ObjectStore)(nil), // 24: kratos.api.AdminBackend.ObjectStore - (*durationpb.Duration)(nil), // 25: google.protobuf.Duration + (*AdminBackend_Storage)(nil), // 20: kratos.api.AdminBackend.Storage + (*AdminBackend_Qiniu)(nil), // 21: kratos.api.AdminBackend.Qiniu + (*AdminBackend_ObjectStore)(nil), // 22: kratos.api.AdminBackend.ObjectStore + (*durationpb.Duration)(nil), // 23: google.protobuf.Duration } -var file_internal_conf_conf_proto_depIdxs = []int32{ +var file_conf_conf_proto_depIdxs = []int32{ 1, // 0: kratos.api.Bootstrap.server:type_name -> kratos.api.Server 2, // 1: kratos.api.Bootstrap.data:type_name -> kratos.api.Data 3, // 2: kratos.api.Bootstrap.admin:type_name -> kratos.api.AdminBackend @@ -2520,60 +2268,55 @@ var file_internal_conf_conf_proto_depIdxs = []int32{ 10, // 10: kratos.api.AdminBackend.captcha:type_name -> kratos.api.AdminBackend.Captcha 11, // 11: kratos.api.AdminBackend.local:type_name -> kratos.api.AdminBackend.Local 12, // 12: kratos.api.AdminBackend.email:type_name -> kratos.api.AdminBackend.Email - 22, // 13: kratos.api.AdminBackend.storage:type_name -> kratos.api.AdminBackend.Storage + 20, // 13: kratos.api.AdminBackend.storage:type_name -> kratos.api.AdminBackend.Storage 13, // 14: kratos.api.AdminBackend.media:type_name -> kratos.api.AdminBackend.Media 14, // 15: kratos.api.AdminBackend.disk_list:type_name -> kratos.api.AdminBackend.Disk 15, // 16: kratos.api.AdminBackend.system:type_name -> kratos.api.AdminBackend.System 16, // 17: kratos.api.AdminBackend.zap:type_name -> kratos.api.AdminBackend.Zap 17, // 18: kratos.api.AdminBackend.cors:type_name -> kratos.api.AdminBackend.CORS 19, // 19: kratos.api.AdminBackend.app:type_name -> kratos.api.AdminBackend.App - 20, // 20: kratos.api.AdminBackend.websocket:type_name -> kratos.api.AdminBackend.WebSocket - 21, // 21: kratos.api.AdminBackend.mq:type_name -> kratos.api.AdminBackend.MQ - 25, // 22: kratos.api.Server.HTTP.timeout:type_name -> google.protobuf.Duration - 25, // 23: kratos.api.Data.Redis.read_timeout:type_name -> google.protobuf.Duration - 25, // 24: kratos.api.Data.Redis.write_timeout:type_name -> google.protobuf.Duration - 7, // 25: kratos.api.Data.Mongo.hosts:type_name -> kratos.api.Data.MongoHost - 25, // 26: kratos.api.AdminBackend.JWT.expires_time:type_name -> google.protobuf.Duration - 25, // 27: kratos.api.AdminBackend.JWT.buffer_time:type_name -> google.protobuf.Duration - 25, // 28: kratos.api.AdminBackend.Captcha.store_expiration:type_name -> google.protobuf.Duration - 18, // 29: kratos.api.AdminBackend.CORS.whitelist:type_name -> kratos.api.AdminBackend.CORSRule - 25, // 30: kratos.api.AdminBackend.WebSocket.write_wait:type_name -> google.protobuf.Duration - 25, // 31: kratos.api.AdminBackend.WebSocket.pong_wait:type_name -> google.protobuf.Duration - 25, // 32: kratos.api.AdminBackend.WebSocket.ping_period:type_name -> google.protobuf.Duration - 23, // 33: kratos.api.AdminBackend.Storage.qiniu:type_name -> kratos.api.AdminBackend.Qiniu - 24, // 34: kratos.api.AdminBackend.Storage.aliyun_oss:type_name -> kratos.api.AdminBackend.ObjectStore - 24, // 35: kratos.api.AdminBackend.Storage.huawei_obs:type_name -> kratos.api.AdminBackend.ObjectStore - 24, // 36: kratos.api.AdminBackend.Storage.tencent_cos:type_name -> kratos.api.AdminBackend.ObjectStore - 24, // 37: kratos.api.AdminBackend.Storage.aws_s3:type_name -> kratos.api.AdminBackend.ObjectStore - 24, // 38: kratos.api.AdminBackend.Storage.cloudflare_r2:type_name -> kratos.api.AdminBackend.ObjectStore - 24, // 39: kratos.api.AdminBackend.Storage.minio:type_name -> kratos.api.AdminBackend.ObjectStore - 40, // [40:40] is the sub-list for method output_type - 40, // [40:40] is the sub-list for method input_type - 40, // [40:40] is the sub-list for extension type_name - 40, // [40:40] is the sub-list for extension extendee - 0, // [0:40] is the sub-list for field type_name + 23, // 20: kratos.api.Server.HTTP.timeout:type_name -> google.protobuf.Duration + 23, // 21: kratos.api.Data.Redis.read_timeout:type_name -> google.protobuf.Duration + 23, // 22: kratos.api.Data.Redis.write_timeout:type_name -> google.protobuf.Duration + 7, // 23: kratos.api.Data.Mongo.hosts:type_name -> kratos.api.Data.MongoHost + 23, // 24: kratos.api.AdminBackend.JWT.expires_time:type_name -> google.protobuf.Duration + 23, // 25: kratos.api.AdminBackend.JWT.buffer_time:type_name -> google.protobuf.Duration + 23, // 26: kratos.api.AdminBackend.Captcha.store_expiration:type_name -> google.protobuf.Duration + 18, // 27: kratos.api.AdminBackend.CORS.whitelist:type_name -> kratos.api.AdminBackend.CORSRule + 21, // 28: kratos.api.AdminBackend.Storage.qiniu:type_name -> kratos.api.AdminBackend.Qiniu + 22, // 29: kratos.api.AdminBackend.Storage.aliyun_oss:type_name -> kratos.api.AdminBackend.ObjectStore + 22, // 30: kratos.api.AdminBackend.Storage.huawei_obs:type_name -> kratos.api.AdminBackend.ObjectStore + 22, // 31: kratos.api.AdminBackend.Storage.tencent_cos:type_name -> kratos.api.AdminBackend.ObjectStore + 22, // 32: kratos.api.AdminBackend.Storage.aws_s3:type_name -> kratos.api.AdminBackend.ObjectStore + 22, // 33: kratos.api.AdminBackend.Storage.cloudflare_r2:type_name -> kratos.api.AdminBackend.ObjectStore + 22, // 34: kratos.api.AdminBackend.Storage.minio:type_name -> kratos.api.AdminBackend.ObjectStore + 35, // [35:35] is the sub-list for method output_type + 35, // [35:35] is the sub-list for method input_type + 35, // [35:35] is the sub-list for extension type_name + 35, // [35:35] is the sub-list for extension extendee + 0, // [0:35] is the sub-list for field type_name } -func init() { file_internal_conf_conf_proto_init() } -func file_internal_conf_conf_proto_init() { - if File_internal_conf_conf_proto != nil { +func init() { file_conf_conf_proto_init() } +func file_conf_conf_proto_init() { + if File_conf_conf_proto != nil { return } type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), - RawDescriptor: unsafe.Slice(unsafe.StringData(file_internal_conf_conf_proto_rawDesc), len(file_internal_conf_conf_proto_rawDesc)), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_conf_conf_proto_rawDesc), len(file_conf_conf_proto_rawDesc)), NumEnums: 0, - NumMessages: 25, + NumMessages: 23, NumExtensions: 0, NumServices: 0, }, - GoTypes: file_internal_conf_conf_proto_goTypes, - DependencyIndexes: file_internal_conf_conf_proto_depIdxs, - MessageInfos: file_internal_conf_conf_proto_msgTypes, + GoTypes: file_conf_conf_proto_goTypes, + DependencyIndexes: file_conf_conf_proto_depIdxs, + MessageInfos: file_conf_conf_proto_msgTypes, }.Build() - File_internal_conf_conf_proto = out.File - file_internal_conf_conf_proto_goTypes = nil - file_internal_conf_conf_proto_depIdxs = nil + File_conf_conf_proto = out.File + file_conf_conf_proto_goTypes = nil + file_conf_conf_proto_depIdxs = nil } diff --git a/internal/conf/conf.proto b/internal/conf/conf.proto index 13522f7..391489d 100644 --- a/internal/conf/conf.proto +++ b/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; diff --git a/internal/data/config_store.go b/internal/data/config_store.go index 7adb3d6..3b0dba9 100644 --- a/internal/data/config_store.go +++ b/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) } diff --git a/internal/data/config_watch.go b/internal/data/config_watch.go index 1917280..f679566 100644 --- a/internal/data/config_watch.go +++ b/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) diff --git a/internal/data/data.go b/internal/data/data.go index 0b24268..fa6a8c6 100644 --- a/internal/data/data.go +++ b/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) diff --git a/internal/data/initialization_backend.go b/internal/data/initialization_backend.go index 933f673..cb825ba 100644 --- a/internal/data/initialization_backend.go +++ b/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 } diff --git a/internal/data/integration_config.go b/internal/data/integration_config.go index 492c988..e5e59c7 100644 --- a/internal/data/integration_config.go +++ b/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{ diff --git a/internal/data/integration_runtime.go b/internal/data/integration_runtime.go new file mode 100644 index 0000000..fd359ca --- /dev/null +++ b/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 +} diff --git a/internal/data/repository/integration_config.go b/internal/data/repository/integration_config.go index a649587..6926c69 100644 --- a/internal/data/repository/integration_config.go +++ b/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 { diff --git a/internal/data/repository/provider.go b/internal/data/repository/provider.go index e8f1b80..218c1a4 100644 --- a/internal/data/repository/provider.go +++ b/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 } diff --git a/internal/data/websocket_config.go b/internal/data/websocket_config.go deleted file mode 100644 index 93858d0..0000000 --- a/internal/data/websocket_config.go +++ /dev/null @@ -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(¤t).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(¤t).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) -} diff --git a/internal/data/websocket_config_test.go b/internal/data/websocket_config_test.go deleted file mode 100644 index cbe2c64..0000000 --- a/internal/data/websocket_config_test.go +++ /dev/null @@ -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) - } -} diff --git a/internal/initialize/configuration.go b/internal/initialize/configuration.go index 1689b0c..81be75b 100644 --- a/internal/initialize/configuration.go +++ b/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 == "******" } diff --git a/internal/integrationruntime/store.go b/internal/integrationruntime/store.go new file mode 100644 index 0000000..7c7b529 --- /dev/null +++ b/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 +} diff --git a/internal/integrationruntime/store_test.go b/internal/integrationruntime/store_test.go new file mode 100644 index 0000000..4be9dd7 --- /dev/null +++ b/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) + } +} diff --git a/pkg/mq/mq.go b/pkg/mq/mq.go index f06a0b7..0df0e98 100644 --- a/pkg/mq/mq.go +++ b/pkg/mq/mq.go @@ -35,6 +35,16 @@ type Client interface { Close() error } +// Registry exposes named broker clients while preserving Client as the +// default EMQX/MQTT boundary for existing modules. +type Registry interface { + Client(provider string) Client + PublishTo(context.Context, string, string, []byte, byte, bool) error + SubscribeTo(context.Context, string, string, byte, Handler) error + UnsubscribeFrom(context.Context, string, ...string) error + ConnectedTo(provider string) bool +} + type Config struct { Enabled bool Broker string diff --git a/web/src/pathInfo.json b/web/src/pathInfo.json index bc346f5..e26b714 100644 --- a/web/src/pathInfo.json +++ b/web/src/pathInfo.json @@ -55,6 +55,7 @@ "/src/view/system/security/forceChangePassword.vue": "ForceChangePassword", "/src/view/system/security/index.vue": "SecurityConfig", "/src/view/system/state.vue": "State", + "/src/view/systemTools/integration/config.vue": "IntegrationConfig", "/src/view/systemTools/logViewer/index.vue": "LogViewer", "/src/view/systemTools/sysError/sysError.vue": "SysError", "/src/view/systemTools/system/system.vue": "Config", diff --git a/web/src/view/systemTools/integration/config.vue b/web/src/view/systemTools/integration/config.vue new file mode 100644 index 0000000..8246360 --- /dev/null +++ b/web/src/view/systemTools/integration/config.vue @@ -0,0 +1,762 @@ + + + + +