package export

import (
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/essential/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/essential"
)

type ExportLifecycleSmokeCase struct {
	cases.BaseCase
	provider CloudProviderType
}

func NewExportLifecycleSmokeCase(provider CloudProviderType) cases.E2ECase {
	return &ExportLifecycleSmokeCase{provider: provider}
}
func (c *ExportLifecycleSmokeCase) Name() string {
	return fmt.Sprintf("%sEssentialV1ExportLifecycleSmoke", c.provider)
}
func (c *ExportLifecycleSmokeCase) Tags() map[string]string {
	return map[string]string{cases.TagKeyDomain: config.DomainExport, cases.TagKeyAPIModes: common.APIModesBoth}
}
func (c *ExportLifecycleSmokeCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	region := config.GlobalConfig.DefaultRegionOf(c.provider)
	return map[string]DependencyRequirement{
		"essential": &TiDBServerlessRequirement{
			Dedicated: true,
			TiDBServerlessDimensions: TiDBServerlessDimensions{
				CloudProvider: c.provider,
				Region:        region,
				ServicePlan:   essential.ServicePlanEssential,
			},
		},
	}
}

func (c *ExportLifecycleSmokeCase) Run() {
	cluster := c.TiDBServerless("essential")
	c.Step("verify Essential v1 cluster", func() error { return common.ValidateCluster(cluster) })
	c.Step("write and verify export data", func() error {
		_, err := common.WriteMarker(cluster, c.Name()+"-export-"+common.UniqueRunSuffix())
		return err
	})
	target, err := exportTarget(c.provider)
	c.Step("validate export target configuration", func() error { return err })
	prefix := strings.TrimRight(target.TargetURI, "/") + "/" + strings.ToLower(c.Name()) + "-" + common.UniqueRunSuffix() + "/"
	c.Step("validate export destination URI", func() error { return essential.ValidateExportURI(prefix, c.provider) })
	req := essential.ExportRequest{OrgID: cluster.OrgID, ProjectID: cluster.ProjectID, ClusterID: cluster.ClusterID, TargetURI: prefix, Region: target.Region, Endpoint: target.Endpoint, RoleARN: target.RoleARN, AccessKeyID: target.AccessKeyID, AccessKeySecret: target.AccessKeySecret, Provider: c.provider, DisplayName: c.Name()}
	client := essential.NewExportClient(c.Logger())
	var id string
	// AWS staging currently denies runner access to the target bucket. Keep
	// object inspection enabled for AliCloud, but skip it for AWS until the IAM
	// fixture regression is fixed.
	if inspectExportTargetObjects(c.provider) {
		// Defers are registered in reverse dependency order. Each cleanup phase
		// is independent, so FailNow/Goexit in task cleanup still runs object
		// cleanup.
		defer func() {
			c.Step("delete and verify export object prefix", func() error {
				return essential.DeleteObjectPrefix(c.CleanupContext(), req)
			})
		}()
	}
	defer func() {
		if id != "" {
			c.Expect("export cleanup deletion completes", func() error {
				exists, cleanupErr := client.Exists(c.CleanupContext(), req, id)
				if cleanupErr != nil {
					return cleanupErr
				}
				if exists {
					return fmt.Errorf("export %s still exists during cleanup", id)
				}
				return nil
			}, 30, 15*time.Second)
		}
	}()
	defer func() {
		if id != "" {
			c.RetryDo("delete export", func() error { return client.Delete(c.CleanupContext(), req, id) }, 3, time.Minute)
		}
	}()
	c.RetryDo("create export", func() error { var createErr error; id, createErr = client.Create(c.Context(), req); return createErr }, 3, time.Minute)
	c.Expect("export completes", func() error {
		state, getErr := client.Get(c.Context(), req, id)
		if getErr != nil {
			return getErr
		}
		if err := common.ValidateExportState(state); err != nil {
			return err
		}
		return nil
	}, 120, 30*time.Second)
	if inspectExportTargetObjects(c.provider) {
		c.Step("verify non-empty export objects", func() error {
			return essential.VerifyNonEmptyObjects(c.Context(), req)
		})
	}
	c.Step("delete completed export", func() error {
		return client.Delete(c.Context(), req, id)
	})
	c.Expect("export deleted", func() error {
		exists, err := client.Exists(c.Context(), req, id)
		if err != nil {
			return err
		}
		if exists {
			return fmt.Errorf("export %s still exists", id)
		}
		return nil
	}, 30, 15*time.Second)
	id = ""
}