package api

import (
	"bufio"
	"context"
	"encoding/json"
	"net/http"
	"runtime"
	"strings"
	"sync"
	"testing"
	"time"
)

// sseEvent is one parsed SSE frame: the event name and the JSON payload.
type sseEvent struct {
	name string
	data string
}

// openSSE opens GET /logs/stream and returns a reader for the frames. The
// stream stays open until the context the caller passed is cancelled or the
// body is closed, so every test here has to do one or the other.
func openSSE(t *testing.T, h *harness, ctx context.Context) (*http.Response, *bufio.Reader) {
	t.Helper()
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, h.ts.URL+"/api/logs/stream", nil)
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("X-API-Key", "rw-key")
	res, err := h.ts.Client().Do(req)
	if err != nil {
		t.Fatalf("open the log stream: %v", err)
	}
	t.Cleanup(func() { _ = res.Body.Close() })
	return res, bufio.NewReader(res.Body)
}

// nextEvent reads until a complete frame has arrived. Comment frames (the
// keep-alive ": ping") and the blank separator line are skipped.
func nextEvent(t *testing.T, br *bufio.Reader) sseEvent {
	t.Helper()
	var ev sseEvent
	for {
		line, err := br.ReadString('\n')
		if err != nil {
			t.Fatalf("read the log stream: %v", err)
		}
		line = strings.TrimRight(line, "\r\n")
		switch {
		case strings.HasPrefix(line, "event: "):
			ev.name = strings.TrimPrefix(line, "event: ")
		case strings.HasPrefix(line, "data: "):
			ev.data = strings.TrimPrefix(line, "data: ")
		case line == "":
			if ev.name != "" || ev.data != "" {
				return ev
			}
		}
	}
}

// freshUnsubSignal gives the SSE test its own one-shot unsubscribe channel.
// The harness closes a single shared channel through a sync.Once, so without
// this only the first stream test in the package could observe it.
func freshUnsubSignal(h *harness) <-chan struct{} {
	h.fake.mu.Lock()
	defer h.fake.mu.Unlock()
	h.fake.unsub = make(chan struct{})
	h.fake.unsubOnce = sync.Once{}
	return h.fake.unsub
}

// TestSSEStreamHeadersAndEvents pins the SSE contract the live log view
// depends on: the streaming content type, the two headers that stop an
// intermediary from buffering the stream, a "ready" event so the UI can show it
// is connected, and exactly one "entry" event per query-log row.
func TestSSEStreamHeadersAndEvents(t *testing.T) {
	h := newHarness(t, nil)
	unsub := freshUnsubSignal(h)
	// Queued before the stream opens: the fake's channel is buffered, so this
	// is a deterministic ordering and not a race on the subscription.
	want := []LogEntry{
		{ID: 1, Name: "a.example.com.", QTypeStr: "A", RCode: 0, Client: "198.51.100.1"},
		{ID: 2, Name: "b.example.com.", QTypeStr: "AAAA", RCode: 3, Client: "198.51.100.2", Blocked: true},
		{ID: 3, Name: "c.example.com.", QTypeStr: "TXT", RCode: 0, Client: "198.51.100.3"},
	}
	for _, e := range want {
		h.fake.stream <- e
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	res, br := openSSE(t, h, ctx)

	if got := res.Header.Get("Content-Type"); got != "text/event-stream" {
		t.Errorf("Content-Type = %q, want text/event-stream", got)
	}
	// no-store, not no-cache: this stream is never replayable.
	if got := res.Header.Get("Cache-Control"); got != "no-store" {
		t.Errorf("Cache-Control = %q, want no-store", got)
	}
	// Without this an nginx in front buffers the stream and the live view shows
	// nothing until the buffer fills.
	if got := res.Header.Get("X-Accel-Buffering"); got != "no" {
		t.Errorf("X-Accel-Buffering = %q, want no", got)
	}

	ready := nextEvent(t, br)
	if ready.name != "ready" {
		t.Fatalf("first event = %q, want ready", ready.name)
	}
	var readyBody map[string]any
	if err := json.Unmarshal([]byte(ready.data), &readyBody); err != nil {
		t.Fatalf("ready payload is not JSON: %v: %q", err, ready.data)
	}
	if readyBody["version"] != "test" {
		t.Errorf("ready payload = %v, want it to carry the version", readyBody)
	}
	for i, e := range want {
		got := nextEvent(t, br)
		if got.name != "entry" {
			t.Fatalf("event %d = %q, want entry", i, got.name)
		}
		var gotEntry LogEntry
		if err := json.Unmarshal([]byte(got.data), &gotEntry); err != nil {
			t.Fatalf("entry payload is not JSON: %v: %q", err, got.data)
		}
		if gotEntry.ID != e.ID || gotEntry.Name != e.Name || gotEntry.Client != e.Client {
			t.Errorf("entry %d = %+v, want %+v", i, gotEntry, e)
		}
	}
	cancel()
	waitFor(t, 2*time.Second, "the subscriber to be released", func() bool {
		return h.fake.subscribers() == 0
	})
	select {
	case <-unsub:
	default:
		t.Error("the handler returned without running the unsubscribe hook")
	}
}

// TestSSEStreamUnsubscribesOnClientDisconnect checks a client that goes away
// does not leave a handler parked on the log channel: the request context is
// cancelled when the connection drops, and the handler must return and run its
// cancel hook, or every reconnect leaks a goroutine and a log subscriber.
func TestSSEStreamUnsubscribesOnClientDisconnect(t *testing.T) {
	h := newHarness(t, nil)
	unsub := freshUnsubSignal(h)
	ctx, cancel := context.WithCancel(context.Background())
	_, br := openSSE(t, h, ctx)
	if got := nextEvent(t, br); got.name != "ready" {
		t.Fatalf("first event = %q, want ready", got.name)
	}
	waitFor(t, 2*time.Second, "the handler to subscribe", func() bool {
		return h.fake.subscribers() == 1
	})

	// Disconnect: dropping the request context closes the connection, which is
	// what the server sees when a browser tab is closed.
	cancel()

	waitFor(t, 2*time.Second, "the handler to unsubscribe", func() bool {
		return h.fake.subscribers() == 0
	})
	select {
	case <-unsub:
	case <-time.After(2 * time.Second):
		t.Fatal("the unsubscribe hook never ran after the client disconnected")
	}
}

// TestSSEStreamLeavesNoGoroutinesBehind opens and closes the stream repeatedly
// and asserts the goroutine count stays bounded. The bound, not an exact count:
// the HTTP client and server keep per-connection goroutines that wind down on
// their own, and a test that demands an exact number is a flake. The property
// worth having is that the count does not grow with the number of streams.
func TestSSEStreamLeavesNoGoroutinesBehind(t *testing.T) {
	h := newHarness(t, nil)
	runtime.GC() // keep the baseline honest: the count is what it is after a GC
	before := runtime.NumGoroutine()
	for i := 0; i < 25; i++ {
		ctx, cancel := context.WithCancel(context.Background())
		_, br := openSSE(t, h, ctx)
		if got := nextEvent(t, br); got.name != "ready" {
			cancel()
			t.Fatalf("stream %d: first event = %q, want ready", i, got.name)
		}
		cancel()
	}
	waitFor(t, 2*time.Second, "every subscriber to be released", func() bool {
		return h.fake.subscribers() == 0
	})
	const bound = 8 // per-connection goroutines that may still be winding down
	waitFor(t, 3*time.Second, "the goroutine count to settle", func() bool {
		return runtime.NumGoroutine() <= before+bound
	})
}

// TestSSEStreamEndsWhenTheSourceCloses covers the other exit from the handler
// loop: the hook's channel being closed, which is what a shutting-down log ring
// does. The response has to end rather than hang, and the unsubscribe hook must
// still run — a handler that only returns on the request context is a handler
// that leaks when the producer goes away.
func TestSSEStreamEndsWhenTheSourceCloses(t *testing.T) {
	h := newHarness(t, nil)
	unsub := freshUnsubSignal(h)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	// Closing the hook's channel is what a shutting-down log ring does.
	close(h.fake.stream)
	_, br := openSSE(t, h, ctx)
	if got := nextEvent(t, br); got.name != "ready" {
		t.Fatalf("first event = %q, want ready", got.name)
	}
	if _, err := br.ReadString('\n'); err == nil {
		t.Error("the stream carried on after its source closed")
	}
	waitFor(t, 2*time.Second, "the handler to unsubscribe", func() bool {
		return h.fake.subscribers() == 0
	})
	select {
	case <-unsub:
	case <-time.After(2 * time.Second):
		t.Fatal("the unsubscribe hook never ran after the source closed")
	}
}

// TestSSEStreamRequiresAuth checks the stream is behind the same auth as the
// rest of the API: it is a live view of every client address and domain the
// instance sees, so an unauthenticated subscribe is a data leak, not a
// cosmetic one.
func TestSSEStreamRequiresAuth(t *testing.T) {
	h := newHarness(t, nil)
	res := h.do(http.MethodGet, "/api/logs/stream", "")
	if res.status != http.StatusUnauthorized {
		t.Fatalf("GET /api/logs/stream without a key = %d, want 401: %s", res.status, res.body)
	}
	if h.fake.count("SubscribeLog") != 0 {
		t.Error("an unauthenticated request reached the subscribe hook")
	}
}