package spindle

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/jetstream"
)

type ticketLog struct {
	mu      sync.Mutex
	done    []string
	dropped []string
}

func (l *ticketLog) snapshot() (done, dropped []string) {
	l.mu.Lock()
	defer l.mu.Unlock()
	return slices.Clone(l.done), slices.Clone(l.dropped)
}

type namedTicket struct {
	name string
	log  *ticketLog
}

func (t namedTicket) Done() {
	t.log.mu.Lock()
	defer t.log.mu.Unlock()
	t.log.done = append(t.log.done, t.name)
}

func (t namedTicket) Drop() {
	t.log.mu.Lock()
	defer t.log.mu.Unlock()
	t.log.dropped = append(t.log.dropped, t.name)
}

func (sh *ingestShards) wait() { sh.wg.Wait() }

func dispatchAll(t *testing.T, ctx context.Context, sh *ingestShards, log *ticketLog, did string, rkeys ...string) {
	t.Helper()
	for _, rkey := range rkeys {
		if err := sh.dispatch(ctx, shardEvent(did, rkey), namedTicket{rkey, log}); err != nil {
			t.Fatal(err)
		}
	}
}

func ticketDone(log *ticketLog, name string) bool {
	done, _ := log.snapshot()
	return slices.Contains(done, name)
}

func shardEvent(did, rkey string) *jetstream.Event {
	return &jetstream.Event{Did: did, Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
		Operation: jetstream.CommitOperationCreate, Collection: org_tangled.GraphCollaboratorNSID, RKey: rkey,
	}}
}

func otherShardKey(key string) string {
	for i := 0; ; i++ {
		if other := fmt.Sprintf("did:plc:other%d", i); ingestShardOf(other) != ingestShardOf(key) {
			return other
		}
	}
}

type appliedLog struct {
	mu   sync.Mutex
	seen []string
}

func (a *appliedLog) add(s string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.seen = append(a.seen, s)
}

func (a *appliedLog) has(s string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return slices.Contains(a.seen, s)
}

func (a *appliedLog) list() []string {
	a.mu.Lock()
	defer a.mu.Unlock()
	return slices.Clone(a.seen)
}

func TestShardsKeepARepoInOrderWhileOtherReposGoOn(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	var applied appliedLog
	gate := make(chan struct{})
	sh := newIngestShards(ctx, func(_ context.Context, e *jetstream.Event) error {
		if e.Commit.RKey == "x1" {
			<-gate
		}
		applied.add(e.Commit.RKey)
		return nil
	})
	t.Cleanup(func() { cancel(); sh.wait() })

	x := "did:plc:x"
	y := otherShardKey(x)
	var log ticketLog
	dispatchAll(t, ctx, sh, &log, x, "x1", "x2")
	dispatchAll(t, ctx, sh, &log, y, "y1", "y2")

	eventually(t, 2*time.Second, "the other repo to apply while x1 is stuck", func() bool { return applied.has("y2") })
	time.Sleep(50 * time.Millisecond)
	if applied.has("x1") || applied.has("x2") {
		t.Fatalf("x ran ahead of its stuck first event, applied = %v", applied.list())
	}
	if ticketDone(&log, "x1") || ticketDone(&log, "x2") {
		t.Fatal("a ticket finished before its event applied")
	}

	close(gate)
	eventually(t, 2*time.Second, "x to finish", func() bool { return applied.has("x2") })
	var xs []string
	for _, rkey := range applied.list() {
		if rkey[0] == 'x' {
			xs = append(xs, rkey)
		}
	}
	if !slices.Equal(xs, []string{"x1", "x2"}) {
		t.Fatalf("x applied in the order %v, want x1 then x2", xs)
	}
	eventually(t, 2*time.Second, "every ticket to finish", func() bool {
		done, _ := log.snapshot()
		return len(done) == 4
	})
}

func TestShardsDropEveryTicketTheyDidNotFinishAtShutdown(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	started := make(chan struct{})
	sh := newIngestShards(ctx, func(ctx context.Context, e *jetstream.Event) error {
		if e.Commit.RKey == "x1" {
			close(started)
			<-ctx.Done()
			return ctx.Err()
		}
		return nil
	})

	x := "did:plc:x"
	var log ticketLog
	dispatchAll(t, ctx, sh, &log, x, "x1", "x2", "x3")
	<-started
	cancel()
	sh.wait()

	done, dropped := log.snapshot()
	slices.Sort(dropped)
	if len(done) != 0 || !slices.Equal(dropped, []string{"x1", "x2", "x3"}) {
		t.Fatalf("done = %v, dropped = %v, want all three dropped and none finished", done, dropped)
	}

	err := sh.dispatch(context.Background(), shardEvent(x, "x4"), namedTicket{"x4", &log})
	if err == nil {
		t.Fatal("a stopped shard took a new event")
	}
	if _, dropped := log.snapshot(); !slices.Contains(dropped, "x4") {
		t.Fatalf("dropped = %v, the refused event's ticket was left held", dropped)
	}
}

func TestADeclarationQueuesBehindTheRepoItNames(t *testing.T) {
	repo := syntax.DID("did:plc:repo")
	record, _ := json.Marshal(org_tangled.RepoDeclaration{Repo: repo.String()})
	declaration := &jetstream.Event{Did: "did:plc:owner", Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
		Operation: jetstream.CommitOperationCreate, Collection: org_tangled.RepoDeclarationNSID, RKey: "repo", Record: record,
	}}
	manifest := &jetstream.Event{Did: repo.String(), Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
		Operation: jetstream.CommitOperationUpdate, Collection: org_tangled.RepoManifestNSID, RKey: "self",
	}}
	if ingestShardOf(ingestShardKey(declaration)) != ingestShardOf(ingestShardKey(manifest)) {
		t.Fatal("a declaration and the manifest of the repo it names are on different shards")
	}

	deleted := &jetstream.Event{Did: "did:plc:owner", Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
		Operation: jetstream.CommitOperationDelete, Collection: org_tangled.RepoDeclarationNSID, RKey: "repo",
	}}
	if got := ingestShardKey(deleted); got != "did:plc:owner" {
		t.Fatalf("a deleted declaration has no body, key = %q, want its owner", got)
	}
}

func TestAPanickingHandlerSkipsItsEventAndTheShardGoesOn(t *testing.T) {
	ctx := t.Context()
	var applied appliedLog
	sh := newIngestShards(ctx, func(_ context.Context, e *jetstream.Event) error {
		if e.Commit.RKey == "poison" {
			var m map[string]int
			m["boom"] = 1
		}
		applied.add(e.Commit.RKey)
		return nil
	})

	var log ticketLog
	dispatchAll(t, ctx, sh, &log, "did:plc:x", "poison", "after")
	eventually(t, 5*time.Second, "the event behind the poisoned one to apply", func() bool { return applied.has("after") })

	done, dropped := log.snapshot()
	if !ticketDone(&log, "poison") || len(dropped) != 0 {
		t.Fatalf("done = %v, dropped = %v, want the poisoned event finished so the cursor can pass it", done, dropped)
	}
}

// an unfinished ticket stalls shutdown for jetstream's 30s grace
func TestNoTicketIsLeftHeldWhenShutdownRacesABlockedDispatch(t *testing.T) {
	for round := range 100 {
		ctx, cancel := context.WithCancel(context.Background())
		block := make(chan struct{})
		sh := newIngestShards(ctx, func(ctx context.Context, e *jetstream.Event) error {
			select {
			case <-block:
			case <-ctx.Done():
			}
			return nil
		})

		var log ticketLog
		var senders sync.WaitGroup
		const events = ingestShardQueue + 40
		for i := range events {
			name := fmt.Sprintf("r%d-e%d", round, i)
			senders.Go(func() { _ = sh.dispatch(ctx, shardEvent("did:plc:x", name), namedTicket{name, &log}) })
		}
		time.Sleep(time.Millisecond)
		cancel()
		senders.Wait()
		sh.wait()
		close(block)

		done, dropped := log.snapshot()
		if got := len(done) + len(dropped); got != events {
			t.Fatalf("round %d: %d of %d tickets were Done or Dropped, the rest stay held", round, got, events)
		}
	}
}

func TestAnEventThatFailsOnTheDbIsTriedAgainBeforeItIsGivenUp(t *testing.T) {
	defer func(old time.Duration) { ingestRetryDelay = old }(ingestRetryDelay)
	ingestRetryDelay = time.Millisecond
	ctx := t.Context()
	var calls atomic.Int32
	sh := newIngestShards(ctx, func(context.Context, *jetstream.Event) error {
		if calls.Add(1) < 3 {
			return errors.New("database is locked")
		}
		return nil
	})

	var log ticketLog
	dispatchAll(t, ctx, sh, &log, "did:plc:x", "busy")
	eventually(t, 5*time.Second, "the event to be applied after the db recovered", func() bool { return ticketDone(&log, "busy") })
	if got := calls.Load(); got != 3 {
		t.Fatalf("apply ran %d times, want 3 (two failures and the one that worked)", got)
	}
}

func TestAnEventThatKeepsFailingLeavesTheCursorBehindItAndTheShardGoesOn(t *testing.T) {
	defer func(old time.Duration) { ingestRetryDelay = old }(ingestRetryDelay)
	ingestRetryDelay = time.Millisecond
	ctx := t.Context()
	var applied appliedLog
	sh := newIngestShards(ctx, func(_ context.Context, e *jetstream.Event) error {
		if e.Commit.RKey == "broken" {
			return errors.New("disk I/O error")
		}
		applied.add(e.Commit.RKey)
		return nil
	})

	var log ticketLog
	dispatchAll(t, ctx, sh, &log, "did:plc:x", "broken", "after")
	eventually(t, 5*time.Second, "the event behind the broken one to apply", func() bool { return applied.has("after") })

	done, dropped := log.snapshot()
	if !slices.Equal(dropped, []string{"broken"}) || ticketDone(&log, "broken") {
		t.Fatalf("done = %v, dropped = %v, want the failed event dropped so the cursor can't pass it", done, dropped)
	}
}