package backup

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"
	br_model "github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/br/model"
	"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"
)

// pitrRestoreSmokeCase tests PiTR (Point-in-Time Recovery) restore functionality.
// It covers:
// - Ensure PiTR is enabled and has time ranges available
// - Select a PiTR timestamp within the available range
// - Trigger a PiTR restore to create a new cluster
// - Wait for restore to complete
type pitrRestoreSmokeCase struct {
	BaseCase
	backupValidator   *common.BackupValidator
	clusterApiWrapper *cluster_non_gen.ClusterClient
	cloudProvider     CloudProviderType
}

// NewPiTRRestoreSmokeCase creates a new PiTR restore smoke test case.
func NewPiTRRestoreSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &pitrRestoreSmokeCase{cloudProvider: cloudProvider}
}

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

func (m *pitrRestoreSmokeCase) Tags() map[string]string {
	return map[string]string{TagKeyDomain: TagValueDomainBR}
}

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

	// Initialize validator
	m.backupValidator = common.NewBackupValidator(m.Logger())

	// init reusable api client
	m.clusterApiWrapper = cluster_non_gen.NewClient(m.Logger())

	var restoreID uint64
	var restoredClusterID string
	var backupID uint64
	var restoreTs int64

	// Cleanup
	defer func() {
		// Clean up restored cluster
		if restoredClusterID != "" {
			m.RetryDo("cleanup: delete restored cluster", func() error {
				m.Logger().Info("deleting restored cluster with ID: %s", restoredClusterID)
				return m.clusterApiWrapper.DeleteCluster(tidbCluster.OrgID, tidbCluster.ProjectID, restoredClusterID)
			}, 3, time.Minute)
		} else if restoreClusterName != "" {
			// Fallback cleanup by name if ID is missing (e.g. restore failed early)
			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
				}

				var clusterID string
				for _, c := range resp.Payload.Clusters {
					if c.DisplayName != nil && *c.DisplayName == restoreClusterName {
						clusterID = c.ClusterID
						break
					}
				}

				if clusterID != "" {
					m.Logger().Info("found restored cluster %s (id: %s), deleting it", restoreClusterName, clusterID)
					return m.clusterApiWrapper.DeleteCluster(tidbCluster.OrgID, tidbCluster.ProjectID, clusterID)
				}

				m.Logger().Info("restored cluster %s not found, skip deletion", restoreClusterName)
				return nil
			}, 3, time.Minute)
		}
	}()

	m.
		// Step 1: Ensure auto backup is enabled (PiTR requires snapshot backup base)
		Step("ensure auto backup is enabled", func() error {
			err := m.backupValidator.IsAutoBackupEnabled(tidbCluster)
			if err != nil {
				m.Logger().Info("auto backup not enabled, enabling daily backup")
				return m.backupValidator.EnableDailyAutoBackup(tidbCluster, "03:00", 7)
			}
			return nil
		}).
		// Step 1: Ensure PiTR is enabled
		Step("ensure PiTR is enabled", func() error {
			err := m.backupValidator.IsPiTREnabled(tidbCluster)
			if err != nil {
				m.Logger().Info("PiTR not enabled, enabling it")
				if enableErr := m.backupValidator.EnablePiTR(tidbCluster); enableErr != nil {
					return enableErr
				}

				m.Logger().Info("PiTR is being enabled, trigger a manual backup and wait for it to be active")
				backupName := fmt.Sprintf("pitr-restore-init-backup-%d", time.Now().Unix())
				_, err := m.backupValidator.TriggerSnapshotBackupAndWait(tidbCluster, backupName, 20*time.Minute)
				return err
			}
			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)
		}, 30, 2*time.Minute). // May need to wait for log backup to accumulate

		// Step 4: Get the latest backup for PiTR restore
		Step("get backup for PiTR restore", func() error {
			backup, err := m.backupValidator.GetLatestBackup(tidbCluster)
			if err != nil {
				return err
			}
			backupID = backup.Id
			m.Logger().Info("using backup %d for PiTR restore", backupID)
			return nil
		}).

		// Step 5: Get PiTR time range and select restore timestamp
		Step("select PiTR 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")
			}

			// Select the first available time range
			firstRange := resp.Ranges[0]
			m.Logger().Info("PiTR range: start=%s, end=%s", firstRange.StartTime, firstRange.EndTime)

			// Parse end time and select a restore point 1 minute before the end
			// to ensure it's within the valid range
			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 end time minus 1 minute as restore point to ensure it's within range
			restoreTs = endTime.Add(-1 * time.Minute).Unix()
			m.Logger().Info("calculated restore timestamp: %d (%s)", restoreTs, time.Unix(restoreTs, 0).Format(time.RFC3339))
			return nil
		}).

		// Step 6: Trigger PiTR restore
		RetryDo("trigger PiTR restore", func() error {
			req := &consoleapi.RestoreBackupRequest{
				ClusterName:  restoreClusterName,
				Region:       tidbCluster.Region,
				BackupRegion: tidbCluster.Region,
				Description:  "E2E test PiTR restored cluster",
				Password:     DefaultClusterPassword,
				RestoreMode:  string(br_model.DataflowRestoreModeRESTOREMODEPITR),
				RestoredTs:   restoreTs,
				Port:         DefaultClusterPort,
				Components:   BuildMinimalClusterComponentsForRegion(m.clusterApiWrapper, tidbCluster.OrgID, tidbCluster.ProjectID, string(m.cloudProvider), tidbCluster.Region),
			}
			resp, err := m.backupValidator.Client().RestoreBackup(tidbCluster, backupID, req)
			if err != nil {
				return err
			}
			restoreID = resp.Id
			restoredClusterID = resp.ClusterId
			m.Logger().Info("triggered PiTR restore with ID: %d, target cluster: %s (id: %s)",
				restoreID, restoreClusterName, restoredClusterID)
			return nil
		}, 3, time.Minute).

		// Step 7: Verify restore appears in list
		Expect("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("restore %d found in list, status: %s, mode: %s",
						restoreID, r.Status, r.RestoreMode)
					return nil
				}
			}
			return fmt.Errorf("restore %d not found in list", restoreID)
		}, 5, time.Second*30).

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