package spindle

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"slices"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"tangled.org/core/api/org_tangled"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/spindle/models"
	"tangled.org/core/workflow"
)

func scheduleTestRepo() db.Repo {
	return db.Repo{
		Knot:    "knot.test",
		Owner:   syntax.DID("did:plc:owner"),
		Rkey:    syntax.RecordKey("repo"),
		RepoDid: syntax.DID("did:plc:repo"),
	}
}

func scheduleTestSnapshot() db.ScheduledRepo {
	return db.ScheduledRepo{
		RepoDid: scheduleTestRepo().RepoDid.String(),
		Branch:  "main",
		SHA:     "sha",
	}
}

func mustCron(t *testing.T, expression, name string) scheduledWorkflow {
	t.Helper()
	repo := scheduleTestRepo()
	schedule, err := workflow.ParseCron(expression, "", repo.RepoDid.String()+"\x00build.yml")
	if err != nil {
		t.Fatalf("ParseCron(%q, UTC): %v", expression, err)
	}
	return scheduledWorkflow{
		repo:          repo,
		scheduledRepo: scheduleTestSnapshot(),
		name:          name,
		expression:    expression,
		schedule:      schedule,
	}
}

func newTestScheduler(t *testing.T) *pipelineScheduler {
	t.Helper()
	d, _ := newTestSpindleDB(t)
	return &pipelineScheduler{
		db:              d,
		l:               slog.New(slog.NewTextHandler(io.Discard, nil)),
		byRepo:          make(map[string][]*scheduledWorkflow),
		minimumInterval: 5 * time.Minute,
		concurrency:     1,
		now:             time.Now,
	}
}

// a released claim can be retaken, a completed one can't
func claimable(t *testing.T, s *pipelineScheduler, entry scheduledWorkflow, at time.Time) bool {
	t.Helper()
	ok, err := s.db.ClaimScheduleRun(context.Background(), entry.repo.RepoDid.String(), entry.name, at, s.minimumInterval)
	if err != nil {
		t.Fatal(err)
	}
	return ok
}

func TestPipelineSchedulerDueUsesUTCMinuteAndDeduplicatesWorkflows(t *testing.T) {
	scheduler := newTestScheduler(t)
	scheduler.replaceCachedSchedules(
		scheduleTestRepo().RepoDid.String(),
		[]scheduledWorkflow{mustCron(t, "0 9 * * 1-5", "build.yml"), mustCron(t, "0 9 * * *", "build.yml"), mustCron(t, "0 12 * * *", "noon.yml")},
		time.Date(2026, time.August, 10, 8, 59, 0, 0, time.UTC),
	)
	at := time.Date(2026, time.August, 10, 11, 0, 47, 0, time.FixedZone("local", 2*60*60))
	due := scheduler.due(at)
	group, ok := due[scheduleTestRepo().RepoDid.String()]
	if !ok {
		t.Fatal("expected repository to have due workflows")
	}
	var names []string
	for name := range group.workflows {
		names = append(names, name)
	}
	slices.Sort(names)
	if !slices.Equal(names, []string{"build.yml"}) {
		t.Fatalf("due workflows = %v, want [build.yml]", names)
	}
}

func TestPipelineSchedulerDispatchDueClaimsOccurrenceOnce(t *testing.T) {
	d, _ := newTestSpindleDB(t)
	entry := mustCron(t, "0 9 * * *", "build.yml")
	at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)

	var calls int
	dispatch := func(_ context.Context, repo db.Repo, _ db.ScheduledRepo, workflows []string, scheduledAt time.Time) (models.PipelineId, error) {
		calls++
		if repo.RepoDid != entry.repo.RepoDid || !slices.Equal(workflows, []string{"build.yml"}) || !scheduledAt.Equal(at) {
			t.Fatalf("unexpected dispatch: repo=%s workflows=%v scheduledAt=%s", repo.RepoDid, workflows, scheduledAt)
		}
		return models.PipelineId("pipeline"), nil
	}
	for range 2 {
		scheduler := newTestScheduler(t)
		scheduler.db = d
		scheduler.dispatch = dispatch
		scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))
		scheduler.dispatchDue(context.Background(), at)
	}
	if calls != 1 {
		t.Fatalf("dispatch calls = %d, want 1", calls)
	}
}

func TestPipelineSchedulerDispatchTimeoutReleasesClaim(t *testing.T) {
	entry := mustCron(t, "0 9 * * *", "build.yml")
	at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)

	scheduler := newTestScheduler(t)
	scheduler.dispatchTimeout = 25 * time.Millisecond
	scheduler.dispatch = func(ctx context.Context, _ db.Repo, _ db.ScheduledRepo, _ []string, _ time.Time) (models.PipelineId, error) {
		<-ctx.Done()
		return models.PipelineId(""), ctx.Err()
	}
	scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))

	start := time.Now()
	scheduler.dispatchDue(context.Background(), at)
	if elapsed := time.Since(start); elapsed > 5*time.Second {
		t.Fatalf("dispatchDue returned after %s; want the dispatch timeout to unblock it", elapsed)
	}
	if !claimable(t, scheduler, entry, at) {
		t.Fatal("timed-out dispatch left its schedule claim behind")
	}
}

func TestPipelineSchedulerBoundsConcurrentRepositoryDispatches(t *testing.T) {
	at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
	started := make(chan struct{}, 2)
	release := make(chan struct{})
	var mu sync.Mutex
	active, maxActive := 0, 0
	scheduler := newTestScheduler(t)
	scheduler.concurrency = 2
	scheduler.dispatch = func(_ context.Context, repo db.Repo, _ db.ScheduledRepo, _ []string, _ time.Time) (models.PipelineId, error) {
		mu.Lock()
		active++
		maxActive = max(maxActive, active)
		mu.Unlock()
		started <- struct{}{}
		<-release
		mu.Lock()
		active--
		mu.Unlock()
		return models.PipelineId("pipeline"), nil
	}
	for _, did := range []string{"did:plc:repoa", "did:plc:repob"} {
		entry := mustCron(t, "0 9 * * *", "build.yml")
		entry.repo.RepoDid = syntax.DID(did)
		entry.scheduledRepo.RepoDid = did
		scheduler.replaceCachedSchedules(did, []scheduledWorkflow{entry}, at.Add(-time.Minute))
	}

	done := make(chan struct{})
	go func() {
		scheduler.dispatchDue(context.Background(), at)
		close(done)
	}()
	for range 2 {
		select {
		case <-started:
		case <-time.After(time.Second):
			t.Fatal("repository dispatches did not start concurrently")
		}
	}
	mu.Lock()
	gotMaxActive := maxActive
	mu.Unlock()
	if gotMaxActive != 2 {
		t.Fatalf("maximum concurrent dispatches = %d, want 2", gotMaxActive)
	}
	close(release)
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("dispatchDue did not wait for repository dispatches")
	}
}

func TestPipelineSchedulerReleasesClaimWhenDispatchIsCanceled(t *testing.T) {
	at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
	entry := mustCron(t, "0 9 * * *", "build.yml")
	started := make(chan struct{})
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	scheduler := newTestScheduler(t)
	scheduler.dispatch = func(ctx context.Context, _ db.Repo, _ db.ScheduledRepo, _ []string, _ time.Time) (models.PipelineId, error) {
		close(started)
		<-ctx.Done()
		return models.PipelineId(""), ctx.Err()
	}
	scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))
	done := make(chan struct{})
	go func() {
		scheduler.dispatchDue(ctx, at)
		close(done)
	}()
	<-started
	cancel()
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("canceled dispatch did not finish")
	}
	if !claimable(t, scheduler, entry, at) {
		t.Fatal("canceled dispatch left its schedule claim")
	}
}

func TestPipelineSchedulerEnforcesMinimumRunIntervalAcrossSchedules(t *testing.T) {
	at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
	var entries []scheduledWorkflow
	for _, expression := range []string{"0 9 * * *", "5 9 * * *", "10 9 * * *"} {
		entries = append(entries, mustCron(t, expression, "build.yml"))
	}

	var dispatched []time.Time
	scheduler := newTestScheduler(t)
	scheduler.minimumInterval = 10 * time.Minute
	scheduler.dispatch = func(_ context.Context, _ db.Repo, _ db.ScheduledRepo, _ []string, scheduledAt time.Time) (models.PipelineId, error) {
		dispatched = append(dispatched, scheduledAt)
		return models.PipelineId("pipeline"), nil
	}
	scheduler.replaceCachedSchedules(scheduleTestRepo().RepoDid.String(), entries, at.Add(-time.Minute))
	scheduler.dispatchDue(context.Background(), at)
	scheduler.dispatchDue(context.Background(), at.Add(5*time.Minute))
	scheduler.dispatchDue(context.Background(), at.Add(10*time.Minute))

	want := []time.Time{at, at.Add(10 * time.Minute)}
	if !slices.Equal(dispatched, want) {
		t.Fatalf("scheduled dispatches = %v, want %v", dispatched, want)
	}
}

func TestPipelineSchedulerDispatchFailureClaimLifecycle(t *testing.T) {
	tests := []struct {
		name         string
		pipelineID   models.PipelineId
		wantReleased bool
	}{
		{name: "before pipeline creation", wantReleased: true},
		{name: "after pipeline creation", pipelineID: models.PipelineId("pipeline")},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			entry := mustCron(t, "0 9 * * *", "build.yml")
			at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)

			scheduler := newTestScheduler(t)
			scheduler.dispatch = func(context.Context, db.Repo, db.ScheduledRepo, []string, time.Time) (models.PipelineId, error) {
				return test.pipelineID, errors.New("temporary failure")
			}
			scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))
			scheduler.dispatchDue(context.Background(), at)

			if reclaimed := claimable(t, scheduler, entry, at); reclaimed != test.wantReleased {
				t.Fatalf("schedule claim reclaimed = %v, want %v", reclaimed, test.wantReleased)
			}
			if test.wantReleased {
				return
			}

			var completed int
			if err := scheduler.db.QueryRow(
				`select count(*) from schedule_runs where repo_did = ? and workflow = ? and scheduled_at = ? and pipeline_id = ?`,
				entry.repo.RepoDid.String(), entry.name, at.Unix(), test.pipelineID.String(),
			).Scan(&completed); err != nil {
				t.Fatal(err)
			}
			if completed != 1 {
				t.Fatal("pipeline creation error left its schedule claim incomplete")
			}
		})
	}
}

func TestPipelineSchedulerRestoresPersistedSchedulesWithoutLoadingRepository(t *testing.T) {
	ctx := context.Background()
	scheduler := newTestScheduler(t)
	repo := scheduleTestRepo()
	if err := scheduler.db.AddRepo(repo); err != nil {
		t.Fatalf("AddRepo: %v", err)
	}
	if err := scheduler.db.ReplaceWorkflowSchedules(ctx, db.ScheduledRepo{
		RepoDid: repo.RepoDid.String(),
		Branch:  "main",
		SHA:     "persisted-sha",
	}, []db.WorkflowSchedule{{
		RepoDid:    repo.RepoDid.String(),
		Workflow:   "build.yml",
		Expression: "30 5 * * 1-5",
		Timezone:   "America/New_York",
	}}, time.Now()); err != nil {
		t.Fatalf("ReplaceWorkflowSchedules: %v", err)
	}

	at := time.Date(2026, time.August, 10, 9, 30, 0, 0, time.UTC)
	scheduler.now = func() time.Time { return at.Add(-time.Minute) }
	if err := scheduler.restoreSchedules(ctx); err != nil {
		t.Fatalf("restoreSchedules: %v", err)
	}
	due := scheduler.due(at)
	group, ok := due[repo.RepoDid.String()]
	if !ok {
		t.Fatal("persisted schedule was not restored")
	}
	if group.scheduledRepo.Branch != "main" || group.scheduledRepo.SHA != "persisted-sha" {
		t.Fatalf("restored repository snapshot = %#v, want main/persisted-sha", group.scheduledRepo)
	}
}

type countingSchedule struct {
	next  time.Time
	calls int
}

func (s *countingSchedule) Next(time.Time) time.Time {
	s.calls++
	return s.next
}

func TestPipelineSchedulerDoesNotEvaluateNonDueSchedulesEachMinute(t *testing.T) {
	now := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
	entries := make([]scheduledWorkflow, 10_000)
	schedules := make([]*countingSchedule, len(entries))
	for i := range entries {
		schedules[i] = &countingSchedule{next: now.Add(24 * time.Hour)}
		entries[i] = scheduledWorkflow{
			repo:          scheduleTestRepo(),
			scheduledRepo: scheduleTestSnapshot(),
			name:          "build.yml",
			schedule:      schedules[i],
		}
	}

	scheduler := newTestScheduler(t)
	scheduler.replaceCachedSchedules(scheduleTestRepo().RepoDid.String(), entries, now)
	for _, schedule := range schedules {
		schedule.calls = 0
	}
	if due := scheduler.due(now.Add(time.Minute)); len(due) != 0 {
		t.Fatalf("due repositories = %d, want 0", len(due))
	}
	for i, schedule := range schedules {
		if schedule.calls != 0 {
			t.Fatalf("schedule %d evaluated %d times while not due", i, schedule.calls)
		}
	}
}

type sequenceSchedule struct {
	next []time.Time
}

func (s *sequenceSchedule) Next(time.Time) time.Time {
	if len(s.next) == 0 {
		return time.Time{}
	}
	next := s.next[0]
	s.next = s.next[1:]
	return next
}

func TestPipelineSchedulerDropsSchedulesThatDoNotAdvance(t *testing.T) {
	at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
	tests := []struct {
		name string
		next []time.Time
		due  bool
	}{
		{name: "zero initial occurrence"},
		{name: "zero after due occurrence", next: []time.Time{at, {}}, due: true},
		{name: "same occurrence repeats", next: []time.Time{at, at}, due: true},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			entry := scheduledWorkflow{
				repo:          scheduleTestRepo(),
				scheduledRepo: scheduleTestSnapshot(),
				name:          "build.yml",
				schedule:      &sequenceSchedule{next: test.next},
			}
			scheduler := newTestScheduler(t)
			scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))
			due := scheduler.due(at)
			if got := len(due) > 0; got != test.due {
				t.Fatalf("due = %v, want %v", got, test.due)
			}
			if scheduler.queue.Len() != 0 {
				t.Fatalf("queue length = %d, want 0 after non-advancing occurrence", scheduler.queue.Len())
			}
		})
	}
}

type slowFirstLoad struct {
	started chan struct{}
	gate    chan struct{}
	release func()
	calls   atomic.Int32
	entries []scheduledWorkflow
}

func (l *slowFirstLoad) load(context.Context, db.Repo, time.Time) ([]scheduledWorkflow, db.ScheduledRepo, error) {
	call := int(l.calls.Add(1))
	if call == 1 {
		close(l.started)
		<-l.gate
	}
	return []scheduledWorkflow{l.entries[min(call, len(l.entries))-1]}, scheduleTestSnapshot(), nil
}

func newRefreshTest(t *testing.T, entries ...scheduledWorkflow) (*pipelineScheduler, *slowFirstLoad, db.Repo) {
	t.Helper()
	scheduler := newTestScheduler(t)
	repo := scheduleTestRepo()
	if err := scheduler.db.AddRepo(repo); err != nil {
		t.Fatalf("AddRepo: %v", err)
	}
	slow := &slowFirstLoad{started: make(chan struct{}), gate: make(chan struct{}), entries: entries}
	var once sync.Once
	slow.release = func() { once.Do(func() { close(slow.gate) }) }
	t.Cleanup(slow.release)
	scheduler.load = slow.load
	return scheduler, slow, repo
}

func refreshAsync(s *pipelineScheduler, repo db.Repo) <-chan error {
	return goErr(func() error { return s.RefreshRepo(context.Background(), repo) })
}

func mustRefresh(t *testing.T, done <-chan error) {
	t.Helper()
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("RefreshRepo: %v", err)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("RefreshRepo did not return")
	}
}

func persistedWorkflows(t *testing.T, s *pipelineScheduler) []string {
	t.Helper()
	definitions, err := s.db.WorkflowSchedules(context.Background())
	if err != nil {
		t.Fatalf("WorkflowSchedules: %v", err)
	}
	var names []string
	for _, definition := range definitions {
		names = append(names, definition.Workflow)
	}
	return names
}

func requireNothingInflight(t *testing.T, s *pipelineScheduler) {
	t.Helper()
	s.refreshMu.Lock()
	defer s.refreshMu.Unlock()
	if len(s.inflight) != 0 {
		t.Fatalf("%d repos still tracked once no refresh is running", len(s.inflight))
	}
}

func waitForQueuedRefresh(t *testing.T, s *pipelineScheduler, repo db.Repo) {
	t.Helper()
	eventually(t, 2*time.Second, "the second refresh to queue behind the first", func() bool {
		s.refreshMu.Lock()
		defer s.refreshMu.Unlock()
		e := s.inflight[repo.RepoDid.String()]
		return e != nil && e.users == 2
	})
}

func TestPipelineSchedulerSerializesRepositoryRefreshes(t *testing.T) {
	scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "old.yml"), mustCron(t, "0 9 * * *", "new.yml"))

	firstDone := refreshAsync(scheduler, repo)
	<-slow.started
	secondDone := refreshAsync(scheduler, repo)
	time.Sleep(time.Second)
	if got := slow.calls.Load(); got != 1 {
		t.Errorf("%d loads started before the first refresh completed, want 1", got)
	}
	slow.release()
	mustRefresh(t, firstDone)
	mustRefresh(t, secondDone)

	if got := persistedWorkflows(t, scheduler); !slices.Equal(got, []string{"new.yml"}) {
		t.Fatalf("persisted schedules = %v, want only newest refresh", got)
	}
}

func TestPipelineSchedulerRemovalDoesNotWaitForInflightRefresh(t *testing.T) {
	scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 9 * * *", "build.yml"))

	refreshDone := refreshAsync(scheduler, repo)
	<-slow.started
	removeDone := make(chan error, 1)
	go func() { removeDone <- scheduler.RemoveRepo(context.Background(), repo.RepoDid.String()) }()
	// a clone can take minutes, and removal can't wait on it
	select {
	case err := <-removeDone:
		if err != nil {
			t.Fatalf("RemoveRepo: %v", err)
		}
	case <-time.After(time.Second):
		t.Fatal("repository removal waited for its schedule refresh that was still in flight")
	}
	slow.release()
	mustRefresh(t, refreshDone)
	if got := persistedWorkflows(t, scheduler); len(got) != 0 {
		t.Fatalf("persisted schedules after removal = %v, want none, a load that finished after the removal has to be thrown away", got)
	}
}

func TestAReloadAfterARemovalKeepsItsResultAndNothingStaysInFlight(t *testing.T) {
	scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "old.yml"), mustCron(t, "0 9 * * *", "new.yml"))

	firstDone := refreshAsync(scheduler, repo)
	<-slow.started
	if err := scheduler.RemoveRepo(context.Background(), repo.RepoDid.String()); err != nil {
		t.Fatalf("RemoveRepo: %v", err)
	}
	secondDone := refreshAsync(scheduler, repo)
	time.Sleep(50 * time.Millisecond)
	slow.release()
	mustRefresh(t, firstDone)
	mustRefresh(t, secondDone)

	if got := persistedWorkflows(t, scheduler); !slices.Equal(got, []string{"new.yml"}) {
		t.Fatalf("persisted schedules = %v, want only what the reload after the removal found", got)
	}
	requireNothingInflight(t, scheduler)
}

func TestARefreshQueuedBehindACloneWhenTheRepoIsRemovedThrowsItsResultAway(t *testing.T) {
	scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "wf.yml"))

	firstDone := refreshAsync(scheduler, repo)
	<-slow.started
	secondDone := refreshAsync(scheduler, repo)
	waitForQueuedRefresh(t, scheduler, repo)

	if err := scheduler.RemoveRepo(context.Background(), repo.RepoDid.String()); err != nil {
		t.Fatalf("RemoveRepo: %v", err)
	}
	slow.release()
	mustRefresh(t, firstDone)
	mustRefresh(t, secondDone)

	if got := persistedWorkflows(t, scheduler); len(got) != 0 {
		t.Fatalf("persisted schedules = %v, want none for a repo removed before the queued refresh ran", got)
	}
}

func TestRefreshesAskedForWhileOneIsAlreadyQueuedAreNotQueuedAgain(t *testing.T) {
	scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "wf.yml"))

	firstDone := refreshAsync(scheduler, repo)
	<-slow.started
	secondDone := refreshAsync(scheduler, repo)
	waitForQueuedRefresh(t, scheduler, repo)

	for range 20 {
		select {
		case err := <-refreshAsync(scheduler, repo):
			if err != nil {
				t.Fatalf("RefreshRepo: %v", err)
			}
		case <-time.After(time.Second):
			t.Fatal("a refresh asked for while one is queued waited behind the clone instead of being covered by it")
		}
	}

	slow.release()
	mustRefresh(t, firstDone)
	mustRefresh(t, secondDone)
	if got := slow.calls.Load(); got != 2 {
		t.Fatalf("%d schedule loads, want the running one and the queued one", got)
	}
}

func TestFlapsOfOneRepoLeaveOneWaiterBehindTheCloneNotOnePerRemoval(t *testing.T) {
	scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "wf.yml"))

	firstDone := refreshAsync(scheduler, repo)
	<-slow.started

	const returns = 20
	flaps := make(chan error, returns)
	for range returns {
		if err := scheduler.RemoveRepo(context.Background(), repo.RepoDid.String()); err != nil {
			t.Fatalf("RemoveRepo: %v", err)
		}
		go func() { flaps <- scheduler.RefreshRepo(context.Background(), repo) }()
		time.Sleep(time.Millisecond)
	}
	for range returns - 1 {
		select {
		case err := <-flaps:
			if err != nil {
				t.Fatalf("RefreshRepo: %v", err)
			}
		case <-time.After(2 * time.Second):
			t.Fatal("refreshes of a repo that was removed again are still parked behind the clone, one per removal")
		}
	}

	slow.release()
	mustRefresh(t, firstDone)
	mustRefresh(t, flaps)
	if got := slow.calls.Load(); got != 2 {
		t.Fatalf("%d schedule loads, want the running one and the newest request's", got)
	}
	if got := persistedWorkflows(t, scheduler); !slices.Equal(got, []string{"wf.yml"}) {
		t.Fatalf("persisted schedules = %v, want what the newest request loaded", got)
	}
	requireNothingInflight(t, scheduler)
}

func TestDefaultBranchComesFromTheRefsRecord(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	if _, err := env.s.defaultBranch(context.Background(), env.k, testRepoDid); err == nil {
		t.Fatal("a repo without a refs record has no default branch yet")
	}
	env.knot.put(testRepoDid, org_tangled.GitRefsNSID, knotclient.RefsRkey, org_tangled.GitRefs{
		LexiconTypeID: org_tangled.GitRefsNSID,
		DefaultBranch: "trunk",
	})
	got, err := env.s.defaultBranch(context.Background(), env.k, testRepoDid)
	if err != nil || got != "trunk" {
		t.Fatalf("default branch = %q, %v, want trunk", got, err)
	}
}