package imports

import (
	"context"
	"fmt"

	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	"github.com/crossplane-contrib/provider-aws/apis/eks/manualv1alpha1"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	storagev1 "k8s.io/api/storage/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/rest"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers"
	"github.com/tidbcloud/eks-provider/controllers/imports/imports"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	compatk8s "github.com/tidbcloud/infra-provider/pkg/compat/k8s/dataplane"
	compatnamer "github.com/tidbcloud/infra-provider/pkg/compat/namer"
	compatnodepool "github.com/tidbcloud/infra-provider/pkg/compat/nodepool"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	sdk "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/nodegroup"
	xpnodegroup "github.com/tidbcloud/infra-provider/pkg/nodegroup"
	"github.com/tidbcloud/infra-provider/pkg/nodepool"
	"github.com/tidbcloud/infra-provider/pkg/procedure"
	"github.com/tidbcloud/infra-provider/pkg/util"
	"github.com/tidbcloud/infra-provider/pkg/util/subnet"
)

type EnvBuilder struct {
	InfraClusterCli client.Client
	Config          config.RootConfig
	AWSSVC          procedure.AWSService
}

func (b *EnvBuilder) BuildEnvFromImport(ctx context.Context, i *cloudv1.Import) (imports.Env, error) {
	providerType, err := imports.GetProviderType(i)
	if err != nil {
		return nil, errors.Annotate(err, "get provider type failed")
	}
	iog, err := imports.GetImportObjectGroup(ctx, b.InfraClusterCli, i)
	if err != nil {
		return nil, errors.Annotate(err, "get import object group failed")
	}
	return &Env{
		infraClusterCli:   b.InfraClusterCli,
		config:            b.Config,
		awsSvc:            b.AWSSVC,
		importObjectGroup: iog,
		providerType:      providerType,
	}, nil
}

type Env struct {
	infraClusterCli   client.Client
	config            config.RootConfig
	awsSvc            procedure.AWSService
	importObjectGroup *imports.ImportAPIObjectGroup
	providerType      string
}

var _ imports.Env = &Env{}

func (e *Env) TCName(clusterName string) string {
	return namer.NewNamer().GetTiDBClusterName()
}

func (e *Env) StorageClassName() string {
	return consts.ImportGp3StorageClassName
}

func (e *Env) GetControllerClient(ctx context.Context) (client.Client, error) {
	network := e.importObjectGroup.Network
	if network == nil {
		return nil, errors.New("network CR object is nil")
	}
	if !util.ContainsRegionLabel(network) {
		return nil, fmt.Errorf("region label does not exist")
	}

	scheme, err := k8s.TiDBOperatorScheme()
	if err != nil {
		return nil, fmt.Errorf("unable to get scheme: %v", err)
	}

	g, err := compatk8s.NetworkClientGetter(ctx, e.infraClusterCli, network, scheme)
	if err != nil {
		return nil, err
	}
	return g.GetClient(ctx)
}

func (e *Env) GetKubeConfig(ctx context.Context) (*rest.Config, error) {
	network := e.importObjectGroup.Network
	if network == nil {
		return nil, errors.New("network CR object is nil")
	}
	if !util.ContainsRegionLabel(network) {
		return nil, fmt.Errorf("region label does not exist")
	}
	scheme, err := k8s.TiDBOperatorScheme()
	if err != nil {
		return nil, fmt.Errorf("unable to get scheme: %v", err)
	}
	opts := client.Options{
		Scheme: scheme,
	}
	var g k8s.ClientManager
	switch e.providerType {
	case consts.ProviderTypeEKS:
		g = compatk8s.NewEKSClientManagerCompat(e.infraClusterCli, *network, opts)
	case consts.ProviderTypeGKE:
		g = compatk8s.NewGKEClientManagerCompat(e.infraClusterCli, *network, opts)
	}
	key := types.NamespacedName{
		Name: compatnamer.LegacyBuildClusterNameByNetwork(network),
	}
	cfg, err := g.GetConfig(ctx, key)
	if err != nil {
		return nil, err
	}
	log.Info("get kube config", zap.String("key name", key.Name), zap.Any("cfg", cfg), zap.Any("scheme", scheme), zap.Any("network", network))
	return cfg, nil
}

func (e *Env) PreCreateJobHook(job *batchv1.Job) error {
	cluster := e.importObjectGroup.Cluster
	imp := e.importObjectGroup.Import
	if cluster == nil {
		return errors.New("cluster CR object is nil")
	}
	if imp == nil {
		return errors.New("import CR object is nil")
	}
	// HACK: label the import object
	ctx := context.Background()
	if err := controllers.CheckRegionalLabel(ctx, cluster, imp, e.infraClusterCli, e.config.Network.Region); err != nil {
		return err
	}
	for i := range job.Spec.Template.Spec.Containers {
		c := &job.Spec.Template.Spec.Containers[i]
		if c.Name == "tidb-lightning" {
			c.Image = sdk.OverrideBaseImage(cluster, c.Image)
		}
	}

	return nil
}

func (e *Env) PreCreateSvcHook(svc *corev1.Service) error {
	return nil
}

// GetDataPlaneNamespace get the namespace for import resources in the data plane.
// It implements the Env interface.
func (e *Env) GetDataPlaneNamespace() (string, error) {
	iog := e.importObjectGroup
	if iog == nil {
		return "", errors.New("import object group is nil")
	}
	if iog.Import == nil {
		return "", errors.New("import object is nil")
	}
	return cloudv1.ClusterPrefixId(iog.Import.Spec.ClusterName), nil
}

// GetDataPlaneCertSecretName get the secret name for import resources in the data plane.
// It implements the Env interface.
func (e *Env) GetClusterCertSecretName() string {
	return fmt.Sprintf("%s-cluster-client-secret", e.TCName(""))
}

func (e *Env) GetTiDBServerCertSecretName() string {
	return fmt.Sprintf("%s-tidb-server-secret", e.TCName(""))
}

// TODO: maybe we need remove the credentials from the instance, because we
// have already filed the credential information in the source uri
func awsCredentials(network *cloudv1.Network, i *cloudv1.Import) []byte {
	// role_arn is the ARN of the role to assume
	// external_id is the external ID used to assume the role
	// With the config file, AWS CLI will assume the role specified with the external ID.
	format := `[default]
  role_arn = %s
  external_id = %s
  credential_source = Ec2InstanceMetadata
      `

	return []byte(fmt.Sprintf(format, i.Spec.AccessConfig.RoleArn, network.Status.IAM.ExternalId))
}

func (e *Env) Credentials(ctx context.Context) ([]byte, error) {
	network := e.importObjectGroup.Network
	if network == nil {
		return nil, errors.New("network CR object is nil")
	}
	if e.providerType == consts.ProviderTypeEKS {
		imp := e.importObjectGroup.Import
		if imp == nil {
			return nil, errors.New("import CR object is nil")
		}

		return awsCredentials(network, imp), nil
	}

	// xaccount/gke ignore import.Spec.AccessConfig.RoleArn and always run as node role(default credentials get by SDK)
	return nil, nil
}

func (e *Env) CreateNodeGroup(ctx context.Context, ngSet *nodegroup.Set) (map[string]string, error) {
	network := e.importObjectGroup.Network
	if network == nil {
		return nil, fmt.Errorf("empty network")
	}
	if !util.ContainsRegionLabel(network) {
		return nil, fmt.Errorf("region label does not exist")
	}
	if len(ngSet.NodeGroups) == 0 {
		return nil, nil
	}
	ngSpec := ngSet.NodeGroups[0]
	ngSpecNamer := namer.NewNamer().
		LegacyWithNetworkName(network.Name).
		WithComponent(string(ngSpec.ComponentType)).
		WithNodeGroupSetOwner(ngSet.Name).
		WithClusterName(e.importObjectGroup.Import.Spec.ClusterName)
	ngSpec.Labels = util.ConcatenateTags(ngSpec.Labels, ngSpecNamer.Labels())
	if e.importObjectGroup.Import.Spec.NodeRole != nil {
		ngSpec.NodeRole = e.importObjectGroup.Import.Spec.NodeRole
	}
	switch e.providerType {
	case consts.ProviderTypeEKS:
		ngSpec.Name = tidbNodeGroupName(ngSpec.Name)
		networkSpec, err := eksv1alpha1.EKSNetworkSpecForNetwork(network)
		if err != nil {
			return nil, fmt.Errorf("unable to get network spec: %v", err)
		}
		subnets, err := subnet.DescribeSubnets(e.awsSvc, network)
		if err != nil {
			return nil, err
		}
		group, ngLaunchTemplate, ngMut, launchTemplateMut := xpnodegroup.FromNodegroupSpec(
			ngSpec, network, networkSpec,
			subnet.FindSubnetIdsByZones(subnets, ngSpec.Zones),
			nil,
			nil, true)
		if _, err := controllerutil.CreateOrUpdate(ctx, e.infraClusterCli, ngLaunchTemplate, launchTemplateMut); err != nil {
			return nil, err
		}
		if _, err := controllerutil.CreateOrUpdate(ctx, e.infraClusterCli, group, ngMut); err != nil {
			return nil, err
		}
	case consts.ProviderTypeGKE:
		ngSpec.Name = importNodePoolName(ngSet.Name)
		networkSpec, err := eksv1alpha1.GKENetworkSpecForNetwork(network)
		if err != nil {
			return nil, fmt.Errorf("unable to get network spec: %v", err)
		}
		np, npMut := compatnodepool.GetNodePoolByNGSpecAndNetWork(ngSpec, network, networkSpec.ProjectID)
		if _, err := controllerutil.CreateOrUpdate(ctx, e.infraClusterCli, np, npMut); err != nil {
			return nil, err
		}
	}
	// Currently infra-api does not pass cluster name to the function.
	return ngSpec.Labels, nil
}

func (e *Env) DeleteNodeGroup(ctx context.Context, ngSet *nodegroup.Set) error {
	network := e.importObjectGroup.Network
	if network == nil {
		return errors.New("network CR object is empty")
	}
	if !util.ContainsRegionLabel(network) {
		return fmt.Errorf("region label does not exist")
	}
	switch e.providerType {
	case consts.ProviderTypeEKS:
		if err := e.infraClusterCli.DeleteAllOf(ctx, &manualv1alpha1.NodeGroup{}, namer.NodeGroupSetOwnerSelector(ngSet.Name)); err != nil {
			return err
		}
		// launchtemplate's name is the same as the nodegroup's name
		if err := e.infraClusterCli.DeleteAllOf(ctx, &ec2v1alpha1.LaunchTemplate{}, namer.NodeGroupSetOwnerSelector(ngSet.Name)); err != nil {
			return err
		}
	case consts.ProviderTypeGKE:
		specName := importNodePoolName(ngSet.Name)
		if err := client.IgnoreNotFound(
			nodepool.DeleteNodePoolBySpecNameAndNetWork(ctx, e.infraClusterCli, specName, network)); err != nil {
			return err
		}
	}
	return nil
}

func (e *Env) PreCreateSecretHook(_ *corev1.Secret) error {
	return nil
}

func (e *Env) EnableCheckpoint() bool {
	return false
}

func tidbNodeGroupName(name string) string {
	return "tidbcloud-" + name
}

func importNodePoolName(name string) string {
	return "import-" + name
}

func (e *Env) BuildStorageClass() *storagev1.StorageClass {
	switch e.providerType {
	case consts.ProviderTypeEKS:
		{
			sc := cloudv1.StorageClass{
				Name:       consts.ImportGp3StorageClassName,
				Type:       consts.AWSVolumeTypeGP3,
				IOPS:       consts.ImportGp3IOPS,
				Throughput: consts.ImportGp3ThroughputMBps,
			}
			return controllers.ConvertAWSStorageClass(sc)
		}
	case consts.ProviderTypeGKE:
		{
			bindingMode := storagev1.VolumeBindingWaitForFirstConsumer
			reclaimPolicy := corev1.PersistentVolumeReclaimDelete

			return &storagev1.StorageClass{
				ObjectMeta: metav1.ObjectMeta{
					Name: consts.ImportGp3StorageClassName,
				},
				VolumeBindingMode:    &bindingMode,
				AllowVolumeExpansion: pointer.Bool(true),
				ReclaimPolicy:        &reclaimPolicy,
				Provisioner:          "pd.csi.storage.gke.io",
				Parameters: map[string]string{
					"csi.storage.k8s.io/fstype": "ext4",
					"type":                      consts.GCPVolumeTypePDBalanced,
				},
				MountOptions: []string{"nodelalloc,noatime"},
			}
		}
	}
	return nil
}