glpi-ai-agent/internal/state/store.go

package state

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"

	"github.com/example/glpi-ai-agent/internal/model"
)

type Store struct {
	mu                sync.RWMutex
	path              string
	indexPath         string
	processedVersions map[int64]string
	escalationKeys    map[string]struct{}
	runs              []model.RunRecord
	runIndex          map[string]int
	analysisIndex     map[string]model.AnalysisRun
	maxRuns           int
}

type durableIndex struct {
	ProcessedVersions map[string]string `json:"processed_versions"`
	EscalationKeys    []string          `json:"escalation_keys"`
}

func Open(dir string, maxRuns int) (*Store, error) {
	if err := os.MkdirAll(dir, 0o750); err != nil {
		return nil, err
	}
	s := &Store{
		path:              filepath.Join(dir, "runs.jsonl"),
		indexPath:         filepath.Join(dir, "state-index.json"),
		processedVersions: map[int64]string{},
		escalationKeys:    map[string]struct{}{},
		runIndex:          map[string]int{},
		analysisIndex:     map[string]model.AnalysisRun{},
		maxRuns:           maxRuns,
	}
	// Fail fast during startup if the persistent data path is not writable.
	// A read-only/root-owned Docker volume must not be discovered only after the first ticket.
	f, err := os.OpenFile(s.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o640)
	if err != nil {
		return nil, fmt.Errorf("state directory %q is not writable: %w", dir, err)
	}
	if err := f.Close(); err != nil {
		return nil, fmt.Errorf("close state write probe: %w", err)
	}
	if err := s.loadDurableIndex(); err != nil {
		return nil, err
	}
	if err := s.load(); err != nil {
		return nil, err
	}
	return s, nil
}
func (s *Store) Seen(id int64, version string) bool {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return version != "" && s.processedVersions[id] == version
}
func (s *Store) ProcessedVersionCount() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return len(s.processedVersions)
}

func (s *Store) Append(r model.RunRecord) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	f, err := os.OpenFile(s.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o640)
	if err != nil {
		return err
	}
	enc := json.NewEncoder(f)
	if err = enc.Encode(r); err == nil {
		err = f.Sync()
	}
	cerr := f.Close()
	if err == nil {
		err = cerr
	}
	if err != nil {
		return err
	}
	s.absorbDurableStateLocked(r)
	s.runs = append(s.runs, r)
	if len(s.runs) > s.maxRuns {
		s.runs = s.runs[len(s.runs)-s.maxRuns:]
	}
	s.rebuildIndexesLocked()
	if err := s.persistDurableIndexLocked(); err != nil {
		return fmt.Errorf("persist durable state index: %w", err)
	}
	if info, statErr := os.Stat(s.path); statErr == nil && info.Size() > 64<<20 {
		if compactErr := s.compactLocked(); compactErr != nil {
			return fmt.Errorf("compact state: %w", compactErr)
		}
	}
	return nil
}
func (s *Store) Recent(limit int) []model.RunRecord {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if limit <= 0 || limit > len(s.runs) {
		limit = len(s.runs)
	}
	out := make([]model.RunRecord, limit)
	for i := 0; i < limit; i++ {
		out[i] = s.runs[len(s.runs)-1-i]
	}
	return out
}

func (s *Store) FindRun(runID string) (model.RunRecord, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	i, ok := s.runIndex[runID]
	if !ok || i < 0 || i >= len(s.runs) {
		return model.RunRecord{}, false
	}
	return s.runs[i], true
}

func marksTicketVersionProcessed(r model.RunRecord) bool {
	if r.Outcome == "error" || r.Error != "" || strings.EqualFold(r.Trigger, "scheduled_escalation") {
		return false
	}
	return r.SourceVersion != ""
}

func (s *Store) FindAnalysis(analysisID string) (model.AnalysisRun, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	a, ok := s.analysisIndex[analysisID]
	return a, ok
}

func (s *Store) HasEscalationKey(key string) bool {
	s.mu.RLock()
	defer s.mu.RUnlock()
	_, ok := s.escalationKeys[strings.TrimSpace(key)]
	return ok
}

func (s *Store) load() error {
	f, err := os.Open(s.path)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		return err
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	buf := make([]byte, 64*1024)
	sc.Buffer(buf, 2*1024*1024)
	for sc.Scan() {
		var r model.RunRecord
		if json.Unmarshal(sc.Bytes(), &r) == nil {
			s.absorbDurableStateLocked(r)
			s.runs = append(s.runs, r)
		}
	}
	if len(s.runs) > s.maxRuns {
		s.runs = s.runs[len(s.runs)-s.maxRuns:]
	}
	sort.SliceStable(s.runs, func(i, j int) bool { return s.runs[i].FinishedAt.Before(s.runs[j].FinishedAt) })
	s.rebuildIndexesLocked()
	return sc.Err()
}

func (s *Store) rebuildIndexesLocked() {
	s.runIndex = make(map[string]int, len(s.runs))
	s.analysisIndex = make(map[string]model.AnalysisRun)
	for i, r := range s.runs {
		s.runIndex[r.RunID] = i
		for _, a := range r.Analyses {
			s.analysisIndex[a.AnalysisID] = a
		}
	}
}

func (s *Store) LatestTicketRun(ticketID int64, excludedTriggers ...string) (model.RunRecord, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	excluded := make(map[string]struct{}, len(excludedTriggers))
	for _, trigger := range excludedTriggers {
		excluded[strings.ToLower(strings.TrimSpace(trigger))] = struct{}{}
	}
	for i := len(s.runs) - 1; i >= 0; i-- {
		r := s.runs[i]
		if r.TicketID != ticketID {
			continue
		}
		if _, skip := excluded[strings.ToLower(strings.TrimSpace(r.Trigger))]; skip {
			continue
		}
		return r, true
	}
	return model.RunRecord{}, false
}

func (s *Store) absorbDurableStateLocked(r model.RunRecord) {
	if marksTicketVersionProcessed(r) {
		s.processedVersions[r.TicketID] = r.SourceVersion
	}
	for _, a := range r.Analyses {
		if a.AnalysisType != "escalation" {
			continue
		}
		// New escalation plans persist every successfully executed step independently.
		// This keeps idempotency intact even when a later step in the same plan fails.
		for _, step := range a.Action.Steps {
			if !step.Executed {
				continue
			}
			if key := escalationKeyFromResult(step.Result); key != "" {
				s.escalationKeys[key] = struct{}{}
			}
		}
		// Preserve compatibility with historical single-action audit records.
		if a.Action.Executed {
			if key := escalationKeyFromResult(a.Action.Result); key != "" {
				s.escalationKeys[key] = struct{}{}
			}
		}
	}
}

func escalationKeyFromResult(result string) string {
	parts := map[string]string{}
	for _, part := range strings.Split(result, ";") {
		part = strings.TrimSpace(part)
		for _, name := range []string{"ticket", "level", "action", "target"} {
			prefix := name + "="
			if strings.HasPrefix(part, prefix) && len(part) > len(prefix) {
				parts[name] = part
			}
		}
	}
	if parts["ticket"] == "" || parts["level"] == "" {
		return ""
	}
	key := parts["ticket"] + ";" + parts["level"]
	if parts["action"] != "" {
		key += ";" + parts["action"]
	}
	if parts["target"] != "" {
		key += ";" + parts["target"]
	}
	return key
}

func (s *Store) loadDurableIndex() error {
	b, err := os.ReadFile(s.indexPath)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("read durable state index: %w", err)
	}
	var idx durableIndex
	if err := json.Unmarshal(b, &idx); err != nil {
		return fmt.Errorf("decode durable state index: %w", err)
	}
	for rawID, version := range idx.ProcessedVersions {
		id, err := strconv.ParseInt(rawID, 10, 64)
		if err == nil && id > 0 && version != "" {
			s.processedVersions[id] = version
		}
	}
	for _, key := range idx.EscalationKeys {
		key = strings.TrimSpace(key)
		if key != "" {
			s.escalationKeys[key] = struct{}{}
		}
	}
	return nil
}

func (s *Store) persistDurableIndexLocked() error {
	idx := durableIndex{ProcessedVersions: make(map[string]string, len(s.processedVersions)), EscalationKeys: make([]string, 0, len(s.escalationKeys))}
	for id, version := range s.processedVersions {
		if id > 0 && version != "" {
			idx.ProcessedVersions[strconv.FormatInt(id, 10)] = version
		}
	}
	for key := range s.escalationKeys {
		idx.EscalationKeys = append(idx.EscalationKeys, key)
	}
	sort.Strings(idx.EscalationKeys)
	b, err := json.MarshalIndent(idx, "", "  ")
	if err != nil {
		return err
	}
	tmp := s.indexPath + ".tmp"
	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o640)
	if err != nil {
		return err
	}
	cleanup := func() {
		_ = f.Close()
		_ = os.Remove(tmp)
	}
	if _, err := f.Write(append(b, '\n')); err != nil {
		cleanup()
		return err
	}
	if err := f.Sync(); err != nil {
		cleanup()
		return err
	}
	if err := f.Close(); err != nil {
		_ = os.Remove(tmp)
		return err
	}
	return os.Rename(tmp, s.indexPath)
}

func (s *Store) compactLocked() error {
	tmp := s.path + ".tmp"
	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o640)
	if err != nil {
		return err
	}
	enc := json.NewEncoder(f)
	for _, r := range s.runs {
		if err := enc.Encode(r); err != nil {
			f.Close()
			_ = os.Remove(tmp)
			return err
		}
	}
	if err := f.Sync(); err != nil {
		f.Close()
		_ = os.Remove(tmp)
		return err
	}
	if err := f.Close(); err != nil {
		_ = os.Remove(tmp)
		return err
	}
	return os.Rename(tmp, s.path)
}