package repocache

import (
	"context"
	"database/sql"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"gorm.io/driver/postgres"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

func TestTransitionSnapshotPostgreSQLAvoidsJSONEquality(t *testing.T) {
	db, err := gorm.Open(postgres.New(postgres.Config{Conn: &sql.DB{}, WithoutReturning: true}), &gorm.Config{DryRun: true, DisableAutomaticPing: true, SkipDefaultTransaction: true})
	require.NoError(t, err)
	now := time.Now().UTC()
	cache := legacyWriterRaceCache("prc_postgres_snapshot", now)
	cache.ActiveGeneration = "prcg_active"
	cache.Info = `{"sizeBytes":1}`
	cache.LastSyncAt = ptrStoreTime(now.Add(-time.Hour))
	cache.LastSuccessAt = ptrStoreTime(now.Add(-time.Hour))
	cache.NextSyncAt = ptrStoreTime(now.Add(time.Hour))
	cache.LeaseUntil = ptrStoreTime(now.Add(2 * time.Hour))

	result := whereTransitionSnapshot(db.Model(&Cache{}).Where("id = ?", cache.ID), cache).
		Updates(map[string]any{"status": StatusBuilding, "revision": gorm.Expr("revision + 1")})
	require.NoError(t, result.Error)
	statement := result.Statement.SQL.String()
	parts := strings.SplitN(statement, " WHERE ", 2)
	require.Len(t, parts, 2)
	predicate := parts[1]
	require.NotContains(t, predicate, "info")
	for _, column := range []string{"revision", "status", "active_generation", "auto_refresh_enabled", "refresh_interval_seconds", "lease_owner", "last_sync_at", "last_success_at", "next_sync_at", "lease_until"} {
		require.Contains(t, predicate, column)
	}
}

func TestClaimBuildRetriesAfterLegacyCompleteWithoutRevision(t *testing.T) {
	const cacheID = "prc_legacy_complete_claim"
	store, db, read, mutated := newLegacyWriterRaceStore(t, TransitionClaimBuild, cacheID)
	base := time.Now().UTC()
	cache := legacyWriterRaceCache(cacheID, base)
	cache.Status = StatusBuilding
	cache.ActiveGeneration = "prcg_gen0"
	cache.LeaseOwner = "prca_legacy"
	cache.LeaseUntil = ptrStoreTime(base.Add(time.Minute))
	_, err := store.Create(context.Background(), cache)
	require.NoError(t, err)

	type claimResult struct {
		cache   Cache
		claimed bool
		err     error
	}
	result := make(chan claimResult, 1)
	go func() {
		claimed, ok, claimErr := store.ClaimBuild(context.Background(), cacheID, "prca_current", "current-worker", base.Add(2*time.Minute), time.Hour)
		result <- claimResult{cache: claimed, claimed: ok, err: claimErr}
	}()

	hook := awaitLegacyRaceValue(t, read)
	require.Equal(t, "prcg_gen0", hook.event.Cache.ActiveGeneration)
	legacyCompleteWithoutRevision(t, db, cacheID, "prca_legacy", "prcg_gen1", base.Add(30*time.Second))
	close(hook.release)

	mutation := awaitLegacyRaceValue(t, mutated)
	require.Equal(t, 2, mutation.AttemptNumber)
	claimed := awaitLegacyRaceValue(t, result)
	require.NoError(t, claimed.err)
	require.True(t, claimed.claimed)
	require.Equal(t, StatusBuilding, claimed.cache.Status)
	require.Equal(t, "prcg_gen1", claimed.cache.ActiveGeneration)
	require.Equal(t, "prca_current", claimed.cache.LeaseOwner)
	require.Equal(t, uint64(1), claimed.cache.Revision)
}

func TestCompleteBuildRetriesAfterLegacySettingsUpdateWithoutRevision(t *testing.T) {
	const cacheID = "prc_legacy_settings_complete"
	store, db, read, mutated := newLegacyWriterRaceStore(t, TransitionCompleteBuild, cacheID)
	base := time.Now().UTC()
	completionTime := base.Add(time.Minute)
	cache := legacyWriterRaceCache(cacheID, base)
	cache.Status = StatusBuilding
	cache.ActiveGeneration = "prcg_gen0"
	cache.LeaseOwner = "prca_current"
	cache.LeaseUntil = ptrStoreTime(base.Add(time.Hour))
	cache.UpdatedBy = "current-worker"
	_, err := store.Create(context.Background(), cache)
	require.NoError(t, err)

	result := make(chan cacheResult, 1)
	go func() {
		completed, completeErr := store.CompleteBuild(context.Background(), cacheID, "prca_current", "prcg_gen1", "gen1-commit", `{"sizeBytes":1}`, completionTime)
		result <- cacheResult{cache: completed, err: completeErr}
	}()

	hook := awaitLegacyRaceValue(t, read)
	require.False(t, hook.event.Cache.AutoRefreshEnabled)
	legacyUpdateSettingsWithoutRevision(t, db, cacheID, true, 120, "legacy-operator", base.Add(30*time.Second))
	close(hook.release)

	mutation := awaitLegacyRaceValue(t, mutated)
	require.Equal(t, 2, mutation.AttemptNumber)
	completed := awaitLegacyRaceValue(t, result)
	require.NoError(t, completed.err)
	require.Equal(t, StatusReady, completed.cache.Status)
	require.True(t, completed.cache.AutoRefreshEnabled)
	require.Equal(t, 120, completed.cache.RefreshIntervalSecs)
	require.Equal(t, completionTime.Add(120*time.Second), *completed.cache.NextSyncAt)
	require.Equal(t, "legacy-operator", completed.cache.UpdatedBy)
	require.Equal(t, uint64(1), completed.cache.Revision)
}

func TestUpdateSettingsRetriesAfterLegacyCompleteWithoutRevision(t *testing.T) {
	const cacheID = "prc_legacy_complete_update"
	store, db, read, mutated := newLegacyWriterRaceStore(t, TransitionUpdateSettings, cacheID)
	base := time.Now().UTC()
	cache := legacyWriterRaceCache(cacheID, base)
	cache.Status = StatusBuilding
	cache.ActiveGeneration = "prcg_gen0"
	cache.LeaseOwner = "prca_legacy"
	cache.LeaseUntil = ptrStoreTime(base.Add(time.Hour))
	_, err := store.Create(context.Background(), cache)
	require.NoError(t, err)

	result := make(chan cacheResult, 1)
	go func() {
		updated, updateErr := store.UpdateSettings(context.Background(), cacheID, true, 120, "current-operator", base.Add(time.Minute))
		result <- cacheResult{cache: updated, err: updateErr}
	}()

	hook := awaitLegacyRaceValue(t, read)
	require.Equal(t, StatusBuilding, hook.event.Cache.Status)
	legacyCompleteWithoutRevision(t, db, cacheID, "prca_legacy", "prcg_gen1", base.Add(30*time.Second))
	close(hook.release)

	mutation := awaitLegacyRaceValue(t, mutated)
	require.Equal(t, 2, mutation.AttemptNumber)
	updated := awaitLegacyRaceValue(t, result)
	require.NoError(t, updated.err)
	require.Equal(t, StatusReady, updated.cache.Status)
	require.Equal(t, "prcg_gen1", updated.cache.ActiveGeneration)
	require.True(t, updated.cache.AutoRefreshEnabled)
	require.Equal(t, uint64(1), updated.cache.Revision)
}

func TestCancelBuildRetriesAfterLegacySettingsUpdateWithoutRevision(t *testing.T) {
	const cacheID = "prc_legacy_settings_cancel"
	store, db, read, mutated := newLegacyWriterRaceStore(t, TransitionCancelBuild, cacheID)
	base := time.Now().UTC()
	cache := legacyWriterRaceCache(cacheID, base)
	cache.Status = StatusBuilding
	cache.LeaseOwner = "prca_current"
	cache.LeaseUntil = ptrStoreTime(base.Add(time.Hour))
	_, err := store.Create(context.Background(), cache)
	require.NoError(t, err)

	type cancelResult struct {
		cache     Cache
		cancelled bool
		err       error
	}
	result := make(chan cancelResult, 1)
	go func() {
		cancelled, ok, cancelErr := store.CancelBuild(context.Background(), cacheID, "prca_current", "cancel-operator", base.Add(time.Minute))
		result <- cancelResult{cache: cancelled, cancelled: ok, err: cancelErr}
	}()

	hook := awaitLegacyRaceValue(t, read)
	legacyUpdateSettingsWithoutRevision(t, db, cacheID, true, 120, "legacy-operator", base.Add(30*time.Second))
	close(hook.release)

	mutation := awaitLegacyRaceValue(t, mutated)
	require.Equal(t, 2, mutation.AttemptNumber)
	cancelled := awaitLegacyRaceValue(t, result)
	require.NoError(t, cancelled.err)
	require.True(t, cancelled.cancelled)
	require.Equal(t, StatusCancelled, cancelled.cache.Status)
	require.True(t, cancelled.cache.AutoRefreshEnabled)
	require.Equal(t, 120, cancelled.cache.RefreshIntervalSecs)
	require.Equal(t, "cancel-operator", cancelled.cache.UpdatedBy)
	require.Equal(t, uint64(1), cancelled.cache.Revision)
}

func TestFailBuildRetriesAfterLegacySettingsUpdateWithoutRevision(t *testing.T) {
	const cacheID = "prc_legacy_settings_fail"
	store, db, read, mutated := newLegacyWriterRaceStore(t, TransitionFailBuild, cacheID)
	base := time.Now().UTC()
	cache := legacyWriterRaceCache(cacheID, base)
	cache.Status = StatusBuilding
	cache.LeaseOwner = "prca_current"
	cache.LeaseUntil = ptrStoreTime(base.Add(time.Hour))
	_, err := store.Create(context.Background(), cache)
	require.NoError(t, err)

	result := make(chan error, 1)
	go func() {
		result <- store.FailBuild(context.Background(), cacheID, "prca_current", "current failure", base.Add(time.Minute))
	}()

	hook := awaitLegacyRaceValue(t, read)
	legacyUpdateSettingsWithoutRevision(t, db, cacheID, true, 120, "legacy-operator", base.Add(30*time.Second))
	close(hook.release)

	mutation := awaitLegacyRaceValue(t, mutated)
	require.Equal(t, 2, mutation.AttemptNumber)
	require.NoError(t, awaitLegacyRaceValue(t, result))
	failed, err := store.Get(context.Background(), cacheID)
	require.NoError(t, err)
	require.Equal(t, StatusFailed, failed.Status)
	require.True(t, failed.AutoRefreshEnabled)
	require.Equal(t, 120, failed.RefreshIntervalSecs)
	require.Equal(t, "legacy-operator", failed.UpdatedBy)
	require.Equal(t, uint64(1), failed.Revision)
}

type legacyRaceHook struct {
	event   TransitionEvent
	release chan struct{}
}

type cacheResult struct {
	cache Cache
	err   error
}

func newLegacyWriterRaceStore(t *testing.T, kind TransitionKind, cacheID string) (*Store, *gorm.DB, <-chan legacyRaceHook, <-chan TransitionEvent) {
	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(&Cache{}))
	read := make(chan legacyRaceHook, 1)
	mutated := make(chan TransitionEvent, 1)
	var firstRead sync.Once
	store := New(db, WithTransitionObserver(func(event TransitionEvent) {
		if event.Kind != kind || event.Cache.ID != cacheID {
			return
		}
		switch event.Stage {
		case TransitionStageRead:
			firstRead.Do(func() {
				hook := legacyRaceHook{event: event, release: make(chan struct{})}
				read <- hook
				<-hook.release
			})
		case TransitionStageMutated:
			mutated <- event
		}
	}))
	return store, db, read, mutated
}

func legacyWriterRaceCache(id string, now time.Time) Cache {
	return Cache{
		ID:                  id,
		NormalizedRemoteURL: "https://example.com/" + id + ".git",
		DisplayRemoteURL:    "https://example.com/" + id + ".git",
		Status:              StatusReady,
		RefreshIntervalSecs: 60,
		CreatedBy:           "ops",
		UpdatedBy:           "ops",
		CreatedAt:           now,
		UpdatedAt:           now,
	}
}

// legacyCompleteWithoutRevision models a pre-revision MeshX UPDATE after the
// migration has added the revision column. It deliberately leaves revision at
// zero while changing build and generation fields.
func legacyCompleteWithoutRevision(t *testing.T, db *gorm.DB, id, attemptToken, generation string, now time.Time) {
	t.Helper()
	result := db.Model(&Cache{}).
		Where("id = ? AND status = ? AND lease_owner = ?", id, StatusBuilding, attemptToken).
		Updates(map[string]any{
			"status":            StatusReady,
			"active_generation": generation,
			"last_sync_at":      now,
			"last_success_at":   now,
			"next_sync_at":      nil,
			"last_commit":       "legacy-commit",
			"last_error":        "",
			"info":              `{"sizeBytes":1}`,
			"lease_owner":       "",
			"lease_until":       nil,
			"updated_at":        now,
		})
	require.NoError(t, result.Error)
	require.Equal(t, int64(1), result.RowsAffected)
}

// legacyUpdateSettingsWithoutRevision matches the old writer behavior when
// enabling refresh: it changes settings and updater fields but neither revision
// nor next_sync_at.
func legacyUpdateSettingsWithoutRevision(t *testing.T, db *gorm.DB, id string, enabled bool, intervalSeconds int, actor string, now time.Time) {
	t.Helper()
	result := db.Model(&Cache{}).Where("id = ?", id).Updates(map[string]any{
		"auto_refresh_enabled":     enabled,
		"refresh_interval_seconds": intervalSeconds,
		"updated_by":               actor,
		"updated_at":               now,
	})
	require.NoError(t, result.Error)
	require.Equal(t, int64(1), result.RowsAffected)
}

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

func awaitLegacyRaceValue[T any](t *testing.T, values <-chan T) T {
	t.Helper()
	timer := time.NewTimer(5 * time.Second)
	defer timer.Stop()
	select {
	case value := <-values:
		return value
	case <-timer.C:
		t.Fatal("timed out waiting for mixed-version transition")
		var zero T
		return zero
	}
}