package case_driver

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"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"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/essential"
)

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

func NewTiDBServerlessDependency(lgr log.SimpleLogger) Dependency {
	return &tiDBServerlessDependency{
		lgr:    lgr,
		client: cluster_non_gen.NewClient(lgr),
	}
}

func (t *tiDBServerlessDependency) Type() string {
	return TiDBServerless
}

func (t *tiDBServerlessDependency) GetExistingUsableEntities() ([]DependencyEntity, error) {
	var original []TiDBServerlessEntity

	// enrich tidb cluster info
	for _, id := range GlobalConfig.TiDBServerlessClusterIds {
		var cluster *cluster_non_gen.TidbCloudServerlessCluster
		ok := retry.Do(func() error {
			var err error
			cluster, err = t.client.GetServerlessCluster(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())
		}
		servicePlan := cluster.ServicePlan
		if servicePlan == "Starter" {
			// Empty is the legacy dimension value used by Starter requirements;
			// retain an explicit plan only for Essential and newer suites.
			servicePlan = ""
		}
		original = append(original, TiDBServerlessEntity{
			TiDBServerlessDimensions: TiDBServerlessDimensions{
				CloudProvider: CloudProviderType(cluster.Region.CloudProvider),
				Region:        cluster.Region.RegionID,
				ServicePlan:   servicePlan,
			},
			CreatedByDriver: false,
			ClusterID:       id,
			ProjectID:       GlobalConfig.ProjectID,
			OrgID:           GlobalConfig.OrgID,
			DisplayName:     cluster.DisplayName,
		})
		t.lgr.Info("get existing tidb erverless entity, cluster id: %s", id)
	}

	var ups []*TiDBServerlessEntity
	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 serverless 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 TiDBServerlessEntities(ups).ToDependencyEntities(), nil
}

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

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

		serverlessCluster := v.(*TiDBServerlessEntity)
		err := retry.Do(func() error {
			return t.client.DeleteServerlessCluster(serverlessCluster.OrgID, serverlessCluster.ProjectID, serverlessCluster.ClusterID)
		},
			retry.DelayType(retry.FixedDelay),
			retry.Delay(time.Minute),
			retry.Attempts(3),
			retry.LastErrorOnly(true))

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

func (t *tiDBServerlessDependency) doCreateCluster(ctx context.Context, tidbClusterReqs EntitiesStats) ([]*TiDBServerlessEntity, error) {
	// assemble requirements
	var clusters []*TiDBServerlessEntity
	for _, req := range tidbClusterReqs {
		for i := 0; i < req.Number; i++ {
			dimensions := NewTiDBServerlessDimensions(req.Dimensions)
			clusters = append(clusters, &TiDBServerlessEntity{
				TiDBServerlessDimensions: 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, "serverless", utils.GenerateRandomString(7)),
			})
		}
	}

	// send requests
	for _, up := range clusters {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		req := serverlessClusterCreateParams(up.TiDBServerlessDimensions)
		req.DisplayName = up.DisplayName

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

func serverlessClusterCreateParams(dimensions TiDBServerlessDimensions) *cluster_non_gen.TidbCloudServerlessCluster {
	req := defaultServerlessClusterParams()
	req.Region.CloudProvider = string(dimensions.CloudProvider)
	req.Region.Name = fmt.Sprintf("regions/%s-%s", dimensions.CloudProvider, dimensions.Region)
	if dimensions.ServicePlan != "" {
		req.ServicePlan = dimensions.ServicePlan
		if dimensions.ServicePlan == essential.ServicePlanEssential {
			req.AutoScaling = &cluster_non_gen.V1beta1ClusterAutoScaling{MinRcu: essential.MinRcu, MaxRcu: essential.MaxRcu}
		}
	}
	return req
}

func (t *tiDBServerlessDependency) waitClustersStateActive(ctx context.Context, tidbClusters []*TiDBServerlessEntity) error {
	// clusterId -> status
	status := make(map[string]string, len(tidbClusters))
	for _, v := range tidbClusters {
		status[v.ClusterID] = ""
	}
	// clusterId -> TiDBCluster
	tidbClustersMap := make(map[string]*TiDBServerlessEntity, 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
			}
			resp, err := t.client.GetServerlessCluster(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 resp.State != "ACTIVE" {
				t.lgr.Info("cluster state is not ACTIVE: %s, orgID: %s, projectID: %s, clusterID: %s", resp.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 *TiDBServerlessEntity) bool { return status[up.ClusterID] == "ACTIVE" }) {
			t.lgr.Info("all clusters ACTIVE")
			return nil
		}
		return fmt.Errorf("not all clusters ACTIVE")
	}, 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 *tiDBServerlessDependency) initClustersPublicConnection(clusters []*TiDBServerlessEntity) error {
	for _, v := range clusters {
		err := t.initOneClusterPublicConnection(context.Background(), v)
		if err != nil {
			return err
		}
	}
	return nil
}

func (t *tiDBServerlessDependency) initOneClusterPublicConnection(ctx context.Context, up *TiDBServerlessEntity) error {
	retryOpts := []retry.Option{retry.DelayType(retry.FixedDelay), retry.Delay(time.Minute), retry.Attempts(3), retry.Context(ctx)}
	// set password
	password := genPassword()
	err := retry.Do(func() error {
		return t.client.ChangeServerlessClusterRootPassword(up.OrgID, up.ProjectID, up.ClusterID, password)
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("SetRootPassword failed, error: %s, tidbClusters: %s", err.Error(), misc.JSON(up))
	}

	// get public connection info
	err = retry.Do(func() error {
		resp, ierr := t.client.GetServerlessCluster(up.OrgID, up.ProjectID, up.ClusterID)
		if ierr != nil {
			return ierr
		}
		up.Host = resp.Endpoints.Public.Host
		up.Port = strconv.Itoa(int(resp.Endpoints.Public.Port))
		up.User = resp.UserPrefix + ".root"
		up.Password = password
		up.PLServiceName = getServerlessClusterPLService(resp)
		return nil
	}, retryOpts...)
	if err != nil {
		return fmt.Errorf("ServerlessServiceGetCluster failed, error: %s, tidbClusters: %s", err.Error(), misc.JSON(up))
	}

	return nil
}

func getServerlessClusterPLService(cluster *cluster_non_gen.TidbCloudServerlessCluster) string {
	if cluster.Endpoints.Private.Aws != nil {
		return cluster.Endpoints.Private.Aws.ServiceName
	} else if cluster.Endpoints.Private.Gcp != nil {
		return cluster.Endpoints.Private.Gcp.ServiceAttachmentName
	} else if cluster.Endpoints.Private.Alicloud != nil {
		return cluster.Endpoints.Private.Alicloud.ServiceName
	}
	return ""
}

func defaultServerlessClusterParams() *cluster_non_gen.TidbCloudServerlessCluster {
	defaultRequestBytes := []byte(`
{
    "region": {
        "cloudProvider": "aws",
        "name": "regions/aws-us-east-1"
    },
    "displayName": "Cluster1",
    "spendingLimit": {
        "monthly": 1000
    },
    "encryptionConfig": {
        "enhancedEncryptionEnabled": false
    },
	"servicePlan": "Starter"
}
`)
	var request cluster_non_gen.TidbCloudServerlessCluster
	err := json.Unmarshal(defaultRequestBytes, &request)
	if err != nil {
		panic("unmarshal serverless defaultRequestJson error: " + err.Error())
	}
	return &request
}