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 '', agent_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); CREATE TABLE IF NOT EXISTS customers ( id SERIAL PRIMARY KEY, number TEXT NOT NULL UNIQUE, name TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, display_name TEXT NOT NULL DEFAULT '', role TEXT NOT NULL DEFAULT 'admin', created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), last_login TIMESTAMPTZ ); ` if _, err := d.db.Exec(schema); err != nil { return fmt.Errorf("Schema anlegen: %w", err) } // Agents um customer_id erweitern d.db.Exec("ALTER TABLE agents ADD COLUMN IF NOT EXISTS customer_id INTEGER REFERENCES customers(id)") d.db.Exec("CREATE INDEX IF NOT EXISTS idx_agents_customer ON agents(customer_id)") // Agents um agent_version und update_requested erweitern d.db.Exec("ALTER TABLE agents ADD COLUMN IF NOT EXISTS agent_version TEXT NOT NULL DEFAULT ''") d.db.Exec("ALTER TABLE agents ADD COLUMN IF NOT EXISTS update_requested BOOLEAN NOT NULL DEFAULT FALSE") d.db.Exec("ALTER TABLE agents ADD COLUMN IF NOT EXISTS platform TEXT NOT NULL DEFAULT 'freebsd'") // Agent-Firmware Tabelle (Multi-Plattform) d.db.Exec(`CREATE TABLE IF NOT EXISTS agent_firmware ( version TEXT PRIMARY KEY, hash TEXT NOT NULL, binary_data BYTEA NOT NULL, uploaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW() )`) // Platform-Spalte hinzufuegen (default freebsd fuer bestehende Eintraege) d.db.Exec("ALTER TABLE agent_firmware DROP CONSTRAINT IF EXISTS agent_firmware_pkey") d.db.Exec("ALTER TABLE agent_firmware ADD COLUMN IF NOT EXISTS platform TEXT NOT NULL DEFAULT 'freebsd'") // Neuer Primary Key: version + platform d.db.Exec(`DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'agent_firmware_version_platform_key') THEN ALTER TABLE agent_firmware ADD CONSTRAINT agent_firmware_version_platform_key UNIQUE (version, platform); END IF; END $$`) // 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)`) // API-Keys Tabelle d.db.Exec(`CREATE TABLE IF NOT EXISTS api_keys ( id SERIAL PRIMARY KEY, name TEXT NOT NULL, key TEXT NOT NULL UNIQUE, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), last_used TIMESTAMPTZ )`) // Installer-Pakete (install.sh + Plugin als ZIP pro Plattform) d.db.Exec(`CREATE TABLE IF NOT EXISTS installers ( id SERIAL PRIMARY KEY, platform TEXT NOT NULL UNIQUE, filename TEXT NOT NULL, data BYTEA NOT NULL, sha256 TEXT NOT NULL, size INTEGER NOT NULL, uploaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW() )`) // Audit-Log Tabelle d.db.Exec(`CREATE TABLE IF NOT EXISTS audit_log ( id SERIAL PRIMARY KEY, timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(), user_name TEXT NOT NULL, action TEXT NOT NULL, target_type TEXT, target_id TEXT, target_name TEXT, details TEXT, source_ip TEXT )`) d.db.Exec(`CREATE INDEX IF NOT EXISTS idx_audit_log_timestamp ON audit_log(timestamp 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, agent_version, platform, registered_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT(id) DO UPDATE SET name=EXCLUDED.name, hostname=EXCLUDED.hostname, ip=EXCLUDED.ip, opnsense_version=EXCLUDED.opnsense_version, agent_version=EXCLUDED.agent_version, platform=EXCLUDED.platform `, agent.ID, agent.Name, agent.Hostname, agent.IP, agent.OPNsenseVersion, agent.AgentVersion, agent.Platform, 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, agent_version, platform, registered_at, last_heartbeat, customer_id, update_requested 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 var custID sql.NullInt64 if err := rows.Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.AgentVersion, &a.Platform, &a.RegisteredAt, &lastHB, &custID, &a.UpdateRequested); err != nil { return nil, err } if lastHB.Valid { a.LastHeartbeat = &lastHB.Time } if custID.Valid { id := int(custID.Int64) a.CustomerID = &id } 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) GetAgentByName(name 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 name = $1 LIMIT 1", name, ).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 var custID sql.NullInt64 err := d.db.QueryRow( "SELECT id, name, hostname, ip, opnsense_version, agent_version, platform, registered_at, last_heartbeat, customer_id, update_requested FROM agents WHERE id = $1", id, ).Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.AgentVersion, &a.Platform, &a.RegisteredAt, &lastHB, &custID, &a.UpdateRequested) if err == sql.ErrNoRows { return nil, nil, nil } if err != nil { return nil, nil, err } if lastHB.Valid { a.LastHeartbeat = &lastHB.Time } if custID.Valid { cid := int(custID.Int64) a.CustomerID = &cid } // 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) UpdateAgentVersion(id, version string) error { _, err := d.db.Exec("UPDATE agents SET agent_version = $1 WHERE id = $2", version, id) return err } // Update-Request Flag setzen/lesen/loeschen func (d *Database) SetUpdateRequest(id string, requested bool) error { _, err := d.db.Exec("UPDATE agents SET update_requested = $1 WHERE id = $2", requested, id) return err } func (d *Database) IsUpdateRequested(id string) bool { var requested bool err := d.db.QueryRow("SELECT update_requested FROM agents WHERE id = $1", id).Scan(&requested) return err == nil && requested } func (d *Database) ClearUpdateRequest(id string) error { return d.SetUpdateRequest(id, false) } // Alle Agents mit Update-Flag setzen func (d *Database) SetUpdateRequestAll(requested bool) (int64, error) { res, err := d.db.Exec("UPDATE agents SET update_requested = $1", requested) if err != nil { return 0, err } return res.RowsAffected() } // Agent-Version direkt aus DB lesen (vor dem Update durch Heartbeat) func (d *Database) GetAgentVersionFromDB(id string) string { var version string d.db.QueryRow("SELECT agent_version FROM agents WHERE id = $1", id).Scan(&version) return version } // Agent-Firmware speichern (pro Plattform) func (d *Database) SaveAgentFirmware(version, platform, hash string, binary []byte) error { _, err := d.db.Exec(` INSERT INTO agent_firmware (version, platform, hash, binary_data, uploaded_at) VALUES ($1, $2, $3, $4, NOW()) ON CONFLICT(version, platform) DO UPDATE SET hash=EXCLUDED.hash, binary_data=EXCLUDED.binary_data, uploaded_at=NOW() `, version, platform, hash, binary) return err } func (d *Database) GetLatestFirmware(platform string) (string, string, []byte, error) { var version, hash string var binary []byte err := d.db.QueryRow("SELECT version, hash, binary_data FROM agent_firmware WHERE platform = $1 ORDER BY uploaded_at DESC LIMIT 1", platform).Scan(&version, &hash, &binary) if err != nil { return "", "", nil, err } return version, hash, binary, nil } func (d *Database) GetFirmwareInfo(platform string) (string, string, int, error) { var version, hash string var size int err := d.db.QueryRow("SELECT version, hash, length(binary_data) FROM agent_firmware WHERE platform = $1 ORDER BY uploaded_at DESC LIMIT 1", platform).Scan(&version, &hash, &size) if err != nil { return "", "", 0, err } return version, hash, size, nil } // Alle Plattform-Firmwares auflisten func (d *Database) GetAllFirmwareInfo() ([]map[string]interface{}, error) { rows, err := d.db.Query(` SELECT DISTINCT ON (platform) platform, version, hash, length(binary_data) as size, uploaded_at FROM agent_firmware ORDER BY platform, uploaded_at DESC `) if err != nil { return nil, err } defer rows.Close() var results []map[string]interface{} for rows.Next() { var platform, version, hash string var size int var uploadedAt time.Time if err := rows.Scan(&platform, &version, &hash, &size, &uploadedAt); err != nil { return nil, err } results = append(results, map[string]interface{}{ "platform": platform, "version": version, "hash": hash, "size": size, "uploaded_at": uploadedAt, }) } return results, nil } // --- Installer --- func (d *Database) SaveInstaller(platform, filename, sha256hash string, data []byte) error { _, err := d.db.Exec(` INSERT INTO installers (platform, filename, data, sha256, size, uploaded_at) VALUES ($1, $2, $3, $4, $5, NOW()) ON CONFLICT (platform) DO UPDATE SET filename=$2, data=$3, sha256=$4, size=$5, uploaded_at=NOW() `, platform, filename, data, sha256hash, len(data)) return err } func (d *Database) GetInstaller(platform string) (string, string, []byte, error) { var filename, sha256hash string var data []byte err := d.db.QueryRow("SELECT filename, sha256, data FROM installers WHERE platform = $1", platform).Scan(&filename, &sha256hash, &data) return filename, sha256hash, data, err } func (d *Database) GetInstallerInfo(platform string) (string, string, int, string, error) { var filename, sha256hash string var size int var uploadedAt time.Time err := d.db.QueryRow("SELECT filename, sha256, size, uploaded_at FROM installers WHERE platform = $1", platform).Scan(&filename, &sha256hash, &size, &uploadedAt) return filename, sha256hash, size, uploadedAt.Format(time.RFC3339), err } func (d *Database) GetAllInstallerInfo() ([]map[string]interface{}, error) { rows, err := d.db.Query("SELECT platform, filename, sha256, size, uploaded_at FROM installers ORDER BY platform") if err != nil { return nil, err } defer rows.Close() var results []map[string]interface{} for rows.Next() { var platform, filename, sha256hash string var size int var uploadedAt time.Time if err := rows.Scan(&platform, &filename, &sha256hash, &size, &uploadedAt); err != nil { return nil, err } results = append(results, map[string]interface{}{ "platform": platform, "filename": filename, "sha256": sha256hash, "size": size, "uploaded_at": uploadedAt, }) } return results, 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() } // --- Agent-Customer Zuordnung --- func (d *Database) AssignAgentCustomer(agentID string, customerID int) error { _, err := d.db.Exec("UPDATE agents SET customer_id = $1 WHERE id = $2", customerID, agentID) return err } func (d *Database) UnassignAgentCustomer(agentID string) error { _, err := d.db.Exec("UPDATE agents SET customer_id = NULL WHERE id = $1", agentID) return err } func (d *Database) GetAgentsByCustomer(customerID int) ([]models.Agent, error) { rows, err := d.db.Query("SELECT id, name, hostname, ip, opnsense_version, registered_at, last_heartbeat, customer_id FROM agents WHERE customer_id = $1 ORDER BY name", customerID) if err != nil { return nil, err } defer rows.Close() var agents []models.Agent for rows.Next() { var a models.Agent var lastHB sql.NullTime var custID sql.NullInt64 if err := rows.Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.RegisteredAt, &lastHB, &custID); err != nil { return nil, err } if lastHB.Valid { a.LastHeartbeat = &lastHB.Time } if custID.Valid { cid := int(custID.Int64) a.CustomerID = &cid } agents = append(agents, a) } if agents == nil { agents = []models.Agent{} } return agents, 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) } // ==================== API-Keys ==================== type APIKey struct { ID int `json:"id"` Name string `json:"name"` Key string `json:"key"` CreatedAt time.Time `json:"created_at"` LastUsed *time.Time `json:"last_used,omitempty"` } func (d *Database) ListAPIKeys() ([]APIKey, error) { rows, err := d.db.Query("SELECT id, name, key, created_at, last_used FROM api_keys ORDER BY created_at") if err != nil { return nil, err } defer rows.Close() var keys []APIKey for rows.Next() { var k APIKey var lastUsed sql.NullTime if err := rows.Scan(&k.ID, &k.Name, &k.Key, &k.CreatedAt, &lastUsed); err != nil { return nil, err } if lastUsed.Valid { k.LastUsed = &lastUsed.Time } keys = append(keys, k) } if keys == nil { keys = []APIKey{} } return keys, rows.Err() } func (d *Database) CreateAPIKey(name, key string) (*APIKey, error) { var k APIKey err := d.db.QueryRow( "INSERT INTO api_keys (name, key) VALUES ($1, $2) RETURNING id, name, key, created_at", name, key, ).Scan(&k.ID, &k.Name, &k.Key, &k.CreatedAt) if err != nil { return nil, err } return &k, nil } func (d *Database) DeleteAPIKey(id int) error { result, err := d.db.Exec("DELETE FROM api_keys WHERE id = $1", id) if err != nil { return err } rows, _ := result.RowsAffected() if rows == 0 { return fmt.Errorf("API-Key nicht gefunden") } return nil } func (d *Database) GetAllAPIKeys() ([]string, error) { rows, err := d.db.Query("SELECT key FROM api_keys") if err != nil { return nil, err } defer rows.Close() var keys []string for rows.Next() { var k string if err := rows.Scan(&k); err != nil { return nil, err } keys = append(keys, k) } return keys, rows.Err() } func (d *Database) UpdateAPIKeyLastUsed(key string) { d.db.Exec("UPDATE api_keys SET last_used = NOW() WHERE key = $1", key) } // MigrateConfigAPIKeys migriert API-Keys aus der Config in die DB (einmalig) // --- Audit Log --- func (d *Database) AddAuditLog(userName, action, targetType, targetID, targetName, details, sourceIP string) error { _, err := d.db.Exec( `INSERT INTO audit_log (user_name, action, target_type, target_id, target_name, details, source_ip) VALUES ($1, $2, $3, $4, $5, $6, $7)`, userName, action, targetType, targetID, targetName, details, sourceIP, ) if err != nil { log.Printf("Audit-Log schreiben fehlgeschlagen: %v", err) } return err } func (d *Database) GetAuditLogs(limit int, offset int, action string, agentID string) ([]models.AuditLogEntry, int, error) { if limit <= 0 { limit = 100 } if limit > 500 { limit = 500 } where := []string{} args := []interface{}{} argIdx := 1 if action != "" { where = append(where, fmt.Sprintf("action = $%d", argIdx)) args = append(args, action) argIdx++ } if agentID != "" { where = append(where, fmt.Sprintf("target_id = $%d", argIdx)) args = append(args, agentID) argIdx++ } whereClause := "" if len(where) > 0 { whereClause = "WHERE " + strings.Join(where, " AND ") } // Total count var total int countQuery := "SELECT COUNT(*) FROM audit_log " + whereClause err := d.db.QueryRow(countQuery, args...).Scan(&total) if err != nil { return nil, 0, err } // Data query := fmt.Sprintf("SELECT id, timestamp, user_name, action, COALESCE(target_type,''), COALESCE(target_id,''), COALESCE(target_name,''), COALESCE(details,''), COALESCE(source_ip,'') FROM audit_log %s ORDER BY timestamp DESC LIMIT $%d OFFSET $%d", whereClause, argIdx, argIdx+1) args = append(args, limit, offset) rows, err := d.db.Query(query, args...) if err != nil { return nil, 0, err } defer rows.Close() var entries []models.AuditLogEntry for rows.Next() { var e models.AuditLogEntry var ts time.Time if err := rows.Scan(&e.ID, &ts, &e.UserName, &e.Action, &e.TargetType, &e.TargetID, &e.TargetName, &e.Details, &e.SourceIP); err != nil { return nil, 0, err } e.Timestamp = ts.Format(time.RFC3339) entries = append(entries, e) } if entries == nil { entries = []models.AuditLogEntry{} } return entries, total, rows.Err() } func (d *Database) MigrateConfigAPIKeys(configKeys []string) { for _, key := range configKeys { var exists bool d.db.QueryRow("SELECT EXISTS(SELECT 1 FROM api_keys WHERE key = $1)", key).Scan(&exists) if !exists { d.db.Exec("INSERT INTO api_keys (name, key) VALUES ($1, $2)", "Default (migriert)", key) log.Printf("API-Key aus Config migriert: %s...", key[:8]) } } }