优化结构

internal/biz/integration/integration_config.go

@@ -57,6 +57,23 @@ type IntegrationConfigRepo interface {
 	DeleteIntegrationConfig(context.Context, string, string) error
 }
 
+// ErrPaymentConfigNotFound marks an absent payment integration row without
+// exposing the storage driver's not-found error to the payment data module.
+var ErrPaymentConfigNotFound = errors.New("支付渠道配置不存在")
+
+// PaymentConfig is the storage-neutral, unmasked snapshot used by payment
+// persistence. It deliberately contains no ORM or table metadata.
+type PaymentConfig struct {
+	Enabled bool
+	Values  json.RawMessage
+}
+
+// PaymentConfigReader is the narrow inversion seam between the payment and
+// integration data modules. The integration module owns its ConfigPO.
+type PaymentConfigReader interface {
+	ReadPaymentConfig(context.Context, string) (*PaymentConfig, error)
+}
+
 type IntegrationConnectionTester interface {
 	TestIntegration(context.Context, *IntegrationConfig) error
 }
@@ -98,9 +115,9 @@ func (uc *IntegrationConfigUsecase) Save(ctx context.Context, config *Integratio
 	if !json.Valid(config.Values) {
 		return errors.New("集成配置必须是合法 JSON")
 	}
-	values := map[string]any{}
+	values, err := decodeIntegrationObject(config.Values)
-	if err := json.Unmarshal(config.Values, &values); err != nil {
+	if err != nil {
-		return errors.New("集成配置必须是 JSON 对象")
+		return err
 	}
 	if definition, ok := IntegrationDefinition(config.Kind, config.Provider); ok {
 		values = mergeIntegrationDefaults(definition.Defaults, values)
@@ -131,9 +148,9 @@ func (uc *IntegrationConfigUsecase) Test(ctx context.Context, config *Integratio
 	if !json.Valid(config.Values) {
 		return errors.New("集成配置必须是合法 JSON")
 	}
-	values := map[string]any{}
+	values, err := decodeIntegrationObject(config.Values)
-	if err := json.Unmarshal(config.Values, &values); err != nil {
+	if err != nil {
-		return errors.New("集成配置必须是 JSON 对象")
+		return err
 	}
 	if definition, ok := IntegrationDefinition(config.Kind, config.Provider); ok {
 		values = mergeIntegrationDefaults(definition.Defaults, values)
@@ -162,6 +179,14 @@ func normalizeIntegrationPart(value string) string {
 	return strings.ToLower(strings.TrimSpace(value))
 }
 
+func decodeIntegrationObject(raw json.RawMessage) (map[string]any, error) {
+	values := map[string]any{}
+	if err := json.Unmarshal(raw, &values); err != nil || values == nil {
+		return nil, errors.New("集成配置必须是 JSON 对象")
+	}
+	return values, nil
+}
+
 func IntegrationDefinitions(kind string) []IntegrationConfigDefinition {
 	kind = normalizeIntegrationPart(kind)
 	definitions := integrationDefinitions[kind]

internal/biz/integration/integration_config_communication_test.go

@@ -96,3 +96,14 @@ func TestIntegrationConfigTestDoesNotPersistCandidate(t *testing.T) {
 		t.Fatalf("tested values = %#v", values)
 	}
 }
+
+func TestIntegrationConfigRejectsJSONNull(t *testing.T) {
+	repo := &integrationConfigRepoTestDouble{}
+	usecase := NewIntegrationConfigUsecase(repo, &integrationConnectionTesterDouble{})
+	if err := usecase.Save(context.Background(), &IntegrationConfig{Kind: IntegrationKindMQ, Provider: "emqx", Values: json.RawMessage("null")}); err == nil {
+		t.Fatal("Save() accepted JSON null as an object")
+	}
+	if err := usecase.Test(context.Background(), &IntegrationConfig{Kind: IntegrationKindMQ, Provider: "emqx", Values: json.RawMessage("null")}); err == nil {
+		t.Fatal("Test() accepted JSON null as an object")
+	}
+}

internal/data/integration/integration_config.go

@@ -27,10 +27,38 @@ func (ConfigPO) TableName() string { return "sys_integration_configs" }
 
 type integrationConfigRepo struct{ data Provider }
 
+type paymentConfigReader struct{ data Provider }
+
 func NewIntegrationConfigRepo(data Provider) integrationbiz.IntegrationConfigRepo {
 	return &integrationConfigRepo{data: data}
 }
 
+// NewPaymentConfigReader exposes only the raw payment configuration needed by
+// the payment data module. The ConfigPO and its table name stay private here.
+func NewPaymentConfigReader(data Provider) integrationbiz.PaymentConfigReader {
+	return &paymentConfigReader{data: data}
+}
+
+func (r *paymentConfigReader) ReadPaymentConfig(ctx context.Context, provider string) (*integrationbiz.PaymentConfig, error) {
+	if r == nil || r.data == nil || r.data.DB() == nil {
+		return nil, errors.New("集成配置数据库未初始化")
+	}
+	provider = strings.ToLower(strings.TrimSpace(provider))
+	if provider == "" {
+		return nil, errors.New("支付渠道不能为空")
+	}
+	var row ConfigPO
+	if err := r.data.DB().WithContext(ctx).
+		Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindPayment, provider).
+		First(&row).Error; err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, integrationbiz.ErrPaymentConfigNotFound
+		}
+		return nil, err
+	}
+	return &integrationbiz.PaymentConfig{Enabled: row.Enabled, Values: append(json.RawMessage(nil), []byte(row.Config)...)}, nil
+}
+
 func (r *integrationConfigRepo) ListIntegrationConfigs(ctx context.Context, kind string) ([]*integrationbiz.IntegrationConfig, error) {
 	var rows []ConfigPO
 	if err := r.data.DB().WithContext(ctx).Where("kind = ?", kind).Order("provider ASC").Find(&rows).Error; err != nil {

internal/data/integration/providers.go

@@ -2,4 +2,4 @@ package integration
 
 import "github.com/google/wire"
 
-var ProviderSet = wire.NewSet(NewIntegrationConfigRepo)
+var ProviderSet = wire.NewSet(NewIntegrationConfigRepo, NewPaymentConfigReader)

internal/data/payment/payment.go

@@ -9,7 +9,6 @@ import (
 	"fmt"
 	integrationbiz "kra/internal/biz/integration"
 	bizpayment "kra/internal/biz/payment"
-	dataintegration "kra/internal/data/integration"
 	"net/url"
 	"strconv"
 	"strings"
@@ -18,67 +17,40 @@ import (
 	datapayment "kra/internal/integration/payment"
 
 	"github.com/google/uuid"
-	"gorm.io/gorm"
 )
 
-type paymentRepo struct{ data Provider }
+type paymentRepo struct {
-
+	data   Provider
-func NewPaymentRepo(data Provider) bizpayment.PaymentRepo { return &paymentRepo{data: data} }
+	config integrationbiz.PaymentConfigReader
-
-func ensurePaymentIntegrationConfigs(db *gorm.DB) error {
-	for _, provider := range bizpayment.SupportedPaymentProviders {
-		var row dataintegration.ConfigPO
-		err := db.Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindPayment, provider).First(&row).Error
-		defaults := integrationbiz.DefaultIntegrationConfig(integrationbiz.IntegrationKindPayment, provider)
-		if errors.Is(err, gorm.ErrRecordNotFound) {
-			encoded, _ := json.Marshal(defaults)
-			if err := db.Create(&dataintegration.ConfigPO{Kind: integrationbiz.IntegrationKindPayment, Provider: provider, Enabled: false, Config: string(encoded)}).Error; err != nil {
-				return err
-			}
-			continue
-		}
-		if err != nil {
-			return err
-		}
-		values := map[string]any{}
-		_ = json.Unmarshal([]byte(row.Config), &values)
-		changed := false
-		for key, value := range defaults {
-			if _, exists := values[key]; !exists {
-				values[key] = value
-				changed = true
-			}
-		}
-		if changed {
-			encoded, _ := json.Marshal(values)
-			if err := db.Model(&row).Update("config", string(encoded)).Error; err != nil {
-				return err
-			}
-		}
-	}
-	return nil
 }
 
-func (r *paymentRepo) row(ctx context.Context, provider string) (*dataintegration.ConfigPO, map[string]any, error) {
+func NewPaymentRepo(data Provider, config integrationbiz.PaymentConfigReader) bizpayment.PaymentRepo {
-	var row dataintegration.ConfigPO
+	return &paymentRepo{data: data, config: config}
-	if err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindPayment, provider).First(&row).Error; err != nil {
+}
-		if errors.Is(err, gorm.ErrRecordNotFound) {
+
-			return nil, nil, bizpayment.ErrPaymentProviderNotFound
+func (r *paymentRepo) values(ctx context.Context, provider string) (map[string]any, error) {
-		}
+	if r == nil || r.config == nil {
-		return nil, nil, err
+		return nil, errors.New("支付配置仓储未接入")
 	}
-	if !row.Enabled {
+	config, err := r.config.ReadPaymentConfig(ctx, provider)
-		return nil, nil, fmt.Errorf("支付渠道 %s 未启用", provider)
+	if err != nil {
+		if errors.Is(err, integrationbiz.ErrPaymentConfigNotFound) {
+			return nil, bizpayment.ErrPaymentProviderNotFound
+		}
+		return nil, err
+	}
+	if config == nil || !config.Enabled {
+		return nil, fmt.Errorf("支付渠道 %s 未启用", provider)
 	}
 	values := map[string]any{}
-	if err := json.Unmarshal([]byte(row.Config), &values); err != nil {
+	if err := json.Unmarshal(config.Values, &values); err != nil {
-		return nil, nil, fmt.Errorf("支付配置格式错误: %w", err)
+		return nil, fmt.Errorf("支付配置格式错误: %w", err)
 	}
-	return &row, values, nil
+	return values, nil
 }
 
 func (r *paymentRepo) adapter(ctx context.Context, provider string) (datapayment.Adapter, map[string]any, error) {
-	_, values, err := r.row(ctx, provider)
+	values, err := r.values(ctx, provider)
 	if err != nil {
 		return nil, nil, err
 	}

internal/data/payment/provider.go

@@ -1,8 +1,6 @@
 package payment
 
-import (
+import "gorm.io/gorm"
-	"gorm.io/gorm"
-)
 
 // Provider is the narrow persistence seam required by payment repositories.
 // Keeping it here lets payment remain an independent data module.

internal/integration/runtimeconfig/store.go

@@ -68,6 +68,13 @@ func (s *Store) Set(config Config) {
 	config = cloneConfig(config)
 	key := configKey(config.Kind, config.Provider)
 	s.mu.Lock()
+	if s.values == nil {
+		s.values = make(map[string]Config)
+	}
+	if previous, exists := s.values[key]; exists && sameConfig(previous, config) {
+		s.mu.Unlock()
+		return
+	}
 	s.values[key] = config
 	callbacks := s.matchingListenersLocked(config.Kind, config.Provider)
 	s.mu.Unlock()
@@ -83,7 +90,12 @@ func (s *Store) Delete(kind, provider string) {
 	kind = strings.ToLower(strings.TrimSpace(kind))
 	provider = strings.ToLower(strings.TrimSpace(provider))
 	s.mu.Lock()
-	delete(s.values, configKey(kind, provider))
+	key := configKey(kind, provider)
+	if _, exists := s.values[key]; !exists {
+		s.mu.Unlock()
+		return
+	}
+	delete(s.values, key)
 	callbacks := s.matchingListenersLocked(kind, provider)
 	s.mu.Unlock()
 	config := Config{Kind: kind, Provider: provider}
@@ -142,6 +154,9 @@ func (s *Store) Subscribe(kind, provider string, callback func(Config)) func() {
 	s.mu.Lock()
 	s.nextID++
 	id := s.nextID
+	if s.listeners == nil {
+		s.listeners = make(map[uint64]listener)
+	}
 	s.listeners[id] = listener{kind: strings.ToLower(strings.TrimSpace(kind)), provider: strings.ToLower(strings.TrimSpace(provider)), callback: callback}
 	s.mu.Unlock()
 	return func() {

internal/integration/runtimeconfig/store_test.go

@@ -63,3 +63,23 @@ func TestStoreReplaceSkipsUnchangedValues(t *testing.T) {
 		t.Fatal("changed replace notification was not delivered")
 	}
 }
+
+func TestStoreSetSkipsUnchangedValues(t *testing.T) {
+	var store Store
+	updates := make(chan Config, 1)
+	stop := store.Subscribe("mq", "rabbitmq", func(config Config) { updates <- config })
+	defer stop()
+	config := Config{Kind: "mq", Provider: "rabbitmq", Enabled: true, Values: json.RawMessage(`{"host":"localhost"}`)}
+	store.Set(config)
+	select {
+	case <-updates:
+	case <-time.After(time.Second):
+		t.Fatal("initial set notification was not delivered")
+	}
+	store.Set(config)
+	select {
+	case update := <-updates:
+		t.Fatalf("unchanged set emitted notification: %#v", update)
+	case <-time.After(20 * time.Millisecond):
+	}
+}

internal/integration/storage/aliyun_storage.go

@@ -75,7 +75,7 @@ func (s *aliyunStorage) DeletePrefix(ctx context.Context, prefix string) error {
 	}
 	cursor := ""
 	for {
-		items, next, more, err := s.List(ctx, prefix, cursor, 1000)
+		items, next, more, err := s.List(ctx, prefix+"/", cursor, 1000)
 		if err != nil {
 			return err
 		}
@@ -97,7 +97,7 @@ func (s *aliyunStorage) List(_ context.Context, prefix, cursor string, limit int
 	if limit <= 0 {
 		limit = 100
 	}
-	result, err := s.bucket.ListObjects(oss.Prefix(s.key(prefix)), oss.Marker(cursor), oss.MaxKeys(limit))
+	result, err := s.bucket.ListObjects(oss.Prefix(boundedPrefix(s.key(prefix), prefix)), oss.Marker(cursor), oss.MaxKeys(limit))
 	if err != nil {
 		return nil, "", false, err
 	}

internal/integration/storage/aws_storage.go

@@ -112,7 +112,7 @@ func (s *awsStorage) DeletePrefix(ctx context.Context, prefix string) error {
 	}
 	cursor := ""
 	for {
-		items, next, more, err := s.List(ctx, prefix, cursor, 1000)
+		items, next, more, err := s.List(ctx, prefix+"/", cursor, 1000)
 		if err != nil {
 			return err
 		}
@@ -134,7 +134,7 @@ func (s *awsStorage) List(ctx context.Context, prefix, cursor string, limit int)
 	if limit <= 0 {
 		limit = 100
 	}
-	input := &s3.ListObjectsV2Input{Bucket: aws.String(s.bucket), Prefix: aws.String(s.key(prefix)), MaxKeys: aws.Int32(int32(limit))}
+	input := &s3.ListObjectsV2Input{Bucket: aws.String(s.bucket), Prefix: aws.String(boundedPrefix(s.key(prefix), prefix)), MaxKeys: aws.Int32(int32(limit))}
 	if cursor != "" {
 		input.ContinuationToken = aws.String(cursor)
 	}

internal/integration/storage/compose_test.go

@@ -100,6 +100,15 @@ func TestNormalizeDeletePrefix(t *testing.T) {
 	}
 }
 
+func TestBoundedPrefixKeepsDirectoryBoundary(t *testing.T) {
+	if got := boundedPrefix("uploads/chunks/1", "uploads/chunks/1/"); got != "uploads/chunks/1/" {
+		t.Fatalf("boundedPrefix() = %q", got)
+	}
+	if got := boundedPrefix("uploads/chunks/1", "uploads/chunks/1"); got != "uploads/chunks/1" {
+		t.Fatalf("boundedPrefix() changed ordinary prefix to %q", got)
+	}
+}
+
 func TestAdvanceDeletePrefixCursor(t *testing.T) {
 	if _, err := advanceDeletePrefixCursor("cursor", "cursor", true); err == nil {
 		t.Fatal("same cursor should fail")

internal/integration/storage/huawei_storage.go

@@ -74,7 +74,7 @@ func (s *huaweiStorage) DeletePrefix(ctx context.Context, prefix string) error {
 	}
 	cursor := ""
 	for {
-		items, next, more, err := s.List(ctx, prefix, cursor, 1000)
+		items, next, more, err := s.List(ctx, prefix+"/", cursor, 1000)
 		if err != nil {
 			return err
 		}
@@ -96,7 +96,7 @@ func (s *huaweiStorage) List(_ context.Context, prefix, cursor string, limit int
 	if limit <= 0 {
 		limit = 100
 	}
-	result, err := s.client.ListObjects(&obs.ListObjectsInput{ListObjsInput: obs.ListObjsInput{Prefix: s.key(prefix), MaxKeys: limit}, Bucket: s.bucket, Marker: cursor})
+	result, err := s.client.ListObjects(&obs.ListObjectsInput{ListObjsInput: obs.ListObjsInput{Prefix: boundedPrefix(s.key(prefix), prefix), MaxKeys: limit}, Bucket: s.bucket, Marker: cursor})
 	if err != nil {
 		return nil, "", false, err
 	}

internal/integration/storage/prefix.go

@@ -39,3 +39,10 @@ func advanceDeletePrefixCursor(current, next string, more bool) (string, error)
 	}
 	return next, nil
 }
+
+func boundedPrefix(key, prefix string) string {
+	if strings.HasSuffix(strings.ReplaceAll(prefix, "\\", "/"), "/") && !strings.HasSuffix(key, "/") {
+		return key + "/"
+	}
+	return key
+}

internal/integration/storage/qiniu_storage.go

@@ -102,7 +102,7 @@ func (s *qiniuStorage) DeletePrefix(ctx context.Context, prefix string) error {
 	}
 	cursor := ""
 	for {
-		items, next, more, err := s.List(ctx, prefix, cursor, 1000)
+		items, next, more, err := s.List(ctx, prefix+"/", cursor, 1000)
 		if err != nil {
 			return err
 		}
@@ -124,7 +124,7 @@ func (s *qiniuStorage) List(ctx context.Context, prefix, cursor string, limit in
 	if limit <= 0 {
 		limit = 100
 	}
-	entries, _, marker, more, err := s.manager.ListFiles(s.config.Bucket, prefix, "", cursor, limit)
+	entries, _, marker, more, err := s.manager.ListFiles(s.config.Bucket, boundedPrefix(strings.TrimPrefix(prefix, "/"), prefix), "", cursor, limit)
 	if err != nil {
 		return nil, "", false, err
 	}

internal/integration/storage/s3_storage.go

@@ -102,7 +102,7 @@ func (s *s3Storage) DeletePrefix(ctx context.Context, prefix string) error {
 	if err != nil {
 		return err
 	}
-	items := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: s.key(prefix), Recursive: true})
+	items := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: boundedPrefix(s.key(prefix), prefix), Recursive: true})
 	for item := range items {
 		if item.Err != nil {
 			return item.Err

internal/integration/storage/tencent_storage.go

@@ -88,7 +88,7 @@ func (s *tencentStorage) DeletePrefix(ctx context.Context, prefix string) error
 	}
 	cursor := ""
 	for {
-		items, next, more, err := s.List(ctx, prefix, cursor, 1000)
+		items, next, more, err := s.List(ctx, prefix+"/", cursor, 1000)
 		if err != nil {
 			return err
 		}
@@ -110,7 +110,7 @@ func (s *tencentStorage) List(ctx context.Context, prefix, cursor string, limit
 	if limit <= 0 {
 		limit = 100
 	}
-	result, _, err := s.client.Bucket.Get(ctx, &cos.BucketGetOptions{Prefix: s.key(prefix), Marker: cursor, MaxKeys: limit})
+	result, _, err := s.client.Bucket.Get(ctx, &cos.BucketGetOptions{Prefix: boundedPrefix(s.key(prefix), prefix), Marker: cursor, MaxKeys: limit})
 	if err != nil {
 		return nil, "", false, err
 	}