sultan/pkg/server/server_test.go

package server

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	"sultan/pkg/auth"
	"sultan/pkg/crypto"
	"sultan/pkg/license"
	"sultan/pkg/notify"
	"sultan/pkg/policy"
	"sultan/pkg/registry"
	"sultan/pkg/store"
	"sultan/pkg/stream"
)

var ctx = context.Background()

// syncBuffer sammelt Log-Ausgaben threadsicher.
type syncBuffer struct {
	mu sync.Mutex
	b  bytes.Buffer
}

func (s *syncBuffer) Write(p []byte) (int, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.b.Write(p)
}
func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }

func eventually(t *testing.T, msg string, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for !cond() {
		if time.Now().After(deadline) {
			t.Fatalf("Bedingung nicht erfüllt: %s", msg)
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// startServer startet einen Server auf freien Ports in einem frischen
// Datenverzeichnis.
func startServer(t *testing.T, mutate ...func(*Config)) (*Server, *syncBuffer) {
	t.Helper()
	logs := &syncBuffer{}
	cfg := Config{
		DataDir:         filepath.Join(t.TempDir(), "suldata"),
		AgentAddr:       "127.0.0.1:0",
		ProviderAddr:    "127.0.0.1:0",
		LanderAddr:      "127.0.0.1:0",
		RefreshInterval: time.Hour,
		Logger:          slog.New(slog.NewTextHandler(logs, nil)),
	}
	for _, m := range mutate {
		m(&cfg)
	}
	s, err := New(ctx, cfg)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	if err := s.Start(); err != nil {
		t.Fatalf("Start: %v", err)
	}
	t.Cleanup(func() {
		sctx, cancel := context.WithTimeout(ctx, 5*time.Second)
		defer cancel()
		s.Shutdown(sctx)
	})
	return s, logs
}

func createTenant(t *testing.T, s *Server, id string, domains ...string) {
	t.Helper()
	if _, err := s.store.CreateTenant(ctx, store.Tenant{
		ID: id, Name: id, Email: id + "@example.com", Tier: store.TierPro, MaxRoutes: 10,
		AllowedDomains: domains, ValidUntil: time.Now().AddDate(1, 0, 0),
	}); err != nil {
		t.Fatal(err)
	}
}

// issueToken erzeugt einen Enrollment-Token wie `sultan token` es tun wird.
func issueToken(t *testing.T, s *Server, tenant string) string {
	t.Helper()
	tok, err := crypto.NewEnrollmentToken(s.Fingerprint(), "")
	if err != nil {
		t.Fatal(err)
	}
	if err := s.store.AddEnrollmentToken(ctx, tok.Secret, tenant, time.Now().Add(time.Hour)); err != nil {
		t.Fatal(err)
	}
	return tok.String()
}

// testAgent spielt die Agent-Seite mit Pinning und Signatur.
type testAgent struct {
	base   string
	hc     *http.Client
	signer *auth.Signer
	secret string
}

func newAgent(t *testing.T, s *Server, token string) *testAgent {
	t.Helper()
	tok, err := crypto.ParseEnrollmentToken(token)
	if err != nil {
		t.Fatal(err)
	}
	priv, _ := auth.GenerateKey()
	return &testAgent{
		base:   "https://" + s.AgentAddr(),
		hc:     &http.Client{Transport: &http.Transport{TLSClientConfig: crypto.PinnedTLSConfig(tok.Fingerprint)}},
		signer: &auth.Signer{Key: priv},
		secret: tok.Secret,
	}
}

func (a *testAgent) do(t *testing.T, method, path string, body any) *http.Response {
	t.Helper()
	var rd io.Reader
	if body != nil {
		b, _ := json.Marshal(body)
		rd = bytes.NewReader(b)
	}
	req, _ := http.NewRequest(method, a.base+path, rd)
	if err := a.signer.Sign(req); err != nil {
		t.Fatal(err)
	}
	resp, err := a.hc.Do(req)
	if err != nil {
		t.Fatalf("%s %s: %v", method, path, err)
	}
	t.Cleanup(func() { resp.Body.Close() })
	return resp
}

func (a *testAgent) enroll(t *testing.T, nodeID string) *http.Response {
	return a.do(t, http.MethodPost, "/api/v1/enroll", EnrollRequest{Token: a.secret, NodeID: nodeID, Hostname: "host-" + nodeID})
}

// connect hält den Stream im Hintergrund offen, bis cancel aufgerufen wird.
func (a *testAgent) connect(t *testing.T) (cancel func()) {
	t.Helper()
	cancel, _ = a.connectEvents(t)
	return cancel
}

// connectEvents hält den Stream offen und liefert die empfangenen Events.
func (a *testAgent) connectEvents(t *testing.T) (cancel func(), events <-chan stream.Event) {
	t.Helper()
	sctx, cancel := context.WithCancel(ctx)
	ch := make(chan stream.Event, 64)
	c := &stream.Client{URL: a.base + "/api/v1/stream", HTTPClient: a.hc, Prepare: a.signer.Sign, MinBackoff: 10 * time.Millisecond}
	done := make(chan struct{})
	go func() {
		c.Run(sctx, func(ev stream.Event) error {
			select {
			case ch <- ev:
			default:
			}
			return nil
		})
		close(done)
	}()
	t.Cleanup(func() { cancel(); <-done })
	return cancel, ch
}

// nextEvent wartet auf das nächste Event des Typs typ.
func nextEvent(t *testing.T, events <-chan stream.Event, typ stream.EventType) stream.Event {
	t.Helper()
	deadline := time.After(3 * time.Second)
	for {
		select {
		case ev := <-events:
			if ev.Type == typ {
				return ev
			}
		case <-deadline:
			t.Fatalf("kein Event %s empfangen", typ)
		}
	}
}

func wpState(domain string) registry.NodeState {
	return registry.NodeState{Workloads: []registry.Workload{{
		Name: "wordpress", Host: "100.64.0.42",
		Labels: map[string]string{
			"traefik.enable":                                    "true",
			"traefik.http.routers.wp.rule":                      "Host(`" + domain + "`)",
			"traefik.http.services.wp.loadbalancer.server.port": "80",
		},
	}}}
}

func providerConfig(t *testing.T, s *Server) string {
	t.Helper()
	resp, err := http.Get("http://" + s.ProviderAddr() + "/v1/traefik-config")
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(resp.Body)
	return string(b)
}

func TestConfigDefaults(t *testing.T) {
	c := Config{}.withDefaults()
	if c.DataDir != "./suldata" || c.AgentAddr != ":8443" || c.ProviderAddr != "127.0.0.1:8080" {
		t.Errorf("Defaults: %+v", c)
	}
	if c.InterceptorURL != "http://127.0.0.1:8081" || c.LanderAddr != "127.0.0.1:8081" || c.RefreshInterval != time.Minute || c.Logger == nil {
		t.Errorf("Defaults: %+v", c)
	}
}

func TestNew_CreatesDataDir(t *testing.T) {
	dir := filepath.Join(t.TempDir(), "a", "suldata")
	s, err := New(ctx, Config{DataDir: dir, Logger: slog.New(slog.DiscardHandler)})
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	defer s.Shutdown(ctx)
	info, err := os.Stat(dir)
	if err != nil || info.Mode().Perm() != 0o700 {
		t.Fatalf("Datenverzeichnis: %v, %v", info, err)
	}
	for _, f := range []string{"sultan.db", "server.key", "server.crt", "master.key"} {
		if _, err := os.Stat(filepath.Join(dir, f)); err != nil {
			t.Errorf("%s fehlt: %v", f, err)
		}
	}
	if info, _ := os.Stat(filepath.Join(dir, "master.key")); info.Mode().Perm() != 0o600 {
		t.Errorf("master.key hat %o", info.Mode().Perm())
	}
}

// Nach einem Neustart müssen Zertifikat und Master-Key gleich bleiben,
// sonst sperren sich alle Agents aus bzw. alle Lizenzen werden ungültig.
func TestNew_StableIdentityAcrossRestarts(t *testing.T) {
	dir := filepath.Join(t.TempDir(), "suldata")
	cfg := Config{DataDir: dir, Logger: slog.New(slog.DiscardHandler)}
	a, err := New(ctx, cfg)
	if err != nil {
		t.Fatal(err)
	}
	fp, mk := a.Fingerprint(), a.MasterPublicKey()
	a.Shutdown(ctx)

	b, err := New(ctx, cfg)
	if err != nil {
		t.Fatalf("Neustart: %v", err)
	}
	defer b.Shutdown(ctx)
	if b.Fingerprint() != fp || !b.MasterPublicKey().Equal(mk) {
		t.Error("Identität hat sich beim Neustart geändert")
	}
}

func TestProviderPort(t *testing.T) {
	s, _ := startServer(t)
	if got := providerConfig(t, s); got != "{}" {
		t.Errorf("leere Konfiguration = %q", got)
	}
	resp, err := http.Get("http://" + s.ProviderAddr() + "/healthz")
	if err != nil {
		t.Fatal(err)
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Errorf("/healthz: %d", resp.StatusCode)
	}
	// Agent-Routen gibt es auf dem Klartext-Port nicht.
	resp, _ = http.Post("http://"+s.ProviderAddr()+"/api/v1/enroll", "application/json", nil)
	resp.Body.Close()
	if resp.StatusCode != http.StatusNotFound {
		t.Errorf("Agent-Route auf Provider-Port: %d", resp.StatusCode)
	}
}

func TestAgentPort_NoPlainHTTP(t *testing.T) {
	s, _ := startServer(t)
	resp, err := http.Get("http://" + s.AgentAddr() + "/api/v1/stream")
	if err == nil {
		defer resp.Body.Close()
		if resp.StatusCode == http.StatusOK {
			t.Error("Agent-Port antwortet im Klartext mit 200")
		}
	}
}

// Ablauf wie im Betrieb: Enrollment, Stream, Zustand melden, Route bei
// Traefik, Agent weg → Route weg und Knoten offline.
func TestEndToEnd(t *testing.T) {
	s, _ := startServer(t)
	createTenant(t, s, "kunde", "kunde.de", "*.kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))

	resp := agent.enroll(t, "node-1")
	if resp.StatusCode != http.StatusCreated {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("Enrollment: %d %s", resp.StatusCode, b)
	}
	var er EnrollResponse
	json.NewDecoder(resp.Body).Decode(&er)
	if er.NodeID != "node-1" || er.TenantID != "kunde" || er.MasterPublicKey != auth.EncodePublicKey(s.MasterPublicKey()) {
		t.Errorf("Antwort: %+v", er)
	}

	stop := agent.connect(t)
	eventually(t, "Stream verbunden", func() bool { return s.hub.Connected("node-1") })
	eventually(t, "Knoten online", func() bool {
		n, _ := s.store.GetNode(ctx, "node-1")
		return n.Status == store.NodeOnline
	})

	if resp := agent.do(t, http.MethodPost, "/api/v1/state", wpState("kunde.de")); resp.StatusCode != http.StatusNoContent {
		t.Fatalf("State: %d", resp.StatusCode)
	}
	if cfg := providerConfig(t, s); !strings.Contains(cfg, "kunde__node-1__wp") || !strings.Contains(cfg, "http://100.64.0.42:80") {
		t.Errorf("Traefik-Konfiguration: %s", cfg)
	}

	// Ungeplant getrennt: Routen bleiben als Wartungsseite, solange die Lease gilt.
	stop()
	eventually(t, "Wartungsseite nach Trennung", func() bool {
		return strings.Contains(providerConfig(t, s), policy.InterceptorService)
	})
	eventually(t, "Knoten offline", func() bool {
		n, _ := s.store.GetNode(ctx, "node-1")
		return n.Status == store.NodeOffline
	})
}

func TestEnroll_Errors(t *testing.T) {
	s, _ := startServer(t)
	createTenant(t, s, "kunde", "kunde.de")

	used := newAgent(t, s, issueToken(t, s, "kunde"))
	if r := used.enroll(t, "n1"); r.StatusCode != http.StatusCreated {
		t.Fatalf("erstes Enrollment: %d", r.StatusCode)
	}
	again := newAgent(t, s, issueToken(t, s, "kunde"))
	again.secret = used.secret // bereits eingelöst
	if r := again.enroll(t, "n2"); r.StatusCode != http.StatusForbidden {
		t.Errorf("verbrauchter Token: %d", r.StatusCode)
	}

	unknown := newAgent(t, s, issueToken(t, s, "kunde"))
	unknown.secret = crypto.TokenPrefix + strings.Repeat("A", 43)
	if r := unknown.enroll(t, "n3"); r.StatusCode != http.StatusForbidden {
		t.Errorf("unbekannter Token: %d", r.StatusCode)
	}

	if r := newAgent(t, s, issueToken(t, s, "kunde")).enroll(t, "a__b"); r.StatusCode != http.StatusBadRequest {
		t.Errorf("ungültige Knoten-ID: %d", r.StatusCode)
	}
	if r := newAgent(t, s, issueToken(t, s, "kunde")).enroll(t, "n1"); r.StatusCode != http.StatusConflict {
		t.Errorf("doppelte Knoten-ID: %d", r.StatusCode)
	}

	// Ohne Signatur gibt es nichts.
	plain := newAgent(t, s, issueToken(t, s, "kunde"))
	req, _ := http.NewRequest(http.MethodPost, plain.base+"/api/v1/enroll", strings.NewReader(`{}`))
	resp, err := plain.hc.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusUnauthorized {
		t.Errorf("unsigniert: %d", resp.StatusCode)
	}
}

func TestState_Errors(t *testing.T) {
	s, _ := startServer(t)
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")

	// Zustand nur bei offenem Stream (Lebendigkeit hängt am Stream).
	if r := agent.do(t, http.MethodPost, "/api/v1/state", wpState("kunde.de")); r.StatusCode != http.StatusConflict {
		t.Errorf("ohne Stream: %d", r.StatusCode)
	}
	agent.connect(t)
	eventually(t, "Stream", func() bool { return s.hub.Connected("n1") })
	req, _ := http.NewRequest(http.MethodPost, agent.base+"/api/v1/state", strings.NewReader("kein json"))
	agent.signer.Sign(req)
	resp, err := agent.hc.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusBadRequest {
		t.Errorf("kaputtes JSON: %d", resp.StatusCode)
	}

	stranger := newAgent(t, s, issueToken(t, s, "kunde")) // nicht enrollt
	if r := stranger.do(t, http.MethodPost, "/api/v1/state", wpState("kunde.de")); r.StatusCode != http.StatusUnauthorized {
		t.Errorf("nicht registriert: %d", r.StatusCode)
	}
}

func TestState_ViolationsAreLogged(t *testing.T) {
	s, logs := startServer(t)
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	agent.connect(t)
	eventually(t, "Stream", func() bool { return s.hub.Connected("n1") })

	if r := agent.do(t, http.MethodPost, "/api/v1/state", wpState("bank.de")); r.StatusCode != http.StatusNoContent {
		t.Fatalf("State: %d", r.StatusCode)
	}
	if cfg := providerConfig(t, s); cfg != "{}" {
		t.Errorf("fremde Domain ausgeliefert: %s", cfg)
	}
	out := logs.String()
	if !strings.Contains(out, "Policy-Verstoß") || !strings.Contains(out, "kind=domain") || !strings.Contains(out, "node=n1") {
		t.Errorf("Audit-Log fehlt:\n%s", out)
	}
}

func TestRefreshLoopAppliesTenantChanges(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.RefreshInterval = 20 * time.Millisecond })
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	agent.connect(t)
	eventually(t, "Stream", func() bool { return s.hub.Connected("n1") })
	agent.do(t, http.MethodPost, "/api/v1/state", wpState("kunde.de"))

	if err := s.store.SetTenantStatus(ctx, "kunde", store.TenantSuspendedPayment); err != nil {
		t.Fatal(err)
	}
	eventually(t, "Umleitung zum Interceptor", func() bool {
		return strings.Contains(providerConfig(t, s), policy.InterceptorService)
	})
}

// Der Regelfall beim Kunden: Der Dienst meldet eine Domain, die erst danach
// beim Mandanten eingetragen wird. Die Route erscheint ohne neue Meldung des
// Agents; entzieht man die Domain wieder, verschwindet sie.
func TestRefreshLoopAppliesDomainChanges(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.RefreshInterval = 20 * time.Millisecond })
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	agent.connect(t)
	eventually(t, "Stream", func() bool { return s.hub.Connected("n1") })
	agent.do(t, http.MethodPost, "/api/v1/state", wpState("neu.de"))
	eventually(t, "Verstoß: keine Route", func() bool { return providerConfig(t, s) == "{}" })

	setDomains := func(domains ...string) {
		t.Helper()
		tn, err := s.store.GetTenant(ctx, "kunde")
		if err != nil {
			t.Fatal(err)
		}
		tn.AllowedDomains = domains
		if _, err := s.store.UpdateTenant(ctx, tn); err != nil {
			t.Fatal(err)
		}
	}
	setDomains("kunde.de", "neu.de")
	eventually(t, "Route nach Freigabe", func() bool { return strings.Contains(providerConfig(t, s), "neu.de") })
	setDomains("kunde.de")
	eventually(t, "Route nach Entzug weg", func() bool { return providerConfig(t, s) == "{}" })
}

func TestDirectory(t *testing.T) {
	s, _ := startServer(t)
	createTenant(t, s, "kunde", "kunde.de")
	if _, err := s.store.CreateNode(ctx, store.Node{ID: "n1", TenantID: "kunde", Hostname: "h", PublicKey: "k", IPAddress: "1.2.3.4"}); err != nil {
		t.Fatal(err)
	}
	d := &directory{store: s.store}

	nc, err := d.NodeContext(ctx, "n1")
	if err != nil {
		t.Fatal(err)
	}
	if nc.Tenant.ID != "kunde" || nc.License.TenantID != "kunde" || nc.License.MaxRoutes != 10 ||
		!nc.License.AllowsDomain("kunde.de") || nc.Drained {
		t.Errorf("NodeContext: %+v", nc)
	}
	s.store.SetNodeStatus(ctx, "n1", store.NodeDrained)
	if nc, _ := d.NodeContext(ctx, "n1"); !nc.Drained {
		t.Error("gedraint nicht erkannt")
	}
	if _, err := d.NodeContext(ctx, "fehlt"); !errors.Is(err, registry.ErrUnknownNode) {
		t.Errorf("unbekannter Knoten: %v", err)
	}
}

// Herunterfahren darf nicht an offenen SSE-Streams hängen bleiben.
func TestRun_ShutdownWithOpenStream(t *testing.T) {
	cfg := Config{
		DataDir: filepath.Join(t.TempDir(), "suldata"), AgentAddr: "127.0.0.1:0", ProviderAddr: "127.0.0.1:0",
		LanderAddr: "127.0.0.1:0", Logger: slog.New(slog.DiscardHandler),
	}
	s, err := New(ctx, cfg)
	if err != nil {
		t.Fatal(err)
	}
	runCtx, cancel := context.WithCancel(ctx)
	done := make(chan error, 1)
	go func() { done <- s.Run(runCtx) }()
	eventually(t, "gestartet", func() bool { return s.AgentAddr() != "" && !strings.HasSuffix(s.AgentAddr(), ":0") })

	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	agent.connect(t)
	eventually(t, "Stream", func() bool { return s.hub.Connected("n1") })

	cancel()
	select {
	case err := <-done:
		if err != nil {
			t.Errorf("Run: %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("Run hängt beim Herunterfahren")
	}
}

func TestStart_PortInUse(t *testing.T) {
	a, _ := startServer(t)
	b, err := New(ctx, Config{
		DataDir: filepath.Join(t.TempDir(), "suldata"), AgentAddr: a.AgentAddr(), ProviderAddr: "127.0.0.1:0",
		LanderAddr: "127.0.0.1:0", Logger: slog.New(slog.DiscardHandler),
	})
	if err != nil {
		t.Fatal(err)
	}
	defer b.Shutdown(ctx)
	if err := b.Start(); err == nil {
		t.Error("belegter Port: erwartet Fehler")
	}
}

// Wie Traefik: Konfiguration lesen, Router zum Interceptor finden, dessen
// Header-Middleware anwenden und den Lander aufrufen.
func TestLander_EndToEndMaintenance(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.ProURL = "https://hoster.example/pro" })
	if s.LanderAddr() == "" || strings.HasSuffix(s.LanderAddr(), ":0") {
		t.Fatalf("Lander-Adresse: %q", s.LanderAddr())
	}
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	agent.connect(t)
	eventually(t, "Stream", func() bool { return s.hub.Connected("n1") })
	st := wpState("kunde.de")
	st.DownServices = []string{"wp"}
	if r := agent.do(t, http.MethodPost, "/api/v1/state", st); r.StatusCode != http.StatusNoContent {
		t.Fatalf("State: %d", r.StatusCode)
	}

	var cfg struct {
		HTTP struct {
			Routers map[string]struct {
				Service     string
				Middlewares []string
			}
			Services map[string]struct {
				LoadBalancer struct{ Servers []struct{ URL string } }
			}
			Middlewares map[string]struct {
				Headers struct{ CustomRequestHeaders map[string]string }
			}
		}
	}
	if err := json.Unmarshal([]byte(providerConfig(t, s)), &cfg); err != nil {
		t.Fatal(err)
	}
	router := cfg.HTTP.Routers["kunde__n1__wp"]
	interceptor := cfg.HTTP.Services[router.Service].LoadBalancer.Servers[0].URL
	if interceptor != "http://"+s.LanderAddr() {
		t.Fatalf("Interceptor-URL %q, erwartet den gebundenen Lander %q", interceptor, s.LanderAddr())
	}
	req, _ := http.NewRequest(http.MethodGet, interceptor+"/shop", nil)
	for _, mw := range router.Middlewares {
		for k, v := range cfg.HTTP.Middlewares[mw].Headers.CustomRequestHeaders {
			req.Header.Set(k, v)
		}
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	if resp.StatusCode != http.StatusServiceUnavailable || resp.Header.Get("Retry-After") != "60" ||
		!strings.Contains(string(body), "Gleich wieder da") {
		t.Errorf("Lander: %d %q\n%s", resp.StatusCode, resp.Header.Get("Retry-After"), body)
	}
}

func TestLander_CustomTemplatesFromDataDir(t *testing.T) {
	s, _ := startServer(t)
	dir := filepath.Join(s.cfg.DataDir, "landing")
	os.MkdirAll(dir, 0o755)
	os.WriteFile(filepath.Join(dir, "maintenance.html"), []byte("Eigene Seite {{.Tenant}}"), 0o644)
	req, _ := http.NewRequest(http.MethodGet, "http://"+s.LanderAddr()+"/", nil)
	req.Header.Set("X-Sultan-Stage", "maintenance")
	req.Header.Set("X-Sultan-Tenant", "kunde")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	if body, _ := io.ReadAll(resp.Body); string(body) != "Eigene Seite kunde" {
		t.Errorf("body = %q", body)
	}
}

func TestLander_Disabled(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.LanderAddr = "none"; c.InterceptorURL = "http://lander.intern:8081" })
	if s.LanderAddr() != "" {
		t.Errorf("Lander läuft trotz none: %q", s.LanderAddr())
	}
	if s.registry.Policy.InterceptorURL != "http://lander.intern:8081" {
		t.Errorf("explizite Interceptor-URL überschrieben: %q", s.registry.Policy.InterceptorURL)
	}
}

// Die öffentliche Adresse wird im Store gemerkt, damit `sultan token invite`
// (eigener Prozess) sie in Einladungen einbetten kann.
func TestPublicURLs(t *testing.T) {
	dir := filepath.Join(t.TempDir(), "suldata")
	discard := slog.New(slog.DiscardHandler)
	if _, err := New(ctx, Config{DataDir: dir, PublicURLs: []string{"https://ok.example:8443", "http://ingress:8443"}, Logger: discard}); err == nil {
		t.Error("http-URL akzeptiert")
	}
	s, err := New(ctx, Config{DataDir: dir, PublicURLs: []string{"https://Ingress.example:8443/", "https://100.64.0.1:8443"}, Logger: discard})
	if err != nil {
		t.Fatal(err)
	}
	want := []string{"https://ingress.example:8443", "https://100.64.0.1:8443"}
	if got, err := s.PublicURLs(ctx); err != nil || !reflect.DeepEqual(got, want) {
		t.Errorf("gespeichert: %v %v", got, err)
	}
	s.Shutdown(ctx)

	// Neustart ohne --public-url: gemerkte Adressen bleiben.
	s, err = New(ctx, Config{DataDir: dir, Logger: discard})
	if err != nil {
		t.Fatal(err)
	}
	defer s.Shutdown(ctx)
	if got, _ := s.PublicURLs(ctx); !reflect.DeepEqual(got, want) {
		t.Errorf("nach Neustart: %v", got)
	}
}

// Beim Verbinden bekommt der Agent die vollständige Adressliste (signiert).
func TestConfigCommandOnConnect(t *testing.T) {
	urls := []string{"https://ingress.example:8443", "https://100.64.0.1:8443"}
	s, _ := startServer(t, func(c *Config) { c.PublicURLs = urls })
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	_, events := agent.connectEvents(t)

	ev := nextEvent(t, events, stream.EventNodeCommand)
	if err := stream.VerifyEvent(ev, s.MasterPublicKey()); err != nil {
		t.Fatal(err)
	}
	cmd, err := stream.ParseNodeCommand(ev)
	if err != nil || cmd.Command != stream.CmdConfig || !reflect.DeepEqual(cmd.ServerURLs, urls) {
		t.Errorf("config-Befehl: %+v %v", cmd, err)
	}
}

func TestNoConfigCommandWithoutPublicURLs(t *testing.T) {
	s, _ := startServer(t)
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	_, events := agent.connectEvents(t)
	nextEvent(t, events, stream.EventLeaseRenew)
	select {
	case ev := <-events:
		if ev.Type == stream.EventNodeCommand {
			t.Errorf("config ohne Adressen gesendet: %s", ev.Payload)
		}
	case <-time.After(200 * time.Millisecond):
	}
}

// connectedAgent: enrollt, verbindet und meldet wpState(kunde.de).
func connectedAgent(t *testing.T, s *Server, node string) (*testAgent, func()) {
	t.Helper()
	if _, err := s.store.GetTenant(ctx, "kunde"); err != nil {
		createTenant(t, s, "kunde", "kunde.de")
	}
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	if r := agent.enroll(t, node); r.StatusCode != http.StatusCreated {
		t.Fatalf("Enrollment: %d", r.StatusCode)
	}
	stop := agent.connect(t)
	eventually(t, "Stream", func() bool { return s.hub.Connected(node) })
	if r := agent.do(t, http.MethodPost, "/api/v1/state", wpState("kunde.de")); r.StatusCode != http.StatusNoContent {
		t.Fatalf("State: %d", r.StatusCode)
	}
	return agent, stop
}

func TestMaintenance_AgentEndpoint(t *testing.T) {
	s, logs := startServer(t)
	agent, _ := connectedAgent(t, s, "n1")

	resp := agent.do(t, http.MethodPost, "/api/v1/maintenance", MaintenanceRequest{
		Enabled: true, Duration: "2h", Reason: "Kernel 6.12", PublicReason: "Sicherheitsupdate",
	})
	if resp.StatusCode != http.StatusNoContent {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("Wartung an: %d %s", resp.StatusCode, b)
	}
	n, _ := s.store.GetNode(ctx, "n1")
	if m := n.Maintenance; m == nil || m.Reason != "Kernel 6.12" || m.PublicReason != "Sicherheitsupdate" ||
		time.Until(m.Until) < 119*time.Minute || time.Until(m.Until) > 121*time.Minute {
		t.Errorf("gespeichert: %+v", m)
	}
	// Sofort wirksam, nicht erst beim nächsten Refresh.
	if cfg := providerConfig(t, s); !strings.Contains(cfg, policy.InterceptorService) {
		t.Errorf("Wartung nicht sofort wirksam: %s", cfg)
	}
	if !strings.Contains(logs.String(), "Kernel 6.12") {
		t.Error("interner Grund nicht protokolliert")
	}

	if r := agent.do(t, http.MethodPost, "/api/v1/maintenance", MaintenanceRequest{Enabled: false}); r.StatusCode != http.StatusNoContent {
		t.Fatalf("Wartung aus: %d", r.StatusCode)
	}
	if n, _ := s.store.GetNode(ctx, "n1"); n.Maintenance != nil {
		t.Error("Wartung nicht beendet")
	}
	if cfg := providerConfig(t, s); strings.Contains(cfg, policy.InterceptorService) {
		t.Errorf("nach Wartungsende noch umgeleitet: %s", cfg)
	}
}

func TestMaintenance_DefaultAndUnlimited(t *testing.T) {
	s, _ := startServer(t)
	agent, _ := connectedAgent(t, s, "n1")

	agent.do(t, http.MethodPost, "/api/v1/maintenance", MaintenanceRequest{Enabled: true})
	n, _ := s.store.GetNode(ctx, "n1")
	if d := time.Until(n.Maintenance.Until); d < 239*time.Minute || d > 241*time.Minute {
		t.Errorf("Default-Dauer %v, erwartet 4h", d)
	}
	agent.do(t, http.MethodPost, "/api/v1/maintenance", MaintenanceRequest{Enabled: true, Duration: "0"})
	if n, _ := s.store.GetNode(ctx, "n1"); n.Maintenance == nil || !n.Maintenance.Until.IsZero() {
		t.Errorf("Dauer 0 = unbefristet: %+v", n.Maintenance)
	}
}

func TestMaintenance_InvalidRequests(t *testing.T) {
	s, _ := startServer(t)
	agent, _ := connectedAgent(t, s, "n1")
	for name, req := range map[string]any{
		"Dauer unlesbar":   MaintenanceRequest{Enabled: true, Duration: "bald"},
		"Dauer negativ":    MaintenanceRequest{Enabled: true, Duration: "-1h"},
		"Grund zu lang":    MaintenanceRequest{Enabled: true, PublicReason: strings.Repeat("x", 301)},
		"Grund mehrzeilig": MaintenanceRequest{Enabled: true, PublicReason: "Zeile 1\nZeile 2"},
		"kein JSON-Objekt": "an",
	} {
		if r := agent.do(t, http.MethodPost, "/api/v1/maintenance", req); r.StatusCode != http.StatusBadRequest {
			t.Errorf("%s: Status %d", name, r.StatusCode)
		}
	}
	plain, _ := http.NewRequest(http.MethodPost, agent.base+"/api/v1/maintenance", strings.NewReader(`{"enabled":true}`))
	if resp, err := agent.hc.Do(plain); err != nil || resp.StatusCode != http.StatusUnauthorized {
		t.Errorf("unsigniert: %v %v", resp, err)
	}
}

// Reboot während der Wartung: Routen bleiben als Wartungsseite stehen, bis
// die Wartung endet – erst dann verschwinden sie.
func TestMaintenance_DisconnectKeepsRoutesUntilEnd(t *testing.T) {
	// Kurze Lease, damit nach Wartungsende nicht die Lease die Routen hält.
	s, _ := startServer(t, func(c *Config) { c.RefreshInterval = 20 * time.Millisecond; c.LeaseDuration = 150 * time.Millisecond })
	agent, stop := connectedAgent(t, s, "n1")
	agent.do(t, http.MethodPost, "/api/v1/maintenance", MaintenanceRequest{Enabled: true, Duration: "400ms"})

	stop()
	eventually(t, "Knoten getrennt", func() bool { return !s.hub.Connected("n1") })
	time.Sleep(100 * time.Millisecond) // einige Refresh-Runden
	if cfg := providerConfig(t, s); !strings.Contains(cfg, policy.InterceptorService) {
		t.Fatalf("Routen während der Wartung entfernt: %s", cfg)
	}
	eventually(t, "Routen nach Wartungsende entfernt", func() bool { return providerConfig(t, s) == "{}" })
}

// Per CLI (anderer Prozess, direkt im Store) gesetzte Wartung wirkt beim nächsten Refresh.
func TestMaintenance_FromStoreAppliedOnRefresh(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.RefreshInterval = 20 * time.Millisecond })
	connectedAgent(t, s, "n1")
	if err := s.store.SetNodeMaintenance(ctx, "n1", store.Maintenance{Until: time.Now().Add(time.Hour)}); err != nil {
		t.Fatal(err)
	}
	eventually(t, "Umleitung", func() bool { return strings.Contains(providerConfig(t, s), policy.InterceptorService) })
}

func TestDirectory_Maintenance(t *testing.T) {
	s, _ := startServer(t)
	createTenant(t, s, "kunde", "kunde.de")
	s.store.CreateNode(ctx, store.Node{ID: "n1", TenantID: "kunde", Hostname: "h", PublicKey: "k", IPAddress: "1.2.3.4"})
	until := time.Now().Add(time.Hour).UTC().Truncate(time.Second)
	s.store.SetNodeMaintenance(ctx, "n1", store.Maintenance{Until: until, Reason: "intern", PublicReason: "öffentlich"})
	nc, err := (&directory{store: s.store}).NodeContext(ctx, "n1")
	if err != nil {
		t.Fatal(err)
	}
	if nc.Maintenance == nil || !nc.Maintenance.Until.Equal(until) || nc.Maintenance.PublicReason != "öffentlich" {
		t.Errorf("Maintenance: %+v", nc.Maintenance)
	}
}

func TestLease_IssuedOnConnectAndSigned(t *testing.T) {
	s, _ := startServer(t)
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	_, events := agent.connectEvents(t)

	ev := nextEvent(t, events, stream.EventLeaseRenew)
	if err := stream.VerifyEvent(ev, s.MasterPublicKey()); err != nil {
		t.Fatalf("Event nicht vom Server signiert: %v", err)
	}
	l, err := license.ParseLease(ev.Payload, s.MasterPublicKey())
	if err != nil {
		t.Fatalf("Lease ungültig: %v", err)
	}
	if l.NodeID != "n1" || l.TenantID != "kunde" {
		t.Errorf("Lease: %+v", l)
	}
	if d := time.Until(l.ValidUntil); d < 7*24*time.Hour-time.Minute || d > 7*24*time.Hour {
		t.Errorf("Pro-Lease läuft %v, erwartet 7 Tage", d)
	}
	stored, err := s.store.GetLease(ctx, "n1")
	if err != nil || !stored.ValidUntil.Equal(l.ValidUntil) {
		t.Errorf("gespeicherte Lease: %+v %v", stored, err)
	}
}

func TestLease_Renewal(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.RefreshInterval = 20 * time.Millisecond; c.LeaseDuration = 300 * time.Millisecond })
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	_, events := agent.connectEvents(t)

	first, _ := license.ParseLease(nextEvent(t, events, stream.EventLeaseRenew).Payload, s.MasterPublicKey())
	second, _ := license.ParseLease(nextEvent(t, events, stream.EventLeaseRenew).Payload, s.MasterPublicKey())
	if !second.ValidUntil.After(first.ValidUntil) {
		t.Errorf("nicht verlängert: %v → %v", first.ValidUntil, second.ValidUntil)
	}
}

// Ungeplantes Trennen: Wartungsseite bis zum Ablauf der Lease, danach weg.
func TestDisconnect_RoutesUntilLeaseExpiry(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.RefreshInterval = 20 * time.Millisecond; c.LeaseDuration = 600 * time.Millisecond })
	_, stop := connectedAgent(t, s, "n1")
	stop()
	eventually(t, "Wartungsseite", func() bool { return strings.Contains(providerConfig(t, s), policy.InterceptorService) })
	eventually(t, "Routen nach Lease-Ablauf entfernt", func() bool { return providerConfig(t, s) == "{}" })
}

// drain entfernt die Routen eines getrennten Knotens sofort – trotz gültiger Lease.
func TestDisconnect_DrainRemovesImmediately(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.RefreshInterval = 20 * time.Millisecond })
	_, stop := connectedAgent(t, s, "n1")
	stop()
	eventually(t, "Wartungsseite", func() bool { return strings.Contains(providerConfig(t, s), policy.InterceptorService) })
	if err := s.store.SetNodeStatus(ctx, "n1", store.NodeDrained); err != nil {
		t.Fatal(err)
	}
	eventually(t, "drain entfernt Routen", func() bool { return providerConfig(t, s) == "{}" })
}

// Befehle aus der Warteschlange (die CLI ist ein eigener Prozess) werden
// zugestellt – auch solche, die vor dem Verbinden eingereiht wurden.
func TestNodeCommands_Delivered(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.RefreshInterval = 20 * time.Millisecond })
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.enroll(t, "n1")
	s.store.EnqueueNodeCommand(ctx, "n1", stream.CmdResync, "")

	_, events := agent.connectEvents(t)
	ev := nextEvent(t, events, stream.EventNodeCommand)
	if err := stream.VerifyEvent(ev, s.MasterPublicKey()); err != nil {
		t.Fatalf("Befehl nicht signiert: %v", err)
	}
	if cmd, err := stream.ParseNodeCommand(ev); err != nil || cmd.Command != stream.CmdResync {
		t.Errorf("Befehl: %+v %v", cmd, err)
	}

	s.store.EnqueueNodeCommand(ctx, "n1", stream.CmdDrain, "Umzug")
	ev = nextEvent(t, events, stream.EventNodeCommand)
	if cmd, _ := stream.ParseNodeCommand(ev); cmd.Command != stream.CmdDrain || cmd.Reason != "Umzug" {
		t.Errorf("zweiter Befehl: %+v", cmd)
	}
	eventually(t, "als gesendet markiert", func() bool {
		cmds, _ := s.store.PendingNodeCommands(ctx, "n1")
		return len(cmds) == 0
	})
}

func TestRotateKey(t *testing.T) {
	s, _ := startServer(t)
	agent, _ := connectedAgent(t, s, "n1")
	other, _ := connectedAgent(t, s, "n2")
	newKey, _ := auth.GenerateKey()
	newPub := auth.EncodePublicKey(newKey.Public().(ed25519.PublicKey))
	req := RotateKeyRequest{PublicKey: newPub, Proof: auth.RotationProof("n1", newKey)}

	if r := agent.do(t, http.MethodPost, "/api/v1/rotate-key", RotateKeyRequest{PublicKey: newPub, Proof: "AAAA"}); r.StatusCode != http.StatusBadRequest {
		t.Errorf("falscher Besitznachweis: %d", r.StatusCode)
	}
	if r := agent.do(t, http.MethodPost, "/api/v1/rotate-key", req); r.StatusCode != http.StatusNoContent {
		t.Fatalf("Rotation: %d", r.StatusCode)
	}
	if n, _ := s.store.GetNode(ctx, "n1"); n.PublicKey != newPub {
		t.Error("Schlüssel nicht getauscht")
	}
	if r := agent.do(t, http.MethodPost, "/api/v1/rotate-key", req); r.StatusCode != http.StatusUnauthorized {
		t.Errorf("alter Schlüssel noch gültig: %d", r.StatusCode)
	}
	oldSigner := agent.signer
	agent.signer = &auth.Signer{Key: newKey}
	if r := agent.do(t, http.MethodPost, "/api/v1/rotate-key", req); r.StatusCode != http.StatusNoContent {
		t.Errorf("Wiederholung mit neuem Schlüssel (Wiederaufnahme nach Absturz): %d", r.StatusCode)
	}
	agent.signer = oldSigner

	// Der Besitznachweis ist an die Knoten-ID gebunden: Ein Nachweis für n2
	// gilt nicht für n1.
	_ = other
	another, _ := auth.GenerateKey()
	anotherPub := auth.EncodePublicKey(another.Public().(ed25519.PublicKey))
	agent.signer = &auth.Signer{Key: newKey}
	foreign := RotateKeyRequest{PublicKey: anotherPub, Proof: auth.RotationProof("n2", another)}
	if r := agent.do(t, http.MethodPost, "/api/v1/rotate-key", foreign); r.StatusCode != http.StatusBadRequest {
		t.Errorf("Nachweis für anderen Knoten: %d", r.StatusCode)
	}
}

func TestDirectory_FeaturesPaymentDisconnected(t *testing.T) {
	s, _ := startServer(t)
	if _, err := s.store.CreateTenant(ctx, store.Tenant{
		ID: "kunde", Name: "k", Email: "k@x.de", Tier: store.TierPro, MaxRoutes: 5,
		AllowedDomains: []string{"kunde.de"}, ValidUntil: time.Now().AddDate(1, 0, 0),
		Features: []string{"custom_middleware"},
	}); err != nil {
		t.Fatal(err)
	}
	s.store.CreateNode(ctx, store.Node{ID: "n1", TenantID: "kunde", Hostname: "h", PublicKey: "k", IPAddress: "1.2.3.4", Status: store.NodeOffline})
	s.store.SetTenantPayment(ctx, "kunde", time.Now().Add(-10*24*time.Hour))

	nc, err := (&directory{store: s.store}).NodeContext(ctx, "n1")
	if err != nil {
		t.Fatal(err)
	}
	if !nc.License.HasFeature(license.FeatureCustomMiddleware) {
		t.Error("Features nicht in der Lizenz")
	}
	if nc.Tenant.DunningLevel != 1 {
		t.Errorf("Mahnstufe %d, erwartet 1 (10 Tage überfällig)", nc.Tenant.DunningLevel)
	}
	if !nc.Disconnected {
		t.Error("getrennter Knoten nicht erkannt")
	}
}

// recorder ist ein Benachrichtigungsziel für Tests; failFirst lässt die
// ersten n Zustellungen scheitern.
type recorder struct {
	mu        sync.Mutex
	msgs      []notify.Message
	failFirst int
}

func (r *recorder) Send(_ context.Context, m notify.Message) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.failFirst > 0 {
		r.failFirst--
		return errors.New("Ziel nicht erreichbar")
	}
	r.msgs = append(r.msgs, m)
	return nil
}

func (r *recorder) matching(substr string) []notify.Message {
	r.mu.Lock()
	defer r.mu.Unlock()
	var out []notify.Message
	for _, m := range r.msgs {
		if strings.Contains(m.Title+" "+m.Body, substr) {
			out = append(out, m)
		}
	}
	return out
}

func fastJobs(rec *recorder) func(*Config) {
	return func(c *Config) {
		c.RefreshInterval = 20 * time.Millisecond
		c.Notify = rec
	}
}

func commercialTenant(t *testing.T, s *Server, id string, validFor time.Duration) {
	t.Helper()
	if _, err := s.store.CreateTenant(ctx, store.Tenant{
		ID: id, Name: id, Email: id + "@example.com", Tier: store.TierPro, MaxRoutes: 10,
		AllowedDomains: []string{"kunde.de"}, ValidUntil: time.Now().Add(validFor),
	}); err != nil {
		t.Fatal(err)
	}
}

func TestCheckmkEndpoint(t *testing.T) {
	s, _ := startServer(t)
	commercialTenant(t, s, "kunde", 20*24*time.Hour)
	connectedAgent(t, s, "n1")

	resp, err := http.Get("http://" + s.ProviderAddr() + "/metrics/checkmk")
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	if resp.StatusCode != 200 || !strings.HasPrefix(resp.Header.Get("Content-Type"), "text/plain") {
		t.Fatalf("Status %d, Content-Type %q", resp.StatusCode, resp.Header.Get("Content-Type"))
	}
	out := string(body)
	for _, want := range []string{`1 "Sultan_Tenant_kunde" days_left=19;30;14`, `0 "Sultan_Payment_kunde"`, `0 "Sultan_Node_n1"`, `0 "Sultan_Policy_Violations"`} {
		if !strings.Contains(out, want) {
			t.Errorf("fehlt %q in:\n%s", want, out)
		}
	}
}

func TestJob_ExpiresCommercialTenantOnly(t *testing.T) {
	rec := &recorder{}
	s, _ := startServer(t, fastJobs(rec))
	commercialTenant(t, s, "kunde", 150*time.Millisecond)
	if _, err := s.store.CreateTenant(ctx, store.Tenant{
		ID: "homelab", Name: "h", Email: "h@x.de", Tier: store.TierCommunity, MaxRoutes: 5,
		AllowedDomains: []string{"x.de"}, ValidUntil: time.Now().Add(150 * time.Millisecond),
	}); err != nil {
		t.Fatal(err)
	}
	eventually(t, "kommerzieller Vertrag auf expired", func() bool {
		tn, _ := s.store.GetTenant(ctx, "kunde")
		return tn.Status == store.TenantExpired
	})
	time.Sleep(100 * time.Millisecond)
	if tn, _ := s.store.GetTenant(ctx, "homelab"); tn.Status != store.TenantActive {
		t.Errorf("Community abgelaufen: %s", tn.Status)
	}
	eventually(t, "Benachrichtigung Vertragsende", func() bool { return len(rec.matching("abgelaufen")) == 1 })
	if m := rec.matching("abgelaufen")[0]; m.Severity != notify.Critical || !strings.Contains(m.Title, "kunde") {
		t.Errorf("Nachricht: %+v", m)
	}
}

func TestNotifications_ContractOnce(t *testing.T) {
	rec := &recorder{}
	s, _ := startServer(t, fastJobs(rec))
	commercialTenant(t, s, "kunde", 10*24*time.Hour) // innerhalb der 14-Tage-Schwelle
	eventually(t, "Benachrichtigung", func() bool { return len(rec.matching("kunde")) > 0 })
	time.Sleep(150 * time.Millisecond) // mehrere Job-Runden
	if got := rec.matching("kunde"); len(got) != 1 || got[0].Severity != notify.Critical {
		t.Errorf("erwartet genau eine kritische Nachricht, bekommen %+v", got)
	}
}

func TestNotifications_Dunning(t *testing.T) {
	rec := &recorder{}
	s, _ := startServer(t, fastJobs(rec))
	commercialTenant(t, s, "kunde", 365*24*time.Hour)
	s.store.SetTenantPayment(ctx, "kunde", time.Now().Add(-9*24*time.Hour))
	eventually(t, "Mahnstufe 1", func() bool { return len(rec.matching("Mahnstufe 1")) == 1 })
	if m := rec.matching("Mahnstufe 1")[0]; m.Severity != notify.Warning {
		t.Errorf("Schweregrad: %v", m.Severity)
	}
}

func TestNotifications_NodeOffline(t *testing.T) {
	rec := &recorder{}
	s, _ := startServer(t, fastJobs(rec), func(c *Config) { c.OfflineAlertAfter = 100 * time.Millisecond })
	_, stop := connectedAgent(t, s, "n1")
	_, stop2 := connectedAgent(t, s, "n2")
	agent2, _ := connectedAgent(t, s, "n3")
	agent2.do(t, http.MethodPost, "/api/v1/maintenance", MaintenanceRequest{Enabled: true})
	_ = stop2
	stop()
	eventually(t, "Offline-Alarm", func() bool { return len(rec.matching("n1")) == 1 })
	time.Sleep(150 * time.Millisecond)
	if got := rec.matching("n1"); len(got) != 1 || got[0].Severity != notify.Critical {
		t.Errorf("erwartet genau einen Alarm: %+v", got)
	}
	if len(rec.matching("n2")) != 0 || len(rec.matching("n3")) != 0 {
		t.Error("Alarm für verbundenen Knoten")
	}
}

func TestNotifications_NoAlarmDuringMaintenance(t *testing.T) {
	rec := &recorder{}
	s, _ := startServer(t, fastJobs(rec), func(c *Config) { c.OfflineAlertAfter = 50 * time.Millisecond })
	agent, stop := connectedAgent(t, s, "n1")
	agent.do(t, http.MethodPost, "/api/v1/maintenance", MaintenanceRequest{Enabled: true, Duration: "1h"})
	stop()
	time.Sleep(300 * time.Millisecond)
	if got := rec.matching("n1"); len(got) != 0 {
		t.Errorf("Alarm trotz angemeldeter Wartung: %+v", got)
	}
}

func TestNotifications_Violation(t *testing.T) {
	rec := &recorder{}
	s, _ := startServer(t, fastJobs(rec))
	agent, _ := connectedAgent(t, s, "n1")
	agent.do(t, http.MethodPost, "/api/v1/state", wpState("bank.de"))
	eventually(t, "Verstoß gemeldet", func() bool { return len(rec.matching("bank.de")) == 1 })
	if m := rec.matching("bank.de")[0]; m.Severity != notify.Warning || !strings.Contains(m.Body, "n1") {
		t.Errorf("Nachricht: %+v", m)
	}
}

// Scheitert die Zustellung, wird es beim nächsten Durchlauf erneut versucht.
func TestNotifications_RetryOnFailure(t *testing.T) {
	rec := &recorder{failFirst: 2}
	s, _ := startServer(t, fastJobs(rec))
	commercialTenant(t, s, "kunde", 10*24*time.Hour)
	eventually(t, "Zustellung nach Wiederholung", func() bool { return len(rec.matching("kunde")) == 1 })
	time.Sleep(100 * time.Millisecond)
	if n := len(rec.matching("kunde")); n != 1 {
		t.Errorf("%d Nachrichten, erwartet 1", n)
	}
}

// Massenhaftes Durchprobieren von Einladungen wird gebremst – vor der
// Signaturprüfung, pro Client-IP.
func TestEnroll_RateLimited(t *testing.T) {
	s, _ := startServer(t, func(c *Config) { c.EnrollLimit = 3 })
	createTenant(t, s, "kunde", "kunde.de")
	agent := newAgent(t, s, issueToken(t, s, "kunde"))
	agent.secret = crypto.TokenPrefix + strings.Repeat("A", 43) // falsch geraten
	for i := range 3 {
		if r := agent.enroll(t, "n1"); r.StatusCode != http.StatusForbidden {
			t.Fatalf("Versuch %d: %d", i+1, r.StatusCode)
		}
	}
	r := agent.enroll(t, "n1")
	if r.StatusCode != http.StatusTooManyRequests || r.Header.Get("Retry-After") == "" {
		t.Errorf("4. Versuch: %d, Retry-After %q", r.StatusCode, r.Header.Get("Retry-After"))
	}
	// Auch unsignierte Fluten zählen.
	req, _ := http.NewRequest(http.MethodPost, agent.base+"/api/v1/enroll", strings.NewReader(`{}`))
	if resp, err := agent.hc.Do(req); err != nil || resp.StatusCode != http.StatusTooManyRequests {
		t.Errorf("unsigniert: %v %v", resp, err)
	}
}

func TestConfigDefaults_EnrollLimit(t *testing.T) {
	if c := (Config{}).withDefaults(); c.EnrollLimit != 10 {
		t.Errorf("EnrollLimit-Default %d, erwartet 10 pro Minute", c.EnrollLimit)
	}
}

// „Wieder online“ nur nach einem Offline-Alarm und genau einmal pro Ausfall.
func TestNotifications_BackOnline(t *testing.T) {
	rec := &recorder{}
	s, _ := startServer(t, fastJobs(rec), func(c *Config) { c.OfflineAlertAfter = 100 * time.Millisecond })
	agent, stop := connectedAgent(t, s, "n1")
	stop()
	eventually(t, "Offline-Alarm", func() bool { return len(rec.matching("offline")) == 1 })

	agent.connect(t)
	eventually(t, "wieder online", func() bool { return len(rec.matching("wieder online")) == 1 })
	m := rec.matching("wieder online")[0]
	if m.Severity != notify.Info || !strings.Contains(m.Body, "n1") {
		t.Errorf("Nachricht: %+v", m)
	}
	time.Sleep(150 * time.Millisecond)
	if n := len(rec.matching("wieder online")); n != 1 {
		t.Errorf("%d Wieder-online-Nachrichten, erwartet 1", n)
	}
}

func TestNotifications_NoBackOnlineWithoutAlarm(t *testing.T) {
	rec := &recorder{}
	s, _ := startServer(t, fastJobs(rec), func(c *Config) { c.OfflineAlertAfter = time.Hour })
	agent, stop := connectedAgent(t, s, "n1")
	stop()
	eventually(t, "offline", func() bool { n, _ := s.store.GetNode(ctx, "n1"); return n.Status == store.NodeOffline })
	agent.connect(t)
	eventually(t, "online", func() bool { return s.hub.Connected("n1") })
	time.Sleep(150 * time.Millisecond)
	if got := rec.matching("wieder online"); len(got) != 0 {
		t.Errorf("Meldung ohne vorherigen Alarm: %+v", got)
	}
}

func TestConfigDefaults_EnrollLimitDisabled(t *testing.T) {
	if c := (Config{EnrollLimit: -1}).withDefaults(); c.EnrollLimit != -1 {
		t.Errorf("-1 (unbegrenzt) überschrieben: %d", c.EnrollLimit)
	}
}

func TestPublicURLs_BlankEntriesIgnored(t *testing.T) {
	dir := filepath.Join(t.TempDir(), "suldata")
	s, err := New(ctx, Config{DataDir: dir, PublicURLs: []string{"", " ", "https://a.example:8443"}, Logger: slog.New(slog.DiscardHandler)})
	if err != nil {
		t.Fatalf("leere Einträge verhindern den Start: %v", err)
	}
	defer s.Shutdown(ctx)
	if got, _ := s.PublicURLs(ctx); !reflect.DeepEqual(got, []string{"https://a.example:8443"}) {
		t.Errorf("gespeichert: %v", got)
	}
}