package stream

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// eventually wartet, bis cond wahr ist, höchstens zwei Sekunden.
func eventually(t *testing.T, msg string, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for !cond() {
		if time.Now().After(deadline) {
			t.Fatalf("Bedingung nicht erfüllt: %s", msg)
		}
		time.Sleep(2 * time.Millisecond)
	}
}

// hubServer startet einen Testserver, der Verbindungen an h übergibt. Die
// Knoten-ID kommt aus ?node=. Rückgabewerte von Serve landen in errs.
func hubServer(t *testing.T, h *Hub) (srv *httptest.Server, errs chan error) {
	t.Helper()
	errs = make(chan error, 16)
	srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		errs <- h.Serve(w, r, r.URL.Query().Get("node"))
	}))
	t.Cleanup(srv.Close)
	t.Cleanup(h.Close)
	return srv, errs
}

func openStream(t *testing.T, ctx context.Context, srv *httptest.Server, node string) *http.Response {
	t.Helper()
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"?node="+node, nil)
	if err != nil {
		t.Fatal(err)
	}
	resp, err := srv.Client().Do(req)
	if err != nil {
		t.Fatalf("Verbindung fehlgeschlagen: %v", err)
	}
	t.Cleanup(func() { resp.Body.Close() })
	return resp
}

func waitErr(t *testing.T, errs <-chan error) error {
	t.Helper()
	select {
	case err := <-errs:
		return err
	case <-time.After(2 * time.Second):
		t.Fatal("Serve ist nicht zurückgekehrt")
		return nil
	}
}

func TestHub_SendOffline(t *testing.T) {
	var h Hub
	if err := h.Send("node-1", Event{Type: EventNodeCommand}); !errors.Is(err, ErrNodeOffline) {
		t.Errorf("erwartet ErrNodeOffline, bekommen %v", err)
	}
}

func TestHub_Defaults(t *testing.T) {
	var h Hub
	if got := h.keepAlive(); got != 45*time.Second {
		t.Errorf("KeepAlive-Default = %v, erwartet 45s (Spec)", got)
	}
	if got := h.writeTimeout(); got != 10*time.Second {
		t.Errorf("WriteTimeout-Default = %v", got)
	}
	if got := h.bufferSize(); got != 16 {
		t.Errorf("BufferSize-Default = %d", got)
	}
}

func TestHub_DeliversEvent(t *testing.T) {
	h := &Hub{}
	srv, _ := hubServer(t, h)
	resp := openStream(t, context.Background(), srv, "node-1")

	if ct := resp.Header.Get("Content-Type"); ct != "text/event-stream" {
		t.Errorf("Content-Type = %q", ct)
	}
	if cc := resp.Header.Get("Cache-Control"); cc != "no-cache" {
		t.Errorf("Cache-Control = %q", cc)
	}
	eventually(t, "node-1 verbunden", func() bool { return h.Connected("node-1") })

	if err := h.Send("node-1", Event{Type: EventNodeCommand, Payload: json.RawMessage(`{"action":"drain"}`)}); err != nil {
		t.Fatalf("Send: %v", err)
	}
	ev, err := NewDecoder(resp.Body).Next()
	if err != nil {
		t.Fatalf("Next: %v", err)
	}
	if ev.Type != EventNodeCommand || ev.NodeID != "node-1" || string(ev.Payload) != `{"action":"drain"}` {
		t.Errorf("unerwartetes Event: %+v", ev)
	}
	if !strings.HasPrefix(ev.ID, "evt_") || ev.Timestamp == 0 {
		t.Errorf("ID/Timestamp nicht ergänzt: %+v", ev)
	}
}

func TestHub_SendKeepsIDAndTimestamp(t *testing.T) {
	h := &Hub{}
	srv, _ := hubServer(t, h)
	resp := openStream(t, context.Background(), srv, "n")
	eventually(t, "verbunden", func() bool { return h.Connected("n") })

	want := Event{ID: "evt_fix", Timestamp: 42, Type: EventLeaseRenew, NodeID: "n"}
	if err := h.Send("n", Event{ID: "evt_fix", Timestamp: 42, Type: EventLeaseRenew, NodeID: "anderer"}); err != nil {
		t.Fatal(err)
	}
	got, err := NewDecoder(resp.Body).Next()
	if err != nil {
		t.Fatal(err)
	}
	if !reflect.DeepEqual(got, want) {
		t.Errorf("got %+v, erwartet %+v (NodeID wird immer vom Hub gesetzt)", got, want)
	}
}

func TestHub_SendInvalid(t *testing.T) {
	h := &Hub{}
	srv, _ := hubServer(t, h)
	openStream(t, context.Background(), srv, "n")
	eventually(t, "verbunden", func() bool { return h.Connected("n") })

	if err := h.Send("n", Event{}); !errors.Is(err, ErrInvalidEvent) {
		t.Errorf("leerer Typ: erwartet ErrInvalidEvent, bekommen %v", err)
	}
	if err := h.Send("n", Event{Type: EventError, Payload: json.RawMessage(`{`)}); !errors.Is(err, ErrInvalidEvent) {
		t.Errorf("kaputte Payload: erwartet ErrInvalidEvent, bekommen %v", err)
	}
}

func TestHub_KeepAlive(t *testing.T) {
	h := &Hub{KeepAlive: 10 * time.Millisecond}
	srv, _ := hubServer(t, h)
	resp := openStream(t, context.Background(), srv, "n")

	buf := make([]byte, 6)
	if _, err := io.ReadFull(resp.Body, buf); err != nil {
		t.Fatalf("lesen: %v", err)
	}
	if string(buf) != ":\n\n:\n\n" {
		t.Errorf("erwartet zwei Keepalives \":\\n\\n\", bekommen %q", buf)
	}
}

func TestHub_DisconnectDetected(t *testing.T) {
	var mu sync.Mutex
	var connects, disconnects []string
	h := &Hub{
		OnConnect:    func(n string) { mu.Lock(); connects = append(connects, n); mu.Unlock() },
		OnDisconnect: func(n string) { mu.Lock(); disconnects = append(disconnects, n); mu.Unlock() },
	}
	srv, errs := hubServer(t, h)
	ctx, cancel := context.WithCancel(context.Background())
	openStream(t, ctx, srv, "node-1")
	eventually(t, "verbunden", func() bool { return h.Connected("node-1") })

	cancel()
	if err := waitErr(t, errs); !errors.Is(err, context.Canceled) {
		t.Errorf("Serve: erwartet context.Canceled, bekommen %v", err)
	}
	if h.Connected("node-1") {
		t.Error("node-1 noch verbunden")
	}
	if err := h.Send("node-1", Event{Type: EventError}); !errors.Is(err, ErrNodeOffline) {
		t.Errorf("erwartet ErrNodeOffline, bekommen %v", err)
	}
	mu.Lock()
	defer mu.Unlock()
	if !reflect.DeepEqual(connects, []string{"node-1"}) || !reflect.DeepEqual(disconnects, []string{"node-1"}) {
		t.Errorf("Callbacks: connects=%v disconnects=%v", connects, disconnects)
	}
}

func TestHub_ReconnectReplacesOld(t *testing.T) {
	var connects, disconnects atomic.Int32
	h := &Hub{
		OnConnect:    func(string) { connects.Add(1) },
		OnDisconnect: func(string) { disconnects.Add(1) },
	}
	srv, errs := hubServer(t, h)
	old := openStream(t, context.Background(), srv, "n")
	eventually(t, "erste Verbindung", func() bool { return connects.Load() == 1 })
	fresh := openStream(t, context.Background(), srv, "n")

	if err := waitErr(t, errs); !errors.Is(err, ErrReplaced) {
		t.Errorf("alte Verbindung: erwartet ErrReplaced, bekommen %v", err)
	}
	if _, err := io.ReadAll(old.Body); err != nil {
		t.Errorf("alter Stream sollte sauber enden: %v", err)
	}
	if err := h.Send("n", Event{Type: EventLeaseRenew}); err != nil {
		t.Fatal(err)
	}
	ev, err := NewDecoder(fresh.Body).Next()
	if err != nil || ev.Type != EventLeaseRenew {
		t.Errorf("neue Verbindung: ev=%+v err=%v", ev, err)
	}
	if c, d := connects.Load(), disconnects.Load(); c != 2 || d != 0 {
		t.Errorf("Callbacks: connects=%d disconnects=%d, erwartet 2/0 (Ersatz ist kein Offline)", c, d)
	}
}

func TestHub_Nodes(t *testing.T) {
	h := &Hub{}
	srv, _ := hubServer(t, h)
	openStream(t, context.Background(), srv, "b")
	openStream(t, context.Background(), srv, "a")
	eventually(t, "beide verbunden", func() bool { return len(h.Nodes()) == 2 })
	if got := h.Nodes(); !reflect.DeepEqual(got, []string{"a", "b"}) {
		t.Errorf("Nodes = %v, erwartet sortiert [a b]", got)
	}
}

func TestHub_Close(t *testing.T) {
	h := &Hub{}
	srv, errs := hubServer(t, h)
	resp := openStream(t, context.Background(), srv, "n")
	eventually(t, "verbunden", func() bool { return h.Connected("n") })

	h.Close()
	if err := waitErr(t, errs); !errors.Is(err, ErrHubClosed) {
		t.Errorf("Serve: erwartet ErrHubClosed, bekommen %v", err)
	}
	if _, err := io.ReadAll(resp.Body); err != nil {
		t.Errorf("Stream sollte sauber enden: %v", err)
	}
	if err := h.Send("n", Event{Type: EventError}); !errors.Is(err, ErrHubClosed) {
		t.Errorf("Send nach Close: erwartet ErrHubClosed, bekommen %v", err)
	}
	req := httptest.NewRequest(http.MethodGet, "/", nil)
	if err := h.Serve(httptest.NewRecorder(), req, "neu"); !errors.Is(err, ErrHubClosed) {
		t.Errorf("Serve nach Close: erwartet ErrHubClosed, bekommen %v", err)
	}
	h.Close() // zweimal schließen darf nicht paniken
}

// blockingWriter simuliert einen Client, der nicht mehr liest.
type blockingWriter struct {
	header  http.Header
	release chan struct{}
}

func (w *blockingWriter) Header() http.Header         { return w.header }
func (w *blockingWriter) WriteHeader(int)             {}
func (w *blockingWriter) Flush()                      {}
func (w *blockingWriter) Write(p []byte) (int, error) { <-w.release; return len(p), nil }

func TestHub_BufferFull(t *testing.T) {
	h := &Hub{BufferSize: 2}
	w := &blockingWriter{header: http.Header{}, release: make(chan struct{})}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- h.Serve(w, httptest.NewRequest(http.MethodGet, "/", nil).WithContext(ctx), "n") }()
	eventually(t, "verbunden", func() bool { return h.Connected("n") })

	var full bool
	for range 4 {
		if err := h.Send("n", Event{Type: EventError}); errors.Is(err, ErrBufferFull) {
			full = true
		}
	}
	if !full {
		t.Error("erwartet ErrBufferFull bei vollem Puffer")
	}
	cancel()
	close(w.release)
	if err := waitErr(t, done); !errors.Is(err, context.Canceled) {
		t.Errorf("Serve: erwartet context.Canceled, bekommen %v", err)
	}
}

type failingWriter struct{ header http.Header }

var errWrite = errors.New("Verbindung weg")

func (w *failingWriter) Header() http.Header       { return w.header }
func (w *failingWriter) WriteHeader(int)           {}
func (w *failingWriter) Flush()                    {}
func (w *failingWriter) Write([]byte) (int, error) { return 0, errWrite }

func TestHub_WriteErrorEndsServe(t *testing.T) {
	h := &Hub{KeepAlive: 5 * time.Millisecond}
	err := h.Serve(&failingWriter{header: http.Header{}}, httptest.NewRequest(http.MethodGet, "/", nil), "n")
	if !errors.Is(err, errWrite) {
		t.Errorf("erwartet Schreibfehler, bekommen %v", err)
	}
	if h.Connected("n") {
		t.Error("Knoten nach Schreibfehler noch verbunden")
	}
}

type noFlushWriter struct{ header http.Header }

func (w *noFlushWriter) Header() http.Header         { return w.header }
func (w *noFlushWriter) WriteHeader(int)             {}
func (w *noFlushWriter) Write(p []byte) (int, error) { return len(p), nil }

func TestHub_StreamingUnsupported(t *testing.T) {
	var h Hub
	err := h.Serve(&noFlushWriter{header: http.Header{}}, httptest.NewRequest(http.MethodGet, "/", nil), "n")
	if !errors.Is(err, ErrStreamingUnsupported) {
		t.Errorf("erwartet ErrStreamingUnsupported, bekommen %v", err)
	}
	if h.Connected("n") {
		t.Error("Knoten darf nicht registriert sein")
	}
}

func TestHub_ConcurrentSend(t *testing.T) {
	const senders, perSender = 20, 50
	h := &Hub{}
	srv, _ := hubServer(t, h)
	resp := openStream(t, context.Background(), srv, "n")
	eventually(t, "verbunden", func() bool { return h.Connected("n") })

	var wg sync.WaitGroup
	for range senders {
		wg.Go(func() {
			for range perSender {
				for {
					err := h.Send("n", Event{Type: EventServiceState})
					if err == nil {
						break
					}
					if !errors.Is(err, ErrBufferFull) {
						t.Errorf("Send: %v", err)
						return
					}
					time.Sleep(time.Millisecond)
				}
			}
		})
	}
	dec := NewDecoder(resp.Body)
	for i := range senders * perSender {
		if _, err := dec.Next(); err != nil {
			t.Fatalf("Event %d: %v", i, err)
		}
	}
	wg.Wait()
}

func TestHub_ServeRejectsEmptyNodeID(t *testing.T) {
	var connects atomic.Int32
	h := &Hub{OnConnect: func(string) { connects.Add(1) }}
	rec := httptest.NewRecorder()
	err := h.Serve(rec, httptest.NewRequest(http.MethodGet, "/", nil), "")
	if !errors.Is(err, ErrInvalidNodeID) {
		t.Errorf("erwartet ErrInvalidNodeID, bekommen %v", err)
	}
	if rec.Body.Len() != 0 || rec.Header().Get("Content-Type") == "text/event-stream" {
		t.Error("bei ungültiger ID darf kein Stream begonnen werden")
	}
	if connects.Load() != 0 || len(h.Nodes()) != 0 {
		t.Error("leere ID wurde registriert")
	}
}

func TestHub_Disconnect(t *testing.T) {
	var disconnects atomic.Int32
	h := &Hub{OnDisconnect: func(string) { disconnects.Add(1) }}
	srv, errs := hubServer(t, h)
	resp := openStream(t, context.Background(), srv, "n")
	eventually(t, "verbunden", func() bool { return h.Connected("n") })

	if err := h.Disconnect("n"); err != nil {
		t.Fatalf("Disconnect: %v", err)
	}
	// Sofort wirksam, nicht erst wenn Serve zurückkehrt:
	if h.Connected("n") {
		t.Error("Knoten nach Disconnect noch verbunden")
	}
	if err := h.Send("n", Event{Type: EventError}); !errors.Is(err, ErrNodeOffline) {
		t.Errorf("Send nach Disconnect: erwartet ErrNodeOffline, bekommen %v", err)
	}
	if err := waitErr(t, errs); !errors.Is(err, ErrDisconnected) {
		t.Errorf("Serve: erwartet ErrDisconnected, bekommen %v", err)
	}
	if _, err := io.ReadAll(resp.Body); err != nil {
		t.Errorf("Stream sollte sauber enden: %v", err)
	}
	if n := disconnects.Load(); n != 1 {
		t.Errorf("OnDisconnect %d-mal aufgerufen, erwartet 1", n)
	}
	if err := h.Disconnect("n"); !errors.Is(err, ErrNodeOffline) {
		t.Errorf("zweites Disconnect: erwartet ErrNodeOffline, bekommen %v", err)
	}
}

// Sobald der Client die Antwort-Header hat, muss der Knoten registriert sein –
// sonst scheitert ein direkt folgender Request ("Stream nicht verbunden").
func TestHub_ConnectedWhenHeadersArrive(t *testing.T) {
	for range 50 {
		h := &Hub{}
		srv, _ := hubServer(t, h)
		openStream(t, context.Background(), srv, "n")
		if !h.Connected("n") {
			t.Fatal("Header empfangen, Knoten aber noch nicht registriert")
		}
	}
}