feat: add persisted notification scheduler
This commit is contained in:
parent
55edaa0209
commit
67753927cb
|
|
@ -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
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
})
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue