package backup

import (
	"fmt"
	"strings"
	"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"
)

// backupExportSmokeCase verifies a complete backup export for an externally
// managed test bucket. Exported target objects are removed by bucket lifecycle
// policies; the case owns its source backups and restores backup settings.
type backupExportSmokeCase struct {
	BaseCase
	backupValidator *common.BackupValidator
	cloudProvider   CloudProviderType
}

type backupExportResource struct {
	backupID uint64
	exportID uint64
}

func NewBackupExportSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &backupExportSmokeCase{cloudProvider: cloudProvider}
}

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

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

func (m *backupExportSmokeCase) 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 *backupExportSmokeCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")
	m.backupValidator = common.NewBackupValidator(m.Logger())

	var resources []*backupExportResource
	var secondaryRegion string
	dualRegionEnabledByCase := false

	// Register setting cleanup first so it runs after export and backup cleanup.
	defer func() {
		if !dualRegionEnabledByCase {
			return
		}
		m.RetryDo("cleanup: restore single-region backup setting", func() error {
			return m.backupValidator.DisableDualRegion(tidbCluster)
		}, DefaultRetryAttempts, DefaultRetryInterval).
			Expect("single-region backup setting is restored", func() error {
				return m.expectConfiguredBackupRegions(tidbCluster, []string{tidbCluster.Region})
			}, 60, 10*time.Second)
	}()

	defer func() {
		for i := len(resources) - 1; i >= 0; i-- {
			resource := resources[i]
			if resource.backupID == 0 {
				continue
			}
			backupID := resource.backupID
			m.RetryDo(fmt.Sprintf("cleanup: delete source backup %d", backupID), func() error {
				return m.backupValidator.Client().DeleteBackup(tidbCluster, backupID)
			}, DefaultRetryAttempts, DefaultRetryInterval).
				Expect(fmt.Sprintf("source backup %d is deleted", backupID), func() error {
					return m.backupValidator.IsBackupNotInList(tidbCluster, backupID)
				}, 5, time.Minute)
		}
	}()

	defer func() {
		for i := len(resources) - 1; i >= 0; i-- {
			resource := resources[i]
			if resource.exportID == 0 {
				continue
			}
			m.RetryDo(fmt.Sprintf("cleanup: cancel active export %d", resource.exportID), func() error {
				return m.cancelExportIfActive(tidbCluster, resource.backupID, resource.exportID)
			}, DefaultRetryAttempts, DefaultRetryInterval)
		}
	}()

	primary := &backupExportResource{}
	resources = append(resources, primary)
	m.Step("create a single-region source snapshot backup", func() error {
		var err error
		primary.backupID, err = m.backupValidator.TriggerSnapshotBackupAndWait(
			tidbCluster,
			GenerateBackupName("e2e-backup-export"),
			DefaultBackupTimeout,
		)
		return err
	}).
		Step("verify single-region backup export profile", func() error {
			return m.verifyExportProfile(tidbCluster, primary.backupID)
		}).
		Step("create single-region backup export", func() error {
			return m.createExport(tidbCluster, primary)
		}).
		Step("wait for single-region backup export success", func() error {
			return m.backupValidator.WaitForExportComplete(tidbCluster, primary.exportID, 60*time.Minute)
		}).
		Step("enable dual-region backup", func() error {
			var err error
			secondaryRegion, err = m.backupValidator.GetSecondaryRegion(tidbCluster, false)
			if err != nil {
				return err
			}
			if err := m.backupValidator.EnableDualRegion(tidbCluster, secondaryRegion); err != nil {
				return err
			}
			dualRegionEnabledByCase = true
			return m.backupValidator.WaitForBackupSettingReady(tidbCluster, DefaultBackupTimeout)
		}).
		Step("verify dual-region backup setting", func() error {
			return m.expectConfiguredBackupRegions(tidbCluster, []string{tidbCluster.Region, secondaryRegion})
		})

	dualRegion := &backupExportResource{}
	resources = append(resources, dualRegion)
	m.Step("create a dual-region source snapshot backup", func() error {
		var err error
		dualRegion.backupID, err = m.backupValidator.TriggerSnapshotBackupAndWait(
			tidbCluster,
			GenerateBackupName("e2e-dual-region-backup-export"),
			DefaultBackupTimeout,
		)
		return err
	}).
		Step("verify snapshot backup covers both regions", func() error {
			return m.expectBackupRegions(
				tidbCluster,
				dualRegion.backupID,
				[]string{tidbCluster.Region, secondaryRegion},
			)
		}).
		Step("verify dual-region backup export profile", func() error {
			return m.verifyExportProfile(tidbCluster, dualRegion.backupID)
		}).
		Step("create dual-region backup export", func() error {
			return m.createExport(tidbCluster, dualRegion)
		}).
		Step("wait for dual-region backup export success", func() error {
			return m.backupValidator.WaitForExportComplete(tidbCluster, dualRegion.exportID, 60*time.Minute)
		})
}

func (m *backupExportSmokeCase) verifyExportProfile(tidbCluster *TiDBDedicatedEntity, backupID uint64) error {
	profile, err := m.backupValidator.Client().GetBackupExportProfile(tidbCluster, backupID)
	if err != nil {
		return err
	}
	switch m.cloudProvider {
	case AWS:
		if strings.TrimSpace(profile.S3RoleArn) == "" {
			return fmt.Errorf("backup export profile has an empty AWS role ARN")
		}
	case GCP:
		if strings.TrimSpace(profile.GcpServiceAccount) == "" {
			return fmt.Errorf("backup export profile has an empty GCP service account")
		}
	default:
		return fmt.Errorf("backup export is not supported for provider %s", m.cloudProvider)
	}
	return nil
}

func (m *backupExportSmokeCase) createExport(
	tidbCluster *TiDBDedicatedEntity,
	resource *backupExportResource,
) error {
	req, err := m.buildExportRequest()
	if err != nil {
		return err
	}
	resp, err := m.backupValidator.Client().CreateBackupExport(tidbCluster, resource.backupID, req)
	if err != nil {
		return err
	}
	if resp.ExportId == 0 {
		return fmt.Errorf("create backup export returned an empty export ID")
	}
	resource.exportID = resp.ExportId
	m.Logger().Info("created backup export %d for backup %d", resource.exportID, resource.backupID)
	return nil
}

func (m *backupExportSmokeCase) expectConfiguredBackupRegions(
	tidbCluster *TiDBDedicatedEntity,
	expected []string,
) error {
	setting, err := m.backupValidator.Client().QueryBackupSetting(tidbCluster)
	if err != nil {
		return err
	}
	if setting.Status != string(br_model.DataflowBackupSettingStatusNormal) {
		return fmt.Errorf("backup setting is not ready: %s", setting.Status)
	}
	actual := make([]string, 0, len(setting.Regions))
	for _, region := range setting.Regions {
		if region != nil {
			actual = append(actual, region.Name)
		}
	}
	return validateRegionNames(actual, expected)
}

func (m *backupExportSmokeCase) expectBackupRegions(
	tidbCluster *TiDBDedicatedEntity,
	backupID uint64,
	expected []string,
) error {
	resp, err := m.backupValidator.Client().ListBackups(tidbCluster, 1, 100)
	if err != nil {
		return err
	}
	for _, backup := range resp.Items {
		if backup.Id != backupID {
			continue
		}
		actual := make([]string, 0, len(backup.Regions))
		for _, region := range backup.Regions {
			actual = append(actual, region.Name)
		}
		return validateRegionNames(actual, expected)
	}
	return fmt.Errorf("backup %d not found in cluster %s", backupID, tidbCluster.ClusterID)
}

func validateRegionNames(actual, expected []string) error {
	actualSet := make(map[string]struct{}, len(actual))
	for _, region := range actual {
		if strings.TrimSpace(region) == "" {
			return fmt.Errorf("actual backup regions contain an empty region: %v", actual)
		}
		actualSet[region] = struct{}{}
	}
	if len(actualSet) != len(expected) {
		return fmt.Errorf("backup regions mismatch: expected %v, actual %v", expected, actual)
	}
	for _, region := range expected {
		if _, ok := actualSet[region]; !ok {
			return fmt.Errorf("backup regions mismatch: expected %v, actual %v", expected, actual)
		}
	}
	return nil
}

func (m *backupExportSmokeCase) buildExportRequest() (*consoleapi.CreateBackupExportRequest, error) {
	return buildBackupExportRequest(m.cloudProvider, config.GlobalConfig.BackupExport)
}

func buildBackupExportRequest(
	cloudProvider CloudProviderType,
	targets config.BackupExportConfig,
) (*consoleapi.CreateBackupExportRequest, error) {
	switch cloudProvider {
	case AWS:
		target := targets.AWS
		if !strings.HasPrefix(target.TargetURI, "s3://") || !strings.HasSuffix(target.TargetURI, "/") {
			return nil, fmt.Errorf("invalid AWS backup export target URI %q", target.TargetURI)
		}
		if strings.TrimSpace(target.Region) == "" {
			return nil, fmt.Errorf("AWS backup export target region is empty")
		}
		return &consoleapi.CreateBackupExportRequest{
			S3: &consoleapi.BackupExportS3Request{
				BucketURI: target.TargetURI,
				Region:    target.Region,
			},
		}, nil
	case GCP:
		targetURI := targets.GCP.TargetURI
		if strings.HasPrefix(targetURI, "gcs://") {
			targetURI = "gs://" + strings.TrimPrefix(targetURI, "gcs://")
		}
		if !strings.HasPrefix(targetURI, "gs://") || !strings.HasSuffix(targetURI, "/") {
			return nil, fmt.Errorf("invalid GCP backup export target URI %q", targetURI)
		}
		return &consoleapi.CreateBackupExportRequest{
			Gcs: &consoleapi.BackupExportGCSRequest{BucketURI: targetURI},
		}, nil
	default:
		return nil, fmt.Errorf("backup export is not supported for provider %s", cloudProvider)
	}
}

func (m *backupExportSmokeCase) cancelExportIfActive(
	tidbCluster *TiDBDedicatedEntity,
	backupID uint64,
	exportID uint64,
) error {
	resp, err := m.backupValidator.Client().ListBackupExports(tidbCluster, 1, 100)
	if err != nil {
		return err
	}
	for _, export := range resp.Items {
		if export.Id != exportID {
			continue
		}
		switch export.Status {
		case string(br_model.DataflowBackupExportStatusPENDING), string(br_model.DataflowBackupExportStatusRUNNING):
			return m.backupValidator.Client().CancelBackupExport(tidbCluster, backupID, exportID)
		case string(br_model.DataflowBackupExportStatusSUCCESS),
			string(br_model.DataflowBackupExportStatusFAILED),
			string(br_model.DataflowBackupExportStatusCANCELED):
			return nil
		default:
			return fmt.Errorf("export %d has unknown status %q", exportID, export.Status)
		}
	}
	return fmt.Errorf("export %d not found for cluster %s", exportID, tidbCluster.ClusterID)
}