package meta

import (
	"errors"
	"fmt"
	"github.com/guo-shaoge/tiflash-autoscale/pkg/metric"
	"sync"
	"sync/atomic"
	"time"
)

type PodsFilter func(*PodDesc) bool

type TenantStatus uint

const (
	TenantStatusResumed TenantStatus = iota
	TenantStatusResuming
	TenantStatusPaused
)

func (s TenantStatus) ToString() string {
	switch s {
	case TenantStatusResumed:
		return "resumed"
	case TenantStatusResuming:
		return "resuming"
	case TenantStatusPaused:
		return "paused"
	default:
		panic("unexpected tenant status")
	}
}

type TenantDesc struct {
	name   string
	rgName metric.ResourceGroupNameType
	mu     struct {
		sync.RWMutex
		status      TenantStatus
		podsMap     PodsMap
		enableShare bool
		// use ch instead of cv because channel support timeout.
		// but it may create many channels.
		allocPodsFinishCh chan bool
		lastResumedTs     time.Time
		lastMergeTs       time.Time
		// TODO init
		lastMemLimitRecoveryTs time.Time
		lastEvictTs            time.Time
	}

	isAllocatingPods atomic.Bool
}

// NewTenatnDesc returns new tenant
func NewTenantDesc(name string, rgName metric.ResourceGroupNameType, enableShare bool) *TenantDesc {
	res := &TenantDesc{
		name:   name,
		rgName: rgName,
	}
	res.mu.status = TenantStatusPaused
	res.mu.podsMap = make(PodsMap)
	res.mu.enableShare = enableShare
	res.mu.allocPodsFinishCh = make(chan bool)
	return res
}

func (t *TenantDesc) GetName() string {
	return t.name
}

func (t *TenantDesc) GetRGName() metric.ResourceGroupNameType {
	return t.rgName
}

// ToString return tenant info by string.
func (t *TenantDesc) ToString() string {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return fmt.Sprintf("name: %s, share: %v, status: %s", t.name, t.mu.enableShare, t.mu.status.ToString())
}

func (t *TenantDesc) ToResumed() {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.mu.status = TenantStatusResumed
	t.mu.lastResumedTs = time.Now()
}

// // SyncResume will return `minPods`(it's just a hint).
// // 1. get current podsMap and return (will delete unusable pods) (will change to TenantStatusResumed)
// // 2. if not enough, take pods from podMgr
// // Called by Http.SyncResume()
// func (t *TenantDesc) SyncResume(ctx context.Context, minPod int, sourceMod string, traceId string) (res PodsMap, err error) {
// 	if minPod <= 0 {
// 		return nil, errors.New("min pod should be greater than zero when SyncResume")
// 	}
//
// 	_ = t.ToResumingIfNecessary()
// 	defer func() {
// 		// change to resumed if got any pods.
// 		if len(res) > 0 {
// 			t.ToResumed()
// 		}
// 	}()
//
// 	enableShare := t.GetEnableShare()
//
// 	_, usablePodsMap := t.RemoveUnusablePods()
// 	if len(usablePodsMap) >= minPod {
// 		return usablePodsMap, nil
// 	}
//
// 	newPods, err := podMgr.SyncTakePods(minPod - len(usablePodsMap), t.name, enableShare, sourceMod, traceId)
// 	if err != nil {
// 		return nil, err
// 	}
//
// 	t.addPodsByDesc(newPods)
// 	_, usablePodsMap = t.RemoveUnusablePods()
// 	return usablePodsMap, nil
// }

// return true if pause ok.
// called by TenantMgr.analyzeTaskLoop()
func (t *TenantDesc) AutoPause(reason string) (bool, string) {
	t.mu.Lock()
	defer t.mu.Unlock()

	var errMsg string
	switch t.mu.status {
	case TenantStatusResumed:
		for name, podDesc := range t.mu.podsMap {
			if _, _, err := podDesc.Deref(t.name, reason); err != nil {
				errMsg += err.Error()
			}
			delete(t.mu.podsMap, name)
		}
		t.mu.status = TenantStatusPaused
		return true, errMsg
	case TenantStatusResuming:
		return false, ""
	case TenantStatusPaused:
		return true, ""
	default:
		panic("unexpected tenant status")
	}
}

// // called by resumed tenant to resize pods. difference between TenantDesc.SyncResume:
// //    1. this func can also deref pods.
// //    2. this func is only for resumed tenants.
// // called by TenantMgr.analyzeTaskLoop()
// func (t *TenantDesc) SyncResizePods(expectedNum uint, sourceMod string, tarceId string) (delta int, err error) {
// 	lockChecker := func() bool {
// 		t.mu.Lock()
// 		defer t.mu.Unlock()
// 		if t.mu.Status != TenantStatusResumed {
// 			return false
// 		}
// 		delta = len(t.mu.podsMap) - int(expectedNum)
// 		return true
// 	}
//
// 	needResize := lockChecker()
// 	if !needResize {
// 		return 0, nil
// 	}
//
// 	if expectedNum == 0 {
// 		return 0, fmt.Errorf("should not resize pod to zero for resumed tenant")
// 	}
//
// 	if delta == 0 {
// 		return 0, nil
// 	} else if delta > 0 {
// 		newPods, err := t.SyncAddPodsByNum(delta, sourceMod, traceId)
// 		return len(newPods), err
// 	} else {
// 		removed, errMsg := t.removePodsByNum(uint(-delta))
// 		return removed, errors.New(errMsg)
// 	}
// }

func (t *TenantDesc) CheckResumedAndGetPodsNum(expectedNum uint) (bool, int) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	if t.mu.status != TenantStatusResumed {
		return false, 0
	}
	delta := int(expectedNum) - len(t.mu.podsMap)
	return true, delta
}

// // call podMgr to take pods from podMgr.
// // called by ResizePods(called by tryScale) and tryEvict(to get new pods after tenant be evicted)
// func (t *TenantDesc) SyncAddPodsByNum(delta uint, sourceMod string, traceId string) ([]*PodDesc, error) {
// 	enableShare := t.GetEnableShare()
// 	res, err := podMgr.SyncTakePods(delta, t.name, t.enableShare, sourceMod, traceId)
// 	if err != nil {
// 		return nil, err
// 	}
// 	t.addPodsByDesc(res)
// 	return res, nil
// }

// ReplaceMergePods will replace all pods according to replaceInfo.
// called by TenantMgr.doAnalyzeMergeSharePods.
func (t *TenantDesc) ReplaceMergePods(replaceInfo map[string]*PodDesc) (string, error) {
	for fromPodName, toPodDesc := range replaceInfo {
		if !toPodDesc.IsUsable() {
			delete(replaceInfo, fromPodName)
		}
	}
	if len(replaceInfo) <= 0 {
		return "", nil
	}

	var replaceMsg string
	mergeTs := time.Now()
	fromPodsMap := make(PodsMap, len(replaceInfo))
	toPods := make([]*PodDesc, 0, len(replaceInfo))
	var err error
	func() {
		t.mu.Lock()
		defer t.mu.Unlock()
		replaceMsg += "before:"
		for podName, _ := range t.mu.podsMap {
			replaceMsg += fmt.Sprintf("%v,", podName)
		}
		for fromPodName, fromPodDesc := range t.mu.podsMap {
			toPodDesc, ok := replaceInfo[fromPodName]
			if !ok {
				continue
			}

			var i int
			maxIter := 5000
			for {
				// Why need for loop:
				// replace info: pod1 -> pod2; pod2 -> pod3
				// final: we need replace pod1 to pod3.
				if i >= maxIter {
					err = fmt.Errorf("the replace info recursive larget than max(%v)", maxIter)
					return
				}
				i++

				toPodName := toPodDesc.GetName()
				toPodDescIter, ok := replaceInfo[toPodName]
				if ok {
					toPodDesc = toPodDescIter
				} else {
					break
				}
			}

			fromPodsMap[fromPodName] = fromPodDesc
			toPods = append(toPods, toPodDesc)
			// santity check
			if toPodDesc.GetName() == fromPodDesc.GetName() {
				err = fmt.Errorf("should not replace the same pod %v", toPodDesc.GetName())
				return
			}
			delete(t.mu.podsMap, fromPodName)
			t.mu.podsMap[toPodDesc.GetName()] = toPodDesc
		}
		replaceMsg += ";after:"
		for podName, _ := range t.mu.podsMap {
			replaceMsg += fmt.Sprintf("%v,", podName)
		}
		replaceMsg += fmt.Sprintf(";fromPods len: %v", len(fromPodsMap))
		replaceMsg += fmt.Sprintf(";toPods len: %v", len(toPods))
	}()

	if err != nil {
		return "", err
	}

	var errMsg string
	for _, pod := range fromPodsMap {
		if _, _, err1 := pod.Deref(t.name, "ReplaceMergePods"); err1 != nil {
			errMsg += fmt.Sprintf("Deref %v failed: %v;", pod.GetName(), err1.Error())
		}
	}
	for _, pod := range toPods {
		// If this tenant already owned this pod, skip ref.
		if pod.HasTenant(t.name) {
			continue
		}
		if _, _, err1 := pod.Ref(t.name, "ReplaceMergePods"); err1 != nil {
			errMsg += fmt.Sprintf("Ref %v failed: %v;", pod.GetName(), err1.Error())
		}
		pod.UpdateLastMergeTs(mergeTs)
	}
	if len(errMsg) != 0 {
		return replaceMsg, errors.New(errMsg)
	}
	if len(fromPodsMap) > 0 {
		t.UpdateLastMergeTs(mergeTs)
	}
	// TODO respect init core/min core!!!!

	return replaceMsg, nil
}

// remove pods from podsMap and Deref it.
// called by TenantDesc.ResizePods(called by TenantMgr.analyzeTaskLoop).
func (t *TenantDesc) RemovePodsByNum(delta uint, reason string) (removed uint, errMsg string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	for podName, podDesc := range t.mu.podsMap {
		if removed < delta {
			_, _, err := podDesc.Deref(t.name, reason)
			if err != nil {
				errMsg += err.Error()
			}
			delete(t.mu.podsMap, podName)
			removed++
		} else {
			break
		}
	}
	return
}

func (t *TenantDesc) RemovePodByName(name string, reason string) error {
	t.mu.Lock()
	defer t.mu.Unlock()
	podDesc, ok := t.mu.podsMap[name]
	if ok {
		_, _, err := podDesc.Deref(t.name, reason)
		delete(t.mu.podsMap, name)
		return err
	}
	return nil
}

func (t *TenantDesc) RemovePodsByName(names []string, reason string) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	for _, podName := range names {
		podDesc, ok := t.mu.podsMap[podName]
		if ok {
			_, _, err := podDesc.Deref(t.name, reason)
			delete(t.mu.podsMap, podName)
			return err
		}
	}
	return nil
}

// IsResumed return bool
func (t *TenantDesc) IsResumed() bool {
	return t.GetStatus() == TenantStatusResumed
}

// GetStatus return status
func (t *TenantDesc) GetStatus() TenantStatus {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.mu.status
}

// GetStatusStr return status string.
func (t *TenantDesc) GetStatusStr() string {
	return t.GetStatus().ToString()
}

// TODO other places to call this
// RemoveUnsablePods will Deref all unusable pods.
func (t *TenantDesc) RemoveUnusablePods() (uint, PodsMap) {
	delNum, res, deleted := func() (uint, PodsMap, []*PodDesc) {
		// TODO lock nested!!!
		deleted := make([]*PodDesc, 0, 3)
		t.mu.Lock()
		defer t.mu.Unlock()

		var delNum uint
		for name, pod := range t.mu.podsMap {
			if !pod.IsUsable() {
				deleted = append(deleted, pod)
				delete(t.mu.podsMap, name)
				delNum++
			}
		}
		res := make(PodsMap, len(t.mu.podsMap))
		for k, v := range t.mu.podsMap {
			res[k] = v
		}
		return delNum, res, deleted
	}()
	for _, pod := range deleted {
		pod.Deref(t.name, "RemoveUnusablePods")
	}
	return delNum, res
}

// GetPodsNum return pods num.
// called by TenantMgr.analyzeTaskLoop(tryScale)
func (t *TenantDesc) GetPodsNum() int {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return len(t.mu.podsMap)
}

func (t *TenantDesc) GetUsablePodsNum() (cnt uint) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	for _, podDesc := range t.mu.podsMap {
		if podDesc.IsUsable() {
			cnt++
		}
	}
	return cnt
}

// GetCPUMetricWindow returns all pods' CPU metric window.
// called by TenantMgr.analyzeTaskLoop(tryEvict/doAnalyzeMergeSharePods)
func (t *TenantDesc) GetCPUMetricWindow() (podsWindow metric.PodsCPUMetricWindow) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	podsWindow = make(metric.PodsCPUMetricWindow, len(t.mu.podsMap))
	for podName, podDesc := range t.mu.podsMap {
		if podDesc.IsUsable() {
			podsWindow[podName] = podDesc.GetCPUMetricWindow()
		}
	}
	return podsWindow
}

// GetTaskCntMetricWindow returns all pods' taskCnt metric window.
// called by TenantMgr.analyzeTaskLoop(tryAutoPause)
func (t *TenantDesc) GetTaskCntMetricWindow() (podsWindow metric.PodsTaskCntMetricWindow) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	podsWindow = make(metric.PodsTaskCntMetricWindow, len(t.mu.podsMap))
	for podName, podDesc := range t.mu.podsMap {
		if podDesc.IsUsable() {
			podsWindow[podName] = podDesc.GetTaskCntMetricWindow()
		}
	}
	return podsWindow
}

// GetRUMetricWindow returns all pods' RU metric window.
// called by TenantMgr.analyzeTaskLoop(tryEvict)
func (t *TenantDesc) GetRUMetricWindow() (podsWindow metric.PodsRUMetricWindow) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	podsWindow = make(metric.PodsRUMetricWindow, len(t.mu.podsMap))
	for podName, podDesc := range t.mu.podsMap {
		if podDesc.IsUsable() {
			podsWindow[podName] = podDesc.GetRUMetricWindow()
		}
	}
	return podsWindow
}

// GetOOMMetricWindow returns all pods' OOM metric window.
// called by TenantMgr.analyzeTaskLoop(tryScale)
func (t *TenantDesc) GetOOMMetricWindow() (podsWindow metric.PodsOOMMetricWindow) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	podsWindow = make(metric.PodsOOMMetricWindow, len(t.mu.podsMap))
	for podName, podDesc := range t.mu.podsMap {
		if podDesc.IsUsable() {
			podsWindow[podName] = podDesc.GetOOMMetricWindow()
		}
	}
	return podsWindow
}

// HasSharingPods return true if got sharing pods.
// called by TenantMgr.analyzeTaskLoop(before tryEvict)
func (t *TenantDesc) HasSharingPods() bool {
	t.mu.RLock()
	defer t.mu.RUnlock()
	for _, pod := range t.mu.podsMap {
		if pod.IsSharing() {
			return true
		}
	}
	return false
}

// ToResumingIfNecessary change status to resuming.
// called by http SyncResume()
func (t *TenantDesc) ToResumingIfNecessary() (TenantStatus, bool) {
	t.mu.Lock()
	defer t.mu.Unlock()
	ori := t.mu.status
	if t.mu.status == TenantStatusPaused {
		t.mu.status = TenantStatusResuming
		return ori, true
	}
	return ori, false
}

// UpdateEnableShare will update enalbeShare, will be called in analyzeTaskLoop()
// called by TenantMgr.doAnalyzeMain to update periodically.
// return true when ori is share and new is not share.
func (t *TenantDesc) UpdateEnableShare(enable bool) bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	ori := t.mu.enableShare
	t.mu.enableShare = enable
	return ori == true && enable == false
}

// GetEnableShare return enableShare.
func (t *TenantDesc) GetEnableShare() bool {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.mu.enableShare
}

// HasSharingPods return true if got any sharing pods.
func (t *TenantDesc) HasShareingPods() bool {
	t.mu.RLock()
	defer t.mu.RUnlock()
	for _, pod := range t.mu.podsMap {
		if pod.IsSharing() {
			return true
		}
	}
	return false
}

// GetTopo return fqdn addr for all pods of this tenant.
func (t *TenantDesc) GetTopo() (res []string, names []string) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	for _, podDesc := range t.mu.podsMap {
		res = append(res, podDesc.GetIP())
		names = append(names, podDesc.GetName())
	}
	return res, names
}

// GetTopoFQDN return fqdn addr for all pods of this tenant.
func (t *TenantDesc) GetTopoFQDN() (res []string, names []string) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	for _, podDesc := range t.mu.podsMap {
		res = append(res, podDesc.GetFQDN())
		names = append(names, podDesc.GetName())
	}
	return res, names
}

func (t *TenantDesc) AddPodsByDesc(newPods []*PodDesc) {
	if len(newPods) <= 0 {
		return
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	for _, pod := range newPods {
		t.mu.podsMap[pod.GetName()] = pod
	}
}

// GetPods return all pods that filter return true.
func (t *TenantDesc) GetPods(filter PodsFilter) (res PodsMap) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	res = make(PodsMap)
	for k, v := range t.mu.podsMap {
		if filter == nil || filter(v) {
			res[k] = v
		}
	}
	return res
}

func (t *TenantDesc) GetPodByName(name string) *PodDesc {
	t.mu.RLock()
	defer t.mu.RUnlock()
	if podDesc, ok := t.mu.podsMap[name]; ok {
		return podDesc
	}
	return nil
}

func (t *TenantDesc) WaitAllocPodsFinish(timeout time.Duration) {
	var ch chan bool
	deadline := time.Now().Add(timeout)
	for {
		t.mu.Lock()
		if t.mu.status != TenantStatusResuming {
			t.mu.Unlock()
			return
		}
		ch = t.mu.allocPodsFinishCh
		t.mu.Unlock()

		select {
		case <-ch:
			continue
		case <-time.After(time.Until(deadline)):
			return
		}
	}
}

func (t *TenantDesc) BroadcastAllocPodsFinish() {
	t.mu.Lock()
	close(t.mu.allocPodsFinishCh)
	t.mu.allocPodsFinishCh = make(chan bool)
	t.mu.Unlock()
}

func (t *TenantDesc) ToResumedOrPaused() bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	if len(t.mu.podsMap) == 0 {
		t.mu.status = TenantStatusPaused
		return false
	} else {
		t.mu.status = TenantStatusResumed
		t.mu.lastResumedTs = time.Now()
		return true
	}
}

func (t *TenantDesc) GetLastResumedTs() time.Time {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.mu.lastResumedTs
}

func (t *TenantDesc) UpdateLastMergeTs(ts time.Time) {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.mu.lastMergeTs = ts
}

func (t *TenantDesc) GetLastMergeTs() time.Time {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.mu.lastMergeTs
}

func (t *TenantDesc) UpdateLastEvictTs(ts time.Time) {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.mu.lastEvictTs = ts
}

func (t *TenantDesc) GetLastEvictTs() time.Time {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.mu.lastEvictTs
}

func (t *TenantDesc) IsAllocatingPods() bool {
	return t.isAllocatingPods.Load()
}

func (t *TenantDesc) SetIsAllocatingPods(val bool) bool {
	return t.isAllocatingPods.CompareAndSwap(!val, val)
}

func (t *TenantDesc) UpdateMemLimitRecoveryTs() {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.mu.lastMemLimitRecoveryTs = time.Now()
}

func (t *TenantDesc) MemLimitRecoveryCoolDown(cooldown uint) (bool, string) {
	t.mu.RLock()
	ts := t.mu.lastMemLimitRecoveryTs
	t.mu.RUnlock()

	okTs := ts.Add(time.Duration(cooldown) * time.Second)
	return time.Now().After(okTs), fmt.Sprintf("last mem limit recovery: %v, ok ts: %v, cooldown sec: %v", ts, okTs, cooldown)
}

func (t *TenantDesc) DerefAllSharingPods(reason string) (errMsg string) {
	t.mu.Lock()
	defer t.mu.Unlock()

	for podName, podDesc := range t.mu.podsMap {
		if podDesc.IsSharing() {
			if _, _, err := podDesc.Deref(t.name, reason); err != nil {
				errMsg += err.Error()
			}
			delete(t.mu.podsMap, podName)
		}
	}
	return
}