package apiserver

import (
	"fmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/controller"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/label"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sync"
	"time"
)

type ClusterStatusMonitor struct {
	statusCache          sync.Map
	storageCache         sync.Map
	statusChangeListener sync.Map

	store *Store
}

type clusterStatus struct {
	status        string
	lastCheckTime time.Time
}

//cluster configurations that can not be updated, we only get it from helm once
type unchangeableClusterValue struct {
	pdStorageClass      string
	tikvStorageClass    string
	monitorStorageClass string
}

type NotifyFunc func(name, newStatus, oldStatus string)

func (monitor *ClusterStatusMonitor) Start(stop chan struct{}) {
	go func() {
		monitor.startClusterController(stop)
		ticker := time.NewTicker(time.Second * 1800)
		for {
			select {
			case <-stop:
				return
			case <-ticker.C:
				monitor.statusCache.Range(monitor.updateClusterStatus)
			}
		}
	}()
}

func (monitor *ClusterStatusMonitor) AddStatusChangeListener(notifyFunc NotifyFunc) {
	monitor.statusChangeListener.Store(notifyFunc, struct{}{})
}

func (monitor *ClusterStatusMonitor) GetClusterStatus(tc *v1alpha1.TidbCluster) string {
	name := tc.Namespace
	cs, ok := monitor.statusCache.Load(name)
	if ok {
		return cs.(*clusterStatus).status
	}
	return monitor.tryStoreClusterStatus(tc)
}

func (monitor *ClusterStatusMonitor) getClusterStatusWithTimeout(tc *v1alpha1.TidbCluster) string {
	c := make(chan string, 1)
	go func() {
		st, _ := monitor.store.Controller().GetClusterStatus(tc)
		c <- st
	}()
	select {
	case status := <-c:
		return status
	case <-time.After(time.Second * 2):
		return controller.InstanceStatusStatusFailed
	}
}

func (monitor *ClusterStatusMonitor) GetPDStorageClass(name string) string {
	return monitor.getStorageClassByComponent(name, label.PDLabelVal)
}

func (monitor *ClusterStatusMonitor) GetTiKVStorageClass(name string) string {
	return monitor.getStorageClassByComponent(name, label.TiKVLabelVal)
}

func (monitor *ClusterStatusMonitor) GetMonitorStorageClass(name string) string {
	return monitor.getStorageClassByComponent(name, "monitor")
}

func (monitor *ClusterStatusMonitor) getStorageClassByComponent(name string, component string) string {
	values, err := monitor.getUnchangeableClusterValue(name)
	if err != nil {
		return "UnknownStorageClass"
	}
	switch component {
	case label.PDLabelVal:
		return values.pdStorageClass
	case label.TiKVLabelVal:
		return values.tikvStorageClass
	default:
		return values.monitorStorageClass
	}
}

func (monitor *ClusterStatusMonitor) getUnchangeableClusterValue(name string) (*unchangeableClusterValue, error) {
	cachedValues, ok := monitor.storageCache.Load(name)
	if ok {
		return cachedValues.(*unchangeableClusterValue), nil
	}
	tidbClusterValues, err := monitor.store.Operator.GetTiDBCluster(name)
	if err != nil {
		msg := fmt.Errorf("failed to get tidb cluster. err: %v", err)
		log.Error("failed to get tidb cluster.", zap.String("instance", name), zap.Error(err))
		return nil, errors.NewInternalError(msg)
	}
	m, err := monitor.store.TidbClusterClient.PingcapV1alpha1().TidbMonitors(name).Get(name, metav1.GetOptions{})
	if err != nil {
		msg := fmt.Errorf("failed to get monitor. err: %v", err)
		log.Error("failed to get monitor", zap.String("instance", name), zap.Error(err))
		return nil, errors.NewInternalError(msg)
	}

	newCacheItem := &unchangeableClusterValue{
		pdStorageClass:      *tidbClusterValues.Spec.PD.StorageClassName,
		tikvStorageClass:    *tidbClusterValues.Spec.TiKV.StorageClassName,
		monitorStorageClass: m.Spec.Storage,
	}
	monitor.storageCache.Store(name, newCacheItem)
	return newCacheItem, nil
}

func (monitor *ClusterStatusMonitor) updateClusterStatus(key, value interface{}) bool {
	name := key.(string)
	status := value.(*clusterStatus)
	oldStatus := status.status

	tc, err := monitor.store.GetTiDBCluster(name)
	if err == nil {
		status.status = monitor.getClusterStatusWithTimeout(tc)
		status.lastCheckTime = time.Now()
		monitor.fireStatusChangedEvent(name, status, oldStatus)
	}
	return true
}

func (monitor *ClusterStatusMonitor) fireStatusChangedEvent(name string, status *clusterStatus, oldStatus string) {
	if status.status != oldStatus {
		log.Info("cluster status is changed, notify listeners", zap.String("cluster", name))
		monitor.statusChangeListener.Range(func(key, value interface{}) bool {
			key.(NotifyFunc)(name, status.status, oldStatus)
			return true
		})
	}
}

func (monitor *ClusterStatusMonitor) startClusterController(stop chan struct{}) {
	startController(monitor.store,
		func(s string) { monitor.statusCache.Delete(s) },
		func(tc *v1alpha1.TidbCluster) { monitor.tryStoreClusterStatus(tc) },
		stop)
}

func (monitor *ClusterStatusMonitor) tryStoreClusterStatus(tc *v1alpha1.TidbCluster) string {
	name := tc.Namespace
	status := monitor.getClusterStatusWithTimeout(tc)
	if status != controller.InstanceStatusStatusCreating {
		monitor.statusCache.Store(name, &clusterStatus{
			status:        status,
			lastCheckTime: time.Now(),
		})
	}
	return status
}