bee/audit/internal/webui/metricsdb.go

package webui

import (
	"database/sql"
	"encoding/csv"
	"io"
	"math"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"bee/audit/internal/platform"
	_ "modernc.org/sqlite"
)

const metricsDBPath = "/appdata/bee/metrics.db"

// MetricsDB persists live metric samples to SQLite.
type MetricsDB struct {
	db    *sql.DB
	clock seqClock
}

// seqClock generates the timestamps stored on metric rows. It is seeded once
// from the wall clock (or the newest persisted row) and thereafter only ever
// advances by the *monotonic* elapsed time between Write calls. A system-clock
// step — an NTP correction, a timezone change, a manual `date` — can therefore
// never punch a gap into the history or send timestamps backwards, which is
// what used to smear every chart into a single diagonal line after the clock
// was changed. Charts count backwards from the newest sample when rendering,
// so the absolute value of the seed does not matter; only the spacing does.
type seqClock struct {
	mu       sync.Mutex
	lastReal time.Time // wall reading from the previous tick (carries a monotonic component)
	cur      float64   // current device time, unix-seconds
	lastInt  int64     // last integer ts handed out, to keep the PK strictly increasing
}

func (c *seqClock) seed(base float64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.lastReal = time.Now()
	c.cur = base
	// One below floor(base) so the first tick (near-zero elapsed) can still
	// hand out floor(base) itself instead of being bumped forward.
	c.lastInt = int64(math.Floor(base)) - 1
}

// tick advances the clock by the monotonic time elapsed since the previous
// call and returns the next row timestamp (unix-seconds, strictly increasing).
func (c *seqClock) tick() int64 {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.advanceLocked()
	ts := int64(math.Floor(c.cur)) // truncate, matching time.Time.Unix semantics
	if ts <= c.lastInt {
		ts = c.lastInt + 1
	}
	c.lastInt = ts
	return ts
}

// deviceNow returns the current device time, advancing the clock so the
// wall↔device relationship stays anchored to the latest observation.
func (c *seqClock) deviceNow() float64 {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.advanceLocked()
	return c.cur
}

// advanceLocked moves cur forward by the monotonic time elapsed since the
// previous observation. c.mu must be held.
func (c *seqClock) advanceLocked() {
	now := time.Now()
	elapsed := now.Sub(c.lastReal).Seconds() // monotonic: immune to wall-clock steps
	if elapsed < 0 || elapsed > 24*3600 {
		elapsed = 0
	}
	c.lastReal = now
	c.cur += elapsed
}

func (m *MetricsDB) Close() error {
	if m == nil || m.db == nil {
		return nil
	}
	return m.db.Close()
}

// openMetricsDB opens (or creates) the metrics database at the given path.
func openMetricsDB(path string) (*MetricsDB, error) {
	if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
		return nil, err
	}
	db, err := sql.Open("sqlite", path+"?_journal=WAL&_busy_timeout=5000")
	if err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(1)
	if err := initMetricsSchema(db); err != nil {
		_ = db.Close()
		return nil, err
	}
	m := &MetricsDB{db: db}
	var maxTS sql.NullInt64
	_ = db.QueryRow(`SELECT MAX(ts) FROM sys_metrics`).Scan(&maxTS)
	if maxTS.Valid && maxTS.Int64 > 0 {
		m.clock.seed(float64(maxTS.Int64))
	} else {
		m.clock.seed(float64(time.Now().UnixNano()) / 1e9)
	}
	return m, nil
}

func initMetricsSchema(db *sql.DB) error {
	_, err := db.Exec(`
CREATE TABLE IF NOT EXISTS sys_metrics (
  ts           INTEGER NOT NULL,
  cpu_load_pct REAL,
  mem_load_pct REAL,
  power_w      REAL,
  power_source TEXT,
  power_mode   TEXT,
  power_reason TEXT,
  PRIMARY KEY (ts)
);
CREATE TABLE IF NOT EXISTS gpu_metrics (
  ts            INTEGER NOT NULL,
  gpu_index     INTEGER NOT NULL,
  temp_c        REAL,
  usage_pct     REAL,
  mem_usage_pct REAL,
  power_w       REAL,
  clock_mhz     REAL,
  mem_clock_mhz REAL,
  PRIMARY KEY (ts, gpu_index)
);
CREATE TABLE IF NOT EXISTS fan_metrics (
  ts   INTEGER NOT NULL,
  name TEXT NOT NULL,
  rpm  REAL,
  PRIMARY KEY (ts, name)
);
CREATE TABLE IF NOT EXISTS temp_metrics (
  ts      INTEGER NOT NULL,
  name    TEXT NOT NULL,
  grp     TEXT NOT NULL,
  celsius REAL,
  PRIMARY KEY (ts, name)
);
`)
	if err != nil {
		return err
	}
	if err := ensureMetricsColumn(db, "gpu_metrics", "clock_mhz", "REAL"); err != nil {
		return err
	}
	if err := ensureMetricsColumn(db, "gpu_metrics", "mem_clock_mhz", "REAL"); err != nil {
		return err
	}
	if err := ensureMetricsColumn(db, "sys_metrics", "power_source", "TEXT"); err != nil {
		return err
	}
	if err := ensureMetricsColumn(db, "sys_metrics", "power_mode", "TEXT"); err != nil {
		return err
	}
	return ensureMetricsColumn(db, "sys_metrics", "power_reason", "TEXT")
}

func ensureMetricsColumn(db *sql.DB, table, column, definition string) error {
	rows, err := db.Query("PRAGMA table_info(" + table + ")")
	if err != nil {
		return err
	}
	defer rows.Close()

	for rows.Next() {
		var cid int
		var name, ctype string
		var notNull, pk int
		var dflt sql.NullString
		if err := rows.Scan(&cid, &name, &ctype, &notNull, &dflt, &pk); err != nil {
			return err
		}
		if strings.EqualFold(name, column) {
			return nil
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	_, err = db.Exec("ALTER TABLE " + table + " ADD COLUMN " + column + " " + definition)
	return err
}

// Write inserts one sample into all relevant tables. The row timestamp is
// taken from the monotonic seqClock, not from s.Timestamp, so a system-clock
// change never corrupts the stored series.
func (m *MetricsDB) Write(s platform.LiveMetricSample) error {
	return m.writeAt(m.clock.tick(), s)
}

// writeAt inserts one sample at an explicit timestamp. Production code uses
// Write; writeAt exists so tests can lay down a series with known spacing.
func (m *MetricsDB) writeAt(ts int64, s platform.LiveMetricSample) error {
	tx, err := m.db.Begin()
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	_, err = tx.Exec(
		`INSERT OR REPLACE INTO sys_metrics(ts,cpu_load_pct,mem_load_pct,power_w,power_source,power_mode,power_reason) VALUES(?,?,?,?,?,?,?)`,
		ts, s.CPULoadPct, s.MemLoadPct, s.PowerW, s.PowerSource, s.PowerMode, s.PowerReason,
	)
	if err != nil {
		return err
	}
	for _, g := range s.GPUs {
		_, err = tx.Exec(
			`INSERT OR REPLACE INTO gpu_metrics(ts,gpu_index,temp_c,usage_pct,mem_usage_pct,power_w,clock_mhz,mem_clock_mhz) VALUES(?,?,?,?,?,?,?,?)`,
			ts, g.GPUIndex, g.TempC, g.UsagePct, g.MemUsagePct, g.PowerW, g.ClockMHz, g.MemClockMHz,
		)
		if err != nil {
			return err
		}
	}
	for _, f := range s.Fans {
		_, err = tx.Exec(
			`INSERT OR REPLACE INTO fan_metrics(ts,name,rpm) VALUES(?,?,?)`,
			ts, f.Name, f.RPM,
		)
		if err != nil {
			return err
		}
	}
	for _, t := range s.Temps {
		_, err = tx.Exec(
			`INSERT OR REPLACE INTO temp_metrics(ts,name,grp,celsius) VALUES(?,?,?,?)`,
			ts, t.Name, t.Group, t.Celsius,
		)
		if err != nil {
			return err
		}
	}
	return tx.Commit()
}

// Downsample reduces density of old metrics rows to 1 sample per minute.
// Rows older than downsampleAge (but within retain) are thinned; rows newer
// than downsampleAge keep full 5-second resolution. For each 60-second bucket
// the row with the smallest ts is kept; the rest are deleted. This trims ~92 %
// of rows in that window while preserving the overall shape of every chart.
//
// The window is measured back from the newest sample (the device clock), not
// from the wall clock, so it is unaffected by system-clock changes.
//
// Called hourly by the metrics collector background goroutine.
func (m *MetricsDB) Downsample(downsampleAge, retain time.Duration) error {
	if m == nil || m.db == nil {
		return nil
	}
	now := m.clock.deviceNow()
	start := int64(now - retain.Seconds())
	end := int64(now - downsampleAge.Seconds())
	if end <= start {
		return nil
	}
	// For each table: delete rows in [start, end) whose ts is NOT the minimum
	// ts in its 60-second bucket (ts/60 integer division = bucket ID).
	for _, table := range []string{"sys_metrics", "gpu_metrics", "fan_metrics", "temp_metrics"} {
		_, err := m.db.Exec(`
DELETE FROM `+table+` WHERE ts >= ? AND ts < ?
  AND ts NOT IN (
    SELECT MIN(ts) FROM `+table+`
    WHERE ts >= ? AND ts < ?
    GROUP BY ts / 60
  )`, start, end, start, end)
		if err != nil {
			return err
		}
	}
	return nil
}

// Prune deletes rows older than retain (measured back from the newest sample,
// so it is unaffected by system-clock changes) from every metrics table.
// Called hourly by the metrics collector to keep the DB size bounded.
func (m *MetricsDB) Prune(retain time.Duration) error {
	if m == nil || m.db == nil {
		return nil
	}
	cutTS := int64(m.clock.deviceNow() - retain.Seconds())
	for _, table := range []string{"sys_metrics", "gpu_metrics", "fan_metrics", "temp_metrics"} {
		if _, err := m.db.Exec("DELETE FROM "+table+" WHERE ts < ?", cutTS); err != nil {
			return err
		}
	}
	_, _ = m.db.Exec("PRAGMA wal_checkpoint(TRUNCATE)")
	return nil
}

// LoadRecent returns up to n samples in chronological order (oldest first).
func (m *MetricsDB) LoadRecent(n int) ([]platform.LiveMetricSample, error) {
	return m.loadSamples(`SELECT ts,cpu_load_pct,mem_load_pct,power_w,IFNULL(power_source,''),IFNULL(power_mode,''),IFNULL(power_reason,'') FROM (SELECT ts,cpu_load_pct,mem_load_pct,power_w,power_source,power_mode,power_reason FROM sys_metrics ORDER BY ts DESC LIMIT ?) ORDER BY ts`, n)
}

// LoadAll returns all persisted samples in chronological order (oldest first).
func (m *MetricsDB) LoadAll() ([]platform.LiveMetricSample, error) {
	return m.loadSamples(`SELECT ts,cpu_load_pct,mem_load_pct,power_w,IFNULL(power_source,''),IFNULL(power_mode,''),IFNULL(power_reason,'') FROM sys_metrics ORDER BY ts`, nil)
}

// LoadBetween returns samples in chronological order within the given time window.
func (m *MetricsDB) LoadBetween(start, end time.Time) ([]platform.LiveMetricSample, error) {
	if m == nil {
		return nil, nil
	}
	if start.IsZero() || end.IsZero() {
		return nil, nil
	}
	if end.Before(start) {
		start, end = end, start
	}
	// Row timestamps track the monotonic device clock, which is seeded from the
	// wall clock and advances at the real rate, so ts stays aligned with wall
	// time for the task-window correlation done here. (System-clock *steps* are
	// absorbed by the device clock rather than corrupting the series.)
	return m.loadSamples(
		`SELECT ts,cpu_load_pct,mem_load_pct,power_w,IFNULL(power_source,''),IFNULL(power_mode,''),IFNULL(power_reason,'') FROM sys_metrics WHERE ts>=? AND ts<=? ORDER BY ts`,
		start.Unix(), end.Unix(),
	)
}

// loadSamples reconstructs LiveMetricSample rows from the normalized tables.
func (m *MetricsDB) loadSamples(query string, args ...any) ([]platform.LiveMetricSample, error) {
	rows, err := m.db.Query(query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	type sysRow struct {
		ts            int64
		cpu, mem, pwr float64
		powerSource   string
		powerMode     string
		powerReason   string
	}
	var sysRows []sysRow
	for rows.Next() {
		var r sysRow
		if err := rows.Scan(&r.ts, &r.cpu, &r.mem, &r.pwr, &r.powerSource, &r.powerMode, &r.powerReason); err != nil {
			continue
		}
		sysRows = append(sysRows, r)
	}
	if len(sysRows) == 0 {
		return nil, nil
	}
	// Collect min/max ts for range query
	minTS := sysRows[0].ts
	maxTS := sysRows[len(sysRows)-1].ts

	// Load GPU rows in range
	type gpuKey struct {
		ts  int64
		idx int
	}
	gpuData := map[gpuKey]platform.GPUMetricRow{}
	gRows, err := m.db.Query(
		`SELECT ts,gpu_index,temp_c,usage_pct,mem_usage_pct,power_w,IFNULL(clock_mhz,0),IFNULL(mem_clock_mhz,0) FROM gpu_metrics WHERE ts>=? AND ts<=? ORDER BY ts,gpu_index`,
		minTS, maxTS,
	)
	if err == nil {
		defer gRows.Close()
		for gRows.Next() {
			var ts int64
			var g platform.GPUMetricRow
			if err := gRows.Scan(&ts, &g.GPUIndex, &g.TempC, &g.UsagePct, &g.MemUsagePct, &g.PowerW, &g.ClockMHz, &g.MemClockMHz); err == nil {
				gpuData[gpuKey{ts, g.GPUIndex}] = g
			}
		}
	}

	// Load fan rows in range
	type fanKey struct {
		ts   int64
		name string
	}
	fanData := map[fanKey]float64{}
	fRows, err := m.db.Query(
		`SELECT ts,name,rpm FROM fan_metrics WHERE ts>=? AND ts<=?`, minTS, maxTS,
	)
	if err == nil {
		defer fRows.Close()
		for fRows.Next() {
			var ts int64
			var name string
			var rpm float64
			if err := fRows.Scan(&ts, &name, &rpm); err == nil {
				fanData[fanKey{ts, name}] = rpm
			}
		}
	}

	// Load temp rows in range
	type tempKey struct {
		ts   int64
		name string
	}
	tempData := map[tempKey]platform.TempReading{}
	tRows, err := m.db.Query(
		`SELECT ts,name,grp,celsius FROM temp_metrics WHERE ts>=? AND ts<=?`, minTS, maxTS,
	)
	if err == nil {
		defer tRows.Close()
		for tRows.Next() {
			var ts int64
			var t platform.TempReading
			if err := tRows.Scan(&ts, &t.Name, &t.Group, &t.Celsius); err == nil {
				tempData[tempKey{ts, t.Name}] = t
			}
		}
	}

	// Collect unique GPU indices and fan/temp names from loaded data.
	// Sort each list so that sample reconstruction is deterministic regardless
	// of Go's non-deterministic map iteration order.
	seenGPU := map[int]bool{}
	var gpuIndices []int
	for k := range gpuData {
		if !seenGPU[k.idx] {
			seenGPU[k.idx] = true
			gpuIndices = append(gpuIndices, k.idx)
		}
	}
	sort.Ints(gpuIndices)

	seenFan := map[string]bool{}
	var fanNames []string
	for k := range fanData {
		if !seenFan[k.name] {
			seenFan[k.name] = true
			fanNames = append(fanNames, k.name)
		}
	}
	sort.Strings(fanNames)

	seenTemp := map[string]bool{}
	var tempNames []string
	for k := range tempData {
		if !seenTemp[k.name] {
			seenTemp[k.name] = true
			tempNames = append(tempNames, k.name)
		}
	}
	sort.Strings(tempNames)

	samples := make([]platform.LiveMetricSample, len(sysRows))
	for i, r := range sysRows {
		s := platform.LiveMetricSample{
			Timestamp:   time.Unix(r.ts, 0).UTC(),
			CPULoadPct:  r.cpu,
			MemLoadPct:  r.mem,
			PowerW:      r.pwr,
			PowerSource: r.powerSource,
			PowerMode:   r.powerMode,
			PowerReason: r.powerReason,
		}
		for _, idx := range gpuIndices {
			if g, ok := gpuData[gpuKey{r.ts, idx}]; ok {
				s.GPUs = append(s.GPUs, g)
			}
		}
		for _, name := range fanNames {
			if rpm, ok := fanData[fanKey{r.ts, name}]; ok {
				s.Fans = append(s.Fans, platform.FanReading{Name: name, RPM: rpm})
			}
		}
		for _, name := range tempNames {
			if t, ok := tempData[tempKey{r.ts, name}]; ok {
				s.Temps = append(s.Temps, t)
			}
		}
		samples[i] = s
	}
	return samples, nil
}

// ExportCSV writes all sys+gpu data as CSV to w.
func (m *MetricsDB) ExportCSV(w io.Writer) error {
	rows, err := m.db.Query(`
		SELECT s.ts, s.cpu_load_pct, s.mem_load_pct, s.power_w,
		       g.gpu_index, g.temp_c, g.usage_pct, g.mem_usage_pct, g.power_w,
		       g.clock_mhz, g.mem_clock_mhz
		FROM sys_metrics s
		LEFT JOIN gpu_metrics g ON g.ts = s.ts
		ORDER BY s.ts, g.gpu_index
	`)
	if err != nil {
		return err
	}
	defer rows.Close()

	cw := csv.NewWriter(w)
	_ = cw.Write([]string{"ts", "cpu_load_pct", "mem_load_pct", "sys_power_w", "gpu_index", "gpu_temp_c", "gpu_usage_pct", "gpu_mem_pct", "gpu_power_w", "gpu_clock_mhz", "gpu_mem_clock_mhz"})
	for rows.Next() {
		var ts int64
		var cpu, mem, pwr float64
		var gpuIdx sql.NullInt64
		var gpuTemp, gpuUse, gpuMem, gpuPow, gpuClock, gpuMemClock sql.NullFloat64
		if err := rows.Scan(&ts, &cpu, &mem, &pwr, &gpuIdx, &gpuTemp, &gpuUse, &gpuMem, &gpuPow, &gpuClock, &gpuMemClock); err != nil {
			continue
		}
		row := []string{
			strconv.FormatInt(ts, 10),
			strconv.FormatFloat(cpu, 'f', 2, 64),
			strconv.FormatFloat(mem, 'f', 2, 64),
			strconv.FormatFloat(pwr, 'f', 1, 64),
		}
		if gpuIdx.Valid {
			row = append(row,
				strconv.FormatInt(gpuIdx.Int64, 10),
				strconv.FormatFloat(gpuTemp.Float64, 'f', 1, 64),
				strconv.FormatFloat(gpuUse.Float64, 'f', 1, 64),
				strconv.FormatFloat(gpuMem.Float64, 'f', 1, 64),
				strconv.FormatFloat(gpuPow.Float64, 'f', 1, 64),
				strconv.FormatFloat(gpuClock.Float64, 'f', 1, 64),
				strconv.FormatFloat(gpuMemClock.Float64, 'f', 1, 64),
			)
		} else {
			row = append(row, "", "", "", "", "", "", "")
		}
		_ = cw.Write(row)
	}
	cw.Flush()
	return cw.Error()
}