package warningnotify import ( "context" "encoding/json" "github.com/gogf/gf/v2/errors/gerror" "sort" "strings" ) type Collector struct{ batches map[string]*Batch } func NewCollector() *Collector { return &Collector{batches: make(map[string]*Batch)} } func (c *Collector) Add(ctx context.Context, events []Event) error { for _, batch := range BuildBatches(events) { switch batch.Method { case MethodPlatform: deliveries, err := resolvePlatformDeliveries(ctx, batch) if err != nil { return err } for _, delivery := range deliveries { c.merge(delivery.Batch, &delivery.Group) } case MethodWechat: for _, delivery := range resolveWechatDeliveries(ctx, batch) { c.merge(delivery.Batch, &delivery.Group) } default: c.merge(batch, nil) } } return nil } func (c *Collector) merge(batch Batch, recipient *platformRecipientGroup) { keyBytes, _ := json.Marshal(struct { Group groupKey Recipient *platformRecipientGroup }{groupKeyForEvent(batch.Method, batch.Recipients, Event{Business: batch.Business, TenantID: batch.TenantID, DeptID: batch.DeptID, CreatedBy: batch.CreatedBy, Category: batch.Category}), recipient}) key := string(keyBytes) current := c.batches[key] if current == nil { copy := batch copy.Items = nil copy.TotalCount = 0 copy.prepared = recipient current = © c.batches[key] = current } current.TotalCount += len(batch.Items) current.WarnStage = max(current.WarnStage, batch.WarnStage) current.HighestLevel = max(current.HighestLevel, highestLevel(batch.Items)) current.AffectOperation = current.AffectOperation || hasAffectOperation(batch.Items) latest := latestEvent(batch.Items).WarnTime if latest.After(current.LatestTime) { current.LatestTime = latest } current.Items = append(current.Items, batch.Items...) if recipient != nil { sort.SliceStable(current.Items, func(i, j int) bool { if current.Items[i].WarnLevel != current.Items[j].WarnLevel { return current.Items[i].WarnLevel > current.Items[j].WarnLevel } if current.Items[i].WarnStage != current.Items[j].WarnStage { return current.Items[i].WarnStage > current.Items[j].WarnStage } return current.Items[i].ItemCode < current.Items[j].ItemCode }) if len(current.Items) > 5 { preview := append([]Event(nil), current.Items[:5]...) current.Items = preview } } } func (c *Collector) Flush(ctx context.Context) error { senderMu.RLock() sender := currentSender senderMu.RUnlock() if sender == nil { sender = LogSender{} } keys := make([]string, 0, len(c.batches)) for key := range c.batches { keys = append(keys, key) } sort.Strings(keys) var failures []string for _, key := range keys { if err := ctx.Err(); err != nil { return err } batch := c.batches[key] applyContent(batch) if err := sender.Send(ctx, *batch); err != nil { failures = append(failures, err.Error()) } } if len(failures) > 0 { return gerror.New(strings.Join(failures, "; ")) } return nil } func batchItemCount(batch Batch) int { if batch.TotalCount > 0 { return batch.TotalCount } return len(batch.Items) } func batchHighestLevel(batch Batch) int { return max(batch.HighestLevel, highestLevel(batch.Items)) } func batchHasAffectOperation(batch Batch) bool { return batch.AffectOperation || hasAffectOperation(batch.Items) }