break
ff44305d0e
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 10m19s
285 lines
10 KiB
Go
285 lines
10 KiB
Go
// Package integrationlog 提供外部交互尝试的可靠持久化能力。
|
|
package integrationlog
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"gorm.io/datatypes"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
|
|
"github.com/break/junhong_cmp_fiber/internal/model"
|
|
"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"
|
|
)
|
|
|
|
// Attempt 描述一次外部调用前必须持久化的稳定事实。
|
|
type Attempt struct {
|
|
IntegrationID string
|
|
Provider string
|
|
Direction string
|
|
Operation string
|
|
ExternalID *string
|
|
ResourceType string
|
|
ResourceID *string
|
|
ResourceKey *string
|
|
TriggerSource *string
|
|
TriggerScene *string
|
|
TriggerSeries *string
|
|
ScheduledAt *time.Time
|
|
StartedAt *time.Time
|
|
Attempt int
|
|
RequestSummary any
|
|
Metadata any
|
|
RequestID *string
|
|
CorrelationID *string
|
|
AuditEventID *uint
|
|
InitialResult string
|
|
RecoveryStrategy *string
|
|
}
|
|
|
|
// Completion 描述外部尝试从待处理状态进入终态的结果。
|
|
type Completion struct {
|
|
Result string
|
|
HTTPStatus int
|
|
ProviderCode string
|
|
ProviderMessage string
|
|
ResponseSummary any
|
|
DurationMS int64
|
|
StateChanged bool
|
|
AuditEventID *uint
|
|
RecoveryStrategy string
|
|
}
|
|
|
|
// InboundAttempt 描述业务处理前必须保存的入站回调安全事实。
|
|
type InboundAttempt struct {
|
|
IntegrationID string
|
|
IdempotencyKey string
|
|
Provider string
|
|
Operation string
|
|
ExternalID string
|
|
ResourceType string
|
|
ResourceID *string
|
|
ResourceKey *string
|
|
RawPayload []byte
|
|
ContentType string
|
|
RequestID *string
|
|
CorrelationID *string
|
|
}
|
|
|
|
// Repository 负责创建稳定尝试及受控地进入终态。
|
|
type Repository struct {
|
|
db *gorm.DB
|
|
now func() time.Time
|
|
}
|
|
|
|
// NewRepository 创建 Integration Log Repository。
|
|
func NewRepository(db *gorm.DB) *Repository {
|
|
return &Repository{db: db, now: time.Now}
|
|
}
|
|
|
|
// Start 在实际调用外部系统前持久化尝试事实。
|
|
func (r *Repository) Start(ctx context.Context, input Attempt) (*model.IntegrationLog, error) {
|
|
if r == nil || r.db == nil {
|
|
return nil, pkgerrors.New(pkgerrors.CodeInvalidStatus, "Integration Log 数据库未配置")
|
|
}
|
|
if err := validateAttempt(input); err != nil {
|
|
return nil, err
|
|
}
|
|
requestSummary, err := marshalSummary(input.RequestSummary)
|
|
if err != nil {
|
|
return nil, pkgerrors.Wrap(pkgerrors.CodeInvalidParam, err, "Integration Log 请求摘要无效")
|
|
}
|
|
metadata, err := marshalSummary(input.Metadata)
|
|
if err != nil {
|
|
return nil, pkgerrors.Wrap(pkgerrors.CodeInvalidParam, err, "Integration Log 元数据无效")
|
|
}
|
|
if input.IntegrationID == "" {
|
|
input.IntegrationID = uuid.NewString()
|
|
}
|
|
if input.Attempt <= 0 {
|
|
input.Attempt = 1
|
|
}
|
|
if input.StartedAt == nil {
|
|
startedAt := r.now().UTC()
|
|
input.StartedAt = &startedAt
|
|
}
|
|
result := input.InitialResult
|
|
if result == "" {
|
|
result = constants.IntegrationResultPending
|
|
}
|
|
resourceType := optionalString(input.ResourceType)
|
|
log := &model.IntegrationLog{
|
|
IntegrationID: input.IntegrationID, Provider: input.Provider, Direction: input.Direction,
|
|
Operation: input.Operation, ExternalID: input.ExternalID, ResourceType: resourceType,
|
|
ResourceID: input.ResourceID, ResourceKey: input.ResourceKey, TriggerSource: input.TriggerSource,
|
|
TriggerScene: input.TriggerScene, TriggerSeries: input.TriggerSeries, ScheduledAt: input.ScheduledAt,
|
|
StartedAt: input.StartedAt, Attempt: input.Attempt, Result: result,
|
|
RequestSummary: requestSummary, Metadata: metadata, RequestID: input.RequestID,
|
|
CorrelationID: input.CorrelationID, AuditEventID: input.AuditEventID,
|
|
RecoveryStrategy: input.RecoveryStrategy,
|
|
}
|
|
if err := r.db.WithContext(ctx).Create(log).Error; err != nil {
|
|
return nil, pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "写入 Integration Log 失败")
|
|
}
|
|
return log, nil
|
|
}
|
|
|
|
// Complete 仅允许把待处理尝试条件更新为一个公开终态。
|
|
func (r *Repository) Complete(ctx context.Context, integrationID string, completion Completion) (*model.IntegrationLog, error) {
|
|
if r == nil || r.db == nil {
|
|
return nil, pkgerrors.New(pkgerrors.CodeInvalidStatus, "Integration Log 数据库未配置")
|
|
}
|
|
if integrationID == "" || !isTerminalResult(completion.Result) {
|
|
return nil, pkgerrors.New(pkgerrors.CodeInvalidParam, "Integration Log 终态参数无效")
|
|
}
|
|
if completion.Result == constants.IntegrationResultUnknown && strings.TrimSpace(completion.RecoveryStrategy) == "" {
|
|
return nil, pkgerrors.New(pkgerrors.CodeInvalidParam, "结果未知必须记录明确恢复策略")
|
|
}
|
|
responseSummary, err := marshalSummary(completion.ResponseSummary)
|
|
if err != nil {
|
|
return nil, pkgerrors.Wrap(pkgerrors.CodeInvalidParam, err, "Integration Log 响应摘要无效")
|
|
}
|
|
updates := map[string]any{
|
|
"result": completion.Result, "duration_ms": completion.DurationMS,
|
|
"state_changed": completion.StateChanged, "response_summary": responseSummary,
|
|
"updated_at": r.now().UTC(),
|
|
}
|
|
if completion.HTTPStatus != 0 {
|
|
updates["http_status"] = completion.HTTPStatus
|
|
}
|
|
if completion.ProviderCode != "" {
|
|
updates["provider_code"] = completion.ProviderCode
|
|
}
|
|
if completion.ProviderMessage != "" {
|
|
updates["provider_message"] = sanitizer.TextSummary(completion.ProviderMessage)
|
|
}
|
|
if completion.AuditEventID != nil {
|
|
updates["audit_event_id"] = completion.AuditEventID
|
|
}
|
|
if completion.RecoveryStrategy != "" {
|
|
updates["recovery_strategy"] = completion.RecoveryStrategy
|
|
}
|
|
result := r.db.WithContext(ctx).Model(&model.IntegrationLog{}).
|
|
Where("integration_id = ? AND result = ?", integrationID, constants.IntegrationResultPending).
|
|
Updates(updates)
|
|
if result.Error != nil {
|
|
return nil, pkgerrors.Wrap(pkgerrors.CodeDatabaseError, result.Error, "终结 Integration Log 失败")
|
|
}
|
|
if result.RowsAffected != 1 {
|
|
return nil, pkgerrors.New(pkgerrors.CodeConflict, "Integration Log 已进入终态或不存在")
|
|
}
|
|
var saved model.IntegrationLog
|
|
if err := r.db.WithContext(ctx).Where("integration_id = ?", integrationID).First(&saved).Error; err != nil {
|
|
return nil, pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "读取 Integration Log 终态失败")
|
|
}
|
|
return &saved, nil
|
|
}
|
|
|
|
// RecordInbound 在业务处理前幂等保存入站回调的安全摘要。
|
|
func (r *Repository) RecordInbound(ctx context.Context, input InboundAttempt) (*model.IntegrationLog, bool, error) {
|
|
if r == nil || r.db == nil {
|
|
return nil, false, pkgerrors.New(pkgerrors.CodeInvalidStatus, "Integration Log 数据库未配置")
|
|
}
|
|
if input.Provider == "" || input.Operation == "" || input.IdempotencyKey == "" || len(input.RawPayload) == 0 {
|
|
return nil, false, pkgerrors.New(pkgerrors.CodeInvalidParam, "入站 Integration Log 参数无效")
|
|
}
|
|
if input.IntegrationID == "" {
|
|
input.IntegrationID = uuid.NewString()
|
|
}
|
|
hash := sha256.Sum256(input.RawPayload)
|
|
summary, err := marshalSummary(map[string]any{
|
|
"content_type": input.ContentType,
|
|
"payload_bytes": len(input.RawPayload),
|
|
"content_hash": hex.EncodeToString(hash[:]),
|
|
})
|
|
if err != nil {
|
|
return nil, false, pkgerrors.Wrap(pkgerrors.CodeInvalidParam, err, "入站 Integration Log 摘要无效")
|
|
}
|
|
now := r.now().UTC()
|
|
log := &model.IntegrationLog{
|
|
IntegrationID: input.IntegrationID, IdempotencyKey: &input.IdempotencyKey,
|
|
Provider: input.Provider, Direction: constants.IntegrationDirectionInbound, Operation: input.Operation,
|
|
ExternalID: optionalString(input.ExternalID), ResourceType: optionalString(input.ResourceType),
|
|
ResourceID: input.ResourceID, ResourceKey: input.ResourceKey, StartedAt: &now, Attempt: 1,
|
|
Result: constants.IntegrationResultPending, RequestSummary: summary,
|
|
ContentHash: hex.EncodeToString(hash[:]), RequestID: input.RequestID, CorrelationID: input.CorrelationID,
|
|
}
|
|
result := r.db.WithContext(ctx).Clauses(clause.OnConflict{
|
|
Columns: []clause.Column{{Name: "provider"}, {Name: "operation"}, {Name: "idempotency_key"}},
|
|
DoNothing: true,
|
|
}).Create(log)
|
|
if result.Error != nil {
|
|
return nil, false, pkgerrors.Wrap(pkgerrors.CodeDatabaseError, result.Error, "写入入站 Integration Log 失败")
|
|
}
|
|
if result.RowsAffected == 1 {
|
|
return log, true, nil
|
|
}
|
|
var existing model.IntegrationLog
|
|
if err := r.db.WithContext(ctx).Where(
|
|
"provider = ? AND operation = ? AND idempotency_key = ?", input.Provider, input.Operation, input.IdempotencyKey,
|
|
).First(&existing).Error; err != nil {
|
|
return nil, false, pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "读取重复入站 Integration Log 失败")
|
|
}
|
|
if existing.ContentHash != hex.EncodeToString(hash[:]) {
|
|
return nil, false, pkgerrors.New(pkgerrors.CodeConflict, "入站幂等标识对应的载荷不一致")
|
|
}
|
|
return &existing, false, nil
|
|
}
|
|
|
|
func validateAttempt(input Attempt) error {
|
|
if input.Provider == "" || input.Operation == "" {
|
|
return pkgerrors.New(pkgerrors.CodeInvalidParam, "Integration Log 提供方和操作不能为空")
|
|
}
|
|
if input.Direction != constants.IntegrationDirectionInbound && input.Direction != constants.IntegrationDirectionOutbound {
|
|
return pkgerrors.New(pkgerrors.CodeInvalidParam, "Integration Log 方向无效")
|
|
}
|
|
if input.InitialResult != "" && input.InitialResult != constants.IntegrationResultPending && !isUnsentResult(input.InitialResult) {
|
|
return pkgerrors.New(pkgerrors.CodeInvalidParam, "Integration Log 初始结果只能是待处理或未发送终态")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func isTerminalResult(result string) bool {
|
|
switch result {
|
|
case constants.IntegrationResultSuccess, constants.IntegrationResultFailed, constants.IntegrationResultUnknown,
|
|
constants.IntegrationResultNotFound, constants.IntegrationResultInvalidPayload, constants.IntegrationResultIgnored,
|
|
constants.IntegrationResultMerged, constants.IntegrationResultRateLimited, constants.IntegrationResultCompleted,
|
|
constants.IntegrationResultCancelled:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func isUnsentResult(result string) bool {
|
|
switch result {
|
|
case constants.IntegrationResultIgnored, constants.IntegrationResultMerged, constants.IntegrationResultRateLimited,
|
|
constants.IntegrationResultCompleted, constants.IntegrationResultCancelled:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func marshalSummary(value any) (datatypes.JSON, error) {
|
|
if value == nil {
|
|
return nil, nil
|
|
}
|
|
encoded, err := sanitizer.MarshalSummary(value)
|
|
return datatypes.JSON(encoded), err
|
|
}
|
|
|
|
func optionalString(value string) *string {
|
|
if value == "" {
|
|
return nil
|
|
}
|
|
return &value
|
|
}
|