feat(automations): 自动化引擎（cron 调度 + 事件触发 + AI 管线动作 + 提醒调度，bus automation.result/reminder.trigger）
This commit is contained in:
Aska committed
1 parent
a3f6f4f687
commit
ff34b62079
10 files changed
+1818
No files matched your search
@@ -0,0 +1,169 @@
|
||||
package automations
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
|
||||
)
|
||||
|
||||
// Action is a unit of work executed when a schedule, event rule or reminder
|
||||
// fires. Execute runs the action with the trigger payload and returns a result
|
||||
// (or an error, wrapped with %w). Implementations must be safe for concurrent use
|
||||
// because the engine may invoke them from multiple goroutines.
|
||||
//
|
||||
// The three concrete implementations are FuncAction (a plain function), ToolAction
|
||||
// (an sdk.Tool) and AIAction (a prompt run through the AI pipeline).
|
||||
type Action interface {
|
||||
// Name identifies the action in logs and in the published Result.
|
||||
Name() string
|
||||
// Execute runs the action with the trigger payload.
|
||||
Execute(ctx context.Context, payload any) (any, error)
|
||||
}
|
||||
|
||||
// Func is a plain function action.
|
||||
type Func func(ctx context.Context, payload any) (any, error)
|
||||
|
||||
// FuncAction adapts a plain function to the Action interface.
|
||||
type FuncAction struct {
|
||||
name string
|
||||
fn Func
|
||||
}
|
||||
|
||||
// NewFuncAction wraps fn as a named action. It returns an error wrapping
|
||||
// ErrNilAction when name is blank or fn is nil.
|
||||
func NewFuncAction(name string, fn Func) (*FuncAction, error) {
|
||||
if strings.TrimSpace(name) == "" {
|
||||
return nil, fmt.Errorf("automations: func action: %w", ErrNilAction)
|
||||
}
|
||||
if fn == nil {
|
||||
return nil, fmt.Errorf("automations: func action %q: %w", name, ErrNilAction)
|
||||
}
|
||||
return &FuncAction{name: name, fn: fn}, nil
|
||||
}
|
||||
|
||||
// Name returns the action name.
|
||||
func (f *FuncAction) Name() string { return f.name }
|
||||
|
||||
// Execute runs the wrapped function with the payload.
|
||||
func (f *FuncAction) Execute(ctx context.Context, payload any) (any, error) {
|
||||
return f.fn(ctx, payload)
|
||||
}
|
||||
|
||||
// ToolAction adapts an sdk.Tool to the Action interface.
|
||||
type ToolAction struct {
|
||||
Tool sdk.Tool
|
||||
}
|
||||
|
||||
// NewToolAction wraps a registered tool as a named action. A nil tool returns an
|
||||
// error wrapping ErrNilAction.
|
||||
func NewToolAction(t sdk.Tool) (*ToolAction, error) {
|
||||
if t == nil {
|
||||
return nil, fmt.Errorf("automations: tool action: %w", ErrNilAction)
|
||||
}
|
||||
return &ToolAction{Tool: t}, nil
|
||||
}
|
||||
|
||||
// Name returns the underlying tool's name.
|
||||
func (t *ToolAction) Name() string { return t.Tool.Name() }
|
||||
|
||||
// Execute delegates to the underlying tool.
|
||||
func (t *ToolAction) Execute(ctx context.Context, payload any) (any, error) {
|
||||
return t.Tool.Execute(ctx, payload)
|
||||
}
|
||||
|
||||
// AIAction drives a prompt through the AI pipeline (loop or agent) and returns
|
||||
// the generated text. sessionID scopes the conversation for tracing/persistence.
|
||||
type AIAction struct {
|
||||
name string
|
||||
prompt string
|
||||
sessionID string
|
||||
pipeline Pipeline
|
||||
}
|
||||
|
||||
// NewAIAction builds a prompting action. name must be non-blank and p must be
|
||||
// non-nil; otherwise an error wrapping ErrNilAction or ErrNoPipeline is returned.
|
||||
func NewAIAction(name, prompt, sessionID string, p Pipeline) (*AIAction, error) {
|
||||
if strings.TrimSpace(name) == "" {
|
||||
return nil, fmt.Errorf("automations: ai action: %w", ErrNilAction)
|
||||
}
|
||||
if p == nil {
|
||||
return nil, fmt.Errorf("automations: ai action %q: %w", name, ErrNoPipeline)
|
||||
}
|
||||
return &AIAction{name: name, prompt: prompt, sessionID: sessionID, pipeline: p}, nil
|
||||
}
|
||||
|
||||
// Name returns the action name.
|
||||
func (a *AIAction) Name() string { return a.name }
|
||||
|
||||
// Execute runs the prompt (with the payload appended) through the pipeline and
|
||||
// returns the generated text as the result.
|
||||
func (a *AIAction) Execute(ctx context.Context, payload any) (any, error) {
|
||||
msgs := []model.Message{model.TextMessage(model.RoleUser, a.promptMessage(payload))}
|
||||
res, err := a.pipeline.Run(ctx, a.sessionID, msgs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("automations: ai action %q: %w", a.name, err)
|
||||
}
|
||||
return res.Text, nil
|
||||
}
|
||||
|
||||
// promptMessage renders the prompt plus a readable form of the trigger payload.
|
||||
func (a *AIAction) promptMessage(payload any) string {
|
||||
if payload == nil {
|
||||
return a.prompt
|
||||
}
|
||||
if s, ok := payload.(string); ok {
|
||||
return a.prompt + "\nTrigger payload: " + s
|
||||
}
|
||||
if b, err := json.Marshal(payload); err == nil {
|
||||
return a.prompt + "\nTrigger payload: " + string(b)
|
||||
}
|
||||
return a.prompt + "\nTrigger payload: " + fmt.Sprint(payload)
|
||||
}
|
||||
|
||||
// runAction executes a with payload under the configured deadline and publishes
|
||||
// the outcome to SubjectResult. It is the single place where automation execution
|
||||
// and result reporting are wired, so every trigger kind reports consistently. A
|
||||
// panic is recovered and recorded as a failed result so a misbehaving action
|
||||
// cannot crash the calling goroutine.
|
||||
func runAction(ctx context.Context, cfg Config, b bus.Bus, log *slog.Logger, kind, ruleID string, a Action, payload any) {
|
||||
if a == nil {
|
||||
return
|
||||
}
|
||||
execCtx := ctx
|
||||
if cfg.ActionTimeout > 0 {
|
||||
var cancel context.CancelFunc
|
||||
execCtx, cancel = context.WithTimeout(ctx, cfg.ActionTimeout)
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
out, err := executeSafely(execCtx, a, payload)
|
||||
res := Result{Kind: kind, RuleID: ruleID, Action: a.Name(), Payload: payload, Output: out}
|
||||
if err != nil {
|
||||
res.Error = err.Error()
|
||||
log.Error("automation action failed", "kind", kind, "rule_id", ruleID, "action", a.Name(), "err", err)
|
||||
} else {
|
||||
log.Info("automation action ran", "kind", kind, "rule_id", ruleID, "action", a.Name())
|
||||
}
|
||||
|
||||
if cerr := b.Publish(ctx, SubjectResult, res); cerr != nil {
|
||||
log.Error("publish automation result", "kind", kind, "rule_id", ruleID, "err", cerr)
|
||||
}
|
||||
}
|
||||
|
||||
// executeSafely runs a.Execute, converting a panic into an error so the action
|
||||
// reports a failed result instead of crashing the engine goroutine.
|
||||
func executeSafely(ctx context.Context, a Action, payload any) (out any, err error) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
out = nil
|
||||
err = fmt.Errorf("automations: action %q panicked: %v", a.Name(), r)
|
||||
}
|
||||
}()
|
||||
return a.Execute(ctx, payload)
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
package automations
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// fakePipeline is an in-memory Pipeline used to observe the prompt an AIAction
|
||||
// sends and to return a canned text. It is the mock for the loop/agent pipeline;
|
||||
// it implements the Pipeline interface, not a concrete loop.
|
||||
type fakePipeline struct {
|
||||
mu sync.Mutex
|
||||
text string
|
||||
err error
|
||||
msgs []model.Message
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakePipeline) Run(ctx context.Context, sessionID string, msgs []model.Message) (*loop.Result, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.calls++
|
||||
f.msgs = append(f.msgs, msgs...)
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return &loop.Result{Text: f.text, StopReason: "stop", Steps: 1}, nil
|
||||
}
|
||||
|
||||
func (f *fakePipeline) lastMessage() model.Message {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if len(f.msgs) == 0 {
|
||||
return model.Message{}
|
||||
}
|
||||
return f.msgs[len(f.msgs)-1]
|
||||
}
|
||||
|
||||
func TestAIAction_RunsPipelineAndPublishesResult(t *testing.T) {
|
||||
t.Parallel()
|
||||
b := bus.New()
|
||||
defer b.Close()
|
||||
|
||||
got := make(chan bus.Event, 1)
|
||||
unsub, err := b.Subscribe(context.Background(), SubjectResult, func(_ context.Context, ev bus.Event) error {
|
||||
got <- ev
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("subscribe: %v", err)
|
||||
}
|
||||
defer unsub()
|
||||
|
||||
p := &fakePipeline{text: "hello from pipeline"}
|
||||
act, err := NewAIAction("gen_greeting", "You are a greeter", "sess-1", p)
|
||||
if err != nil {
|
||||
t.Fatalf("NewAIAction: %v", err)
|
||||
}
|
||||
|
||||
s, err := New(Config{}, Deps{Bus: b})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
if _, err := s.AddSchedule(ctx, "@every 50ms", act); err != nil {
|
||||
t.Fatalf("AddSchedule: %v", err)
|
||||
}
|
||||
if err := s.Start(ctx); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
|
||||
var res Result
|
||||
select {
|
||||
case ev := <-got:
|
||||
v, ok := ev.Payload.(Result)
|
||||
if !ok {
|
||||
t.Fatalf("unexpected payload type: %T", ev.Payload)
|
||||
}
|
||||
res = v
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("AI action did not fire")
|
||||
}
|
||||
|
||||
if res.Kind != "schedule" {
|
||||
t.Errorf("kind = %q, want schedule", res.Kind)
|
||||
}
|
||||
if res.Action != "gen_greeting" {
|
||||
t.Errorf("action = %q, want gen_greeting", res.Action)
|
||||
}
|
||||
if res.Output != "hello from pipeline" {
|
||||
t.Errorf("output = %#v, want hello from pipeline", res.Output)
|
||||
}
|
||||
last := p.lastMessage()
|
||||
if len(last.Content) == 0 || last.Content[0].Text != "You are a greeter" {
|
||||
t.Errorf("prompt = %#v, want the configured prompt", last.Content)
|
||||
}
|
||||
if p.calls < 1 {
|
||||
t.Errorf("pipeline not called")
|
||||
}
|
||||
|
||||
if err := s.Stop(ctx); err != nil {
|
||||
t.Fatalf("Stop: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAIAction_RequiresPipeline(t *testing.T) {
|
||||
t.Parallel()
|
||||
if _, err := NewAIAction("x", "p", "s", nil); err == nil {
|
||||
t.Fatal("expected error for nil pipeline")
|
||||
} else if !errors.Is(err, ErrNoPipeline) {
|
||||
t.Fatalf("err = %v, want ErrNoPipeline", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAIAction_PropagatesPipelineError(t *testing.T) {
|
||||
t.Parallel()
|
||||
p := &fakePipeline{err: context.DeadlineExceeded}
|
||||
act, err := NewAIAction("x", "p", "s", p)
|
||||
if err != nil {
|
||||
t.Fatalf("NewAIAction: %v", err)
|
||||
}
|
||||
if _, err := act.Execute(context.Background(), nil); err == nil {
|
||||
t.Fatal("expected error from pipeline")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,176 @@
|
||||
// Package automations implements the Cyrene v2 automation engine and reminder
|
||||
// scheduler (DevDocs/design/04-运行时.md §8.5). It is the "how she does things on
|
||||
// her own" layer: schedules and event rules fire a named action, the action runs
|
||||
// (either a plain function, a registered tool, or a prompt through the AI
|
||||
// loop/agent pipeline), and every outcome is published back to the bus so upper
|
||||
// layers can react or push notifications.
|
||||
//
|
||||
// The package is split into five files:
|
||||
//
|
||||
// - automations.go — shared types: Config, Deps, the Pipeline abstraction, the
|
||||
// published Result and the sentinel errors.
|
||||
// - scheduler.go — the cron-driven Scheduler (robfig/cron/v3).
|
||||
// - rule.go — the Rule model and the event-driven Engine.
|
||||
// - action.go — the Action interface plus func/tool/AI implementations.
|
||||
// - reminder.go — the time-based ReminderStore.
|
||||
// - service.go — the Service that composes the three for lifecycle control.
|
||||
//
|
||||
// Execution model: every trigger (cron fire, event match, reminder due) resolves
|
||||
// to the same runAction, which executes the Action under an optional deadline and
|
||||
// publishes a Result to the "automation.result" subject. This keeps reporting and
|
||||
// logging consistent across all trigger kinds (DevDocs/design/04-运行时.md §8.5:
|
||||
// task output returns to the message queue/agent).
|
||||
//
|
||||
// Identity and capability boundaries (Devedocs global red lines): automation
|
||||
// never decides who is allowed to do anything. It only fires named actions;
|
||||
// authorization stays in the channel/policy layer.
|
||||
package automations
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// Bus subjects published by the automations engine. Upper layers subscribe to
|
||||
// these to observe outcomes or push notifications.
|
||||
const (
|
||||
// SubjectResult is the subject every executed automation publishes its
|
||||
// Result to.
|
||||
SubjectResult = "automation.result"
|
||||
// SubjectTrigger is the subject a due Reminder is published to. Upper layers
|
||||
// subscribe to it to surface the reminder to the user.
|
||||
SubjectTrigger = "reminder.trigger"
|
||||
)
|
||||
|
||||
// Sentinel errors returned by this package. Use errors.Is to test for them.
|
||||
var (
|
||||
// ErrNilBus is returned when a constructor receives a nil bus.
|
||||
ErrNilBus = errors.New("automations: nil bus")
|
||||
// ErrNilAction is returned when an action is nil or unnamed.
|
||||
ErrNilAction = errors.New("automations: nil action")
|
||||
// ErrInvalidRule is returned when a rule is malformed.
|
||||
ErrInvalidRule = errors.New("automations: invalid rule")
|
||||
// ErrInvalidCron is returned when a cron expression fails to parse.
|
||||
ErrInvalidCron = errors.New("automations: invalid cron expression")
|
||||
// ErrNotFound is returned when a rule or reminder is missing.
|
||||
ErrNotFound = errors.New("automations: not found")
|
||||
// ErrDuplicate is returned when a rule or reminder id is already registered.
|
||||
ErrDuplicate = errors.New("automations: already exists")
|
||||
// ErrNoPipeline is returned when an AI action is built without a pipeline.
|
||||
ErrNoPipeline = errors.New("automations: no AI pipeline configured")
|
||||
)
|
||||
|
||||
// Config configures the automations engine.
|
||||
type Config struct {
|
||||
// Location is the timezone used for cron schedules and reminder firing. Nil
|
||||
// uses time.Local.
|
||||
Location *time.Location
|
||||
// WithSeconds enables 6-field cron specs (with a leading seconds field) in
|
||||
// addition to the standard 5-field format.
|
||||
WithSeconds bool
|
||||
// ActionTimeout bounds a single action execution. Non-positive disables the
|
||||
// deadline.
|
||||
ActionTimeout time.Duration
|
||||
// ReminderInterval is how often the reminder loop polls for due reminders.
|
||||
// Non-positive selects defaultReminderInterval.
|
||||
ReminderInterval time.Duration
|
||||
}
|
||||
|
||||
// Deps is the injected dependency set for the automations engine.
|
||||
type Deps struct {
|
||||
// Bus is the in-process event bus. Required.
|
||||
Bus bus.Bus
|
||||
// Loop is the AI pipeline (the agent loop). When set, an AI action runs
|
||||
// through it. Agent is used only when Loop is nil.
|
||||
Loop *loop.Loop
|
||||
// Agent is the single-step agent, used as the AI pipeline when Loop is nil.
|
||||
Agent *agent.Agent
|
||||
// Log is the base logger. A nil logger is replaced with a discard logger
|
||||
// tagged with source "services/automations".
|
||||
Log *slog.Logger
|
||||
// Now supplies the current time for reminders. Nil uses time.Now.
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
// Pipeline is the AI-execution surface an automation action runs through. It is
|
||||
// satisfied by *loop.Loop directly, so a real pipeline can be injected without
|
||||
// an adapter. Tests inject a fake to observe the prompt and return canned text.
|
||||
type Pipeline interface {
|
||||
Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error)
|
||||
}
|
||||
|
||||
// AgentPipeline adapts an *agent.Agent to Pipeline by running a single step and
|
||||
// returning its text as the result. It is used when only an agent (not a full
|
||||
// loop) is injected.
|
||||
type AgentPipeline struct {
|
||||
Agent *agent.Agent
|
||||
}
|
||||
|
||||
// Run runs one agent step and returns its text as the loop-style Result.
|
||||
func (a *AgentPipeline) Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error) {
|
||||
if a.Agent == nil {
|
||||
return nil, fmt.Errorf("automations: agent pipeline: %w", ErrNoPipeline)
|
||||
}
|
||||
sr, err := a.Agent.RunStep(ctx, sessionID, messages)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("automations: agent pipeline run: %w", err)
|
||||
}
|
||||
return &loop.Result{Text: sr.Text, StopReason: sr.StopReason, Steps: 1}, nil
|
||||
}
|
||||
|
||||
// pipeline returns the resolved AI pipeline from deps, or nil when neither a
|
||||
// loop nor an agent is configured.
|
||||
func (d Deps) pipeline() Pipeline {
|
||||
if d.Loop != nil {
|
||||
return d.Loop
|
||||
}
|
||||
if d.Agent != nil {
|
||||
return &AgentPipeline{Agent: d.Agent}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Result is the structured outcome of an executed automation, published to
|
||||
// SubjectResult. Upper layers subscribe to it to react or push notifications.
|
||||
type Result struct {
|
||||
// Kind classifies the trigger: "schedule", "event" or "reminder".
|
||||
Kind string `json:"kind"`
|
||||
// RuleID identifies the schedule/rule/reminder that fired.
|
||||
RuleID string `json:"rule_id"`
|
||||
// Action is the action name that ran.
|
||||
Action string `json:"action"`
|
||||
// Payload is the trigger payload (nil for a cron fire).
|
||||
Payload any `json:"payload"`
|
||||
// Output is the action's returned value, or nil on error.
|
||||
Output any `json:"output"`
|
||||
// Error carries the action's error message when the run failed.
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// sourcedLog returns a base logger tagged with the module source
|
||||
// "services/automations", replacing a nil logger with a discard one so callers
|
||||
// can always log safely.
|
||||
func sourcedLog(l *slog.Logger) *slog.Logger {
|
||||
if l == nil {
|
||||
l = slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
}
|
||||
return logger.WithSource(l, "services/automations")
|
||||
}
|
||||
|
||||
// resolveNow returns f when non-nil, otherwise time.Now.
|
||||
func resolveNow(f func() time.Time) func() time.Time {
|
||||
if f == nil {
|
||||
return time.Now
|
||||
}
|
||||
return f
|
||||
}
|
||||
@@ -0,0 +1,211 @@
|
||||
package automations
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
)
|
||||
|
||||
const (
|
||||
// defaultReminderInterval is how often the reminder loop polls for due
|
||||
// reminders when Config.ReminderInterval is non-positive.
|
||||
defaultReminderInterval = 500 * time.Millisecond
|
||||
// reminderStopWait bounds how long Stop waits for the reminder loop to exit.
|
||||
reminderStopWait = 5 * time.Second
|
||||
)
|
||||
|
||||
// Reminder is a single delayed notification.
|
||||
type Reminder struct {
|
||||
// ID uniquely identifies the reminder.
|
||||
ID string
|
||||
// At is the time the reminder should fire.
|
||||
At time.Time
|
||||
// Message is the payload surfaced to the user when it fires.
|
||||
Message string
|
||||
}
|
||||
|
||||
// ReminderStore schedules delayed notifications and publishes each due reminder
|
||||
// to SubjectTrigger so an upper layer can push it to the user. It is safe for
|
||||
// concurrent use.
|
||||
type ReminderStore struct {
|
||||
cfg Config
|
||||
bus bus.Bus
|
||||
log *slog.Logger
|
||||
now func() time.Time
|
||||
interval time.Duration
|
||||
|
||||
mu sync.RWMutex
|
||||
items map[string]Reminder
|
||||
cancel context.CancelFunc
|
||||
done chan struct{}
|
||||
started bool
|
||||
}
|
||||
|
||||
// NewReminder builds a ReminderStore. The bus in deps is required; a nil bus
|
||||
// returns an error wrapping ErrNilBus. cfg.ReminderInterval defaults to
|
||||
// defaultReminderInterval.
|
||||
func NewReminder(cfg Config, deps Deps) (*ReminderStore, error) {
|
||||
if deps.Bus == nil {
|
||||
return nil, fmt.Errorf("automations: new reminder: %w", ErrNilBus)
|
||||
}
|
||||
interval := cfg.ReminderInterval
|
||||
if interval <= 0 {
|
||||
interval = defaultReminderInterval
|
||||
}
|
||||
return &ReminderStore{
|
||||
cfg: cfg,
|
||||
bus: deps.Bus,
|
||||
log: sourcedLog(deps.Log),
|
||||
now: resolveNow(deps.Now),
|
||||
interval: interval,
|
||||
items: make(map[string]Reminder),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Add registers a reminder. It returns an error wrapping ErrInvalidRule when the
|
||||
// id is blank or At is zero, and ErrDuplicate when the id is already registered.
|
||||
func (rs *ReminderStore) Add(ctx context.Context, r Reminder) error {
|
||||
if r.ID == "" || r.At.IsZero() {
|
||||
return fmt.Errorf("automations: add reminder: %w", ErrInvalidRule)
|
||||
}
|
||||
rs.mu.Lock()
|
||||
defer rs.mu.Unlock()
|
||||
if _, exists := rs.items[r.ID]; exists {
|
||||
return fmt.Errorf("automations: add reminder %q: %w", r.ID, ErrDuplicate)
|
||||
}
|
||||
rs.items[r.ID] = r
|
||||
rs.log.Info("reminder added", "id", r.ID, "at", r.At)
|
||||
return nil
|
||||
}
|
||||
|
||||
// List returns a snapshot of the registered reminders, sorted by At ascending.
|
||||
func (rs *ReminderStore) List() []Reminder {
|
||||
rs.mu.RLock()
|
||||
defer rs.mu.RUnlock()
|
||||
out := make([]Reminder, 0, len(rs.items))
|
||||
for _, r := range rs.items {
|
||||
out = append(out, r)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].At.Before(out[j].At) })
|
||||
return out
|
||||
}
|
||||
|
||||
// Cancel removes a reminder by id. A blank id or a missing id returns an error
|
||||
// wrapping ErrInvalidRule or ErrNotFound respectively.
|
||||
func (rs *ReminderStore) Cancel(ctx context.Context, id string) error {
|
||||
if id == "" {
|
||||
return fmt.Errorf("automations: cancel reminder: %w", ErrInvalidRule)
|
||||
}
|
||||
rs.mu.Lock()
|
||||
defer rs.mu.Unlock()
|
||||
if _, exists := rs.items[id]; !exists {
|
||||
return fmt.Errorf("automations: cancel reminder %q: %w", id, ErrNotFound)
|
||||
}
|
||||
delete(rs.items, id)
|
||||
rs.log.Info("reminder canceled", "id", id)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Start launches the polling loop that fires due reminders. Start is idempotent;
|
||||
// it derives the loop's cancellation from ctx.
|
||||
func (rs *ReminderStore) Start(ctx context.Context) error {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
rs.mu.Lock()
|
||||
if rs.started {
|
||||
rs.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
loopCtx, cancel := context.WithCancel(ctx)
|
||||
rs.cancel = cancel
|
||||
rs.done = make(chan struct{})
|
||||
rs.started = true
|
||||
rs.mu.Unlock()
|
||||
go rs.loop(loopCtx)
|
||||
rs.log.Info("reminder loop started")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop stops the polling loop and waits (bounded by reminderStopWait) for it to
|
||||
// exit. It is idempotent.
|
||||
func (rs *ReminderStore) Stop(ctx context.Context) error {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
rs.mu.Lock()
|
||||
if !rs.started {
|
||||
rs.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
rs.started = false
|
||||
cancel := rs.cancel
|
||||
rs.cancel = nil
|
||||
done := rs.done
|
||||
rs.mu.Unlock()
|
||||
|
||||
if cancel != nil {
|
||||
cancel()
|
||||
}
|
||||
if done != nil {
|
||||
select {
|
||||
case <-done:
|
||||
case <-ctx.Done():
|
||||
return fmt.Errorf("automations: reminder stop: %w", ctx.Err())
|
||||
case <-time.After(reminderStopWait):
|
||||
rs.log.Warn("reminder stop timed out")
|
||||
}
|
||||
}
|
||||
rs.log.Info("reminder loop stopped")
|
||||
return nil
|
||||
}
|
||||
|
||||
// loop is the polling goroutine. It owns the done channel and closes it on exit.
|
||||
func (rs *ReminderStore) loop(ctx context.Context) {
|
||||
defer func() {
|
||||
rs.mu.Lock()
|
||||
if rs.done != nil {
|
||||
close(rs.done)
|
||||
rs.done = nil
|
||||
}
|
||||
rs.mu.Unlock()
|
||||
}()
|
||||
ticker := time.NewTicker(rs.interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
rs.fireDue(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// fireDue publishes every reminder whose At is at or before now and removes it
|
||||
// from the store. Publishing happens outside the lock so the loop does not hold
|
||||
// the mutex across the bus fan-out.
|
||||
func (rs *ReminderStore) fireDue(ctx context.Context) {
|
||||
now := rs.now()
|
||||
var due []Reminder
|
||||
rs.mu.Lock()
|
||||
for id, r := range rs.items {
|
||||
if !now.Before(r.At) {
|
||||
due = append(due, r)
|
||||
delete(rs.items, id)
|
||||
}
|
||||
}
|
||||
rs.mu.Unlock()
|
||||
for _, r := range due {
|
||||
if err := rs.bus.Publish(ctx, SubjectTrigger, r); err != nil {
|
||||
rs.log.Error("publish reminder", "id", r.ID, "err", err)
|
||||
continue
|
||||
}
|
||||
rs.log.Info("reminder fired", "id", r.ID, "at", r.At, "message", r.Message)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
package automations
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
)
|
||||
|
||||
func TestReminder_AddListCancel(t *testing.T) {
|
||||
t.Parallel()
|
||||
rs, err := NewReminder(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("NewReminder: %v", err)
|
||||
}
|
||||
base := time.Now().Add(time.Hour)
|
||||
r1 := Reminder{ID: "later", At: base.Add(time.Hour), Message: "b"}
|
||||
r2 := Reminder{ID: "sooner", At: base, Message: "a"}
|
||||
|
||||
if err := rs.Add(context.Background(), r1); err != nil {
|
||||
t.Fatalf("Add r1: %v", err)
|
||||
}
|
||||
if err := rs.Add(context.Background(), r2); err != nil {
|
||||
t.Fatalf("Add r2: %v", err)
|
||||
}
|
||||
if err := rs.Add(context.Background(), r2); !errors.Is(err, ErrDuplicate) {
|
||||
t.Fatalf("duplicate: err = %v, want ErrDuplicate", err)
|
||||
}
|
||||
|
||||
list := rs.List()
|
||||
if len(list) != 2 {
|
||||
t.Fatalf("len(List) = %d, want 2", len(list))
|
||||
}
|
||||
// Sorted by At ascending.
|
||||
if list[0].ID != "sooner" || list[1].ID != "later" {
|
||||
t.Fatalf("not sorted by At: %+v", list)
|
||||
}
|
||||
|
||||
if err := rs.Cancel(context.Background(), "sooner"); err != nil {
|
||||
t.Fatalf("Cancel: %v", err)
|
||||
}
|
||||
if got := len(rs.List()); got != 1 {
|
||||
t.Fatalf("len(List) after cancel = %d, want 1", got)
|
||||
}
|
||||
if err := rs.Cancel(context.Background(), "sooner"); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("cancel missing: err = %v, want ErrNotFound", err)
|
||||
}
|
||||
if err := rs.Add(context.Background(), Reminder{ID: "", At: base}); !errors.Is(err, ErrInvalidRule) {
|
||||
t.Fatalf("empty id: err = %v, want ErrInvalidRule", err)
|
||||
}
|
||||
if err := rs.Add(context.Background(), Reminder{ID: "x"}); !errors.Is(err, ErrInvalidRule) {
|
||||
t.Fatalf("zero At: err = %v, want ErrInvalidRule", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReminder_FiresOnDue(t *testing.T) {
|
||||
t.Parallel()
|
||||
b := bus.New()
|
||||
defer b.Close()
|
||||
|
||||
got := make(chan bus.Event, 1)
|
||||
unsub, err := b.Subscribe(context.Background(), SubjectTrigger, func(_ context.Context, ev bus.Event) error {
|
||||
got <- ev
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("subscribe: %v", err)
|
||||
}
|
||||
defer unsub()
|
||||
|
||||
rs, err := NewReminder(Config{ReminderInterval: 10 * time.Millisecond}, Deps{Bus: b})
|
||||
if err != nil {
|
||||
t.Fatalf("NewReminder: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
r := Reminder{ID: "wake", At: time.Now().Add(20 * time.Millisecond), Message: "time to focus"}
|
||||
if err := rs.Add(ctx, r); err != nil {
|
||||
t.Fatalf("Add: %v", err)
|
||||
}
|
||||
if err := rs.Start(ctx); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case ev := <-got:
|
||||
rem, ok := ev.Payload.(Reminder)
|
||||
if !ok {
|
||||
t.Fatalf("payload type = %T, want Reminder", ev.Payload)
|
||||
}
|
||||
if rem.ID != "wake" || rem.Message != "time to focus" {
|
||||
t.Fatalf("reminder = %+v", rem)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("reminder did not fire")
|
||||
}
|
||||
if err := rs.Stop(ctx); err != nil {
|
||||
t.Fatalf("Stop: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReminder_StartStopIdempotent(t *testing.T) {
|
||||
t.Parallel()
|
||||
rs, err := NewReminder(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("NewReminder: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
if err := rs.Start(ctx); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
if err := rs.Start(ctx); err != nil {
|
||||
t.Fatalf("second Start: %v", err)
|
||||
}
|
||||
if err := rs.Stop(ctx); err != nil {
|
||||
t.Fatalf("Stop: %v", err)
|
||||
}
|
||||
if err := rs.Stop(ctx); err != nil {
|
||||
t.Fatalf("second Stop: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReminder_ConcurrentAddCancelList(t *testing.T) {
|
||||
t.Parallel()
|
||||
rs, err := NewReminder(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("NewReminder: %v", err)
|
||||
}
|
||||
const n = 50
|
||||
done := make(chan struct{})
|
||||
for i := 0; i < n; i++ {
|
||||
go func(i int) {
|
||||
defer func() { done <- struct{}{} }()
|
||||
id := "r"
|
||||
_ = rs.Add(context.Background(), Reminder{ID: id, At: time.Now().Add(time.Hour), Message: "m"})
|
||||
_ = rs.List()
|
||||
_ = rs.Cancel(context.Background(), id)
|
||||
}(i)
|
||||
}
|
||||
for i := 0; i < n; i++ {
|
||||
<-done
|
||||
}
|
||||
}
|
||||
|
||||
func TestService_ReminderLifecycle(t *testing.T) {
|
||||
t.Parallel()
|
||||
b := bus.New()
|
||||
defer b.Close()
|
||||
s, err := NewService(Config{}, Deps{Bus: b})
|
||||
if err != nil {
|
||||
t.Fatalf("NewService: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
if err := s.AddReminder(ctx, Reminder{ID: "x", At: time.Now().Add(time.Hour), Message: "m"}); err != nil {
|
||||
t.Fatalf("AddReminder: %v", err)
|
||||
}
|
||||
if got := len(s.Reminders()); got != 1 {
|
||||
t.Fatalf("len(Reminders) = %d, want 1", got)
|
||||
}
|
||||
if err := s.CancelReminder(ctx, "x"); err != nil {
|
||||
t.Fatalf("CancelReminder: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,206 @@
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,233 @@
|
||||
package automations
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
)
|
||||
|
||||
func TestEngine_EventMatchFiresOnlyMatchingRule(t *testing.T) {
|
||||
t.Parallel()
|
||||
b := bus.New()
|
||||
defer b.Close()
|
||||
eng, err := NewEngine(Config{}, Deps{Bus: b})
|
||||
if err != nil {
|
||||
t.Fatalf("NewEngine: %v", err)
|
||||
}
|
||||
|
||||
var ranA, ranB atomic.Int32
|
||||
actA, err := NewFuncAction("on_device", func(context.Context, any) (any, error) {
|
||||
ranA.Add(1)
|
||||
return nil, nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction A: %v", err)
|
||||
}
|
||||
actB, err := NewFuncAction("on_temp", func(context.Context, any) (any, error) {
|
||||
ranB.Add(1)
|
||||
return nil, nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction B: %v", err)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
if err := eng.AddEvent(ctx, Rule{ID: "r1", Trigger: TriggerEvent, Expression: "iot.device_update", Action: actA}); err != nil {
|
||||
t.Fatalf("AddEvent r1: %v", err)
|
||||
}
|
||||
if err := eng.AddEvent(ctx, Rule{ID: "r2", Trigger: TriggerEvent, Expression: "iot.room_temperature", Action: actB}); err != nil {
|
||||
t.Fatalf("AddEvent r2: %v", err)
|
||||
}
|
||||
if err := eng.Start(ctx); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
|
||||
if err := b.Publish(ctx, "iot.device_update", map[string]any{"value": 42}); err != nil {
|
||||
t.Fatalf("Publish: %v", err)
|
||||
}
|
||||
|
||||
waitFor(t, 2*time.Second, func() bool { return ranA.Load() > 0 })
|
||||
if ranA.Load() != 1 {
|
||||
t.Errorf("matching rule ran %d times, want 1", ranA.Load())
|
||||
}
|
||||
if ranB.Load() != 0 {
|
||||
t.Errorf("non-matching rule ran %d times, want 0", ranB.Load())
|
||||
}
|
||||
if err := eng.Stop(ctx); err != nil {
|
||||
t.Fatalf("Stop: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngine_AddEventValidation(t *testing.T) {
|
||||
t.Parallel()
|
||||
eng, err := NewEngine(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("NewEngine: %v", err)
|
||||
}
|
||||
act, err := NewFuncAction("x", func(context.Context, any) (any, error) { return nil, nil })
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction: %v", err)
|
||||
}
|
||||
|
||||
// Wrong trigger for an event rule.
|
||||
err = eng.AddEvent(context.Background(), Rule{ID: "c", Trigger: TriggerCron, Expression: "*/5 * * * *", Action: act})
|
||||
if !errors.Is(err, ErrInvalidRule) {
|
||||
t.Fatalf("wrong trigger: err = %v, want ErrInvalidRule", err)
|
||||
}
|
||||
// Missing id.
|
||||
err = eng.AddEvent(context.Background(), Rule{ID: "", Trigger: TriggerEvent, Expression: "iot.*", Action: act})
|
||||
if !errors.Is(err, ErrInvalidRule) {
|
||||
t.Fatalf("missing id: err = %v, want ErrInvalidRule", err)
|
||||
}
|
||||
// Nil action.
|
||||
err = eng.AddEvent(context.Background(), Rule{ID: "ok", Trigger: TriggerEvent, Expression: "iot.*"})
|
||||
if !errors.Is(err, ErrInvalidRule) {
|
||||
t.Fatalf("nil action: err = %v, want ErrInvalidRule", err)
|
||||
}
|
||||
// Duplicate.
|
||||
if err := eng.AddEvent(context.Background(), Rule{ID: "d", Trigger: TriggerEvent, Expression: "iot.*", Action: act}); err != nil {
|
||||
t.Fatalf("AddEvent d: %v", err)
|
||||
}
|
||||
err = eng.AddEvent(context.Background(), Rule{ID: "d", Trigger: TriggerEvent, Expression: "iot.*", Action: act})
|
||||
if !errors.Is(err, ErrDuplicate) {
|
||||
t.Fatalf("duplicate: err = %v, want ErrDuplicate", err)
|
||||
}
|
||||
// Missing removal.
|
||||
err = eng.RemoveRule(context.Background(), "nope")
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("remove missing: err = %v, want ErrNotFound", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngine_RemoveRuleStopsFiring(t *testing.T) {
|
||||
t.Parallel()
|
||||
b := bus.New()
|
||||
defer b.Close()
|
||||
eng, err := NewEngine(Config{}, Deps{Bus: b})
|
||||
if err != nil {
|
||||
t.Fatalf("NewEngine: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
var ran atomic.Int32
|
||||
act, err := NewFuncAction("x", func(context.Context, any) (any, error) { ran.Add(1); return nil, nil })
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction: %v", err)
|
||||
}
|
||||
if err := eng.AddEvent(ctx, Rule{ID: "r", Trigger: TriggerEvent, Expression: "iot.device_update", Action: act}); err != nil {
|
||||
t.Fatalf("AddEvent: %v", err)
|
||||
}
|
||||
if err := eng.Start(ctx); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
if err := b.Publish(ctx, "iot.device_update", nil); err != nil {
|
||||
t.Fatalf("Publish: %v", err)
|
||||
}
|
||||
waitFor(t, 2*time.Second, func() bool { return ran.Load() > 0 })
|
||||
|
||||
if err := eng.RemoveRule(ctx, "r"); err != nil {
|
||||
t.Fatalf("RemoveRule: %v", err)
|
||||
}
|
||||
// Give a removed subscription a chance to (incorrectly) receive the next event.
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
before := ran.Load()
|
||||
if err := b.Publish(ctx, "iot.device_update", nil); err != nil {
|
||||
t.Fatalf("Publish 2: %v", err)
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if got := ran.Load(); got != before {
|
||||
t.Fatalf("rule fired after removal: %d -> %d", before, got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngine_RulesSnapshot(t *testing.T) {
|
||||
t.Parallel()
|
||||
eng, err := NewEngine(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("NewEngine: %v", err)
|
||||
}
|
||||
act, err := NewFuncAction("x", func(context.Context, any) (any, error) { return nil, nil })
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction: %v", err)
|
||||
}
|
||||
if err := eng.AddEvent(context.Background(), Rule{ID: "a", Trigger: TriggerEvent, Expression: "iot.*", Action: act}); err != nil {
|
||||
t.Fatalf("AddEvent a: %v", err)
|
||||
}
|
||||
if err := eng.AddEvent(context.Background(), Rule{ID: "b", Trigger: TriggerEvent, Expression: "iot.temp", Action: act}); err != nil {
|
||||
t.Fatalf("AddEvent b: %v", err)
|
||||
}
|
||||
rules := eng.Rules()
|
||||
if len(rules) != 2 {
|
||||
t.Fatalf("len(Rules) = %d, want 2", len(rules))
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngine_ConcurrentAddRemoveAndPublish(t *testing.T) {
|
||||
t.Parallel()
|
||||
b := bus.New()
|
||||
defer b.Close()
|
||||
eng, err := NewEngine(Config{}, Deps{Bus: b})
|
||||
if err != nil {
|
||||
t.Fatalf("NewEngine: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
act, err := NewFuncAction("x", func(context.Context, any) (any, error) { return nil, nil })
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction: %v", err)
|
||||
}
|
||||
|
||||
const n = 50
|
||||
done := make(chan struct{})
|
||||
for i := 0; i < n; i++ {
|
||||
go func(i int) {
|
||||
defer func() { done <- struct{}{} }()
|
||||
id := "r"
|
||||
_ = eng.AddEvent(ctx, Rule{ID: id, Trigger: TriggerEvent, Expression: "iot.*", Action: act})
|
||||
_ = b.Publish(ctx, "iot.device_update", i)
|
||||
_ = eng.RemoveRule(ctx, id)
|
||||
}(i)
|
||||
}
|
||||
for i := 0; i < n; i++ {
|
||||
<-done
|
||||
}
|
||||
}
|
||||
|
||||
func TestService_AddRuleRoutesCronAndRejectsInvalid(t *testing.T) {
|
||||
t.Parallel()
|
||||
s, err := NewService(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("NewService: %v", err)
|
||||
}
|
||||
act, err := NewFuncAction("cron", func(context.Context, any) (any, error) { return nil, nil })
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction: %v", err)
|
||||
}
|
||||
if err := s.AddRule(context.Background(), Rule{ID: "c1", Trigger: TriggerCron, Expression: "*/5 * * * *", Action: act}); err != nil {
|
||||
t.Fatalf("valid cron rule: %v", err)
|
||||
}
|
||||
err = s.AddRule(context.Background(), Rule{ID: "c2", Trigger: TriggerCron, Expression: "not a cron", Action: act})
|
||||
if !errors.Is(err, ErrInvalidCron) {
|
||||
t.Fatalf("invalid cron rule: err = %v, want ErrInvalidCron", err)
|
||||
}
|
||||
if err := s.RemoveRule(context.Background(), "c1"); err != nil {
|
||||
t.Fatalf("RemoveRule cron: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func waitFor(t *testing.T, timeout time.Duration, cond func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
if cond() {
|
||||
return
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,208 @@
|
||||
package automations
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
"github.com/robfig/cron/v3"
|
||||
)
|
||||
|
||||
// schedulerStopWait bounds how long Stop waits for in-flight schedule jobs to
|
||||
// finish before giving up and returning. It prevents a blocked job from leaking
|
||||
// the Stop caller.
|
||||
const schedulerStopWait = 5 * time.Second
|
||||
|
||||
// ScheduleID aliases the cron entry id so callers only need this package's type.
|
||||
type ScheduleID = cron.EntryID
|
||||
|
||||
// parserFor builds the cron spec parser matching cfg. When WithSeconds is set the
|
||||
// parser accepts a leading seconds field (6-field specs); otherwise it accepts
|
||||
// the standard 5-field format. Both accept descriptors such as "@every".
|
||||
func parserFor(cfg Config) cron.Parser {
|
||||
if cfg.WithSeconds {
|
||||
return cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
|
||||
}
|
||||
return cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
|
||||
}
|
||||
|
||||
// cronLogger adapts an *slog.Logger to the cron.Logger interface so cron's own
|
||||
// operational messages carry the "services/automations" source tag.
|
||||
type cronLogger struct {
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
func (c cronLogger) Info(msg string, keysAndValues ...any) { c.log.Info(msg, keysAndValues...) }
|
||||
|
||||
func (c cronLogger) Error(err error, msg string, keysAndValues ...any) {
|
||||
c.log.Error(msg, append([]any{"error", err}, keysAndValues...)...)
|
||||
}
|
||||
|
||||
// Scheduler is the cron-driven automation scheduler. It wraps a robfig/cron/v3
|
||||
// scheduler and runs a registered Action each time its cron expression fires,
|
||||
// publishing the outcome to the bus. It is safe for concurrent use.
|
||||
type Scheduler struct {
|
||||
cfg Config
|
||||
bus bus.Bus
|
||||
log *slog.Logger
|
||||
cron *cron.Cron
|
||||
|
||||
mu sync.RWMutex
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
started bool
|
||||
}
|
||||
|
||||
// New builds a Scheduler. The bus in deps is required; a nil bus returns an error
|
||||
// wrapping ErrNilBus. cfg.Location defaults to time.Local.
|
||||
func New(cfg Config, deps Deps) (*Scheduler, error) {
|
||||
if deps.Bus == nil {
|
||||
return nil, fmt.Errorf("automations: new scheduler: %w", ErrNilBus)
|
||||
}
|
||||
loc := cfg.Location
|
||||
if loc == nil {
|
||||
loc = time.Local
|
||||
}
|
||||
log := sourcedLog(deps.Log)
|
||||
s := &Scheduler{
|
||||
cfg: cfg,
|
||||
bus: deps.Bus,
|
||||
log: log,
|
||||
cron: cron.New(
|
||||
cron.WithLocation(loc),
|
||||
cron.WithParser(parserFor(cfg)),
|
||||
cron.WithLogger(cronLogger{log: log}),
|
||||
),
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// ValidateCron reports whether expr is a valid cron expression for this
|
||||
// scheduler's configured parser. A parse failure is wrapped with ErrInvalidCron.
|
||||
func (s *Scheduler) ValidateCron(expr string) error {
|
||||
if _, err := parserFor(s.cfg).Parse(expr); err != nil {
|
||||
return fmt.Errorf("automations: invalid cron %q: %w: %v", expr, ErrInvalidCron, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// AddSchedule registers action to run whenever cronExpr fires, returning its
|
||||
// schedule id. The cron expression is validated against the scheduler's parser;
|
||||
// an invalid expression returns an error wrapping ErrInvalidCron. ctx is used
|
||||
// only to bound this call; it does not govern later fires which use the
|
||||
// scheduler's own live context.
|
||||
func (s *Scheduler) AddSchedule(ctx context.Context, cronExpr string, action Action) (ScheduleID, error) {
|
||||
if action == nil {
|
||||
return 0, fmt.Errorf("automations: add schedule: %w", ErrNilAction)
|
||||
}
|
||||
if _, err := parserFor(s.cfg).Parse(cronExpr); err != nil {
|
||||
return 0, fmt.Errorf("automations: add schedule %q: %w: %v", cronExpr, ErrInvalidCron, err)
|
||||
}
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return 0, fmt.Errorf("automations: add schedule %q: %w", cronExpr, err)
|
||||
}
|
||||
id, err := s.cron.AddFunc(cronExpr, func() { s.fire(cronExpr, action) })
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("automations: add schedule %q: %w: %v", cronExpr, ErrInvalidCron, err)
|
||||
}
|
||||
s.log.Info("schedule added", "expr", cronExpr, "action", action.Name(), "id", id)
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// RemoveSchedule removes a previously added schedule by id. It is safe to call on
|
||||
// an unknown id.
|
||||
func (s *Scheduler) RemoveSchedule(id ScheduleID) {
|
||||
s.cron.Remove(id)
|
||||
s.log.Info("schedule removed", "id", id)
|
||||
}
|
||||
|
||||
// Entries returns a snapshot of the registered schedules, ordered by the next run
|
||||
// time. It is only populated once the scheduler has been started.
|
||||
func (s *Scheduler) Entries() []cron.Entry {
|
||||
return s.cron.Entries()
|
||||
}
|
||||
|
||||
// Entry returns the schedule registered under id, or the zero Entry when the id
|
||||
// is unknown.
|
||||
func (s *Scheduler) Entry(id ScheduleID) cron.Entry {
|
||||
return s.cron.Entry(id)
|
||||
}
|
||||
|
||||
// Start begins firing registered schedules. Start derives the scheduler's live
|
||||
// context from ctx; cancelling ctx (or calling Stop) stops the scheduler and
|
||||
// cancels in-flight jobs. Start is idempotent.
|
||||
func (s *Scheduler) Start(ctx context.Context) error {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
s.mu.Lock()
|
||||
if s.started {
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
s.ctx, s.cancel = context.WithCancel(ctx)
|
||||
s.started = true
|
||||
s.mu.Unlock()
|
||||
s.cron.Start()
|
||||
s.log.Info("scheduler started")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop stops the scheduler, cancels in-flight jobs and waits (bounded by
|
||||
// schedulerStopWait) for them to finish. It is idempotent.
|
||||
func (s *Scheduler) Stop(ctx context.Context) error {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
s.mu.Lock()
|
||||
if !s.started {
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
s.started = false
|
||||
if s.cancel != nil {
|
||||
s.cancel()
|
||||
s.cancel = nil
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
done := s.cron.Stop()
|
||||
select {
|
||||
case <-done.Done():
|
||||
s.log.Info("scheduler stopped")
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return fmt.Errorf("automations: scheduler stop: %w", ctx.Err())
|
||||
case <-time.After(schedulerStopWait):
|
||||
s.log.Warn("scheduler stop timed out")
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// liveCtx returns the scheduler's live context, or Background before Start.
|
||||
func (s *Scheduler) liveCtx() context.Context {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
if s.ctx != nil {
|
||||
return s.ctx
|
||||
}
|
||||
return context.Background()
|
||||
}
|
||||
|
||||
// fire runs one scheduled action. It recovers panics so a misbehaving action
|
||||
// cannot take down the cron scheduler goroutine, and reports its outcome to the
|
||||
// bus through runAction.
|
||||
func (s *Scheduler) fire(cronExpr string, action Action) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
s.log.Error("schedule job panicked", "expr", cronExpr, "panic", r)
|
||||
}
|
||||
}()
|
||||
runAction(s.liveCtx(), s.cfg, s.bus, s.log, "schedule", cronExpr, action, nil)
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
package automations
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
)
|
||||
|
||||
func TestValidateCron(t *testing.T) {
|
||||
t.Parallel()
|
||||
s, err := New(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
expr string
|
||||
cfg Config
|
||||
want bool
|
||||
}{
|
||||
{name: "standard 5-field", expr: "*/5 * * * *", want: true},
|
||||
{name: "standard 5-field with names", expr: "0 9 * * MON", want: true},
|
||||
{name: "descriptor every", expr: "@every 1h30m", want: true},
|
||||
{name: "descriptor midnight", expr: "@midnight", want: true},
|
||||
{name: "6-field rejected by default", expr: "* * * * * *", want: false},
|
||||
{name: "6-field accepted with seconds", expr: "*/10 * * * * *", cfg: Config{WithSeconds: true}, want: true},
|
||||
{name: "garbage rejected", expr: "not a cron", want: false},
|
||||
{name: "empty rejected", expr: "", want: false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
ss := s
|
||||
if tt.cfg.WithSeconds {
|
||||
ss, err = New(tt.cfg, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
}
|
||||
err := ss.ValidateCron(tt.expr)
|
||||
if got := err == nil; got != tt.want {
|
||||
t.Fatalf("ValidateCron(%q) err = %v, wantValid=%v", tt.expr, err, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAddSchedule_InvalidCronRejected(t *testing.T) {
|
||||
t.Parallel()
|
||||
s, err := New(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
act, err := NewFuncAction("x", func(context.Context, any) (any, error) { return nil, nil })
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction: %v", err)
|
||||
}
|
||||
if _, err := s.AddSchedule(context.Background(), "not a cron", act); !errors.Is(err, ErrInvalidCron) {
|
||||
t.Fatalf("err = %v, want ErrInvalidCron", err)
|
||||
}
|
||||
if _, err := s.AddSchedule(context.Background(), "* * * * * *", act); !errors.Is(err, ErrInvalidCron) {
|
||||
t.Fatalf("6-field without seconds: err = %v, want ErrInvalidCron", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestScheduler_RejectsNilAction(t *testing.T) {
|
||||
t.Parallel()
|
||||
s, err := New(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
if _, err := s.AddSchedule(context.Background(), "*/5 * * * *", nil); !errors.Is(err, ErrNilAction) {
|
||||
t.Fatalf("err = %v, want ErrNilAction", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestScheduler_StartStopIdempotent(t *testing.T) {
|
||||
t.Parallel()
|
||||
s, err := New(Config{}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
if err := s.Start(ctx); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
if err := s.Start(ctx); err != nil {
|
||||
t.Fatalf("second Start: %v", err)
|
||||
}
|
||||
if err := s.Stop(ctx); err != nil {
|
||||
t.Fatalf("Stop: %v", err)
|
||||
}
|
||||
if err := s.Stop(ctx); err != nil {
|
||||
t.Fatalf("second Stop: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestScheduler_FiresOnDue(t *testing.T) {
|
||||
t.Parallel()
|
||||
b := bus.New()
|
||||
defer b.Close()
|
||||
|
||||
got := make(chan bus.Event, 1)
|
||||
unsub, err := b.Subscribe(context.Background(), SubjectResult, func(_ context.Context, ev bus.Event) error {
|
||||
got <- ev
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("subscribe: %v", err)
|
||||
}
|
||||
defer unsub()
|
||||
|
||||
s, err := New(Config{}, Deps{Bus: b})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
act, err := NewFuncAction("tick", func(context.Context, any) (any, error) { return "ok", nil })
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
if _, err := s.AddSchedule(ctx, "@every 50ms", act); err != nil {
|
||||
t.Fatalf("AddSchedule: %v", err)
|
||||
}
|
||||
if err := s.Start(ctx); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case ev := <-got:
|
||||
res, ok := ev.Payload.(Result)
|
||||
if !ok {
|
||||
t.Fatalf("payload type = %T, want Result", ev.Payload)
|
||||
}
|
||||
if res.Kind != "schedule" || res.Action != "tick" || res.Output != "ok" {
|
||||
t.Fatalf("result = %+v, want {schedule tick ok}", res)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("schedule did not fire")
|
||||
}
|
||||
if err := s.Stop(ctx); err != nil {
|
||||
t.Fatalf("Stop: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestScheduler_WithSecondsAcceptsSixField(t *testing.T) {
|
||||
t.Parallel()
|
||||
s, err := New(Config{WithSeconds: true}, Deps{Bus: bus.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
act, err := NewFuncAction("tick", func(context.Context, any) (any, error) { return nil, nil })
|
||||
if err != nil {
|
||||
t.Fatalf("NewFuncAction: %v", err)
|
||||
}
|
||||
if _, err := s.AddSchedule(context.Background(), "*/1 * * * * *", act); err != nil {
|
||||
t.Fatalf("6-field with seconds rejected: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,149 @@
|
||||
package automations
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// Service composes the Scheduler (cron), the Engine (event rules) and the
|
||||
// ReminderStore into a single automation engine with one lifecycle. It routes
|
||||
// event rules to the Engine and cron rules to the Scheduler by Trigger kind.
|
||||
type Service struct {
|
||||
cfg Config
|
||||
deps Deps
|
||||
log *slog.Logger
|
||||
|
||||
sched *Scheduler
|
||||
engine *Engine
|
||||
remind *ReminderStore
|
||||
|
||||
mu sync.RWMutex
|
||||
cronRules map[string]ScheduleID // rule id -> schedule id
|
||||
}
|
||||
|
||||
// NewService builds an automation engine from cfg and deps. The bus in deps is
|
||||
// required. It also resolves the AI pipeline used by AIAction construction.
|
||||
func NewService(cfg Config, deps Deps) (*Service, error) {
|
||||
if deps.Bus == nil {
|
||||
return nil, fmt.Errorf("automations: new service: %w", ErrNilBus)
|
||||
}
|
||||
sched, err := New(cfg, deps)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
engine, err := NewEngine(cfg, deps)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
remind, err := NewReminder(cfg, deps)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Service{
|
||||
cfg: cfg,
|
||||
deps: deps,
|
||||
log: sourcedLog(deps.Log),
|
||||
sched: sched,
|
||||
engine: engine,
|
||||
remind: remind,
|
||||
cronRules: make(map[string]ScheduleID),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Pipeline returns the service's resolved AI pipeline, or nil when neither a loop
|
||||
// nor an agent is configured.
|
||||
func (s *Service) Pipeline() Pipeline { return s.deps.pipeline() }
|
||||
|
||||
// Scheduler returns the underlying cron scheduler.
|
||||
func (s *Service) Scheduler() *Scheduler { return s.sched }
|
||||
|
||||
// Engine returns the underlying event rule engine.
|
||||
func (s *Service) Engine() *Engine { return s.engine }
|
||||
|
||||
// ReminderStore returns the underlying reminder store.
|
||||
func (s *Service) ReminderStore() *ReminderStore { return s.remind }
|
||||
|
||||
// AddSchedule delegates to the Scheduler.
|
||||
func (s *Service) AddSchedule(ctx context.Context, cronExpr string, action Action) (ScheduleID, error) {
|
||||
return s.sched.AddSchedule(ctx, cronExpr, action)
|
||||
}
|
||||
|
||||
// AddRule routes an event rule to the Engine and a cron rule to the Scheduler. A
|
||||
// malformed rule returns an error wrapping ErrInvalidRule.
|
||||
func (s *Service) AddRule(ctx context.Context, r Rule) error {
|
||||
if !r.valid() {
|
||||
return fmt.Errorf("automations: add rule %q: %w", r.ID, ErrInvalidRule)
|
||||
}
|
||||
switch r.Trigger {
|
||||
case TriggerEvent:
|
||||
return s.engine.AddEvent(ctx, r)
|
||||
case TriggerCron:
|
||||
id, err := s.sched.AddSchedule(ctx, r.Expression, r.Action)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.cronRules[r.ID] = id
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
default:
|
||||
return fmt.Errorf("automations: add rule %q: %w", r.ID, ErrInvalidRule)
|
||||
}
|
||||
}
|
||||
|
||||
// RemoveRule removes an event or cron rule by id. It returns an error wrapping
|
||||
// ErrNotFound when no such rule exists.
|
||||
func (s *Service) RemoveRule(ctx context.Context, id string) error {
|
||||
s.mu.Lock()
|
||||
if schedID, ok := s.cronRules[id]; ok {
|
||||
delete(s.cronRules, id)
|
||||
s.mu.Unlock()
|
||||
s.sched.RemoveSchedule(schedID)
|
||||
return nil
|
||||
}
|
||||
s.mu.Unlock()
|
||||
return s.engine.RemoveRule(ctx, id)
|
||||
}
|
||||
|
||||
// AddReminder delegates to the ReminderStore.
|
||||
func (s *Service) AddReminder(ctx context.Context, r Reminder) error {
|
||||
return s.remind.Add(ctx, r)
|
||||
}
|
||||
|
||||
// Reminders returns a sorted snapshot of registered reminders.
|
||||
func (s *Service) Reminders() []Reminder { return s.remind.List() }
|
||||
|
||||
// CancelReminder delegates to the ReminderStore.
|
||||
func (s *Service) CancelReminder(ctx context.Context, id string) error {
|
||||
return s.remind.Cancel(ctx, id)
|
||||
}
|
||||
|
||||
// Start starts the scheduler, the event engine and the reminder loop.
|
||||
func (s *Service) Start(ctx context.Context) error {
|
||||
if err := s.sched.Start(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.engine.Start(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return s.remind.Start(ctx)
|
||||
}
|
||||
|
||||
// Stop stops the reminder loop, the event engine and the scheduler, collecting
|
||||
// errors from each.
|
||||
func (s *Service) Stop(ctx context.Context) error {
|
||||
var errs []error
|
||||
if err := s.remind.Stop(ctx); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
if err := s.engine.Stop(ctx); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
if err := s.sched.Stop(ctx); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
Reference in new issue
Block a user