Maven/cmd/mavend/factenrichment.go

// factenrichment.go — resolves the Subject of entity-aware facts against
// Nexus, turning free-text ("the espresso machine", "Kate") into a canonical
// entity_id (Vikunja #279). Mirrors Praxis's async enrichment worker
// (internal/enrichment/worker.go there): a plain poll-queue-update loop, no
// shared state with the tick loop beyond the store and ecosystemWiring both
// already hold.
package main

import (
	"context"
	"log"
	"sync"
	"time"

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

// factEnrichmentWorker polls store.PendingFactResolutions and resolves each
// one's Subject through ecosystemWiring.resolveEntityReference. A nil nexus
// client (ecosystem.nexus unconfigured) makes every tick a no-op rather than
// erroring — entity-aware memory degrades to "facts just aren't tagged",
// never to a crash loop.
type factEnrichmentWorker struct {
	store    *store.Store
	eco      *ecosystemWiring
	interval time.Duration
	batch    int // facts resolved per tick; keeps a single slow tick bounded
	now      func() time.Time

	// Retry state for facts whose resolution failed transiently. Kept in
	// memory rather than in the DB: a restart legitimately retries
	// everything, and the backoff exists to spare a struggling Nexus, not
	// to be durable. A fact is never given up on — degraded means slower,
	// not dropped.
	mu      sync.Mutex
	attempt map[int64]int       // fact id → consecutive failures
	nextTry map[int64]time.Time // fact id → earliest retry
}

// enrichmentScanLimit bounds how deep a single tick (or status report) walks
// the pending queue looking for facts whose backoff has elapsed. The queue is
// ordered by id, so without a scan the oldest facts hold every batch slot
// whether or not they are eligible, and one permanently failing fact stalls
// every younger one behind it.
const enrichmentScanLimit = 1000

// enrichmentBackoff is the wait before retrying a fact after n consecutive
// failures, capped so a long Nexus outage still retries about hourly.
func enrichmentBackoff(n int) time.Duration {
	d := time.Minute
	for i := 1; i < n && d < time.Hour; i++ {
		d *= 2
	}
	if d > time.Hour {
		d = time.Hour
	}
	return d
}

func newFactEnrichmentWorker(st *store.Store, eco *ecosystemWiring, interval time.Duration) *factEnrichmentWorker {
	return &factEnrichmentWorker{
		store:    st,
		eco:      eco,
		interval: interval,
		batch:    20,
		now:      time.Now,
		attempt:  map[int64]int{},
		nextTry:  map[int64]time.Time{},
	}
}

// enrichmentStatus is what the worker reports about its own health: how many
// facts are waiting, how many of those are currently in backoff, and the worst
// retry count among them. Degradation is reported, never hidden — a Nexus that
// has been down all day must be visible as a backlog, not as facts that
// silently never got tagged.
//
// All three numbers describe the same set of rows, the first
// enrichmentScanLimit pending facts. Counting Pending over a thousand rows
// while counting InBackoff over the twenty that reached the head of a batch
// described two different populations under one struct.
type enrichmentStatus struct {
	Pending     int
	InBackoff   int
	MaxAttempts int
	Scanned     int // rows the other three counts were taken over
}

func (w *factEnrichmentWorker) status(ctx context.Context) enrichmentStatus {
	var st enrichmentStatus
	pending, err := w.store.PendingFactResolutions(ctx, enrichmentScanLimit)
	if err != nil {
		log.Printf("factenrichment: status: %v", err)
		return st
	}
	st.Pending = len(pending)
	st.Scanned = len(pending)
	w.mu.Lock()
	defer w.mu.Unlock()
	now := w.now()
	for _, f := range pending {
		if next, ok := w.nextTry[f.ID]; ok && now.Before(next) {
			st.InBackoff++
		}
		if n := w.attempt[f.ID]; n > st.MaxAttempts {
			st.MaxAttempts = n
		}
	}
	return st
}

func (w *factEnrichmentWorker) run(ctx context.Context) {
	if w.eco == nil || w.eco.nexus == nil {
		log.Printf("factenrichment: nexus not configured, worker idle")
		<-ctx.Done()
		return
	}

	ticker := time.NewTicker(w.interval)
	defer ticker.Stop()

	w.tick(ctx)
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			w.tick(ctx)
		}
	}
}

func (w *factEnrichmentWorker) tick(ctx context.Context) {
	// Scan past the facts that are still in backoff instead of letting them
	// occupy the batch. The queue is ordered by id, so the oldest facts are
	// pulled first whether or not they are eligible: twenty facts Nexus keeps
	// rejecting would otherwise hold every slot forever and enrichment would
	// stop with no error and no log line, because a tick that skips everything
	// fails nothing.
	pending, err := w.store.PendingFactResolutions(ctx, enrichmentScanLimit)
	if err != nil {
		log.Printf("factenrichment: list pending: %v", err)
		return
	}
	w.forgetDeparted(pending)
	skipped, failed, attempted := 0, 0, 0
	for _, f := range pending {
		if attempted >= w.batch {
			break
		}
		if !w.due(f.ID) {
			skipped++
			continue
		}
		attempted++
		if !w.resolveOne(ctx, f) {
			failed++
		}
	}
	if failed > 0 {
		log.Printf("factenrichment: %d/%d resolutions failed this tick, %d held in backoff",
			failed, attempted, skipped)
	}
	// Report the backlog every tick, not only when something failed: the
	// stalled state worth seeing is the one where nothing failed because
	// nothing was attempted.
	if st := w.status(ctx); st.Pending > 0 {
		log.Printf("factenrichment: %d facts pending entity resolution, %d in backoff, worst attempt %d (scanned %d)",
			st.Pending, st.InBackoff, st.MaxAttempts, st.Scanned)
	}
}

// forgetDeparted drops retry state for facts that are no longer pending. A
// fact can leave the queue without ever resolving here — voided, or resolved
// by a later write — and its entries would otherwise live as long as the
// process does.
func (w *factEnrichmentWorker) forgetDeparted(pending []store.Fact) {
	live := make(map[int64]struct{}, len(pending))
	for _, f := range pending {
		live[f.ID] = struct{}{}
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	for id := range w.attempt {
		if _, ok := live[id]; !ok {
			delete(w.attempt, id)
		}
	}
	for id := range w.nextTry {
		if _, ok := live[id]; !ok {
			delete(w.nextTry, id)
		}
	}
}

// due reports whether a fact's backoff window has elapsed.
func (w *factEnrichmentWorker) due(id int64) bool {
	w.mu.Lock()
	defer w.mu.Unlock()
	next, ok := w.nextTry[id]
	return !ok || !w.now().Before(next)
}

// resolveOne resolves one pending fact. It returns false when the attempt
// failed transiently: the fact stays pending and is retried on a backoff.
func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) bool {
	entityID, _, ambiguous, err := w.eco.resolveEntityReference(ctx, f.Subject, nil)
	if err != nil {
		// Transient (Nexus unreachable) — leave pending, back off, retry later.
		// The subject is his words: log its length, the way the trace does.
		log.Printf("factenrichment: resolve fact %d subject %s: %v", f.ID, redactSubject(f.Subject), err)
		w.backOff(f.ID)
		return false
	}
	state := store.ResolutionNotFound
	switch {
	case entityID != "":
		state = store.ResolutionResolved
	case len(ambiguous) > 0:
		state = store.ResolutionAmbiguous
	}
	if err := w.store.ResolveFactEntity(ctx, f.ID, entityID, state); err != nil {
		// A failed write leaves the fact pending exactly like a failed resolve
		// does, so it gets the same pacing. Clearing the counters first meant
		// this one path retried every tick, at full rate, with no ceiling.
		log.Printf("factenrichment: record resolution for fact %d: %v", f.ID, err)
		w.backOff(f.ID)
		return false
	}
	w.mu.Lock()
	delete(w.attempt, f.ID)
	delete(w.nextTry, f.ID)
	w.mu.Unlock()
	return true
}

// backOff records one more consecutive failure for a fact and pushes its next
// attempt out accordingly.
func (w *factEnrichmentWorker) backOff(id int64) {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.attempt[id]++
	w.nextTry[id] = w.now().Add(enrichmentBackoff(w.attempt[id]))
}