package dataplane

import (
	"context"
	"encoding/base64"
	"fmt"
	"sync"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// TODO(shenjun): no need to use cache if dataplane repo is datapalne mrg repo
type managerImpl struct {
	dataplaneRepo repo.DataplaneRepo

	lock sync.RWMutex
	// host to client
	cache map[string]cacheItem

	// name to host
	nameToHost  map[string]string
	hostToNames map[string]sets.Set[string]

	ctx context.Context
}

type cacheItem struct {
	cancelFn context.CancelFunc
	client   client.Client
	restCfg  *rest.Config
}

func NewManager(localKubeCfg *rest.Config, dataplaneRepo repo.DataplaneRepo) Manager {
	m := &managerImpl{
		dataplaneRepo: dataplaneRepo,
		cache:         make(map[string]cacheItem),
		nameToHost:    map[string]string{},
		hostToNames:   map[string]sets.Set[string]{},
		ctx:           context.Background(),
	}

	err := m.register(LocalName, localKubeCfg)
	if err != nil {
		panic("failed to register local client")
	}

	return m
}

func (m *managerImpl) isRegistered(name string) bool {
	_, ok := m.getCacheItem(name)
	return ok
}

func (m *managerImpl) register(name string, cfg *rest.Config) error {
	m.lock.Lock()
	defer m.lock.Unlock()

	item, ok := m.cache[cfg.Host]
	if ok {
		m.update(name, cfg.Host, nil)
		return nil
	}

	ctx, cancel := context.WithCancel(m.ctx)
	c, err := k8s.NewClient(ctx, cfg)
	if err != nil {
		cancel()
		tilog.Error(m.ctx, "new k8s client with rest config failed", zap.String("cfg", cfg.String()), zap.Error(err))
		return err
	}

	item.client = c
	item.cancelFn = cancel

	item.restCfg = cfg

	m.update(name, cfg.Host, &item)

	return nil
}

func (m *managerImpl) GetClient(name string) (client.Client, error) {
	item, ok := m.getCacheItem(name)
	if ok {
		// local client not need to check
		if name == LocalName {
			return item.client, nil
		}

		// check if client is still alive
		if healthCheck(m.ctx, item) {
			return item.client, nil
		}

		tilog.Info(m.ctx, "client is dead, re-registering", zap.String("name", name))
		m.delete(name, item.restCfg.Host)
	}

	// register client if not found
	if err := m.registerRESTConfig(name); err != nil {
		return nil, err
	}

	item, ok = m.getCacheItem(name)
	if !ok {
		return nil, fmt.Errorf("cfg must be registered before get: %s", name)
	}

	return item.client, nil
}

func (m *managerImpl) GetRestConfig(name string) (*rest.Config, error) {
	item, ok := m.getCacheItem(name)
	if ok {
		// local client not need to check
		if name == LocalName {
			return item.restCfg, nil
		}

		// check if client is still alive
		if healthCheck(m.ctx, item) {
			return item.restCfg, nil
		}

		tilog.Info(m.ctx, "client is dead, re-registering", zap.String("name", name))
		m.delete(name, item.restCfg.Host)
	}

	// register client if not found
	if err := m.registerRESTConfig(name); err != nil {
		return nil, err
	}

	item, ok = m.getCacheItem(name)
	if !ok {
		return nil, fmt.Errorf("cfg must be registered before get: %s", name)
	}

	return item.restCfg, nil
}

func (m *managerImpl) registerRESTConfig(name string) error {
	var (
		err        error
		kubeConfig *rest.Config
	)

	dataplaneProvider, err := m.dataplaneRepo.LoadByName(m.ctx, name)
	if err != nil {
		if status.Code(err) == codes.NotFound || status.Code(err) == codes.Unavailable {
			tilog.Warn(m.ctx, "dataplane provider not found or unavailable, maybe not ready",
				zap.String("dataplaneName", name),
				zap.Error(err))
			return fmt.Errorf("dataplane provider not ready: %w: %w", ErrDataplaneProviderNotReady, err)
		}

		tilog.Error(m.ctx, "failed to load dataplane provider", zap.Error(err))
		return err
	}

	// double check
	if m.isRegistered(dataplaneProvider.GetName()) {
		return nil
	}

	switch dataplaneProvider.GetType() {
	case apiv1.ProviderType_AWS:
		if dataplaneProvider.GetKubeconfig() != "" {
			tilog.Info(m.ctx, "registering aws dataplane provider, use kubeconfig directly",
				zap.String("dataplane", dataplaneProvider.GetName()))
			decoded, err := base64.StdEncoding.DecodeString(dataplaneProvider.GetKubeconfig())
			if err != nil {
				return fmt.Errorf("failed to base64 decode kubeconfig: %w", err)
			}
			kubeConfig, err = NewRESTConfigForKubeConfig(decoded)
			if err != nil {
				return fmt.Errorf("failed to create aws rest config for kubeconfig: %w", err)
			}
		}
	case apiv1.ProviderType_GCP:
		tilog.Info(m.ctx, "registering gcp dataplane provider", zap.String("dataplane", dataplaneProvider.GetName()))
		decoded, err := base64.StdEncoding.DecodeString(dataplaneProvider.GetKubeconfig())
		if err != nil {
			return fmt.Errorf("failed to base64 decode kubeconfig: %w", err)
		}
		kubeConfig, err = NewRESTConfigForKubeConfig(decoded)
		if err != nil {
			return fmt.Errorf("failed to create gcp rest config for kubeconfig: %w", err)
		}
	case apiv1.ProviderType_AZURE:
		tilog.Info(m.ctx, "registering azure dataplane provider", zap.String("dataplane", dataplaneProvider.GetName()))
		decoded, err := base64.StdEncoding.DecodeString(dataplaneProvider.GetKubeconfig())
		if err != nil {
			return fmt.Errorf("failed to base64 decode kubeconfig: %w", err)
		}
		kubeConfig, err = NewRESTConfigForKubeConfig(decoded)
		if err != nil {
			return fmt.Errorf("failed to create azure rest config for kubeconfig: %w", err)
		}
	case apiv1.ProviderType_ALICLOUD:
		if dataplaneProvider.GetKubeconfig() != "" {
			tilog.Info(m.ctx, "registering aws dataplane provider, use kubeconfig directly",
				zap.String("dataplane", dataplaneProvider.GetName()))
			decoded, err := base64.StdEncoding.DecodeString(dataplaneProvider.GetKubeconfig())
			if err != nil {
				return fmt.Errorf("failed to base64 decode kubeconfig: %w", err)
			}
			kubeConfig, err = NewRESTConfigForKubeConfig(decoded)
			if err != nil {
				return fmt.Errorf("failed to create alicloud rest config for kubeconfig: %w", err)
			}
		}
	default:
		return fmt.Errorf("unsupported dataplane provider: %s", dataplaneProvider.GetName())
	}
	return m.register(dataplaneProvider.GetName(), kubeConfig)
}

func healthCheck(ctx context.Context, item *cacheItem) bool {
	// use no cache client to avoid cache issue
	clientset, _ := kubernetes.NewForConfig(item.restCfg)
	pods, err := clientset.CoreV1().Pods(metav1.NamespaceSystem).List(ctx, metav1.ListOptions{Watch: false})
	if err != nil {
		tilog.Error(ctx, "check dataplane client failed", zap.String("msg", "failed to list pods in kube-system"), zap.Error(err))
		return false
	}
	if len(pods.Items) == 0 {
		tilog.Error(ctx, "check dataplane client failed", zap.String("msg", "no pods found in kube-system namespace"))
		return false
	}

	return true
}

func (m *managerImpl) getCacheItem(name string) (*cacheItem, bool) {
	m.lock.RLock()
	defer m.lock.RUnlock()
	host, ok := m.nameToHost[name]
	if !ok {
		return nil, false
	}

	item, ok := m.cache[host]
	return &item, ok
}

func (m *managerImpl) update(name, host string, item *cacheItem) {
	oldHost, ok := m.nameToHost[name]
	if ok && oldHost != host {
		m.deleteLocked(name, oldHost)
	}

	m.nameToHost[name] = host
	s, ok := m.hostToNames[host]
	if !ok {
		s = sets.New[string]()
	}
	s.Insert(name)
	m.hostToNames[host] = s

	if item != nil {
		m.cache[host] = *item
	}
}

func (m *managerImpl) delete(name, host string) {
	m.lock.Lock()
	defer m.lock.Unlock()
	m.deleteLocked(name, host)
}

// deleteLocked is the internal delete method that assumes the lock is already held.
func (m *managerImpl) deleteLocked(name, host string) {
	s, ok := m.hostToNames[host]
	if ok {
		s.Delete(name)
	}

	delete(m.hostToNames, host)
	item, ok := m.cache[host]
	if ok {
		// call cancel fn to stop client cache
		item.cancelFn()
	}
	delete(m.cache, host)
	delete(m.nameToHost, name)
}