edgeguard-native/cmd/edgeguard-ctl/cluster_reconcile.go

package main

import (
	"fmt"
	"os"
	"sort"
	"strings"
)

// cmdClusterReconcileReplication bringt Publication, Grants und Subscription
// idempotent in den Soll-Zustand. Verhindert die zwei Fehlermodi, die sich
// beim Nachrüsten von Features zeigen:
//   - Eine `FOR TABLE`-Publication nimmt später per Migration hinzugekommene
//     Shared-Tables NICHT automatisch auf → sie replizieren nie (Standby
//     läuft nach Failover ohne WAF/OIDC/DHCP/RADIUS-Config).
//   - Der GRANT SELECT für den Replikations-User ist ein Snapshot bei Setup;
//     neue Tabellen fehlen → tablesync hängt in `d` (Permission).
//
// Rollen-Selbsterkennung (idempotent, läuft im postinst nach migrate):
//
//	Publisher (hat Publication):
//	  - GRANT SELECT auf ALLE Tabellen (+ DEFAULT PRIVILEGES) für den
//	    Replikations-User.
//	  - Publication-Mitgliedschaft angleichen: fehlende Shared-Tables ADD,
//	    fälschlich enthaltene node-lokale (localOnlyTables) DROP.
//	Subscriber (hat Subscription):
//	  - Frisch zu synchronisierende Shared-Tables lokal TRUNCATE (Primary =
//	    Source of Truth; verhindert Duplicate-Key beim Initial-COPY einer
//	    per Migration seed-befüllten Singleton-Tabelle), dann REFRESH.
//	Single-Node (weder noch): nichts zu tun.
//
// Best-effort: Fehler werden geloggt, brechen aber ein Paket-Upgrade nie ab.
func cmdClusterReconcileReplication(_ []string) int {
	hasPub := psqlDBBool("edgeguard",
		fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_publication WHERE pubname='%s')", egPubName))
	hasSub := psqlDBBool("edgeguard",
		fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_subscription WHERE subname='%s')", egSubName))

	switch {
	case hasPub:
		reconcilePublisher()
	case hasSub:
		reconcileSubscriber()
	default:
		// Standalone-Node — keine Replikation eingerichtet.
	}
	return 0
}

// reconcilePublisher gleicht Grants + Publication-Mitgliedschaft an.
func reconcilePublisher() {
	// 1. Grants IMMER neu setzen (idempotent, deckt neue Tabellen ab).
	grantSQL := fmt.Sprintf(
		"GRANT SELECT ON ALL TABLES IN SCHEMA public TO %s;\n"+
			"ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO %s;",
		egReplUser, egReplUser)
	if err := psqlDBExec("edgeguard", grantSQL); err != nil {
		fmt.Fprintln(os.Stderr, "reconcile: GRANT SELECT fehlgeschlagen:", err)
	} else {
		fmt.Printf("✓ reconcile: SELECT-Grants für %q aktualisiert\n", egReplUser)
	}

	// 2. Publication-Mitgliedschaft angleichen.
	desired, err := desiredSharedTables()
	if err != nil {
		fmt.Fprintln(os.Stderr, "reconcile: Tabellen-Liste:", err)
		return
	}
	current, err := publicationTables()
	if err != nil {
		fmt.Fprintln(os.Stderr, "reconcile: Publication-Liste:", err)
		return
	}
	desiredSet := toSet(desired)
	currentSet := toSet(current)

	var toAdd, toDrop []string
	for _, t := range desired {
		if !currentSet[t] {
			toAdd = append(toAdd, t)
		}
	}
	for _, t := range current {
		if !desiredSet[t] {
			toDrop = append(toDrop, t) // node-lokale, die fälschlich drin sind
		}
	}
	sort.Strings(toAdd)
	sort.Strings(toDrop)

	if len(toAdd) > 0 {
		if err := psqlDBExec("edgeguard", fmt.Sprintf(
			"ALTER PUBLICATION %s ADD TABLE %s;", egPubName, strings.Join(toAdd, ", "))); err != nil {
			fmt.Fprintln(os.Stderr, "reconcile: ADD TABLE fehlgeschlagen:", err)
		} else {
			fmt.Printf("✓ reconcile: %d Tabelle(n) zur Publication hinzugefügt: %s\n",
				len(toAdd), strings.Join(toAdd, ", "))
		}
	}
	if len(toDrop) > 0 {
		if err := psqlDBExec("edgeguard", fmt.Sprintf(
			"ALTER PUBLICATION %s DROP TABLE %s;", egPubName, strings.Join(toDrop, ", "))); err != nil {
			fmt.Fprintln(os.Stderr, "reconcile: DROP TABLE fehlgeschlagen:", err)
		} else {
			fmt.Printf("✓ reconcile: %d node-lokale Tabelle(n) aus Publication entfernt: %s\n",
				len(toDrop), strings.Join(toDrop, ", "))
		}
	}
	if len(toAdd) == 0 && len(toDrop) == 0 {
		fmt.Println("✓ reconcile: Publication bereits im Soll-Zustand")
	}
}

// reconcileSubscriber zieht neu publizierte Tabellen nach: erst lokal leeren
// (Primary = Source of Truth, verhindert Duplicate-Key beim Initial-COPY),
// dann REFRESH PUBLICATION. Bereits synchronisierte Tabellen bleiben unberührt.
func reconcileSubscriber() {
	desired, err := desiredSharedTables()
	if err != nil {
		fmt.Fprintln(os.Stderr, "reconcile: Tabellen-Liste:", err)
		return
	}
	synced, err := subscriptionRelTables()
	if err != nil {
		fmt.Fprintln(os.Stderr, "reconcile: subscription_rel-Liste:", err)
		return
	}
	syncedSet := toSet(synced)

	var fresh []string
	for _, t := range desired {
		if !syncedSet[t] {
			fresh = append(fresh, t)
		}
	}
	sort.Strings(fresh)

	if len(fresh) > 0 {
		// Nur frisch zu synchronisierende Shared-Tables leeren — nie eine
		// bereits replizierte oder node-lokale Tabelle.
		if err := psqlDBExec("edgeguard",
			fmt.Sprintf("TRUNCATE %s;", strings.Join(fresh, ", "))); err != nil {
			fmt.Fprintln(os.Stderr, "reconcile: TRUNCATE (neue Tabellen) fehlgeschlagen:", err)
		} else {
			fmt.Printf("✓ reconcile: %d neue Tabelle(n) für Initial-Sync geleert: %s\n",
				len(fresh), strings.Join(fresh, ", "))
		}
	}

	// REFRESH ist NICHT transaktionssicher → einzelnes Statement, autocommit.
	if err := psqlDBExec("edgeguard",
		fmt.Sprintf("ALTER SUBSCRIPTION %s REFRESH PUBLICATION;", egSubName)); err != nil {
		fmt.Fprintln(os.Stderr, "reconcile: REFRESH PUBLICATION fehlgeschlagen:", err)
	} else {
		fmt.Printf("✓ reconcile: Subscription %q refresht\n", egSubName)
	}
}

// desiredSharedTables = alle public-Tabellen minus localOnlyTables.
func desiredSharedTables() ([]string, error) {
	out, err := psqlDBRun("edgeguard", []string{"-tA", "-c",
		`SELECT tablename FROM pg_tables WHERE schemaname='public'`})
	if err != nil {
		return nil, fmt.Errorf("list tables: %w", err)
	}
	return filterSharedTables(splitLines(string(out))), nil
}

// filterSharedTables entfernt localOnlyTables aus der Tabellenliste. Pure
// Funktion — unit-testbar.
func filterSharedTables(all []string) []string {
	excluded := toSet(localOnlyTables)
	var out []string
	for _, t := range all {
		if t != "" && !excluded[t] {
			out = append(out, t)
		}
	}
	sort.Strings(out)
	return out
}

// publicationTables listet die aktuell in edgeguard_shared publizierten Tabellen.
func publicationTables() ([]string, error) {
	out, err := psqlDBRun("edgeguard", []string{"-tA", "-c",
		fmt.Sprintf("SELECT tablename FROM pg_publication_tables WHERE pubname='%s'", egPubName)})
	if err != nil {
		return nil, err
	}
	return splitLines(string(out)), nil
}

// subscriptionRelTables listet die Tabellen, die die Subscription bereits kennt.
func subscriptionRelTables() ([]string, error) {
	out, err := psqlDBRun("edgeguard", []string{"-tA", "-c",
		fmt.Sprintf(`SELECT c.relname FROM pg_subscription_rel r
		  JOIN pg_class c ON c.oid = r.srrelid
		  JOIN pg_subscription s ON s.oid = r.srsubid
		 WHERE s.subname = '%s'`, egSubName)})
	if err != nil {
		return nil, err
	}
	return splitLines(string(out)), nil
}

func psqlDBBool(db, sql string) bool {
	out, err := psqlDBRun(db, []string{"-tA", "-c", sql})
	if err != nil {
		return false
	}
	return strings.TrimSpace(string(out)) == "t"
}

func splitLines(s string) []string {
	var out []string
	for _, l := range strings.Split(strings.TrimSpace(s), "\n") {
		if l = strings.TrimSpace(l); l != "" {
			out = append(out, l)
		}
	}
	return out
}

func toSet(items []string) map[string]bool {
	m := make(map[string]bool, len(items))
	for _, it := range items {
		m[it] = true
	}
	return m
}