package crrf

import (
	"fmt"
	"time"

	. "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"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cluster/client/cluster_service"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cluster_non_gen"
)

// crrfRestoreSmokeCase tests the CRRF restore flow.
// It covers:
//   - Verify CRRF is enabled with a valid checkpoint
//   - Trigger a restore with is_crrf=true, sourcing data from the secondary region bucket
//   - Verify the restore job appears in the list with the correct is_crrf flag
//   - Wait for the restore to complete
//   - Verify the restored cluster is created successfully
//
// Prerequisites:
//   - The source cluster must be AWS-based with CRRF enabled
//   - The CRRF checkpoint must be non-empty (log backup running in secondary region)
type crrfRestoreSmokeCase struct {
	BaseCase
	backupValidator   *common.BackupValidator
	clusterApiWrapper *cluster_non_gen.ClusterClient
	cloudProvider     CloudProviderType
}

// NewCRRFRestoreSmokeCase creates a new CRRF restore smoke test case.
func NewCRRFRestoreSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &crrfRestoreSmokeCase{cloudProvider: cloudProvider}
}

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

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

func (m *crrfRestoreSmokeCase) 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 *crrfRestoreSmokeCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")
	restoreClusterName := GenerateClusterName("e2e-crrf-restore")

	m.backupValidator = common.NewBackupValidator(m.Logger())
	m.clusterApiWrapper = cluster_non_gen.NewClient(m.Logger())

	// CRRF is AWS-only
	if m.cloudProvider != AWS {
		m.Logger().Info("skipping CRRF restore smoke test for non-AWS provider %s", m.cloudProvider)
		return
	}

	var (
		restoreID         uint64
		restoredClusterID string
		backupID          uint64
		restoreTs         int64
		secondaryRegion   string
	)

	// Cleanup: delete the restored cluster regardless of test outcome
	defer func() {
		if restoredClusterID != "" {
			m.RetryDo("cleanup: delete restored cluster", func() error {
				m.Logger().Info("deleting CRRF-restored cluster id=%s", restoredClusterID)
				return m.clusterApiWrapper.DeleteCluster(tidbCluster.OrgID, tidbCluster.ProjectID, restoredClusterID)
			}, 3, time.Minute)
		} else if restoreClusterName != "" {
			// Fallback: find by name if cluster ID wasn't captured
			m.RetryDo("cleanup: delete restored cluster by name", func() error {
				cli := consoleapi.ClusterClient().ClusterService
				pageSize := int32(100)
				params := cluster_service.NewClusterServiceListClustersParams().
					WithProjectID(tidbCluster.ProjectID).
					WithPageSize(&pageSize)

				resp, err := cli.ClusterServiceListClusters(params)
				if err != nil {
					return err
				}
				for _, c := range resp.Payload.Clusters {
					if c.DisplayName != nil && *c.DisplayName == restoreClusterName {
						m.Logger().Info("found cluster %s (id=%s), deleting", restoreClusterName, c.ClusterID)
						return m.clusterApiWrapper.DeleteCluster(tidbCluster.OrgID, tidbCluster.ProjectID, c.ClusterID)
					}
				}
				m.Logger().Info("cluster %s not found, skipping cleanup", restoreClusterName)
				return nil
			}, 3, time.Minute)
		}
	}()

	m.
		Step("ensure CRRF is enabled", func() error {
			resp, err := m.backupValidator.Client().QueryBackupSetting(tidbCluster)
			if err != nil {
				return err
			}
			if resp.CrrfStatus != "enabled" {
				if err := m.enableCRRFAutomatically(tidbCluster); err != nil {
					return err
				}
			}
			return nil
		}).
		Expect("CRRF status is enabled in backup setting", func() error {
			return m.backupValidator.ExpectCRRFStatus(tidbCluster, "enabled")
		}, 30, 30*time.Second).
		Step("resolve secondary region from backup setting", func() error {
			resp, err := m.backupValidator.Client().QueryBackupSetting(tidbCluster)
			if err != nil {
				return err
			}
			secondaryRegion, err = GetSecondaryRegionFromBackupSetting(tidbCluster.Region, resp.Regions)
			return err
		}).
		// Step 2: Get the latest backup for restore
		Step("get backup for CRRF restore", func() error {
			backup, err := m.backupValidator.GetLatestBackup(tidbCluster)
			if err != nil {
				return err
			}
			backupID = backup.Id
			m.Logger().Info("using backup id=%d for CRRF restore", backupID)
			return nil
		}).
		// Step 3: Wait for PiTR time ranges to be available
		Expect("PiTR time ranges are available", func() error {
			return m.backupValidator.HasPiTRTimeRanges(tidbCluster)
		}, 10, 1*time.Minute). // May need to wait for log backup to accumulate
		// Step 4: Select a PiTR timestamp from available ranges
		Step("select CRRF restore timestamp", func() error {
			resp, err := m.backupValidator.Client().ListPiTRTimeRanges(tidbCluster)
			if err != nil {
				return err
			}
			if len(resp.Ranges) == 0 {
				return fmt.Errorf("no PiTR time ranges available for CRRF restore")
			}
			firstRange := resp.Ranges[0]
			m.Logger().Info("PiTR range for CRRF: start=%s, end=%s", firstRange.StartTime, firstRange.EndTime)

			endTime, err := time.Parse(time.RFC3339, firstRange.EndTime)
			if err != nil {
				return fmt.Errorf("failed to parse end time %s: %w", firstRange.EndTime, err)
			}

			// Use 1 minute before the end of the available range as a safe restore point
			restoreTs = endTime.Add(-1 * time.Minute).Unix()
			m.Logger().Info("CRRF restore timestamp: %d (%s)", restoreTs, time.Unix(restoreTs, 0).Format(time.RFC3339))
			return nil
		}).

		// Step 5: Trigger restore in secondary region.
		RetryDo("trigger CRRF restore", func() error {
			req := BuildCRRFRestoreRequestForRegion(m.clusterApiWrapper, tidbCluster.OrgID, tidbCluster.ProjectID, string(m.cloudProvider), restoreClusterName, secondaryRegion, restoreTs)
			resp, err := m.backupValidator.Client().RestoreBackup(tidbCluster, backupID, req)
			if err != nil {
				return err
			}
			restoreID = resp.Id
			restoredClusterID = resp.ClusterId
			m.Logger().Info("triggered CRRF restore: restore_id=%d, target_cluster=%s (id=%s)",
				restoreID, restoreClusterName, restoredClusterID)
			return nil
		}, 3, time.Minute).

		// Step 6: Verify the restore job appears in the list
		Expect("CRRF restore is in list", func() error {
			resp, err := m.backupValidator.Client().ListRestores(tidbCluster, 1, 100)
			if err != nil {
				return err
			}
			for _, r := range resp.Items {
				if r.Id == restoreID {
					m.Logger().Info("CRRF restore %d found in list: status=%s, mode=%s",
						restoreID, r.Status, r.RestoreMode)
					return nil
				}
			}
			return fmt.Errorf("CRRF restore %d not found in list", restoreID)
		}, 5, 30*time.Second).

		// Step 7: Wait for the restore to complete
		Step("wait for CRRF restore to complete", func() error {
			return m.backupValidator.WaitForRestoreComplete(tidbCluster, restoreID, 60*time.Minute)
		})
}

func (m *crrfRestoreSmokeCase) enableCRRFAutomatically(tidbCluster *TiDBDedicatedEntity) error {
	secondaryRegion, err := m.backupValidator.GetSecondaryRegion(tidbCluster, true)
	if err != nil {
		return err
	}
	if err := m.backupValidator.EnableDualRegion(tidbCluster, secondaryRegion); err != nil {
		return err
	}
	if err := m.backupValidator.WaitForBackupSettingReady(tidbCluster, 5*time.Minute); err != nil {
		return err
	}
	if err := m.backupValidator.EnablePiTR(tidbCluster); err != nil {
		return err
	}
	_, err = m.backupValidator.TriggerSnapshotBackupAndWait(
		tidbCluster,
		GenerateClusterName("crrf-trigger"),
		30*time.Minute,
	)
	return err
}