package eventstream

import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/gorilla/websocket"
	"tangled.org/core/notifier"
)

type memSource struct {
	mu     sync.Mutex
	events []Event
}

func (s *memSource) add(ev Event) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.events = append(s.events, ev)
}

func (s *memSource) GetEvents(cursorTid string, limit int) ([]Event, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := []Event{}
	for _, ev := range s.events {
		if ev.Tid > cursorTid {
			out = append(out, ev)
			if len(out) == limit {
				break
			}
		}
	}
	return out, nil
}

func mkEvent(i int) Event {
	return Event{
		Tid:       fmt.Sprintf("tid-%04d", i),
		Rkey:      fmt.Sprintf("rk-%04d", i),
		Nsid:      "sh.tangled.test",
		EventJson: json.RawMessage(fmt.Sprintf(`{"i":%d}`, i)),
		Created:   int64(i),
	}
}

func startServer(t *testing.T, src Backend, cfg StreamConfig) (string, *notifier.Notifier, <-chan error) {
	t.Helper()
	n := notifier.New()
	cfg.Backend = src
	cfg.Notifier = &n
	cfg.Logger = slog.New(slog.NewTextHandler(io.Discard, nil))

	errCh := make(chan error, 1)
	mux := http.NewServeMux()
	mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
		errCh <- Stream(w, r, cfg)
	})
	srv := httptest.NewServer(mux)
	t.Cleanup(srv.Close)
	wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "/events"
	return wsURL, &n, errCh
}

func dial(t *testing.T, wsURL, cursor string) *websocket.Conn {
	t.Helper()
	if cursor != "" {
		wsURL += "?cursor=" + cursor
	}
	c, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	t.Cleanup(func() { c.Close() })
	return c
}

func readN(t *testing.T, c *websocket.Conn, n int) []Event {
	t.Helper()
	c.SetReadDeadline(time.Now().Add(2 * time.Second))
	out := make([]Event, 0, n)
	for range n {
		_, msg, err := c.ReadMessage()
		if err != nil {
			t.Fatalf("read message at %d/%d: %v", len(out), n, err)
		}
		var ev Event
		if err := json.Unmarshal(msg, &ev); err != nil {
			t.Fatalf("unmarshal: %v", err)
		}
		out = append(out, ev)
	}
	return out
}

func TestStream_DrainStopsOnShortBatch(t *testing.T) {
	src := &memSource{}
	for i := range 7 {
		src.add(mkEvent(i))
	}

	wsURL, _, errCh := startServer(t, src, StreamConfig{
		BatchSize:          3,
		MaxBatchesPerDrain: 10,
	})
	c := dial(t, wsURL, "")

	got := readN(t, c, 7)
	for i, ev := range got {
		if ev.Tid != fmt.Sprintf("tid-%04d", i) {
			t.Fatalf("event %d: got tid=%q", i, ev.Tid)
		}
	}

	c.Close()
	select {
	case err := <-errCh:
		if err != nil && !isCloseErr(err) {
			t.Fatalf("server error: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("server did not exit")
	}
}

func TestStream_DrainHitsCap_ReturnsErrDrainCap(t *testing.T) {
	src := &memSource{}
	for i := range 5 {
		src.add(mkEvent(i))
	}

	wsURL, _, errCh := startServer(t, src, StreamConfig{
		BatchSize:          2,
		MaxBatchesPerDrain: 2,
	})
	c := dial(t, wsURL, "")

	got := readN(t, c, 4)
	if len(got) != 4 {
		t.Fatalf("want 4 events before cap, got %d", len(got))
	}
	if got[3].Tid != "tid-0003" {
		t.Fatalf("last delivered tid = %q, want tid-0003", got[3].Tid)
	}

	select {
	case err := <-errCh:
		if !errors.Is(err, ErrDrainCap) {
			t.Fatalf("want ErrDrainCap, got %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("server did not return cap error")
	}
}

func TestStream_CursorResume(t *testing.T) {
	src := &memSource{}
	for i := range 5 {
		src.add(mkEvent(i))
	}

	wsURL, _, errCh := startServer(t, src, StreamConfig{
		BatchSize:          10,
		MaxBatchesPerDrain: 10,
	})
	c := dial(t, wsURL, "tid-0002")

	got := readN(t, c, 2)
	if got[0].Tid != "tid-0003" || got[1].Tid != "tid-0004" {
		t.Fatalf("resume from cursor: got %q,%q want tid-0003,tid-0004", got[0].Tid, got[1].Tid)
	}

	c.Close()
	<-errCh
}

func TestStream_LiveDelivery(t *testing.T) {
	src := &memSource{}

	wsURL, n, errCh := startServer(t, src, StreamConfig{
		BatchSize:          10,
		MaxBatchesPerDrain: 10,
	})
	c := dial(t, wsURL, "")

	src.add(mkEvent(42))
	n.NotifyAll()

	got := readN(t, c, 1)
	if got[0].Tid != "tid-0042" {
		t.Fatalf("live event tid = %q, want tid-0042", got[0].Tid)
	}

	c.Close()
	<-errCh
}

func TestStream_LiveBurstExceedsBatchSize_DrainsAll(t *testing.T) {
	src := &memSource{}

	wsURL, n, errCh := startServer(t, src, StreamConfig{
		BatchSize:          5,
		MaxBatchesPerDrain: 100,
	})
	c := dial(t, wsURL, "")

	const burst = 17
	for i := range burst {
		src.add(mkEvent(i))
	}
	n.NotifyAll()

	got := readN(t, c, burst)
	if len(got) != burst {
		t.Fatalf("got %d events, want %d", len(got), burst)
	}
	for i, ev := range got {
		want := fmt.Sprintf("tid-%04d", i)
		if ev.Tid != want {
			t.Fatalf("event %d: got %q want %q", i, ev.Tid, want)
		}
	}

	c.Close()
	<-errCh
}

func isCloseErr(err error) bool {
	if err == nil {
		return false
	}
	if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
		return true
	}
	return strings.Contains(err.Error(), "use of closed network connection") ||
		strings.Contains(err.Error(), "websocket: close") ||
		strings.Contains(err.Error(), "broken pipe")
}
