edgeguard-native/internal/handlers/cluster_repair.go

package handlers

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"os/exec"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/gin-gonic/gin"

	"git.netcell-it.de/projekte/edgeguard-native/internal/handlers/response"
	"git.netcell-it.de/projekte/edgeguard-native/internal/models"
)

// Replication-Repair ("Resync erzwingen") für das Config-Drift-Banner.
//
// Drift entsteht, wenn ein Peer einen anderen config_hash hat als dieser
// Node — entweder weil die Logical-Replication-Subscription gestört ist
// oder weil direkt in die DB des Subscribers geschrieben wurde. Die
// Reparatur baut die Subscription neu auf und kopiert alle geteilten
// Tabellen frisch vom Primary (einseitig: Primary = Source of Truth).
//
// Der Resync MUSS auf dem Standby/Subscriber laufen (nur der hat eine
// Subscription). Operatoren erreichen die UI aber über die VIP, die immer
// auf den Primary zeigt. Deshalb:
//
//   - Auf dem Primary geklickt → Dispatch via mTLS an den Standby
//     (POST /agent/cluster/repair-replication), der dort lokal läuft.
//   - Auf dem Standby direkt geklickt → läuft lokal.
//
// Die eigentliche Arbeit läuft — analog zum Rolling-Update — in einer
// transienten systemd-Unit, die das bereits getestete
// `edgeguard-ctl cluster-setup-standby <primary>` ausführt.

const (
	repairUnitName   = "edgeguard-repair-replication.service"
	repairScriptPath = "/var/lib/edgeguard/repair-replication.sh"
	repairAgentPath  = "/agent/cluster/repair-replication"
)

// validRepairHost erlaubt nur IPv4/IPv6/Hostnamen — der Wert landet in
// einem Bash-Script das als root läuft, also strikt validieren (defense
// in depth, auch wenn er aus ha_nodes stammt).
var validRepairHost = regexp.MustCompile(`^[A-Za-z0-9._:-]{1,253}$`)

// RepairReplication ist der UI-Endpoint. Läuft der lokale Node als
// Primary, wird der Resync an den Standby-Peer delegiert; auf dem Standby
// selbst läuft er lokal.
func (h *ClusterHandler) RepairReplication(c *gin.Context) {
	if h.Store == nil {
		response.Internal(c, errors.New("cluster store unavailable"))
		return
	}
	all, err := h.Store.List(c.Request.Context())
	if err != nil {
		response.Internal(c, err)
		return
	}
	local := findNode(all, h.LocalID)
	standby := findByPGRole(all, "standby")

	// Primary → an den Standby delegieren.
	if local != nil && local.PGRole == "primary" {
		if h.Aggregator == nil {
			response.BadRequest(c, errors.New("kein mTLS-Aggregator verfügbar — Resync nicht delegierbar"))
			return
		}
		if standby == nil {
			response.BadRequest(c, errors.New("kein Standby-Node gefunden, an den der Resync delegiert werden könnte"))
			return
		}
		res := h.Aggregator.PostPeer(c.Request.Context(), *standby, repairAgentPath)
		if !res.OK {
			response.Internal(c, fmt.Errorf("Resync auf %s anstoßen: %s", standby.FQDN, res.Err))
			return
		}
		slog.Info("cluster: replication repair delegated to standby", "standby", standby.FQDN)
		if h.Audit != nil {
			_ = h.Audit.Log(c.Request.Context(), actorOf(c), "cluster.repair-replication",
				standby.FQDN, gin.H{"target": "standby", "standby": standby.FQDN}, h.NodeID)
		}
		response.Accepted(c, gin.H{"dispatched": true, "target": "standby", "standby_fqdn": standby.FQDN})
		return
	}

	// Standby (oder Direktzugriff) → lokal ausführen.
	host, err := h.runLocalRepair(c.Request.Context(), all)
	if err != nil {
		response.BadRequest(c, err)
		return
	}
	if h.Audit != nil {
		_ = h.Audit.Log(c.Request.Context(), actorOf(c), "cluster.repair-replication",
			host, gin.H{"target": "local", "primary": host}, h.NodeID)
	}
	response.Accepted(c, gin.H{"dispatched": true, "target": "local", "primary": host})
}

// AgentRepairReplication wird vom Primary via mTLS auf dem Standby
// aufgerufen und startet dort den lokalen Resync.
func (h *ClusterHandler) AgentRepairReplication(c *gin.Context) {
	if h.Store == nil {
		response.Internal(c, errors.New("cluster store unavailable"))
		return
	}
	all, err := h.Store.List(c.Request.Context())
	if err != nil {
		response.Internal(c, err)
		return
	}
	host, err := h.runLocalRepair(c.Request.Context(), all)
	if err != nil {
		response.BadRequest(c, err)
		return
	}
	slog.Info("cluster: replication repair triggered by peer", "primary", host, "node", h.LocalID)
	if h.Audit != nil {
		_ = h.Audit.Log(c.Request.Context(), "cluster-peer", "cluster.repair-replication",
			host, gin.H{"target": "local", "primary": host, "via": "agent"}, h.NodeID)
	}
	response.Accepted(c, gin.H{"dispatched": true, "primary": host})
}

// runLocalRepair startet den Resync auf DIESEM Node. Verweigert auf dem
// Primary (kein Subscriber). Gibt den ermittelten Primary-Host zurück.
func (h *ClusterHandler) runLocalRepair(_ context.Context, all []models.HANode) (string, error) {
	local := findNode(all, h.LocalID)
	primary := findByPGRole(all, "primary")

	if local != nil && local.PGRole == "primary" {
		return "", errors.New("dieser Node ist der PostgreSQL-Primary — Resync läuft nur auf einem Standby/Subscriber")
	}
	if primary == nil {
		return "", errors.New("kein PostgreSQL-Primary im Cluster gefunden — Resync-Quelle unbekannt")
	}
	if primary.ID == h.LocalID {
		return "", errors.New("der lokale Node ist als Primary markiert — Resync nicht möglich")
	}

	host := pickPrimaryHost(primary)
	if host == "" {
		return "", errors.New("Primary hat keine erreichbare IP/FQDN in ha_nodes")
	}
	if !validRepairHost.MatchString(host) {
		return "", fmt.Errorf("ungültige Primary-Adresse: %q", host)
	}
	if st := repairUnitState(); st == "activating" || st == "active" {
		return "", errors.New("Resync läuft bereits")
	}

	script := fmt.Sprintf(`#!/bin/bash
set -uo pipefail
echo "[repair] resync der Logical-Replication-Subscription von Primary %[1]s"
/usr/bin/edgeguard-ctl cluster-setup-standby %[1]s
rc=$?
if [ "$rc" -ne 0 ]; then
    echo "[repair] cluster-setup-standby fehlgeschlagen (rc=$rc)"
    exit "$rc"
fi
echo "[repair] abgeschlossen — config_hash wird beim nächsten Cluster-Status neu berechnet"
rm -f %[2]s
`, host, repairScriptPath)

	if err := os.WriteFile(repairScriptPath, []byte(script), 0o755); err != nil {
		return "", fmt.Errorf("write repair script: %w", err)
	}
	_ = exec.Command("sudo", "-n", "/usr/bin/systemctl", "reset-failed", repairUnitName).Run()
	cmd := exec.Command("sudo", "-n", "/usr/bin/systemd-run",
		"--unit="+repairUnitName,
		"--description=EdgeGuard replication repair",
		"--collect",
		"bash", repairScriptPath)
	if err := cmd.Run(); err != nil {
		return "", fmt.Errorf("systemd-run failed: %w", err)
	}
	slog.Info("cluster: replication repair dispatched (local)", "primary", host, "node", h.LocalID)
	return host, nil
}

// repairStatusResponse spiegelt den Zustand der transienten Repair-Unit.
type repairStatusResponse struct {
	Phase      string   `json:"phase"` // idle | running | success | failed
	State      string   `json:"state"`
	Result     string   `json:"result"`
	ExitCode   int      `json:"exit_code"`
	StartedAt  string   `json:"started_at,omitempty"`
	FinishedAt string   `json:"finished_at,omitempty"`
	Log        []string `json:"log"`
}

// RepairReplicationStatus liest den Job-Zustand. Auf dem Primary wird der
// Status vom Standby-Peer geholt (dort läuft der Job); sonst lokal.
func (h *ClusterHandler) RepairReplicationStatus(c *gin.Context) {
	if h.Store != nil {
		if all, err := h.Store.List(c.Request.Context()); err == nil {
			local := findNode(all, h.LocalID)
			standby := findByPGRole(all, "standby")
			if local != nil && local.PGRole == "primary" && h.Aggregator != nil && standby != nil {
				results := h.Aggregator.FanOut(c.Request.Context(),
					[]models.HANode{*standby}, repairAgentPath+"/status", h.LocalID)
				if len(results) == 1 && results[0].OK && len(results[0].Data) > 0 {
					c.Data(200, "application/json", wrapEnvelope(results[0].Data))
					return
				}
				// Peer nicht erreichbar → idle zurückgeben statt Fehler,
				// damit das UI-Polling nicht hart abbricht.
				response.OK(c, repairStatusResponse{Phase: "idle", Log: []string{}})
				return
			}
		}
	}
	response.OK(c, localRepairStatus())
}

// AgentRepairReplicationStatus liefert den lokalen Job-Zustand an den
// abfragenden Primary.
func (h *ClusterHandler) AgentRepairReplicationStatus(c *gin.Context) {
	response.OK(c, localRepairStatus())
}

// wrapEnvelope verpackt eine bereits entpackte data-Payload wieder in die
// Standard-Envelope, damit das UI (isEnvelope) sie konsistent liest.
func wrapEnvelope(data []byte) []byte {
	out := []byte(`{"data":`)
	out = append(out, data...)
	out = append(out, []byte(`,"error":null,"message":"ok"}`)...)
	return out
}

// localRepairStatus liest den Zustand der lokalen Repair-Unit aus systemd
// (analog UpgradeStatus). Quelle der Wahrheit für Job-Ende ist die Unit.
func localRepairStatus() repairStatusResponse {
	out := repairStatusResponse{Phase: "idle", Log: []string{}}

	if data, err := exec.Command("systemctl", "show", repairUnitName,
		"--no-page",
		"-p", "ActiveState",
		"-p", "Result",
		"-p", "ExecMainStatus",
		"-p", "ExecMainStartTimestamp",
		"-p", "ExecMainExitTimestamp",
	).CombinedOutput(); err == nil {
		for _, line := range strings.Split(string(data), "\n") {
			kv := strings.SplitN(strings.TrimSpace(line), "=", 2)
			if len(kv) != 2 {
				continue
			}
			switch kv[0] {
			case "ActiveState":
				out.State = kv[1]
			case "Result":
				out.Result = kv[1]
			case "ExecMainStatus":
				out.ExitCode, _ = strconv.Atoi(kv[1])
			case "ExecMainStartTimestamp":
				if t, err := time.Parse("Mon 2006-01-02 15:04:05 MST", kv[1]); err == nil {
					out.StartedAt = t.UTC().Format(time.RFC3339)
				}
			case "ExecMainExitTimestamp":
				if t, err := time.Parse("Mon 2006-01-02 15:04:05 MST", kv[1]); err == nil {
					out.FinishedAt = t.UTC().Format(time.RFC3339)
				}
			}
		}
	}

	switch out.State {
	case "activating", "active", "deactivating":
		out.Phase = "running"
	case "failed":
		out.Phase = "failed"
	case "inactive":
		if out.Result == "success" && out.ExitCode == 0 && out.FinishedAt != "" {
			out.Phase = "success"
		} else if out.Result != "" && out.Result != "success" {
			out.Phase = "failed"
		}
	}

	if data, err := exec.Command("journalctl",
		"-u", repairUnitName,
		"--no-pager", "-n", "100", "-o", "cat",
	).CombinedOutput(); err == nil {
		lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n")
		if !(len(lines) == 1 && (lines[0] == "" || strings.HasPrefix(lines[0], "-- No entries"))) {
			out.Log = lines
		}
	}
	return out
}

// findNode / findByPGRole: kleine Helfer über die ha_nodes-Liste.
func findNode(nodes []models.HANode, id string) *models.HANode {
	for i := range nodes {
		if nodes[i].ID == id {
			return &nodes[i]
		}
	}
	return nil
}

func findByPGRole(nodes []models.HANode, role string) *models.HANode {
	for i := range nodes {
		if nodes[i].PGRole == role {
			return &nodes[i]
		}
	}
	return nil
}

// pickPrimaryHost wählt die beste erreichbare Adresse des Primary:
// Mgmt-IP → Internal-IP → Public-IP → FQDN. Strippt eine etwaige
// CIDR-Maske (inet-Spalten können "10.0.0.5/32" liefern).
func pickPrimaryHost(n *models.HANode) string {
	for _, cand := range []*string{n.MgmtIP, n.InternalIP, n.PublicIP} {
		if cand != nil {
			if h := strings.TrimSpace(strings.SplitN(*cand, "/", 2)[0]); h != "" {
				return h
			}
		}
	}
	return strings.TrimSpace(n.FQDN)
}

// repairUnitState gibt den ActiveState der Repair-Unit zurück ("" wenn
// unbekannt). Für den Doppelstart-Schutz.
func repairUnitState() string {
	out, err := exec.Command("systemctl", "show", repairUnitName, "--no-page", "-p", "ActiveState").CombinedOutput()
	if err != nil {
		return ""
	}
	for _, line := range strings.Split(string(out), "\n") {
		if kv := strings.SplitN(strings.TrimSpace(line), "=", 2); len(kv) == 2 && kv[0] == "ActiveState" {
			return kv[1]
		}
	}
	return ""
}