package rules import ( "context" "git.pengzhan.dev/haopengzhan/rsspipe/internal/db" "git.pengzhan.dev/haopengzhan/rsspipe/internal/events" "git.pengzhan.dev/haopengzhan/rsspipe/internal/logger" "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" "go.uber.org/zap" ) type WorkerDispatcher interface { Dispatch(ctx context.Context, action models.RuleAction, entry models.FeedEntry, depth int) } type Engine struct { db *db.DB bus *events.EventBus evaluator *Evaluator dispatcher WorkerDispatcher log *zap.Logger } func NewEngine(database *db.DB, bus *events.EventBus, dispatcher WorkerDispatcher) *Engine { return &Engine{ db: database, bus: bus, evaluator: NewEvaluator(), dispatcher: dispatcher, log: logger.Named("rules-engine"), } } // Start registers event bus subscriptions to evaluate rules reactively. func (e *Engine) Start(ctx context.Context) { e.bus.Subscribe(events.EventPostCreated, e.handleEvent) e.bus.Subscribe(events.EventPostUpdated, e.handleEvent) e.bus.Subscribe(events.EventVariantUpdated, e.handleEvent) e.bus.Subscribe(events.EventManualTrigger, e.handleEvent) e.log.Info("Rules engine started, listening for events") } func (e *Engine) handleEvent(ctx context.Context, evt events.Event) error { trigger := string(evt.Type) rules, err := e.db.GetActiveRules(trigger) if err != nil { e.log.Error("Failed to fetch active rules", zap.String("trigger", trigger), zap.Error(err)) return err } if len(rules) == 0 { return nil } // Fetch full entry with latest variants if entry ID is present entry := evt.Entry if entry.ID != "" { if full, err := e.db.GetEntry(entry.ID); err == nil && full != nil { entry = *full } } for _, r := range rules { if e.evaluator.Evaluate(r, entry) { e.log.Info("Rule matched", zap.String("rule_id", r.ID), zap.String("rule_name", r.Name), zap.String("entry_id", entry.ID), zap.Int("actions_count", len(r.Actions)), ) for _, action := range r.Actions { if e.dispatcher != nil { e.dispatcher.Dispatch(ctx, action, entry, evt.Depth+1) } } } } return nil }