优化结构

internal/integration/connectivity.go

@@ -0,0 +1,86 @@
+package integration
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"kra/internal/biz"
+	"kra/internal/integration/mq"
+	"kra/internal/integration/runtimeconfig"
+	websocketintegration "kra/internal/integration/websocket"
+)
+
+// ConnectivityTester probes candidate communication settings without changing
+// the active clients or writing anything to sys_integration_configs.
+type ConnectivityTester struct {
+	store *runtimeconfig.Store
+}
+
+func NewConnectivityTester(store *runtimeconfig.Store) *ConnectivityTester {
+	return &ConnectivityTester{store: store}
+}
+
+func (t *ConnectivityTester) TestIntegration(ctx context.Context, config *biz.IntegrationConfig) error {
+	if config == nil {
+		return errors.New("集成配置请求为空")
+	}
+	values := map[string]any{}
+	if err := json.Unmarshal(config.Values, &values); err != nil {
+		return fmt.Errorf("解析集成配置失败: %w", err)
+	}
+	if err := t.restoreMaskedSecrets(config.Kind, config.Provider, values); err != nil {
+		return err
+	}
+	raw, err := json.Marshal(values)
+	if err != nil {
+		return fmt.Errorf("编码集成配置失败: %w", err)
+	}
+	switch strings.ToLower(strings.TrimSpace(config.Kind)) {
+	case biz.IntegrationKindMQ:
+		return mq.TestConfig(ctx, config.Provider, raw)
+	case biz.IntegrationKindWebSocket:
+		if strings.ToLower(strings.TrimSpace(config.Provider)) != websocketintegration.ProviderMelody {
+			return fmt.Errorf("不支持的 WebSocket provider %q", config.Provider)
+		}
+		return websocketintegration.TestConfig(ctx, raw)
+	default:
+		return fmt.Errorf("不支持测试集成类型 %q", config.Kind)
+	}
+}
+
+func (t *ConnectivityTester) restoreMaskedSecrets(kind, provider string, values map[string]any) error {
+	definition, ok := biz.IntegrationDefinition(kind, provider)
+	if !ok {
+		return fmt.Errorf("不支持的集成 %s/%s", kind, provider)
+	}
+	masked := make(map[string]struct{})
+	for _, field := range definition.Fields {
+		if field.Secret {
+			masked[field.Key] = struct{}{}
+		}
+	}
+	if len(masked) == 0 {
+		return nil
+	}
+	currentValues := map[string]any{}
+	if t != nil && t.store != nil {
+		if current, exists := t.store.Get(kind, provider); exists {
+			_ = json.Unmarshal(current.Values, &currentValues)
+		}
+	}
+	for key := range masked {
+		value, _ := values[key].(string)
+		if strings.TrimSpace(value) != "******" {
+			continue
+		}
+		prior, _ := currentValues[key].(string)
+		if strings.TrimSpace(prior) == "" || strings.TrimSpace(prior) == "******" {
+			return fmt.Errorf("配置字段 %s 已脱敏，请重新填写后再测试", key)
+		}
+		values[key] = prior
+	}
+	return nil
+}

internal/integration/mq/emqx.go

@@ -72,6 +72,51 @@ func storeConfig(store *runtimeconfig.Store, provider string) runtimeconfig.Conf
 	return config
 }
 
+// TestConfig creates a short-lived provider client and closes it immediately.
+// For RabbitMQ this also checks the configured exchange and queue topology.
+func TestConfig(ctx context.Context, provider string, raw json.RawMessage) error {
+	if ctx != nil {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+	}
+	provider = strings.ToLower(strings.TrimSpace(provider))
+	if provider == ProviderEMQX {
+		values := map[string]any{}
+		if err := json.Unmarshal(raw, &values); err != nil {
+			return fmt.Errorf("decode %s configuration: %w", provider, err)
+		}
+		baseID := configText(values, "client_id")
+		values["client_id"] = fmt.Sprintf("%s-test-%d", baseID, time.Now().UnixNano())
+		encoded, err := json.Marshal(values)
+		if err != nil {
+			return fmt.Errorf("encode %s test configuration: %w", provider, err)
+		}
+		raw = encoded
+	}
+	client, err := newProviderClient(provider, raw)
+	if err != nil {
+		return err
+	}
+	if client == nil || !client.Connected() {
+		if client != nil {
+			_ = client.Close()
+		}
+		return platformmq.ErrUnavailable
+	}
+	closeErr := client.Close()
+	if ctx != nil {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+	}
+	return closeErr
+}
+
 func (r *Reloadable) apply(provider string, config runtimeconfig.Config) {
 	r.opMu.Lock()
 	defer r.opMu.Unlock()

internal/integration/provider.go

@@ -20,6 +20,8 @@ var ProviderSet = wire.NewSet(
 	cache.New,
 	email.NewEmailRepo,
 	storage.NewFileStorage,
+	NewConnectivityTester,
+	wire.Bind(new(biz.IntegrationConnectionTester), new(*ConnectivityTester)),
 	mqintegration.New,
 	wire.Bind(new(mq.Client), new(*mqintegration.Reloadable)),
 	wire.Bind(new(mq.Registry), new(*mqintegration.Reloadable)),

internal/integration/websocket/server.go

@@ -1,15 +1,18 @@
 package websocket
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
+	"net/http/httptest"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
+	gorillawebsocket "github.com/gorilla/websocket"
 	melody "github.com/olahol/melody"
 	"kra/internal/integration/runtimeconfig"
 	platformws "kra/pkg/websocket"
@@ -57,6 +60,71 @@ func storeConfig(store *runtimeconfig.Store) runtimeconfig.Config {
 	return config
 }
 
+// TestConfig performs a local WebSocket handshake using a temporary server
+// built from the candidate settings. It does not touch the live endpoint.
+func TestConfig(ctx context.Context, raw json.RawMessage) error {
+	values := map[string]any{}
+	if err := json.Unmarshal(raw, &values); err != nil {
+		return fmt.Errorf("decode websocket configuration: %w", err)
+	}
+	path := text(values, "path")
+	if path == "" {
+		path = "/ws"
+	}
+	if !strings.HasPrefix(path, "/") {
+		path = "/" + path
+	}
+	temporary := platformws.New(platformws.Config{
+		WriteWait:                 durationValue(values, "write_wait", 10*time.Second),
+		PongWait:                  durationValue(values, "pong_wait", 60*time.Second),
+		PingPeriod:                durationValue(values, "ping_period", 54*time.Second),
+		MaxMessageSize:            int64Value(values, "max_message_size"),
+		MessageBufferSize:         int(intValue(values, "message_buffer_size")),
+		ConcurrentMessageHandling: boolValue(values, "concurrent_message_handling"),
+		AllowOrigins:              stringList(values, "allow_origins"),
+	})
+	defer temporary.Close()
+	httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != path {
+			http.NotFound(w, r)
+			return
+		}
+		if err := temporary.HandleRequest(w, r); err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+		}
+	}))
+	defer httpServer.Close()
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	headers := http.Header{}
+	origins := stringList(values, "allow_origins")
+	if len(origins) > 0 {
+		origin := origins[0]
+		if origin == "*" {
+			origin = "http://localhost"
+		}
+		headers.Set("Origin", origin)
+	}
+	wsURL := "ws" + strings.TrimPrefix(httpServer.URL, "http") + path
+	connection, response, err := gorillawebsocket.DefaultDialer.DialContext(ctx, wsURL, headers)
+	if response != nil && response.Body != nil {
+		_ = response.Body.Close()
+	}
+	if err != nil {
+		return fmt.Errorf("websocket handshake failed: %w", err)
+	}
+	if connection == nil {
+		return errors.New("websocket handshake returned an empty connection")
+	}
+	return connection.Close()
+}
+
 func (s *Server) apply(config runtimeconfig.Config) {
 	if s == nil {
 		return