rsspipe/internal/rules/engine.go

package rules

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"

	"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 JobEnqueuer interface {
	Enqueue(ctx context.Context, jobType string, payload map[string]interface{}, idempotencyKey string) (*models.Job, error)
}

type Engine struct {
	db         *db.DB
	bus        *events.EventBus
	evaluator  *Evaluator
	dispatcher WorkerDispatcher
	enqueuer   JobEnqueuer
	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"),
	}
}

func (e *Engine) SetEnqueuer(enqueuer JobEnqueuer) {
	e.enqueuer = enqueuer
}

// Start registers event bus subscriptions to evaluate rules reactively.
func (e *Engine) Start(ctx context.Context) {
	e.bus.Subscribe(events.EventEntryTransition, e.handleTransitionEvent)
	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) handleTransitionEvent(ctx context.Context, evt events.Event) error {
	rules, err := e.db.ListRules()
	if err != nil {
		e.log.Error("Failed to fetch rules for transition", zap.Error(err))
		return err
	}

	newEntry := evt.NewEntry
	if newEntry == nil {
		newEntry = &evt.Entry
	}

	for _, r := range rules {
		if !r.IsActive {
			continue
		}

		// Check trigger type compatibility
		trigger := r.Trigger
		if trigger != "on_update" && trigger != "on_entry_transition" && trigger != "on_variant_updated" && trigger != "on_post_updated" && trigger != "on_post_created" && trigger != "variant_change" {
			continue
		}

		if e.evaluator.EvaluateTransition(r, evt.OldEntry, newEntry, evt.Diff) {
			e.log.Info("Transition rule matched",
				zap.String("rule_id", r.ID),
				zap.String("rule_name", r.Name),
				zap.String("entry_id", newEntry.ID),
				zap.Int("actions_count", len(r.Actions)),
			)

			e.executeRuleActions(ctx, r, *newEntry, evt.Depth)
		}
	}

	return nil
}

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
	}

	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.EvaluateTransition(r, evt.OldEntry, &entry, evt.Diff) {
			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)),
			)

			e.executeRuleActions(ctx, r, entry, evt.Depth)
		}
	}

	return nil
}

func (e *Engine) executeRuleActions(ctx context.Context, r models.Rule, entry models.FeedEntry, depth int) {
	for _, action := range r.Actions {
		// Prefer enqueuing async jobs to the persistent task queue if available
		if e.enqueuer != nil {
			payload := make(map[string]interface{})
			if action.Params != nil {
				for k, v := range action.Params {
					payload[k] = v
				}
			}
			payload["entry_id"] = entry.ID
			payload["feed_id"] = entry.FeedID
			payload["rule_id"] = r.ID
			payload["rule_name"] = r.Name

			// Generate deterministic idempotency key
			rawKey := fmt.Sprintf("%s:%s:%s:%d:%d", r.ID, entry.ID, action.Type, entry.CreatedAt.Unix(), entry.PublishedAt.Unix())
			h := sha256.Sum256([]byte(rawKey))
			idempotencyKey := hex.EncodeToString(h[:16])

			_, err := e.enqueuer.Enqueue(ctx, action.Type, payload, idempotencyKey)
			if err != nil {
				e.log.Error("Failed to enqueue job for rule action",
					zap.String("rule_id", r.ID),
					zap.String("action", action.Type),
					zap.Error(err),
				)
			}
		} else if e.dispatcher != nil {
			e.dispatcher.Dispatch(ctx, action, entry, depth+1)
		}
	}
}