package packagetrafficalert import ( "context" "strconv" "gorm.io/gorm" notificationapp "github.com/break/junhong_cmp_fiber/internal/application/notification" app "github.com/break/junhong_cmp_fiber/internal/application/packagetrafficalert" "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit" "github.com/break/junhong_cmp_fiber/internal/infrastructure/messaging/outbox" "github.com/break/junhong_cmp_fiber/internal/model" "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" ) // AlertWriter 在同一 GORM 事务内写入预警事实、可靠通知事件与成功审计。 // // 依据 ENG-OUTBOX-001,业务事实与可靠异步副作用必须同事务写 Outbox; // 依据 ENG-TX-001,成功必达的审计与关键状态事实同事务,且事务内不持有不可回滚的外部 I/O。 // 事务粒度为一个命中资产:单资产失败不阻塞其余资产,锁持有时间可控。 type AlertWriter struct { db *gorm.DB store *postgres.PackageTrafficAlertStore outbox *outbox.Repository auditWriter *audit.Writer } // NewAlertWriter 创建套餐真流量预警写入 Adapter。 func NewAlertWriter(db *gorm.DB, store *postgres.PackageTrafficAlertStore, repository *outbox.Repository, auditWriter *audit.Writer) *AlertWriter { return &AlertWriter{db: db, store: store, outbox: repository, auditWriter: auditWriter} } // SaveAlert 幂等创建预警,并在同一事务内写入通知事件与审计。 // 唯一键冲突(同一使用记录 + 同一阈值快照)视为已处理:不写事件、不写审计,直接提交。 func (w *AlertWriter) SaveAlert(ctx context.Context, candidate app.AlertCandidate) (bool, error) { if w == nil || w.db == nil || w.store == nil { return false, errors.New(errors.CodeInternalError, "套餐真流量达量预警写入 Adapter 未配置") } if w.auditWriter == nil { return false, errors.New(errors.CodeInvalidStatus, "套餐真流量达量预警统一审计接缝未配置") } alert := candidate.Alert created := false err := w.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { store := w.store.WithTx(tx) inserted, insertErr := store.CreateAlertIdempotent(ctx, &alert) if insertErr != nil { return errors.Wrap(errors.CodeDatabaseError, insertErr, "创建套餐真流量达量预警失败") } if !inserted { return nil } created = true if candidate.Notification != nil { eventID, appendErr := w.appendNotificationEvent(ctx, tx, alert, *candidate.Notification) if appendErr != nil { return appendErr } if updateErr := tx.WithContext(ctx).Model(&model.PackageTrafficAlert{}). Where("id = ?", alert.ID).Update("notification_event_id", eventID).Error; updateErr != nil { return errors.Wrap(errors.CodeDatabaseError, updateErr, "回填预警通知事件ID失败") } alert.NotificationEventID = eventID } return w.appendTriggerAudit(ctx, tx, alert) }) if err != nil { return false, err } return created, nil } // appendNotificationEvent 幂等追加动态通知事件。 // 接收人是触发时冻结的业务员账号(TargetKind=account),投递期不再重新解析店铺业务员, // 因此业务员变更不会把通知送给不属于它的新账号,也不会在触发时无业务员的情况下补发。 func (w *AlertWriter) appendNotificationEvent(ctx context.Context, tx *gorm.DB, alert model.PackageTrafficAlert, request app.NotificationRequest) (string, error) { if w.outbox == nil { return "", errors.New(errors.CodeInternalError, "套餐真流量达量预警 Outbox 仓储未配置") } eventID := app.EventIDFor(alert.PackageUsageID, alert.ThresholdPercentSnapshot) alertID := strconv.FormatUint(uint64(alert.ID), 10) expiresAt := request.ExpiresAt _, err := w.outbox.AppendIdempotent(ctx, tx, outbox.Envelope{ EventID: eventID, EventType: constants.OutboxEventTypeAdminDynamicNotification, PayloadVersion: constants.NotificationPayloadVersionV1, AggregateType: constants.PackageTrafficAlertScanAggregateType, AggregateID: alertID, ResourceType: constants.NotificationRefTypePackageTrafficAlert, ResourceID: alertID, BusinessKey: eventID, Payload: notificationapp.AdminDynamicPayload{ TargetKind: constants.NotificationTargetKindAccount, TargetID: request.RecipientAccountID, NotificationType: constants.NotificationTypePackageTrafficAlert, TemplateData: request.TemplateData, RefType: constants.NotificationRefTypePackageTrafficAlert, RefID: alertID, RefKey: alert.AssetIdentifierSnapshot, ExpiresAt: &expiresAt, }, }) if err != nil { return "", errors.Wrap(errors.CodeDatabaseError, err, "写入套餐真流量达量预警通知事件失败") } return eventID, nil } // appendTriggerAudit 在业务事务内追加达量预警审计事实。 // 审计动作的 actor/source 固定为系统任务与 Worker,与注册表声明一致。 func (w *AlertWriter) appendTriggerAudit(ctx context.Context, tx *gorm.DB, alert model.PackageTrafficAlert) error { alertID := strconv.FormatUint(uint64(alert.ID), 10) usageID := strconv.FormatUint(uint64(alert.PackageUsageID), 10) summary := "创建套餐真流量达量预警" if alert.NotificationEventID == "" { summary = "创建套餐真流量达量预警(触发时无有效业务员,未生成通知)" } // 使用 AppendAndGet:达量预警审计属于「要求成功必达」的事实,失败必须回滚整个事务(ENG-TX-001)。 if _, err := w.auditWriter.AppendAndGet(ctx, tx, audit.AppendInput{ EventID: audit.TaskEventID(constants.AuditResourcePackageTrafficAlert, alert.ID, "trigger"), ActionCode: constants.AuditActionPackageTrafficAlertTriggered, Summary: summary, // Actor.ID 必须非空且稳定:审计写入要求操作者标识,缺省会静默失败。 Actor: audit.ActorInput{ Kind: constants.AuditActorSystemTask, ID: constants.TaskTypePackageTrafficAlertScan, Name: "套餐真流量达量扫描", }, Source: constants.AuditSourceWorker, ScopeType: constants.AuditScopePlatform, Result: constants.AuditResultSuccess, Resources: []audit.ResourceInput{{ Type: constants.AuditResourcePackageTrafficAlert, ID: &alertID, Key: usageID, DisplayName: alert.PackageNameSnapshot, Relation: constants.AuditResourceRelationPrimary, Role: constants.AuditResourceRolePackageTrafficAlertTarget, IdentitySnapshot: alertAuditIdentity(alert), AfterData: alertAuditSnapshot(alert), }}, }); err != nil { return err } return nil } // alertAuditIdentity 返回预警审计身份快照,字段落在注册表白名单内。 func alertAuditIdentity(alert model.PackageTrafficAlert) map[string]any { identity := map[string]any{ "id": alert.ID, "package_usage_id": alert.PackageUsageID, "package_id": alert.PackageID, "asset_type": alert.AssetType, "asset_id": alert.AssetID, "threshold_percent_snapshot": alert.ThresholdPercentSnapshot, "used_mb_snapshot": alert.UsedMBSnapshot, "limit_mb_snapshot": alert.LimitMBSnapshot, "usage_percent_snapshot": alert.UsagePercentSnapshot, "shop_id": alert.ShopIDSnapshot, } if alert.BusinessOwnerAccountIDSnapshot != nil { identity["business_owner_account_id"] = *alert.BusinessOwnerAccountIDSnapshot } else { identity["business_owner_account_id"] = nil } return identity } // alertAuditSnapshot 返回预警审计的 after 快照,用于忠实还原触发时的冻结事实。 func alertAuditSnapshot(alert model.PackageTrafficAlert) map[string]any { snapshot := alertAuditIdentity(alert) snapshot["rule_id"] = alert.RuleID snapshot["asset_identifier"] = alert.AssetIdentifierSnapshot snapshot["card_identifier"] = alert.CardIdentifierSnapshot snapshot["counterpart_identifier"] = alert.CounterpartIdentifierSnapshot snapshot["package_name"] = alert.PackageNameSnapshot snapshot["shop_name"] = alert.ShopNameSnapshot snapshot["notification_event_id"] = alert.NotificationEventID if alert.BusinessOwnerAccountIDSnapshot != nil { snapshot["business_owner_account_id"] = *alert.BusinessOwnerAccountIDSnapshot } else { snapshot["business_owner_account_id"] = nil } if alert.ExpiresAtSnapshot != nil { snapshot["expires_at"] = alert.ExpiresAtSnapshot.UTC() } else { snapshot["expires_at"] = nil } snapshot["triggered_at"] = alert.TriggeredAt.UTC() return snapshot }