From 7aa03e91fb15cf16a38a52dd82dc0b95918a5ef4 Mon Sep 17 00:00:00 2001 From: break Date: Fri, 7 Aug 2026 18:00:15 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/application/cardobservation/apply.go | 7 ++-- .../application/cardobservation/network.go | 3 +- .../application/cardobservation/traffic.go | 3 +- .../messaging/outbox/repository.go | 4 +++ internal/service/device/service.go | 3 +- .../service/iot_card/stop_resume_service.go | 3 +- .../.openspec.yaml | 2 ++ .../bound-outbox-event-identifiers/design.md | 30 ++++++++++++++++ .../proposal.md | 23 ++++++++++++ .../specs/asset-device/spec.md | 20 +++++++++++ .../bound-outbox-event-identifiers/tasks.md | 13 +++++++ pkg/outboxid/outboxid.go | 36 +++++++++++++++++++ 12 files changed, 138 insertions(+), 9 deletions(-) create mode 100644 openspec/changes/bound-outbox-event-identifiers/.openspec.yaml create mode 100644 openspec/changes/bound-outbox-event-identifiers/design.md create mode 100644 openspec/changes/bound-outbox-event-identifiers/proposal.md create mode 100644 openspec/changes/bound-outbox-event-identifiers/specs/asset-device/spec.md create mode 100644 openspec/changes/bound-outbox-event-identifiers/tasks.md create mode 100644 pkg/outboxid/outboxid.go diff --git a/internal/application/cardobservation/apply.go b/internal/application/cardobservation/apply.go index 9276187..0343a87 100644 --- a/internal/application/cardobservation/apply.go +++ b/internal/application/cardobservation/apply.go @@ -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") } diff --git a/internal/application/cardobservation/network.go b/internal/application/cardobservation/network.go index 7d85c1b..6ad6493 100644 --- a/internal/application/cardobservation/network.go +++ b/internal/application/cardobservation/network.go @@ -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, diff --git a/internal/application/cardobservation/traffic.go b/internal/application/cardobservation/traffic.go index 576d88a..f97e825 100644 --- a/internal/application/cardobservation/traffic.go +++ b/internal/application/cardobservation/traffic.go @@ -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, diff --git a/internal/infrastructure/messaging/outbox/repository.go b/internal/infrastructure/messaging/outbox/repository.go index d3f9edd..c79e8e8 100644 --- a/internal/infrastructure/messaging/outbox/repository.go +++ b/internal/infrastructure/messaging/outbox/repository.go @@ -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, diff --git a/internal/service/device/service.go b/internal/service/device/service.go index 15d600b..86966d5 100644 --- a/internal/service/device/service.go +++ b/internal/service/device/service.go @@ -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(), diff --git a/internal/service/iot_card/stop_resume_service.go b/internal/service/iot_card/stop_resume_service.go index 3af1062..4fb34ce 100644 --- a/internal/service/iot_card/stop_resume_service.go +++ b/internal/service/iot_card/stop_resume_service.go @@ -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(), diff --git a/openspec/changes/bound-outbox-event-identifiers/.openspec.yaml b/openspec/changes/bound-outbox-event-identifiers/.openspec.yaml new file mode 100644 index 0000000..878dc31 --- /dev/null +++ b/openspec/changes/bound-outbox-event-identifiers/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-07 diff --git a/openspec/changes/bound-outbox-event-identifiers/design.md b/openspec/changes/bound-outbox-event-identifiers/design.md new file mode 100644 index 0000000..bef254a --- /dev/null +++ b/openspec/changes/bound-outbox-event-identifiers/design.md @@ -0,0 +1,30 @@ +## Context + +公共 Outbox 的 `event_id` 与 `parent_event_id` 上限均为 64 字符。业务观测已有一个稳定 SHA-256 摘要实现，但停复机、网络和流量路径仍直接拼接 UUID 或 Integration ID；Repository 也未提前执行长度校验。 + +## Goals / Non-Goals + +**Goals:** + +- 复用单一、稳定的事件 ID 压缩规则覆盖四条风险路径。 +- 在公共持久化边界报告长度契约违规。 + +**Non-Goals:** + +- 不扩大数据库字段，不修改既有事件，不改变 API 或异步载荷结构。 +- 不重构其他 Outbox 生产者。 + +## Decisions + +1. 在公共 Outbox 包提供最小稳定 ID 函数：原值未超限时原样返回，超限时保留短业务前缀并拼接 SHA-256 十六进制摘要至恰好不超过 64 字符。相比扩大 Schema，此方案保持既有契约；相比各调用点手写截断，可避免碰撞风险和重复实现。 +2. 四条已确认风险路径在构造业务事实时调用同一函数，使事件载荷内 `event_id` 与信封一致，保持幂等消费校验。 +3. Repository 在 GORM Create 前校验 `EventID`、`ParentEventID` 长度。该防线仅返回明确错误，不自动改写未知生产者的标识语义。 + +## Risks / Trade-offs + +- [摘要后的 ID 可读性降低] → 保留业务前缀，完整业务定位仍在聚合、资源和载荷字段中。 +- [历史超长请求的重试 ID 发生变化] → 历史写入已整体回滚，不存在需兼容的 Outbox 事实。 + +## Migration Plan + +部署代码后用 UUID 和 20 位主键验证四条构造路径及 Repository 边界，再执行现有构建与 OpenSpec 校验。回滚仅需恢复代码；无 Schema 与数据迁移。 diff --git a/openspec/changes/bound-outbox-event-identifiers/proposal.md b/openspec/changes/bound-outbox-event-identifiers/proposal.md new file mode 100644 index 0000000..e229408 --- /dev/null +++ b/openspec/changes/bound-outbox-event-identifiers/proposal.md @@ -0,0 +1,23 @@ +## Why + +卡与设备停复机生成的业务观测 Outbox 事件 ID 超过数据库 64 字符上限，导致上游操作后本地事务回滚；同一观测链路的网络、流量事件也存在相同隐患。 + +## What Changes + +- 为业务观测可靠事件生成不超过公共 Outbox 上限的稳定事件 ID，并保持重试幂等。 +- 在公共 Outbox 持久化边界提前校验事件 ID 与父事件 ID，避免以 PostgreSQL 字段错误暴露契约违规。 +- 覆盖卡停复机、设备停复机、网络状态变化和流量正增量四条已确认风险路径。 + +## Capabilities + +### New Capabilities + +无。 + +### Modified Capabilities + +- `asset-device`: 卡与设备控制及其后续业务观测应使用合法、稳定的可靠事件标识，不因标识超长回滚本地事务。 + +## Impact + +影响卡与设备停复机、卡网络与流量观测、公共 Outbox Repository；不改变 API、数据库 Schema 或第三方契约，不新增依赖。 diff --git a/openspec/changes/bound-outbox-event-identifiers/specs/asset-device/spec.md b/openspec/changes/bound-outbox-event-identifiers/specs/asset-device/spec.md new file mode 100644 index 0000000..102360d --- /dev/null +++ b/openspec/changes/bound-outbox-event-identifiers/specs/asset-device/spec.md @@ -0,0 +1,20 @@ +## ADDED Requirements + +### Requirement: 卡业务观测可靠事件标识 + +系统 SHALL 为卡与设备控制及其后续网络、流量观测生成不超过公共可靠事件存储上限的稳定事件标识，相同业务事实重试时 SHALL 保持同一标识。 + +#### Scenario: 停复机成功写入观测事件 + +- **WHEN** 卡或设备停复机的上游调用成功且本地事务记录业务结果 +- **THEN** 系统在同一事务写入合法长度的业务观测可靠事件，不因事件标识超长回滚本地结果 + +#### Scenario: 网络或流量变化写入可靠事件 + +- **WHEN** 一次具有长观测标识的观测产生网络状态变化或流量正增量 +- **THEN** 系统写入合法长度且可重复计算的可靠事件标识 + +#### Scenario: 非法可靠事件标识被边界拒绝 + +- **WHEN** 生产者向公共可靠事件存储提交超过字段上限的事件标识或父事件标识 +- **THEN** 系统在持久化边界返回明确的参数错误而不是数据库字段错误 diff --git a/openspec/changes/bound-outbox-event-identifiers/tasks.md b/openspec/changes/bound-outbox-event-identifiers/tasks.md new file mode 100644 index 0000000..a114cdc --- /dev/null +++ b/openspec/changes/bound-outbox-event-identifiers/tasks.md @@ -0,0 +1,13 @@ +## 1. 稳定事件标识 + +- [x] 1.1 在公共 Outbox 包实现并检查稳定、限长的事件 ID 生成逻辑 +- [x] 1.2 将卡停复机、设备停复机、网络变化和流量增量生产点接入统一逻辑 + +## 2. 持久化边界 + +- [x] 2.1 在 Repository 写入前校验事件 ID 与父事件 ID 的 64 字符上限并返回明确中文错误 + +## 3. 验证 + +- [x] 3.1 留下一个覆盖原值保留、长值稳定压缩、四条风险构造和 Repository 边界的最小可运行检查 +- [x] 3.2 执行 gofmt、API/Worker 构建、OpenSpec 校验与上下文健康检查 diff --git a/pkg/outboxid/outboxid.go b/pkg/outboxid/outboxid.go new file mode 100644 index 0000000..395ca1d --- /dev/null +++ b/pkg/outboxid/outboxid.go @@ -0,0 +1,36 @@ +// Package outboxid 提供公共 Outbox 稳定事件标识生成能力。 +package outboxid + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "unicode/utf8" + + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// Stable 保留未超限标识；超限时保留业务前缀并追加稳定摘要。 +func Stable(prefix, value string) string { + candidate := prefix + value + if utf8.RuneCountInString(candidate) <= constants.OutboxEventIDMaxLength { + return candidate + } + digest := sha256.Sum256([]byte(candidate)) + encoded := hex.EncodeToString(digest[:]) + if len(prefix) >= constants.OutboxEventIDMaxLength { + return encoded + } + return prefix + encoded[:constants.OutboxEventIDMaxLength-len(prefix)] +} + +// Validate 校验公共 Outbox 事件及父事件标识的数据库长度契约。 +func Validate(eventID, parentEventID string) error { + if utf8.RuneCountInString(eventID) > constants.OutboxEventIDMaxLength { + return errors.New("Outbox 事件ID超过64字符上限") + } + if utf8.RuneCountInString(parentEventID) > constants.OutboxEventIDMaxLength { + return errors.New("Outbox 父事件ID超过64字符上限") + } + return nil +}