package repocache

import (
	"context"
	"encoding/json"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	gogit "github.com/go-git/go-git/v5"
	"github.com/go-git/go-git/v5/config"
	"github.com/go-git/go-git/v5/plumbing/object"
	"github.com/stretchr/testify/require"
	store "github.com/tidbcloud/AgentTeam/internal/store/repocache"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

func TestNormalizePublicRemoteURL(t *testing.T) {
	normalized, err := NormalizePublicRemoteURL(context.Background(), "https://93.184.216.34/Org/Repo")
	require.NoError(t, err)
	require.Equal(t, "https://93.184.216.34/Org/Repo.git", normalized)

	for _, raw := range []string{
		"http://93.184.216.34/org/repo.git",
		"ssh://git@93.184.216.34/org/repo.git",
		"https://token@93.184.216.34/org/repo.git",
		"https://127.0.0.1/org/repo.git",
	} {
		_, err := NormalizePublicRemoteURL(context.Background(), raw)
		require.Error(t, err, raw)
	}

	_, err = NormalizePublicRemoteURL(context.Background(), "https://93.184.216.34/"+strings.Repeat("a", maxNormalizedURLLength)+".git")
	require.ErrorContains(t, err, "must not exceed 768 characters")
}

func TestBuildPublishesBareMirrorAndRefreshesIt(t *testing.T) {
	ctx := context.Background()
	upstream := filepath.Join(t.TempDir(), "upstream.git")
	_, err := gogit.PlainInit(upstream, true)
	require.NoError(t, err)
	seedUpstream(t, upstream, "first")

	service, repository := newServiceForTest(t)
	now := time.Now().UTC()
	created, err := repository.Create(ctx, store.Cache{ID: "prc_test", NormalizedRemoteURL: upstream, DisplayRemoteURL: upstream, Status: store.StatusPending, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	built, err := service.Build(ctx, created.ID, "ops")
	require.NoError(t, err)
	require.Equal(t, store.StatusReady, built.Status)
	require.NotEmpty(t, built.ActiveGeneration)
	var info cacheInfo
	require.NoError(t, json.Unmarshal([]byte(built.Info), &info))
	require.Positive(t, info.SizeBytes)
	firstPath := generationPathForTest(t, service, built)
	require.Equal(t, directorySizeForTest(t, firstPath), info.SizeBytes)
	cachePath, found, err := service.ResolveReady(ctx, upstream)
	require.NoError(t, err)
	// Local paths deliberately are not accepted by the public URL resolver.
	require.False(t, found)
	require.Empty(t, cachePath)
	firstRepo, err := gogit.PlainOpen(firstPath)
	require.NoError(t, err)
	firstCommit := mirrorHEAD(firstRepo)

	seedUpstream(t, upstream, "second")
	refreshed, err := service.Build(ctx, created.ID, "ops")
	require.NoError(t, err)
	require.NotEqual(t, built.ActiveGeneration, refreshed.ActiveGeneration)
	require.NotEqual(t, built.LastCommit, refreshed.LastCommit)
	require.NoError(t, json.Unmarshal([]byte(refreshed.Info), &info))
	refreshedPath := generationPathForTest(t, service, refreshed)
	require.Equal(t, directorySizeForTest(t, refreshedPath), info.SizeBytes)
	refreshedRepo, err := gogit.PlainOpen(refreshedPath)
	require.NoError(t, err)
	require.Equal(t, refreshed.LastCommit, mirrorHEAD(refreshedRepo))
	_, err = os.Stat(filepath.Join(refreshedPath, "objects", "info", "alternates"))
	require.ErrorIs(t, err, os.ErrNotExist, "published generations must not depend on the previous generation's objects")
	firstRepo, err = gogit.PlainOpen(firstPath)
	require.NoError(t, err)
	require.Equal(t, firstCommit, mirrorHEAD(firstRepo), "refresh must not mutate the serving generation")
	entries, err := os.ReadDir(service.root)
	require.NoError(t, err)
	for _, entry := range entries {
		require.NotContains(t, entry.Name(), ".build-")
	}
}

func TestResolveReadyKeepsServingPublishedGenerationWhileRefreshIsBuildingOrFailed(t *testing.T) {
	ctx := context.Background()
	service, repository := newServiceForTest(t)
	now := time.Now().UTC()
	remoteURL := "https://93.184.216.34/org/repo.git"
	generation := "prcg_serving"
	path, err := service.generationPath("prc_serving", generation)
	require.NoError(t, err)
	require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700))
	_, err = gogit.PlainInit(path, true)
	require.NoError(t, err)
	lastSuccess := now.Add(-time.Hour)
	_, err = repository.Create(ctx, store.Cache{
		ID:                  "prc_serving",
		NormalizedRemoteURL: remoteURL,
		DisplayRemoteURL:    remoteURL,
		Status:              store.StatusReady,
		ActiveGeneration:    generation,
		RefreshIntervalSecs: 60,
		LastSuccessAt:       &lastSuccess,
		CreatedBy:           "ops",
		UpdatedBy:           "ops",
		CreatedAt:           now,
		UpdatedAt:           now,
	})
	require.NoError(t, err)

	claimed, ok, err := repository.ClaimBuild(ctx, "prc_serving", "prca_serving", "worker", now, time.Hour)
	require.NoError(t, err)
	require.True(t, ok)
	require.Equal(t, store.StatusBuilding, claimed.Status)
	resolved, found, err := service.ResolveReady(ctx, remoteURL)
	require.NoError(t, err)
	require.True(t, found)
	require.Equal(t, path, resolved.Path)

	require.NoError(t, repository.FailBuild(ctx, "prc_serving", "prca_serving", "upstream unavailable", now.Add(time.Minute)))
	resolved, found, err = service.ResolveReady(ctx, remoteURL)
	require.NoError(t, err)
	require.True(t, found)
	require.Equal(t, path, resolved.Path)
}

func TestFailedRefreshDoesNotReplacePublishedGeneration(t *testing.T) {
	ctx := context.Background()
	upstream := filepath.Join(t.TempDir(), "upstream.git")
	_, err := gogit.PlainInit(upstream, true)
	require.NoError(t, err)
	seedUpstream(t, upstream, "first")
	service, repository := newServiceForTest(t)
	now := time.Now().UTC()
	created, err := repository.Create(ctx, store.Cache{ID: "prc_failure", NormalizedRemoteURL: upstream, DisplayRemoteURL: upstream, Status: store.StatusPending, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	built, err := service.Build(ctx, created.ID, "ops")
	require.NoError(t, err)
	activePath := generationPathForTest(t, service, built)

	require.NoError(t, os.RemoveAll(upstream))
	_, err = service.Build(ctx, created.ID, "ops")
	require.ErrorContains(t, err, "refresh public repository cache generation")
	failed, err := repository.Get(ctx, created.ID)
	require.NoError(t, err)
	require.Equal(t, store.StatusFailed, failed.Status)
	require.Equal(t, built.ActiveGeneration, failed.ActiveGeneration)
	require.Equal(t, built.LastCommit, failed.LastCommit)
	_, err = gogit.PlainOpen(activePath)
	require.NoError(t, err)
}

func TestLegacyReadyCacheStaysServingDuringGenerationRefresh(t *testing.T) {
	ctx := context.Background()
	service, repository := newServiceForTest(t)
	now := time.Now().UTC()
	remoteURL := "https://93.184.216.34/org/legacy.git"
	legacyPath, err := service.legacyCachePath("prc_legacy")
	require.NoError(t, err)
	_, err = gogit.PlainInit(legacyPath, true)
	require.NoError(t, err)
	lastSuccess := now.Add(-time.Hour)
	_, err = repository.Create(ctx, store.Cache{ID: "prc_legacy", NormalizedRemoteURL: remoteURL, DisplayRemoteURL: remoteURL, Status: store.StatusReady, RefreshIntervalSecs: 60, LastSuccessAt: &lastSuccess, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	_, claimed, err := repository.ClaimBuild(ctx, "prc_legacy", "prca_legacy", "worker", now, time.Hour)
	require.NoError(t, err)
	require.True(t, claimed)

	resolved, found, err := service.ResolveReady(ctx, remoteURL)
	require.NoError(t, err)
	require.True(t, found)
	require.Equal(t, legacyPath, resolved.Path)
}

func TestCleanupStaleGenerationsKeepsActiveAndRecentReaders(t *testing.T) {
	service, repository := newServiceForTest(t)
	service.buildTimeout = time.Minute
	now := time.Now().UTC()
	service.now = func() time.Time { return now }
	cacheID := "prc_cleanup"
	active := "prcg_active"
	recent := "prcg_recent"
	expired := "prcg_expired"
	orphan := "prcg_orphan"
	_, err := repository.Create(context.Background(), store.Cache{ID: cacheID, NormalizedRemoteURL: "https://93.184.216.34/org/cleanup.git", DisplayRemoteURL: "https://93.184.216.34/org/cleanup.git", Status: store.StatusReady, ActiveGeneration: active, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	for _, generation := range []string{active, recent, expired, orphan} {
		path, err := service.generationPath(cacheID, generation)
		require.NoError(t, err)
		require.NoError(t, os.MkdirAll(path, 0o700))
	}
	for generation, retiredAt := range map[string]time.Time{
		recent:  now.Add(-5 * time.Minute),
		expired: now.Add(-20 * time.Minute),
	} {
		marker, err := service.retirementMarkerPath(cacheID, generation)
		require.NoError(t, err)
		require.NoError(t, os.MkdirAll(filepath.Dir(marker), 0o700))
		require.NoError(t, os.WriteFile(marker, nil, 0o600))
		require.NoError(t, os.Chtimes(marker, retiredAt, retiredAt))
	}
	orphanPath, err := service.generationPath(cacheID, orphan)
	require.NoError(t, err)
	require.NoError(t, os.Chtimes(orphanPath, now.Add(-61*time.Minute), now.Add(-61*time.Minute)))
	legacyPath, err := service.legacyCachePath(cacheID)
	require.NoError(t, err)
	require.NoError(t, os.MkdirAll(legacyPath, 0o700))
	legacyMarker, err := service.retirementMarkerPath(cacheID, "")
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(legacyMarker, nil, 0o600))
	require.NoError(t, os.Chtimes(legacyMarker, now.Add(-20*time.Minute), now.Add(-20*time.Minute)))

	processed, err := service.CleanupStaleGenerations(context.Background(), 1)
	require.NoError(t, err)
	require.Equal(t, 1, processed)
	for _, generation := range []string{active, recent} {
		path, err := service.generationPath(cacheID, generation)
		require.NoError(t, err)
		_, err = os.Stat(path)
		require.NoError(t, err)
	}
	for _, generation := range []string{expired, orphan} {
		path, err := service.generationPath(cacheID, generation)
		require.NoError(t, err)
		_, err = os.Stat(path)
		require.ErrorIs(t, err, os.ErrNotExist)
	}
	_, err = os.Stat(legacyPath)
	require.ErrorIs(t, err, os.ErrNotExist)
}

func TestCleanupStaleGenerationsAdvancesBoundedCursor(t *testing.T) {
	service, repository := newServiceForTest(t)
	service.buildTimeout = time.Minute
	now := time.Now().UTC()
	service.now = func() time.Time { return now }
	paths := make(map[string]string)
	for _, cacheID := range []string{"prc_cleanup_a", "prc_cleanup_b"} {
		_, err := repository.Create(context.Background(), store.Cache{ID: cacheID, NormalizedRemoteURL: "https://93.184.216.34/org/" + cacheID + ".git", DisplayRemoteURL: "https://93.184.216.34/org/" + cacheID + ".git", Status: store.StatusPending, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
		require.NoError(t, err)
		path, err := service.generationPath(cacheID, "prcg_orphan")
		require.NoError(t, err)
		require.NoError(t, os.MkdirAll(path, 0o700))
		require.NoError(t, os.Chtimes(path, now.Add(-61*time.Minute), now.Add(-61*time.Minute)))
		paths[cacheID] = path
	}

	processed, err := service.CleanupStaleGenerations(context.Background(), 1)
	require.NoError(t, err)
	require.Equal(t, 1, processed)
	_, err = os.Stat(paths["prc_cleanup_a"])
	require.ErrorIs(t, err, os.ErrNotExist)
	_, err = os.Stat(paths["prc_cleanup_b"])
	require.NoError(t, err)

	processed, err = service.CleanupStaleGenerations(context.Background(), 1)
	require.NoError(t, err)
	require.Equal(t, 1, processed)
	_, err = os.Stat(paths["prc_cleanup_b"])
	require.ErrorIs(t, err, os.ErrNotExist)
}

func TestLegacyCacheFirstRefreshPublishesGenerationAndPeriodicCleanupRetiresLegacy(t *testing.T) {
	ctx := context.Background()
	upstream := filepath.Join(t.TempDir(), "upstream.git")
	_, err := gogit.PlainInit(upstream, true)
	require.NoError(t, err)
	seedUpstream(t, upstream, "first")

	service, repository := newServiceForTest(t)
	service.buildTimeout = time.Minute
	now := time.Now().UTC()
	service.now = func() time.Time { return now }
	cacheID := "prc_legacy_upgrade"
	legacyPath, err := service.legacyCachePath(cacheID)
	require.NoError(t, err)
	_, err = gogit.PlainClone(legacyPath, true, &gogit.CloneOptions{URL: upstream, Mirror: true})
	require.NoError(t, err)
	lastSuccess := now.Add(-time.Hour)
	_, err = repository.Create(ctx, store.Cache{ID: cacheID, NormalizedRemoteURL: upstream, DisplayRemoteURL: upstream, Status: store.StatusReady, RefreshIntervalSecs: 60, LastSuccessAt: &lastSuccess, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	seedUpstream(t, upstream, "second")

	refreshed, err := service.Build(ctx, cacheID, "ops")
	require.NoError(t, err)
	require.NotEmpty(t, refreshed.ActiveGeneration)
	activePath := generationPathForTest(t, service, refreshed)
	activeRepo, err := gogit.PlainOpen(activePath)
	require.NoError(t, err)
	require.Equal(t, refreshed.LastCommit, mirrorHEAD(activeRepo))
	_, err = os.Stat(legacyPath)
	require.NoError(t, err, "legacy readers must remain valid through the retirement window")
	legacyMarker, err := service.retirementMarkerPath(cacheID, "")
	require.NoError(t, err)
	_, err = os.Stat(legacyMarker)
	require.NoError(t, err)

	now = now.Add(service.buildTimeout + retiredGenerationBuffer + time.Minute)
	processed, err := service.CleanupStaleGenerations(ctx, 10)
	require.NoError(t, err)
	require.Equal(t, 1, processed)
	_, err = os.Stat(legacyPath)
	require.ErrorIs(t, err, os.ErrNotExist)
	_, err = os.Stat(activePath)
	require.NoError(t, err)
}

func TestConcurrentBuildClaimIsRejected(t *testing.T) {
	service, repository := newServiceForTest(t)
	now := time.Now().UTC()
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_busy", NormalizedRemoteURL: "https://93.184.216.34/org/repo.git", DisplayRemoteURL: "https://93.184.216.34/org/repo.git", Status: store.StatusBuilding, LeaseOwner: "worker-a", LeaseUntil: ptrTime(now.Add(time.Hour)), RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	_, err = service.Build(context.Background(), "prc_busy", "worker-b")
	require.ErrorContains(t, err, "already building")
}

func TestExpiredBuildAttemptCannotMutateNewAttemptWithSameActor(t *testing.T) {
	_, repository := newServiceForTest(t)
	now := time.Now().UTC()
	oldGeneration := "prcg_old"
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_attempt_fence", NormalizedRemoteURL: "https://93.184.216.34/org/fence.git", DisplayRemoteURL: "https://93.184.216.34/org/fence.git", Status: store.StatusReady, ActiveGeneration: oldGeneration, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	_, claimed, err := repository.ClaimBuild(context.Background(), "prc_attempt_fence", "prca_old", "meshx-public-repo-cache-scheduler", now, time.Minute)
	require.NoError(t, err)
	require.True(t, claimed)
	_, err = repository.CompleteBuild(context.Background(), "prc_attempt_fence", "prca_old", "prcg_expired", "expired-commit", `{"sizeBytes":1}`, now.Add(90*time.Second))
	require.ErrorContains(t, err, "build lease was lost")
	require.ErrorContains(t, repository.FailBuild(context.Background(), "prc_attempt_fence", "prca_old", "expired failure", now.Add(90*time.Second)), "build lease was lost")
	_, claimed, err = repository.ClaimBuild(context.Background(), "prc_attempt_fence", "prca_new", "meshx-public-repo-cache-scheduler", now.Add(2*time.Minute), time.Hour)
	require.NoError(t, err)
	require.True(t, claimed)

	_, err = repository.CompleteBuild(context.Background(), "prc_attempt_fence", "prca_old", "prcg_stale", "stale-commit", `{"sizeBytes":1}`, now.Add(3*time.Minute))
	require.ErrorContains(t, err, "build lease was lost")
	require.ErrorContains(t, repository.FailBuild(context.Background(), "prc_attempt_fence", "prca_old", "stale failure", now.Add(3*time.Minute)), "build lease was lost")
	_, _, err = repository.CancelBuild(context.Background(), "prc_attempt_fence", "prca_old", "ops", now.Add(3*time.Minute))
	require.ErrorContains(t, err, "no longer building")

	cache, err := repository.Get(context.Background(), "prc_attempt_fence")
	require.NoError(t, err)
	require.Equal(t, store.StatusBuilding, cache.Status)
	require.Equal(t, "prca_new", cache.LeaseOwner)
	require.Equal(t, "meshx-public-repo-cache-scheduler", cache.UpdatedBy)
	require.Equal(t, oldGeneration, cache.ActiveGeneration)
}

func TestBuildLeaseDurationCoversConfiguredTimeout(t *testing.T) {
	require.Equal(t, defaultLeaseDuration, buildLeaseDuration(time.Minute))
	require.Equal(t, 75*time.Minute, buildLeaseDuration(time.Hour))
}

func TestCancelBuildStopsActiveBuildAndReleasesLease(t *testing.T) {
	service, repository := newServiceForTest(t)
	now := time.Now().UTC()
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_cancel", NormalizedRemoteURL: "https://93.184.216.34/org/repo.git", DisplayRemoteURL: "https://93.184.216.34/org/repo.git", Status: store.StatusBuilding, LeaseOwner: "prca_cancel", LeaseUntil: ptrTime(now.Add(time.Hour)), RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	buildCtx, buildCancel := context.WithCancel(context.Background())
	service.builds["prca_cancel"] = &activeBuild{attemptToken: "prca_cancel", cancel: buildCancel}
	updated, err := service.CancelBuild(buildCtx, "prc_cancel", "cancel-user")
	require.NoError(t, err)
	require.Equal(t, store.StatusCancelled, updated.Status)
	require.Equal(t, "build cancelled", updated.LastError)
	require.Empty(t, updated.LeaseOwner)
	require.Nil(t, updated.LeaseUntil)
	require.Equal(t, "cancel-user", updated.UpdatedBy)
	require.ErrorIs(t, buildCtx.Err(), context.Canceled)
}

func TestCancelBuildReleasesRemoteInstanceLease(t *testing.T) {
	service, repository := newServiceForTest(t)
	now := time.Now().UTC()
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_remote", NormalizedRemoteURL: "https://93.184.216.34/org/remote.git", DisplayRemoteURL: "https://93.184.216.34/org/remote.git", Status: store.StatusBuilding, LeaseOwner: "prca_remote", LeaseUntil: ptrTime(now.Add(time.Hour)), RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	updated, err := service.CancelBuild(context.Background(), "prc_remote", "cancel-user")
	require.NoError(t, err)
	require.Equal(t, store.StatusCancelled, updated.Status)
	require.Empty(t, updated.LeaseOwner)
}

func TestCancelBuildUsesCurrentDBAttemptWhenLocalAttemptIsStale(t *testing.T) {
	service, repository := newServiceForTest(t)
	now := time.Now().UTC()
	service.now = func() time.Time { return now.Add(3 * time.Minute) }
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_routed_cancel", NormalizedRemoteURL: "https://93.184.216.34/org/routed.git", DisplayRemoteURL: "https://93.184.216.34/org/routed.git", Status: store.StatusReady, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	_, claimed, err := repository.ClaimBuild(context.Background(), "prc_routed_cancel", "prca_local_stale", "meshx-public-repo-cache-scheduler", now, time.Minute)
	require.NoError(t, err)
	require.True(t, claimed)
	staleCtx, cancelStale := context.WithCancel(context.Background())
	t.Cleanup(cancelStale)
	staleBuild := &activeBuild{attemptToken: "prca_local_stale", cancel: cancelStale}
	service.builds["prca_local_stale"] = staleBuild
	_, claimed, err = repository.ClaimBuild(context.Background(), "prc_routed_cancel", "prca_remote_current", "meshx-public-repo-cache-scheduler", now.Add(2*time.Minute), time.Hour)
	require.NoError(t, err)
	require.True(t, claimed)

	updated, err := service.CancelBuild(context.Background(), "prc_routed_cancel", "cancel-user")
	require.NoError(t, err)
	require.Equal(t, store.StatusCancelled, updated.Status)
	require.Empty(t, updated.LeaseOwner)
	require.NoError(t, staleCtx.Err(), "cancel routed to the stale instance must not cancel an unrelated local attempt")
	require.Same(t, staleBuild, service.builds["prca_local_stale"])
	_, err = repository.CompleteBuild(context.Background(), "prc_routed_cancel", "prca_remote_current", "prcg_cancelled", "cancelled-commit", `{"sizeBytes":1}`, now.Add(4*time.Minute))
	require.ErrorContains(t, err, "build lease was lost")
}

func TestBuildAndCancelCloseClaimBeforeRegisterRace(t *testing.T) {
	service, repository := newServiceForTest(t)
	ctx := context.Background()
	now := time.Now().UTC()
	_, err := repository.Create(ctx, store.Cache{ID: "prc_register_cancelled", NormalizedRemoteURL: "https://93.184.216.34/org/register-cancelled.git", DisplayRemoteURL: "https://93.184.216.34/org/register-cancelled.git", Status: store.StatusReady, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	claimed := make(chan attemptHookEvent, 1)
	service.attemptHooks.afterClaim = func(hookCtx context.Context, cacheID, attemptToken string) {
		event := attemptHookEvent{ctx: hookCtx, cacheID: cacheID, attemptToken: attemptToken, release: make(chan struct{})}
		claimed <- event
		<-event.release
	}
	buildResult := make(chan cacheCallResult, 1)
	go func() {
		cache, buildErr := service.Build(ctx, "prc_register_cancelled", "worker")
		buildResult <- cacheCallResult{cache: cache, err: buildErr}
	}()

	claimEvent := awaitTestValue(t, claimed)
	require.Equal(t, "prc_register_cancelled", claimEvent.cacheID)
	updated, err := service.CancelBuild(ctx, "prc_register_cancelled", "cancel-user")
	require.NoError(t, err)
	require.Equal(t, store.StatusCancelled, updated.Status)
	close(claimEvent.release)
	result := awaitTestValue(t, buildResult)
	require.ErrorContains(t, result.err, "build lease was lost")
	require.NotContains(t, service.builds, claimEvent.attemptToken)
}

func TestBuildReleasesLeaseWhenRequestIsCancelledBeforeLeaseRecheck(t *testing.T) {
	claimMutated := make(chan transitionHookEvent, 1)
	var cancelRequest context.CancelFunc
	service, repository := newServiceForTestWithObserver(t, func(event store.TransitionEvent) {
		if event.Kind != store.TransitionClaimBuild || event.Stage != store.TransitionStageMutated || event.Cache.ID != "prc_cancelled_recheck" {
			return
		}
		cancelRequest()
		hook := transitionHookEvent{event: event, release: make(chan struct{})}
		claimMutated <- hook
		<-hook.release
	})
	now := time.Now().UTC()
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_cancelled_recheck", NormalizedRemoteURL: "https://93.184.216.34/org/cancelled-recheck.git", DisplayRemoteURL: "https://93.184.216.34/org/cancelled-recheck.git", Status: store.StatusReady, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	requestCtx, cancel := context.WithCancel(context.Background())
	cancelRequest = cancel
	buildResult := make(chan cacheCallResult, 1)
	go func() {
		cache, buildErr := service.Build(requestCtx, "prc_cancelled_recheck", "worker")
		buildResult <- cacheCallResult{cache: cache, err: buildErr}
	}()

	mutation := awaitTestValue(t, claimMutated)
	require.ErrorIs(t, requestCtx.Err(), context.Canceled)
	close(mutation.release)
	result := awaitTestValue(t, buildResult)
	require.ErrorIs(t, result.err, context.Canceled)
	cache, err := repository.Get(context.Background(), "prc_cancelled_recheck")
	require.NoError(t, err)
	require.Equal(t, store.StatusFailed, cache.Status)
	require.Empty(t, cache.LeaseOwner)
	require.Nil(t, cache.LeaseUntil)
	require.Contains(t, cache.LastError, context.Canceled.Error())
	require.NotContains(t, service.builds, mutation.event.AttemptToken)
}

func TestClaimBuildRetriesWhenActiveGenerationChangesAfterRead(t *testing.T) {
	claimRead := make(chan transitionHookEvent, 1)
	claimMutated := make(chan store.TransitionEvent, 1)
	var firstClaimRead sync.Once
	var cancelBuild context.CancelFunc
	service, repository := newServiceForTestWithObserver(t, func(event store.TransitionEvent) {
		if event.Kind != store.TransitionClaimBuild || event.Cache.ID != "prc_generation_race" {
			return
		}
		switch event.Stage {
		case store.TransitionStageRead:
			firstClaimRead.Do(func() {
				hook := transitionHookEvent{event: event, release: make(chan struct{})}
				claimRead <- hook
				<-hook.release
			})
		case store.TransitionStageMutated:
			cancelBuild()
			claimMutated <- event
		}
	})
	base := time.Now().UTC()
	service.now = func() time.Time { return base.Add(2 * time.Minute) }
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_generation_race", NormalizedRemoteURL: "https://93.184.216.34/org/generation-race.git", DisplayRemoteURL: "https://93.184.216.34/org/generation-race.git", Status: store.StatusBuilding, ActiveGeneration: "prcg_gen0", LeaseOwner: "prca_completing", LeaseUntil: ptrTime(base.Add(time.Minute)), RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "worker-b", CreatedAt: base, UpdatedAt: base})
	require.NoError(t, err)

	requestCtx, cancel := context.WithCancel(context.Background())
	cancelBuild = cancel
	buildResult := make(chan cacheCallResult, 1)
	go func() {
		cache, buildErr := service.Build(requestCtx, "prc_generation_race", "worker-a")
		buildResult <- cacheCallResult{cache: cache, err: buildErr}
	}()

	read := awaitTestValue(t, claimRead)
	require.Equal(t, "prcg_gen0", read.event.Cache.ActiveGeneration)
	completed, err := repository.CompleteBuild(context.Background(), "prc_generation_race", "prca_completing", "prcg_gen1", "gen1-commit", `{"sizeBytes":1}`, base.Add(30*time.Second))
	require.NoError(t, err)
	require.Equal(t, "prcg_gen1", completed.ActiveGeneration)
	close(read.release)
	mutation := awaitTestValue(t, claimMutated)
	require.Equal(t, 2, mutation.AttemptNumber)
	require.Equal(t, "prcg_gen1", mutation.Cache.ActiveGeneration)
	require.ErrorIs(t, awaitTestValue(t, buildResult).err, context.Canceled)
	final, err := repository.Get(context.Background(), "prc_generation_race")
	require.NoError(t, err)
	require.Equal(t, "prcg_gen1", final.ActiveGeneration)
	require.Equal(t, store.StatusFailed, final.Status)
}

func TestCompleteBuildRetriesWhenRefreshSettingsChangeAfterRead(t *testing.T) {
	completeRead := make(chan transitionHookEvent, 1)
	var firstCompleteRead sync.Once
	service, repository := newServiceForTestWithObserver(t, func(event store.TransitionEvent) {
		if event.Kind != store.TransitionCompleteBuild || event.Stage != store.TransitionStageRead || event.Cache.ID != "prc_settings_race" {
			return
		}
		firstCompleteRead.Do(func() {
			hook := transitionHookEvent{event: event, release: make(chan struct{})}
			completeRead <- hook
			<-hook.release
		})
	})
	base := time.Now().UTC()
	settingTime := base.Add(30 * time.Second)
	completionTime := base.Add(time.Minute)
	service.now = func() time.Time { return settingTime }
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_settings_race", NormalizedRemoteURL: "https://93.184.216.34/org/settings-race.git", DisplayRemoteURL: "https://93.184.216.34/org/settings-race.git", Status: store.StatusBuilding, ActiveGeneration: "prcg_gen0", LeaseOwner: "prca_settings", LeaseUntil: ptrTime(base.Add(time.Hour)), RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "worker", CreatedAt: base, UpdatedAt: base})
	require.NoError(t, err)

	completeResult := make(chan cacheCallResult, 1)
	go func() {
		cache, completeErr := repository.CompleteBuild(context.Background(), "prc_settings_race", "prca_settings", "prcg_gen1", "gen1-commit", `{"sizeBytes":1}`, completionTime)
		completeResult <- cacheCallResult{cache: cache, err: completeErr}
	}()

	read := awaitTestValue(t, completeRead)
	require.False(t, read.event.Cache.AutoRefreshEnabled)
	updated, err := service.Update(context.Background(), "prc_settings_race", UpdateInput{AutoRefreshEnabled: true, RefreshIntervalSecs: 120, Actor: "operator"})
	require.NoError(t, err)
	require.True(t, updated.AutoRefreshEnabled)
	require.Equal(t, settingTime.Add(120*time.Second), *updated.NextSyncAt)
	close(read.release)
	completed := awaitTestValue(t, completeResult)
	require.NoError(t, completed.err)
	require.True(t, completed.cache.AutoRefreshEnabled)
	require.Equal(t, 120, completed.cache.RefreshIntervalSecs)
	require.Equal(t, completionTime.Add(120*time.Second), *completed.cache.NextSyncAt)
	final, err := repository.Get(context.Background(), "prc_settings_race")
	require.NoError(t, err)
	require.Equal(t, completionTime.Add(120*time.Second), *final.NextSyncAt)
	require.Equal(t, uint64(2), final.Revision)
}

func TestCompleteBuildRetriesWhenSameSettingsUpdateChangesActorAfterRead(t *testing.T) {
	completeRead := make(chan transitionHookEvent, 1)
	completeMutated := make(chan store.TransitionEvent, 1)
	var firstCompleteRead sync.Once
	service, repository := newServiceForTestWithObserver(t, func(event store.TransitionEvent) {
		if event.Kind != store.TransitionCompleteBuild || event.Cache.ID != "prc_same_settings_race" {
			return
		}
		switch event.Stage {
		case store.TransitionStageRead:
			firstCompleteRead.Do(func() {
				hook := transitionHookEvent{event: event, release: make(chan struct{})}
				completeRead <- hook
				<-hook.release
			})
		case store.TransitionStageMutated:
			completeMutated <- event
		}
	})
	base := time.Now().UTC()
	settingTime := base.Add(30 * time.Second)
	completionTime := base.Add(time.Minute)
	service.now = func() time.Time { return settingTime }
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_same_settings_race", NormalizedRemoteURL: "https://93.184.216.34/org/same-settings-race.git", DisplayRemoteURL: "https://93.184.216.34/org/same-settings-race.git", Status: store.StatusBuilding, ActiveGeneration: "prcg_gen0", LeaseOwner: "prca_same_settings", LeaseUntil: ptrTime(base.Add(time.Hour)), AutoRefreshEnabled: true, RefreshIntervalSecs: 120, CreatedBy: "ops", UpdatedBy: "worker", CreatedAt: base, UpdatedAt: base})
	require.NoError(t, err)

	completeResult := make(chan cacheCallResult, 1)
	go func() {
		cache, completeErr := repository.CompleteBuild(context.Background(), "prc_same_settings_race", "prca_same_settings", "prcg_gen1", "gen1-commit", `{"sizeBytes":1}`, completionTime)
		completeResult <- cacheCallResult{cache: cache, err: completeErr}
	}()

	read := awaitTestValue(t, completeRead)
	require.Equal(t, "worker", read.event.Cache.UpdatedBy)
	updated, err := service.Update(context.Background(), "prc_same_settings_race", UpdateInput{AutoRefreshEnabled: true, RefreshIntervalSecs: 120, Actor: "operator"})
	require.NoError(t, err)
	require.Equal(t, uint64(1), updated.Revision)
	require.Equal(t, "operator", updated.UpdatedBy)
	close(read.release)

	mutation := awaitTestValue(t, completeMutated)
	require.Equal(t, 2, mutation.AttemptNumber)
	completed := awaitTestValue(t, completeResult)
	require.NoError(t, completed.err)
	require.Equal(t, uint64(2), completed.cache.Revision)
	require.Equal(t, "operator", completed.cache.UpdatedBy)
	require.Equal(t, completionTime.Add(120*time.Second), *completed.cache.NextSyncAt)
}

func TestUpdateSettingsRetriesWhenBuildCompletesAfterRead(t *testing.T) {
	updateRead := make(chan transitionHookEvent, 1)
	updateMutated := make(chan store.TransitionEvent, 1)
	var firstUpdateRead sync.Once
	service, repository := newServiceForTestWithObserver(t, func(event store.TransitionEvent) {
		if event.Kind != store.TransitionUpdateSettings || event.Cache.ID != "prc_update_race" {
			return
		}
		switch event.Stage {
		case store.TransitionStageRead:
			firstUpdateRead.Do(func() {
				hook := transitionHookEvent{event: event, release: make(chan struct{})}
				updateRead <- hook
				<-hook.release
			})
		case store.TransitionStageMutated:
			updateMutated <- event
		}
	})
	base := time.Now().UTC()
	completionTime := base.Add(30 * time.Second)
	settingTime := base.Add(time.Minute)
	service.now = func() time.Time { return settingTime }
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_update_race", NormalizedRemoteURL: "https://93.184.216.34/org/update-race.git", DisplayRemoteURL: "https://93.184.216.34/org/update-race.git", Status: store.StatusBuilding, ActiveGeneration: "prcg_gen0", LeaseOwner: "prca_update", LeaseUntil: ptrTime(base.Add(time.Hour)), RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "worker", CreatedAt: base, UpdatedAt: base})
	require.NoError(t, err)

	updateResult := make(chan cacheCallResult, 1)
	go func() {
		cache, updateErr := service.Update(context.Background(), "prc_update_race", UpdateInput{AutoRefreshEnabled: true, RefreshIntervalSecs: 120, Actor: "operator"})
		updateResult <- cacheCallResult{cache: cache, err: updateErr}
	}()

	read := awaitTestValue(t, updateRead)
	require.Equal(t, store.StatusBuilding, read.event.Cache.Status)
	require.Equal(t, "prcg_gen0", read.event.Cache.ActiveGeneration)
	require.Equal(t, uint64(0), read.event.Cache.Revision)
	completed, err := repository.CompleteBuild(context.Background(), "prc_update_race", "prca_update", "prcg_gen1", "gen1-commit", `{"sizeBytes":1}`, completionTime)
	require.NoError(t, err)
	require.Equal(t, uint64(1), completed.Revision)
	close(read.release)

	mutation := awaitTestValue(t, updateMutated)
	require.Equal(t, 2, mutation.AttemptNumber)
	require.Equal(t, uint64(2), mutation.Cache.Revision)
	updated := awaitTestValue(t, updateResult)
	require.NoError(t, updated.err)
	require.Equal(t, store.StatusReady, updated.cache.Status)
	require.Equal(t, "prcg_gen1", updated.cache.ActiveGeneration)
	require.Empty(t, updated.cache.LeaseOwner)
	require.Nil(t, updated.cache.LeaseUntil)
	require.True(t, updated.cache.AutoRefreshEnabled)
	require.Equal(t, 120, updated.cache.RefreshIntervalSecs)
	require.Equal(t, settingTime.Add(120*time.Second), *updated.cache.NextSyncAt)
	require.Equal(t, "operator", updated.cache.UpdatedBy)
	require.Equal(t, uint64(2), updated.cache.Revision)

	final, err := repository.Get(context.Background(), "prc_update_race")
	require.NoError(t, err)
	require.Equal(t, updated.cache, final)
}

func TestUpdateSettingsSameValuesStillAdvancesRevision(t *testing.T) {
	service, repository := newServiceForTest(t)
	now := time.Now().UTC().Truncate(time.Second)
	service.now = func() time.Time { return now }
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_noop_update", NormalizedRemoteURL: "https://93.184.216.34/org/noop-update.git", DisplayRemoteURL: "https://93.184.216.34/org/noop-update.git", Status: store.StatusReady, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "operator", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	first, err := service.Update(context.Background(), "prc_noop_update", UpdateInput{AutoRefreshEnabled: false, RefreshIntervalSecs: 60, Actor: "operator"})
	require.NoError(t, err)
	require.Equal(t, uint64(1), first.Revision)
	second, err := service.Update(context.Background(), "prc_noop_update", UpdateInput{AutoRefreshEnabled: false, RefreshIntervalSecs: 60, Actor: "operator"})
	require.NoError(t, err)
	require.Equal(t, uint64(2), second.Revision)
	require.Equal(t, store.StatusReady, second.Status)
	require.Equal(t, "operator", second.UpdatedBy)
	require.Nil(t, second.NextSyncAt)
}

func TestCancelBuildCancelsExactAttemptAcrossNewClaim(t *testing.T) {
	cancelled := make(chan transitionHookEvent, 1)
	var firstCancel sync.Once
	service, repository := newServiceForTestWithObserver(t, func(event store.TransitionEvent) {
		if event.Kind != store.TransitionCancelBuild || event.Stage != store.TransitionStageMutated || event.Cache.ID != "prc_overlapping_cancel" {
			return
		}
		firstCancel.Do(func() {
			hook := transitionHookEvent{event: event, release: make(chan struct{})}
			cancelled <- hook
			<-hook.release
		})
	})
	ctx := context.Background()
	upstream := filepath.Join(t.TempDir(), "overlapping-upstream.git")
	_, err := gogit.PlainInit(upstream, true)
	require.NoError(t, err)
	seedUpstream(t, upstream, "overlapping")
	now := time.Now().UTC()
	_, err = repository.Create(ctx, store.Cache{ID: "prc_overlapping_cancel", NormalizedRemoteURL: upstream, DisplayRemoteURL: upstream, Status: store.StatusReady, RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	registered := make(chan attemptHookEvent, 2)
	service.attemptHooks.afterRegister = func(hookCtx context.Context, cacheID, attemptToken string) {
		event := attemptHookEvent{ctx: hookCtx, cacheID: cacheID, attemptToken: attemptToken, release: make(chan struct{})}
		registered <- event
		<-event.release
	}

	oldBuildResult := make(chan cacheCallResult, 1)
	go func() {
		cache, buildErr := service.Build(ctx, "prc_overlapping_cancel", "worker-b")
		oldBuildResult <- cacheCallResult{cache: cache, err: buildErr}
	}()
	oldAttempt := awaitTestValue(t, registered)

	cancelRequestCtx, cancelRequest := context.WithCancel(ctx)
	cancelResult := make(chan cacheCallResult, 1)
	go func() {
		cache, cancelErr := service.CancelBuild(cancelRequestCtx, "prc_overlapping_cancel", "cancel-user")
		cancelResult <- cacheCallResult{cache: cache, err: cancelErr}
	}()
	cancelEvent := awaitTestValue(t, cancelled)
	require.Equal(t, oldAttempt.attemptToken, cancelEvent.event.AttemptToken)

	newBuildResult := make(chan cacheCallResult, 1)
	go func() {
		cache, buildErr := service.Build(ctx, "prc_overlapping_cancel", "worker-c")
		newBuildResult <- cacheCallResult{cache: cache, err: buildErr}
	}()
	newAttempt := awaitTestValue(t, registered)
	require.NotEqual(t, oldAttempt.attemptToken, newAttempt.attemptToken)

	cancelRequest()
	require.ErrorIs(t, cancelRequestCtx.Err(), context.Canceled)
	close(cancelEvent.release)
	result := awaitTestValue(t, cancelResult)
	require.NoError(t, result.err)
	require.Equal(t, store.StatusCancelled, result.cache.Status)
	require.ErrorIs(t, oldAttempt.ctx.Err(), context.Canceled)
	require.NoError(t, newAttempt.ctx.Err(), "cancelling the retired attempt must not cancel the new attempt")

	close(oldAttempt.release)
	require.Error(t, awaitTestValue(t, oldBuildResult).err)
	updated, err := service.CancelBuild(ctx, "prc_overlapping_cancel", "cancel-new-attempt")
	require.NoError(t, err)
	require.Equal(t, store.StatusCancelled, updated.Status)
	require.ErrorIs(t, newAttempt.ctx.Err(), context.Canceled)
	close(newAttempt.release)
	require.Error(t, awaitTestValue(t, newBuildResult).err)
	final, err := repository.Get(ctx, "prc_overlapping_cancel")
	require.NoError(t, err)
	require.Equal(t, store.StatusCancelled, final.Status)
	require.Empty(t, final.LeaseOwner)
}

func TestCancelledBuildCannotSwitchActiveGeneration(t *testing.T) {
	service, repository := newServiceForTest(t)
	now := time.Now().UTC()
	oldGeneration := "prcg_old"
	_, err := repository.Create(context.Background(), store.Cache{ID: "prc_fenced", NormalizedRemoteURL: "https://93.184.216.34/org/fenced.git", DisplayRemoteURL: "https://93.184.216.34/org/fenced.git", Status: store.StatusBuilding, ActiveGeneration: oldGeneration, LeaseOwner: "prca_remote", LeaseUntil: ptrTime(now.Add(time.Hour)), RefreshIntervalSecs: 60, CreatedBy: "ops", UpdatedBy: "ops", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	_, err = service.CancelBuild(context.Background(), "prc_fenced", "cancel-user")
	require.NoError(t, err)
	_, err = repository.CompleteBuild(context.Background(), "prc_fenced", "prca_remote", "prcg_new", "new-commit", `{"sizeBytes":1}`, now.Add(time.Minute))
	require.ErrorContains(t, err, "build lease was lost")
	cache, err := repository.Get(context.Background(), "prc_fenced")
	require.NoError(t, err)
	require.Equal(t, store.StatusCancelled, cache.Status)
	require.Equal(t, oldGeneration, cache.ActiveGeneration)
}

func newServiceForTest(t *testing.T) (*Service, *store.Store) {
	return newServiceForTestWithObserver(t, nil)
}

func newServiceForTestWithObserver(t *testing.T, observer func(store.TransitionEvent)) (*Service, *store.Store) {
	t.Helper()
	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "cache.db")), &gorm.Config{})
	require.NoError(t, err)
	require.NoError(t, db.AutoMigrate(&store.Cache{}))
	repository := store.New(db, store.WithTransitionObserver(observer))
	service := New(repository, filepath.Join(t.TempDir(), "cache"), "", time.Minute)
	return service, repository
}

func seedUpstream(t *testing.T, upstream, content string) {
	t.Helper()
	working := t.TempDir()
	repo, err := gogit.PlainInit(working, false)
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(filepath.Join(working, "README.md"), []byte(content), 0o600))
	worktree, err := repo.Worktree()
	require.NoError(t, err)
	_, err = worktree.Add("README.md")
	require.NoError(t, err)
	_, err = worktree.Commit(content, &gogit.CommitOptions{Author: testSignature()})
	require.NoError(t, err)
	_, err = repo.CreateRemote(&config.RemoteConfig{Name: "origin", URLs: []string{upstream}})
	require.NoError(t, err)
	require.NoError(t, repo.Push(&gogit.PushOptions{RemoteName: "origin", RefSpecs: []config.RefSpec{"+refs/heads/*:refs/heads/*"}}))
}

func testSignature() *object.Signature {
	return &object.Signature{Name: "test", Email: "test@example.com", When: time.Now()}
}

func ptrTime(value time.Time) *time.Time { return &value }

type attemptHookEvent struct {
	ctx          context.Context
	cacheID      string
	attemptToken string
	release      chan struct{}
}

type cacheCallResult struct {
	cache store.Cache
	err   error
}

type transitionHookEvent struct {
	event   store.TransitionEvent
	release chan struct{}
}

const testOrchestrationTimeout = 30 * time.Second

func awaitTestValue[T any](t *testing.T, values <-chan T) T {
	t.Helper()
	timer := time.NewTimer(testOrchestrationTimeout)
	defer timer.Stop()
	select {
	case value := <-values:
		return value
	case <-timer.C:
		t.Fatal("timed out waiting for test orchestration")
		var zero T
		return zero
	}
}

func generationPathForTest(t *testing.T, service *Service, cache store.Cache) string {
	t.Helper()
	path, err := service.generationPath(cache.ID, cache.ActiveGeneration)
	require.NoError(t, err)
	return path
}

func directorySizeForTest(t *testing.T, root string) int64 {
	t.Helper()
	var total int64
	require.NoError(t, filepath.Walk(root, func(_ string, info os.FileInfo, err error) error {
		if err == nil && info.Mode().IsRegular() {
			total += info.Size()
		}
		return err
	}))
	return total
}