rmm2/backend/db/postgres.go

package db

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"strconv"
	"strings"
	"time"

	"github.com/cynfo/rmm-backend/models"

	_ "github.com/jackc/pgx/v5/stdlib"
)

type Database struct {
	db *sql.DB
}

type DBConfig struct {
	Host     string `yaml:"host"`
	Port     int    `yaml:"port"`
	User     string `yaml:"user"`
	Password string `yaml:"password"`
	DBName   string `yaml:"dbname"`
	SSLMode  string `yaml:"sslmode"`
}

func New(cfg DBConfig) (*Database, error) {
	dsn := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s",
		cfg.Host, cfg.Port, cfg.User, cfg.Password, cfg.DBName, cfg.SSLMode)

	db, err := sql.Open("pgx", dsn)
	if err != nil {
		return nil, fmt.Errorf("Datenbank oeffnen: %w", err)
	}

	// Connection Pool konfigurieren
	db.SetMaxOpenConns(25)
	db.SetMaxIdleConns(5)
	db.SetConnMaxLifetime(5 * time.Minute)

	// Verbindung testen
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := db.PingContext(ctx); err != nil {
		return nil, fmt.Errorf("Datenbank-Verbindung fehlgeschlagen: %w", err)
	}

	d := &Database{db: db}
	if err := d.migrate(); err != nil {
		return nil, fmt.Errorf("Migration fehlgeschlagen: %w", err)
	}

	log.Printf("PostgreSQL verbunden: %s@%s:%d/%s", cfg.User, cfg.Host, cfg.Port, cfg.DBName)
	return d, nil
}

func (d *Database) migrate() error {
	// Relationale Tabellen
	schema := `
	CREATE TABLE IF NOT EXISTS agents (
		id TEXT PRIMARY KEY,
		name TEXT NOT NULL,
		hostname TEXT NOT NULL,
		ip TEXT NOT NULL,
		opnsense_version TEXT NOT NULL DEFAULT '',
		registered_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
		last_heartbeat TIMESTAMPTZ
	);

	CREATE TABLE IF NOT EXISTS system_data (
		agent_id TEXT PRIMARY KEY REFERENCES agents(id) ON DELETE CASCADE,
		data_json JSONB NOT NULL,
		updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
	);

	CREATE TABLE IF NOT EXISTS config_backups (
		id SERIAL PRIMARY KEY,
		agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
		hash TEXT NOT NULL,
		size INTEGER NOT NULL,
		config_xml TEXT NOT NULL,
		created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
	);

	CREATE INDEX IF NOT EXISTS idx_config_backups_agent ON config_backups(agent_id, created_at DESC);
	CREATE INDEX IF NOT EXISTS idx_config_backups_hash ON config_backups(agent_id, hash);
	`

	if _, err := d.db.Exec(schema); err != nil {
		return fmt.Errorf("Schema anlegen: %w", err)
	}

	// Metrics Hypertable
	metricsSchema := `
	CREATE TABLE IF NOT EXISTS metrics (
		time TIMESTAMPTZ NOT NULL,
		agent_id TEXT NOT NULL,
		metric TEXT NOT NULL,
		value DOUBLE PRECISION NOT NULL,
		tags JSONB
	);
	`
	if _, err := d.db.Exec(metricsSchema); err != nil {
		return fmt.Errorf("Metrics-Tabelle anlegen: %w", err)
	}

	// In Hypertable umwandeln (ignoriert wenn schon Hypertable)
	var isHypertable bool
	err := d.db.QueryRow(`
		SELECT EXISTS (
			SELECT 1 FROM timescaledb_information.hypertables 
			WHERE hypertable_name = 'metrics'
		)
	`).Scan(&isHypertable)

	if err != nil || !isHypertable {
		_, err = d.db.Exec(`SELECT create_hypertable('metrics', 'time', if_not_exists => TRUE)`)
		if err != nil {
			log.Printf("Hypertable-Warnung (ggf. schon vorhanden): %v", err)
		} else {
			log.Println("Metrics Hypertable erstellt")
		}
	}

	// Indices fuer Metrics
	d.db.Exec(`CREATE INDEX IF NOT EXISTS idx_metrics_agent_metric ON metrics(agent_id, metric, time DESC)`)

	// Retention Policy (90 Tage)
	d.setupRetention()

	// Compression Policy
	d.setupCompression()

	return nil
}

func (d *Database) setupRetention() {
	// Pruefen ob schon eine Policy existiert
	var count int
	d.db.QueryRow(`
		SELECT COUNT(*) FROM timescaledb_information.jobs 
		WHERE hypertable_name = 'metrics' AND proc_name = 'policy_retention'
	`).Scan(&count)

	if count == 0 {
		_, err := d.db.Exec(`SELECT add_retention_policy('metrics', INTERVAL '90 days')`)
		if err != nil {
			log.Printf("Retention Policy Warnung: %v", err)
		} else {
			log.Println("Retention Policy: 90 Tage fuer metrics")
		}
	}
}

func (d *Database) setupCompression() {
	// Compression aktivieren
	d.db.Exec(`
		ALTER TABLE metrics SET (
			timescaledb.compress,
			timescaledb.compress_segmentby = 'agent_id, metric',
			timescaledb.compress_orderby = 'time DESC'
		)
	`)

	var count int
	d.db.QueryRow(`
		SELECT COUNT(*) FROM timescaledb_information.jobs 
		WHERE hypertable_name = 'metrics' AND proc_name = 'policy_compression'
	`).Scan(&count)

	if count == 0 {
		_, err := d.db.Exec(`SELECT add_compression_policy('metrics', INTERVAL '7 days')`)
		if err != nil {
			log.Printf("Compression Policy Warnung: %v", err)
		} else {
			log.Println("Compression Policy: nach 7 Tagen fuer metrics")
		}
	}
}

func (d *Database) Close() error {
	return d.db.Close()
}

// --- Agent Methoden ---

func (d *Database) RegisterAgent(agent *models.Agent) error {
	_, err := d.db.Exec(`
		INSERT INTO agents (id, name, hostname, ip, opnsense_version, registered_at)
		VALUES ($1, $2, $3, $4, $5, $6)
		ON CONFLICT(id) DO UPDATE SET
			name=EXCLUDED.name,
			hostname=EXCLUDED.hostname,
			ip=EXCLUDED.ip,
			opnsense_version=EXCLUDED.opnsense_version
	`, agent.ID, agent.Name, agent.Hostname, agent.IP, agent.OPNsenseVersion, agent.RegisteredAt.UTC())
	return err
}

func (d *Database) SaveSystemData(agentID string, data *models.SystemData) error {
	now := time.Now().UTC()

	jsonData, err := json.Marshal(data)
	if err != nil {
		return fmt.Errorf("JSON Marshal: %w", err)
	}

	tx, err := d.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()

	// Heartbeat aktualisieren
	res, err := tx.Exec("UPDATE agents SET last_heartbeat = $1 WHERE id = $2", now, agentID)
	if err != nil {
		return err
	}
	rows, _ := res.RowsAffected()
	if rows == 0 {
		return fmt.Errorf("Agent %s nicht gefunden", agentID)
	}

	// Systemdaten upsert
	_, err = tx.Exec(`
		INSERT INTO system_data (agent_id, data_json, updated_at)
		VALUES ($1, $2, $3)
		ON CONFLICT(agent_id) DO UPDATE SET
			data_json=EXCLUDED.data_json,
			updated_at=EXCLUDED.updated_at
	`, agentID, string(jsonData), now)
	if err != nil {
		return err
	}

	// Metriken extrahieren und schreiben
	if err := d.writeMetrics(tx, agentID, now, data); err != nil {
		log.Printf("Metriken schreiben Warnung: %v", err)
		// Kein Abbruch — Systemdaten sind wichtiger
	}

	return tx.Commit()
}

// writeMetrics extrahiert Metriken aus den Systemdaten und schreibt sie in die Hypertable
func (d *Database) writeMetrics(tx *sql.Tx, agentID string, ts time.Time, data *models.SystemData) error {
	stmt, err := tx.Prepare(`INSERT INTO metrics (time, agent_id, metric, value, tags) VALUES ($1, $2, $3, $4, $5)`)
	if err != nil {
		return err
	}
	defer stmt.Close()

	// CPU
	if data.CPU.UsagePercent > 0 {
		stmt.Exec(ts, agentID, "cpu_usage", data.CPU.UsagePercent, nil)
	}

	// Memory
	if data.Memory.TotalBytes > 0 {
		stmt.Exec(ts, agentID, "memory_total_bytes", float64(data.Memory.TotalBytes), nil)
		stmt.Exec(ts, agentID, "memory_used_bytes", float64(data.Memory.UsedBytes), nil)
		usedPercent := float64(data.Memory.UsedBytes) / float64(data.Memory.TotalBytes) * 100
		stmt.Exec(ts, agentID, "memory_used_percent", usedPercent, nil)
	}

	// Uptime
	if data.UptimeSeconds > 0 {
		stmt.Exec(ts, agentID, "uptime_seconds", float64(data.UptimeSeconds), nil)
	}

	// Disks
	for _, disk := range data.Disks {
		if disk.TotalBytes > 0 {
			usedPercent := float64(disk.UsedBytes) / float64(disk.TotalBytes) * 100
			tags, _ := json.Marshal(map[string]string{"mountpoint": disk.MountPoint, "dataset": disk.Filesystem})
			stmt.Exec(ts, agentID, "disk_used_percent", usedPercent, string(tags))
			stmt.Exec(ts, agentID, "disk_used_bytes", float64(disk.UsedBytes), string(tags))
		}
	}

	// Network Interfaces
	for _, iface := range data.NetworkInterfaces {
		if iface.RxBytes > 0 || iface.TxBytes > 0 {
			tags, _ := json.Marshal(map[string]string{"interface": iface.Name})
			stmt.Exec(ts, agentID, "interface_rx_bytes", float64(iface.RxBytes), string(tags))
			stmt.Exec(ts, agentID, "interface_tx_bytes", float64(iface.TxBytes), string(tags))
		}
	}

	// Gateways (Loss und Delay sind Strings wie "0.0%" und "1.2ms")
	for _, gw := range data.Gateways {
		if gw.Delay != "" {
			tags, _ := json.Marshal(map[string]string{"gateway": gw.Name})
			// Delay parsen ("1.2ms" -> 1.2)
			delayStr := strings.TrimSuffix(strings.TrimSpace(gw.Delay), "ms")
			if delay, err := strconv.ParseFloat(delayStr, 64); err == nil {
				stmt.Exec(ts, agentID, "gateway_rtt_ms", delay, string(tags))
			}
			// Loss parsen ("0.0%" -> 0.0)
			lossStr := strings.TrimSuffix(strings.TrimSpace(gw.Loss), "%")
			if loss, err := strconv.ParseFloat(lossStr, 64); err == nil {
				stmt.Exec(ts, agentID, "gateway_loss_percent", loss, string(tags))
			}
		}
	}

	return nil
}

func (d *Database) GetAgents() ([]models.Agent, error) {
	rows, err := d.db.Query("SELECT id, name, hostname, ip, opnsense_version, registered_at, last_heartbeat FROM agents ORDER BY name")
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var agents []models.Agent
	for rows.Next() {
		var a models.Agent
		var lastHB sql.NullTime

		if err := rows.Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.RegisteredAt, &lastHB); err != nil {
			return nil, err
		}
		if lastHB.Valid {
			a.LastHeartbeat = &lastHB.Time
		}
		agents = append(agents, a)
	}
	if agents == nil {
		agents = []models.Agent{}
	}
	return agents, rows.Err()
}

func (d *Database) GetAgentByHostname(hostname string) (*models.Agent, error) {
	var a models.Agent
	var lastHB sql.NullTime

	err := d.db.QueryRow(
		"SELECT id, name, hostname, ip, opnsense_version, registered_at, last_heartbeat FROM agents WHERE hostname = $1 LIMIT 1",
		hostname,
	).Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.RegisteredAt, &lastHB)

	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	if lastHB.Valid {
		a.LastHeartbeat = &lastHB.Time
	}
	return &a, nil
}

func (d *Database) GetAgent(id string) (*models.Agent, *models.SystemData, error) {
	var a models.Agent
	var lastHB sql.NullTime

	err := d.db.QueryRow(
		"SELECT id, name, hostname, ip, opnsense_version, registered_at, last_heartbeat FROM agents WHERE id = $1",
		id,
	).Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.RegisteredAt, &lastHB)

	if err == sql.ErrNoRows {
		return nil, nil, nil
	}
	if err != nil {
		return nil, nil, err
	}
	if lastHB.Valid {
		a.LastHeartbeat = &lastHB.Time
	}

	// Systemdaten laden
	var dataJSON sql.NullString
	err = d.db.QueryRow("SELECT data_json FROM system_data WHERE agent_id = $1", id).Scan(&dataJSON)
	if err != nil && err != sql.ErrNoRows {
		return nil, nil, err
	}

	var sysData *models.SystemData
	if dataJSON.Valid {
		sysData = &models.SystemData{}
		if err := json.Unmarshal([]byte(dataJSON.String), sysData); err != nil {
			return &a, nil, nil
		}
	}

	return &a, sysData, nil
}

func (d *Database) GetSystemData(agentID string) (*models.SystemData, error) {
	var dataJSON string
	err := d.db.QueryRow("SELECT data_json FROM system_data WHERE agent_id = $1", agentID).Scan(&dataJSON)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}

	var data models.SystemData
	if err := json.Unmarshal([]byte(dataJSON), &data); err != nil {
		return nil, err
	}
	return &data, nil
}

func (d *Database) DeleteAgent(id string) (bool, error) {
	res, err := d.db.Exec("DELETE FROM agents WHERE id = $1", id)
	if err != nil {
		return false, err
	}
	rows, _ := res.RowsAffected()
	return rows > 0, nil
}

// --- Config Backup Methoden ---

func (d *Database) SaveConfigBackup(agentID, hash string, size int, configXML string) (int64, bool, error) {
	var existingID int64
	err := d.db.QueryRow("SELECT id FROM config_backups WHERE agent_id = $1 AND hash = $2 LIMIT 1", agentID, hash).Scan(&existingID)
	if err == nil {
		return existingID, false, nil
	}

	var id int64
	err = d.db.QueryRow(
		"INSERT INTO config_backups (agent_id, hash, size, config_xml) VALUES ($1, $2, $3, $4) RETURNING id",
		agentID, hash, size, configXML,
	).Scan(&id)
	if err != nil {
		return 0, false, fmt.Errorf("Backup speichern: %w", err)
	}
	return id, true, nil
}

func (d *Database) GetConfigBackups(agentID string, limit int) ([]map[string]interface{}, error) {
	if limit <= 0 {
		limit = 50
	}
	rows, err := d.db.Query(
		"SELECT id, hash, size, created_at FROM config_backups WHERE agent_id = $1 ORDER BY created_at DESC LIMIT $2",
		agentID, limit,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var backups []map[string]interface{}
	for rows.Next() {
		var id int64
		var hash string
		var size int
		var createdAt time.Time
		if err := rows.Scan(&id, &hash, &size, &createdAt); err != nil {
			return nil, err
		}
		backups = append(backups, map[string]interface{}{
			"id":         id,
			"hash":       hash,
			"size":       size,
			"created_at": createdAt.Format(time.RFC3339),
		})
	}
	if backups == nil {
		backups = []map[string]interface{}{}
	}
	return backups, rows.Err()
}

func (d *Database) GetConfigBackup(agentID string, backupID int64) (map[string]interface{}, error) {
	var hash, configXML string
	var size int
	var id int64
	var createdAt time.Time

	err := d.db.QueryRow(
		"SELECT id, hash, size, config_xml, created_at FROM config_backups WHERE agent_id = $1 AND id = $2",
		agentID, backupID,
	).Scan(&id, &hash, &size, &configXML, &createdAt)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return map[string]interface{}{
		"id":         id,
		"hash":       hash,
		"size":       size,
		"config_xml": configXML,
		"created_at": createdAt.Format(time.RFC3339),
	}, nil
}

func (d *Database) GetLatestConfigBackup(agentID string) (map[string]interface{}, error) {
	var hash, configXML string
	var size int
	var id int64
	var createdAt time.Time

	err := d.db.QueryRow(
		"SELECT id, hash, size, config_xml, created_at FROM config_backups WHERE agent_id = $1 ORDER BY created_at DESC LIMIT 1",
		agentID,
	).Scan(&id, &hash, &size, &configXML, &createdAt)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return map[string]interface{}{
		"id":         id,
		"hash":       hash,
		"size":       size,
		"config_xml": configXML,
		"created_at": createdAt.Format(time.RFC3339),
	}, nil
}

func (d *Database) DeleteConfigBackup(agentID string, backupID int64) (bool, error) {
	res, err := d.db.Exec("DELETE FROM config_backups WHERE agent_id = $1 AND id = $2", agentID, backupID)
	if err != nil {
		return false, err
	}
	rows, _ := res.RowsAffected()
	return rows > 0, nil
}

// --- Metrics Abfragen ---

// GetMetrics holt Metriken fuer einen Agent in einem Zeitraum
func (d *Database) GetMetrics(agentID, metric string, from, to time.Time, limit int) ([]map[string]interface{}, error) {
	if limit <= 0 {
		limit = 1000
	}

	query := `SELECT time, metric, value, tags FROM metrics 
		WHERE agent_id = $1 AND time >= $2 AND time <= $3`
	args := []interface{}{agentID, from, to}

	if metric != "" {
		query += " AND metric = $4"
		args = append(args, metric)
	}

	query += " ORDER BY time DESC LIMIT $" + fmt.Sprintf("%d", len(args)+1)
	args = append(args, limit)

	rows, err := d.db.Query(query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var results []map[string]interface{}
	for rows.Next() {
		var ts time.Time
		var m string
		var v float64
		var tags sql.NullString
		if err := rows.Scan(&ts, &m, &v, &tags); err != nil {
			return nil, err
		}
		entry := map[string]interface{}{
			"time":   ts.Format(time.RFC3339),
			"metric": m,
			"value":  v,
		}
		if tags.Valid {
			var t map[string]interface{}
			json.Unmarshal([]byte(tags.String), &t)
			entry["tags"] = t
		}
		results = append(results, entry)
	}
	if results == nil {
		results = []map[string]interface{}{}
	}
	return results, rows.Err()
}

// GetMetricsSummary holt aggregierte Metriken (avg/min/max) in Zeitbuckts
func (d *Database) GetMetricsSummary(agentID, metric, bucket string, from, to time.Time) ([]map[string]interface{}, error) {
	if bucket == "" {
		bucket = "5 minutes"
	}

	rows, err := d.db.Query(`
		SELECT time_bucket($1::interval, time) AS t,
			AVG(value) AS avg_value,
			MIN(value) AS min_value,
			MAX(value) AS max_value,
			COUNT(*) AS samples
		FROM metrics
		WHERE agent_id = $2 AND metric = $3 AND time >= $4 AND time <= $5
		GROUP BY t
		ORDER BY t DESC
	`, bucket, agentID, metric, from, to)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var results []map[string]interface{}
	for rows.Next() {
		var t time.Time
		var avg, min, max float64
		var samples int
		if err := rows.Scan(&t, &avg, &min, &max, &samples); err != nil {
			return nil, err
		}
		results = append(results, map[string]interface{}{
			"time":    t.Format(time.RFC3339),
			"avg":     avg,
			"min":     min,
			"max":     max,
			"samples": samples,
		})
	}
	if results == nil {
		results = []map[string]interface{}{}
	}
	return results, rows.Err()
}