修复

2026-08-11 15:27:18 +08:00
parent 629609a556
commit 2a7a8fb49d
5 changed files with 16112 additions and 18 deletions
--- a/.scratch/tech-global-audit/审计覆盖清单.json
+++ b/.scratch/tech-global-audit/审计覆盖清单.json
--- a/internal/infrastructure/audit/registry.go
+++ b/internal/infrastructure/audit/registry.go
@@ -187,6 +187,8 @@ func NewRegistry() *Registry {
 		constants.AuditActionOutboxExpiredLeaseReleased,
 		"人工释放 Outbox 过期租约",
 	)
+	integrationAttemptStarted := integrationAction(constants.AuditActionIntegrationAttemptStarted, "记录外部交互开始")
+	integrationInboundReceived := integrationAction(constants.AuditActionIntegrationInboundReceived, "记录外部入站回调")
 	iotCardImportTaskCreated := taskAction(constants.AuditActionIotCardImportTaskCreated, "创建 IoT 卡导入任务", constants.AuditResourceIotCardImportTask, constants.AuditActorAccount, constants.AuditSourceAdminAPI)
 	iotCardImportTaskCompleted := taskAction(constants.AuditActionIotCardImportTaskCompleted, "完成 IoT 卡导入任务", constants.AuditResourceIotCardImportTask, constants.AuditActorSystemTask, constants.AuditSourceWorker)
 	deviceImportTaskCreated := taskAction(constants.AuditActionDeviceImportTaskCreated, "创建设备导入任务", constants.AuditResourceDeviceImportTask, constants.AuditActorAccount, constants.AuditSourceAdminAPI)
@@ -563,6 +565,8 @@ func NewRegistry() *Registry {
 			constants.AuditActionPaymentCreated:                        paymentCreated,
 			constants.AuditActionPaymentConfirmed:                      paymentConfirmed,
 			constants.AuditActionPaymentFailed:                         paymentFailed,
+			constants.AuditActionIntegrationAttemptStarted:             integrationAttemptStarted,
+			constants.AuditActionIntegrationInboundReceived:            integrationInboundReceived,
 			constants.AuditActionAgentRechargeCreated:                  agentRechargeCreated,
 			constants.AuditActionAgentRechargeCredited:                 agentRechargeCredited,
 			constants.AuditActionAgentRechargeClosed:                   agentRechargeClosed,
@@ -1196,6 +1200,23 @@ func outboxRecoveryAction(code, name string) ActionDefinition {
 	}
 }

+func integrationAction(code, name string) ActionDefinition {
+	return ActionDefinition{
+		Code: code, Name: name, Category: constants.AuditCategoryReliability, Risk: constants.AuditRiskNormal,
+		PrimaryResource: constants.AuditResourceIntegrationLog, AllowedActor: constants.AuditActorSystemTask,
+		Source: constants.AuditSourceWorker, RequireTransaction: false,
+		DefaultVisibility: constants.AuditSubjectInternalOnly,
+		AllowedVisibility: []string{constants.AuditSubjectInternalOnly},
+		AllowedOrigins: []ActionOrigin{
+			{Actor: constants.AuditActorAccount, Source: constants.AuditSourceAdminAPI},
+			{Actor: constants.AuditActorPersonalCustomer, Source: constants.AuditSourcePersonalAPI},
+			{Actor: constants.AuditActorOpenAPI, Source: constants.AuditSourceOpenAPI},
+			{Actor: constants.AuditActorExternalSystem, Source: constants.AuditSourceCallback},
+			{Actor: constants.AuditActorScheduledJob, Source: constants.AuditSourceScheduler},
+		},
+	}
+}
+
 func connectionConfigAction(code, name, resourceType, risk string) ActionDefinition {
 	return ActionDefinition{
 		Code: code, Name: name, Category: constants.AuditCategoryConfiguration, Risk: risk,
--- a/internal/infrastructure/audit/writer.go
+++ b/internal/infrastructure/audit/writer.go
@@ -872,27 +872,33 @@ func (w *Writer) WriteRecovery(ctx context.Context, tx *gorm.DB, recovery outbox

 // Append 在调用方提供的 GORM 事务中顺序追加事件及资源。
 func (w *Writer) Append(ctx context.Context, tx *gorm.DB, input AppendInput) error {
+	_, err := w.AppendAndGet(ctx, tx, input)
+	return err
+}
+
+// AppendAndGet 追加事件并返回已持久化的审计事件，幂等重放返回已有事件。
+func (w *Writer) AppendAndGet(ctx context.Context, tx *gorm.DB, input AppendInput) (*model.AuditEvent, error) {
 	if w == nil || w.registry == nil || tx == nil {
-		return pkgerrors.New(pkgerrors.CodeInvalidStatus, "统一审计 Writer 未正确配置")
+		return nil, pkgerrors.New(pkgerrors.CodeInvalidStatus, "统一审计 Writer 未正确配置")
 	}
 	input = fillFromContext(ctx, input)
 	action, ok := w.registry.Action(input.ActionCode)
 	if !ok {
-		return pkgerrors.New(pkgerrors.CodeInvalidParam, "审计动作未注册")
+		return nil, pkgerrors.New(pkgerrors.CodeInvalidParam, "审计动作未注册")
 	}
 	if !actionAllowsOrigin(action, input.Actor.Kind, input.Source) || input.Actor.ID == "" {
-		return pkgerrors.New(pkgerrors.CodeInvalidParam, "审计操作者或入口不符合动作注册规则")
+		return nil, pkgerrors.New(pkgerrors.CodeInvalidParam, "审计操作者或入口不符合动作注册规则")
 	}
 	if !validResult(input.Result) || len(input.Resources) == 0 {
-		return pkgerrors.New(pkgerrors.CodeInvalidParam, "审计结果非法或缺少资源")
+		return nil, pkgerrors.New(pkgerrors.CodeInvalidParam, "审计结果非法或缺少资源")
 	}
 	resources, err := w.buildResources(input.Resources, action)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	metadata, err := safeObject(input.Metadata)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	occurredAt := input.OccurredAt
 	if occurredAt.IsZero() {
@@ -913,7 +919,7 @@ func (w *Writer) Append(ctx context.Context, tx *gorm.DB, input AppendInput) err
 	}
 	event.ContentHash, err = contentHash(event, resources)
 	if err != nil {
-		return pkgerrors.Wrap(pkgerrors.CodeInvalidParam, err, "计算审计内容哈希失败")
+		return nil, pkgerrors.Wrap(pkgerrors.CodeInvalidParam, err, "计算审计内容哈希失败")
 	}
 	event.EventID = input.EventID
 	if event.EventID == "" {
@@ -923,19 +929,23 @@ func (w *Writer) Append(ctx context.Context, tx *gorm.DB, input AppendInput) err
 		Columns: []clause.Column{{Name: "event_id"}}, DoNothing: true,
 	}).Create(&event)
 	if create.Error != nil {
-		return pkgerrors.Wrap(pkgerrors.CodeDatabaseError, create.Error, "写入审计事件失败")
+		return nil, pkgerrors.Wrap(pkgerrors.CodeDatabaseError, create.Error, "写入审计事件失败")
 	}
 	if create.RowsAffected == 0 {
-		return nil
+		var existing model.AuditEvent
+		if err := tx.WithContext(ctx).Where("event_id = ?", event.EventID).First(&existing).Error; err != nil {
+			return nil, pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "读取已存在审计事件失败")
+		}
+		return &existing, nil
 	}
 	for index := range resources {
 		resources[index].AuditEventID = event.ID
 		resources[index].CreatedAt = event.CreatedAt
 	}
 	if err := tx.WithContext(ctx).Create(&resources).Error; err != nil {
-		return pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "写入审计事件资源失败")
+		return nil, pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "写入审计事件资源失败")
 	}
-	return nil
+	return &event, nil
 }

 func actionAllowsOrigin(action ActionDefinition, actor, source string) bool {
--- a/internal/infrastructure/integrationlog/repository.go
+++ b/internal/infrastructure/integrationlog/repository.go
@@ -10,11 +10,14 @@ import (
 	"unicode/utf8"

 	"github.com/google/uuid"
+	"go.uber.org/zap"
 	"gorm.io/datatypes"
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"

+	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
 	"github.com/break/junhong_cmp_fiber/internal/model"
+	"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
 	"github.com/break/junhong_cmp_fiber/pkg/constants"
 	pkgerrors "github.com/break/junhong_cmp_fiber/pkg/errors"
 	"github.com/break/junhong_cmp_fiber/pkg/sanitizer"
@@ -57,7 +60,6 @@ type Completion struct {
 	StateChanged        bool
 	ResourceID          *string
 	ResourceKey         *string
-	AuditEventID        *uint
 	RecoveryStrategy    string
 }

@@ -75,17 +77,19 @@ type InboundAttempt struct {
 	ContentType    string
 	RequestID      *string
 	CorrelationID  *string
+	AuditEventID   *uint
 }

 // Repository 负责创建稳定尝试及受控地进入终态。
 type Repository struct {
-	db  *gorm.DB
-	now func() time.Time
+	db    *gorm.DB
+	now   func() time.Time
+	audit *audit.Writer
 }

 // NewRepository 创建 Integration Log Repository。
 func NewRepository(db *gorm.DB) *Repository {
-	return &Repository{db: db, now: time.Now}
+	return &Repository{db: db, now: time.Now, audit: audit.NewWriter(audit.NewRegistry(), nil)}
 }

 // Start 在实际调用外部系统前持久化尝试事实。
@@ -107,6 +111,12 @@ func (r *Repository) Start(ctx context.Context, input Attempt) (*model.Integrati
 	if input.IntegrationID == "" {
 		input.IntegrationID = uuid.NewString()
 	}
+	if input.AuditEventID == nil && input.TriggerSeries != nil {
+		input.AuditEventID = r.auditEventIDForSeries(ctx, *input.TriggerSeries)
+	}
+	if input.AuditEventID == nil {
+		input.AuditEventID = r.recordAuditEvent(ctx, constants.AuditActionIntegrationAttemptStarted, input.IntegrationID, input.Provider, input.Direction, input.Operation, input.ResourceType, input.ResourceID, input.ResourceKey, input.CorrelationID)
+	}
 	autoAttempt := input.Attempt <= 0 && input.TriggerSeries != nil
 	if input.Attempt <= 0 {
 		input.Attempt = 1
@@ -193,9 +203,6 @@ func (r *Repository) Complete(ctx context.Context, integrationID string, complet
 	} else if completion.ProviderMessage != "" {
 		updates["provider_message"] = sanitizer.TextSummary(completion.ProviderMessage)
 	}
-	if completion.AuditEventID != nil {
-		updates["audit_event_id"] = completion.AuditEventID
-	}
 	if completion.ResourceID != nil {
 		updates["resource_id"] = completion.ResourceID
 	}
@@ -236,6 +243,9 @@ func (r *Repository) RecordInbound(ctx context.Context, input InboundAttempt) (*
 	if input.IntegrationID == "" {
 		input.IntegrationID = uuid.NewString()
 	}
+	if input.AuditEventID == nil {
+		input.AuditEventID = r.recordAuditEvent(ctx, constants.AuditActionIntegrationInboundReceived, input.IntegrationID, input.Provider, constants.IntegrationDirectionInbound, input.Operation, input.ResourceType, input.ResourceID, input.ResourceKey, input.CorrelationID)
+	}
 	triggerSeries := input.IntegrationID
 	hash := sha256.Sum256(input.RawPayload)
 	summary, err := marshalSummary(map[string]any{
@@ -255,6 +265,7 @@ func (r *Repository) RecordInbound(ctx context.Context, input InboundAttempt) (*
 		TriggerSeries: &triggerSeries,
 		Result:        constants.IntegrationResultPending, RequestSummary: summary,
 		ContentHash: hex.EncodeToString(hash[:]), RequestID: input.RequestID, CorrelationID: input.CorrelationID,
+		AuditEventID: input.AuditEventID,
 	}
 	result := r.db.WithContext(ctx).Clauses(clause.OnConflict{
 		Columns:   []clause.Column{{Name: "provider"}, {Name: "operation"}, {Name: "idempotency_key"}},
@@ -278,6 +289,80 @@ func (r *Repository) RecordInbound(ctx context.Context, input InboundAttempt) (*
 	return &existing, false, nil
 }

+// recordAuditEvent 为新的外部交互建立稳定审计关联；审计失败不丢失外部事实。
+func (r *Repository) recordAuditEvent(ctx context.Context, actionCode, integrationID, provider, direction, operation, resourceType string, resourceID, resourceKey, correlationID *string) *uint {
+	if r == nil || r.db == nil || r.audit == nil {
+		zap.L().Warn("Integration Log 缺少审计关联", zap.String("integration_id", integrationID), zap.String("reason", "审计 Writer 未配置"))
+		return nil
+	}
+	value := auditcontext.From(ctx)
+	if !validAuditOrigin(value.ActorKind, value.ActorID, value.Source) {
+		value.ActorKind = constants.AuditActorSystemTask
+		value.ActorID = "integration_log"
+		value.Source = constants.AuditSourceWorker
+	}
+	event, err := r.audit.AppendAndGet(ctx, r.db, audit.AppendInput{
+		EventID:    "integration:" + integrationID,
+		ActionCode: actionCode, Summary: "记录外部交互审计关联",
+		Actor:  audit.ActorInput{Kind: value.ActorKind, ID: value.ActorID, Name: value.ActorName, ShopID: value.ActorShopID, EnterpriseID: value.ActorEnterpriseID},
+		Source: value.Source, ScopeType: constants.AuditScopePlatform,
+		Result: constants.AuditResultSuccess, RequestID: textValue(correlationID), CorrelationID: textValue(correlationID),
+		Resources: []audit.ResourceInput{{
+			Type: constants.AuditResourceIntegrationLog, Key: integrationID, DisplayName: operation,
+			Relation: constants.AuditResourceRelationPrimary, Role: constants.AuditResourceRoleWorkerIntegration,
+			IdentitySnapshot: map[string]any{
+				"integration_id": integrationID, "provider": provider, "direction": direction, "operation": operation,
+				"resource_type": resourceType, "resource_id": textValue(resourceID), "resource_key": textValue(resourceKey), "correlation_id": textValue(correlationID),
+			},
+		}},
+	})
+	if err != nil {
+		zap.L().Warn("Integration Log 缺少审计关联", zap.String("integration_id", integrationID), zap.Error(err))
+		return nil
+	}
+	return &event.ID
+}
+
+func (r *Repository) auditEventIDForSeries(ctx context.Context, triggerSeries string) *uint {
+	if r == nil || r.db == nil || triggerSeries == "" {
+		return nil
+	}
+	var log model.IntegrationLog
+	if err := r.db.WithContext(ctx).Where("trigger_series = ? AND audit_event_id IS NOT NULL", triggerSeries).Order("attempt DESC").First(&log).Error; err != nil {
+		return nil
+	}
+	return log.AuditEventID
+}
+
+func validAuditOrigin(actorKind, actorID, source string) bool {
+	if actorID == "" {
+		return false
+	}
+	switch source {
+	case constants.AuditSourceAdminAPI:
+		return actorKind == constants.AuditActorAccount
+	case constants.AuditSourcePersonalAPI:
+		return actorKind == constants.AuditActorPersonalCustomer
+	case constants.AuditSourceOpenAPI:
+		return actorKind == constants.AuditActorOpenAPI
+	case constants.AuditSourceCallback:
+		return actorKind == constants.AuditActorExternalSystem
+	case constants.AuditSourceScheduler:
+		return actorKind == constants.AuditActorScheduledJob
+	case constants.AuditSourceWorker:
+		return actorKind == constants.AuditActorSystemTask
+	default:
+		return false
+	}
+}
+
+func textValue(value *string) string {
+	if value == nil {
+		return ""
+	}
+	return *value
+}
+
 // ClaimExpiredInboundPending 原子认领已超过处理租约的入站 pending 记录。
 func (r *Repository) ClaimExpiredInboundPending(ctx context.Context, integrationID string, lease time.Duration) (bool, error) {
 	if r == nil || r.db == nil {
--- a/pkg/constants/audit.go
+++ b/pkg/constants/audit.go
@@ -413,6 +413,10 @@ const (
 	AuditActionOutboxReplayed = "outbox.replayed"
 	// AuditActionOutboxExpiredLeaseReleased 表示人工释放 Outbox 过期租约。
 	AuditActionOutboxExpiredLeaseReleased = "outbox.expired_lease_released"
+	// AuditActionIntegrationAttemptStarted 表示外部交互开始前已记录的审计事实。
+	AuditActionIntegrationAttemptStarted = "integration.attempt_started"
+	// AuditActionIntegrationInboundReceived 表示入站回调处理前已记录的审计事实。
+	AuditActionIntegrationInboundReceived = "integration.inbound_received"
 	// AuditActionIotCardImportTaskCreated 表示创建 IoT 卡导入任务。
 	AuditActionIotCardImportTaskCreated = "iot_card_import_task.create"
 	// AuditActionIotCardImportTaskCompleted 表示 IoT 卡导入任务完成。