package crrf

import (
	"errors"
	"fmt"
	"sort"
	"time"

	"github.com/avast/retry-go/v4"
	. "github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/dedicated/common"
	"github.com/tidbcloud/dataflow-service-e2e/internal/config"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi"
)

// crrfEnableDisableSmokeCase tests CRRF auto lifecycle through backup-setting APIs.
type crrfEnableDisableSmokeCase struct {
	BaseCase
	backupValidator *common.BackupValidator
	cloudProvider   CloudProviderType
}

// NewCRRFEnableDisableSmokeCase creates a new CRRF enable/disable smoke test case.
// CRRF is AWS-only; other cloud providers will skip gracefully.
func NewCRRFEnableDisableSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &crrfEnableDisableSmokeCase{cloudProvider: cloudProvider}
}

func (m *crrfEnableDisableSmokeCase) Name() string {
	return fmt.Sprintf("%sCRRFEnableDisableSmoke", m.cloudProvider)
}

func (m *crrfEnableDisableSmokeCase) Tags() map[string]string {
	return map[string]string{TagKeyDomain: TagValueDomainCRRF}
}

func (m *crrfEnableDisableSmokeCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: true,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: m.cloudProvider,
				Region:        config.GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
	}
}

func (m *crrfEnableDisableSmokeCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")

	m.backupValidator = common.NewBackupValidator(m.Logger())

	// Only run on AWS clusters since CRRF depends on S3 CRR
	if m.cloudProvider != AWS {
		m.Logger().Info("skipping CRRF enable/disable smoke test for non-AWS provider %s", m.cloudProvider)
		return
	}

	var (
		secondaryRegion       string
		originalPiTREnabled   bool
		originalRegions       []string
		originalStateCaptured bool
	)

	defer func() {
		if !originalStateCaptured {
			return
		}

		var (
			cleanupErr           error
			restoreApplied       bool
			shouldRestorePiTR    bool
			shouldRestoreRegions bool
		)

		runCleanupWithRetry := func(action string, fn func() error) error {
			m.Logger().Info("BEGIN-cleanup '%s'", action)
			err := retry.Do(
				fn,
				retry.Attempts(3),
				retry.DelayType(retry.FixedDelay),
				retry.Delay(20*time.Second),
			)
			if err != nil {
				m.Logger().Error("FAIL-cleanup '%s', error: %s", action, err.Error())
				return err
			}
			m.Logger().Info("END-cleanup '%s'", action)
			return nil
		}

		waitForRestoreConvergence := func(timeout time.Duration) error {
			deadline := time.Now().Add(timeout)
			pollInterval := 10 * time.Second
			consecutiveReadyMatches := 0

			for time.Now().Before(deadline) {
				resp, err := m.backupValidator.Client().QueryBackupSetting(tidbCluster)
				if err != nil {
					consecutiveReadyMatches = 0
					m.Logger().Warn("cleanup: query backup setting while waiting for restore convergence failed: %v", err)
					time.Sleep(pollInterval)
					continue
				}

				currentPiTREnabled := resp.ContinuousBackupSchedule != nil && resp.ContinuousBackupSchedule.Enable
				regionsMatch := equalStringSet(regionNamesFromResponse(resp.Regions), originalRegions)
				pitrMatch := currentPiTREnabled == originalPiTREnabled
				ready := resp.Status == "Normal"

				if pitrMatch && regionsMatch && ready {
					consecutiveReadyMatches++
					if consecutiveReadyMatches >= 2 {
						return nil
					}
				} else {
					consecutiveReadyMatches = 0
				}

				m.Logger().Info("cleanup: waiting for restore convergence, status=%s pitrMatch=%t regionsMatch=%t",
					resp.Status, pitrMatch, regionsMatch)
				time.Sleep(pollInterval)
			}

			return fmt.Errorf("timeout waiting for backup setting restore convergence")
		}

		resp, err := m.backupValidator.Client().QueryBackupSetting(tidbCluster)
		if err != nil {
			cleanupErr = errors.Join(cleanupErr, fmt.Errorf("cleanup: query backup setting before restore: %w", err))
			shouldRestorePiTR = true
			shouldRestoreRegions = true
		} else {
			currentPiTREnabled := resp.ContinuousBackupSchedule != nil && resp.ContinuousBackupSchedule.Enable
			shouldRestorePiTR = currentPiTREnabled != originalPiTREnabled
			shouldRestoreRegions = !equalStringSet(regionNamesFromResponse(resp.Regions), originalRegions)
		}

		if !shouldRestorePiTR && !shouldRestoreRegions {
			if cleanupErr != nil {
				m.Step("cleanup: fail on unrecoverable errors", func() error {
					return cleanupErr
				})
			}
			return
		}

		restoreRequests := buildRestoreBackupSettingRequests(
			originalPiTREnabled,
			originalRegions,
			shouldRestorePiTR,
			shouldRestoreRegions,
		)
		for idx, req := range restoreRequests {
			requestSeq := idx + 1
			waitAction := fmt.Sprintf("cleanup: wait backup setting ready before restore request #%d", requestSeq)
			if err := runCleanupWithRetry(waitAction, func() error {
				return m.backupValidator.WaitForBackupSettingReady(tidbCluster, 2*time.Minute)
			}); err != nil {
				cleanupErr = errors.Join(cleanupErr, fmt.Errorf("%s: %w", waitAction, err))
				continue
			}

			restoreAction := fmt.Sprintf("cleanup: restore backup setting request #%d", requestSeq)
			if err := runCleanupWithRetry(restoreAction, func() error {
				return m.backupValidator.Client().ModifyBackupSetting(tidbCluster, req)
			}); err != nil {
				cleanupErr = errors.Join(cleanupErr, fmt.Errorf("%s: %w", restoreAction, err))
			} else {
				restoreApplied = true
			}
		}

		if restoreApplied {
			if err := waitForRestoreConvergence(3 * time.Minute); err != nil {
				cleanupErr = errors.Join(cleanupErr, fmt.Errorf("cleanup: verify restore convergence: %w", err))
			}
		}

		if cleanupErr != nil {
			m.Step("cleanup: fail on unrecoverable errors", func() error {
				return cleanupErr
			})
		}
	}()

	m.Step("capture original backup setting", func() error {
		resp, err := m.backupValidator.Client().QueryBackupSetting(tidbCluster)
		if err != nil {
			return err
		}
		originalPiTREnabled = resp.ContinuousBackupSchedule != nil && resp.ContinuousBackupSchedule.Enable
		originalRegions = regionNamesFromResponse(resp.Regions)
		originalStateCaptured = true
		return nil
	})

	m.
		Step("enable dual-region", func() error {
			var err error
			secondaryRegion, err = m.backupValidator.GetSecondaryRegion(tidbCluster, false)
			if err != nil {
				return err
			}
			return m.backupValidator.EnableDualRegion(tidbCluster, secondaryRegion)
		}).
		Expect("dual-region reflected in backup setting", func() error {
			resp, err := m.backupValidator.Client().QueryBackupSetting(tidbCluster)
			if err != nil {
				return err
			}
			confirmedSecondary, err := GetSecondaryRegionFromBackupSetting(tidbCluster.Region, resp.Regions)
			if err != nil {
				return err
			}
			if confirmedSecondary != secondaryRegion {
				return fmt.Errorf("secondary region mismatch: expected %s, got %s", secondaryRegion, confirmedSecondary)
			}
			return nil
		}, 10, 10*time.Second).
		Step("wait backup setting ready after dual-region enable", func() error {
			return m.backupValidator.WaitForBackupSettingReady(tidbCluster, 5*time.Minute)
		}).
		Step("enable PiTR", func() error {
			return m.backupValidator.EnablePiTR(tidbCluster)
		}).
		Step("trigger snapshot backup for CRRF activation", func() error {
			_, err := m.backupValidator.TriggerSnapshotBackupAndWait(
				tidbCluster,
				GenerateClusterName("crrf-trigger"),
				30*time.Minute,
			)
			return err
		}).
		Expect("CRRF auto enabled", func() error {
			return m.backupValidator.ExpectCRRFStatus(tidbCluster, "enabled")
		}, 30, 30*time.Second).
		Step("wait backup setting ready before disabling PiTR", func() error {
			return m.backupValidator.WaitForBackupSettingReady(tidbCluster, 5*time.Minute)
		}).
		Step("disable PiTR to auto disable CRRF", func() error {
			return m.backupValidator.DisablePiTR(tidbCluster)
		}).
		Expect("CRRF auto disabled", func() error {
			return m.backupValidator.ExpectCRRFStatus(tidbCluster, "disabled")
		}, 30, 20*time.Second)
}

func regionNamesFromResponse(regions []*consoleapi.RegionInfoResp) []string {
	regionNames := make([]string, 0, len(regions))
	for _, region := range regions {
		if region != nil && region.Name != "" {
			regionNames = append(regionNames, region.Name)
		}
	}
	return regionNames
}

func regionRequestsFromNames(regionNames []string) []*consoleapi.RegionInfoRequest {
	regions := make([]*consoleapi.RegionInfoRequest, 0, len(regionNames))
	for _, regionName := range regionNames {
		regions = append(regions, &consoleapi.RegionInfoRequest{Name: regionName})
	}
	return regions
}

func equalStringSet(left, right []string) bool {
	if len(left) != len(right) {
		return false
	}
	leftCopy := append([]string(nil), left...)
	rightCopy := append([]string(nil), right...)
	sort.Strings(leftCopy)
	sort.Strings(rightCopy)
	for i := range leftCopy {
		if leftCopy[i] != rightCopy[i] {
			return false
		}
	}
	return true
}

func buildRestoreBackupSettingRequests(
	originalPiTREnabled bool,
	originalRegions []string,
	shouldRestorePiTR bool,
	shouldRestoreRegions bool,
) []*consoleapi.ModifyBackupSettingRequest {
	requests := make([]*consoleapi.ModifyBackupSettingRequest, 0, 2)
	if shouldRestorePiTR {
		requests = append(requests, &consoleapi.ModifyBackupSettingRequest{
			ContinuousBackupSchedule: &consoleapi.ContinuousBackupScheduleRequest{
				Enable: originalPiTREnabled,
			},
		})
	}
	if shouldRestoreRegions {
		requests = append(requests, &consoleapi.ModifyBackupSettingRequest{
			Regions: regionRequestsFromNames(originalRegions),
		})
	}
	return requests
}