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); CREATE TABLE IF NOT EXISTS agent_events ( id SERIAL PRIMARY KEY, agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE, event_type TEXT NOT NULL, message TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX IF NOT EXISTS idx_agent_events_agent ON agent_events(agent_id, created_at DESC); ` 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() } // --- Agent Events Methoden --- // InsertAgentEvent schreibt ein Event in die agent_events Tabelle func (d *Database) InsertAgentEvent(agentID, eventType, message string) error { _, err := d.db.Exec( "INSERT INTO agent_events (agent_id, event_type, message) VALUES ($1, $2, $3)", agentID, eventType, message, ) if err != nil { log.Printf("Event schreiben fehlgeschlagen: %v", err) } return err } // GetAgentEvents holt Events fuer einen bestimmten Agent func (d *Database) GetAgentEvents(agentID string, eventType string, limit int) ([]models.AgentEvent, error) { if limit <= 0 { limit = 100 } query := "SELECT id, agent_id, event_type, COALESCE(message, ''), created_at FROM agent_events WHERE agent_id = $1" args := []interface{}{agentID} if eventType != "" { query += " AND event_type = $2" args = append(args, eventType) } query += fmt.Sprintf(" ORDER BY created_at DESC LIMIT $%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 events []models.AgentEvent for rows.Next() { var e models.AgentEvent if err := rows.Scan(&e.ID, &e.AgentID, &e.EventType, &e.Message, &e.CreatedAt); err != nil { return nil, err } events = append(events, e) } if events == nil { events = []models.AgentEvent{} } return events, rows.Err() } // GetAllAgentEvents holt Events fuer alle Agents func (d *Database) GetAllAgentEvents(eventType string, limit int) ([]models.AgentEvent, error) { if limit <= 0 { limit = 100 } query := "SELECT id, agent_id, event_type, COALESCE(message, ''), created_at FROM agent_events" args := []interface{}{} if eventType != "" { query += " WHERE event_type = $1" args = append(args, eventType) } query += fmt.Sprintf(" ORDER BY created_at DESC LIMIT $%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 events []models.AgentEvent for rows.Next() { var e models.AgentEvent if err := rows.Scan(&e.ID, &e.AgentID, &e.EventType, &e.Message, &e.CreatedAt); err != nil { return nil, err } events = append(events, e) } if events == nil { events = []models.AgentEvent{} } return events, rows.Err() } // GetAgentStatus berechnet den Status eines Agents aus last_heartbeat func (d *Database) GetAgentStatus(agentID string) string { var lastHB sql.NullTime err := d.db.QueryRow("SELECT last_heartbeat FROM agents WHERE id = $1", agentID).Scan(&lastHB) if err != nil || !lastHB.Valid { return "unknown" } return models.CalculateAgentStatus(&lastHB.Time) }