package jetstream

import (
	"context"
	"errors"
	"math/rand/v2"
	"sync"
	"testing"
	"time"
)

func TestTheCursorNeverPassesAnUnfinishedEvent(t *testing.T) {
	const n = 64
	for range 200 {
		j := &JetstreamClient{}
		tickets := make([]*Ticket, n)
		for i := range tickets {
			tickets[i] = j.ticket(int64(100 + i*10))
		}
		rand.Shuffle(n, func(a, b int) { tickets[a], tickets[b] = tickets[b], tickets[a] })

		done := map[int64]bool{}
		for _, tk := range tickets {
			done[tk.us] = true
			tk.Done()
			for i := range n {
				us := int64(100 + i*10)
				if !done[us] {
					if got := j.lastSeenUs.Load(); got > us {
						t.Fatalf("cursor %d passed the unfinished event at %d", got, us)
					}
					break
				}
			}
		}
		if got, want := j.lastSeenUs.Load(), int64(100+(n-1)*10+1); got != want {
			t.Fatalf("cursor = %d once everything is done, want %d", got, want)
		}
	}
}

func TestAHeldEventKeepsTheCursorBehindIt(t *testing.T) {
	j := &JetstreamClient{wantedDids: make(Set[string])}
	var held *Ticket
	wrapped := j.withDidFilter(func(ctx context.Context, e *Event) error {
		if e.TimeUS == 100 {
			held = Hold(ctx)
		}
		return nil
	})

	_ = wrapped(context.Background(), &Event{Did: "did:plc:a", TimeUS: 100})
	_ = wrapped(context.Background(), &Event{Did: "did:plc:a", TimeUS: 200})
	if got := j.lastSeenUs.Load(); got != 0 {
		t.Fatalf("cursor = %d with an earlier event still held, want it untouched", got)
	}

	held.Done()
	if got := j.lastSeenUs.Load(); got != 201 {
		t.Fatalf("cursor = %d after the held event finished, want 201", got)
	}
}

func TestHoldOutsideTheClientIsANoOp(t *testing.T) {
	if tk := Hold(context.Background()); tk != nil {
		t.Fatalf("Hold = %+v with no event in the context", tk)
	}
	var none *Ticket
	none.Done()
}

func TestTicketsFinishingAtOnceStillLandOnTheNewestEvent(t *testing.T) {
	const n = 500
	j := &JetstreamClient{}
	tickets := make([]*Ticket, n)
	for i := range tickets {
		tickets[i] = j.ticket(int64(i + 1))
	}
	var wg sync.WaitGroup
	for _, tk := range tickets {
		wg.Go(tk.Done)
	}
	wg.Wait()
	if got := j.lastSeenUs.Load(); got != n+1 {
		t.Fatalf("cursor = %d, want %d", got, n+1)
	}
	if len(j.marks.pending) != 0 {
		t.Fatalf("%d events still pending", len(j.marks.pending))
	}
}

func TestShutdownWaitsForHeldEventsBeforeTheLastSave(t *testing.T) {
	db := &fakeCursorDB{savedErr: errors.New("no row")}
	j, cancel := startClient(t, refusedEndpoint, db, false)
	ticket := j.ticket(7)
	ticket.holding.Store(true)
	j.holding.Add(1)

	cancel()
	select {
	case <-j.Done():
		t.Fatal("the client finished while a processor still held an event")
	case <-time.After(200 * time.Millisecond):
	}

	ticket.Drop()
	requireStopsWithSavedCursor(t, j, db, "the client never finished after the held event was dropped")
}

// replays come out of time order, and the cursor tracks the max, not the last one
func TestTheCursorLandsOnTheNewestEventReleasedTogether(t *testing.T) {
	j := &JetstreamClient{}
	a, b, replayed := j.ticket(100), j.ticket(200), j.ticket(100)
	b.Done()
	replayed.Done()
	a.Done()
	if got := j.lastSeenUs.Load(); got != 201 {
		t.Fatalf("cursor = %d, want 201 once the event at 200 is done too", got)
	}
}