package meta

import (
	"errors"
	"fmt"
	"github.com/guo-shaoge/tiflash-autoscale/pkg/metric"
	rpc "github.com/guo-shaoge/tiflash-autoscale/pkg/rpc/supervisorclient"
	"github.com/guo-shaoge/tiflash-autoscale/pkg/util/log/logger"
	"github.com/guo-shaoge/tiflash-autoscale/supervisor_proto/supervisor"
	"go.uber.org/zap"
	"strings"
	"sync"
	"time"
)

type PodsMap map[string]*PodDesc
type PodStatus uint

const (
	PodStatusReadyUnassign PodStatus = iota
	PodStatusUnassigning
	PodStatusUnassigned

	PodStatusRunning
	PodStatusSharing

	// TODO
	PodStatusLogicalError

	// when pod cannot recovery(unassign fail too many times),
	// action: will call k8s api to delete it
	PodStatusNeedDelete

	// k8s watcher got DeleteEvent, and will be tagged PodStatusDeleted.
	// action: will delete from podsMap.
	PodStatusDeleted

	PodStatusPending
)

// ToString() return string value of PodStatus.
func (s PodStatus) ToString() string {
	switch s {
	case PodStatusReadyUnassign:
		return "ready_unassign"
	case PodStatusUnassigning:
		return "unassigning"
	case PodStatusUnassigned:
		return "unassigned"
	case PodStatusRunning:
		return "running"
	case PodStatusSharing:
		return "sharing"
	case PodStatusLogicalError:
		return "logical_error"
	case PodStatusDeleted:
		return "delete"
	case PodStatusPending:
		return "pending"
	case PodStatusNeedDelete:
		return "need_delete"
	default:
		panic("unknown pod status")
	}
}

type PodDesc struct {
	name string
	// namespace of cloneset
	namespace string
	mu        struct {
		sync.RWMutex
		ip string

		status PodStatus

		// Record when the pod starts to become pending status,
		// so we can decide if this pod is pending too long and we can remove it.
		lastPendingTs time.Time

		mainTenant  string
		guestTenant map[string]struct{}

		cpuWindow     metric.CPUMetricWindow
		ruWindow      metric.RUMetricWindow
		oomWindow     metric.OOMMetricWindow
		taskCntWindow metric.TaskCntMetricWindow

		lastShareTs time.Time
		lastMergeTs time.Time
	}
}

// NewPodDesc return a new Pod.
func NewPodDesc(podName string, podIP string, namespace string, pending bool) *PodDesc {
	res := &PodDesc{
		name:      podName,
		namespace: namespace,
	}
	if pending {
		res.mu.status = PodStatusPending
	} else {
		res.mu.status = PodStatusUnassigned
	}
	res.mu.ip = podIP
	now := time.Now()
	res.mu.lastShareTs = now
	res.mu.lastMergeTs = now
	res.mu.guestTenant = make(map[string]struct{})
	// skip init lastPendingTs, so it's a min ts.
	return res
}

func (p *PodDesc) GetName() string {
	return p.name
}

// ToUnassignedStatus change status to unassigned.
// called by UnassigningPool.AddAndTriggerUnassign()
func (p *PodDesc) ToUnassigned() bool {
	p.mu.Lock()
	defer p.mu.Unlock()

	if p.mu.status == PodStatusUnassigning {
		p.mu.status = PodStatusUnassigned
		return true
	}
	return false
}

func (p *PodDesc) ToLogicalError() {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.mu.status = PodStatusLogicalError
}

func (p *PodDesc) ToDeleted() {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.mu.status = PodStatusDeleted
}

func (p *PodDesc) ToPending() {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.mu.status = PodStatusPending

	// p.mu.lastPendingTs == time.Time{} means this pod is ok before,
	// this is the first time to become pending.
	// if branch is to avoid a pod keep receiving pending event, and keep updating lastPendingTs.
	if p.mu.lastPendingTs.IsZero() {
		p.mu.lastPendingTs = time.Now()
	}
}

func (p *PodDesc) ToNeedDelete() {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.mu.status = PodStatusNeedDelete
}

func (p *PodDesc) GetLastPendingTs() time.Time {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.mu.lastPendingTs
}

// GetStatus return pod status.
func (p *PodDesc) GetStatus() PodStatus {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.mu.status
}

func (p *PodDesc) GetStatusStr() string {
	return p.GetStatus().ToString()
}

// SharingRef is the combine of okToShareWithoutLock and refWithoutLock(), to make they as atomic.
// 1. check if pod is ok to share.
// 2. if ok to share, call Ref() and mark share timestamp.
// called by PodMgr.SyncTakePods()(called by SharePool.tryTake())
func (p *PodDesc) SharingRef(tenantName string, shareTs time.Time,
	shareLowCPULimit float64, shareCoolDownSec uint,
	metricWindowSec uint, metricIntervalSec uint, coreNumPerPod uint,
	stillShareWhenMetricNotEnough bool) (string, bool, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	msg, ok := p.okToShareWithoutLock(shareLowCPULimit, shareCoolDownSec,
		metricWindowSec, metricIntervalSec, coreNumPerPod, stillShareWhenMetricNotEnough)
	if !ok {
		return msg, false, nil
	}
	_, _, err := p.refWithoutLock(tenantName)
	if err != nil {
		return "", false, err
	}

	p.updateLastShareTsWithoutLock(shareTs)
	return "", true, nil
}

// Ref ref one pod.
// called by PodMgr.syncRemovePodsFromWarmupPool and
func (p *PodDesc) Ref(tenantName string, reason string) (PodStatus, PodStatus, error) {
	logger.RootLogger.Debug("pod Ref", zap.String("tenant", tenantName),
		zap.String("pod", p.name), zap.String("reason", reason))
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.refWithoutLock(tenantName)
}

func (p *PodDesc) refWithoutLock(tenantName string) (PodStatus, PodStatus, error) {
	from := p.mu.status
	switch p.mu.status {
	case PodStatusReadyUnassign:
		p.mu.status = PodStatusRunning
		p.mu.mainTenant = tenantName
		to := p.mu.status
		return from, to, nil
	case PodStatusUnassigning:
		to := from
		return from, to, errors.New("cannot ref Unassigning pod")
	case PodStatusUnassigned:
		if len(p.mu.mainTenant) != 0 {
			return from, from, fmt.Errorf("pod is Unassigned, but mainTenant(%s) is not empty", p.mu.mainTenant)
		}
		if len(p.mu.guestTenant) != 0 {
			return from, from, fmt.Errorf("pod is Unassigned, but guestTenant(%v) is not empty", p.mu.guestTenant)
		}
		p.mu.status = PodStatusRunning
		p.mu.mainTenant = tenantName
		to := p.mu.status
		return from, to, nil
	case PodStatusRunning:
		if len(p.mu.mainTenant) == 0 {
			return from, from, fmt.Errorf("main tenant of pod(%v) should not be empty for PodStatusRunning", p.name)
		}
		if len(p.mu.guestTenant) != 0 {
			return from, from, fmt.Errorf("guest tenant of pod(%v) should be empty for PodStatusRunning", p.name)
		}
		if p.mu.mainTenant == tenantName {
			return from, from, fmt.Errorf("should not share same tenant(%v,%v)", p.name, tenantName)
		}
		if _, ok := p.mu.guestTenant[tenantName]; ok {
			return from, from, fmt.Errorf("should not share same tenant(%v,%v)", p.name, tenantName)
		}
		to := PodStatusSharing
		p.mu.guestTenant[tenantName] = struct{}{}
		p.mu.status = to
		return from, to, nil
	case PodStatusSharing:
		if len(p.mu.mainTenant) == 0 {
			return from, from, fmt.Errorf("main tenant of pod(%v) should not be empty for PodStatusSharing", p.name)
		}
		if len(p.mu.guestTenant) == 0 {
			return from, from, errors.New("guest tenant should not be empty for PodStatusSharing")
		}
		if p.mu.mainTenant == tenantName {
			return from, from, fmt.Errorf("should not share same tenant(%v,%v)", p.name, tenantName)
		}
		to := PodStatusSharing
		p.mu.guestTenant[tenantName] = struct{}{}
		return from, to, nil
	case PodStatusLogicalError, PodStatusDeleted, PodStatusPending, PodStatusNeedDelete:
		return from, from, fmt.Errorf("should not ref pod(%v) of %v", p.name, from.ToString())
	default:
		return from, from, fmt.Errorf("unexpected PodStatus(%v)", from)
	}
}

// Deref change pod status.
func (p *PodDesc) Deref(tenantName string, reason string) (PodStatus, PodStatus, error) {
	logger.RootLogger.Debug("pod Deref", zap.String("tenant", tenantName),
		zap.String("pod", p.name), zap.String("reason", reason), zap.String("status", p.GetStatusStr()))
	p.mu.Lock()
	defer p.mu.Unlock()
	from := p.mu.status
	switch p.mu.status {
	case PodStatusReadyUnassign, PodStatusUnassigning, PodStatusUnassigned:
		return from, from, fmt.Errorf("should not deref pod(%v) of %v", p.name, from.ToString())
	case PodStatusRunning:
		if len(p.mu.mainTenant) == 0 {
			return from, from, fmt.Errorf("main tenant of pod(%v) should not be empty for PodStatusRunning", p.name)
		}
		if len(p.mu.guestTenant) != 0 {
			return from, from, fmt.Errorf("guest tenant of pod(%v) should be empty for PodStatusRunning", p.name)
		}
		if p.mu.mainTenant != tenantName {
			return from, from, fmt.Errorf("cannot deref pod(%s) that tenant(%v) doesn't own", p.name, tenantName)
		}
		p.mu.status = PodStatusReadyUnassign
		p.mu.mainTenant = ""
		to := p.mu.status
		return from, to, nil
	case PodStatusSharing:
		if len(p.mu.mainTenant) == 0 {
			return from, from, fmt.Errorf("main tenant of pod(%v) should not be empty for PodStatusSharing", p.name)
		}
		if len(p.mu.guestTenant) == 0 {
			return from, from, fmt.Errorf("guest tenant of pod(%v) should not be empty for PodStatusSharing", p.name)
		}
		if p.mu.mainTenant == tenantName {
			for name, _ := range p.mu.guestTenant {
				p.mu.mainTenant = name
				delete(p.mu.guestTenant, name)
				break
			}
		} else {
			for name, _ := range p.mu.guestTenant {
				delete(p.mu.guestTenant, name)
				break
			}
		}
		if len(p.mu.mainTenant) == 0 {
			// todo remove panic
			panic("should not be here")
		}
		if len(p.mu.guestTenant) == 0 {
			p.mu.status = PodStatusRunning
		}
		to := p.mu.status
		return from, to, nil
	case PodStatusLogicalError, PodStatusDeleted, PodStatusPending, PodStatusNeedDelete:
		return from, from, fmt.Errorf("should not deref pod(%v) of %v", p.name, from.ToString())
	default:
		return from, from, fmt.Errorf("unexpected PodStatus(%v)", from)
	}
}

// IsUsable return true when pod is running or sharing.
func (p *PodDesc) IsUsable() bool {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.mu.status == PodStatusRunning || p.mu.status == PodStatusSharing
}

// HasTenant return true when pods is serving this tenant.
func (p *PodDesc) HasTenant(name string) bool {
	p.mu.RLock()
	defer p.mu.RUnlock()
	if p.mu.mainTenant == name {
		return true
	}
	_, ok := p.mu.guestTenant[name]
	return ok
}

// 1. send gprc and wait result
// called by PodMgr.SyncTakePods()
func (p *PodDesc) SendAssignRPC(pdAddr string, tiflashVer string, logger *zap.Logger) (*supervisor.Result, error) {
	// TODO tenantName shared/real tenant name
	return rpc.AssignTenant(p.GetIP(), "", pdAddr, tiflashVer, logger)
}

// 1. change status to unassigning
// 2. send grpc and wait result
// called by StoppingPool.triggerUnassign()
func (p *PodDesc) SendUnassignRPC(logger *zap.Logger) (*supervisor.Result, error) {
	return rpc.UnassignTenant(p.GetIP(), "", false, logger)
}

// OkToMerge return true when pod is ok to be a merge candidate.
// merge conditions:
// 1. merge cooldown has finished. and
// 2. avgCPU is less equal to mergeLowCPULimit.
func (p *PodDesc) OkToMerge(mergeLowCPULimit float64, mergeCoolDownSec uint,
	metricWindowSec uint, metricIntervalSec uint, coreNumPerPod uint) (msg string, avgCPU float64, ok bool) {
	ts := p.GetLastMergeTs()
	msg = fmt.Sprintf("pod(%s) not ok to be a merge candidate, because ", p.name)
	okTs := ts.Add(time.Duration(mergeCoolDownSec) * time.Second)
	if time.Now().Before(okTs) {
		msg += fmt.Sprintf("merge cooldown not finish(lastMerge: %v, okToMerge: %v)", ts, okTs)
		return msg, avgCPU, false
	}

	window := p.GetCPUMetricWindow()
	err := metric.IsWindowValid(metricWindowSec, metricIntervalSec, len(window))
	if err != nil {
		msg += err.Error()
		return msg, avgCPU, false
	}

	avgCPU = window.GetAvgCPU()
	lowCPULimit := mergeLowCPULimit * float64(coreNumPerPod)
	if avgCPU > lowCPULimit {
		msg += fmt.Sprintf("avgCPU(%v) is greater than mergeLowCPULimit(%v)", avgCPU, lowCPULimit)
		return msg, avgCPU, false
	}
	return "", avgCPU, true
}

// okToShareWithoutLock return true when pod is ok to share.
// share condition:
// 1. share cooldown has finished. and
// 2. avgCPU is less equal to shareLowCPULimit.
// called by ShareingRef(called by podMgr.SyncTakePods(called by SharePool.tryTake))
func (p *PodDesc) okToShareWithoutLock(shareLowCPULimit float64, shareCoolDownSec uint,
	metricWindowSec uint, metricIntervalSec uint, coreNumPerPod uint,
	stillShareWhenMetricNotEnough bool) (msg string, ok bool) {
	ts := p.GetLastShareTsWithoutLock()
	msg = fmt.Sprintf("pod(%s) not ok to be share, because ", p.name)

	if shareCoolDownSec > 0 {
		okTs := ts.Add(time.Duration(shareCoolDownSec) * time.Second)
		if time.Now().Before(okTs) {
			msg += fmt.Sprintf("share cooldown not finish(lastShare: %v, okToShare: %v)", ts, okTs)
			return msg, false
		}
	}

	window := p.GetCPUMetricWindowWithoutLock()
	err := metric.IsWindowValid(metricWindowSec, metricIntervalSec, len(window))
	if err != nil {
		msg += err.Error()
		if stillShareWhenMetricNotEnough {
			msg += ", but still share"
			return msg, true
		}

		return msg, false
	}

	avgCPU := window.GetAvgCPU()
	if avgCPU > shareLowCPULimit*float64(coreNumPerPod) {
		msg += fmt.Sprintf("avgCPU(%v) is greater than shareLowCPULimit(%v)", avgCPU, shareLowCPULimit)
		return msg, false
	}
	return "", true
}

// GetCPUMetricWindow return cpu metrics
func (p *PodDesc) GetCPUMetricWindow() metric.CPUMetricWindow {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.GetCPUMetricWindowWithoutLock()
}

// GetCPUMetricWindowWithoutLock return cpu metrics
func (p *PodDesc) GetCPUMetricWindowWithoutLock() metric.CPUMetricWindow {
	return p.mu.cpuWindow
}

// GetTaskCntMetricWindow return TaskCnt metrics
func (p *PodDesc) GetTaskCntMetricWindow() metric.TaskCntMetricWindow {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.mu.taskCntWindow
}

// GetOOMMetricWindow return OOM metrics
func (p *PodDesc) GetOOMMetricWindow() metric.OOMMetricWindow {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.mu.oomWindow
}

// GetRUMetricWindow return RU metrics
func (p *PodDesc) GetRUMetricWindow() metric.RUMetricWindow {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.mu.ruWindow
}

// UpdateMetrics update all metrics.
// called by PodMgr.UpdateMetricsByCopy(called by MetricMgr.collectMetricLoop)
func (p *PodDesc) UpdateMetrics(
	cpu metric.CPUMetricWindow,
	taskCnt metric.TaskCntMetricWindow,
	oom metric.OOMMetricWindow,
	ru metric.RUMetricWindow) {

	p.mu.Lock()
	defer p.mu.Unlock()

	p.mu.cpuWindow = cpu
	p.mu.taskCntWindow = taskCnt
	p.mu.oomWindow = oom
	p.mu.ruWindow = ru
}

// UpdateLastShareTs update share ts.
// called by SharingRef(PodMgr.SyncTakePods()(called by SharePool.tryTake()))
func (p *PodDesc) updateLastShareTsWithoutLock(now time.Time) {
	p.mu.lastShareTs = now
}

// UpdateLastMergeTs update Merge ts.
// called by TenantDesc.ReplaceMergePods(called by TenantMgr.doAnalyzeMergePods)
func (p *PodDesc) UpdateLastMergeTs(now time.Time) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.mu.lastMergeTs = now
}

func (p *PodDesc) GetFQDN() string {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return fmt.Sprintf("%s.%s.pod.cluster.local",
		strings.Replace(p.mu.ip, ".", "-", -1), p.namespace)
}

func (p *PodDesc) GetIP() string {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.mu.ip
}

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

func (p *PodDesc) GetLastShareTsWithoutLock() time.Time {
	return p.mu.lastShareTs
}

func (p *PodDesc) IsSharing() bool {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.mu.status == PodStatusSharing
}

func (p *PodDesc) IsRunning() bool {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.mu.status == PodStatusRunning
}

func (p *PodDesc) GetTenantCnt() uint {
	p.mu.RLock()
	defer p.mu.RUnlock()
	if len(p.mu.mainTenant) == 0 {
		return 0
	}
	return uint(len(p.mu.guestTenant) + 1)
}

func (p *PodDesc) CheckAndToUnassigning() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.mu.status == PodStatusReadyUnassign || p.mu.status == PodStatusLogicalError {
		p.mu.status = PodStatusUnassigning
		return true
	}
	return false
}

func (p *PodDesc) IsDeleted() bool {
	return p.GetStatus() == PodStatusDeleted
}

func (p *PodDesc) IsUnassigned() bool {
	return p.GetStatus() == PodStatusUnassigned
}

func (p *PodDesc) IsPending() bool {
	return p.GetStatus() == PodStatusPending
}

func (p *PodDesc) ChangeIPAndFromPendingToUnassigned(podIP string) bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.mu.ip = podIP
	// todo need check or not?
	if p.mu.status == PodStatusPending {
		p.mu.status = PodStatusUnassigned
		p.mu.lastPendingTs = time.Time{}
		return true
	}
	return false
}

func (p *PodDesc) ToString() string {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.ToStringWithoutLock()
}

func (p *PodDesc) ToStringWithoutLock() string {
	var tenantMsg string
	if len(p.mu.mainTenant) != 0 {
		tenantMsg = fmt.Sprintf("%v;", p.mu.mainTenant)
	}
	for t, _ := range p.mu.guestTenant {
		tenantMsg += fmt.Sprintf("%v;", t)
	}
	return fmt.Sprintf("{name: %v, ip: %v, status: %v, tenants: %v}",
		p.name, p.mu.ip, p.mu.status.ToString(), tenantMsg)
}