package eventconsumer

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

	"tangled.org/core/eventconsumer/cursor"
	"tangled.org/core/eventstream"
	"tangled.org/core/notifier"
	"tangled.org/core/tid"
)

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

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

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

func mkTids(n int) []string {
	g := tid.MonotonicGenerator{}
	out := make([]string, n)
	for i := range n {
		out[i] = g.FromNanos(int64(i+1) * 1_000_000_000)
	}
	return out
}

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

func startEventServer(t *testing.T, src *memSrc) (Source, *notifier.Notifier) {
	t.Helper()
	n := notifier.New()
	mux := http.NewServeMux()
	mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
		_ = eventstream.Stream(w, r, eventstream.StreamConfig{
			Backend:            src,
			Notifier:           &n,
			Logger:             slog.New(slog.NewTextHandler(io.Discard, nil)),
			BatchSize:          5,
			MaxBatchesPerDrain: 100,
		})
	})
	srv := httptest.NewServer(mux)
	t.Cleanup(srv.Close)
	addr := strings.TrimPrefix(srv.URL, "http://")
	return Source{Kind: "test", Host: addr}, &n
}

func TestConsumer_DrainAdvancesCursor(t *testing.T) {
	tids := mkTids(8)
	src := &memSrc{}
	for i, t := range tids {
		src.add(mkEv(t, i))
	}

	source, _ := startEventServer(t, src)

	store := &cursor.MemoryStore{}
	seenMu := sync.Mutex{}
	seen := []string{}

	cfg := ConsumerConfig{
		ProcessFunc: func(ctx context.Context, _ Source, msg eventstream.Event) error {
			seenMu.Lock()
			seen = append(seen, msg.Tid)
			seenMu.Unlock()
			return nil
		},
		WorkerCount:       1,
		QueueSize:         16,
		ConnectionTimeout: 2 * time.Second,
		CursorStore:       store,
		URLFunc:           DefaultURL(true),
		Logger:            slog.New(slog.NewTextHandler(io.Discard, nil)),
	}
	c := NewConsumer(cfg)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	c.Start(ctx)
	c.AddSource(ctx, source)

	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		seenMu.Lock()
		n := len(seen)
		seenMu.Unlock()
		if n >= 8 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}

	seenMu.Lock()
	defer seenMu.Unlock()
	if len(seen) != 8 {
		t.Fatalf("processed %d events, want 8: %v", len(seen), seen)
	}
	for i, got := range seen {
		if got != tids[i] {
			t.Fatalf("event %d: got %q want %q", i, got, tids[i])
		}
	}

	finalCursor := store.Get(source.Key())
	if finalCursor != tids[7] {
		t.Fatalf("cursor = %q, want %q", finalCursor, tids[7])
	}
}

func TestConsumer_CursorMonotonic_OutOfOrderWorkers(t *testing.T) {
	tids := mkTids(4)
	src := &memSrc{}
	for i, t := range tids {
		src.add(mkEv(t, i))
	}

	source, _ := startEventServer(t, src)

	store := &cursor.MemoryStore{}

	releaseFirst := make(chan struct{})
	processed := make(chan string, 4)

	cfg := ConsumerConfig{
		ProcessFunc: func(ctx context.Context, _ Source, msg eventstream.Event) error {
			if msg.Tid == tids[0] {
				<-releaseFirst
			}
			processed <- msg.Tid
			return nil
		},
		WorkerCount:       4,
		QueueSize:         16,
		ConnectionTimeout: 2 * time.Second,
		CursorStore:       store,
		URLFunc:           DefaultURL(true),
		Logger:            slog.New(slog.NewTextHandler(io.Discard, nil)),
	}
	c := NewConsumer(cfg)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	c.Start(ctx)
	c.AddSource(ctx, source)

	for range 3 {
		select {
		case <-processed:
		case <-time.After(3 * time.Second):
			t.Fatal("timed out waiting for tids 1-3 to be processed")
		}
	}

	if cur := store.Get(source.Key()); cur != tids[3] {
		t.Fatalf("cursor before slow worker finished = %q, want %q", cur, tids[3])
	}

	close(releaseFirst)
	select {
	case <-processed:
	case <-time.After(3 * time.Second):
		t.Fatal("timed out waiting for slow worker")
	}

	if cur := store.Get(source.Key()); cur != tids[3] {
		t.Fatalf("cursor regressed after slow worker: %q, want %q", cur, tids[3])
	}
}

func TestConsumer_StopTerminatesWithoutCtxCancel(t *testing.T) {
	src := &memSrc{}
	source, _ := startEventServer(t, src)

	cfg := ConsumerConfig{
		ProcessFunc:       func(ctx context.Context, _ Source, _ eventstream.Event) error { return nil },
		WorkerCount:       2,
		QueueSize:         8,
		ConnectionTimeout: 2 * time.Second,
		CursorStore:       &cursor.MemoryStore{},
		URLFunc:           DefaultURL(true),
		Logger:            slog.New(slog.NewTextHandler(io.Discard, nil)),
	}
	c := NewConsumer(cfg)

	c.Start(context.Background())
	c.AddSource(context.Background(), source)

	done := make(chan struct{})
	go func() {
		c.Stop()
		close(done)
	}()

	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("Stop did not return within 5s")
	}
}

func TestConsumer_ResumesFromStoredCursor(t *testing.T) {
	tids := mkTids(5)
	src := &memSrc{}
	for i, t := range tids {
		src.add(mkEv(t, i))
	}

	source, _ := startEventServer(t, src)

	store := &cursor.MemoryStore{}
	store.Set(source.Key(), tids[2])

	seenMu := sync.Mutex{}
	seen := []string{}

	cfg := ConsumerConfig{
		ProcessFunc: func(ctx context.Context, _ Source, msg eventstream.Event) error {
			seenMu.Lock()
			seen = append(seen, msg.Tid)
			seenMu.Unlock()
			return nil
		},
		WorkerCount:       1,
		QueueSize:         16,
		ConnectionTimeout: 2 * time.Second,
		CursorStore:       store,
		URLFunc:           DefaultURL(true),
		Logger:            slog.New(slog.NewTextHandler(io.Discard, nil)),
	}
	c := NewConsumer(cfg)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	c.Start(ctx)
	c.AddSource(ctx, source)

	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		seenMu.Lock()
		n := len(seen)
		seenMu.Unlock()
		if n >= 2 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}

	seenMu.Lock()
	defer seenMu.Unlock()
	if len(seen) < 2 {
		t.Fatalf("processed %d events, want 2: %v", len(seen), seen)
	}
	if seen[0] != tids[3] || seen[1] != tids[4] {
		t.Fatalf("resumed events = %v, want [%q %q]", seen, tids[3], tids[4])
	}
}
