package repo_imp

import (
	"go/ast"
	"go/parser"
	"go/token"
	"strings"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
)

// TestCleanDeletedWithoutBackup_UsesPluckThenDelete verifies that CleanDeletedWithoutBackup
// uses a two-step approach (Pluck IDs, then Delete by IDs) instead of the old
// single-step DELETE WHERE id IN (SELECT ... LIMIT ...) which triggers MySQL Error 1235.
//
// This test parses the source code AST to verify the fix pattern is in place,
// ensuring we don't regress to the broken SQL pattern.
func TestCleanDeletedWithoutBackup_UsesPluckThenDelete(t *testing.T) {
	fset := token.NewFileSet()
	f, err := parser.ParseFile(fset, "backup_imp.go", nil, 0)
	require.NoError(t, err, "failed to parse backup_imp.go")

	var foundPluck, foundDeleteByIDs, foundVirtualHistoryGuard bool
	var foundInSubquery bool

	ast.Inspect(f, func(n ast.Node) bool {
		call, ok := n.(*ast.CallExpr)
		if !ok {
			return true
		}
		for _, arg := range call.Args {
			if lit, ok := arg.(*ast.BasicLit); ok && strings.Contains(lit.Value, "backup_schedule_histories") {
				foundVirtualHistoryGuard = true
			}
		}
		sel, ok := call.Fun.(*ast.SelectorExpr)
		if !ok {
			return true
		}

		switch sel.Sel.Name {
		case "Pluck":
			// Check for Pluck("id", &ids) pattern
			if len(call.Args) >= 2 {
				if lit, ok := call.Args[0].(*ast.BasicLit); ok && strings.Contains(lit.Value, "id") {
					foundPluck = true
				}
			}
		case "Delete":
			// Check that the chain before Delete doesn't use a subquery with Limit
			// (the old broken pattern was: Where("id IN (?)", subQuery).Delete(...))
			foundDeleteByIDs = true
		}

		// Detect the old broken pattern: Where("id IN (?)", subQuery).Delete(...)
		// where subQuery had a Limit
		if sel.Sel.Name == "Where" && len(call.Args) >= 2 {
			if lit, ok := call.Args[0].(*ast.BasicLit); ok {
				val := lit.Value
				if strings.Contains(val, "IN (?)") || strings.Contains(val, "IN(?)") {
					// Check if the second arg is the subQuery variable (not a simple ids slice)
					if ident, ok := call.Args[1].(*ast.Ident); ok && ident.Name == "subQuery" {
						foundInSubquery = true
					}
				}
			}
		}

		return true
	})

	assert.True(t, foundPluck,
		"CleanDeletedWithoutBackup should use Pluck to query IDs first")
	assert.True(t, foundDeleteByIDs,
		"CleanDeletedWithoutBackup should have a Delete call")
	assert.False(t, foundInSubquery,
		"CleanDeletedWithoutBackup should NOT use Where('id IN (?)', subQuery) — "+
			"this generates LIMIT inside IN subquery which triggers MySQL Error 1235")
	assert.True(t, foundVirtualHistoryGuard,
		"CleanDeletedWithoutBackup should preserve schedules needed by unexpired virtual backups")
}

func TestDeleteExpiredBackupScheduleHistoryUsesPhysicalBackupTime(t *testing.T) {
	fset := token.NewFileSet()
	f, err := parser.ParseFile(fset, "backup_imp.go", nil, 0)
	require.NoError(t, err, "failed to parse backup_imp.go")

	var foundWatermarkCondition, foundPhysicalExpiryCondition bool
	for _, decl := range f.Decls {
		fn, ok := decl.(*ast.FuncDecl)
		if !ok || fn.Name.Name != "DeleteExpired" {
			continue
		}
		ast.Inspect(fn.Body, func(n ast.Node) bool {
			condition, ok := n.(*ast.BasicLit)
			if !ok {
				return true
			}
			foundWatermarkCondition = foundWatermarkCondition ||
				strings.Contains(condition.Value, "watermark.slot_time >=")
			foundPhysicalExpiryCondition = foundPhysicalExpiryCondition ||
				strings.Contains(condition.Value, "DATE_ADD(slot.backup_time")
			return true
		})
	}

	assert.True(t, foundWatermarkCondition,
		"DeleteExpired should wait until slot synchronization has passed the history")
	assert.True(t, foundPhysicalExpiryCondition,
		"DeleteExpired should use the resolved physical backup time")
}

func TestBuildBackupScheduleHistoryScopeMatchUsesEntityDefaults(t *testing.T) {
	savedConfig := config.Global()
	config.Set(&config.CentralConfig{ServerConfig: config.ServerConfig{Stage: "staging"}})
	t.Cleanup(func() {
		config.Set(savedConfig)
	})

	scopeMatch, args := buildBackupScheduleHistoryScopeMatch("slot")

	assert.Equal(t,
		"slot.cloud_provider = COALESCE(NULLIF(backup_schedule_histories.cloud_provider, ''), ?)"+
			" AND slot.region = backup_schedule_histories.region"+
			" AND slot.kubernetes = COALESCE(NULLIF(backup_schedule_histories.kubernetes, ''), "+
			"CONCAT(?, LOWER(backup_schedule_histories.region), ?))",
		scopeMatch,
	)
	prefix, suffix := entity.DefaultKubernetesNameParts()
	assert.Equal(t, []any{entity.DefaultCloudProvider, prefix, suffix}, args)
}

func TestBuildBackupScheduleHistoryRegionFilter(t *testing.T) {
	filter, args := buildBackupScheduleHistoryRegionFilter([]entity.ProviderRegion{
		{Provider: config.ProviderAWS, Region: "aws-us-east-1"},
		{Provider: config.ProviderGCP, Region: "gcp-us-west1"},
	})

	assert.Equal(t,
		"(((cloud_provider = ? OR cloud_provider = '' OR cloud_provider IS NULL) AND region = ?) OR "+
			"(cloud_provider = ? AND region = ?))",
		filter,
	)
	assert.Equal(t, []any{config.ProviderAWS, "aws-us-east-1", config.ProviderGCP, "gcp-us-west1"}, args)
}

func TestUserBackupReadsExcludeTombstonedRestorePoints(t *testing.T) {
	fset := token.NewFileSet()
	f, err := parser.ParseFile(fset, "backup_imp.go", nil, 0)
	require.NoError(t, err)

	var foundHelper, foundRecycledFilter bool
	ast.Inspect(f, func(n ast.Node) bool {
		lit, ok := n.(*ast.BasicLit)
		if !ok {
			return true
		}
		if strings.Contains(lit.Value, "bt.schedule_time = user_backups.schedule_time") {
			foundHelper = true
		}
		if strings.Contains(lit.Value, "bt.schedule_time = ub.schedule_time") {
			foundRecycledFilter = true
		}
		return true
	})

	assert.True(t, foundHelper, "persisted list/get/count reads should apply restore-point tombstones")
	assert.True(t, foundRecycledFilter, "recycled instance discovery should ignore tombstoned persisted backups")
}

func TestTombstoneCleanupPreservesUnexpiredPersistedRestorePoints(t *testing.T) {
	fset := token.NewFileSet()
	f, err := parser.ParseFile(fset, "backup_imp.go", nil, 0)
	require.NoError(t, err)

	var foundPersistedExpiryGuard bool
	ast.Inspect(f, func(n ast.Node) bool {
		lit, ok := n.(*ast.BasicLit)
		if ok && strings.Contains(lit.Value, "ub.expire_time >=") &&
			strings.Contains(lit.Value, "backup_tombstones.schedule_time") {
			foundPersistedExpiryGuard = true
		}
		return true
	})

	assert.True(t, foundPersistedExpiryGuard,
		"tombstone cleanup should retain restore points with unexpired persisted backups")
}

func TestMarkDeletedPreservesClosedHistoryBoundaries(t *testing.T) {
	fset := token.NewFileSet()
	f, err := parser.ParseFile(fset, "backup_imp.go", nil, 0)
	require.NoError(t, err, "failed to parse backup_imp.go")

	var foundCoalesce, foundClusterFilter, foundCurrentOnlyFilter bool
	for _, decl := range f.Decls {
		fn, ok := decl.(*ast.FuncDecl)
		if !ok || fn.Name.Name != "MarkDeleted" {
			continue
		}
		ast.Inspect(fn.Body, func(n ast.Node) bool {
			literal, ok := n.(*ast.BasicLit)
			if !ok {
				return true
			}
			foundCoalesce = foundCoalesce || strings.Contains(literal.Value, "COALESCE(valid_to, ?)")
			foundClusterFilter = foundClusterFilter || strings.Contains(literal.Value, "cluster_id = ?")
			foundCurrentOnlyFilter = foundCurrentOnlyFilter || strings.Contains(literal.Value, "valid_to IS NULL")
			return true
		})
	}

	assert.True(t, foundCoalesce, "MarkDeleted should preserve valid_to on already-closed histories")
	assert.True(t, foundClusterFilter, "MarkDeleted should update every history for the deleted cluster")
	assert.False(t, foundCurrentOnlyFilter, "MarkDeleted must not mark only the current history")
}