132 lines
3.1 KiB
Go
132 lines
3.1 KiB
Go
package state
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"sync"
|
|
|
|
"github.com/example/glpi-ai-agent/internal/model"
|
|
)
|
|
|
|
type Store struct {
|
|
mu sync.RWMutex
|
|
path string
|
|
processed map[string]struct{}
|
|
runs []model.RunRecord
|
|
maxRuns int
|
|
}
|
|
|
|
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"), processed: map[string]struct{}{}, 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.load(); err != nil {
|
|
return nil, err
|
|
}
|
|
return s, nil
|
|
}
|
|
func key(id int64, version string) string { return fmtKey(id, version) }
|
|
func fmtKey(id int64, version string) string {
|
|
b, _ := json.Marshal([]any{id, version})
|
|
return string(b)
|
|
}
|
|
func (s *Store) Seen(id int64, version string) bool {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
_, ok := s.processed[key(id, version)]
|
|
return ok
|
|
}
|
|
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
|
|
}
|
|
if r.Outcome != "error" && r.Error == "" {
|
|
s.processed[key(r.TicketID, r.SourceVersion)] = struct{}{}
|
|
}
|
|
s.runs = append(s.runs, r)
|
|
if len(s.runs) > s.maxRuns {
|
|
s.runs = s.runs[len(s.runs)-s.maxRuns:]
|
|
}
|
|
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()
|
|
for i := len(s.runs) - 1; i >= 0; i-- {
|
|
if s.runs[i].RunID == runID {
|
|
return s.runs[i], true
|
|
}
|
|
}
|
|
return model.RunRecord{}, false
|
|
}
|
|
|
|
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.runs = append(s.runs, r)
|
|
if r.Outcome != "error" && r.Error == "" {
|
|
s.processed[key(r.TicketID, r.SourceVersion)] = struct{}{}
|
|
}
|
|
}
|
|
}
|
|
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) })
|
|
return sc.Err()
|
|
}
|