package spindle

import (
	"context"
	"fmt"
	"net/http"
	"slices"
	"testing"
	"time"

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

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/spindle/db"
)

func TestARepoJobOnAHangingKnotStopsAtItsBudgetAndBacksTheRepoOff(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)
	if err := env.s.markRepo(testRepoDid); err != nil {
		t.Fatal(err)
	}
	defer func(old time.Duration) { repoJobBudget = old }(repoJobBudget)
	repoJobBudget = 100 * time.Millisecond

	hang := make(chan struct{})
	t.Cleanup(func() { close(hang) })
	env.knot.beforeAnswer = func(*http.Request) { <-hang }

	finished := make(chan struct{})
	go func() {
		env.s.runRepoJob(context.Background(), testRepoDid)
		close(finished)
	}()
	select {
	case <-finished:
	case <-time.After(5 * time.Second):
		t.Fatal("a repo job on a hanging knot kept its worker far past the budget")
	}

	if attempts, marked, err := env.s.db.RepoRetryRow(testRepoDid); err != nil || !marked || attempts != 1 {
		t.Fatalf("attempts = %d, exists %v, %v, want the repo backed off after running out of time", attempts, marked, err)
	}
	if _, due, err := env.s.db.RepoRetryDue(testRepoDid, env.s.retries.clock()); err != nil || due {
		t.Fatalf("due = %v, %v, want the repo to wait out a backoff before the next job", due, err)
	}
}

func TestARepoJobWorksThroughALimitedNumberOfMarksThenYieldsItsWorker(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)
	const marks = recordMarksPerJob + 50
	for i := range marks {
		r := db.RecordRetry{RepoDid: testRepoDid, Collection: org_tangled.TrackTicketNSID, Rkey: fmt.Sprintf("3kticket%05d", i)}
		if err := env.s.db.MarkRecordDirty(r, time.Now(), recordMarksPerRepo); err != nil {
			t.Fatal(err)
		}
	}

	env.s.runRepoJob(context.Background(), testRepoDid)

	if left := dueRecords(t, env); len(left) != 50 {
		t.Fatalf("%d record marks left, want the %d past the per-job limit", len(left), 50)
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if did, ok := env.s.retries.jobs.next(ctx); !ok || did != testRepoDid {
		t.Fatalf("next job = %q, %v, want the repo queued again for the rest of its marks", did, ok)
	}
}

func fairJobs(pool int, priority ...string) (*repoJobs, map[syntax.DID]string) {
	knots := map[syntax.DID]string{}
	set := map[string]bool{}
	for _, knot := range priority {
		set[knot] = true
	}
	j := &repoJobs{}
	j.configure(pool, func(did syntax.DID) string { return knots[did] }, set)
	return j, knots
}

func kickRepos(j *repoJobs, knots map[syntax.DID]string, knot string, n int) []syntax.DID {
	var dids []syntax.DID
	for i := range n {
		did := syntax.DID(fmt.Sprintf("did:plc:%s%03d", knot, i))
		knots[did] = knot
		j.kick(did)
		dids = append(dids, did)
	}
	return dids
}

func within(d time.Duration, next func(context.Context) (syntax.DID, bool)) (syntax.DID, bool) {
	ctx, cancel := context.WithTimeout(context.Background(), d)
	defer cancel()
	return next(ctx)
}

func startAll(j *repoJobs, knots map[syntax.DID]string) (started []syntax.DID, perKnot map[string]int) {
	perKnot = map[string]int{}
	for {
		did, ok := within(100*time.Millisecond, j.next)
		if !ok {
			return started, perKnot
		}
		started = append(started, did)
		perKnot[knots[did]]++
	}
}

func TestAnIdlePoolKeepsAReserveAndUnservedReposShareAHalf(t *testing.T) {
	for _, tc := range []struct {
		name        string
		pool, repos int
		knot        string
		want        int
	}{
		{"an eighth of a large pool", 64, 80, "a", 56},
		{"unserved repos get half the pool", 8, 10, "", 4},
	} {
		t.Run(tc.name, func(t *testing.T) {
			j, knots := fairJobs(tc.pool)
			kickRepos(j, knots, tc.knot, tc.repos)
			if started, _ := startAll(j, knots); len(started) != tc.want {
				t.Fatalf("%d jobs started on a pool of %d, want %d", len(started), tc.pool, tc.want)
			}
		})
	}
}

func TestOneKnotOnAnIdlePoolLeavesTheReserveForANewcomer(t *testing.T) {
	j, knots := fairJobs(16)
	kickRepos(j, knots, "a", 30)
	if started, _ := startAll(j, knots); len(started) != 14 {
		t.Fatalf("%d jobs started on a pool of 16 with only one knot waiting, want all but the reserve of 2", len(started))
	}

	newcomer := kickRepos(j, knots, "b", 1)[0]
	if did, ok := within(time.Second, j.next); !ok || did != newcomer {
		t.Fatalf("next = %q, %v, want a new knot to get a worker at once while another is busy", did, ok)
	}
}

func TestKnotsWithJobsWaitingEachGetAnEqualShare(t *testing.T) {
	j, knots := fairJobs(8)
	kickRepos(j, knots, "a", 8)
	kickRepos(j, knots, "b", 8)
	_, perKnot := startAll(j, knots)
	if perKnot["a"] != 4 || perKnot["b"] != 4 {
		t.Fatalf("running per knot = %v, want 4 each while both have jobs waiting", perKnot)
	}
}

func TestASlowKnotWithManyReposCantHoldMoreThanItsShareWhileAnotherKnotWaits(t *testing.T) {
	j, knots := fairJobs(8)
	kickRepos(j, knots, "hostile", 30)
	healthy := kickRepos(j, knots, "healthy", 1)[0]

	var firstFive []syntax.DID
	var hostile int
	for range 5 {
		did, ok := within(time.Second, j.next)
		if !ok {
			t.Fatalf("only %d jobs started, want the healthy knot's repo to get a worker", len(firstFive))
		}
		firstFive = append(firstFive, did)
		if knots[did] == "hostile" {
			hostile++
		}
	}
	if hostile != 4 || !slices.Contains(firstFive, healthy) {
		t.Fatalf("hostile knot started %d of the first 5 (%v), want 4 and the healthy repo among them", hostile, firstFive)
	}
}

func TestAFinishedJobGivesItsSlotToAWaitingOne(t *testing.T) {
	j, knots := fairJobs(8)
	kickRepos(j, knots, "a", 5)
	kickRepos(j, knots, "b", 5)
	started, perKnot := startAll(j, knots)
	if perKnot["a"] != 4 {
		t.Fatalf("running per knot = %v, want knot a at its share of 4", perKnot)
	}

	j.finish(started[0])
	if _, ok := within(time.Second, j.next); !ok {
		t.Fatal("a finished job didn't free a slot for a waiting one")
	}
}

func TestTheRepoJobWorkerCountComesFromTheConfig(t *testing.T) {
	env := newEnsureEnv(t)
	if got := env.s.repoJobWorkerCount(); got != defaultRepoJobWorkers {
		t.Fatalf("workers = %d with nothing configured, want the default %d", got, defaultRepoJobWorkers)
	}
	env.s.cfg.Server.RepoJobWorkers = 7
	if got := env.s.repoJobWorkerCount(); got != 7 {
		t.Fatalf("workers = %d, want the configured 7", got)
	}
}

func TestAPriorityKnotsJobRunsAtOnceWhileAHostileKnotHoldsTheSharedPool(t *testing.T) {
	j, knots := fairJobs(8, "priority.example.com")
	kickRepos(j, knots, "hostile", 30)
	if started, _ := startAll(j, knots); len(started) != 6 {
		t.Fatalf("%d hostile jobs started, want them held to the shared pool less its reserve", len(started))
	}

	vip := kickRepos(j, knots, "priority.example.com", 1)[0]
	if did, ok := within(time.Second, j.nextPriority); !ok || did != vip {
		t.Fatalf("priority worker got %q, %v, want the priority knot's job at once", did, ok)
	}
}

func TestANonPriorityKnotCantUseThePriorityWorkers(t *testing.T) {
	j, knots := fairJobs(8, "priority.example.com")
	kickRepos(j, knots, "other", 5)
	if did, ok := within(200*time.Millisecond, j.nextPriority); ok {
		t.Fatalf("a priority worker took %q of a knot that isn't a priority one", did)
	}
}

func TestAPriorityKnotsJobCanAlsoUseASharedWorker(t *testing.T) {
	j, knots := fairJobs(8, "priority.example.com")
	vip := kickRepos(j, knots, "priority.example.com", 1)[0]
	if did, ok := within(time.Second, j.next); !ok || did != vip {
		t.Fatalf("shared worker got %q, %v, want it to take a priority knot's job under the usual rules", did, ok)
	}
}

func TestPriorityKnotsMatchAsDidsAndAsHosts(t *testing.T) {
	env := newEnsureEnv(t)
	env.s.cfg.Server.PriorityKnots = []string{"did:web:knot1.tangled.sh", " KNOT2.example.com ", "did:web:knot3.example.com%3A8443"}
	got := env.s.priorityKnots()
	for _, want := range []string{"knot1.tangled.sh", "knot2.example.com", "knot3.example.com:8443"} {
		if !got[want] {
			t.Fatalf("priority knots = %v, want %s to match however it was written", got, want)
		}
	}
	byDid, byHost := priorityKnotHosts([]string{"did:web:knot1.tangled.sh"}), priorityKnotHosts([]string{"knot1.tangled.sh"})
	if len(byDid) != 1 || !byDid["knot1.tangled.sh"] || !byHost["knot1.tangled.sh"] {
		t.Fatalf("a knot's did gave %v and its host %v, want the same key", byDid, byHost)
	}
}

func TestPriorityWorkersOnlyExistWithPriorityKnotsAndDefaultToAQuarter(t *testing.T) {
	env := newEnsureEnv(t)
	env.s.cfg.Server.RepoJobWorkers = 64
	if got := env.s.priorityJobWorkerCount(); got != 0 {
		t.Fatalf("priority workers = %d with no priority knots, want none", got)
	}
	env.s.cfg.Server.PriorityKnots = []string{"knot1.tangled.sh"}
	if got := env.s.priorityJobWorkerCount(); got != 16 {
		t.Fatalf("priority workers = %d, want a quarter of 64", got)
	}
	env.s.cfg.Server.PriorityRepoJobWorkers = 5
	if got := env.s.priorityJobWorkerCount(); got != 5 {
		t.Fatalf("priority workers = %d, want the configured 5", got)
	}
}