Maven/internal/loop/gather.go

// loop/gather.go — the ONE impure piece in the loop.
//
// Gather builds a State snapshot under the store lock at the start of each
// tick. From there, every predicate and the gate are pure functions over State.
//
// Why centralize the I/O: the loop is "dumb + deterministic", the spec
// repeatedly enforces a no-I/O contract on predicates. Centralizing read here
// makes the contract checkable (anywhere outside gather.go doing I/O is a bug).
package loop

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"time"

	"github.com/kami/maven/internal/calendar"
	"github.com/kami/maven/internal/store"
)

// Gatherer — holds nothing mutable; the Store is the only dependency. The
// daemon runs one Gatherer per tick.
type Gatherer struct {
	store      *store.Store
	rules      []Rule
	quietStart string // "HH:MM" local time, "" = disabled
	quietEnd   string // "HH:MM" local time, "" = disabled
}

func NewGatherer(s *store.Store, rules []Rule) *Gatherer {
	return &Gatherer{store: s, rules: rules}
}

// SetQuietHours enables the time-window schedule check. Times are local.
func (g *Gatherer) SetQuietHours(start, end string) {
	g.quietStart = start
	g.quietEnd = end
}

// GatherState — reads the store ONCE and assembles the snapshot the pure Tick
// will operate on.
//
// Reads the loop needs:
//   - presence: probes (ts per signal key), current bucket, compute score, resolve.
//   - facts: every key any rule's Predicate OR InertWhenNoData names.
//   - last nudge per rule (for cooldown).
//   - due reminders (the loop reuses the loop for reminders; we gather them here).
//   - env flags QuietHours / CalendarBusy — read as facts (kind=config/env).
//
// All reads share a single read-only transaction for a consistent snapshot.
func (g *Gatherer) GatherState(ctx context.Context, now time.Time) (State, []store.Reminder, error) {
	// presence first — pure function over probes. the gate checks the bucket;
	// delivery (later) checks the score.
	probes, err := g.store.PresenceProbes(ctx)
	if err != nil {
		return State{}, nil, err
	}
	lastBucket, _, _, err := g.store.LoadPresenceState(ctx)
	if err != nil {
		return State{}, nil, err
	}
	score := store.PresenceScore(now, probes)
	bucket := store.Resolve(score, lastBucket)

	// collect every key any rule references (predicate + inert list).
	// Daemon rules are small (≤ ~30, per the "revisit at 30 rules" line);
	// a single map over rules per tick is negligible at 60s cadence.
	wanted := make(map[string]struct{})
	for _, r := range g.rules {
		// We don't introspect the predicate closure (Go can't); the rule author
		// declares InertWhenNoData for keys the predicate reads. Reuse that list.
		for _, k := range r.InertWhenNoData {
			wanted[k] = struct{}{}
		}
	}
	// presence signal keys live in facts too — included via the probes path,
	// but also surface via Fact() for rules that want direct access (e.g. break).
	for _, sig := range store.PresenceSignals {
		wanted[sig.Key] = struct{}{}
	}

	facts := make(map[string]store.Fact, len(wanted))
	for k := range wanted {
		f, err := g.store.LatestFact(ctx, k)
		if err == nil {
			facts[k] = f
			continue
		}
		if err == store.ErrNoFact {
			continue // missing ⇒ shut up; the gate handles it
		}
		return State{}, nil, err
	}

	// prefix families — the keys a rule cannot name at wiring time (one fact
	// per kuma monitor). Loaded into the same map; State.FactsUnder reads them.
	for _, r := range g.rules {
		for _, p := range r.WantPrefixes {
			fam, err := g.store.LatestFactsByPrefix(ctx, p)
			if err != nil {
				return State{}, nil, err
			}
			for _, f := range fam {
				facts[f.Key] = f
			}
		}
	}

	// last nudge per rule + cooldown-until derived from the active cooldown.
	// "active" = the feedback tuner's persisted base if one exists, else the
	// rule's static Base. LatestFactBySource is the trust-by-provenance read
	// (a module that doesn't own the `feedback` source can't poison a rule's
	// cooldown once the auth source-scope lands — same shape as ServiceDownRule).
	lastNudge := make(map[string]store.Nudge, len(g.rules))
	cooldownUntil := make(map[string]time.Time, len(g.rules))
	for _, r := range g.rules {
		base := r.Cooldown.Base
		if f, err := g.store.LatestFactBySource(ctx, FeedbackKey(r), FeedbackSource); err == nil {
			if tuned, ok := ParseCooldownFact(f); ok {
				base = tuned
			}
		} else if err != store.ErrNoFact {
			return State{}, nil, err
		}
		n, err := g.store.LastNudge(ctx, r.Name)
		if err == nil {
			lastNudge[r.Name] = n
			cooldownUntil[r.Name] = CooldownFor(base, n.Ts)
			continue
		}
		if err == store.ErrNudgeNotFound {
			continue // never fired → no cooldown
		}
		return State{}, nil, err
	}

	// live snoozes — "leave me alone until X", per rule. The `snoozed` outcome
	// on the nudges table is the whole record; the store turns it into an
	// expiry. Absent rules mean "not snoozed", which is what the gate reads.
	snoozeUntil, err := g.store.SnoozedUntil(ctx, now)
	if err != nil {
		return State{}, nil, err
	}

	// env flags — QuietHours / CalendarBusy as config facts.
	// QuietHours: presence != reachability, sleep/quiet-hours handled separately
	// in the gate. We read a config `quiet_hours` fact for the boolean.
	var quiet bool
	if f, ok := readFact(ctx, g.store, "quiet_hours"); ok {
		quiet = f.Value == "true" || f.Value == `"true"`
	}
	// Schedule-based quiet hours: if a time window is configured AND we're
	// inside it, quiet is true regardless of the config fact. The voice
	// toggle extends activation independently — both can fire at once.
	if g.quietStart != "" && g.quietEnd != "" && inQuietWindow(now, g.quietStart, g.quietEnd) {
		quiet = true
	}
	var calBusy bool
	if f, ok := readFact(ctx, g.store, "calendar_busy"); ok {
		calBusy = f.Value == "true" || f.Value == `"true"`
	}
	// An ambient meeting suppresses a nudge too (Vikunja #513). It writes
	// calendar_event_* and never calendar_busy, which is the CalDAV poller's
	// level, so before this a low-confidence meeting was good enough to recite
	// out loud and not good enough to stop a nudge during it. That is
	// backwards: being wrong here costs one nudge he did not get.
	//
	// The expiry is the event's own span, which is why there is no new level
	// and no interval to choose. A poller re-asserts a level every cycle and a
	// notification arrives once; an event that already ended covers nothing,
	// and one that has not started yet covers nothing either.
	if !calBusy {
		calBusy = g.eventCoversNow(ctx, now)
	}

	// due reminders — gate-bypassing class. read here, the daemon emits them.
	due, err := g.store.DueReminders(ctx, now)
	if err != nil {
		return State{}, nil, err
	}
	due = collapseReminders(due)

	s := State{
		Now:           now,
		Presence:      bucket,
		PresenceScore: score,
		Facts:         facts,
		LastNudge:     lastNudge,
		SnoozeUntil:   snoozeUntil,
		CooldownUntil: cooldownUntil,
		QuietHours:    quiet,
		CalendarBusy:  calBusy,
	}
	return s, due, nil
}

// inQuietWindow returns true when `now` falls within the daily window
// [start, end). start/end are "HH:MM" in local time. Windows that cross
// midnight (start > end) are handled: "23:00"-"08:00" means quiet from
// 23:00 to 08:00 the next day.
func inQuietWindow(now time.Time, start, end string) bool {
	startH, startM, ok1 := parseHHMM(start)
	endH, endM, ok2 := parseHHMM(end)
	if !ok1 || !ok2 {
		return false
	}
	nowMin := now.Hour()*60 + now.Minute()
	startMin := startH*60 + startM
	endMin := endH*60 + endM
	if startMin < endMin {
		return nowMin >= startMin && nowMin < endMin
	}
	// Crossing midnight: e.g., 23:00-08:00.
	return nowMin >= startMin || nowMin < endMin
}

func parseHHMM(s string) (hour, min int, ok bool) {
	if len(s) != 5 || s[2] != ':' {
		return 0, 0, false
	}
	h := int(s[0]-'0')*10 + int(s[1]-'0')
	m := int(s[3]-'0')*10 + int(s[4]-'0')
	if h < 0 || h > 23 || m < 0 || m > 59 {
		return 0, 0, false
	}
	return h, m, true
}

// eventCoversNow reports whether any stored calendar event covers this instant.
// Read from the event facts themselves, so it holds for exactly as long as the
// meeting does — see the note at the call site.
//
// A read failure answers false: a meeting nobody can read about is not a reason
// to go quiet.
func (g *Gatherer) eventCoversNow(ctx context.Context, now time.Time) bool {
	fam, err := g.store.LatestFactsByPrefix(ctx, calendar.EventKeyPrefix)
	if err != nil {
		return false
	}
	for _, f := range fam {
		start, end, ok := calendar.FactSpan(f.Key, f.Value, now.Location())
		if !ok {
			continue
		}
		if !now.Before(start) && now.Before(end) {
			return true
		}
	}
	return false
}

func readFact(ctx context.Context, s *store.Store, key string) (store.Fact, bool) {
	f, err := s.LatestFact(ctx, key)
	if err != nil {
		return store.Fact{}, false
	}
	return f, true
}

// collapseReminders — when multiple reminders are due at once (e.g. after
// the daemon was offline), collapse them into digest reminders to avoid a
// burst of individual notifications. A previously phrased delivery group is
// kept intact and separate from newly-due reminders: otherwise one new row
// joining a failed bundle would force the old bundle through the model again.
// The synthetic digest (ID=0) carries the originals in Collapsed; the
// dispatcher completes them only after the digest actually delivers.
func collapseReminders(due []store.Reminder) []store.Reminder {
	if len(due) <= 1 {
		return due
	}

	// Every reminder without a delivery group is part of the new bundle for
	// this tick. Persisted groups each retain their own bundle identity.
	const ungrouped = "\x00"
	type reminderGroup struct {
		key       string
		reminders []store.Reminder
	}
	var groups []reminderGroup
	byKey := make(map[string]int)
	for _, r := range due {
		key := r.DeliveryGroup
		if key == "" {
			key = ungrouped
		}
		i, ok := byKey[key]
		if !ok {
			i = len(groups)
			byKey[key] = i
			groups = append(groups, reminderGroup{key: key})
		}
		groups[i].reminders = append(groups[i].reminders, r)
	}
	// DueReminders was ordered globally. Grouping can move rows together, so
	// restore earliest-first ordering between the resulting deliveries.
	sort.SliceStable(groups, func(i, j int) bool {
		a := groups[i].reminders[0]
		b := groups[j].reminders[0]
		if a.NextFireTs.Equal(b.NextFireTs) {
			return a.ID < b.ID
		}
		return a.NextFireTs.Before(b.NextFireTs)
	})

	out := make([]store.Reminder, 0, len(groups))
	for _, group := range groups {
		if len(group.reminders) == 1 {
			out = append(out, group.reminders[0])
			continue
		}
		out = append(out, collapseReminderGroup(group.reminders))
	}
	return out
}

func collapseReminderGroup(due []store.Reminder) store.Reminder {
	// Build a summary payload.
	var items []string
	earliest := due[0].FireTs
	for _, r := range due {
		// Each reminder's Payload is JSON. Try to extract a "text" field;
		// fall back to the raw payload.
		var parsed struct {
			Text string `json:"text"`
		}
		if json.Unmarshal([]byte(r.Payload), &parsed) == nil && parsed.Text != "" {
			items = append(items, parsed.Text)
		} else {
			items = append(items, r.Payload)
		}
		if r.FireTs.Before(earliest) {
			earliest = r.FireTs
		}
	}
	summary := fmt.Sprintf("You have %d pending reminders", len(due))
	digestPayload, _ := json.Marshal(map[string]any{
		"text":  summary,
		"items": items,
	})

	// Return a single synthetic digest reminder. ID=0 signals "digest" to
	// the dispatcher, which completes the Collapsed originals on success.
	return store.Reminder{
		ID:        0,
		FireTs:    earliest,
		Payload:   string(digestPayload),
		Status:    store.ReminderPending,
		Collapsed: due,
	}
}