Cyrene_v2/internal/services/automations/reminder.go

package automations

import (
	"context"
	"fmt"
	"log/slog"
	"sort"
	"sync"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)

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 `json:"id"`
	// At is the time the reminder should fire.
	At time.Time `json:"at"`
	// Message is the payload surfaced to the user when it fires.
	Message string `json:"message"`
}

// 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", trace.RedactText(r.Message))
	}
}