package events import ( "context" "sync" "git.pengzhan.dev/haopengzhan/rsspipe/internal/logger" "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" "go.uber.org/zap" ) type EventType string const ( EventPostCreated EventType = "on_post_created" EventPostUpdated EventType = "on_post_updated" EventVariantUpdated EventType = "on_variant_updated" EventManualTrigger EventType = "manual" ) type Event struct { Type EventType Entry models.FeedEntry VariantKey string // populated when Type == EventVariantUpdated Depth int DirectAttrs map[string]interface{} } type EventHandler func(ctx context.Context, evt Event) error type EventBus struct { mu sync.RWMutex subscribers map[EventType][]EventHandler log *zap.Logger maxDepth int } func NewEventBus() *EventBus { return &EventBus{ subscribers: make(map[EventType][]EventHandler), log: logger.Named("eventbus"), maxDepth: 5, } } func (b *EventBus) Subscribe(eventType EventType, handler EventHandler) { b.mu.Lock() defer b.mu.Unlock() b.subscribers[eventType] = append(b.subscribers[eventType], handler) } func (b *EventBus) Publish(ctx context.Context, evt Event) { if evt.Depth > b.maxDepth { b.log.Warn("Event dropped: recursion depth limit exceeded", zap.String("type", string(evt.Type)), zap.String("entry_id", evt.Entry.ID), zap.Int("depth", evt.Depth), ) return } b.mu.RLock() handlers := append([]EventHandler{}, b.subscribers[evt.Type]...) wildcardHandlers := append([]EventHandler{}, b.subscribers["*"]...) b.mu.RUnlock() b.log.Debug("Event published", zap.String("type", string(evt.Type)), zap.String("entry_id", evt.Entry.ID), zap.String("variant_key", evt.VariantKey), zap.Int("depth", evt.Depth), zap.Int("subscriber_count", len(handlers)+len(wildcardHandlers)), ) allHandlers := append(handlers, wildcardHandlers...) for _, h := range allHandlers { go func(fn EventHandler) { defer func() { if r := recover(); r != nil { b.log.Error("Event handler panic recovered", zap.Any("panic", r)) } }() if err := fn(ctx, evt); err != nil { b.log.Error("Event handler error", zap.String("type", string(evt.Type)), zap.String("entry_id", evt.Entry.ID), zap.Error(err), ) } }(h) } }