package integrationlog_test import ( "context" "fmt" "strings" "sync" "testing" "gorm.io/gorm" "github.com/break/junhong_cmp_fiber/internal/infrastructure/integrationlog" "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/internal/testutil" "github.com/break/junhong_cmp_fiber/pkg/constants" ) func TestOutboundAttemptPersistsBeforeConditionalCompletion(t *testing.T) { db := testutil.NewPostgresTransaction(t) createTemporaryIntegrationLogTable(t, db) repository := integrationlog.NewRepository(db) attempt, err := repository.Start(context.Background(), integrationlog.Attempt{ IntegrationID: "integration-outbound-1", Provider: "gateway", Direction: constants.IntegrationDirectionOutbound, Operation: "query_card_status", ResourceType: "iot_card", ResourceID: testutil.StringPointer("1001"), RequestSummary: map[string]any{ "iccid": "8986001234567890123", "access_token": "must-not-persist", "credential": "must-not-persist", "private_url": "https://must-not-persist.example", }, }) if err != nil { t.Fatalf("持久化外部尝试失败:%v", err) } if attempt.Result != constants.IntegrationResultPending { t.Fatalf("调用前必须是待处理状态,得到 %q", attempt.Result) } if strings.Contains(string(attempt.RequestSummary), "must-not-persist") { t.Fatalf("请求摘要泄露禁止字段:%s", attempt.RequestSummary) } auditEventID := uint(77) completed, err := repository.Complete(context.Background(), attempt.IntegrationID, integrationlog.Completion{ Result: constants.IntegrationResultSuccess, HTTPStatus: 200, ProviderCode: "0", ProviderMessage: "查询成功", StateChanged: true, AuditEventID: &auditEventID, ResponseSummary: map[string]any{"status": "active", "secret": "must-not-persist"}, }) if err != nil { t.Fatalf("终结外部尝试失败:%v", err) } if completed.Result != constants.IntegrationResultSuccess || !completed.StateChanged || completed.AuditEventID == nil || *completed.AuditEventID != auditEventID { t.Fatalf("外部尝试终态错误:%+v", completed) } if completed.ProviderMessage == nil || strings.Contains(*completed.ProviderMessage, "查询成功") { t.Fatalf("不可信渠道消息必须转换为不可逆摘要:%+v", completed.ProviderMessage) } if _, err := repository.Complete(context.Background(), attempt.IntegrationID, integrationlog.Completion{ Result: constants.IntegrationResultFailed, }); err == nil { t.Fatal("既有终态不得被重复完成改写") } } func TestIntegrationResultNamesCoverEveryPublicTerminalState(t *testing.T) { for _, result := range []string{ constants.IntegrationResultSuccess, constants.IntegrationResultFailed, constants.IntegrationResultUnknown, constants.IntegrationResultNotFound, constants.IntegrationResultInvalidPayload, constants.IntegrationResultIgnored, constants.IntegrationResultMerged, constants.IntegrationResultRateLimited, constants.IntegrationResultCompleted, constants.IntegrationResultCancelled, } { if constants.IntegrationResultName(result) == "" { t.Fatalf("公开终态 %q 缺少中文名称", result) } } } func TestInboundAttemptIsIdempotentAndNeverPersistsRawPayload(t *testing.T) { db := testutil.NewPostgresTransaction(t) createTemporaryIntegrationLogTable(t, db) repository := integrationlog.NewRepository(db) input := integrationlog.InboundAttempt{ IntegrationID: "integration-inbound-1", IdempotencyKey: "wechat:callback:transaction-1", Provider: "wechat", Operation: "payment_callback", ExternalID: "transaction-1", RawPayload: []byte(`{"sign":"raw-signature","ciphertext":"raw-ciphertext"}`), ContentType: "application/json", } first, created, err := repository.RecordInbound(context.Background(), input) if err != nil || !created { t.Fatalf("首次保存入站尝试失败:created=%v err=%v", created, err) } second, created, err := repository.RecordInbound(context.Background(), input) if err != nil || created { t.Fatalf("重复入站尝试应返回既有事实:created=%v err=%v", created, err) } if first.ID != second.ID || first.ContentHash == "" { t.Fatalf("重复回调未复用稳定事实或缺少内容哈希:first=%+v second=%+v", first, second) } serialized := string(first.RequestSummary) if strings.Contains(serialized, "raw-signature") || strings.Contains(serialized, "raw-ciphertext") { t.Fatalf("入站摘要泄露原始载荷:%s", serialized) } conflict := input conflict.IntegrationID = "integration-inbound-conflict" conflict.RawPayload = []byte(`{"different":true}`) if _, _, err := repository.RecordInbound(context.Background(), conflict); err == nil { t.Fatal("相同入站幂等标识对应不同载荷时必须拒绝") } } func TestUnknownResultRequiresExplicitRecoveryAndUnsentAttemptIsTerminal(t *testing.T) { db := testutil.NewPostgresTransaction(t) createTemporaryIntegrationLogTable(t, db) repository := integrationlog.NewRepository(db) pending, err := repository.Start(context.Background(), integrationlog.Attempt{ IntegrationID: "integration-unknown-1", Provider: "wecom", Direction: constants.IntegrationDirectionOutbound, Operation: "submit_approval", }) if err != nil { t.Fatalf("准备结果未知尝试失败:%v", err) } if _, err := repository.Complete(context.Background(), pending.IntegrationID, integrationlog.Completion{ Result: constants.IntegrationResultUnknown, }); err == nil { t.Fatal("结果未知必须记录明确恢复策略") } unknown, err := repository.Complete(context.Background(), pending.IntegrationID, integrationlog.Completion{ Result: constants.IntegrationResultUnknown, RecoveryStrategy: "按原外部单号查询结果,禁止盲目重发", }) if err != nil || unknown.RecoveryStrategy == nil { t.Fatalf("保存结果未知及恢复策略失败:log=%+v err=%v", unknown, err) } for index, terminal := range []string{ constants.IntegrationResultIgnored, constants.IntegrationResultMerged, constants.IntegrationResultRateLimited, constants.IntegrationResultCompleted, constants.IntegrationResultCancelled, } { unsent, err := repository.Start(context.Background(), integrationlog.Attempt{ IntegrationID: fmt.Sprintf("integration-unsent-%d", index), Provider: "gateway", Direction: constants.IntegrationDirectionOutbound, Operation: "query_card_status", InitialResult: terminal, }) if err != nil || unsent.Result != terminal || unsent.HTTPStatus != nil { t.Fatalf("未发送终态不得伪造 HTTP 结果:log=%+v err=%v", unsent, err) } if _, err := repository.Complete(context.Background(), unsent.IntegrationID, integrationlog.Completion{ Result: constants.IntegrationResultSuccess, }); err == nil { t.Fatal("未发送终态不得被后续完成改写") } } if _, err := repository.Start(context.Background(), integrationlog.Attempt{ Provider: "gateway", Direction: constants.IntegrationDirectionOutbound, Operation: "query_card_status", InitialResult: constants.IntegrationResultSuccess, }); err == nil { t.Fatal("实际调用结果不得绕过 pending 直接写终态") } } func TestExplicitFailureAndConcurrentCompletionKeepFirstTerminalFact(t *testing.T) { db := testutil.NewPostgresDatabase(t) integrationIDs := []string{"integration-failed-1", "integration-concurrent-1"} t.Cleanup(func() { if err := db.Where("integration_id IN ?", integrationIDs).Delete(&model.IntegrationLog{}).Error; err != nil { t.Errorf("清理 Integration Log 并发测试数据失败:%v", err) } }) if err := db.Where("integration_id IN ?", integrationIDs).Delete(&model.IntegrationLog{}).Error; err != nil { t.Fatalf("准备 Integration Log 并发测试隔离数据失败:%v", err) } repository := integrationlog.NewRepository(db) failedAttempt, err := repository.Start(context.Background(), integrationlog.Attempt{ IntegrationID: "integration-failed-1", Provider: "gateway", Direction: constants.IntegrationDirectionOutbound, Operation: "query_card_status", }) if err != nil { t.Fatalf("准备明确失败尝试失败:%v", err) } failed, err := repository.Complete(context.Background(), failedAttempt.IntegrationID, integrationlog.Completion{ Result: constants.IntegrationResultFailed, ProviderCode: "UPSTREAM_REJECTED", ProviderMessage: "上游明确拒绝", }) if err != nil || failed.Result != constants.IntegrationResultFailed { t.Fatalf("明确失败未保存为失败终态:log=%+v err=%v", failed, err) } attempt, err := repository.Start(context.Background(), integrationlog.Attempt{ IntegrationID: "integration-concurrent-1", Provider: "gateway", Direction: constants.IntegrationDirectionOutbound, Operation: "query_card_status", }) if err != nil { t.Fatalf("准备并发终结尝试失败:%v", err) } results := make(chan error, 2) var group sync.WaitGroup for _, terminal := range []string{constants.IntegrationResultSuccess, constants.IntegrationResultFailed} { group.Add(1) go func(result string) { defer group.Done() _, completeErr := repository.Complete(context.Background(), attempt.IntegrationID, integrationlog.Completion{Result: result}) results <- completeErr }(terminal) } group.Wait() close(results) successes := 0 for completeErr := range results { if completeErr == nil { successes++ } } if successes != 1 { t.Fatalf("并发终结必须且只能一个成功,实际成功 %d 次", successes) } var saved model.IntegrationLog if err := db.Where("integration_id = ?", attempt.IntegrationID).First(&saved).Error; err != nil { t.Fatalf("读取并发终态失败:%v", err) } if saved.Result != constants.IntegrationResultSuccess && saved.Result != constants.IntegrationResultFailed { t.Fatalf("并发终结未保存明确成功或失败:%+v", saved) } } func createTemporaryIntegrationLogTable(t *testing.T, db *gorm.DB) { t.Helper() if err := db.Exec(`CREATE TEMP TABLE tb_integration_log ( id bigserial PRIMARY KEY, integration_id varchar(64) NOT NULL UNIQUE, idempotency_key varchar(160), provider varchar(32) NOT NULL, direction varchar(16) NOT NULL, operation varchar(64) NOT NULL, external_id varchar(128), resource_type varchar(64), resource_id varchar(128), resource_key varchar(128), trigger_source varchar(32), trigger_scene varchar(128), trigger_series varchar(64), scheduled_at timestamptz, started_at timestamptz, attempt integer NOT NULL DEFAULT 1, result varchar(20) NOT NULL, http_status integer, provider_code varchar(64), provider_message varchar(500), request_summary jsonb, response_summary jsonb, content_hash varchar(64), duration_ms bigint NOT NULL DEFAULT 0, state_changed boolean NOT NULL DEFAULT false, metadata jsonb, recovery_strategy varchar(255), request_id varchar(64), correlation_id varchar(64), audit_event_id bigint, created_at timestamptz NOT NULL DEFAULT NOW(), updated_at timestamptz NOT NULL DEFAULT NOW(), CONSTRAINT uq_test_integration_idempotency UNIQUE (provider, operation, idempotency_key) ) ON COMMIT DROP`).Error; err != nil { t.Fatalf("创建 Integration Log 测试表失败:%v", err) } }