csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 8m25s
Details

Browse Source
This commit is contained in:
break
2026-08-07 18:00:15 +08:00
parent 3421b5f106
commit 7aa03e91fb
12 changed files with 138 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
internal/application/cardobservation/apply.go
View File

@@ -3,8 +3,6 @@ package cardobservation
import (
"context"
"crypto/sha256"
"encoding/hex"
"strconv"
"time"
@@ -13,6 +11,7 @@ import (
"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/outboxid"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
@@ -257,7 +256,5 @@ func workerObservationAudited(ctx context.Context) bool {
func realnameChangedEventID(cardID uint, observationID string) string {
prefix := "card-realname:"
digest := sha256.Sum256([]byte(strconv.FormatUint(uint64(cardID), 10) + ":" + observationID + ":changed"))
// 使用稳定摘要保留幂等语义，同时严格服从公共 Outbox 的字段长度上限。
return prefix + hex.EncodeToString(digest[:])[:constants.OutboxEventIDMaxLength-len(prefix)]
return outboxid.Stable(prefix, strconv.FormatUint(uint64(cardID), 10)+":"+observationID+":changed")
}

3
internal/application/cardobservation/network.go
View File

@@ -9,6 +9,7 @@ import (
"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"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
@@ -78,7 +79,7 @@ func (s *Service) ApplyNetworkObservation(ctx context.Context, observation domai
return errors.New(errors.CodeConflict, "卡网络事实已被其他请求更新")
}
if decision.StatusChanged {
eventID := "card-network:" + strconv.FormatUint(uint64(card.ID), 10) + ":" + observation.Metadata.ObservationID + ":changed"
eventID := outboxid.Stable("card-network:", strconv.FormatUint(uint64(card.ID), 10)+":"+observation.Metadata.ObservationID+":changed")
if err := s.eventWriter.AppendNetwork(ctx, tx, NetworkChangedEvent{
EventID: eventID, CardID: card.ID, BeforeStatus: card.NetworkStatus, AfterStatus: decision.AfterStatus,
GatewayExtend: decision.GatewayExtend, ObservedAt: observation.Metadata.ObservedAt,

3
internal/application/cardobservation/traffic.go
View File

@@ -12,6 +12,7 @@ import (
"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 是卡流量正增量的可靠领域事实。
@@ -69,7 +70,7 @@ func (s *Service) ApplyTrafficObservation(ctx context.Context, observation domai
return errors.New(errors.CodeConflict, "卡流量基线已被其他请求更新")
}
if decision.IncrementMB > 0 {
eventID := "card-traffic:" + strconv.FormatUint(uint64(card.ID), 10) + ":" + observation.Metadata.ObservationID + ":incremented"
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,

4
internal/infrastructure/messaging/outbox/repository.go
View File

@@ -17,6 +17,7 @@ import (
"github.com/break/junhong_cmp_fiber/pkg/asynctask"
"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/pkg/outboxid"
)
// Envelope 是业务 UseCase 在事务内追加的公共事件信封。
@@ -85,6 +86,9 @@ func (r *Repository) append(ctx context.Context, tx *gorm.DB, envelope Envelope,
if envelope.ParentEventID == "" {
envelope.ParentEventID = linkage.ParentEventID
}
if err := outboxid.Validate(envelope.EventID, envelope.ParentEventID); err != nil {
return nil, err
}
now := time.Now().UTC()
event := &model.OutboxEvent{
EventID: envelope.EventID, EventType: envelope.EventType, PayloadVersion: envelope.PayloadVersion,

3
internal/service/device/service.go
View File

@@ -24,6 +24,7 @@ import (
"github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/logger"
"github.com/break/junhong_cmp_fiber/pkg/middleware"
"github.com/break/junhong_cmp_fiber/pkg/outboxid"
)
type Service struct {
@@ -1721,7 +1722,7 @@ func (s *Service) updateCardAndAppendNetworkSeries(
requestID = *value
}
return s.observationSeriesEvents.AppendSeriesRequested(ctx, tx, cardObservationApp.SeriesRequestedEvent{
EventID: "card-observation:network-command:" + operationID,
EventID: outboxid.Stable("card-observation:network-command:", operationID),
Scene: scene, ResourceType: constants.CardObservationResourceTypeCard, ResourceID: card.ID,
SyncTypes: []string{constants.CardObservationSyncTypeNetwork}, ExpectedValue: expected,
Source: constants.CardObservationSourceBusinessEvent, OccurredAt: time.Now().UTC(),

3
internal/service/iot_card/stop_resume_service.go
View File

@@ -19,6 +19,7 @@ import (
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/outboxid"
)
// StopResumeServiceInterface 停复机服务接口
@@ -893,7 +894,7 @@ func (s *StopResumeService) updateCardAndAppendNetworkSeries(
return nil
}
return s.observationSeriesEvents.AppendSeriesRequested(ctx, tx, cardObservationApp.SeriesRequestedEvent{
EventID: "card-observation:network-command:" + operationID,
EventID: outboxid.Stable("card-observation:network-command:", operationID),
Scene: scene, ResourceType: constants.CardObservationResourceTypeCard, ResourceID: card.ID,
SyncTypes: []string{constants.CardObservationSyncTypeNetwork}, ExpectedValue: expected,
Source: constants.CardObservationSourceBusinessEvent, OccurredAt: time.Now().UTC(),