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 }