EvoBGP/internal/pgmonitor/snapshot.go

package pgmonitor

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
)

type snapshotRow struct {
	ID          string
	CollectedAt time.Time
	Payload     json.RawMessage
}

func (s *Service) loadSnapshot(ctx context.Context, id string, maxAge time.Duration) (snapshotRow, bool, error) {
	var row snapshotRow
	err := s.pool.QueryRow(ctx, `
		SELECT id, collected_at, payload_json
		FROM postgres_monitor_snapshot
		WHERE id = $1 AND collected_at >= $2`,
		id, time.Now().UTC().Add(-maxAge)).Scan(&row.ID, &row.CollectedAt, &row.Payload)
	if err != nil {
		return snapshotRow{}, false, nil
	}
	return row, true, nil
}

func (s *Service) UpsertSnapshot(ctx context.Context, id string, payload any) error {
	if s == nil || s.pool == nil {
		return fmt.Errorf("pgmonitor: postgres not configured")
	}
	b, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	_, err = s.pool.Exec(ctx, `
		INSERT INTO postgres_monitor_snapshot (id, collected_at, payload_json)
		VALUES ($1, now(), $2::jsonb)
		ON CONFLICT (id) DO UPDATE SET collected_at = EXCLUDED.collected_at, payload_json = EXCLUDED.payload_json`,
		id, string(b))
	return err
}

func decodePayload(raw json.RawMessage, dest any) error {
	return json.Unmarshal(raw, dest)
}

// RefreshMetricsSnapshot stores overview and tables for heavy reads.
func (s *Service) RefreshMetricsSnapshot(ctx context.Context) error {
	ov, err := s.fetchOverview(ctx)
	if err != nil {
		return err
	}
	if err := s.UpsertSnapshot(ctx, "overview", ov); err != nil {
		return err
	}
	tables, err := queryTables(ctx, s.pool, 50)
	if err != nil {
		return err
	}
	return s.UpsertSnapshot(ctx, "tables", tables)
}

// AggregateSlowQueries stores top statements snapshot.
func (s *Service) AggregateSlowQueries(ctx context.Context, limit int) error {
	if !s.statementsQueryable(ctx) {
		return s.UpsertSnapshot(ctx, "slow_queries", []QueryStat{})
	}
	items, err := queryTopStatements(ctx, s.pool, clampLimit(limit, 20, 100))
	if err != nil {
		if isPgStatStatementsUnavailable(err) {
			s.markStatementsUnavailable()
			return s.UpsertSnapshot(ctx, "slow_queries", []QueryStat{})
		}
		return err
	}
	return s.UpsertSnapshot(ctx, "slow_queries", items)
}

// EstimateTableBloat refreshes bloat heuristics on tables snapshot.
func (s *Service) EstimateTableBloat(ctx context.Context) error {
	tables, err := queryTables(ctx, s.pool, 100)
	if err != nil {
		return err
	}
	return s.UpsertSnapshot(ctx, "table_bloat", tables)
}

// AnalyzeIndexUsage stores unused indexes.
func (s *Service) AnalyzeIndexUsage(ctx context.Context) error {
	rows, err := s.pool.Query(ctx, `
		SELECT indexrelname, idx_scan, pg_relation_size(indexrelid)
		FROM pg_stat_user_indexes
		WHERE schemaname = 'public' AND idx_scan = 0
		ORDER BY pg_relation_size(indexrelid) DESC
		LIMIT 50`)
	if err != nil {
		return fmt.Errorf("pgmonitor: index usage: %w", err)
	}
	defer rows.Close()
	type unused struct {
		Index     string `json:"index"`
		IdxScan   int64  `json:"idx_scan"`
		SizeBytes int64  `json:"size_bytes"`
	}
	var items []unused
	for rows.Next() {
		var u unused
		if err := rows.Scan(&u.Index, &u.IdxScan, &u.SizeBytes); err != nil {
			return err
		}
		items = append(items, u)
	}
	return s.UpsertSnapshot(ctx, "unused_indexes", items)
}

// DetectAutovacuumLag stores tables with high dead tuple ratio.
func (s *Service) DetectAutovacuumLag(ctx context.Context) error {
	rows, err := s.pool.Query(ctx, `
		SELECT relname, n_dead_tup, last_autovacuum,
		       CASE WHEN n_live_tup + n_dead_tup > 0
		            THEN round(n_dead_tup::numeric / (n_live_tup + n_dead_tup), 4) ELSE 0 END
		FROM pg_stat_user_tables
		WHERE schemaname = 'public' AND n_dead_tup > 1000
		ORDER BY n_dead_tup DESC
		LIMIT 30`)
	if err != nil {
		return fmt.Errorf("pgmonitor: autovacuum lag: %w", err)
	}
	defer rows.Close()
	type lagRow struct {
		Relname        string     `json:"relname"`
		DeadTuples     int64      `json:"n_dead_tup"`
		LastAutovacuum *time.Time `json:"last_autovacuum,omitempty"`
		Ratio          float64    `json:"ratio"`
	}
	var items []lagRow
	for rows.Next() {
		var r lagRow
		if err := rows.Scan(&r.Relname, &r.DeadTuples, &r.LastAutovacuum, &r.Ratio); err != nil {
			return err
		}
		items = append(items, r)
	}
	return s.UpsertSnapshot(ctx, "autovacuum_lag", items)
}

// RunPeriodicAnalyzerJobs runs all snapshot analyzers (for scheduler).
func RunPeriodicAnalyzerJobs(ctx context.Context, pool *pgxpool.Pool) {
	s := NewService(pool)
	if s == nil {
		return
	}
	_ = s.RefreshMetricsSnapshot(ctx)
	_ = s.AggregateSlowQueries(ctx, 30)
	_ = s.EstimateTableBloat(ctx)
	_ = s.AnalyzeIndexUsage(ctx)
	_ = s.DetectAutovacuumLag(ctx)
}