junhong_cmp_fiber/internal/application/cardobservation/traffic.go

package cardobservation

import (
	"context"
	"strconv"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	domain "github.com/break/junhong_cmp_fiber/internal/domain/cardobservation"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
	"github.com/break/junhong_cmp_fiber/pkg/outboxid"
)

// TrafficIncrementedEvent 是卡流量正增量的可靠领域事实。
type TrafficIncrementedEvent struct {
	EventID       string    `json:"event_id"`
	CardID        uint      `json:"card_id"`
	IncrementMB   float64   `json:"increment_mb"`
	ObservedAt    time.Time `json:"observed_at"`
	Source        string    `json:"source"`
	Scene         string    `json:"scene"`
	RequestID     string    `json:"request_id,omitempty"`
	CorrelationID string    `json:"correlation_id,omitempty"`
}

// ApplyTrafficObservation 串行应用流量读数，并在正增量时同事务写 Outbox。
func (s *Service) ApplyTrafficObservation(ctx context.Context, observation domain.TrafficObservation) (domain.TrafficDecision, error) {
	if s == nil || s.db == nil || s.eventWriter == nil {
		return domain.TrafficDecision{}, errors.New(errors.CodeInternalError, "卡流量观测能力未完整配置")
	}
	var decision domain.TrafficDecision
	var auditedCard *model.IotCard
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var card model.IotCard
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", observation.CardID).First(&card).Error; err != nil {
			if err == gorm.ErrRecordNotFound {
				return errors.New(errors.CodeNotFound, "IoT卡不存在")
			}
			return errors.Wrap(errors.CodeDatabaseError, err, "锁定IoT卡流量事实失败")
		}
		auditedCard = &card
		nextDecision, decisionErr := domain.ApplyTraffic(domain.CardTrafficSnapshot{
			CardID: card.ID, DataUsageMB: card.DataUsageMB, CurrentMonthUsageMB: card.CurrentMonthUsageMB,
			CurrentMonthStartDate: card.CurrentMonthStartDate, LastMonthTotalMB: card.LastMonthTotalMB,
			LastGatewayReadingMB: card.LastGatewayReadingMB,
		}, observation)
		if decisionErr != nil {
			return decisionErr
		}
		decision = nextDecision
		updates := map[string]any{
			"last_data_check_at": observation.Metadata.ObservedAt, "last_sync_time": observation.Metadata.ObservedAt,
			"current_month_start_date": decision.CurrentMonthStartDate, "last_month_total_mb": decision.LastMonthTotalMB,
			"current_month_usage_mb": decision.CurrentMonthUsageMB, "data_usage_mb": decision.DataUsageMB,
		}
		if decision.ReadingAccepted {
			updates["last_gateway_reading_mb"] = decision.LastGatewayReadingMB
		}
		result := tx.Model(&model.IotCard{}).
			Where("id = ? AND last_gateway_reading_mb = ?", card.ID, card.LastGatewayReadingMB).
			Updates(updates)
		if result.Error != nil {
			return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新卡流量事实失败")
		}
		if result.RowsAffected != 1 {
			return errors.New(errors.CodeConflict, "卡流量基线已被其他请求更新")
		}
		if decision.IncrementMB > 0 {
			eventID := outboxid.Stable("card-traffic:", strconv.FormatUint(uint64(card.ID), 10)+":"+observation.Metadata.ObservationID+":incremented")
			if err := s.eventWriter.AppendTraffic(ctx, tx, TrafficIncrementedEvent{
				EventID: eventID, CardID: card.ID, IncrementMB: decision.IncrementMB,
				ObservedAt: observation.Metadata.ObservedAt, Source: observation.Metadata.Source,
				Scene: observation.Metadata.Scene, RequestID: observation.Metadata.RequestID,
				CorrelationID: observation.Metadata.CorrelationID,
			}); err != nil {
				return err
			}
		}
		stateChanged := decision.IncrementMB != 0 || decision.CrossMonth || decision.LastGatewayReadingMB != card.LastGatewayReadingMB
		if actionCode, audited := manualRefreshAuditAction(ctx); observation.Metadata.Source == constants.CardObservationSourceManualSync && stateChanged && audited {
			if s.auditWriter == nil {
				return errors.New(errors.CodeInternalError, "卡状态统一审计能力未配置")
			}
			if err := s.auditWriter.WriteCardStateAudit(ctx, tx, StateAudit{
				ActionCode: actionCode,
				Summary:    "人工刷新 IoT 卡流量",
				Card:       &card,
				BeforeData: map[string]any{
					"data_usage_mb": card.DataUsageMB, "current_month_usage_mb": card.CurrentMonthUsageMB,
					"current_month_start_date": card.CurrentMonthStartDate, "last_month_total_mb": card.LastMonthTotalMB,
					"last_gateway_reading_mb": card.LastGatewayReadingMB,
				},
				AfterData: map[string]any{
					"data_usage_mb": decision.DataUsageMB, "current_month_usage_mb": decision.CurrentMonthUsageMB,
					"current_month_start_date": decision.CurrentMonthStartDate, "last_month_total_mb": decision.LastMonthTotalMB,
					"last_gateway_reading_mb": decision.LastGatewayReadingMB, "increment_mb": decision.IncrementMB,
				},
			}); err != nil {
				return err
			}
		} else if workerObservationAudited(ctx) && stateChanged {
			if s.auditWriter == nil {
				return errors.New(errors.CodeInternalError, "卡状态统一审计能力未配置")
			}
			if err := s.auditWriter.WriteCardStateAudit(ctx, tx, StateAudit{
				ActionCode: constants.AuditActionIotCardWorkerTrafficSynced,
				Summary:    "Worker 同步 IoT 卡流量事实", Card: &card,
				IntegrationID: observation.Metadata.ObservationID,
				BeforeData: map[string]any{
					"data_usage_mb": card.DataUsageMB, "current_month_usage_mb": card.CurrentMonthUsageMB,
					"current_month_start_date": card.CurrentMonthStartDate, "last_month_total_mb": card.LastMonthTotalMB,
					"last_gateway_reading_mb": card.LastGatewayReadingMB,
				},
				AfterData: map[string]any{
					"data_usage_mb": decision.DataUsageMB, "current_month_usage_mb": decision.CurrentMonthUsageMB,
					"current_month_start_date": decision.CurrentMonthStartDate, "last_month_total_mb": decision.LastMonthTotalMB,
					"last_gateway_reading_mb": decision.LastGatewayReadingMB, "increment_mb": decision.IncrementMB,
				},
			}); err != nil {
				return err
			}
		}
		return nil
	})
	if err != nil {
		if workerObservationAudited(ctx) && auditedCard != nil && s.auditWriter != nil {
			s.auditWriter.WriteCardStateFailure(ctx, StateAudit{
				ActionCode: constants.AuditActionIotCardWorkerTrafficSynced,
				Summary:    "Worker 同步 IoT 卡流量事实失败", Card: auditedCard,
				IntegrationID: observation.Metadata.ObservationID,
			}, err)
		}
		return domain.TrafficDecision{}, err
	}
	if s.cache != nil {
		s.cache.Invalidate(ctx, observation.CardID)
	}
	return decision, nil
}