bee/audit/internal/app/component_status_db.go

package app

import (
	"encoding/json"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

// ComponentStatusDB is a persistent, append-only store of hardware component health records.
// Records are keyed by component identity strings (e.g. "pcie:0000:c8:00.0", "storage:nvme0n1").
// Once a component is marked Warning or Critical, subsequent OK entries do not downgrade it —
// the component stays at the highest observed severity until explicitly reset.
type ComponentStatusDB struct {
	path    string
	mu      sync.Mutex
	records map[string]*ComponentStatusRecord
}

// ComponentStatusRecord holds the current and historical health of one hardware component.
type ComponentStatusRecord struct {
	ComponentKey  string                 `json:"component_key"`
	Status        string                 `json:"status"` // "OK", "Warning", "Critical", "Unknown"
	LastCheckedAt time.Time              `json:"last_checked_at"`
	LastChangedAt time.Time              `json:"last_changed_at"`
	ErrorSummary  string                 `json:"error_summary,omitempty"`
	History       []ComponentStatusEntry `json:"history"`
}

// ComponentStatusEntry is one observation written to a component's history.
type ComponentStatusEntry struct {
	At     time.Time `json:"at"`
	Status string    `json:"status"`
	Source string    `json:"source"` // e.g. "sat:nvidia", "sat:memory", "watchdog:kmsg"
	Detail string    `json:"detail,omitempty"`
}

// OpenComponentStatusDB opens (or creates) the JSON status DB at path.
func OpenComponentStatusDB(path string) (*ComponentStatusDB, error) {
	db := &ComponentStatusDB{
		path:    path,
		records: make(map[string]*ComponentStatusRecord),
	}
	if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
		return nil, err
	}
	data, err := readFileLimited(path, 10<<20)
	if err != nil && !os.IsNotExist(err) {
		return nil, err
	}
	if len(data) > 0 {
		var records []ComponentStatusRecord
		if err := json.Unmarshal(data, &records); err == nil {
			for i := range records {
				db.records[records[i].ComponentKey] = &records[i]
			}
		}
	}
	return db, nil
}

// Record writes one observation for the given component key.
// source is a short label like "sat:nvidia" or "watchdog:kmsg".
// status is "OK", "Warning", "Critical", or "Unknown".
// OK never downgrades an existing Warning or Critical status.
func (db *ComponentStatusDB) Record(key, source, status, detail string) {
	if db == nil || strings.TrimSpace(key) == "" {
		return
	}
	db.mu.Lock()
	defer db.mu.Unlock()

	db.reloadLocked()

	now := time.Now().UTC()
	rec, exists := db.records[key]
	if !exists {
		rec = &ComponentStatusRecord{ComponentKey: key}
		db.records[key] = rec
	}
	rec.LastCheckedAt = now

	// History records status *transitions*, per the ingest contract
	// (bible-local/docs/hardware-ingest-contract.md: "История переходов
	// статусов" — status_history is a transition log, not a per-poll
	// journal). Skip the append when this source's last recorded status for
	// this key is unchanged, or a continuously-polled component (e.g. the
	// PSU watchdog, every 60s indefinitely) grows an unbounded run of
	// identical "still OK" entries and the file never stops growing.
	// LastCheckedAt above already carries "we still saw this as of now" for
	// a steady-state component, so nothing is lost by not repeating it here.
	if last := lastEntryFromSource(rec.History, source); last == nil || last.Status != status {
		rec.History = append(rec.History, ComponentStatusEntry{At: now, Status: status, Source: source, Detail: detail})
	}

	// Status merge: OK never downgrades Warning/Critical.
	newSev := componentSeverity(status)
	curSev := componentSeverity(rec.Status)
	if newSev > curSev {
		rec.Status = status
		rec.LastChangedAt = now
		rec.ErrorSummary = detail
	} else if rec.Status == "" {
		rec.Status = status
		rec.LastChangedAt = now
	}

	_ = db.saveLocked()
}

// lastEntryFromSource returns the most recent history entry recorded by the
// given source, or nil if that source has never reported for this key.
func lastEntryFromSource(history []ComponentStatusEntry, source string) *ComponentStatusEntry {
	for i := len(history) - 1; i >= 0; i-- {
		if history[i].Source == source {
			return &history[i]
		}
	}
	return nil
}

// Get returns the current record for a component key.
func (db *ComponentStatusDB) Get(key string) (ComponentStatusRecord, bool) {
	if db == nil {
		return ComponentStatusRecord{}, false
	}
	db.mu.Lock()
	defer db.mu.Unlock()
	r, ok := db.records[key]
	if !ok {
		return ComponentStatusRecord{}, false
	}
	return *r, true
}

// All returns a snapshot of all records.
func (db *ComponentStatusDB) All() []ComponentStatusRecord {
	if db == nil {
		return nil
	}
	db.mu.Lock()
	defer db.mu.Unlock()
	out := make([]ComponentStatusRecord, 0, len(db.records))
	for _, r := range db.records {
		out = append(out, *r)
	}
	return out
}

// reloadLocked merges on-disk state into memory before this process applies
// its own update. component-status.json is shared by the long-lived bee-web
// process (writing PSU/kmsg watchdog records roughly every 60s) and each
// short-lived "bee bee-worker" subprocess spawned per SAT task (writing GPU/
// CPU/memory/storage records once on completion) — each holds its own
// in-memory copy backed by the same file. Without this reload, saveLocked
// below would dump the caller's stale in-memory map over the file and erase
// whatever the other process wrote in between: e.g. a GPU SAT task's
// pcie:gpu:nvidia record, written by a worker subprocess, silently vanishing
// the next time the main process's health poller ticks and saves its own
// GPU-blind snapshot. Only keys with a newer LastCheckedAt on disk are
// pulled in, so this process's own pending (not-yet-saved) update for key is
// never discarded by its own reload.
func (db *ComponentStatusDB) reloadLocked() {
	data, err := readFileLimited(db.path, 10<<20)
	if err != nil || len(data) == 0 {
		return
	}
	var onDisk []ComponentStatusRecord
	if err := json.Unmarshal(data, &onDisk); err != nil {
		return
	}
	for i := range onDisk {
		key := onDisk[i].ComponentKey
		if existing, ok := db.records[key]; ok && !onDisk[i].LastCheckedAt.After(existing.LastCheckedAt) {
			continue
		}
		rec := onDisk[i]
		db.records[key] = &rec
	}
}

func (db *ComponentStatusDB) saveLocked() error {
	records := make([]ComponentStatusRecord, 0, len(db.records))
	for _, r := range db.records {
		records = append(records, *r)
	}
	data, err := json.MarshalIndent(records, "", "  ")
	if err != nil {
		return err
	}
	return os.WriteFile(db.path, data, 0644)
}

// componentSeverity returns a numeric severity so higher values win.
func componentSeverity(status string) int {
	switch strings.TrimSpace(status) {
	case "Critical":
		return 3
	case "Warning":
		return 2
	case "OK":
		return 1
	default:
		return 0
	}
}

// ApplySATResultToDB reads a SAT summary.txt from the run directory next to archivePath
// and writes component status records to db for the given SAT target.
// archivePath may be either a bare .tar.gz path or "Archive written to /path/foo.tar.gz".
func ApplySATResultToDB(db *ComponentStatusDB, target, archivePath string) {
	if db == nil || strings.TrimSpace(archivePath) == "" {
		return
	}
	archivePath = extractArchivePath(archivePath)
	if archivePath == "" {
		return
	}
	runDir := strings.TrimSuffix(archivePath, ".tar.gz")
	data, err := os.ReadFile(filepath.Join(runDir, "summary.txt"))
	if err != nil {
		return
	}
	kv := parseSATKV(string(data))
	overall := strings.ToUpper(strings.TrimSpace(kv["overall_status"]))
	if overall == "" {
		return
	}

	source := "sat:" + target
	dbStatus := satStatusToDBStatus(overall)

	// Map SAT target to component keys. GPU targets are keyed by vendor, not
	// by the raw target string: "nvidia" (Check tier) and "nvidia-stress" /
	// "nvidia-targeted-stress" (Load/Burn tier) all exercise the same
	// physical GPUs, so they must share one severity-tracked record — a
	// severity-1 Check run after a severity-3 Load failure must not lose
	// that failure just because it ran more recently. Recording each target
	// under its own key (the previous behavior) also silently broke
	// applyComponentStatusDB below, which expects "pcie:gpu:<vendor>" and
	// otherwise fails to match any real BDF.
	switch target {
	case "nvidia", "nvidia-targeted-stress", "nvidia-compute", "nvidia-targeted-power", "nvidia-pulse",
		"nvidia-interconnect", "nvidia-bandwidth", "nvidia-stress", "nvidia-config":
		db.Record("pcie:gpu:nvidia", source, dbStatus, target+" SAT: "+overall)
	case "amd", "amd-stress", "amd-mem", "amd-bandwidth":
		db.Record("pcie:gpu:amd", source, dbStatus, target+" SAT: "+overall)
	case "memory", "memory-stress", "sat-stress":
		db.Record("memory:all", source, dbStatus, target+" SAT: "+overall)
	case "cpu", "platform-stress":
		db.Record("cpu:all", source, dbStatus, target+" SAT: "+overall)
	case "storage":
		// Try to record per-device if available in summary.
		recordedAny := false
		for key, val := range kv {
			if !strings.HasSuffix(key, "_status") || key == "overall_status" {
				continue
			}
			base := strings.TrimSuffix(key, "_status")
			idx := strings.Index(base, "_")
			if idx <= 0 {
				continue
			}
			devName := base[:idx]
			devStatus := satStatusToDBStatus(strings.ToUpper(strings.TrimSpace(val)))
			db.Record("storage:"+devName, source, devStatus, "storage SAT: "+val)
			recordedAny = true
		}
		if !recordedAny {
			db.Record("storage:all", source, dbStatus, "storage SAT: "+overall)
		}
	}
}

func satStatusToDBStatus(overall string) string {
	switch overall {
	case "OK":
		return "OK"
	case "FAILED":
		return "Warning"
	case "PARTIAL", "UNSUPPORTED":
		return "Unknown"
	default:
		return "Unknown"
	}
}

// ExtractArchivePath extracts a bare path from a string that may be
// "Archive written to /path/to/run-dir" or already a bare path.
func ExtractArchivePath(s string) string {
	return extractArchivePath(s)
}

// ReadSATOverallStatus reads the overall_status value from the summary.txt
// file located in the run directory alongside archivePath.
// Returns "" if the file cannot be read.
func ReadSATOverallStatus(archivePath string) string {
	if strings.TrimSpace(archivePath) == "" {
		return ""
	}
	runDir := strings.TrimSuffix(archivePath, ".tar.gz")
	data, err := os.ReadFile(filepath.Join(runDir, "summary.txt"))
	if err != nil {
		return ""
	}
	kv := parseSATKV(string(data))
	return strings.ToUpper(strings.TrimSpace(kv["overall_status"]))
}

func extractArchivePath(s string) string {
	s = strings.TrimSpace(s)
	if rest, ok := strings.CutPrefix(s, "Archive written to "); ok {
		return strings.TrimSpace(rest)
	}
	return s
}

func parseSATKV(raw string) map[string]string {
	kv := make(map[string]string)
	for _, line := range strings.Split(raw, "\n") {
		k, v, ok := strings.Cut(strings.TrimSpace(line), "=")
		if ok {
			kv[strings.TrimSpace(k)] = strings.TrimSpace(v)
		}
	}
	return kv
}