syncova-backup/packages/hypervisor/store.go

package hypervisor

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

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/syncova/syncova/packages/platform/crypto"
)

// ErrClusterNotFound meldet einen unbekannten Verbund.
var ErrClusterNotFound = errors.New("der virtualisierungsverbund wurde nicht gefunden")

// ErrClusterInUse meldet einen Verbund, auf den noch Sicherungsquellen verweisen.
//
// Ihn zu löschen nähme den betroffenen Aufträgen ihre Grundlage — und zwar
// stillschweigend, bis der nächste Lauf scheitert.
var ErrClusterInUse = errors.New("auf diesen verbund verweisen noch sicherungsquellen")

// Store ist die Datenzugriffsschicht der Virtualisierungsumgebungen.
type Store struct {
	// connectionPool ist der Datenbankpool der Control Plane.
	connectionPool *pgxpool.Pool
	// secretStore ver- und entschlüsselt Zugangsdaten.
	//
	// Ohne ihn lässt sich kein Verbund einrichten: Ein API-Token im Klartext in
	// der Datenbank wäre ein Verstoß gegen PROMPT.md §140 — und wer die
	// Datenbank sichert, sicherte die Zugangsdaten gleich mit.
	secretStore crypto.SecretStore
}

// NewStore erzeugt die Datenzugriffsschicht.
func NewStore(connectionPool *pgxpool.Pool, secretStore crypto.SecretStore) (*Store, error) {
	if connectionPool == nil {
		return nil, errors.New("der speicher braucht einen datenbankpool")
	}

	if secretStore == nil {
		return nil, errors.New("ohne schluesselmaterial lassen sich keine zugangsdaten ablegen; " +
			"setzen sie SYNCOVA_ENCRYPTION_KEYS")
	}

	return &Store{connectionPool: connectionPool, secretStore: secretStore}, nil
}

// clusterColumns ist die gemeinsame Spaltenliste aller Leseabfragen.
//
// An einer Stelle, weil zwei auseinanderlaufende Listen genau dann auffallen,
// wenn ein Feld fehlt und niemand weiß, warum.
const clusterColumns = `
	id, name, api_endpoint, api_token_id, tls_fingerprint, backup_storage_id,
	archive_transport, archive_mount_roots, ssh_username, ssh_port,
	ssh_host_fingerprints, keep_archive_on_node, status, last_error,
	last_seen_at, last_discovery_at, created_at, updated_at`

// CreateCluster richtet einen Verbund ein.
func (store *Store) CreateCluster(createContext context.Context, input ClusterInput) (*Cluster, error) {
	if validationError := input.Validate(); validationError != nil {
		return nil, validationError
	}

	tokenCiphertext, tokenKeyVersion, encryptError := store.secretStore.Encrypt([]byte(input.APITokenSecret))
	if encryptError != nil {
		return nil, fmt.Errorf("das api-token liess sich nicht verschluesseln: %w", encryptError)
	}

	var (
		keyCiphertext []byte
		keyVersion    *string
	)

	if input.SSHPrivateKeyPEM != "" {
		encryptedKey, version, keyEncryptError := store.secretStore.Encrypt([]byte(input.SSHPrivateKeyPEM))
		if keyEncryptError != nil {
			return nil, fmt.Errorf("der ssh-schluessel liess sich nicht verschluesseln: %w", keyEncryptError)
		}

		keyCiphertext = encryptedKey
		keyVersion = &version
	}

	mountRootsJSON, hostFingerprintsJSON, encodeError := encodeMaps(input.ArchiveMountRoots, input.SSHHostFingerprints)
	if encodeError != nil {
		return nil, encodeError
	}

	const insertStatement = `
		INSERT INTO proxmox_clusters (
			name, api_endpoint, api_token_id, api_token_ciphertext, api_token_key_version,
			tls_fingerprint, backup_storage_id, archive_transport, archive_mount_roots,
			ssh_username, ssh_port, ssh_private_key_ciphertext, ssh_private_key_version,
			ssh_host_fingerprints, keep_archive_on_node)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
		RETURNING ` + clusterColumns

	createdCluster, scanError := scanCluster(store.connectionPool.QueryRow(createContext, insertStatement,
		input.Name, input.APIEndpoint, input.APITokenID, tokenCiphertext, tokenKeyVersion,
		emptyToNil(input.TLSFingerprint), input.BackupStorageID, string(input.ArchiveTransport), mountRootsJSON,
		emptyToNil(input.SSHUsername), zeroToNil(input.SSHPort), keyCiphertext, keyVersion,
		hostFingerprintsJSON, input.KeepArchiveOnNode))
	if scanError != nil {
		return nil, fmt.Errorf("der verbund liess sich nicht anlegen: %w", scanError)
	}

	return createdCluster, nil
}

// ListClusters liefert alle eingerichteten Verbünde.
func (store *Store) ListClusters(listContext context.Context) ([]Cluster, error) {
	const selectStatement = `SELECT ` + clusterColumns + ` FROM proxmox_clusters ORDER BY name`

	resultRows, queryError := store.connectionPool.Query(listContext, selectStatement)
	if queryError != nil {
		return nil, fmt.Errorf("die verbuende liessen sich nicht lesen: %w", queryError)
	}
	defer resultRows.Close()

	clusterList := make([]Cluster, 0, 4)

	for resultRows.Next() {
		scannedCluster, scanError := scanCluster(resultRows)
		if scanError != nil {
			return nil, scanError
		}

		clusterList = append(clusterList, *scannedCluster)
	}

	return clusterList, resultRows.Err()
}

// GetCluster liefert einen einzelnen Verbund.
func (store *Store) GetCluster(readContext context.Context, clusterIdentifier uuid.UUID) (*Cluster, error) {
	const selectStatement = `SELECT ` + clusterColumns + ` FROM proxmox_clusters WHERE id = $1`

	foundCluster, scanError := scanCluster(store.connectionPool.QueryRow(readContext, selectStatement, clusterIdentifier))
	if errors.Is(scanError, pgx.ErrNoRows) {
		return nil, ErrClusterNotFound
	}

	if scanError != nil {
		return nil, fmt.Errorf("der verbund liess sich nicht lesen: %w", scanError)
	}

	return foundCluster, nil
}

// LoadCredentials entschlüsselt die Zugangsdaten eines Verbunds.
//
// Ein eigener Aufruf und nicht Teil von GetCluster: Wer einen Verbund nur
// anzeigt, soll die Geheimnisse gar nicht erst im Speicher haben.
func (store *Store) LoadCredentials(readContext context.Context,
	clusterIdentifier uuid.UUID) (*ClusterCredentials, error) {
	const selectStatement = `
		SELECT api_token_ciphertext, api_token_key_version,
		       ssh_private_key_ciphertext, ssh_private_key_version
		FROM proxmox_clusters WHERE id = $1`

	var (
		tokenCiphertext []byte
		tokenVersion    string
		keyCiphertext   []byte
		keyVersion      *string
	)

	scanError := store.connectionPool.QueryRow(readContext, selectStatement, clusterIdentifier).Scan(
		&tokenCiphertext, &tokenVersion, &keyCiphertext, &keyVersion)
	if errors.Is(scanError, pgx.ErrNoRows) {
		return nil, ErrClusterNotFound
	}

	if scanError != nil {
		return nil, fmt.Errorf("die zugangsdaten liessen sich nicht lesen: %w", scanError)
	}

	tokenSecret, decryptError := store.secretStore.Decrypt(tokenCiphertext, tokenVersion)
	if decryptError != nil {
		return nil, fmt.Errorf("das api-token liess sich nicht entschluesseln: %w", decryptError)
	}

	credentials := &ClusterCredentials{APITokenSecret: string(tokenSecret)}

	if len(keyCiphertext) > 0 && keyVersion != nil {
		privateKey, keyDecryptError := store.secretStore.Decrypt(keyCiphertext, *keyVersion)
		if keyDecryptError != nil {
			return nil, fmt.Errorf("der ssh-schluessel liess sich nicht entschluesseln: %w", keyDecryptError)
		}

		credentials.SSHPrivateKeyPEM = privateKey
	}

	return credentials, nil
}

// RecordConnectionResult vermerkt das Ergebnis einer Verbindungsprüfung.
func (store *Store) RecordConnectionResult(recordContext context.Context, clusterIdentifier uuid.UUID,
	status ClusterStatus, connectionError error) error {
	const updateStatement = `
		UPDATE proxmox_clusters
		SET status = $2,
		    last_error = $3,
		    -- Der Zeitpunkt wird nur bei Erfolg fortgeschrieben. Ihn bei jedem
		    -- Versuch zu setzen machte aus "zuletzt erreicht" ein "zuletzt
		    -- versucht" — und ein seit Wochen toter Verbund sähe frisch aus.
		    last_seen_at = CASE WHEN $2 = 'reachable' THEN NOW() ELSE last_seen_at END,
		    updated_at = NOW()
		WHERE id = $1`

	var errorMessage *string

	if connectionError != nil {
		message := connectionError.Error()
		errorMessage = &message
	}

	_, executeError := store.connectionPool.Exec(recordContext, updateStatement,
		clusterIdentifier, string(status), errorMessage)
	if executeError != nil {
		return fmt.Errorf("das pruefergebnis liess sich nicht vermerken: %w", executeError)
	}

	return nil
}

// DeleteCluster entfernt einen Verbund.
func (store *Store) DeleteCluster(deleteContext context.Context, clusterIdentifier uuid.UUID) error {
	// Erst die Verwendung prüfen: Der Fremdschlüssel steht auf RESTRICT, aber
	// eine Fremdschlüsselverletzung ist für den Aufrufer keine Auskunft.
	const usageStatement = `SELECT COUNT(*) FROM backup_job_sources WHERE cluster_id = $1`

	var sourceCount int

	if countError := store.connectionPool.QueryRow(deleteContext, usageStatement,
		clusterIdentifier).Scan(&sourceCount); countError != nil {
		return fmt.Errorf("die verwendung liess sich nicht pruefen: %w", countError)
	}

	if sourceCount > 0 {
		return fmt.Errorf("%w (%d quelle(n))", ErrClusterInUse, sourceCount)
	}

	commandTag, deleteError := store.connectionPool.Exec(deleteContext,
		`DELETE FROM proxmox_clusters WHERE id = $1`, clusterIdentifier)
	if deleteError != nil {
		return fmt.Errorf("der verbund liess sich nicht entfernen: %w", deleteError)
	}

	if commandTag.RowsAffected() == 0 {
		return ErrClusterNotFound
	}

	return nil
}

// ListVirtualMachines liefert den aufgenommenen Gästebestand.
func (store *Store) ListVirtualMachines(listContext context.Context,
	clusterIdentifier *uuid.UUID) ([]VirtualMachine, error) {
	const selectStatement = `
		SELECT vm.id, vm.cluster_id, vm.host_id, vm.provider_vm_id, vm.name, vm.guest_kind,
		       vm.status, vm.cpu_count, vm.memory_bytes, host.node_name,
		       vm.disks_json, vm.guest_agent_running, vm.last_discovered_at, vm.missing_since
		FROM virtual_machines vm
		LEFT JOIN proxmox_hosts host ON host.id = vm.host_id
		WHERE ($1::uuid IS NULL OR vm.cluster_id = $1)
		ORDER BY vm.name`

	resultRows, queryError := store.connectionPool.Query(listContext, selectStatement, clusterIdentifier)
	if queryError != nil {
		return nil, fmt.Errorf("der gaestebestand liess sich nicht lesen: %w", queryError)
	}
	defer resultRows.Close()

	machineList := make([]VirtualMachine, 0, 16)

	for resultRows.Next() {
		var (
			scannedMachine  VirtualMachine
			nodeName        *string
			cpuCount        *int
			memoryBytes     *int64
			diskDefinitions []byte
			guestStatus     *string
		)

		scanError := resultRows.Scan(&scannedMachine.ID, &scannedMachine.ClusterID, &scannedMachine.HostID,
			&scannedMachine.ProviderVMID, &scannedMachine.Name, &scannedMachine.GuestKind,
			&guestStatus, &cpuCount, &memoryBytes, &nodeName, &diskDefinitions,
			&scannedMachine.GuestAgentRunning, &scannedMachine.LastDiscoveredAt, &scannedMachine.MissingSince)
		if scanError != nil {
			return nil, fmt.Errorf("ein gast liess sich nicht lesen: %w", scanError)
		}

		scannedMachine.Status = stringValue(guestStatus)
		scannedMachine.NodeName = stringValue(nodeName)

		if cpuCount != nil {
			scannedMachine.CPUCount = *cpuCount
		}

		if memoryBytes != nil {
			scannedMachine.MemoryBytes = *memoryBytes
		}

		scannedMachine.DiskCount, scannedMachine.ExcludedDiskCount = countDisks(diskDefinitions)

		machineList = append(machineList, scannedMachine)
	}

	return machineList, resultRows.Err()
}

// FindVirtualMachine sucht einen Gast anhand seiner Providerkennung.
func (store *Store) FindVirtualMachine(readContext context.Context, clusterIdentifier uuid.UUID,
	providerVMID string) (*VirtualMachine, error) {
	machineList, listError := store.ListVirtualMachines(readContext, &clusterIdentifier)
	if listError != nil {
		return nil, listError
	}

	for _, candidateMachine := range machineList {
		if candidateMachine.ProviderVMID == providerVMID {
			return &candidateMachine, nil
		}
	}

	return nil, fmt.Errorf("der gast %q ist im bestand des verbunds nicht enthalten", providerVMID)
}

// countDisks zählt gesicherte und ausgenommene Platten.
//
// Eine beschädigte Plattenliste wird als „keine Angabe" gewertet und nicht als
// „keine Platten": Der Unterschied entscheidet darüber, ob jemand eine
// unvollständige Maschine für vollständig hält.
func countDisks(diskDefinitions []byte) (diskCount int, excludedCount int) {
	if len(diskDefinitions) == 0 {
		return 0, 0
	}

	var decodedDisks []struct {
		ExcludedFromBackup bool `json:"excluded_from_backup"`
	}

	if decodeError := json.Unmarshal(diskDefinitions, &decodedDisks); decodeError != nil {
		return 0, 0
	}

	for _, singleDisk := range decodedDisks {
		if singleDisk.ExcludedFromBackup {
			excludedCount++

			continue
		}

		diskCount++
	}

	return diskCount, excludedCount
}

// rowScanner deckt QueryRow und Rows gleichermaßen ab.
type rowScanner interface {
	// Scan liest eine Zeile.
	Scan(destinations ...any) error
}

// scanCluster liest eine Verbundzeile.
func scanCluster(scanner rowScanner) (*Cluster, error) {
	var (
		scannedCluster       Cluster
		transportKind        string
		mountRootsJSON       []byte
		hostFingerprintsJSON []byte
		tlsFingerprint       *string
		sshUsername          *string
		sshPort              *int
		lastError            *string
		statusValue          string
	)

	scanError := scanner.Scan(&scannedCluster.ID, &scannedCluster.Name, &scannedCluster.APIEndpoint,
		&scannedCluster.APITokenID, &tlsFingerprint, &scannedCluster.BackupStorageID,
		&transportKind, &mountRootsJSON, &sshUsername, &sshPort, &hostFingerprintsJSON,
		&scannedCluster.KeepArchiveOnNode, &statusValue, &lastError,
		&scannedCluster.LastSeenAt, &scannedCluster.LastDiscoveryAt,
		&scannedCluster.CreatedAt, &scannedCluster.UpdatedAt)
	if scanError != nil {
		return nil, scanError
	}

	scannedCluster.ArchiveTransport = TransportKind(transportKind)
	scannedCluster.Status = ClusterStatus(statusValue)
	scannedCluster.TLSFingerprint = stringValue(tlsFingerprint)
	scannedCluster.SSHUsername = stringValue(sshUsername)
	scannedCluster.LastError = stringValue(lastError)

	if sshPort != nil {
		scannedCluster.SSHPort = *sshPort
	}

	if decodeError := json.Unmarshal(mountRootsJSON, &scannedCluster.ArchiveMountRoots); decodeError != nil {
		return nil, fmt.Errorf("die speicherzuordnung liess sich nicht lesen: %w", decodeError)
	}

	if decodeError := json.Unmarshal(hostFingerprintsJSON, &scannedCluster.SSHHostFingerprints); decodeError != nil {
		return nil, fmt.Errorf("die wirtsschluessel liessen sich nicht lesen: %w", decodeError)
	}

	return &scannedCluster, nil
}

// encodeMaps kodiert die beiden Zuordnungen für JSONB-Spalten.
func encodeMaps(mountRoots map[string]string, hostFingerprints map[string]string) ([]byte, []byte, error) {
	if mountRoots == nil {
		mountRoots = map[string]string{}
	}

	if hostFingerprints == nil {
		hostFingerprints = map[string]string{}
	}

	encodedMountRoots, mountError := json.Marshal(mountRoots)
	if mountError != nil {
		return nil, nil, fmt.Errorf("die speicherzuordnung liess sich nicht kodieren: %w", mountError)
	}

	encodedFingerprints, fingerprintError := json.Marshal(hostFingerprints)
	if fingerprintError != nil {
		return nil, nil, fmt.Errorf("die wirtsschluessel liessen sich nicht kodieren: %w", fingerprintError)
	}

	return encodedMountRoots, encodedFingerprints, nil
}

// emptyToNil macht aus einer leeren Zeichenkette ein NULL.
//
// Nötig, weil eine leere Zeichenkette und „nicht gesetzt" in der Datenbank
// verschiedene Dinge sind — ein leerer Fingerabdruck sähe sonst aus wie ein
// hinterlegter.
func emptyToNil(rawValue string) *string {
	if rawValue == "" {
		return nil
	}

	return &rawValue
}

// zeroToNil macht aus einer Null ein NULL.
func zeroToNil(rawValue int) *int {
	if rawValue == 0 {
		return nil
	}

	return &rawValue
}

// stringValue liest einen Zeiger auf eine Zeichenkette.
func stringValue(rawValue *string) string {
	if rawValue == nil {
		return ""
	}

	return *rawValue
}

// ListHosts liefert die Knoten eines Verbunds.
func (store *Store) ListHosts(listContext context.Context, clusterIdentifier uuid.UUID) ([]Host, error) {
	const selectStatement = `
		SELECT id, cluster_id, node_name, status, cpu_count, memory_bytes, last_seen_at
		FROM proxmox_hosts
		WHERE cluster_id = $1
		ORDER BY node_name`

	resultRows, queryError := store.connectionPool.Query(listContext, selectStatement, clusterIdentifier)
	if queryError != nil {
		return nil, fmt.Errorf("die knoten liessen sich nicht lesen: %w", queryError)
	}
	defer resultRows.Close()

	hostList := make([]Host, 0, 4)

	for resultRows.Next() {
		var (
			scannedHost Host
			cpuCount    *int
			memoryBytes *int64
		)

		if scanError := resultRows.Scan(&scannedHost.ID, &scannedHost.ClusterID, &scannedHost.NodeName,
			&scannedHost.Status, &cpuCount, &memoryBytes, &scannedHost.LastSeenAt); scanError != nil {
			return nil, fmt.Errorf("ein knoten liess sich nicht lesen: %w", scanError)
		}

		if cpuCount != nil {
			scannedHost.CPUCount = *cpuCount
		}

		if memoryBytes != nil {
			scannedHost.MemoryBytes = *memoryBytes
		}

		hostList = append(hostList, scannedHost)
	}

	return hostList, resultRows.Err()
}

// ErrVirtualMachineNotFound meldet einen unbekannten Gast.
var ErrVirtualMachineNotFound = errors.New("der gast wurde im bestand nicht gefunden")

// GetVirtualMachine liefert einen einzelnen Gast anhand seiner Kennung.
func (store *Store) GetVirtualMachine(readContext context.Context,
	machineIdentifier uuid.UUID) (*VirtualMachine, error) {
	// Über die Listenabfrage, damit die Auswertung der Plattenliste an einer
	// Stelle bleibt: Zwei Auswertungen derselben Spalte laufen auseinander, und
	// man merkt es erst, wenn jemand nachrechnet.
	machineList, listError := store.ListVirtualMachines(readContext, nil)
	if listError != nil {
		return nil, listError
	}

	for _, candidateMachine := range machineList {
		if candidateMachine.ID == machineIdentifier {
			return &candidateMachine, nil
		}
	}

	return nil, ErrVirtualMachineNotFound
}