package imports

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"

	"github.com/tidbcloud/aws-shared-provider/cmd/manager/app/config"
	controllerutils "github.com/tidbcloud/aws-shared-provider/pkg/controllers/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/pd"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tikv"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	infra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	infraclient "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client"
	"github.com/tidbcloud/tidb-management-service/api/serverless_infra/meta"
	corev1 "k8s.io/api/core/v1"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	tidbPort                        = 4000
	lightningConfigFileName         = "tidb-lightning.toml"
	extraLightningConfigKey         = "tidb-lightning.toml"
	extraVipLightningConfigKey      = "vip-tidb-lightning.toml"
	extraSuperVipLightningConfigKey = "super-vip-tidb-lightning.toml"
	importCmName                    = "import-cm"

	cpuResourceLimitKey              = "cpu-resource-limit"
	cpuResourceRequestKey            = "cpu-resource-request"
	memoryResourceLimitKey           = "mem-resource-limit"
	memoryResourceRequestKey         = "mem-resource-request"
	vipCpuResourceLimitKey           = "vip-cpu-resource-limit"
	vipCpuResourceRequestKey         = "vip-cpu-resource-request"
	vipMemoryResourceLimitKey        = "vip-mem-resource-limit"
	vipMemoryResourceRequestKey      = "vip-mem-resource-request"
	superVipCpuResourceLimitKey      = "super-vip-cpu-resource-limit"
	superVipCpuResourceRequestKey    = "super-vip-cpu-resource-request"
	superVipMemoryResourceLimitKey   = "super-vip-mem-resource-limit"
	superVipMemoryResourceRequestKey = "super-vip-mem-resource-request"
	nodeSelectorKey                  = "node-selector"
	tolerationsKey                   = "tolerations"
	vipNodeSelectorKey               = "vip-node-selector"
	vipTolerationsKey                = "vip-tolerations"
	superVipNodeSelectorKey          = "super-vip-node-selector"
	superVipTolerationsKey           = "super-vip-tolerations"

	defaultCpuResourceRequest    = "1"
	defaultCpuResourceLimit      = "2"
	defaultMemoryResourceRequest = "2Gi"
	defaultMemoryResourceLimit   = "12Gi" // for parquet file, lightning need more memory

	enableCheckpointKey = "enable-checkpoint"

	// TODO: should not hard code pool name.
	vipTidbPool            = "vip-tidb-pool"
	superVipTidbPool       = "super-vip-tidb-pool"
	canaryVipTidbPool      = "canary-vip-tidb-pool"
	canarySuperVipTidbPool = "canary-super-vip-tidb-pool"

	importSuperVIPAnno = types.KeyPrefix + "import-super-vip"

	importClusterCertName = "import-cluster-secret"
)

type ServerlessTierEnv struct {
	importObj      *cloudv1.Import
	clusterObj     *infra.Cluster
	regionalEksCli client.Client
	managerCfg     *config.Config
	tikvApi        tikv.Api
	pdClient       pd.HttpPdClient
}

func (e *ServerlessTierEnv) GetControllerClient() client.Client {
	return e.regionalEksCli
}

func (e *ServerlessTierEnv) StorageClassName() string {
	return types.DefaultStorageClassName
}

func (e *ServerlessTierEnv) ImportJobNamespace() string {
	return e.managerCfg.ImportJobsNamespace
}

func (e *ServerlessTierEnv) EnableCheckpoint(ctx context.Context) bool {
	importCm, err := e.getImportCm(ctx)
	if err != nil {
		ctrl.Log.Error(err, "get import cm")
		return false
	}
	return enableCheckpoint(importCm)
}

func enableCheckpoint(importCm *corev1.ConfigMap) bool {
	enabled, ok := importCm.Data[enableCheckpointKey]
	return ok && enabled == "true"
}

func (e *ServerlessTierEnv) GetClusterCertSecretName() string {
	// If not enable TLS in manager, we will not mount cert secret to import job.
	if e.managerCfg.ImportEnableTLS {
		return importClusterCertName
	}
	return ""
}

func (e *ServerlessTierEnv) buildUrl(url string) string {
	scheme := "http"
	if e.managerCfg.ImportEnableTLS {
		scheme = "https"
	}
	return fmt.Sprintf("%s://%s", scheme, url)
}

func (e *ServerlessTierEnv) getImportCm(ctx context.Context) (*corev1.ConfigMap, error) {
	importCm := &corev1.ConfigMap{}
	err := e.regionalEksCli.Get(ctx, client.ObjectKey{
		Name:      importCmName,
		Namespace: e.managerCfg.ImportJobsNamespace,
	}, importCm)
	return importCm, err
}

func (e *ServerlessTierEnv) getImportTaskID() (int64, error) {
	// import task name format: import-<id>
	idStr := strings.Split(e.importObj.Name, "-")[1]
	id, err := strconv.ParseInt(idStr, 10, 64)
	if err != nil {
		ctrl.Log.Error(err, "parse import task id", "idStr", idStr)
		return 0, err
	}
	return id, nil
}

func (e *ServerlessTierEnv) CancelLoadDataTasks(ctx context.Context, importTaskID int64) error {
	clusterID, err := e.pdClient.GetClusterID(ctx)
	if err != nil {
		return err
	}
	err = e.tikvApi.CancelLoadDataTasks(ctx, clusterID, e.clusterObj.Status.KeyspaceId, importTaskID)
	if err != nil {
		return err
	}

	return nil
}

type ServerlessTierEnvBuilder struct {
	infraClusterCli    infraclient.Client
	regionalClusterCli client.Client
	managerCfg         *config.Config
	tikvApi            tikv.Api
	pdClient           pd.HttpPdClient
}

// NewServerlessTierEnvBuilder creates a `ServerlessTierEnvBuilder` instance.
func NewServerlessTierEnvBuilder(
	infraClusterCli infraclient.Client,
	sharedClusterCli client.Client,
	managerCfg *config.Config,
	tikvApi tikv.Api,
	pdClient pd.HttpPdClient,
) *ServerlessTierEnvBuilder {
	return &ServerlessTierEnvBuilder{
		infraClusterCli:    infraClusterCli,
		regionalClusterCli: sharedClusterCli,
		managerCfg:         managerCfg,
		tikvApi:            tikvApi,
		pdClient:           pdClient,
	}
}

// BuildEnvFromImport builds an import Env from an import API object.
// It implements the EnvBuilder interface.
// It creates an import env based on an infra cluster.
func (b *ServerlessTierEnvBuilder) BuildEnvFromImport(
	ctx context.Context,
	i *cloudv1.Import,
) (*ServerlessTierEnv, error) {
	env := &ServerlessTierEnv{
		importObj:      i,
		regionalEksCli: b.regionalClusterCli,
		managerCfg:     b.managerCfg,
		tikvApi:        b.tikvApi,
		pdClient:       b.pdClient,
	}
	c, err := getInfraCluster(ctx, i, b)
	if err != nil {
		if errors.Is(err, infraclient.ErrNotFound) {
			ctrl.Log.Info("cluster not found for import", "import", i.Name, "cluster", i.Spec.ClusterName)
			return env, nil
		}
		return nil, err
	}
	env.clusterObj = c
	return env, nil
}

func getInfraCluster(ctx context.Context, i *cloudv1.Import, b *ServerlessTierEnvBuilder) (*infra.Cluster, error) {
	if strings.Contains(i.Namespace, "test") {
		/* e2e test namespace */
		cluster := &infra.Cluster{ObjectMeta: meta.ObjectMeta{Namespace: i.Namespace, Name: i.Spec.ClusterName}}
		err := b.infraClusterCli.Get(ctx, cluster)
		return cluster, err
	}
	nsPrefix, projectId, err := controllerutils.ParseNamespacePrefixAndProjectID(i.Namespace)
	if err != nil {
		return nil, err
	}
	return controllerutils.GetInfraCluster(ctx, b.infraClusterCli, nsPrefix, i.Spec.ClusterName, projectId)
}