break
5e552d99bc
Constraint: 在线热修前必须保存当前迭代分支全部有效代码进展 Confidence: medium Scope-risk: broad Directive: 后续修改需保持审计事件与业务事务边界一致 Tested: git diff --cached --check Not-tested: 未运行全量测试，提交用于切换分支前保存既有工作
295 lines
11 KiB
Go
295 lines
11 KiB
Go
package device
|
|
|
|
import (
|
|
"context"
|
|
stderrors "errors"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"gorm.io/gorm"
|
|
|
|
"github.com/break/junhong_cmp_fiber/internal/gateway"
|
|
"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
|
"github.com/break/junhong_cmp_fiber/internal/infrastructure/integrationlog"
|
|
"github.com/break/junhong_cmp_fiber/internal/model"
|
|
assetAuditSvc "github.com/break/junhong_cmp_fiber/internal/service/asset_audit"
|
|
"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
|
|
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
|
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
|
)
|
|
|
|
type deviceGatewayIntegrationLog interface {
|
|
Start(ctx context.Context, input integrationlog.Attempt) (*model.IntegrationLog, error)
|
|
Complete(ctx context.Context, integrationID string, completion integrationlog.Completion) (*model.IntegrationLog, error)
|
|
}
|
|
|
|
// SetGatewayIntegrationLog 注入设备外部命令的 Integration Log 接缝。
|
|
func (s *Service) SetGatewayIntegrationLog(integration deviceGatewayIntegrationLog) {
|
|
s.gatewayIntegration = integration
|
|
}
|
|
|
|
type deviceGatewayResource struct {
|
|
Type string
|
|
ID string
|
|
Key string
|
|
ExternalID string
|
|
RequestSummary map[string]any
|
|
}
|
|
|
|
type deviceGatewayAttempt struct {
|
|
log *model.IntegrationLog
|
|
startedAt time.Time
|
|
}
|
|
|
|
type deviceGatewayAttemptObserver struct {
|
|
service *Service
|
|
operation string
|
|
scene string
|
|
seriesKey string
|
|
resource deviceGatewayResource
|
|
current *deviceGatewayAttempt
|
|
successful *deviceGatewayAttempt
|
|
integration string
|
|
unknown bool
|
|
}
|
|
|
|
func (o *deviceGatewayAttemptObserver) BeforeAttempt(ctx context.Context, attempt int) error {
|
|
started, err := o.service.startDeviceGatewayAttempt(ctx, o.operation, o.scene, o.seriesKey, attempt, o.resource)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
o.current = started
|
|
o.integration = started.log.IntegrationID
|
|
return nil
|
|
}
|
|
|
|
func (o *deviceGatewayAttemptObserver) AfterAttempt(ctx context.Context, _ int, callErr error) error {
|
|
if isDeviceGatewayTimeout(callErr) {
|
|
o.unknown = true
|
|
}
|
|
if callErr == nil {
|
|
o.successful = o.current
|
|
o.current = nil
|
|
return nil
|
|
}
|
|
err := o.service.completeDeviceGatewayAttempt(ctx, o.current, callErr, false)
|
|
o.current = nil
|
|
return err
|
|
}
|
|
|
|
func (o *deviceGatewayAttemptObserver) completeSuccess(ctx context.Context, stateChanged bool) error {
|
|
return o.service.completeDeviceGatewayAttempt(ctx, o.successful, nil, stateChanged)
|
|
}
|
|
|
|
func (s *Service) startDeviceGatewayAttempt(
|
|
ctx context.Context,
|
|
operation, scene, seriesKey string,
|
|
attempt int,
|
|
resource deviceGatewayResource,
|
|
) (*deviceGatewayAttempt, error) {
|
|
if s == nil || s.gatewayIntegration == nil {
|
|
return nil, errors.New(errors.CodeInvalidStatus, "设备 Gateway Integration Log 接缝未配置")
|
|
}
|
|
linkage := auditcontext.From(ctx)
|
|
triggerSource := linkage.Source
|
|
if triggerSource == "" {
|
|
triggerSource = "service"
|
|
}
|
|
triggerSeries := uuid.NewSHA1(uuid.NameSpaceOID, []byte("gateway-device-command:"+seriesKey+":"+operation+":"+resource.Type+":"+resource.ID)).String()
|
|
var requestID, correlationID *string
|
|
if linkage.RequestID != "" {
|
|
requestID = &linkage.RequestID
|
|
}
|
|
if linkage.CorrelationID != "" {
|
|
correlationID = &linkage.CorrelationID
|
|
} else {
|
|
correlationID = requestID
|
|
}
|
|
log, err := s.gatewayIntegration.Start(ctx, integrationlog.Attempt{
|
|
Provider: constants.IntegrationProviderGateway, Direction: constants.IntegrationDirectionOutbound,
|
|
Operation: operation, ExternalID: &resource.ExternalID,
|
|
ResourceType: resource.Type, ResourceID: &resource.ID, ResourceKey: &resource.Key,
|
|
TriggerSource: &triggerSource, TriggerScene: &scene, TriggerSeries: &triggerSeries,
|
|
Attempt: attempt, RequestID: requestID, CorrelationID: correlationID,
|
|
RequestSummary: resource.RequestSummary,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &deviceGatewayAttempt{log: log, startedAt: time.Now()}, nil
|
|
}
|
|
|
|
func (s *Service) completeDeviceGatewayAttempt(ctx context.Context, attempt *deviceGatewayAttempt, callErr error, stateChanged bool) error {
|
|
if attempt == nil || attempt.log == nil {
|
|
return nil
|
|
}
|
|
completion := integrationlog.Completion{
|
|
Result: constants.IntegrationResultSuccess, DurationMS: time.Since(attempt.startedAt).Milliseconds(),
|
|
StateChanged: stateChanged, ResponseSummary: map[string]any{"result": "success"},
|
|
}
|
|
if callErr != nil {
|
|
completion.Result = constants.IntegrationResultFailed
|
|
completion.SafeProviderMessage = "Gateway 设备命令失败"
|
|
completion.ResponseSummary = map[string]any{"result": "failed"}
|
|
if isDeviceGatewayTimeout(callErr) {
|
|
completion.Result = constants.IntegrationResultUnknown
|
|
completion.SafeProviderMessage = "Gateway 设备命令结果未知"
|
|
completion.ResponseSummary = map[string]any{"result": "unknown"}
|
|
completion.RecoveryStrategy = constants.GatewayDeviceCommandUnknownRecoveryStrategy
|
|
}
|
|
}
|
|
_, err := s.gatewayIntegration.Complete(ctx, attempt.log.IntegrationID, completion)
|
|
return err
|
|
}
|
|
|
|
type deviceGatewayCommand struct {
|
|
ActionCode string
|
|
Summary string
|
|
Operation string
|
|
Scene string
|
|
RequestSummary map[string]any
|
|
Metadata map[string]any
|
|
TargetCard *model.IotCard
|
|
Call func(context.Context) error
|
|
}
|
|
|
|
func (s *Service) executeDeviceGatewayCommand(ctx context.Context, device *model.Device, command deviceGatewayCommand) error {
|
|
deviceID := strconv.FormatUint(uint64(device.ID), 10)
|
|
seriesKey := deviceCommandSeriesKey(ctx)
|
|
observer := &deviceGatewayAttemptObserver{
|
|
service: s, operation: command.Operation, scene: command.Scene, seriesKey: seriesKey,
|
|
resource: deviceGatewayResource{
|
|
Type: constants.AuditResourceDevice, ID: deviceID, Key: audit.DeviceResourceKey(device),
|
|
ExternalID: device.IMEI, RequestSummary: command.RequestSummary,
|
|
},
|
|
}
|
|
callErr := command.Call(gateway.WithAttemptObserver(ctx, observer))
|
|
metadata := cloneDeviceCommandMetadata(command.Metadata)
|
|
metadata["integration_id"] = observer.integration
|
|
if callErr != nil {
|
|
result := constants.AuditResultFailed
|
|
summary := command.Summary + "失败"
|
|
if observer.unknown {
|
|
result = constants.AuditResultUnknown
|
|
summary = command.Summary + "结果未知"
|
|
}
|
|
s.recordDeviceCommandAudit(ctx, command.ActionCode, summary, result, device, command.TargetCard, nil, nil, metadata, callErr)
|
|
return callErr
|
|
}
|
|
if err := observer.completeSuccess(ctx, false); err != nil {
|
|
s.recordDeviceCommandAudit(ctx, command.ActionCode, command.Summary+"结果未知", constants.AuditResultUnknown,
|
|
device, command.TargetCard, nil, nil, metadata, err)
|
|
return errors.Wrap(errors.CodeDatabaseError, err, "终结设备 Gateway Integration Log 失败")
|
|
}
|
|
s.recordDeviceCommandAudit(ctx, command.ActionCode, command.Summary, constants.AuditResultSuccess,
|
|
device, command.TargetCard, nil, nil, metadata, nil)
|
|
return nil
|
|
}
|
|
|
|
func cloneDeviceCommandMetadata(source map[string]any) map[string]any {
|
|
result := make(map[string]any, len(source)+1)
|
|
for key, value := range source {
|
|
result[key] = value
|
|
}
|
|
return result
|
|
}
|
|
|
|
func deviceCommandSeriesKey(ctx context.Context) string {
|
|
linkage := auditcontext.From(ctx)
|
|
if linkage.CorrelationID != "" {
|
|
return linkage.CorrelationID
|
|
}
|
|
if linkage.RequestID != "" {
|
|
return linkage.RequestID
|
|
}
|
|
return uuid.NewString()
|
|
}
|
|
|
|
func isDeviceGatewayTimeout(err error) bool {
|
|
var appErr *errors.AppError
|
|
return stderrors.As(err, &appErr) && appErr != nil && appErr.Code == errors.CodeGatewayTimeout
|
|
}
|
|
|
|
func (s *Service) appendDeviceCommandAudit(
|
|
ctx context.Context,
|
|
tx *gorm.DB,
|
|
actionCode, summary, result string,
|
|
device *model.Device,
|
|
targetCard *model.IotCard,
|
|
cardBefore, cardAfter, metadata map[string]any,
|
|
businessErr error,
|
|
) error {
|
|
if s.auditWriter == nil || device == nil || device.ID == 0 {
|
|
return errors.New(errors.CodeInvalidStatus, "设备命令统一审计接缝未配置或资源不完整")
|
|
}
|
|
cardReferences, _, err := loadDeviceCardAuditReferences(ctx, tx, []*model.Device{device}, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if targetCard != nil && targetCard.ID > 0 {
|
|
targetID := strconv.FormatUint(uint64(targetCard.ID), 10)
|
|
found := false
|
|
for i := range cardReferences[device.ID] {
|
|
resource := &cardReferences[device.ID][i]
|
|
if resource.Type == constants.AuditResourceIotCard && resource.ID != nil && *resource.ID == targetID {
|
|
found = true
|
|
if cardBefore != nil || cardAfter != nil {
|
|
resource.Relation = constants.AuditResourceRelationAffected
|
|
resource.BeforeData = cardBefore
|
|
resource.AfterData = cardAfter
|
|
resource.SubjectVisibility = constants.AuditSubjectResult
|
|
resource.SubjectSummary = summary
|
|
} else {
|
|
resource.Role = constants.AuditResourceRoleDeviceCommandTargetCard
|
|
}
|
|
}
|
|
}
|
|
if !found {
|
|
cardReferences[device.ID] = append(cardReferences[device.ID], audit.ResourceInput{
|
|
Type: constants.AuditResourceIotCard, ID: &targetID,
|
|
Key: audit.IotCardResourceKey(targetCard), DisplayName: targetCard.ICCID,
|
|
Relation: constants.AuditResourceRelationReference, Role: constants.AuditResourceRoleDeviceCommandTargetCard,
|
|
IdentitySnapshot: audit.IotCardIdentitySnapshot(targetCard),
|
|
SubjectVisibility: constants.AuditSubjectInternalOnly,
|
|
})
|
|
}
|
|
}
|
|
deviceID := strconv.FormatUint(uint64(device.ID), 10)
|
|
resources := []audit.ResourceInput{{
|
|
Type: constants.AuditResourceDevice, ID: &deviceID,
|
|
Key: audit.DeviceResourceKey(device), DisplayName: device.VirtualNo,
|
|
Relation: constants.AuditResourceRelationPrimary, Role: constants.AuditResourceRoleDeviceTarget,
|
|
IdentitySnapshot: audit.DeviceIdentitySnapshot(device),
|
|
SubjectVisibility: constants.AuditSubjectResult, SubjectSummary: summary,
|
|
}}
|
|
resources = append(resources, cardReferences[device.ID]...)
|
|
errorCode, errorSummary := assetAuditSvc.BuildErrorInfo(businessErr)
|
|
return s.auditWriter.Append(ctx, tx, audit.AppendInput{
|
|
ActionCode: actionCode, Summary: summary, ScopeType: constants.AuditScopePlatform,
|
|
Result: result, ErrorCode: errorCode, ErrorSummary: errorSummary,
|
|
Metadata: metadata, Resources: resources,
|
|
})
|
|
}
|
|
|
|
func (s *Service) recordDeviceCommandAudit(
|
|
ctx context.Context,
|
|
actionCode, summary, result string,
|
|
device *model.Device,
|
|
targetCard *model.IotCard,
|
|
cardBefore, cardAfter, metadata map[string]any,
|
|
businessErr error,
|
|
) {
|
|
if s.db == nil || s.auditWriter == nil || device == nil || device.ID == 0 {
|
|
recordDeviceAuditSecondaryFailure(ctx, actionCode, 0, businessErr, errors.New(errors.CodeInvalidStatus, "设备命令统一审计接缝未配置或资源不完整"))
|
|
return
|
|
}
|
|
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
return s.appendDeviceCommandAudit(ctx, tx, actionCode, summary, result, device, targetCard, cardBefore, cardAfter, metadata, businessErr)
|
|
}); err != nil {
|
|
recordDeviceAuditSecondaryFailure(ctx, actionCode, device.ID, businessErr, err)
|
|
}
|
|
}
|
|
|
|
var _ deviceGatewayIntegrationLog = (*integrationlog.Repository)(nil)
|