From 67753927cb98636e4834eb19c5f5eea458ccc0e1 Mon Sep 17 00:00:00 2001 From: mirivlad Date: Tue, 14 Jul 2026 02:34:00 +0800 Subject: [PATCH] feat: add persisted notification scheduler --- internal/core/notifications/manager.go | 254 ++++++++++++++++++++ internal/core/notifications/manager_test.go | 131 ++++++++++ internal/core/notifications/store.go | 80 ++++++ 3 files changed, 465 insertions(+) create mode 100644 internal/core/notifications/manager.go create mode 100644 internal/core/notifications/manager_test.go create mode 100644 internal/core/notifications/store.go diff --git a/internal/core/notifications/manager.go b/internal/core/notifications/manager.go new file mode 100644 index 0000000..4bd4592 --- /dev/null +++ b/internal/core/notifications/manager.go @@ -0,0 +1,254 @@ +// Package notifications persists plugin-owned reminder schedules and delivers +// due entries through a desktop-native sender. +package notifications + +import ( + "context" + "fmt" + "sort" + "strings" + "sync" + "time" +) + +const ( + maxRequestsPerPlugin = 500 + maxIDLength = 256 + maxTitleLength = 512 + maxBodyLength = 4096 +) + +// VaultPath provides a stable operation while a vault is open. +type VaultPath interface { + WithOpenPath(func(string) error) error +} + +// Request is a desired scheduled notification in a plugin's own namespace. +type Request struct { + ID string `json:"id"` + DueAt string `json:"dueAt"` + Title string `json:"title"` + Body string `json:"body,omitempty"` +} + +// Item is the persisted form of a plugin notification schedule. +type Item struct { + PluginID string `json:"pluginId"` + ID string `json:"id"` + DueAt string `json:"dueAt"` + Title string `json:"title"` + Body string `json:"body,omitempty"` + SentForDueAt string `json:"sentForDueAt,omitempty"` +} + +// Sender delivers one native desktop notification. +type Sender interface { + Send(context.Context, Item) error +} + +// Manager owns canonical schedules for every authorized plugin. +type Manager struct { + mu sync.Mutex + vault VaultPath + sender Sender + now func() time.Time + + items []Item + loaded bool + stop chan struct{} +} + +// New creates a notification manager. The supplied clock is used by tests; a +// nil clock uses the current time. +func New(vault VaultPath, sender Sender, now func() time.Time) *Manager { + if now == nil { + now = time.Now + } + return &Manager{vault: vault, sender: sender, now: now} +} + +// Replace atomically replaces one plugin's desired schedules. +func (m *Manager) Replace(pluginID string, requests []Request) error { + if err := validateRequests(pluginID, requests); err != nil { + return err + } + m.mu.Lock() + defer m.mu.Unlock() + if err := m.loadLocked(); err != nil { + return err + } + + previous := make(map[string]Item) + kept := make([]Item, 0, len(m.items)+len(requests)) + for _, item := range m.items { + if item.PluginID == pluginID { + previous[item.ID] = item + continue + } + kept = append(kept, item) + } + for _, request := range requests { + item := Item{ + PluginID: pluginID, + ID: request.ID, + DueAt: request.DueAt, + Title: request.Title, + Body: request.Body, + } + if old, ok := previous[request.ID]; ok && old.DueAt == request.DueAt { + item.SentForDueAt = old.SentForDueAt + } + kept = append(kept, item) + } + sortItems(kept) + if err := m.writeLocked(kept); err != nil { + return err + } + m.items = kept + return nil +} + +// Clear removes every schedule owned by a plugin. +func (m *Manager) Clear(pluginID string) error { + return m.Replace(pluginID, nil) +} + +// Items returns a snapshot of the current schedule state. +func (m *Manager) Items() []Item { + m.mu.Lock() + defer m.mu.Unlock() + if err := m.loadLocked(); err != nil { + return nil + } + return append([]Item(nil), m.items...) +} + +// Path reports the canonical schedule path while the vault is open. +func (m *Manager) Path() string { + if m == nil || m.vault == nil { + return "" + } + var path string + if err := m.vault.WithOpenPath(func(vaultPath string) error { + path = schedulePath(vaultPath) + return nil + }); err != nil { + return "" + } + return path +} + +// Start immediately checks due schedules, then repeats every 30 seconds. +func (m *Manager) Start(ctx context.Context) { + if m == nil { + return + } + m.mu.Lock() + if m.stop != nil { + m.mu.Unlock() + return + } + stop := make(chan struct{}) + m.stop = stop + m.mu.Unlock() + + go func() { + m.Tick(ctx) + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for { + select { + case <-ticker.C: + m.Tick(ctx) + case <-stop: + return + case <-ctx.Done(): + return + } + } + }() +} + +// Stop stops a running scheduler. It is safe to call repeatedly. +func (m *Manager) Stop() { + if m == nil { + return + } + m.mu.Lock() + defer m.mu.Unlock() + if m.stop == nil { + return + } + close(m.stop) + m.stop = nil +} + +// Tick delivers each overdue schedule once. A sender failure leaves the item +// pending for a later tick. +func (m *Manager) Tick(ctx context.Context) { + if m == nil || m.sender == nil { + return + } + m.mu.Lock() + defer m.mu.Unlock() + if err := m.loadLocked(); err != nil { + return + } + now := m.now().UTC() + for index, item := range m.items { + dueAt, err := time.Parse(time.RFC3339, item.DueAt) + if err != nil || dueAt.After(now) || item.SentForDueAt == item.DueAt { + continue + } + if err := m.sender.Send(ctx, item); err != nil { + continue + } + updated := append([]Item(nil), m.items...) + updated[index].SentForDueAt = item.DueAt + if err := m.writeLocked(updated); err != nil { + continue + } + m.items = updated + } +} + +func validateRequests(pluginID string, requests []Request) error { + if strings.TrimSpace(pluginID) == "" || strings.ContainsAny(pluginID, "/\\") { + return fmt.Errorf("invalid plugin ID %q", pluginID) + } + if len(requests) > maxRequestsPerPlugin { + return fmt.Errorf("plugin %q has more than %d notification schedules", pluginID, maxRequestsPerPlugin) + } + seen := make(map[string]bool, len(requests)) + for _, request := range requests { + if request.ID == "" || len(request.ID) > maxIDLength { + return fmt.Errorf("notification ID is empty or too long") + } + if seen[request.ID] { + return fmt.Errorf("duplicate notification ID %q", request.ID) + } + seen[request.ID] = true + if !strings.HasSuffix(request.DueAt, "Z") { + return fmt.Errorf("notification %q dueAt must be UTC RFC3339", request.ID) + } + if _, err := time.Parse(time.RFC3339, request.DueAt); err != nil { + return fmt.Errorf("notification %q has invalid dueAt: %w", request.ID, err) + } + if strings.TrimSpace(request.Title) == "" || len(request.Title) > maxTitleLength { + return fmt.Errorf("notification %q title is empty or too long", request.ID) + } + if len(request.Body) > maxBodyLength { + return fmt.Errorf("notification %q body is too long", request.ID) + } + } + return nil +} + +func sortItems(items []Item) { + sort.Slice(items, func(left, right int) bool { + if items[left].PluginID != items[right].PluginID { + return items[left].PluginID < items[right].PluginID + } + return items[left].ID < items[right].ID + }) +} diff --git a/internal/core/notifications/manager_test.go b/internal/core/notifications/manager_test.go new file mode 100644 index 0000000..e7aed2a --- /dev/null +++ b/internal/core/notifications/manager_test.go @@ -0,0 +1,131 @@ +package notifications + +import ( + "context" + "errors" + "path/filepath" + "testing" + "time" +) + +type testVault struct { + path string +} + +func (v testVault) WithOpenPath(operation func(string) error) error { + return operation(v.path) +} + +type fakeSender struct { + err error + items []Item +} + +func (s *fakeSender) Send(_ context.Context, item Item) error { + s.items = append(s.items, item) + return s.err +} + +func testManager(t *testing.T, sender *fakeSender, now time.Time) *Manager { + t.Helper() + return New(testVault{path: t.TempDir()}, sender, func() time.Time { return now }) +} + +func request(id, dueAt string) Request { + return Request{ID: id, DueAt: dueAt, Title: "Reminder " + id, Body: "Todo " + id} +} + +func TestReplaceRemovesStalePluginSchedules(t *testing.T) { + manager := testManager(t, &fakeSender{}, time.Date(2026, 7, 14, 9, 0, 0, 0, time.UTC)) + if err := manager.Replace("verstak.todo", []Request{ + request("first", "2026-07-14T10:00:00Z"), + request("second", "2026-07-14T11:00:00Z"), + }); err != nil { + t.Fatalf("initial replace: %v", err) + } + if err := manager.Replace("verstak.todo", []Request{request("second", "2026-07-14T11:00:00Z")}); err != nil { + t.Fatalf("stale-item replace: %v", err) + } + + items := manager.Items() + if len(items) != 1 || items[0].PluginID != "verstak.todo" || items[0].ID != "second" { + t.Fatalf("items = %#v, want only second Todo reminder", items) + } +} + +func TestReplaceReschedulesDeliveredItem(t *testing.T) { + now := time.Date(2026, 7, 14, 12, 0, 0, 0, time.UTC) + sender := &fakeSender{} + manager := testManager(t, sender, now) + if err := manager.Replace("verstak.todo", []Request{request("todo-1", "2026-07-14T11:00:00Z")}); err != nil { + t.Fatalf("initial replace: %v", err) + } + manager.Tick(context.Background()) + if len(sender.items) != 1 || manager.Items()[0].SentForDueAt == "" { + t.Fatalf("first delivery = %#v, state = %#v", sender.items, manager.Items()) + } + + if err := manager.Replace("verstak.todo", []Request{request("todo-1", "2026-07-14T13:00:00Z")}); err != nil { + t.Fatalf("reschedule: %v", err) + } + if got := manager.Items()[0].SentForDueAt; got != "" { + t.Fatalf("rescheduled item kept old acknowledgment %q", got) + } +} + +func TestTickRetriesFailedSendAndAcknowledgesOneDelivery(t *testing.T) { + now := time.Date(2026, 7, 14, 12, 0, 0, 0, time.UTC) + sender := &fakeSender{err: errors.New("notification unavailable")} + manager := testManager(t, sender, now) + if err := manager.Replace("verstak.todo", []Request{request("todo-1", "2026-07-14T11:00:00Z")}); err != nil { + t.Fatalf("replace: %v", err) + } + + manager.Tick(context.Background()) + if len(sender.items) != 1 || manager.Items()[0].SentForDueAt != "" { + t.Fatalf("failed send was acknowledged: sent=%d items=%#v", len(sender.items), manager.Items()) + } + + sender.err = nil + manager.Tick(context.Background()) + manager.Tick(context.Background()) + if len(sender.items) != 2 { + t.Fatalf("send attempts = %d, want 2", len(sender.items)) + } + if got := manager.Items()[0].SentForDueAt; got != "2026-07-14T11:00:00Z" { + t.Fatalf("sent acknowledgment = %q", got) + } +} + +func TestPersistentSchedulesReloadAndDeliverOverdueOnlyOnce(t *testing.T) { + path := t.TempDir() + now := time.Date(2026, 7, 14, 12, 0, 0, 0, time.UTC) + first := New(testVault{path: path}, &fakeSender{}, func() time.Time { return now }) + if err := first.Replace("verstak.todo", []Request{request("todo-1", "2026-07-14T11:00:00Z")}); err != nil { + t.Fatalf("replace: %v", err) + } + if got, want := filepath.Join(path, ".verstak", "notifications", "schedules.json"), first.Path(); got != want { + t.Fatalf("schedule path = %q, want %q", got, want) + } + + sender := &fakeSender{} + restarted := New(testVault{path: path}, sender, func() time.Time { return now }) + restarted.Tick(context.Background()) + restarted.Tick(context.Background()) + if len(sender.items) != 1 || sender.items[0].ID != "todo-1" { + t.Fatalf("overdue delivery = %#v", sender.items) + } +} + +func TestReplaceRejectsInvalidRequests(t *testing.T) { + manager := testManager(t, &fakeSender{}, time.Now().UTC()) + for _, requests := range [][]Request{ + {{ID: "", DueAt: "2026-07-14T10:00:00Z", Title: "title"}}, + {{ID: "same", DueAt: "2026-07-14T10:00:00Z", Title: "title"}, {ID: "same", DueAt: "2026-07-14T11:00:00Z", Title: "title"}}, + {{ID: "bad-date", DueAt: "not-a-date", Title: "title"}}, + } { + if err := manager.Replace("verstak.todo", requests); err == nil { + t.Fatalf("Replace(%#v) succeeded", requests) + } + } +} diff --git a/internal/core/notifications/store.go b/internal/core/notifications/store.go new file mode 100644 index 0000000..824701d --- /dev/null +++ b/internal/core/notifications/store.go @@ -0,0 +1,80 @@ +package notifications + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" +) + +type storedSchedules struct { + Items []Item `json:"items"` +} + +func schedulePath(vaultPath string) string { + return filepath.Join(vaultPath, ".verstak", "notifications", "schedules.json") +} + +func (m *Manager) loadLocked() error { + if m.loaded { + return nil + } + if m.vault == nil { + return fmt.Errorf("notification vault is not initialized") + } + var loaded []Item + if err := m.vault.WithOpenPath(func(vaultPath string) error { + path := schedulePath(vaultPath) + data, err := os.ReadFile(path) + if os.IsNotExist(err) { + loaded = []Item{} + return nil + } + if err != nil { + return fmt.Errorf("read notification schedules: %w", err) + } + var stored storedSchedules + if err := json.Unmarshal(data, &stored); err != nil { + return fmt.Errorf("decode notification schedules: %w", err) + } + for _, item := range stored.Items { + if err := validateRequests(item.PluginID, []Request{{ID: item.ID, DueAt: item.DueAt, Title: item.Title, Body: item.Body}}); err != nil { + return fmt.Errorf("invalid stored notification schedule: %w", err) + } + loaded = append(loaded, item) + } + sortItems(loaded) + return nil + }); err != nil { + return err + } + m.items = loaded + m.loaded = true + return nil +} + +func (m *Manager) writeLocked(items []Item) error { + if m.vault == nil { + return fmt.Errorf("notification vault is not initialized") + } + return m.vault.WithOpenPath(func(vaultPath string) error { + data, err := json.MarshalIndent(storedSchedules{Items: items}, "", " ") + if err != nil { + return fmt.Errorf("encode notification schedules: %w", err) + } + path := schedulePath(vaultPath) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return fmt.Errorf("create notification schedule directory: %w", err) + } + temporary := filepath.Join(filepath.Dir(path), fmt.Sprintf(".schedules.%d.tmp", time.Now().UnixNano())) + if err := os.WriteFile(temporary, data, 0o644); err != nil { + return fmt.Errorf("write notification schedules: %w", err) + } + if err := os.Rename(temporary, path); err != nil { + _ = os.Remove(temporary) + return fmt.Errorf("replace notification schedules: %w", err) + } + return nil + }) +}