zzet--gortex/internal/telemetry/store.go

package telemetry

import (
	"encoding/json"
	"errors"
	"io/fs"
	"os"
	"path/filepath"
	"sort"
	"strings"
)

// Store persists daily rollups as one JSON file per UTC day under a telemetry
// directory (rollup-YYYY-MM-DD.json). Per-day files let a completed day be
// read, sent, and deleted independently of the day still accumulating.
type Store struct {
	dir string
}

// NewStore roots a store at dir; the directory is created on first write.
func NewStore(dir string) *Store { return &Store{dir: dir} }

const (
	rollupFilePrefix = "rollup-"
	rollupFileSuffix = ".json"
)

func (s *Store) pathForDay(day string) string {
	return filepath.Join(s.dir, rollupFilePrefix+day+rollupFileSuffix)
}

// Load reads a day's rollup, returning an empty rollup (not an error) when the
// day has no file yet — the natural starting point for accumulation.
func (s *Store) Load(day string) (*Rollup, error) {
	b, err := os.ReadFile(s.pathForDay(day))
	if errors.Is(err, fs.ErrNotExist) {
		return &Rollup{Day: day, Counts: map[string]int{}}, nil
	}
	if err != nil {
		return nil, err
	}
	r := &Rollup{}
	if err := json.Unmarshal(b, r); err != nil {
		return nil, err
	}
	if r.Counts == nil {
		r.Counts = map[string]int{}
	}
	return r, nil
}

// Save atomically writes a rollup's day file (temp file + rename), creating the
// telemetry directory on demand. Atomicity guarantees a reader (or a crash)
// never sees a half-written rollup.
func (s *Store) Save(r *Rollup) error {
	if err := os.MkdirAll(s.dir, 0o755); err != nil {
		return err
	}
	b, err := json.MarshalIndent(r, "", "  ")
	if err != nil {
		return err
	}
	tmp, err := os.CreateTemp(s.dir, rollupFilePrefix+r.Day+".*.tmp")
	if err != nil {
		return err
	}
	tmpName := tmp.Name()
	defer func() { _ = os.Remove(tmpName) }() // no-op after a successful rename
	if _, err := tmp.Write(b); err != nil {
		_ = tmp.Close()
		return err
	}
	if err := tmp.Close(); err != nil {
		return err
	}
	return os.Rename(tmpName, s.pathForDay(r.Day))
}

// Merge loads a day, folds r into it, and saves — the recorder's accumulate
// step, so concurrent processes converge on the same day's totals.
func (s *Store) Merge(r *Rollup) error {
	existing, err := s.Load(r.Day)
	if err != nil {
		return err
	}
	existing.Merge(r)
	return s.Save(existing)
}

// Days returns every persisted day, sorted ascending.
func (s *Store) Days() ([]string, error) {
	entries, err := os.ReadDir(s.dir)
	if errors.Is(err, fs.ErrNotExist) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	var days []string
	for _, e := range entries {
		n := e.Name()
		if strings.HasPrefix(n, rollupFilePrefix) && strings.HasSuffix(n, rollupFileSuffix) {
			days = append(days, strings.TrimSuffix(strings.TrimPrefix(n, rollupFilePrefix), rollupFileSuffix))
		}
	}
	sort.Strings(days)
	return days, nil
}

// LoadCompleted returns every rollup whose day is strictly before today (UTC).
// Today is still accumulating, so only completed days are safe to send — this
// is how volume scales with active machines, not with tool calls.
func (s *Store) LoadCompleted(today string) ([]*Rollup, error) {
	days, err := s.Days()
	if err != nil {
		return nil, err
	}
	var out []*Rollup
	for _, d := range days {
		if d >= today {
			continue // today or a future-dated file: still open
		}
		r, err := s.Load(d)
		if err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, nil
}

// Delete removes a day's file (after a successful send). A missing file is not
// an error — delete is idempotent.
func (s *Store) Delete(day string) error {
	err := os.Remove(s.pathForDay(day))
	if errors.Is(err, fs.ErrNotExist) {
		return nil
	}
	return err
}