package k8s

import (
	"context"
	"fmt"
	stdlog "log"
	"strings"
	"sync"

	"helm.sh/helm/v3/pkg/action"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/cli-runtime/pkg/genericclioptions"
	"k8s.io/client-go/rest"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/manager"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/namer"
)

// Manager holds the dataplane k8s controller-runtime manager
// It's used for reconciler to build dataplane k8s client cache
type Manager struct {
	scheme      *runtime.Scheme
	infraClient client.Client // infra-apiserver kube client
	registers   []DataplaneReconcilerRegister

	mu         sync.Mutex
	dataplanes map[string]*DataplaneContext

	errors chan error
}

type DataplaneReconcilerRegister interface {
	Register(nctx *DataplaneContext) error
}

// DataplaneContext holds a dataplane k8s controller-runtime manager
// it implements the ClientGetter, so tasks can use this to retrieve cached k8s client
type DataplaneContext struct {
	Manager     manager.Manager
	KubeCluster KubeCluster
	actionCfg   *action.Configuration

	cancel context.CancelFunc
}

type KubeCluster interface {
	// GetName returns the dataplane k8s cluster name, it's used as the global map key
	GetName() string
	// GetAssumeRole returns the assume role info for aws
	GetAssumeRole() (*AssumeRole, error)
	// GetConfig returns a rest.Config, used to build k8s client and helm client
	GetConfig(context.Context, client.Client, *runtime.Scheme) (*rest.Config, error)
}

type NetworkKubeCluster struct {
	*cloudv1.Network
	// optional, take priority over the .spec.assumeRoleARN in the ProviderConfig referenced by cloudv1.Network.spec.providerConfigReference
	AWS *AssumeRole
}

type ClusterKubeCluster struct {
	*cloudv1.Cluster
	AWS *AssumeRole
}

type AssumeRole struct {
	RoleARN    string
	ExternalID string
}

var (
	// NetworkKubeCluster implements KubeCluster interface
	// It's used in network-controller
	_ KubeCluster = NetworkKubeCluster{}
	// ClusterKubeCluster implments KubeCluster interface
	// It's used in cluster-controller for new 3-layer architecture
	_ KubeCluster = ClusterKubeCluster{}
)

func (c NetworkKubeCluster) GetName() string {
	return client.ObjectKeyFromObject(c.Network).String()
}

func (c NetworkKubeCluster) GetAssumeRole() (*AssumeRole, error) {
	if c.AWS != nil {
		return c.AWS, nil
	}

	spec, err := eksv1alpha1.EKSNetworkSpecForNetwork(c.Network)
	if err != nil {
		return nil, err
	}
	return &AssumeRole{
		RoleARN:    spec.AssumeRoleArn,
		ExternalID: spec.AssumeExternalId,
	}, nil
}

func (c NetworkKubeCluster) GetConfig(ctx context.Context, infraClient client.Client, scheme *runtime.Scheme) (*rest.Config, error) {
	return ConfigFromNetwork(ctx, infraClient, c.Network, c.AWS, scheme)
}

func (c ClusterKubeCluster) GetName() string {
	return c.Cluster.Spec.Kubernetes.Name
}

func (c ClusterKubeCluster) GetAssumeRole() (*AssumeRole, error) {
	if c.AWS != nil {
		return c.AWS, nil
	}
	return nil, fmt.Errorf("non aws provider")
}

func (c ClusterKubeCluster) GetConfig(ctx context.Context, infraClient client.Client, scheme *runtime.Scheme) (*rest.Config, error) {
	k := c.Cluster.Spec.Kubernetes
	pt := c.Cluster.GetProviderType()
	switch pt {
	case cloudv1.ProviderTypeEKS:
		if c.AWS == nil {
			return nil, fmt.Errorf("assume role unspecified for provider type %s", pt)
		}
		region := c.Cluster.GetRegion()
		return dataplane.NewRESTConfigForEKS(k.Name, k.Endpoint, region, c.AWS.RoleARN, c.AWS.ExternalID, k.CAData)
	case cloudv1.ProviderTypeGKE:
		return dataplane.NewRESTConfigForGKE(k.Name, k.Endpoint, k.CAData)
	}

	// if kubeconfigSecretRef is set, get the kubeconfig from the secret
	if k.KubeConfigSecretRef != nil {
		dataplaneKubeData, err := dataplane.GetDataplaneKubeConfigFromMetaSecret(ctx, infraClient, k.KubeConfigSecretRef)
		if err != nil {
			return nil, fmt.Errorf("failed to get kubeconfig from secret %s/%s, err: %w", k.KubeConfigSecretRef.Namespace, k.KubeConfigSecretRef.Name, err)
		}
		return dataplane.NewRESTConfigForKubeConfig(dataplaneKubeData)
	}

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

// NewManager creates a new manager.
func NewManager(infraClient client.Client, registers []DataplaneReconcilerRegister, scheme *runtime.Scheme) *Manager {
	return &Manager{
		scheme:      scheme,
		infraClient: infraClient,
		dataplanes:  make(map[string]*DataplaneContext),
		errors:      make(chan error, 1),
		registers:   registers,
	}
}

func (m *Manager) ClientGetter(ctx context.Context, name string) (ClientGetter, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	nctx, ok := m.dataplanes[name]
	if !ok {
		return nil, fmt.Errorf("dataplane cluster %s unregistered", name)
	}
	return nctx, nil
}

// DataplaneContext implements ClientGetter
var _ ClientGetter = &DataplaneContext{}

func (c *DataplaneContext) GetClient(ctx context.Context) (client.Client, error) {
	// the manager.GetClient() is lightweight
	// so we don't need to cache it in DataplaneContext
	return c.Manager.GetClient(), nil
}

func (c *DataplaneContext) GetHelmConfig(ctx context.Context) (*action.Configuration, error) {
	if c.actionCfg != nil {
		// use cached action.Configuration to avoid initialize a action.Configuration every time
		return c.actionCfg, nil
	}
	restCfg := c.Manager.GetConfig()
	var actionCfg action.Configuration
	// https://github.com/helm/helm/issues/11047
	// helm client can't use rest.Config directly
	// The Manager is constructed from rest.Config and rest.Config is created from configFlags
	// So here we just reverse the process to construct configFlags from rest.Config
	configFlags := genericclioptions.NewConfigFlags(false)
	configFlags.APIServer = &restCfg.Host
	configFlags.Insecure = pointer.Bool(true)
	configFlags.WrapConfigFn = func(_ *rest.Config) *rest.Config {
		return restCfg
	}
	err := actionCfg.Init(
		configFlags,
		HelmNamespace,
		"configmap", // use configmap as the driver
		stdlog.Printf,
	)
	if err != nil {
		return nil, err
	}
	c.actionCfg = &actionCfg
	return c.actionCfg, nil
}

// Errors return the errors channel.
// If some error happens, it means some network manage fails to run
// we should better just exit.
func (m *Manager) Errors() <-chan error {
	return m.errors
}

// Register registers a running network with the manager.
// It is safe to call this method multiple times with same network.
func (m *Manager) Register(ctx context.Context, kc KubeCluster) error {
	nkey := kc.GetName()
	m.mu.Lock()
	defer m.mu.Unlock()
	_, ok := m.dataplanes[nkey]
	if ok {
		return nil
	}

	logger := log.FromContext(ctx)
	logger.Info("registering to manager", "kubernetes", nkey)

	cfg, err := kc.GetConfig(ctx, m.infraClient, m.scheme)
	if err != nil {
		if apierrors.IsNotFound(err) &&
			strings.Contains(err.Error(), "Cluster.eks.aws.crossplane.io") ||
			strings.Contains(err.Error(), "Cluster.container.gcp.crossplane.io") {
			logger.Info("Dataplane K8s cluster is not found, skip network manager registration")
			return nil
		}
		return err
	}

	mgr, err := manager.New(cfg, manager.Options{Scheme: m.scheme, LeaderElection: false, Metrics: metricsserver.Options{BindAddress: "0"}})
	if err != nil {
		return err
	}

	nctx, cancel := context.WithCancel(context.Background())
	dataplaneContext := &DataplaneContext{
		Manager:     mgr,
		KubeCluster: kc,
		cancel:      cancel,
	}

	for _, register := range m.registers {
		err := register.Register(dataplaneContext)
		if err != nil {
			return err
		}
	}

	go func() {
		logger.Info("start manager for dataplane cluster", "kubernetes", nkey)
		err := mgr.Start(nctx)
		if err != nil {
			m.errors <- err
		}
	}()
	m.dataplanes[nkey] = dataplaneContext

	return nil
}

// Deregister deregisters a dataplane cluster and controller with the manager.
// When we delete or hibernates a dataplane kubernetes cluster,
// we need to deregister it from cache and stop the dataplane controller
// It's safe to deregister the kubernetes cluster multiple times
func (m *Manager) Deregister(ctx context.Context, nkey string) {
	m.mu.Lock()
	defer m.mu.Unlock()

	nctx, ok := m.dataplanes[nkey]
	if !ok {
		return
	}

	log.FromContext(ctx).Info("deregistering from manager", "kubernetes", nkey)

	nctx.cancel()
	delete(m.dataplanes, nkey)
}

// Close the manager.
func (m *Manager) Close() {
	m.mu.Lock()
	defer m.mu.Unlock()

	for _, nctx := range m.dataplanes {
		nctx.cancel()
	}
}

// ConfigFromNetwork retrieves a k8s rest.Config from the given network (dataplane)
// The runtime.Scheme is used for dataplane apiserver interaction
func ConfigFromNetwork(ctx context.Context, infraClient client.Client, n *cloudv1.Network, awsAssumeRoleOptional *AssumeRole, scheme *runtime.Scheme) (*rest.Config, error) {
	opts := client.Options{
		Scheme: scheme,
	}
	var g ClientManager
	switch provider := n.Spec.Provider; provider {
	case consts.ProviderTypeEKS:
		g = NewEKSClientManager(infraClient, opts, withAWSAssumeRole(awsAssumeRoleOptional))
	case consts.ProviderTypeGKE:
		g = NewGKEClientManager(infraClient, opts)
	default:
		return nil, fmt.Errorf("unknown provider: %s", provider)
	}
	key := types.NamespacedName{
		Name: namer.LegacyBuildClusterNameByNetwork(n),
	}
	cfg, err := g.GetConfig(ctx, key)
	if err != nil {
		return nil, err
	}

	return cfg, nil
}