// Package outbox 提供公共 Outbox 的只读运行状态投影。 package outbox import ( "context" "time" "gorm.io/gorm" "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/pkg/constants" ) // EventTypeBacklog 表示某事件类型的未投递积压。 type EventTypeBacklog struct { EventType string `json:"event_type"` Count int64 `json:"count"` } // RetryBucket 表示某重试次数上的事件数量。 type RetryBucket struct { RetryCount int `json:"retry_count"` Count int64 `json:"count"` } // Metrics 是 Outbox 运维公开指标投影。 type Metrics struct { PendingCount int64 `json:"pending_count"` OldestPendingAgeSecs int64 `json:"oldest_pending_age_seconds"` DeliveringCount int64 `json:"delivering_count"` ExpiredLeaseCount int64 `json:"expired_lease_count"` DeliveredInWindow int64 `json:"delivered_in_window"` FinalFailedCount int64 `json:"final_failed_count"` DeliverySuccessRate float64 `json:"delivery_success_rate"` RetryDistribution []RetryBucket `json:"retry_distribution"` BacklogByEventType []EventTypeBacklog `json:"backlog_by_event_type"` ObservedAt time.Time `json:"observed_at"` } // Query 查询 PostgreSQL 中的 Outbox 运行事实。 type Query struct { db *gorm.DB now func() time.Time } // NewQuery 创建 Outbox 指标查询。 func NewQuery(db *gorm.DB, now func() time.Time) *Query { if now == nil { now = time.Now } return &Query{db: db, now: now} } // GetMetrics 查询指定统计窗口的积压、成功率、重试和最终失败。 func (q *Query) GetMetrics(ctx context.Context, window time.Duration) (Metrics, error) { now := q.now().UTC() if window <= 0 { window = time.Hour } metrics := Metrics{ObservedAt: now, RetryDistribution: []RetryBucket{}, BacklogByEventType: []EventTypeBacklog{}} base := func() *gorm.DB { return q.db.WithContext(ctx).Model(&model.OutboxEvent{}) } if err := base().Where("status = ?", constants.OutboxStatusPending).Count(&metrics.PendingCount).Error; err != nil { return metrics, err } if err := base().Where("status = ?", constants.OutboxStatusDelivering).Count(&metrics.DeliveringCount).Error; err != nil { return metrics, err } if err := base().Where("status = ? AND lease_expires_at <= ?", constants.OutboxStatusDelivering, now). Count(&metrics.ExpiredLeaseCount).Error; err != nil { return metrics, err } if err := base().Where("status = ?", constants.OutboxStatusFailed).Count(&metrics.FinalFailedCount).Error; err != nil { return metrics, err } var oldestEvent model.OutboxEvent err := base().Select("created_at").Where("status = ?", constants.OutboxStatusPending). Order("created_at ASC").Take(&oldestEvent).Error if err != nil && err != gorm.ErrRecordNotFound { return metrics, err } if err == nil && oldestEvent.CreatedAt.Before(now) { metrics.OldestPendingAgeSecs = int64(now.Sub(oldestEvent.CreatedAt).Seconds()) } windowStart := now.Add(-window) if err := base().Where("status = ? AND delivered_at >= ?", constants.OutboxStatusDelivered, windowStart). Count(&metrics.DeliveredInWindow).Error; err != nil { return metrics, err } var failedInWindow int64 if err := base().Where("status = ? AND updated_at >= ?", constants.OutboxStatusFailed, windowStart). Count(&failedInWindow).Error; err != nil { return metrics, err } totalTerminal := metrics.DeliveredInWindow + failedInWindow if totalTerminal > 0 { metrics.DeliverySuccessRate = float64(metrics.DeliveredInWindow) / float64(totalTerminal) } if err := base().Select("retry_count, COUNT(*) AS count").Where("retry_count > 0"). Group("retry_count").Order("retry_count ASC").Scan(&metrics.RetryDistribution).Error; err != nil { return metrics, err } if err := base().Select("event_type, COUNT(*) AS count").Where("status IN ?", []int{ constants.OutboxStatusPending, constants.OutboxStatusDelivering, constants.OutboxStatusFailed, }).Group("event_type").Order("event_type ASC").Scan(&metrics.BacklogByEventType).Error; err != nil { return metrics, err } return metrics, nil } // Thresholds 是不包含高基数字段的 Outbox 告警阈值。 type Thresholds struct { PendingCount int64 OldestPendingAge time.Duration ExpiredLeaseCount int64 FinalFailedCount int64 } // Alert 是可交给现有告警通道的中文安全摘要。 type Alert struct { Code string `json:"code"` Component string `json:"component"` Count int64 `json:"count"` Summary string `json:"summary"` Window string `json:"window"` } // EvaluateAlerts 根据聚合指标生成低基数告警,不包含 payload 或敏感配置。 func EvaluateAlerts(metrics Metrics, thresholds Thresholds) []Alert { alerts := make([]Alert, 0, 4) if thresholds.PendingCount > 0 && metrics.PendingCount >= thresholds.PendingCount { alerts = append(alerts, Alert{Code: "OUTBOX_PENDING_HIGH", Component: "outbox", Count: metrics.PendingCount, Summary: "Outbox 待投递事件持续积压", Window: "当前快照"}) } if thresholds.OldestPendingAge > 0 && metrics.OldestPendingAgeSecs >= int64(thresholds.OldestPendingAge.Seconds()) { alerts = append(alerts, Alert{Code: "OUTBOX_OLDEST_PENDING_HIGH", Component: "outbox", Count: metrics.OldestPendingAgeSecs, Summary: "Outbox 最老待投递事件超过阈值", Window: "秒"}) } if thresholds.ExpiredLeaseCount > 0 && metrics.ExpiredLeaseCount >= thresholds.ExpiredLeaseCount { alerts = append(alerts, Alert{Code: "OUTBOX_EXPIRED_LEASE_HIGH", Component: "outbox", Count: metrics.ExpiredLeaseCount, Summary: "Outbox 过期租约数量超过阈值", Window: "当前快照"}) } if thresholds.FinalFailedCount > 0 && metrics.FinalFailedCount >= thresholds.FinalFailedCount { alerts = append(alerts, Alert{Code: "OUTBOX_FINAL_FAILED_HIGH", Component: "outbox", Count: metrics.FinalFailedCount, Summary: "Outbox 最终失败事件需要人工处理", Window: "当前快照"}) } return alerts }