package case_driver

import (
	"context"
	"fmt"
	"net"
	"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"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cluster_non_gen"
)

type tiDBDedicatedDependency struct {
	lgr    log.SimpleLogger
	client *cluster_non_gen.ClusterClient
}

const (
	sourceEndpointDNSReadyAttempts = 60
	sourceEndpointDNSReadyDelay    = time.Second * 30
)

func NewTiDBDedicatedDependency(lgr log.SimpleLogger) Dependency {
	return &tiDBDedicatedDependency{
		lgr:    lgr,
		client: cluster_non_gen.NewClient(lgr),
	}
}
func (t *tiDBDedicatedDependency) Type() string {
	return TiDBDedicated
}

func (t *tiDBDedicatedDependency) GetExistingUsableEntities() ([]DependencyEntity, error) {
	var original []TiDBDedicatedEntity

	// enrich tidb cluster info
	for _, id := range GlobalConfig.TiDBDedicatedClusterIds {
		var resp *cluster_non_gen.GetClusterResponse
		ok := retry.Do(func() error {
			var err error
			resp, err = t.client.GetCluster(GlobalConfig.OrgID, GlobalConfig.ProjectID, 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, TiDBDedicatedEntity{
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: CloudProviderType(resp.DedicatedCluster.CloudProvider),
				Region:        regionId2Region(resp.DedicatedCluster.RegionID),
			},
			CreatedByDriver: false,
			ClusterID:       id,
			ProjectID:       GlobalConfig.ProjectID,
			OrgID:           GlobalConfig.OrgID,
			DisplayName:     resp.DedicatedCluster.DisplayName,
		})
		t.lgr.Info("get existing tidb dedicated entity, cluster id: %s", id)
	}

	var ups []*TiDBDedicatedEntity
	for _, it := range original {
		if it.OrgID == GlobalConfig.OrgID &&
			it.ProjectID == GlobalConfig.ProjectID &&
			pie.Contains(GlobalConfig.RegionsOf(it.CloudProvider), it.Region) {
			ups = append(ups, &it)
			t.lgr.Info("add existing tidb dedicated 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 TiDBDedicatedEntities(ups).ToDependencyEntities(), nil
}

func (t *tiDBDedicatedDependency) 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, []*TiDBDedicatedEntity{cluster}); err != nil {
					t.lgr.Error("wait dedicated 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 dedicated cluster public connection failed, cluster: %s, error: %s", misc.JSON(cluster), err.Error())
					return
				}
				select {
				case <-ctx.Done():
					t.lgr.Warn("stop sending dedicated cluster due to context cancellation, cluster: %s, error: %v", misc.JSON(cluster), ctx.Err())
				case out <- cluster:
				}
			}()
		}
		wg.Wait()
	}()

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

		tidb := v.(*TiDBDedicatedEntity)

		err := retry.Do(func() error {
			return t.client.DeleteCluster(tidb.OrgID, tidb.ProjectID, 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.ProjectID, 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 *tiDBDedicatedDependency) doCreateCluster(ctx context.Context, tidbClusterReqs EntitiesStats) ([]*TiDBDedicatedEntity, error) {
	// assemble requirements
	var clusters []*TiDBDedicatedEntity
	for _, req := range tidbClusterReqs {
		for i := 0; i < req.Number; i++ {
			dimensions := NewTiDBDedicatedDimensions(req.Dimensions)
			clusters = append(clusters, &TiDBDedicatedEntity{
				TiDBDedicatedDimensions: dimensions,
				CreatedByDriver:         true,
				ProjectID:               GlobalConfig.ProjectID,
				OrgID:                   GlobalConfig.OrgID,
				DisplayName:             fmt.Sprintf("%s-%s-%s-%s-%s-%s", GlobalConfig.Executor, "e2e-test", dimensions.CloudProvider, dimensions.Region, "dedicated", utils.GenerateRandomString(7)),
			})
		}
	}

	// send requests
	for _, up := range clusters {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		err := t.createCidrIfNeeded(ctx, up)
		if err != nil {
			return nil, fmt.Errorf("create cidr for cluster failed, error: %s, cluster: %s", err.Error(), misc.JSON(up))
		}

		clusterRequest := cluster_non_gen.DefaultCreateClusterRequest()
		clusterRequest.CloudProvider = string(up.CloudProvider)
		clusterRequest.RegionId = fmt.Sprintf("%s-%s", up.CloudProvider, up.Region)
		clusterRequest.DisplayName = up.DisplayName
		if up.CloudProvider != AWS {
			clusterRequest.TikvNodeSetting.StorageType = "Basic"
		}

		clusterId, err := t.client.CreateCluster(up.OrgID, up.ProjectID, clusterRequest)
		if err != nil {
			if !strings.Contains(err.Error(), "Spu not found with specified attributes") {
				return nil, fmt.Errorf("Create cluster %s error: %s", clusterRequest.DisplayName, err.Error())
			}
			// The default node spec is not available in this region; fall back to the smallest available spec.
			t.lgr.Info("default node spec not available in region %s, querying smallest available spec", clusterRequest.RegionId)
			clusterId, err = t.createClusterWithSmallestNodeSpec(ctx, up, clusterRequest)
			if err != nil {
				return nil, fmt.Errorf("Create cluster %s with smallest node spec 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 *tiDBDedicatedDependency) createClusterWithSmallestNodeSpec(ctx context.Context, up *TiDBDedicatedEntity, clusterRequest *cluster_non_gen.CreateClusterRequest) (string, error) {
	if err := ctx.Err(); err != nil {
		return "", err
	}
	regionId := clusterRequest.RegionId

	tidbSpec, err := t.client.GetSmallestNodeSpec(up.OrgID, up.ProjectID, regionId, "TIDB")
	if err != nil {
		return "", fmt.Errorf("get smallest tidb node spec for region %s failed: %s", regionId, err.Error())
	}
	tikvSpec, err := t.client.GetSmallestNodeSpec(up.OrgID, up.ProjectID, regionId, "TIKV")
	if err != nil {
		return "", fmt.Errorf("get smallest tikv node spec for region %s failed: %s", regionId, err.Error())
	}
	t.lgr.Info("selected tidb node spec: %s, tikv node spec: %s for region %s", tidbSpec.NodeSpecKey, tikvSpec.NodeSpecKey, regionId)

	clusterRequest.TidbNodeSetting.NodeSpecKey = tidbSpec.NodeSpecKey
	clusterRequest.TikvNodeSetting.NodeSpecKey = tikvSpec.NodeSpecKey
	clusterRequest.TikvNodeSetting.StorageSizeGi = int(tikvSpec.MinStorageSizeGi)

	return t.client.CreateCluster(up.OrgID, up.ProjectID, clusterRequest)
}

func (t *tiDBDedicatedDependency) createCidrIfNeeded(ctx context.Context, cluster *TiDBDedicatedEntity) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	// list cidr of the project and in cloud provider
	list, err := t.client.ListCidr(cluster.OrgID, cluster.ProjectID, string(cluster.CloudProvider))
	if err != nil {
		return fmt.Errorf("ListCidr error: %s", err.Error())
	}

	// check if it contains the region
	for _, it := range list.Items {
		if it.Region.Name == cluster.Region {
			t.lgr.Info("Cidr already exist for cluster: %s, cidr: %s", misc.JSON(cluster), misc.JSON(it))
			return nil
		}
	}
	t.lgr.Info("Cidr not exist, will create it later for cluster: %s, ", misc.JSON(cluster))

	// if not, create cidr
	_, err = t.client.CreateCidr(cluster.OrgID, cluster.ProjectID, string(cluster.CloudProvider), cluster.Region)
	if err != nil {
		return fmt.Errorf("CreateCidr error: %s", err.Error())
	}
	return nil
}

func (t *tiDBDedicatedDependency) waitClustersStateActive(ctx context.Context, tidbClusters []*TiDBDedicatedEntity) error {
	// clusterId -> status
	status := make(map[string]string, len(tidbClusters))
	for _, v := range tidbClusters {
		status[v.ClusterID] = ""
	}
	// clusterId -> TiDBCluster
	tidbClustersMap := make(map[string]*TiDBDedicatedEntity, 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.ProjectID, v.ClusterID)
			if err != nil {
				t.lgr.Warn("get cluster error: %s, orgID: %s, projectID: %s, clusterID: %s", err.Error(), v.OrgID, v.ProjectID, v.ClusterID)
				status[v.ClusterID] = "ERROR"
				continue
			}
			if cluster.DedicatedCluster.State != "ACTIVE" {
				t.lgr.Info("cluster state is not ACTIVE: %s, orgID: %s, projectID: %s, clusterID: %s", cluster.DedicatedCluster.State, v.OrgID, v.ProjectID, v.ClusterID)
				status[v.ClusterID] = "NOT_ACTIVE"
				continue
			}
			t.lgr.Info("cluster state is ACTIVE, orgID: %s, projectID: %s, clusterID: %s", v.OrgID, v.ProjectID, v.ClusterID)
			status[v.ClusterID] = "ACTIVE"
		}

		if pie.All(pie.Values(tidbClustersMap), func(up *TiDBDedicatedEntity) 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 *tiDBDedicatedDependency) initClustersPublicConnection(clusters []*TiDBDedicatedEntity) error {
	for _, v := range clusters {
		err := t.initOneClusterPublicConnection(context.Background(), v)
		if err != nil {
			return err
		}
	}
	return nil
}

func (t *tiDBDedicatedDependency) initOneClusterPublicConnection(ctx context.Context, up *TiDBDedicatedEntity) error {
	retryOpts := []retry.Option{retry.DelayType(retry.FixedDelay), retry.Delay(time.Minute), retry.Attempts(5), retry.Context(ctx)}
	// get default node group
	var ng *cluster_non_gen.TiDBNodeGroup
	err := retry.Do(func() error {
		var ierr error
		ng, ierr = t.client.GetDefaultNodeGroup(up.OrgID, up.ProjectID, up.ClusterID)
		if ierr != nil {
			return ierr
		}
		publicEndpoint := t.client.GetPublicEndpoint(ng)
		if publicEndpoint == nil {
			return fmt.Errorf("can not find public endpoint of DefaultNodeGroup nodegroup: %s, tidbClusters: %s", misc.JSON(ng), misc.JSON(up))
		}
		return nil
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("get DefaultNodeGroup failed, error: %s, tidbClusters: %s", err.Error(), misc.JSON(up))
	}
	// enable global access
	err = retry.Do(func() error {
		var ierr error
		ierr = t.client.EnableAccessFromAnyWhere(up.OrgID, up.ProjectID, up.ClusterID, ng.TiDBNodeGroupID)
		return ierr
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("EnableAccessFromAnyWhere for DefaultNodeGroup failed, error: %s, nodegroup: %s, tidbClusters: %s",
			err.Error(), misc.JSON(ng), misc.JSON(up))
	}
	// set password
	password := genPassword()
	err = retry.Do(func() error {
		var ierr error
		ierr = t.client.SetRootPassword(up.OrgID, up.ProjectID, up.ClusterID, password)
		return ierr
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("SetRootPassword for DefaultNodeGroup failed, error: %s, nodegroup: %s, tidbClusters: %s",
			err.Error(), misc.JSON(ng), misc.JSON(up))
	}

	// download CA Cert (download URL is short-lived, refresh it on each retry)
	var cacert []byte
	err = retry.Do(func() error {
		var ierr error
		urlresp, ierr := t.client.GetCACertDownloadURL(up.OrgID, up.ProjectID, up.ClusterID)
		if ierr != nil {
			return ierr
		}
		cacert, ierr = t.client.GetCACertContent(urlresp.URL)
		return ierr
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("download CA cert failed, error: %s, nodegroup: %s, tidbClusters: %s",
			err.Error(), misc.JSON(ng), misc.JSON(up))
	}

	// the host of public endpoint may be empty before enable global access.
	// query the public endpoint again
	err = retry.Do(func() error {
		var ierr error
		ng, ierr = t.client.GetDefaultNodeGroup(up.OrgID, up.ProjectID, up.ClusterID)
		if ierr != nil {
			return ierr
		}
		publicEndpoint := t.client.GetPublicEndpoint(ng)
		if publicEndpoint == nil || publicEndpoint.Host == "" {
			return fmt.Errorf("can not find public endpoint of DefaultNodeGroup after global access enabled, nodegroup: %s, tidbClusters: %s", misc.JSON(ng), 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(ng)
	up.Host = publicEndpoint.Host
	up.Port = fmt.Sprintf("%d", publicEndpoint.Port)
	up.User = "root"
	up.Password = password
	up.CACert = string(cacert)

	return t.waitSourceEndpointDNSReady(ctx, up)
}

func (t *tiDBDedicatedDependency) waitSourceEndpointDNSReady(ctx context.Context, up *TiDBDedicatedEntity) error {
	err := retry.Do(func() error {
		if _, err := net.DefaultResolver.LookupHost(ctx, up.Host); err != nil {
			t.lgr.Info("Wait for public endpoint DNS domain to be resolvable")
			return fmt.Errorf("resolve source endpoint %q: %w", up.Host, err)
		}
		return nil
	}, retry.DelayType(retry.FixedDelay), retry.Delay(sourceEndpointDNSReadyDelay), retry.Attempts(sourceEndpointDNSReadyAttempts), retry.Context(ctx), retry.LastErrorOnly(true))
	if err != nil {
		return fmt.Errorf("source endpoint DNS readiness timed out for clusterID %s, host %s: %w", up.ClusterID, up.Host, err)
	}

	t.lgr.Info("source endpoint DNS is ready, clusterID: %s, host: %s", up.ClusterID, up.Host)
	return nil
}

// FIXME: implement a random
func genPassword() string {
	return "EpvcjMaVy3weOau7"
}