From d80710d85a93759423eb5c60ed03d14c2dccb9d3 Mon Sep 17 00:00:00 2001 From: Aska Date: Fri, 21 Aug 2026 12:54:11 +0800 Subject: [PATCH] feat(gateway): automation rule engine and reminder scheduler - engine: 5-field cron parser (*/n, ranges, lists) firing at most once per minute; device-state triggers with edge detection; manual trigger via POST /automation/rules/:id/trigger; actions run through the AI pipeline so rules can call any tool (IoT control, search, ...) - reminders: due-scan scheduler delivering via WebSocket broadcast, optional LLM-polished reminder text with plain-title fallback; Reminder gains Done flag with ListDue/MarkDone - tests: cron matching and validation, once-per-minute dedup, device edge triggering, manual trigger, reminder delivery --- backend/cmd/cyrene/main.go | 56 +++++ backend/services/gateway/api/handlers_more.go | 40 +++- backend/services/gateway/api/router.go | 6 + backend/services/gateway/engine/cron.go | 93 ++++++++ backend/services/gateway/engine/engine.go | 178 ++++++++++++++++ .../services/gateway/engine/engine_test.go | 201 ++++++++++++++++++ backend/services/gateway/engine/reminders.go | 107 ++++++++++ 7 files changed, 679 insertions(+), 2 deletions(-) create mode 100644 backend/services/gateway/engine/cron.go create mode 100644 backend/services/gateway/engine/engine.go create mode 100644 backend/services/gateway/engine/engine_test.go create mode 100644 backend/services/gateway/engine/reminders.go diff --git a/backend/cmd/cyrene/main.go b/backend/cmd/cyrene/main.go index bdf40f2..301429a 100644 --- a/backend/cmd/cyrene/main.go +++ b/backend/cmd/cyrene/main.go @@ -43,6 +43,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/persona" "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/subsession" "git.yeij.top/AskaEth/Cyrene_v2/services/gateway/api" + "git.yeij.top/AskaEth/Cyrene_v2/services/gateway/engine" "git.yeij.top/AskaEth/Cyrene_v2/services/gateway/store" "git.yeij.top/AskaEth/Cyrene_v2/services/gateway/ws" "git.yeij.top/AskaEth/Cyrene_v2/services/memory" @@ -306,6 +307,35 @@ func main() { } }() + // === 12.7 Automation engine === + autoEngine := engine.New( + automationSourceAdapter{gwDeps.Automation}, + func(ctx context.Context, sessionID, userID, message string) { + tokenCh, _, err := orch.ProcessInput(ctx, sessionID, userID, message) + if err != nil { + logger.Infof("[automation] action failed: %v", err) + return + } + for range tokenCh { + } + }, + iotQuerierAdapter{reg}, + 30*time.Second, + ) + autoEngine.Start(context.Background()) + gwDeps.AutomationEng = autoEngine + + // === 12.8 Reminder scheduler === + remSched := engine.NewReminderScheduler( + gwDeps.Reminders, + func(userID, title, body string) { + wsHub.Broadcast(map[string]any{"type": "reminder", "title": title, "content": body}) + }, + &memory.LLMAdapterBridge{Adapter: llmSelector.DefaultAdapter()}, + 30*time.Second, + ) + remSched.Start(context.Background()) + apiRouter := api.New(gwDeps) wsMux := http.NewServeMux() @@ -394,6 +424,32 @@ func (a *perfAdapter) History() []map[string]any { return result } +// automationSourceAdapter adapts api.AutomationStore to engine.RuleSource. +type automationSourceAdapter struct{ s *api.AutomationStore } + +func (a automationSourceAdapter) List() []engine.Rule { + rules := a.s.List() + out := make([]engine.Rule, len(rules)) + for i, r := range rules { + out[i] = engine.Rule{ + ID: r.ID, Name: r.Name, Trigger: r.Trigger, + Condition: r.Condition, Action: r.Action, Enabled: r.Enabled, + } + } + return out +} + +// iotQuerierAdapter adapts the plugin registry to engine.DeviceQuerier. +type iotQuerierAdapter struct{ reg *plugin.Registry } + +func (a iotQuerierAdapter) ListDevices(ctx context.Context) ([]plugin.Device, error) { + iot := a.reg.GetIoTProvider() + if iot == nil { + return nil, fmt.Errorf("no iot provider") + } + return iot.ListDevices(ctx) +} + // chatPersisterAdapter adapts store.ChatStore to aiCoreCtx.Persister. type chatPersisterAdapter struct{ s *store.ChatStore } diff --git a/backend/services/gateway/api/handlers_more.go b/backend/services/gateway/api/handlers_more.go index 7327f74..478da26 100644 --- a/backend/services/gateway/api/handlers_more.go +++ b/backend/services/gateway/api/handlers_more.go @@ -8,6 +8,8 @@ import ( "strings" "sync" "time" + + "git.yeij.top/AskaEth/Cyrene_v2/services/gateway/engine" ) // ===== Automation Rules ===== @@ -57,6 +59,7 @@ type Reminder struct { Title string `json:"title"` DueAt string `json:"due_at"` // RFC3339 UserID string `json:"user_id"` + Done bool `json:"done"` CreatedAt string `json:"created_at"` } @@ -76,6 +79,25 @@ func (s *ReminderStore) List(userID string) []*Reminder { return result } +// ListDue returns undelivered reminders whose time has come. +func (s *ReminderStore) ListDue(now time.Time) []engine.Reminder { + s.mu.RLock(); defer s.mu.RUnlock() + var due []engine.Reminder + for _, r := range s.reminders { + if r.Done { continue } + t, err := time.Parse(time.RFC3339, r.DueAt) + if err != nil || t.After(now) { continue } + due = append(due, engine.Reminder{ID: r.ID, Title: r.Title, DueAt: t, UserID: r.UserID}) + } + return due +} + +// MarkDone marks a reminder as delivered. +func (s *ReminderStore) MarkDone(id string) { + s.mu.Lock(); defer s.mu.Unlock() + if r, ok := s.reminders[id]; ok { r.Done = true } +} + func (s *ReminderStore) Save(r *Reminder) { s.mu.Lock(); defer s.mu.Unlock() if r.ID == "" { r.ID = fmt.Sprintf("rem-%d", time.Now().UnixNano()) } @@ -123,8 +145,22 @@ func (r *Router) automationHandler(w http.ResponseWriter, req *http.Request) { func (r *Router) automationItemHandler(w http.ResponseWriter, req *http.Request) { store := r.deps.Automation if store == nil { writeJSON(w, 503, M{"error": "not available"}); return } - id := strings.TrimPrefix(req.URL.Path, "/api/v1/automation/rules/") - if req.Method == "DELETE" { store.Delete(id); writeJSON(w, 204, nil); return } + rest := strings.TrimPrefix(req.URL.Path, "/api/v1/automation/rules/") + + // POST /api/v1/automation/rules/{id}/trigger — fire immediately + if strings.HasSuffix(rest, "/trigger") { + if req.Method != "POST" { writeJSON(w, 405, M{"error": "POST required"}); return } + if r.deps.AutomationEng == nil { writeJSON(w, 503, M{"error": "automation engine not running"}); return } + id := strings.TrimSuffix(rest, "/trigger") + if err := r.deps.AutomationEng.Trigger(req.Context(), id); err != nil { + writeJSON(w, 404, M{"error": err.Error()}) + return + } + writeJSON(w, 200, M{"ok": true}) + return + } + + if req.Method == "DELETE" { store.Delete(rest); writeJSON(w, 204, nil); return } writeJSON(w, 405, M{"error": "method not allowed"}) } diff --git a/backend/services/gateway/api/router.go b/backend/services/gateway/api/router.go index 9480758..1d20fcb 100644 --- a/backend/services/gateway/api/router.go +++ b/backend/services/gateway/api/router.go @@ -86,6 +86,11 @@ type BridgeConfigManager interface { DeleteAdapter(name string) error } +// AutomationEngine fires automation rules immediately. +type AutomationEngine interface { + Trigger(ctx context.Context, id string) error +} + // ChatStorer persists chat sessions and messages. type ChatStorer interface { CreateSession(ctx context.Context, id, userID, title string) (*store.Session, error) @@ -110,6 +115,7 @@ type Dependencies struct { ChatStore ChatStorer ModelConfig *ModelConfigStore Automation *AutomationStore + AutomationEng AutomationEngine Reminders *ReminderStore PluginList []map[string]string } diff --git a/backend/services/gateway/engine/cron.go b/backend/services/gateway/engine/cron.go new file mode 100644 index 0000000..905302d --- /dev/null +++ b/backend/services/gateway/engine/cron.go @@ -0,0 +1,93 @@ +// Package engine provides the automation rule engine and reminder scheduler. +// Rules fire actions through the AI pipeline (which can call tools like IoT control). +package engine + +import ( + "fmt" + "strconv" + "strings" + "time" +) + +// cronField matches one field of a cron expression. +type cronField struct{ values map[int]bool } + +// parseCronField parses "*", "*/n", "a", "a,b,c", "a-b", "a-b/n". +func parseCronField(s string, min, max int) (cronField, error) { + f := cronField{values: map[int]bool{}} + for _, part := range strings.Split(s, ",") { + part = strings.TrimSpace(part) + if part == "" { + return f, fmt.Errorf("empty cron field part") + } + + step := 1 + base := part + if i := strings.Index(part, "/"); i >= 0 { + base = part[:i] + n, err := strconv.Atoi(part[i+1:]) + if err != nil || n <= 0 { + return f, fmt.Errorf("invalid step %q", part) + } + step = n + } + + lo, hi := min, max + switch { + case base == "*": + case strings.Contains(base, "-"): + segs := strings.SplitN(base, "-", 2) + a, err1 := strconv.Atoi(segs[0]) + b, err2 := strconv.Atoi(segs[1]) + if err1 != nil || err2 != nil { + return f, fmt.Errorf("invalid range %q", base) + } + lo, hi = a, b + default: + n, err := strconv.Atoi(base) + if err != nil { + return f, fmt.Errorf("invalid value %q", base) + } + lo, hi = n, n + } + if lo < min || hi > max || lo > hi { + return f, fmt.Errorf("value out of range %q", base) + } + for v := lo; v <= hi; v += step { + f.values[v] = true + } + } + return f, nil +} + +// CronSchedule is a parsed 5-field cron expression (min hour dom month dow). +type CronSchedule struct { + min, hour, dom, month, dow cronField +} + +// ParseCron parses a standard 5-field cron expression. +func ParseCron(expr string) (*CronSchedule, error) { + parts := strings.Fields(expr) + if len(parts) != 5 { + return nil, fmt.Errorf("cron must have 5 fields, got %d", len(parts)) + } + bounds := [5][2]int{{0, 59}, {0, 23}, {1, 31}, {1, 12}, {0, 6}} + var fields [5]cronField + for i, p := range parts { + f, err := parseCronField(p, bounds[i][0], bounds[i][1]) + if err != nil { + return nil, fmt.Errorf("field %d: %w", i+1, err) + } + fields[i] = f + } + return &CronSchedule{fields[0], fields[1], fields[2], fields[3], fields[4]}, nil +} + +// Match reports whether t matches the schedule (minute precision). +func (c *CronSchedule) Match(t time.Time) bool { + return c.min.values[t.Minute()] && + c.hour.values[t.Hour()] && + c.dom.values[t.Day()] && + c.month.values[int(t.Month())] && + c.dow.values[int(t.Weekday())] +} diff --git a/backend/services/gateway/engine/engine.go b/backend/services/gateway/engine/engine.go new file mode 100644 index 0000000..5a8c74d --- /dev/null +++ b/backend/services/gateway/engine/engine.go @@ -0,0 +1,178 @@ +package engine + +import ( + "context" + "fmt" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/plugin" +) + +// Rule is an automation rule. Trigger types: "time" (cron), "event" (device state). +// The action is a natural-language instruction executed through the AI pipeline, +// so it can call any tool (IoT control, search, etc.). +type Rule struct { + ID string + Name string + Trigger string // "time" | "event" | "webhook" + Condition map[string]string // time: {"cron": "..."}; event: {"device_id","field","value"} + Action string + Enabled bool +} + +// RuleSource provides the current automation rules. +type RuleSource interface { + List() []Rule +} + +// AIFunc executes an action prompt through the AI pipeline (fire and forget). +type AIFunc func(ctx context.Context, sessionID, userID, message string) + +// DeviceQuerier polls IoT device states for event triggers. +type DeviceQuerier interface { + ListDevices(ctx context.Context) ([]plugin.Device, error) +} + +// Engine evaluates automation rules on a ticker. +type Engine struct { + src RuleSource + ai AIFunc + devices DeviceQuerier + interval time.Duration + + mu sync.Mutex + lastCronFire map[string]string // ruleID → fired minute key + deviceMatches map[string]bool // ruleID → condition held last poll + stop chan struct{} + stopOnce sync.Once +} + +// New creates an automation engine. +func New(src RuleSource, ai AIFunc, devices DeviceQuerier, interval time.Duration) *Engine { + if interval <= 0 { + interval = 30 * time.Second + } + return &Engine{ + src: src, + ai: ai, + devices: devices, + interval: interval, + lastCronFire: map[string]string{}, + deviceMatches: map[string]bool{}, + stop: make(chan struct{}), + } +} + +// Start begins the evaluation loop. +func (e *Engine) Start(ctx context.Context) { + go func() { + ticker := time.NewTicker(e.interval) + defer ticker.Stop() + for { + select { + case <-e.stop: + return + case <-ctx.Done(): + return + case t := <-ticker.C: + e.check(ctx, t) + } + } + }() + logger.Infof("[automation] engine started (interval=%v)", e.interval) +} + +// Stop shuts down the engine. +func (e *Engine) Stop() { e.stopOnce.Do(func() { close(e.stop) }) } + +// Trigger fires a rule immediately (manual / webhook). +func (e *Engine) Trigger(ctx context.Context, id string) error { + for _, r := range e.src.List() { + if r.ID == id { + e.fire(ctx, r, "manual") + return nil + } + } + return fmt.Errorf("rule %s not found", id) +} + +// check evaluates all rules once. +func (e *Engine) check(ctx context.Context, now time.Time) { + for _, r := range e.src.List() { + if !r.Enabled { + continue + } + switch r.Trigger { + case "time": + e.checkCron(ctx, r, now) + case "event": + e.checkDevice(ctx, r) + } + } +} + +func (e *Engine) checkCron(ctx context.Context, r Rule, now time.Time) { + expr := r.Condition["cron"] + if expr == "" { + return + } + sched, err := ParseCron(expr) + if err != nil { + return + } + if !sched.Match(now) { + return + } + // Fire at most once per minute per rule + minuteKey := now.Format("2006-01-02 15:04") + e.mu.Lock() + if e.lastCronFire[r.ID] == minuteKey { + e.mu.Unlock() + return + } + e.lastCronFire[r.ID] = minuteKey + e.mu.Unlock() + e.fire(ctx, r, "cron") +} + +func (e *Engine) checkDevice(ctx context.Context, r Rule) { + if e.devices == nil { + return + } + devID, field, want := r.Condition["device_id"], r.Condition["field"], r.Condition["value"] + if devID == "" || field == "" { + return + } + + devices, err := e.devices.ListDevices(ctx) + if err != nil { + return + } + held := false + for _, d := range devices { + if d.ID != devID { + continue + } + if v, ok := d.State[field]; ok && fmt.Sprint(v) == want { + held = true + } + } + + // Edge trigger: only fire when the condition starts holding + e.mu.Lock() + was := e.deviceMatches[r.ID] + e.deviceMatches[r.ID] = held + e.mu.Unlock() + if held && !was { + e.fire(ctx, r, "device_state") + } +} + +func (e *Engine) fire(ctx context.Context, r Rule, reason string) { + logger.Infof("[automation] rule %q fired (%s): %s", r.Name, reason, r.Action) + if e.ai != nil && r.Action != "" { + go e.ai(ctx, "automation-"+r.ID, "automation", r.Action) + } +} diff --git a/backend/services/gateway/engine/engine_test.go b/backend/services/gateway/engine/engine_test.go new file mode 100644 index 0000000..a636e9b --- /dev/null +++ b/backend/services/gateway/engine/engine_test.go @@ -0,0 +1,201 @@ +package engine + +import ( + "context" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/plugin" +) + +// --- cron --- + +func TestParseCronMatch(t *testing.T) { + tests := []struct { + expr string + at time.Time + want bool + }{ + {"* * * * *", time.Date(2026, 8, 21, 10, 30, 0, 0, time.Local), true}, + {"30 10 * * *", time.Date(2026, 8, 21, 10, 30, 0, 0, time.Local), true}, + {"31 10 * * *", time.Date(2026, 8, 21, 10, 30, 0, 0, time.Local), false}, + {"*/15 * * * *", time.Date(2026, 8, 21, 10, 45, 0, 0, time.Local), true}, + {"*/15 * * * *", time.Date(2026, 8, 21, 10, 46, 0, 0, time.Local), false}, + {"0 8-20/2 * * *", time.Date(2026, 8, 21, 10, 0, 0, 0, time.Local), true}, + {"0 8-20/2 * * *", time.Date(2026, 8, 21, 9, 0, 0, 0, time.Local), false}, + {"0 9 21 8 *", time.Date(2026, 8, 21, 9, 0, 0, 0, time.Local), true}, + {"0 9 * * 5", time.Date(2026, 8, 21, 9, 0, 0, 0, time.Local), true}, // 2026-08-21 is Friday + {"0 9 * * 1", time.Date(2026, 8, 21, 9, 0, 0, 0, time.Local), false}, + {"0,30 9 * * *", time.Date(2026, 8, 21, 9, 30, 0, 0, time.Local), true}, + } + for _, tt := range tests { + s, err := ParseCron(tt.expr) + if err != nil { + t.Fatalf("ParseCron(%q): %v", tt.expr, err) + } + if got := s.Match(tt.at); got != tt.want { + t.Errorf("Match(%q, %v) = %v, want %v", tt.expr, tt.at, got, tt.want) + } + } +} + +func TestParseCronInvalid(t *testing.T) { + for _, expr := range []string{"* * * *", "61 * * * *", "a b c d e", "*/0 * * * *", "5-1 * * * *"} { + if _, err := ParseCron(expr); err == nil { + t.Errorf("ParseCron(%q) should fail", expr) + } + } +} + +// --- engine --- + +type fakeRuleSource struct{ rules []Rule } + +func (f fakeRuleSource) List() []Rule { return f.rules } + +type fakeDevices struct{ devices []plugin.Device } + +func (f fakeDevices) ListDevices(context.Context) ([]plugin.Device, error) { return f.devices, nil } + +type firedRecorder struct { + mu sync.Mutex + calls []string +} + +func (f *firedRecorder) record(_ context.Context, sessionID, _, message string) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, sessionID+"|"+message) +} + +func (f *firedRecorder) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.calls) +} + +// eventually waits until the recorder reaches n calls (fires are async). +func (f *firedRecorder) eventually(t *testing.T, n int) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if f.count() >= n { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("expected %d fires, got %d", n, f.count()) +} + +// stillSmall waits briefly and asserts the count stays at n. +func (f *firedRecorder) stillAt(t *testing.T, n int) { + t.Helper() + time.Sleep(60 * time.Millisecond) + if f.count() != n { + t.Fatalf("expected count to stay at %d, got %d", n, f.count()) + } +} + +func TestCronRuleFiresOncePerMinute(t *testing.T) { + now := time.Date(2026, 8, 21, 10, 30, 0, 0, time.Local) + src := fakeRuleSource{rules: []Rule{{ + ID: "r1", Name: "晨检", Trigger: "time", Enabled: true, + Condition: map[string]string{"cron": "30 10 * * *"}, + Action: "检查一下设备", + }}} + rec := &firedRecorder{} + e := New(src, rec.record, nil, time.Minute) + + e.check(context.Background(), now) + e.check(context.Background(), now) // same minute — must not refire + rec.eventually(t, 1) + + e.check(context.Background(), now.Add(time.Minute)) + rec.stillAt(t, 1) +} + +func TestDeviceRuleEdgeTrigger(t *testing.T) { + src := fakeRuleSource{rules: []Rule{{ + ID: "r2", Name: "灯开联动", Trigger: "event", Enabled: true, + Condition: map[string]string{"device_id": "light-1", "field": "power", "value": "true"}, + Action: "灯开了，说点什么", + }}} + devices := &fakeDevices{devices: []plugin.Device{ + {ID: "light-1", State: map[string]any{"power": false}}, + }} + rec := &firedRecorder{} + e := New(src, rec.record, devices, time.Minute) + ctx := context.Background() + + e.check(ctx, time.Now()) // off → no fire + rec.stillAt(t, 0) + + devices.devices[0].State["power"] = true + e.check(ctx, time.Now()) // edge → fire + rec.eventually(t, 1) + + e.check(ctx, time.Now()) // held → no refire + rec.stillAt(t, 1) + + devices.devices[0].State["power"] = false + e.check(ctx, time.Now()) + devices.devices[0].State["power"] = true + e.check(ctx, time.Now()) // new edge → fire again + rec.eventually(t, 2) +} + +func TestManualTrigger(t *testing.T) { + src := fakeRuleSource{rules: []Rule{{ + ID: "r3", Name: "手动", Trigger: "webhook", Enabled: false, Action: "执行", + }}} + rec := &firedRecorder{} + e := New(src, rec.record, nil, time.Minute) + + if err := e.Trigger(context.Background(), "r3"); err != nil { + t.Fatalf("Trigger: %v", err) + } + rec.eventually(t, 1) + if err := e.Trigger(context.Background(), "nope"); err == nil { + t.Error("expected error for unknown rule") + } +} + +// --- reminder scheduler --- + +type fakeReminderSource struct { + reminders []Reminder + done []string +} + +func (f *fakeReminderSource) ListDue(now time.Time) []Reminder { + var out []Reminder + for _, r := range f.reminders { + if !r.DueAt.After(now) { + out = append(out, r) + } + } + return out +} + +func (f *fakeReminderSource) MarkDone(id string) { f.done = append(f.done, id) } + +func TestReminderDelivery(t *testing.T) { + src := &fakeReminderSource{reminders: []Reminder{ + {ID: "rem1", Title: "吃药", DueAt: time.Now().Add(-time.Minute), UserID: "u1"}, + {ID: "rem2", Title: "未来的事", DueAt: time.Now().Add(time.Hour), UserID: "u1"}, + }} + var notified []string + sched := NewReminderScheduler(src, func(userID, title, body string) { + notified = append(notified, userID+"|"+title+"|"+body) + }, nil, time.Minute) + + sched.deliver(context.Background(), time.Now()) + + if len(notified) != 1 || notified[0] != "u1|吃药|吃药" { + t.Errorf("unexpected notifications: %v", notified) + } + if len(src.done) != 1 || src.done[0] != "rem1" { + t.Errorf("expected rem1 marked done, got %v", src.done) + } +} diff --git a/backend/services/gateway/engine/reminders.go b/backend/services/gateway/engine/reminders.go new file mode 100644 index 0000000..61f78df --- /dev/null +++ b/backend/services/gateway/engine/reminders.go @@ -0,0 +1,107 @@ +package engine + +import ( + "context" + "fmt" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "strings" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model" +) + +// Reminder is a due notification for a user. +type Reminder struct { + ID string + Title string + DueAt time.Time + UserID string +} + +// ReminderSource provides due reminders and marks them delivered. +type ReminderSource interface { + // ListDue returns reminders that are due and not yet delivered. + ListDue(now time.Time) []Reminder + // MarkDone marks a reminder as delivered. + MarkDone(id string) +} + +// ReminderNotifier delivers a reminder to the user (e.g. via WebSocket). +type ReminderNotifier func(userID, title, body string) + +// Copywriter polishes reminder text via the LLM (optional). +type Copywriter interface { + Chat(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) +} + +// ReminderScheduler scans for due reminders and delivers them. +type ReminderScheduler struct { + src ReminderSource + notify ReminderNotifier + llm Copywriter + interval time.Duration + stop chan struct{} + stopOnce sync.Once +} + +// NewReminderScheduler creates a scheduler scanning at the given interval. +func NewReminderScheduler(src ReminderSource, notify ReminderNotifier, llm Copywriter, interval time.Duration) *ReminderScheduler { + if interval <= 0 { + interval = 30 * time.Second + } + return &ReminderScheduler{src: src, notify: notify, llm: llm, interval: interval, stop: make(chan struct{})} +} + +// Start begins the scan loop. +func (s *ReminderScheduler) Start(ctx context.Context) { + go func() { + ticker := time.NewTicker(s.interval) + defer ticker.Stop() + for { + select { + case <-s.stop: + return + case <-ctx.Done(): + return + case t := <-ticker.C: + s.deliver(ctx, t) + } + } + }() + logger.Infof("[reminder] scheduler started (interval=%v)", s.interval) +} + +// Stop shuts down the scheduler. +func (s *ReminderScheduler) Stop() { s.stopOnce.Do(func() { close(s.stop) }) } + +func (s *ReminderScheduler) deliver(ctx context.Context, now time.Time) { + for _, r := range s.src.ListDue(now) { + body := s.polish(ctx, r.Title) + logger.Infof("[reminder] due: %q for %s", r.Title, r.UserID) + if s.notify != nil { + s.notify(r.UserID, r.Title, body) + } + s.src.MarkDone(r.ID) + } +} + +// polish asks the LLM for a warmer reminder text; falls back to the title. +func (s *ReminderScheduler) polish(ctx context.Context, title string) string { + if s.llm == nil { + return title + } + c, cancel := context.WithTimeout(ctx, 20*time.Second) + defer cancel() + resp, err := s.llm.Chat(c, []model.LLMMessage{ + {Role: "system", Content: "你是昔涟。用一两句温柔自然的话提醒主人一件事，不要解释，直接输出提醒内容。"}, + {Role: "user", Content: fmt.Sprintf("请提醒我：%s", title)}, + }) + if err != nil { + return title + } + if text := strings.TrimSpace(resp.Content); text != "" && len([]rune(text)) <= 120 { + return text + } + return title +}