package imports

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/rest"
	"sigs.k8s.io/controller-runtime/pkg/client"

	storagev1 "k8s.io/api/storage/v1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	gprovider "github.com/tidbcloud/infra-api/controllers/provider/gardener"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/gardener"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	"github.com/tidbcloud/infra-api/pkg/nodegroup"
	"github.com/tidbcloud/infra-api/pkg/util"
)

// EnvBuilder defines the interfaces to create an import Env.
type EnvBuilder interface {
	// BuildEnvFromImport builds an import Env from an import API object.
	BuildEnvFromImport(ctx context.Context, i *cloudv1.Import) (Env, error)
}

// Env defines the operations in a TiDB Cluster K8S Environment for an import.
type Env interface {
	// TCName get the tidbcluster name create by the specified cluster.
	TCName(clusterName string) string
	// StorageClassName used to create pv as the local storage for lighting.
	StorageClassName() string
	// PreCreateSecretHook is call before we create the secret.
	PreCreateSecretHook(secret *corev1.Secret) error
	// PreCreateJobHook is call before we create the job.
	PreCreateJobHook(job *batchv1.Job) error
	// Credentials gets the credential file content for import.
	Credentials(ctx context.Context) ([]byte, error)
	// CreateNodeGroup creates a node group in the K8S cluster.
	CreateNodeGroup(ctx context.Context, ngs *nodegroup.Set) (map[string]string, error)
	// DeleteNodeGroup deletes the node group in the K8S cluster.
	DeleteNodeGroup(ctx context.Context, ngs *nodegroup.Set) error
	// GetControllerClient gets the controller runtime client for the K8S cluster.
	GetControllerClient(ctx context.Context) (client.Client, error)
	// GetKubeConfig gets the kubeconfig for the K8S cluster.
	GetKubeConfig(ctx context.Context) (*rest.Config, error)
	// PreCreateSvcHook override svc before create.
	PreCreateSvcHook(svc *corev1.Service) error
	// GetDataPlaneNamespace get the namespace for import resources in the data plane.
	GetDataPlaneNamespace() (string, error)
	// GetClusterCertSecretName get the secret name of inter-components mTLS cert. This cert/key/CA Cert is used for
	// inter-components communication with mTLS.
	// This cert can be issued by a different CA cert than tidb-server cert (can be retrieved by GetTiDBServerCertSecretName).
	GetClusterCertSecretName() string
	// GetTiDBServerCertSecretName get the secret name of tidb-server cert which includes the tidb server-side cert/key and the CA cert of server-side cert.
	// This cert can be issued by a different CA cert than inter-components mTLS cert (can be retrieved by GetClusterCertSecretName).
	GetTiDBServerCertSecretName() string
	// EnableCheckpoint enable checkpoint for import.
	EnableCheckpoint() bool
	// BuildStorageClass builds the storage class for the K8S cluster.
	BuildStorageClass() *storagev1.StorageClass
}

// GardenerShootEnv is an Env implementation for Gardener Shoot Cluster.
type GardenerShootEnv struct {
	gardenerEnv       *gardener.GardEnvironment
	importObjectGroup *ImportAPIObjectGroup
	shootName         string
	shootCtx          *gardener.ShootContext
	infraClusterCli   client.Client
	providerType      string
}

// NewGardenerShootEnv creates an import env based on gardener shoot K8S cluster.
func NewGardenerShootEnv(ctx context.Context, gardenerEnv *gardener.GardEnvironment, importObjectGroup *ImportAPIObjectGroup, infraClusterCli client.Client, providerType string) (*GardenerShootEnv, error) {
	if importObjectGroup == nil || importObjectGroup.Network == nil {
		return nil, errors.New("cannot find the network CR object.")
	}
	networkStatus, err := cloudv1.GardenerNetworkStatusFromNetwork(importObjectGroup.Network)
	if err != nil {
		return nil, errors.Annotate(err, "extract gardener network status from network failed.")
	}
	shootName := networkStatus.ShootName
	if len(shootName) == 0 {
		return nil, errors.New("cannot get the shoot name of the network")
	}
	shootCtx, err := gardenerEnv.GetShootContext(ctx, shootName)
	if err != nil {
		return nil, errors.Annotate(err, "get shoot context failed")
	}
	return &GardenerShootEnv{
		gardenerEnv:       gardenerEnv,
		importObjectGroup: importObjectGroup,
		shootName:         shootName,
		shootCtx:          shootCtx,
		providerType:      providerType,
		infraClusterCli:   infraClusterCli,
	}, nil
}

// TCName get the tidbcluster name create by the specified cluster.
// It implements the Env interface.
func (e *GardenerShootEnv) TCName(_ string) string {
	return "db"
}

// GetControllerClient gets the controller runtime client for the K8S cluster.
// It implements the Env interface.
func (e *GardenerShootEnv) GetControllerClient(ctx context.Context) (client.Client, error) {
	return e.shootCtx.Manager.GetClient(), nil
}

// GetKubeConfig gets the kubeconfig for the K8S cluster.
// It implements the Env interface.
func (e *GardenerShootEnv) GetKubeConfig(ctx context.Context) (*rest.Config, error) {
	return e.shootCtx.Manager.GetConfig(), nil
}

// StorageClassName used to create pv as the local storage for lighting.
// It implements the Env interface.
func (e *GardenerShootEnv) StorageClassName() string {
	switch e.providerType {
	case consts.ProviderTypeAWS:
		return k8s.ImportGp3StorageClassName
	default:
		return DefaultStorageClassName
	}
}

// EnableCheckpoint enable checkpoint for import.
func (e *GardenerShootEnv) EnableCheckpoint() bool {
	return false
}

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

// PreCreateJobHook is call before we create the job.
// It implements the Env interface.
func (e *GardenerShootEnv) PreCreateJobHook(_ *batchv1.Job) error {
	return nil
}

// Credentials gets the credential file content for import.
// It implements the Env interface.
func (e *GardenerShootEnv) Credentials(ctx context.Context) ([]byte, error) {
	iog := e.importObjectGroup
	if iog == nil {
		return nil, errors.New("import object group is nil")
	}
	if iog.Import == nil {
		return nil, errors.New("import object is nil")
	}
	if iog.Network == nil {
		return nil, errors.New("network object is nil")
	}
	switch e.providerType {
	case consts.ProviderTypeAWS:
		return awsCredentials(ctx, iog.Network, iog.Import)
	case consts.ProviderTypeGCP:
		return GCPCredentials(ctx, e.infraClusterCli, iog.Network)
	default:
		return nil, errors.Errorf("provider '%s' is not supported yet", e.providerType)
	}
}

func awsCredentials(ctx context.Context, network *cloudv1.Network, i *cloudv1.Import) ([]byte, error) {
	if i.Spec.AccessConfig.RoleArn != "" {
		return awsCrossAccountCredentials(network, i), nil
	}

	roleARN, err := GetImportIamIdentity(network)
	if err != nil {
		return nil, err
	}

	// with AWS_SDK_LOAD_CONFIG setting as true
	// this file ~/.aws/credentials will make the SDK
	// get credentials from ec2 meta and than assume the specified role_arn
	format := `[default]
role_arn = %s
credential_source = Ec2InstanceMetadata
`
	return []byte(fmt.Sprintf(format, roleARN)), nil
}

func awsCrossAccountCredentials(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 GCPCredentials(ctx context.Context, cli client.Client, network *cloudv1.Network) ([]byte, error) {
	saName := gprovider.GetGCPAccessServiceAccountName(network.Spec.Region, network.GetProjectId())
	keyName := gprovider.GetGCPAccessServiceAccountKeyName(saName)
	secretName := types.NamespacedName{Namespace: network.Namespace, Name: keyName}

	var keySecret corev1.Secret
	if err := cli.Get(ctx, secretName, &keySecret); err != nil {
		return nil, err
	}
	return keySecret.Data[gcpCredentialsKey], nil
}

// CreateNodeGroup creates a node group in the K8S cluster.
// It implements the Env interface.
func (e *GardenerShootEnv) CreateNodeGroup(ctx context.Context, ngs *nodegroup.Set) (map[string]string, error) {
	if _, err := e.gardenerEnv.ApplyNodeGroupSet(ctx, e.shootName, ngs, gardener.NodeGroupPolicyScaleOnly); err != nil {
		return nil, errors.Trace(err)
	}
	ng := ngs.NodeGroups[0]
	iog := e.importObjectGroup
	if iog == nil {
		return nil, errors.New("import object group is empty")
	}
	i := iog.Import
	if i == nil {
		return nil, errors.New("import object is empty")
	}
	return map[string]string{
		nodegroup.GardenerWorkerNodeLabelName: nodegroup.NodeGroupName(util.FnvHash(i.ClusterScopeName()), TiDBLightningComName, ng.Zones[0]),
	}, nil
}

// DeleteNodeGroup deletes the node group in the K8S cluster.
// It implements the Env interface.
func (e *GardenerShootEnv) DeleteNodeGroup(ctx context.Context, ngs *nodegroup.Set) error {
	if _, err := e.gardenerEnv.ApplyNodeGroupSet(ctx, e.shootName, ngs, gardener.NodeGroupPolicyScaleOnly); err != nil {
		return errors.Trace(err)
	}
	return nil
}

// PreCreateSvcHook override svc before create.
// It implements the Env interface.
func (e *GardenerShootEnv) 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 *GardenerShootEnv) GetDataPlaneNamespace() (string, error) {
	/*
	   Originally, the data plane namespace was retrieved from the status field of the cluster CR.
	   However, this approach caused the import CR to be dependent on the cluster CR for reconciliation.
	   To make the two types of CR objects independent of each other, we have hardcoded the method to obtain the namespace here.
	   The method for generating the namespace is rarely changed, so it is acceptable to hardcode it in this location.
	*/
	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 *GardenerShootEnv) GetClusterCertSecretName() string {
	return fmt.Sprintf("%s-cluster-client-secret", e.TCName(""))
}
func (e *GardenerShootEnv) GetTiDBServerCertSecretName() string {
	return ""
}

// BuildStorageClass builds the storage class for the K8S cluster.
// In Gargdener arch the storage class is created by the gardener controller.
func (e *GardenerShootEnv) BuildStorageClass() *storagev1.StorageClass {
	return nil
}

// GardenerShootEnvBuilder is a EnvBuilder implementation for building a gardener-shoot-based import environment.
type GardenerShootEnvBuilder struct {
	gardenerEnv     *gardener.GardEnvironment
	infraClusterCli client.Client
}

// NewGardenerShootEnvBuilder creates a `GardenerShootEnvBuilder` instance.
func NewGardenerShootEnvBuilder(infraClusterCli client.Client, g *gardener.GardEnvironment) *GardenerShootEnvBuilder {
	return &GardenerShootEnvBuilder{
		gardenerEnv:     g,
		infraClusterCli: infraClusterCli,
	}
}

// BuildEnvFromImport builds an import Env from an import API object.
// It implements the EnvBuilder interface.
// It creates an import env based on a gardener-shoot cluster.
func (b *GardenerShootEnvBuilder) BuildEnvFromImport(ctx context.Context, i *cloudv1.Import) (Env, error) {
	providerType, err := GetProviderType(i)
	if err != nil {
		return nil, errors.Annotate(err, "get provider type failed")
	}
	iog, err := GetImportObjectGroup(ctx, b.infraClusterCli, i)
	if err != nil {
		return nil, errors.Annotate(err, "get import object group failed")
	}
	return NewGardenerShootEnv(ctx, b.gardenerEnv, iog, b.infraClusterCli, providerType)
}