Cyrene_v2/internal/services/automations/rule.go

package automations

import (
	"context"
	"fmt"
	"log/slog"
	"sync"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
)

// TriggerKind classifies how a Rule fires.
type TriggerKind string

const (
	// TriggerCron fires on a cron schedule (Expression is a cron spec).
	TriggerCron TriggerKind = "cron"
	// TriggerEvent fires on a bus event (Expression is a bus subject pattern).
	TriggerEvent TriggerKind = "event"
)

// Rule is an automation rule: a trigger plus an action.
type Rule struct {
	// ID uniquely identifies the rule.
	ID string
	// Name is a human-readable label.
	Name string
	// Trigger classifies how the rule fires (cron or event).
	Trigger TriggerKind
	// Expression is the cron spec when Trigger is cron, or a bus subject pattern
	// when Trigger is event.
	Expression string
	// Action runs when the rule fires.
	Action Action
}

// valid reports whether the rule carries every field required to fire.
func (r Rule) valid() bool {
	return r.ID != "" && r.Action != nil && r.Expression != "" &&
		(r.Trigger == TriggerCron || r.Trigger == TriggerEvent)
}

// Engine is the event-driven rule engine. It subscribes to bus subjects on behalf
// of event-triggered rules, matches incoming events to rules and runs the rule's
// action, publishing the outcome to the bus. Cron-triggered rules are handled by
// the Scheduler, not here. It is safe for concurrent use.
type Engine struct {
	cfg Config
	bus bus.Bus
	log *slog.Logger

	mu     sync.RWMutex
	rules  map[string]Rule   // rule id -> rule
	unsub  map[string]func() // rule id -> bus unsubscribe func
	ctx    context.Context
	cancel context.CancelFunc
}

// NewEngine builds an event rule engine. The bus in deps is required.
func NewEngine(cfg Config, deps Deps) (*Engine, error) {
	if deps.Bus == nil {
		return nil, fmt.Errorf("automations: new engine: %w", ErrNilBus)
	}
	return &Engine{
		cfg:   cfg,
		bus:   deps.Bus,
		log:   sourcedLog(deps.Log),
		rules: make(map[string]Rule),
		unsub: make(map[string]func()),
	}, nil
}

// AddEvent registers an event-triggered rule. It subscribes to r.Expression on
// the bus; events whose subject matches the pattern run the rule's action. It
// returns an error wrapping ErrInvalidRule when the rule is malformed or not an
// event rule, and ErrDuplicate when the id is already registered. ctx bounds the
// registration call; the subscription itself is tied to the engine's lifecycle.
func (e *Engine) AddEvent(ctx context.Context, r Rule) error {
	if r.Trigger != TriggerEvent || !r.valid() {
		return fmt.Errorf("automations: add event rule %q: %w", r.ID, ErrInvalidRule)
	}
	if err := e.checkCtx(ctx); err != nil {
		return err
	}

	e.mu.Lock()
	if _, exists := e.rules[r.ID]; exists {
		e.mu.Unlock()
		return fmt.Errorf("automations: add event rule %q: %w", r.ID, ErrDuplicate)
	}
	unsub, err := e.bus.Subscribe(context.Background(), r.Expression, func(_ context.Context, ev bus.Event) error {
		e.dispatch(r, ev)
		return nil
	})
	if err != nil {
		e.mu.Unlock()
		return fmt.Errorf("automations: add event rule %q: %w", r.ID, err)
	}
	e.rules[r.ID] = r
	e.unsub[r.ID] = unsub
	e.mu.Unlock()
	e.log.Info("event rule added", "rule_id", r.ID, "sub", r.Expression)
	return nil
}

// Rules returns a snapshot of the registered event rules, in insertion order.
func (e *Engine) Rules() []Rule {
	e.mu.RLock()
	defer e.mu.RUnlock()
	out := make([]Rule, 0, len(e.rules))
	for _, r := range e.rules {
		out = append(out, r)
	}
	return out
}

// RemoveRule unsubscribes and removes the event rule with the given id. It
// returns an error wrapping ErrNotFound when no such rule exists.
func (e *Engine) RemoveRule(ctx context.Context, id string) error {
	e.mu.Lock()
	_, ok := e.rules[id]
	unsub := e.unsub[id]
	delete(e.rules, id)
	delete(e.unsub, id)
	e.mu.Unlock()
	if !ok {
		return fmt.Errorf("automations: remove rule %q: %w", id, ErrNotFound)
	}
	if unsub != nil {
		unsub()
	}
	e.log.Info("event rule removed", "rule_id", id)
	return nil
}

// Start activates in-flight action cancellation for the engine. It creates the
// cancellation context from ctx; cancelling ctx (or Stop) cancels running actions.
// Start is idempotent.
func (e *Engine) Start(ctx context.Context) error {
	if ctx == nil {
		ctx = context.Background()
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.cancel != nil {
		return nil
	}
	e.ctx, e.cancel = context.WithCancel(ctx)
	e.log.Info("event engine started")
	return nil
}

// Stop cancels running actions, unsubscribes every rule and clears the registry.
// It is idempotent.
func (e *Engine) Stop(ctx context.Context) error {
	e.mu.Lock()
	if e.cancel != nil {
		e.cancel()
		e.cancel = nil
	}
	unsubs := e.unsub
	e.rules = make(map[string]Rule)
	e.unsub = make(map[string]func())
	e.mu.Unlock()
	for _, u := range unsubs {
		if u != nil {
			u()
		}
	}
	e.log.Info("event engine stopped")
	return nil
}

// dispatch runs the action for a matched event. It uses the engine's live context
// so Stop cancels in-flight actions, and it skips delivery when the engine is
// already stopped.
func (e *Engine) dispatch(r Rule, ev bus.Event) {
	ctx := e.liveCtx()
	if err := ctx.Err(); err != nil {
		return
	}
	e.log.Info("event matched", "rule_id", r.ID, "subject", ev.Subject)
	runAction(ctx, e.cfg, e.bus, e.log, "event", r.ID, r.Action, ev.Payload)
}

// liveCtx returns the engine's live context, or Background before Start.
func (e *Engine) liveCtx() context.Context {
	e.mu.RLock()
	defer e.mu.RUnlock()
	if e.ctx != nil {
		return e.ctx
	}
	return context.Background()
}

// checkCtx returns ctx.Err() wrapped when ctx is canceled. A nil ctx is treated as
// Background.
func (e *Engine) checkCtx(ctx context.Context) error {
	if ctx == nil {
		return nil
	}
	if err := ctx.Err(); err != nil {
		return fmt.Errorf("automations: %w", err)
	}
	return nil
}