package case_driver

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/avast/retry-go/v4"
	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/dataflow-service-e2e/cases/utils"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/config"
	"github.com/tidbcloud/dataflow-service-e2e/internal/log"
	"github.com/tidbcloud/dataflow-service-e2e/internal/misc"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
	clustercli "github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/consoleapi/cluster_non_gen"
)

type tiDBPremiumDependency struct {
	lgr    log.SimpleLogger
	client *clustercli.ClusterClient
}

func NewTiDBPremiumDependency(lgr log.SimpleLogger) Dependency {
	return &tiDBPremiumDependency{
		lgr:    lgr,
		client: clustercli.NewClient(lgr),
	}
}
func (t *tiDBPremiumDependency) Type() string {
	return TiDBPremium
}

func (t *tiDBPremiumDependency) GetExistingUsableEntities() ([]DependencyEntity, error) {
	var original []TiDBPremiumEntity

	// enrich tidb cluster info
	for _, id := range GlobalConfig.TiDBPremiumClusterIds {
		var resp *clustercli.ClusterResponse
		ok := retry.Do(func() error {
			var err error
			resp, err = t.client.GetCluster(GlobalConfig.OrgID, id)
			if err != nil {
				return err
			}
			return nil
		}, retry.DelayType(retry.FixedDelay), retry.Delay(20*time.Second), retry.Attempts(3))
		if ok != nil {
			return nil, fmt.Errorf("enrich tidb cluster info failed when calling GetCluster, clusterId:%s, error: %s", id, ok.Error())
		}
		original = append(original, TiDBPremiumEntity{
			TiDBPremiumDimensions: TiDBPremiumDimensions{
				CloudProvider: CloudProviderType(resp.CloudProvider),
				Region:        regionId2Region(resp.RegionId),
			},
			CreatedByDriver: false,
			ClusterID:       id,
			OrgID:           GlobalConfig.OrgID,
			DisplayName:     resp.DisplayName,
		})
		t.lgr.Info("get existing tidb Premium entity, cluster id: %s", id)
	}

	var ups []*TiDBPremiumEntity
	for _, it := range original {
		if it.OrgID == GlobalConfig.OrgID &&
			pie.Contains(GlobalConfig.RegionsOf(it.CloudProvider), it.Region) {
			ups = append(ups, &it)
			t.lgr.Info("add existing tidb Premium cluster to dependency, cluster: %+v", it)
		}
	}

	err := t.initClustersPublicConnection(ups)
	if err != nil {
		t.lgr.Error("init existed tidbClusters public connection failed")
		return nil, err
	}

	return TiDBPremiumEntities(ups).ToDependencyEntities(), nil
}

func regionId2Region(regionId string) string {
	parts := strings.Split(regionId, "-")
	return strings.Join(parts[1:], "-")
}

func (t *tiDBPremiumDependency) ProvisionEntities(ctx context.Context, requirementStats EntitiesStats) (<-chan DependencyEntity, error) {
	out := make(chan DependencyEntity)
	tidbClusters, err := t.doCreateCluster(ctx, requirementStats)
	if err != nil {
		close(out)
		return out, err
	}

	go func() {
		defer close(out)
		var wg sync.WaitGroup
		for _, cluster := range tidbClusters {
			cluster := cluster
			wg.Add(1)
			go func() {
				defer wg.Done()
				if err := t.waitClustersStateActive(ctx, []*TiDBPremiumEntity{cluster}); err != nil {
					t.lgr.Error("wait premium cluster active failed, cluster: %s, error: %s", misc.JSON(cluster), err.Error())
					return
				}
				if err := t.initOneClusterPublicConnection(ctx, cluster); err != nil {
					t.lgr.Error("init premium cluster public connection failed, cluster: %s, error: %s", misc.JSON(cluster), err.Error())
					return
				}
				select {
				case <-ctx.Done():
					t.lgr.Warn("stop sending premium cluster due to context cancellation, cluster: %s, error: %v", misc.JSON(cluster), ctx.Err())
				case out <- cluster:
				}
			}()
		}
		wg.Wait()
	}()

	return out, nil
}
func (t *tiDBPremiumDependency) Delete(entities ...DependencyEntity) error {
	var failed []*TiDBPremiumEntity
	for _, v := range entities {
		if !v.IsCreatedByDriver() {
			continue
		}

		tidb := v.(*TiDBPremiumEntity)

		err := retry.Do(func() error {
			return t.client.DeleteCluster(tidb.OrgID, tidb.ClusterID)
		},
			retry.DelayType(retry.FixedDelay),
			retry.Delay(time.Minute),
			retry.Attempts(10), // changefeed deletion take time which may block cluster deletion
			retry.LastErrorOnly(true))

		if err != nil {
			t.lgr.Error("delete cluster with retry failed, orgID: %s, projectID: %s, clusterID: %s, name: %s",
				tidb.OrgID, tidb.ClusterID, tidb.DisplayName)
			failed = append(failed, tidb)
		}
	}
	if len(failed) != 0 {
		return fmt.Errorf("failed to delete all the tidb dedicated cluster, to deleted count: %d, failed count: %d, failed clusters: %s",
			len(entities), len(failed), misc.JSON(failed))
	}
	return nil
}

func (t *tiDBPremiumDependency) doCreateCluster(ctx context.Context, tidbClusterReqs EntitiesStats) ([]*TiDBPremiumEntity, error) {
	// assemble requirements
	var clusters []*TiDBPremiumEntity
	for _, req := range tidbClusterReqs {
		for i := 0; i < req.Number; i++ {
			dimensions := NewTiDBPremiumDimensions(req.Dimensions)
			clusters = append(clusters, &TiDBPremiumEntity{
				TiDBPremiumDimensions: dimensions,
				CreatedByDriver:       true,
				OrgID:                 GlobalConfig.OrgID,
				DisplayName:           fmt.Sprintf("%s-%s-%s-%s-%s-%s", GlobalConfig.Executor, "e2e-test", dimensions.CloudProvider, dimensions.Region, "premium", utils.GenerateRandomString(7)),
			})
		}
	}

	// send requests
	for _, up := range clusters {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		clusterRequest := clustercli.DefaultCreateClusterRequest()
		clusterRequest.CloudProvider = string(up.CloudProvider)
		clusterRequest.RegionId = fmt.Sprintf("%s-%s", up.CloudProvider, up.Region)
		clusterRequest.DisplayName = up.DisplayName

		clusterId, err := t.client.CreateCluster(up.OrgID, clusterRequest)
		if err != nil {
			return nil, fmt.Errorf("Create cluster %s error: %s", clusterRequest.DisplayName, err.Error())
		}
		t.lgr.Info("Create cluster %s successfully, id: %s", clusterRequest.DisplayName, clusterId)
		up.ClusterID = clusterId
	}
	return clusters, nil
}

func (t *tiDBPremiumDependency) waitClustersStateActive(ctx context.Context, tidbClusters []*TiDBPremiumEntity) error {
	// clusterId -> status
	status := make(map[string]string, len(tidbClusters))
	for _, v := range tidbClusters {
		status[v.ClusterID] = ""
	}
	// clusterId -> TiDBCluster
	tidbClustersMap := make(map[string]*TiDBPremiumEntity, len(tidbClusters))
	for _, v := range tidbClusters {
		tidbClustersMap[v.ClusterID] = v
	}

	ok := retry.Do(func() error {
		// collect status
		for _, v := range tidbClustersMap {
			if status[v.ClusterID] == "ACTIVE" {
				continue
			}
			cluster, err := t.client.GetCluster(v.OrgID, v.ClusterID)
			if err != nil {
				t.lgr.Warn("get cluster error: %s, orgID: %s,  clusterID: %s", err.Error(), v.OrgID, v.ClusterID)
				status[v.ClusterID] = "ERROR"
				continue
			}
			if cluster.State != "ACTIVE" {
				t.lgr.Info("cluster state is not ACTIVE: %s, orgID: %s, clusterID: %s", cluster.State, v.OrgID, v.ClusterID)
				status[v.ClusterID] = "NOT_ACTIVE"
				continue
			}
			t.lgr.Info("cluster state is ACTIVE, orgID: %s,clusterID: %s", v.OrgID, v.ClusterID)
			status[v.ClusterID] = "ACTIVE"
		}

		if pie.All(pie.Values(tidbClustersMap), func(up *TiDBPremiumEntity) bool { return status[up.ClusterID] == "ACTIVE" }) {
			t.lgr.Info("all clusters ACTIVE")
			return nil
		}
		return fmt.Errorf("not ")
	}, retry.DelayType(retry.FixedDelay),
		retry.Delay(time.Minute),
		retry.Attempts(60),
		retry.Context(ctx))

	if ok != nil {
		if err := ctx.Err(); err != nil {
			return err
		}
		t.lgr.Error("wait all clusters ACTIVE timeout")
		for k, v := range status {
			t.lgr.Error("[%s] clusterID: %s, clusterName: %s", v, k, tidbClustersMap[k].DisplayName)
		}
		return fmt.Errorf("wait all clusters ACTIVE timeout")
	}
	return nil
}

func (t *tiDBPremiumDependency) initClustersPublicConnection(clusters []*TiDBPremiumEntity) error {
	for _, v := range clusters {
		err := t.initOneClusterPublicConnection(context.Background(), v)
		if err != nil {
			return err
		}
	}
	return nil
}

func (t *tiDBPremiumDependency) initOneClusterPublicConnection(ctx context.Context, up *TiDBPremiumEntity) error {
	retryOpts := []retry.Option{retry.DelayType(retry.FixedDelay), retry.Delay(time.Minute), retry.Attempts(10), retry.Context(ctx)}
	// enable global access
	err := retry.Do(func() error {
		var ierr error
		ierr = t.client.EnableAccessFromAnyWhere(up.OrgID, up.ClusterID)
		return ierr
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("EnableAccessFromAnyWhere failed, error: %s, tidbClusters: %s",
			err.Error(), misc.JSON(up))
	}
	// set password
	password := genPassword()
	err = retry.Do(func() error {
		var ierr error
		ierr = t.client.SetRootPassword(up.OrgID, up.ClusterID, password)
		return ierr
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("SetRootPassword for DefaultNodeGroup failed, error: %s, nodegroup: %s",
			err.Error(), misc.JSON(up))
	}

	// get url of CA cert
	var urlresp *clustercli.CACertDownloadURLResponse
	err = retry.Do(func() error {
		var ierr error
		urlresp, ierr = t.client.GetCACertDownloadURL(up.OrgID, up.ClusterID)
		return ierr
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("GetCACertDownloadURL failed, error: %s, nodegroup: %s",
			err.Error(), misc.JSON(up))
	}

	// download CA Cert
	var cacert []byte
	err = retry.Do(func() error {
		var ierr error
		cacert, ierr = t.client.GetCACertContent(urlresp.URL)
		return ierr
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("GetCACertContent failed, error: %s, nodegroup: %s",
			err.Error(), misc.JSON(up))
	}

	// the host of public endpoint may be empty before enable global access.
	// query the public endpoint again
	var ngCluster *clustercli.ClusterResponse
	err = retry.Do(func() error {
		var ierr error
		ngCluster, ierr = t.client.GetCluster(up.OrgID, up.ClusterID)
		if ierr != nil {
			return ierr
		}
		publicEndpoint := t.client.GetPublicEndpoint(ngCluster)
		if publicEndpoint == nil || publicEndpoint.Host == "" {
			return fmt.Errorf("can not find public endpoint of DefaultNodeGroup after global access enabled,tidbClusters: %s", misc.JSON(up))
		}
		return nil
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("get DefaultNodeGroup failed after global access enabled, error: %s, tidbClusters: %s", err.Error(), misc.JSON(up))
	}

	// set connection info
	publicEndpoint := t.client.GetPublicEndpoint(ngCluster)
	up.Host = publicEndpoint.Host
	up.Port = fmt.Sprintf("%d", publicEndpoint.Port)
	up.User = "root"
	up.Password = password
	up.CACert = string(cacert)

	return nil
}