diff --git a/internal/biz/integration/integration_config.go b/internal/biz/integration/integration_config.go index ed106ad..c8cf680 100644 --- a/internal/biz/integration/integration_config.go +++ b/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{} - if err := json.Unmarshal(config.Values, &values); err != nil { - return errors.New("集成配置必须是 JSON 对象") + values, err := decodeIntegrationObject(config.Values) + if err != nil { + 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{} - if err := json.Unmarshal(config.Values, &values); err != nil { - return errors.New("集成配置必须是 JSON 对象") + values, err := decodeIntegrationObject(config.Values) + if err != nil { + 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] diff --git a/internal/biz/integration/integration_config_communication_test.go b/internal/biz/integration/integration_config_communication_test.go index 23d961e..96c138c 100644 --- a/internal/biz/integration/integration_config_communication_test.go +++ b/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") + } +} diff --git a/internal/data/integration/integration_config.go b/internal/data/integration/integration_config.go index 46589ee..7dea23a 100644 --- a/internal/data/integration/integration_config.go +++ b/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 { diff --git a/internal/data/integration/providers.go b/internal/data/integration/providers.go index a862518..c325c87 100644 --- a/internal/data/integration/providers.go +++ b/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) diff --git a/internal/data/payment/payment.go b/internal/data/payment/payment.go index d4db027..6300522 100644 --- a/internal/data/payment/payment.go +++ b/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 } - -func NewPaymentRepo(data Provider) bizpayment.PaymentRepo { return &paymentRepo{data: data} } - -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 +type paymentRepo struct { + data Provider + config integrationbiz.PaymentConfigReader } -func (r *paymentRepo) row(ctx context.Context, provider string) (*dataintegration.ConfigPO, map[string]any, error) { - var row dataintegration.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, nil, bizpayment.ErrPaymentProviderNotFound - } - return nil, nil, err +func NewPaymentRepo(data Provider, config integrationbiz.PaymentConfigReader) bizpayment.PaymentRepo { + return &paymentRepo{data: data, config: config} +} + +func (r *paymentRepo) values(ctx context.Context, provider string) (map[string]any, error) { + if r == nil || r.config == nil { + return nil, errors.New("支付配置仓储未接入") } - if !row.Enabled { - return nil, nil, fmt.Errorf("支付渠道 %s 未启用", provider) + config, err := r.config.ReadPaymentConfig(ctx, 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 { - return nil, nil, fmt.Errorf("支付配置格式错误: %w", err) + if err := json.Unmarshal(config.Values, &values); err != nil { + 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 } diff --git a/internal/data/payment/provider.go b/internal/data/payment/provider.go index 9c968c7..8e17764 100644 --- a/internal/data/payment/provider.go +++ b/internal/data/payment/provider.go @@ -1,8 +1,6 @@ package payment -import ( - "gorm.io/gorm" -) +import "gorm.io/gorm" // Provider is the narrow persistence seam required by payment repositories. // Keeping it here lets payment remain an independent data module. diff --git a/internal/integration/runtimeconfig/store.go b/internal/integration/runtimeconfig/store.go index efa4553..6e5fef9 100644 --- a/internal/integration/runtimeconfig/store.go +++ b/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() { diff --git a/internal/integration/runtimeconfig/store_test.go b/internal/integration/runtimeconfig/store_test.go index 8327ab1..4e3c432 100644 --- a/internal/integration/runtimeconfig/store_test.go +++ b/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): + } +} diff --git a/internal/integration/storage/aliyun_storage.go b/internal/integration/storage/aliyun_storage.go index aedee5c..31274be 100644 --- a/internal/integration/storage/aliyun_storage.go +++ b/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 } diff --git a/internal/integration/storage/aws_storage.go b/internal/integration/storage/aws_storage.go index 881066d..2938d30 100644 --- a/internal/integration/storage/aws_storage.go +++ b/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) } diff --git a/internal/integration/storage/compose_test.go b/internal/integration/storage/compose_test.go index 75db552..399d979 100644 --- a/internal/integration/storage/compose_test.go +++ b/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") diff --git a/internal/integration/storage/huawei_storage.go b/internal/integration/storage/huawei_storage.go index 26df589..ca9707a 100644 --- a/internal/integration/storage/huawei_storage.go +++ b/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 } diff --git a/internal/integration/storage/prefix.go b/internal/integration/storage/prefix.go index 19b7628..1eded18 100644 --- a/internal/integration/storage/prefix.go +++ b/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 +} diff --git a/internal/integration/storage/qiniu_storage.go b/internal/integration/storage/qiniu_storage.go index 4c6f693..f0eb996 100644 --- a/internal/integration/storage/qiniu_storage.go +++ b/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 } diff --git a/internal/integration/storage/s3_storage.go b/internal/integration/storage/s3_storage.go index 344c4c9..9608487 100644 --- a/internal/integration/storage/s3_storage.go +++ b/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 diff --git a/internal/integration/storage/tencent_storage.go b/internal/integration/storage/tencent_storage.go index a71e85b..cfe5b7d 100644 --- a/internal/integration/storage/tencent_storage.go +++ b/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 }