Pulse/internal/metrics/store.go

// Package metrics provides persistent storage for time-series metrics data
// using SQLite for durability across restarts.
package metrics

import (
	"database/sql"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/rs/zerolog/log"
	_ "modernc.org/sqlite"
)

// Tier represents the granularity of stored metrics
type Tier string

const (
	TierRaw    Tier = "raw"    // Raw data, ~5s intervals
	TierMinute Tier = "minute" // 1-minute averages
	TierHourly Tier = "hourly" // 1-hour averages
	TierDaily  Tier = "daily"  // 1-day averages
)

// MetricPoint represents a single metric data point
type MetricPoint struct {
	Timestamp time.Time
	Value     float64
	Min       float64 // For aggregated data
	Max       float64 // For aggregated data
}

// StoreConfig holds configuration for the metrics store
type StoreConfig struct {
	DBPath          string
	WriteBufferSize int           // Number of records to buffer before batch write
	FlushInterval   time.Duration // Max time between flushes
	RetentionRaw    time.Duration // How long to keep raw data
	RetentionMinute time.Duration // How long to keep minute data
	RetentionHourly time.Duration // How long to keep hourly data
	RetentionDaily  time.Duration // How long to keep daily data
}

// DefaultConfig returns sensible defaults for metrics storage
func DefaultConfig(dataDir string) StoreConfig {
	return StoreConfig{
		DBPath:          filepath.Join(dataDir, "metrics.db"),
		WriteBufferSize: 100,
		FlushInterval:   5 * time.Second,
		RetentionRaw:    2 * time.Hour,
		RetentionMinute: 24 * time.Hour,
		RetentionHourly: 7 * 24 * time.Hour,
		RetentionDaily:  90 * 24 * time.Hour,
	}
}

// bufferedMetric holds a metric waiting to be written
type bufferedMetric struct {
	resourceType string
	resourceID   string
	metricType   string
	value        float64
	timestamp    time.Time
}

// Store provides persistent metrics storage
type Store struct {
	db     *sql.DB
	config StoreConfig

	// Write buffer
	bufferMu sync.Mutex
	buffer   []bufferedMetric

	// Background workers
	stopCh   chan struct{}
	doneCh   chan struct{}
	stopOnce sync.Once
}

// NewStore creates a new metrics store with the given configuration
func NewStore(config StoreConfig) (*Store, error) {
	// Ensure directory exists
	dir := filepath.Dir(config.DBPath)
	if err := os.MkdirAll(dir, 0755); err != nil {
		return nil, fmt.Errorf("failed to create metrics directory: %w", err)
	}

	// Open database with WAL mode for better concurrent access
	db, err := sql.Open("sqlite", config.DBPath+"?_journal_mode=WAL&_busy_timeout=5000")
	if err != nil {
		return nil, fmt.Errorf("failed to open metrics database: %w", err)
	}

	// Configure connection pool (SQLite works best with single writer)
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	db.SetConnMaxLifetime(0)

	store := &Store{
		db:     db,
		config: config,
		buffer: make([]bufferedMetric, 0, config.WriteBufferSize),
		stopCh: make(chan struct{}),
		doneCh: make(chan struct{}),
	}

	// Initialize schema
	if err := store.initSchema(); err != nil {
		db.Close()
		return nil, fmt.Errorf("failed to initialize schema: %w", err)
	}

	// Start background workers
	go store.backgroundWorker()

	log.Info().
		Str("path", config.DBPath).
		Int("bufferSize", config.WriteBufferSize).
		Msg("Metrics store initialized")

	return store, nil
}

// initSchema creates the database schema if it doesn't exist
func (s *Store) initSchema() error {
	schema := `
		-- Main metrics table
		CREATE TABLE IF NOT EXISTS metrics (
			id INTEGER PRIMARY KEY AUTOINCREMENT,
			resource_type TEXT NOT NULL,
			resource_id TEXT NOT NULL,
			metric_type TEXT NOT NULL,
			value REAL NOT NULL,
			min_value REAL,
			max_value REAL,
			timestamp INTEGER NOT NULL,
			tier TEXT NOT NULL DEFAULT 'raw'
		);

		-- Index for efficient queries by resource and time
		CREATE INDEX IF NOT EXISTS idx_metrics_lookup 
		ON metrics(resource_type, resource_id, metric_type, tier, timestamp);

		-- Index for retention pruning
		CREATE INDEX IF NOT EXISTS idx_metrics_tier_time 
		ON metrics(tier, timestamp);

		-- Metadata table for tracking rollup state
		CREATE TABLE IF NOT EXISTS metrics_meta (
			key TEXT PRIMARY KEY,
			value TEXT NOT NULL
		);
	`

	_, err := s.db.Exec(schema)
	if err != nil {
		return fmt.Errorf("failed to create schema: %w", err)
	}

	log.Debug().Msg("Metrics schema initialized")
	return nil
}

// Write adds a metric to the write buffer
func (s *Store) Write(resourceType, resourceID, metricType string, value float64, timestamp time.Time) {
	s.bufferMu.Lock()
	defer s.bufferMu.Unlock()

	s.buffer = append(s.buffer, bufferedMetric{
		resourceType: resourceType,
		resourceID:   resourceID,
		metricType:   metricType,
		value:        value,
		timestamp:    timestamp,
	})

	// Flush if buffer is full
	if len(s.buffer) >= s.config.WriteBufferSize {
		s.flushLocked()
	}
}

// flush writes buffered metrics to the database (caller must hold bufferMu)
func (s *Store) flushLocked() {
	if len(s.buffer) == 0 {
		return
	}

	// Copy buffer for writing
	toWrite := make([]bufferedMetric, len(s.buffer))
	copy(toWrite, s.buffer)
	s.buffer = s.buffer[:0]

	// Write in background to not block callers
	go s.writeBatch(toWrite)
}

// writeBatch writes a batch of metrics to the database
func (s *Store) writeBatch(metrics []bufferedMetric) {
	if len(metrics) == 0 {
		return
	}

	tx, err := s.db.Begin()
	if err != nil {
		log.Error().Err(err).Msg("Failed to begin metrics transaction")
		return
	}

	stmt, err := tx.Prepare(`
		INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier)
		VALUES (?, ?, ?, ?, ?, 'raw')
	`)
	if err != nil {
		tx.Rollback()
		log.Error().Err(err).Msg("Failed to prepare metrics insert")
		return
	}
	defer stmt.Close()

	for _, m := range metrics {
		_, err := stmt.Exec(m.resourceType, m.resourceID, m.metricType, m.value, m.timestamp.Unix())
		if err != nil {
			log.Warn().Err(err).
				Str("resource", m.resourceID).
				Str("metric", m.metricType).
				Msg("Failed to insert metric")
		}
	}

	if err := tx.Commit(); err != nil {
		log.Error().Err(err).Msg("Failed to commit metrics batch")
		return
	}

	log.Debug().Int("count", len(metrics)).Msg("Wrote metrics batch")
}

// Query retrieves metrics for a resource within a time range
func (s *Store) Query(resourceType, resourceID, metricType string, start, end time.Time) ([]MetricPoint, error) {
	// Select appropriate tier based on time range
	tier := s.selectTier(end.Sub(start))

	rows, err := s.db.Query(`
		SELECT timestamp, value, COALESCE(min_value, value), COALESCE(max_value, value)
		FROM metrics
		WHERE resource_type = ? AND resource_id = ? AND metric_type = ? AND tier = ?
		AND timestamp >= ? AND timestamp <= ?
		ORDER BY timestamp ASC
	`, resourceType, resourceID, metricType, string(tier), start.Unix(), end.Unix())
	if err != nil {
		return nil, fmt.Errorf("failed to query metrics: %w", err)
	}
	defer rows.Close()

	var points []MetricPoint
	for rows.Next() {
		var ts int64
		var p MetricPoint
		if err := rows.Scan(&ts, &p.Value, &p.Min, &p.Max); err != nil {
			log.Warn().Err(err).Msg("Failed to scan metric row")
			continue
		}
		p.Timestamp = time.Unix(ts, 0)
		points = append(points, p)
	}

	return points, rows.Err()
}

// QueryAll retrieves all metric types for a resource within a time range
func (s *Store) QueryAll(resourceType, resourceID string, start, end time.Time) (map[string][]MetricPoint, error) {
	tier := s.selectTier(end.Sub(start))

	rows, err := s.db.Query(`
		SELECT metric_type, timestamp, value, COALESCE(min_value, value), COALESCE(max_value, value)
		FROM metrics
		WHERE resource_type = ? AND resource_id = ? AND tier = ?
		AND timestamp >= ? AND timestamp <= ?
		ORDER BY metric_type, timestamp ASC
	`, resourceType, resourceID, string(tier), start.Unix(), end.Unix())
	if err != nil {
		return nil, fmt.Errorf("failed to query all metrics: %w", err)
	}
	defer rows.Close()

	result := make(map[string][]MetricPoint)
	for rows.Next() {
		var metricType string
		var ts int64
		var p MetricPoint
		if err := rows.Scan(&metricType, &ts, &p.Value, &p.Min, &p.Max); err != nil {
			log.Warn().Err(err).Msg("Failed to scan metric row")
			continue
		}
		p.Timestamp = time.Unix(ts, 0)
		result[metricType] = append(result[metricType], p)
	}

	return result, rows.Err()
}

// selectTier chooses the appropriate data tier based on time range
func (s *Store) selectTier(duration time.Duration) Tier {
	switch {
	case duration <= s.config.RetentionRaw:
		return TierRaw
	case duration <= s.config.RetentionMinute:
		return TierMinute
	case duration <= s.config.RetentionHourly:
		return TierHourly
	default:
		return TierDaily
	}
}

// backgroundWorker runs periodic tasks
func (s *Store) backgroundWorker() {
	defer close(s.doneCh)

	flushTicker := time.NewTicker(s.config.FlushInterval)
	rollupTicker := time.NewTicker(5 * time.Minute)
	retentionTicker := time.NewTicker(1 * time.Hour)

	defer flushTicker.Stop()
	defer rollupTicker.Stop()
	defer retentionTicker.Stop()

	for {
		select {
		case <-s.stopCh:
			// Final flush before stopping
			s.Flush()
			return

		case <-flushTicker.C:
			s.Flush()

		case <-rollupTicker.C:
			s.runRollup()

		case <-retentionTicker.C:
			s.runRetention()
		}
	}
}

// Flush writes any buffered metrics to the database
func (s *Store) Flush() {
	s.bufferMu.Lock()
	defer s.bufferMu.Unlock()
	s.flushLocked()
}

// runRollup aggregates raw data into higher tiers
func (s *Store) runRollup() {
	start := time.Now()

	// Rollup raw -> minute (for data older than 5 minutes)
	s.rollupTier(TierRaw, TierMinute, time.Minute, 5*time.Minute)

	// Rollup minute -> hourly (for data older than 1 hour)
	s.rollupTier(TierMinute, TierHourly, time.Hour, time.Hour)

	// Rollup hourly -> daily (for data older than 24 hours)
	s.rollupTier(TierHourly, TierDaily, 24*time.Hour, 24*time.Hour)

	log.Debug().Dur("duration", time.Since(start)).Msg("Metrics rollup completed")
}

// rollupTier aggregates data from one tier to another
func (s *Store) rollupTier(fromTier, toTier Tier, bucketSize, minAge time.Duration) {
	cutoff := time.Now().Add(-minAge).Unix()
	bucketSecs := int64(bucketSize.Seconds())

	// Find distinct resource/metric combinations that need rollup
	rows, err := s.db.Query(`
		SELECT DISTINCT resource_type, resource_id, metric_type
		FROM metrics
		WHERE tier = ? AND timestamp < ?
	`, string(fromTier), cutoff)
	if err != nil {
		log.Error().Err(err).Str("tier", string(fromTier)).Msg("Failed to find rollup candidates")
		return
	}

	var candidates []struct {
		resourceType string
		resourceID   string
		metricType   string
	}

	for rows.Next() {
		var c struct {
			resourceType string
			resourceID   string
			metricType   string
		}
		if err := rows.Scan(&c.resourceType, &c.resourceID, &c.metricType); err == nil {
			candidates = append(candidates, c)
		}
	}
	rows.Close()

	if len(candidates) == 0 {
		return
	}

	// Process each candidate
	for _, c := range candidates {
		s.rollupCandidate(c.resourceType, c.resourceID, c.metricType, fromTier, toTier, bucketSecs, cutoff)
	}
}

// rollupCandidate aggregates a single resource/metric from one tier to another
func (s *Store) rollupCandidate(resourceType, resourceID, metricType string, fromTier, toTier Tier, bucketSecs, cutoff int64) {
	tx, err := s.db.Begin()
	if err != nil {
		return
	}
	defer tx.Rollback()

	// Aggregate data into buckets
	_, err = tx.Exec(`
		INSERT INTO metrics (resource_type, resource_id, metric_type, value, min_value, max_value, timestamp, tier)
		SELECT 
			resource_type, 
			resource_id, 
			metric_type,
			AVG(value) as value,
			MIN(value) as min_value,
			MAX(value) as max_value,
			(timestamp / ?) * ? as bucket_ts,
			?
		FROM metrics
		WHERE resource_type = ? AND resource_id = ? AND metric_type = ? 
		AND tier = ? AND timestamp < ?
		GROUP BY resource_type, resource_id, metric_type, bucket_ts
	`, bucketSecs, bucketSecs, string(toTier), resourceType, resourceID, metricType, string(fromTier), cutoff)

	if err != nil {
		log.Warn().Err(err).
			Str("resource", resourceID).
			Str("from", string(fromTier)).
			Str("to", string(toTier)).
			Msg("Failed to rollup metrics")
		return
	}

	// Delete rolled-up raw data
	_, err = tx.Exec(`
		DELETE FROM metrics
		WHERE resource_type = ? AND resource_id = ? AND metric_type = ?
		AND tier = ? AND timestamp < ?
	`, resourceType, resourceID, metricType, string(fromTier), cutoff)

	if err != nil {
		log.Warn().Err(err).Msg("Failed to delete rolled-up metrics")
		return
	}

	tx.Commit()
}

// runRetention deletes data older than retention period
func (s *Store) runRetention() {
	start := time.Now()
	now := time.Now()

	// Delete old data for each tier
	tiers := []struct {
		tier      Tier
		retention time.Duration
	}{
		{TierRaw, s.config.RetentionRaw},
		{TierMinute, s.config.RetentionMinute},
		{TierHourly, s.config.RetentionHourly},
		{TierDaily, s.config.RetentionDaily},
	}

	var totalDeleted int64
	for _, t := range tiers {
		cutoff := now.Add(-t.retention).Unix()
		result, err := s.db.Exec(`DELETE FROM metrics WHERE tier = ? AND timestamp < ?`, string(t.tier), cutoff)
		if err != nil {
			log.Warn().Err(err).Str("tier", string(t.tier)).Msg("Failed to prune metrics")
			continue
		}
		if affected, _ := result.RowsAffected(); affected > 0 {
			totalDeleted += affected
		}
	}

	if totalDeleted > 0 {
		log.Info().
			Int64("deleted", totalDeleted).
			Dur("duration", time.Since(start)).
			Msg("Metrics retention cleanup completed")
	}
}

// Close shuts down the store gracefully
func (s *Store) Close() error {
	s.stopOnce.Do(func() {
		close(s.stopCh)
	})

	// Wait for background worker to finish
	select {
	case <-s.doneCh:
	case <-time.After(5 * time.Second):
		log.Warn().Msg("Metrics store shutdown timed out")
	}

	return s.db.Close()
}

// Stats holds metrics store statistics
type Stats struct {
	DBPath        string    `json:"dbPath"`
	DBSize        int64     `json:"dbSize"`
	RawCount      int64     `json:"rawCount"`
	MinuteCount   int64     `json:"minuteCount"`
	HourlyCount   int64     `json:"hourlyCount"`
	DailyCount    int64     `json:"dailyCount"`
	TotalWrites   int64     `json:"totalWrites"`
	BufferSize    int       `json:"bufferSize"`
	LastFlush     time.Time `json:"lastFlush"`
	LastRollup    time.Time `json:"lastRollup"`
	LastRetention time.Time `json:"lastRetention"`
}

// GetStats returns storage statistics
func (s *Store) GetStats() Stats {
	stats := Stats{
		DBPath: s.config.DBPath,
	}

	// Count by tier
	rows, err := s.db.Query(`SELECT tier, COUNT(*) FROM metrics GROUP BY tier`)
	if err == nil {
		defer rows.Close()
		for rows.Next() {
			var tier string
			var count int64
			if err := rows.Scan(&tier, &count); err == nil {
				switch tier {
				case "raw":
					stats.RawCount = count
				case "minute":
					stats.MinuteCount = count
				case "hourly":
					stats.HourlyCount = count
				case "daily":
					stats.DailyCount = count
				}
			}
		}
	}

	// Get database size
	if fi, err := os.Stat(s.config.DBPath); err == nil {
		stats.DBSize = fi.Size()
	}

	// Get buffer size
	s.bufferMu.Lock()
	stats.BufferSize = len(s.buffer)
	s.bufferMu.Unlock()

	return stats
}