edgeguard-native/internal/handlers/cluster_repair.go

package handlers

import (
	"context"
	"encoding/json"
	"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).
//
// Rollen-Erkennung: NICHT über ha_nodes.role/pg_role — die sind je Node
// lokal und unzuverlässig (jede Node markiert sich selbst, pg_role bleibt
// 'standalone' bis `promote`). Verlässlich ist die PUBLICATION: nur der
// Primary hat `edgeguard_shared` (pg_publication ist für jeden DB-User
// lesbar). Der Subscriber hat sie nicht → er ist das Resync-Ziel.
//
// Ablauf:
//   - Klick auf dem Primary → Dispatch via mTLS an den Peer
//     (POST /agent/cluster/repair-replication) mit der eigenen Adresse als
//     primary_host; der Peer resynct von dort.
//   - Klick direkt auf dem Subscriber → läuft lokal (Quelle = der Peer).
//
// Die eigentliche Arbeit läuft — analog zum Rolling-Update — in einer
// transienten systemd-Unit, die `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"
	repairPubName    = "edgeguard_shared" // muss zu cmd/edgeguard-ctl egPubName passen
)

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

// repairDispatchBody ist der Body des Agent-Dispatch: der Primary teilt
// dem Subscriber seine Adresse mit, von der resynct werden soll.
type repairDispatchBody struct {
	PrimaryHost string `json:"primary_host"`
}

// RepairReplication ist der UI-Endpoint. Hat dieser Node die Publication
// (= Primary), wird der Resync an den Peer delegiert; sonst (Subscriber)
// läuft er lokal mit dem Peer als Quelle.
func (h *ClusterHandler) RepairReplication(c *gin.Context) {
	if h.Store == nil {
		response.Internal(c, errors.New("cluster store unavailable"))
		return
	}
	ctx := c.Request.Context()
	all, err := h.Store.List(ctx)
	if err != nil {
		response.Internal(c, err)
		return
	}
	local := findNode(all, h.LocalID)
	peer := findOtherPeer(all, h.LocalID)
	if peer == nil {
		response.BadRequest(c, errors.New("kein Peer-Node im Cluster — nichts zu resyncen"))
		return
	}

	if h.nodeHasPublication(ctx) {
		// Primary → an den Subscriber-Peer delegieren, mit eigener Adresse.
		if h.Aggregator == nil {
			response.BadRequest(c, errors.New("kein mTLS-Aggregator verfügbar — Resync nicht delegierbar"))
			return
		}
		primaryHost := pickPrimaryHost(local)
		if primaryHost == "" || !validRepairHost.MatchString(primaryHost) {
			response.BadRequest(c, errors.New("eigene Primary-Adresse (Mgmt/Internal/Public-IP/FQDN) fehlt oder ist ungültig"))
			return
		}
		body, _ := json.Marshal(repairDispatchBody{PrimaryHost: primaryHost})
		res := h.Aggregator.PostPeerWithBody(ctx, *peer, repairAgentPath, body)
		if !res.OK {
			response.Internal(c, fmt.Errorf("Resync auf %s anstoßen: %s", peer.FQDN, res.Err))
			return
		}
		slog.Info("cluster: replication repair delegated", "target", peer.FQDN, "primary_host", primaryHost)
		if h.Audit != nil {
			_ = h.Audit.Log(ctx, actorOf(c), "cluster.repair-replication",
				peer.FQDN, gin.H{"target": "peer", "peer": peer.FQDN, "primary_host": primaryHost}, h.NodeID)
		}
		response.Accepted(c, gin.H{"dispatched": true, "target": "peer", "peer_fqdn": peer.FQDN})
		return
	}

	// Subscriber → lokal ausführen, Quelle = der Peer (Primary).
	host := pickPrimaryHost(peer)
	if err := h.startResync(ctx, host); err != nil {
		response.BadRequest(c, err)
		return
	}
	if h.Audit != nil {
		_ = h.Audit.Log(ctx, 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 Subscriber
// aufgerufen und startet dort den lokalen Resync von primary_host.
func (h *ClusterHandler) AgentRepairReplication(c *gin.Context) {
	if h.Store == nil {
		response.Internal(c, errors.New("cluster store unavailable"))
		return
	}
	ctx := c.Request.Context()
	var body repairDispatchBody
	_ = c.ShouldBindJSON(&body) // best-effort; Fallback unten

	host := strings.TrimSpace(body.PrimaryHost)
	if host == "" {
		// Fallback: Quelle aus ha_nodes (der andere Node).
		if all, err := h.Store.List(ctx); err == nil {
			host = pickPrimaryHost(findOtherPeer(all, h.LocalID))
		}
	}
	if err := h.startResync(ctx, host); 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(ctx, "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})
}

// startResync schreibt das Repair-Script und startet die transiente
// systemd-Unit. Safety-Guard: läuft NIE auf dem Publication-Primary.
func (h *ClusterHandler) startResync(ctx context.Context, primaryHost string) error {
	primaryHost = strings.TrimSpace(primaryHost)
	if primaryHost == "" {
		return errors.New("keine Primary-Adresse für den Resync ermittelbar")
	}
	if !validRepairHost.MatchString(primaryHost) {
		return fmt.Errorf("ungültige Primary-Adresse: %q", primaryHost)
	}
	// Niemals auf dem Primary (Publication-Quelle) resyncen — würde die
	// eigene Config mit sich selbst überschreiben bzw. ist sinnlos.
	if h.nodeHasPublication(ctx) {
		return errors.New("dieser Node ist der Publication-Primary — Resync läuft nur auf einem Subscriber")
	}
	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
`, primaryHost, 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", primaryHost, "node", h.LocalID)
	return nil
}

// nodeHasPublication prüft, ob dieser Node die Replikations-Publication
// besitzt — das verlässliche Primary-Signal. pg_publication ist für jeden
// DB-User lesbar (anders als pg_subscription).
func (h *ClusterHandler) nodeHasPublication(ctx context.Context) bool {
	if h.Store == nil || h.Store.Pool == nil {
		return false
	}
	cctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	var exists bool
	if err := h.Store.Pool.QueryRow(cctx,
		`SELECT EXISTS(SELECT 1 FROM pg_publication WHERE pubname = $1)`, repairPubName,
	).Scan(&exists); err != nil {
		return false
	}
	return exists
}

// 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 Subscriber-Peer geholt (dort läuft der Job); sonst lokal.
func (h *ClusterHandler) RepairReplicationStatus(c *gin.Context) {
	ctx := c.Request.Context()
	if h.Store != nil && h.nodeHasPublication(ctx) && h.Aggregator != nil {
		if all, err := h.Store.List(ctx); err == nil {
			if peer := findOtherPeer(all, h.LocalID); peer != nil {
				results := h.Aggregator.FanOut(ctx,
					[]models.HANode{*peer}, 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 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 liefert die ha_nodes-Row mit der gegebenen ID.
func findNode(nodes []models.HANode, id string) *models.HANode {
	for i := range nodes {
		if nodes[i].ID == id {
			return &nodes[i]
		}
	}
	return nil
}

// findOtherPeer liefert den (einen) anderen Node im 2-Node-Cluster.
// Bevorzugt einen online erreichbaren Peer.
func findOtherPeer(nodes []models.HANode, localID string) *models.HANode {
	var fallback *models.HANode
	for i := range nodes {
		n := &nodes[i]
		if n.ID == localID {
			continue
		}
		if n.Status == "online" {
			return n
		}
		if fallback == nil {
			fallback = n
		}
	}
	return fallback
}

// pickPrimaryHost wählt die beste erreichbare Adresse eines Node:
// 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 {
	if n == nil {
		return ""
	}
	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 ""
}