diff --git a/internal/services/automations/action.go b/internal/services/automations/action.go new file mode 100644 index 0000000..759f046 --- /dev/null +++ b/internal/services/automations/action.go @@ -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) +} diff --git a/internal/services/automations/action_test.go b/internal/services/automations/action_test.go new file mode 100644 index 0000000..fe15a83 --- /dev/null +++ b/internal/services/automations/action_test.go @@ -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") + } +} diff --git a/internal/services/automations/automations.go b/internal/services/automations/automations.go new file mode 100644 index 0000000..c0dbd70 --- /dev/null +++ b/internal/services/automations/automations.go @@ -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 +} diff --git a/internal/services/automations/reminder.go b/internal/services/automations/reminder.go new file mode 100644 index 0000000..01b49c0 --- /dev/null +++ b/internal/services/automations/reminder.go @@ -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) + } +} diff --git a/internal/services/automations/reminder_test.go b/internal/services/automations/reminder_test.go new file mode 100644 index 0000000..0d23a27 --- /dev/null +++ b/internal/services/automations/reminder_test.go @@ -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) + } +} diff --git a/internal/services/automations/rule.go b/internal/services/automations/rule.go new file mode 100644 index 0000000..d9251a3 --- /dev/null +++ b/internal/services/automations/rule.go @@ -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 +} diff --git a/internal/services/automations/rule_test.go b/internal/services/automations/rule_test.go new file mode 100644 index 0000000..34c5cdf --- /dev/null +++ b/internal/services/automations/rule_test.go @@ -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) + } +} diff --git a/internal/services/automations/scheduler.go b/internal/services/automations/scheduler.go new file mode 100644 index 0000000..a5219a8 --- /dev/null +++ b/internal/services/automations/scheduler.go @@ -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) +} diff --git a/internal/services/automations/scheduler_test.go b/internal/services/automations/scheduler_test.go new file mode 100644 index 0000000..e3dbfca --- /dev/null +++ b/internal/services/automations/scheduler_test.go @@ -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) + } +} diff --git a/internal/services/automations/service.go b/internal/services/automations/service.go new file mode 100644 index 0000000..7305a1f --- /dev/null +++ b/internal/services/automations/service.go @@ -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...) +}