// Package task add by blacknc <blacknc@163.com> at 2024/08/28 00:20:44
package task

import (
	"context"
	"fmt"
	"path"
	"time"

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pingcap/tiflow-operator/api/label"
	"go.opentelemetry.io/otel"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/tools/record"
	"sigs.k8s.io/controller-runtime/pkg/client"

	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"

	"github.com/tidbcloud/eks-provider/controllers/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	utilcfg "github.com/tidbcloud/infra-provider/util/config"
)

const (
	Finalizer              = "import.cloud.pingcap.com/finalizer"
	LightningComponent     = "tidb-lightning"
	TiDBServerCertMountDir = "/var/lib/tidb-server-tls"
	ClusterCertMountDir    = "/var/lib/lightning-tls"

	AnnoKeyPrefixUsedByImport = "alpha.usedby.import.cloud.pingcap.com/"
	AnnoValInUse              = "in-use"
)

var (
	Tracer = otel.Tracer("ImportReconciler")
)

type Context struct {
	EnvProvider dataplane.EnvProvider
	Namer       *namer.Namer

	// Injected from the caller
	Local         client.Client
	EventRecorder record.EventRecorder

	// DON'T update cluster in tasks actually
	// DON'T get or set cluster spec in tasks
	// Both will be done after all tasks are finished or aborted.
	// See update.go for more details.
	Cluster *regionalpb.Cluster
	// DON'T MODIFY ClusterRole in tasks
	ClusterRole *regionalpb.ClusterRole
	Kubernetes  *cloudv1.Kubernetes
	Import      *cloudv1.Import

	Config config.RootConfig

	Logger logr.Logger

	ClusterID string
	ProjectID string
	TenantID  string

	// BeginAt denotes the starting time of the task.
	BeginAt *time.Time

	zones            []string
	subnets          []string
	LightningConfig  []byte
	LightningLabels  map[string]string
	LightningVolumes LightningVolumes
}

func (c *Context) GetRegion() string {
	return c.Config.Network.Region
}

func (c *Context) IsDataplaneExist() bool {
	return c.EnvProvider.GetClient() != nil
}

func (c *Context) IsClusterExist() bool {
	return c.Cluster != nil
}

func (c *Context) GetJobName() string {
	return c.Import.Name
}

func (c *Context) Subnets() []string {
	if c.subnets == nil {
		panic("subnets unreachable, may be forget init?")
	}
	return c.subnets
}

type LightningVolumes struct {
	Volumes []corev1.Volume
	Mounts  []corev1.VolumeMount
	Envs    []corev1.EnvVar
}

func (c *Context) Default() (*Context, error) {
	c.Namer = namer.NewNamer().
		WithClusterName(c.GetClusterName()).
		WithTiDBNamespace(c.GetClusterNameSpace())

	providerName := c.Config.Provider
	switch providerName {
	case consts.ProviderTypeEKS:

		// 3-layer architecture must get zones from cluster spec
		// ignoring the zone info in config file
		if cloudaws := c.Kubernetes.Cloud.AWS; cloudaws != nil {
			zones, subnets := cloudaws.SubnetGroups.SubnetsFilterByGroupNameOrZones([]cloudv1.AWSSubnetGroupName{
				cloudv1.AWSSubnetGroupNodes, cloudv1.AWSSubnetGroupSecondaryCIDR,
			}, []string{})

			c.zones = zones
			c.subnets = subnets
		}

	case consts.ProviderTypeGKE:
		if cloudgcp := c.Kubernetes.Cloud.GCP; cloudgcp != nil {
			c.zones = cloudgcp.NodeZones
		}

	case consts.ProviderTypeAKS:
		if cloudaks := c.Kubernetes.Cloud.Azure; cloudaks != nil {
			c.zones = cloudaks.NodeZones
		}

	default:
		return nil, fmt.Errorf("unknown provider type: %s", providerName)
	}

	if len(c.zones) != 3 {
		return nil, fmt.Errorf("available zones %v, the length of zones must be 3", c.zones)
	}

	if c.EnvProvider == nil {
		return nil, errors.New("EnvProvider is required")
	}

	if err := c.parseLightningConfig(); err != nil {
		return nil, err
	}

	c.parseLightningLabels()

	return c, nil
}

func (c *Context) parseLightningLabels() {

	name := c.Import.Name
	tcName := c.Namer.GetTiDBClusterName()
	labels := utils.CopyLabelsForWorkloads(c.Import.Labels)
	labels = common.ConcatenateTags(labels,
		map[string]string{
			label.ComponentLabelKey: LightningComponent,
			label.InstanceLabelKey:  tcName, // "db" for dedicated and "db${clusterName}" for dev-tier now
			label.NameLabelKey:      name,
		})
	c.LightningLabels = labels
}

func (c *Context) parseLightningConfig() error {

	i := c.Import
	var cfg *utilcfg.GenericConfig
	if len(i.Spec.LightningConfig) == 0 {
		return errors.New("lightning config is empty")
	}
	impConfigTOML := new(cloudv1.TomlConfig)
	if err := impConfigTOML.UnmarshalTOML([]byte(i.Spec.LightningConfig)); err != nil {
		return errors.Annotate(err, "parse config TOML data error")
	}
	cfg = utilcfg.New(impConfigTOML.MP)

	if cpuCore := c.Config.ImportConfig.CpuCore; cpuCore > 0 {
		switch i.Spec.Source.Format {
		case cloudv1.DataFormatAuroraParquet, cloudv1.DataFormatRawParquet:
			if cfg.Get("lightning.region-concurrency") == nil {
				// set region-concurrency to 0.7 of the number of CPU cores
				cfg.Set("lightning.region-concurrency", int64(float64(cpuCore)*0.7))
			}
		}
	}

	// use the CA cert in 'db-tidb-server-secret' to verify the tidb server-side cert when connecting to tidb at port 4000.
	// We don't present the client cert/key, because it's not required for enabling TLS.
	cfg.Set("tidb.security.ca-path", path.Join(TiDBServerCertMountDir, "ca.crt"))

	cfgBytes, err := cfg.MarshalTOML()
	if err != nil {
		return err
	}
	c.LightningConfig = cfgBytes
	return nil
}

func (c *Context) GetClusterName() string {
	return c.Import.Spec.ClusterName
}

func (c *Context) GetClusterNameSpace() string {
	return cloudv1.ClusterPrefixId(c.GetClusterName())
}

func (c *Context) GetClusterScopeName() string {
	return fmt.Sprintf("%s-%s", c.GetClusterNameSpace(), c.GetClusterName())
}

// GetAKSProviderConfigName returns the name of the azure Crossplane ProviderConfig object.
func (c *Context) GetAKSProviderConfigName() string {
	// pc-{tidb cluster name}-{hash}-imports
	// all import resources in the same tidb cluster share the same provider config
	return namer.NewName(c.GetClusterScopeName(), "pc", "imports", -1)
}

// GetAWSNodeRoleExternalID returns the external ID for the AWS node role.
func (c *Context) GetAWSNodeRoleExternalID() string {
	if c.ClusterRole == nil {
		panic("cluster role unreachable, may be forget init or cluste role is not exists?")
	}
	return c.ClusterRole.ExternalId
}

// GetEKSProviderConfigName returns the name of the eks Crossplane ProviderConfig object.
func (c *Context) GetEKSProviderConfigName() string {
	// pc-{tidb cluster name}-{hash}-imports
	return namer.NewName(c.GetClusterScopeName(), "pc", "imports", -1)
}

// GetGKEProviderConfigName returns the name of the gke Crossplane ProviderConfig object.
func (c *Context) GetGKEProviderConfigName() string {
	// pc-{tidb cluster name}-{hash}-imports
	// all import resources in the same tidb cluster share the same provider config
	return namer.NewName(c.GetClusterScopeName(), "pc", "imports", -1)
}

func (c *Context) Zones() []string {
	if c.zones == nil {
		panic("zone unreachable, may be forget init?")
	}
	return c.zones
}

type KMSGetter interface {
	Get() string
}

type PCAGetter interface {
	Get() string
}

type GetFunc func() string

func (f GetFunc) Get() string {
	return f()
}

var _ deploy.Task = &BaseTask{}

type BaseTask struct {
	*Context
}

func (t *BaseTask) Name() string {
	return ""
}

func (t *BaseTask) Category() deploy.TaskCategory {
	return deploy.ImportTask
}

func (t *BaseTask) Create(ctx context.Context) deploy.Result {
	return deploy.Done()
}

func (t *BaseTask) PostCreate(ctx context.Context) deploy.Result {
	return deploy.Done()
}

func (t *BaseTask) Delete(ctx context.Context) deploy.Result {
	return deploy.Done()
}

func (t *BaseTask) PostDelete(ctx context.Context) deploy.Result {
	return deploy.Done()
}

// MetricsStartTime will return the `BeginAt` (nil by default). For
// Task whose metrics need to be exported, the `BeginAt` should be
// initialized when the `Create` is being called.
func (t *BaseTask) MetricsStartTime() *time.Time {
	return t.BeginAt
}

func (t *BaseTask) TidbClusterMetaObject() *opv1alpha1.TidbCluster {
	return &opv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.Namer.GetTiDBClusterName(),
			Namespace: t.Namer.GetTiDBNamespace(),
		},
	}
}