package stream

import (
	"context"
	"errors"
	"net/http"
	"net/http/httptest"
	"reflect"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

func TestBackoff(t *testing.T) {
	const minD, maxD = time.Second, 60 * time.Second
	tests := map[int]time.Duration{
		0:   time.Second,
		1:   2 * time.Second,
		5:   32 * time.Second,
		6:   60 * time.Second,
		100: 60 * time.Second,
	}
	for attempt, want := range tests {
		if got := backoff(attempt, minD, maxD); got != want {
			t.Errorf("backoff(%d) = %v, erwartet %v", attempt, got, want)
		}
	}
	if got := backoff(3, 5*time.Second, 5*time.Second); got != 5*time.Second {
		t.Errorf("min == max: %v", got)
	}
}

func TestClient_Defaults(t *testing.T) {
	var c Client
	if got := c.idleTimeout(); got != 90*time.Second {
		t.Errorf("IdleTimeout-Default = %v, erwartet 90s (2 × Keepalive)", got)
	}
	if got := c.minBackoff(); got != time.Second {
		t.Errorf("MinBackoff-Default = %v", got)
	}
	if got := c.maxBackoff(); got != time.Minute {
		t.Errorf("MaxBackoff-Default = %v", got)
	}
	if c.httpClient() != http.DefaultClient {
		t.Error("HTTPClient-Default sollte http.DefaultClient sein")
	}
}

// Ende-zu-Ende: Client verbindet sich über Run mit einem Hub, Prepare setzt
// Authentifizierung, ein vom Hub gesendetes Event kommt beim Handler an.
func TestClient_ReceivesEvents(t *testing.T) {
	h := &Hub{}
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Header.Get("Authorization") != "Bearer geheim" {
			http.Error(w, "nein", http.StatusUnauthorized)
			return
		}
		if r.Header.Get("Accept") != "text/event-stream" {
			http.Error(w, "falscher Accept", http.StatusNotAcceptable)
			return
		}
		h.Serve(w, r, "node-1")
	}))
	t.Cleanup(srv.Close)
	t.Cleanup(h.Close)

	c := &Client{
		URL:     srv.URL,
		Prepare: func(r *http.Request) error { r.Header.Set("Authorization", "Bearer geheim"); return nil },
	}
	got := make(chan Event, 1)
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- c.Run(ctx, func(ev Event) error { got <- ev; return nil }) }()

	eventually(t, "Client verbunden", func() bool { return h.Connected("node-1") })
	if err := h.Send("node-1", Event{Type: EventNodeCommand}); err != nil {
		t.Fatal(err)
	}
	select {
	case ev := <-got:
		if ev.Type != EventNodeCommand || ev.NodeID != "node-1" {
			t.Errorf("unerwartetes Event: %+v", ev)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("kein Event empfangen")
	}
	cancel()
	if err := waitErr(t, done); !errors.Is(err, context.Canceled) {
		t.Errorf("Run: erwartet context.Canceled, bekommen %v", err)
	}
}

func TestClient_UnexpectedResponse(t *testing.T) {
	tests := map[string]http.HandlerFunc{
		"Status 401": func(w http.ResponseWriter, r *http.Request) {
			http.Error(w, "nein", http.StatusUnauthorized)
		},
		"falscher Content-Type": func(w http.ResponseWriter, r *http.Request) {
			w.Header().Set("Content-Type", "text/plain")
			w.Write([]byte("hallo"))
		},
	}
	for name, handler := range tests {
		t.Run(name, func(t *testing.T) {
			srv := httptest.NewServer(handler)
			defer srv.Close()
			c := &Client{URL: srv.URL}
			err := c.Connect(context.Background(), func(Event) error { return nil })
			if !errors.Is(err, ErrUnexpectedResponse) {
				t.Errorf("erwartet ErrUnexpectedResponse, bekommen %v", err)
			}
		})
	}
}

// sseHandler sendet nur die Header und hält die Verbindung ohne Daten offen.
func silentSSE(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "text/event-stream")
	w.WriteHeader(http.StatusOK)
	w.(http.Flusher).Flush()
	<-r.Context().Done()
}

func TestClient_IdleTimeout(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(silentSSE))
	defer srv.Close()
	c := &Client{URL: srv.URL, IdleTimeout: 50 * time.Millisecond}

	start := time.Now()
	err := c.Connect(context.Background(), func(Event) error { return nil })
	if !errors.Is(err, ErrIdleTimeout) {
		t.Fatalf("erwartet ErrIdleTimeout, bekommen %v", err)
	}
	if d := time.Since(start); d > time.Second {
		t.Errorf("Timeout griff erst nach %v", d)
	}
}

func TestClient_KeepAlivePreventsIdleTimeout(t *testing.T) {
	h := &Hub{KeepAlive: 10 * time.Millisecond}
	srv, _ := hubServer(t, h)
	c := &Client{URL: srv.URL + "?node=n", IdleTimeout: 60 * time.Millisecond}

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- c.Connect(ctx, func(Event) error { return nil }) }()

	select {
	case err := <-done:
		t.Fatalf("Verbindung trotz Keepalives beendet: %v", err)
	case <-time.After(300 * time.Millisecond):
	}
	cancel()
	if err := waitErr(t, done); !errors.Is(err, context.Canceled) {
		t.Errorf("erwartet context.Canceled, bekommen %v", err)
	}
}

func TestClient_ServerClosesStream(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
		w.Write([]byte("data: " + evJSON + "\n\n"))
	}))
	defer srv.Close()

	var got []Event
	err := (&Client{URL: srv.URL}).Connect(context.Background(), func(ev Event) error {
		got = append(got, ev)
		return nil
	})
	if !errors.Is(err, ErrStreamClosed) {
		t.Errorf("erwartet ErrStreamClosed, bekommen %v", err)
	}
	if !reflect.DeepEqual(got, []Event{evWant}) {
		t.Errorf("Events: %+v", got)
	}
}

func TestClient_PrepareError(t *testing.T) {
	errSign := errors.New("Signatur fehlgeschlagen")
	c := &Client{URL: "http://127.0.0.1:1", Prepare: func(*http.Request) error { return errSign }}
	if err := c.Connect(context.Background(), nil); !errors.Is(err, errSign) {
		t.Errorf("erwartet Prepare-Fehler, bekommen %v", err)
	}
}

func TestClient_ConnectionRefused(t *testing.T) {
	srv := httptest.NewServer(http.NotFoundHandler())
	url := srv.URL
	srv.Close()
	if err := (&Client{URL: url}).Connect(context.Background(), nil); err == nil {
		t.Error("erwartet Verbindungsfehler")
	}
}

func TestClient_HandlerErrorStopsRun(t *testing.T) {
	var requests atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		requests.Add(1)
		w.Header().Set("Content-Type", "text/event-stream")
		w.Write([]byte("data: " + evJSON + "\n\n"))
	}))
	defer srv.Close()

	errStop := errors.New("Handler bricht ab")
	err := (&Client{URL: srv.URL}).Run(context.Background(), func(Event) error { return errStop })
	if !errors.Is(err, errStop) {
		t.Errorf("erwartet Handler-Fehler, bekommen %v", err)
	}
	if n := requests.Load(); n != 1 {
		t.Errorf("%d Verbindungsversuche, erwartet 1 (kein Reconnect nach Handler-Fehler)", n)
	}
}

// Ablauf: 503, 503, erfolgreicher Stream (endet), danach nur noch 503.
// Erwartet: exponentielles Backoff, Reset nach erfolgreicher Verbindung,
// Deckelung bei MaxBackoff.
func TestClient_RunRetriesWithBackoff(t *testing.T) {
	var requests atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if requests.Add(1) == 3 {
			w.Header().Set("Content-Type", "text/event-stream")
			w.Write([]byte("data: " + evJSON + "\n\n"))
			return
		}
		http.Error(w, "später", http.StatusServiceUnavailable)
	}))
	defer srv.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	var mu sync.Mutex
	var delays []time.Duration
	var reported int
	var events atomic.Int32
	c := &Client{
		URL:        srv.URL,
		MinBackoff: 10 * time.Millisecond,
		MaxBackoff: 40 * time.Millisecond,
		OnError: func(err error, retryIn time.Duration) {
			mu.Lock()
			defer mu.Unlock()
			if err == nil {
				t.Error("OnError ohne Fehler aufgerufen")
			}
			reported++
		},
		sleep: func(ctx context.Context, d time.Duration) error {
			mu.Lock()
			defer mu.Unlock()
			delays = append(delays, d)
			if len(delays) == 6 {
				cancel()
				return ctx.Err()
			}
			return nil
		},
	}
	err := c.Run(ctx, func(Event) error { events.Add(1); return nil })
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("erwartet context.Canceled, bekommen %v", err)
	}
	ms := time.Millisecond
	want := []time.Duration{10 * ms, 20 * ms, 10 * ms, 20 * ms, 40 * ms, 40 * ms}
	mu.Lock()
	defer mu.Unlock()
	if !reflect.DeepEqual(delays, want) {
		t.Errorf("Backoff-Folge = %v, erwartet %v", delays, want)
	}
	if reported != 6 {
		t.Errorf("OnError %d-mal aufgerufen, erwartet 6", reported)
	}
	if events.Load() != 1 {
		t.Errorf("%d Events empfangen, erwartet 1", events.Load())
	}
}

func TestClient_ConnectReturnsHandlerError(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		w.Write([]byte("data: " + evJSON + "\n\n"))
	}))
	defer srv.Close()

	errStop := errors.New("stopp")
	err := (&Client{URL: srv.URL}).Connect(context.Background(), func(Event) error { return errStop })
	if err != errStop {
		t.Errorf("erwartet den unverpackten Handler-Fehler, bekommen %v", err)
	}
}

func TestClient_CustomHTTPClient(t *testing.T) {
	var used atomic.Bool
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
	}))
	defer srv.Close()
	hc := srv.Client()
	base := hc.Transport
	hc.Transport = roundTripFunc(func(r *http.Request) (*http.Response, error) {
		used.Store(true)
		return base.RoundTrip(r)
	})

	err := (&Client{URL: srv.URL, HTTPClient: hc}).Connect(context.Background(), nil)
	if !errors.Is(err, ErrStreamClosed) {
		t.Errorf("erwartet ErrStreamClosed, bekommen %v", err)
	}
	if !used.Load() {
		t.Error("eigener HTTPClient wurde nicht verwendet")
	}
}

type roundTripFunc func(*http.Request) (*http.Response, error)

func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }

func TestSleepCtx(t *testing.T) {
	if err := sleepCtx(context.Background(), time.Millisecond); err != nil {
		t.Errorf("abgelaufener Timer: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	start := time.Now()
	if err := sleepCtx(ctx, time.Hour); !errors.Is(err, context.Canceled) {
		t.Errorf("erwartet context.Canceled, bekommen %v", err)
	}
	if time.Since(start) > time.Second {
		t.Error("sleepCtx hat trotz Abbruch gewartet")
	}
}

// OnConnect wird nach jedem erfolgreichen Verbindungsaufbau aufgerufen –
// nicht bei Fehlantworten.
func TestClient_OnConnect(t *testing.T) {
	var requests atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if requests.Add(1)%2 == 1 {
			http.Error(w, "später", http.StatusServiceUnavailable)
			return
		}
		w.Header().Set("Content-Type", "text/event-stream")
	}))
	defer srv.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	var connects, sleeps atomic.Int32
	c := &Client{
		URL:       srv.URL,
		OnConnect: func() { connects.Add(1) },
		sleep: func(ctx context.Context, d time.Duration) error {
			if sleeps.Add(1) == 4 {
				cancel()
				return ctx.Err()
			}
			return nil
		},
	}
	c.Run(ctx, func(Event) error { return nil })
	if n := connects.Load(); n != 2 {
		t.Errorf("OnConnect %d-mal, erwartet 2 (bei 4 Versuchen, davon 2 erfolgreich)", n)
	}
}