package cron

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"
	meteringsdkcommon "github.com/pingcap/metering_sdk/common"
	apiv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	reservedNodeMeteringTenantLabelKey       = "tags.tidbcloud.com/tenant"
	reservedNodeMeteringSharedPoolLabelKey   = "tags.tidbcloud.com/shared-pool"
	reservedNodeMeteringResourcePoolLabelKey = "tags.tidbcloud.com/resource-pool"
	reservedNodeMeteringClusterLabelKey      = "tags.tidbcloud.com/cluster"
	reservedNodeMeteringComponentLabelKey    = "tags.tidbcloud.com/component"
	reservedNodeMeteringNodeSpecLabelKey     = "node-spec"
	reservedNodeMeteringDedicatedTaintKey    = "dedicated"
	reservedNodeMeteringWorkerTaintValue     = "worker"

	reservedNodeMeteringCategory = "reserved-node"
	reservedNodeMeteringMeter    = "reserved_vcpu_usage"
)

var reservedNodeMeteringAllowedComponents = map[string]struct{}{
	"worker-tidb":     {},
	"pd":              {},
	"tikv-worker":     {},
	"tiproxy":         {},
	"tidb":            {},
	"tikv":            {},
	"write-tiflash":   {},
	"compute-tiflash": {},
}

type ReservedNodeMeteringJob struct {
	checkpointRepo     repo.ReservedNodeMeteringCheckpointRepo
	dataplaneManager   dataplane.Manager
	sharedPoolRepo     repo.SharedPoolRepo
	tidbClusterRepo    repo.TiDBClusterRepo
	lockRepo           repo.LockRepo
	meteringWriter     meteringWriter
	region             string
	cloudProvider      string
	compensationWindow time.Duration
	expression         string
}

func NewReservedNodeMeteringJob(
	checkpointRepo repo.ReservedNodeMeteringCheckpointRepo,
	dataplaneManager dataplane.Manager,
	sharedPoolRepo repo.SharedPoolRepo,
	tidbClusterRepo repo.TiDBClusterRepo,
	lockRepo repo.LockRepo,
	meteringWriter meteringWriter,
	region string,
	cloudProvider string,
	compensationWindow time.Duration,
	expression string,
) *ReservedNodeMeteringJob {
	return &ReservedNodeMeteringJob{
		checkpointRepo:     checkpointRepo,
		dataplaneManager:   dataplaneManager,
		sharedPoolRepo:     sharedPoolRepo,
		tidbClusterRepo:    tidbClusterRepo,
		lockRepo:           lockRepo,
		meteringWriter:     meteringWriter,
		region:             region,
		cloudProvider:      cloudProvider,
		compensationWindow: compensationWindow,
		expression:         expression,
	}
}

func (j *ReservedNodeMeteringJob) Name() string {
	return "ReservedNodeMeteringJob"
}

func (j *ReservedNodeMeteringJob) CronExpression() string {
	return j.expression
}

func (j *ReservedNodeMeteringJob) Execute(ctx context.Context) error {
	unlockFunc, err := j.lock(ctx)
	if err != nil {
		tilog.Warn(ctx, "failed to acquire lock for reserved-node metering, skip this round", zap.Error(err))
		return nil
	}
	defer func() { _ = unlockFunc() }()

	sharedPools, err := j.sharedPoolRepo.List(ctx, nil)
	if err != nil {
		reservedNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
		return fmt.Errorf("failed to list shared pools: %w", err)
	}

	sharedPoolIDs := make([]string, 0, len(sharedPools))
	for _, sp := range sharedPools {
		if sp.GetCapacityMode() != apiv1beta.CapacityMode_TOPOLOGY_DRIVEN {
			continue
		}
		if sp.GetId() != "" {
			sharedPoolIDs = append(sharedPoolIDs, sp.GetId())
		}
	}

	if len(sharedPoolIDs) == 0 {
		return nil
	}

	clusters, err := j.tidbClusterRepo.List(ctx, &repo.TiDBClusterListFilter{SharedPoolIds: sharedPoolIDs, Deleted: false})
	if err != nil {
		reservedNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
		return fmt.Errorf("failed to list tidb clusters: %w", err)
	}

	poolByID := make(map[string]*entity.SharedPool, len(sharedPools))
	for _, sp := range sharedPools {
		poolByID[sp.GetId()] = sp
	}

	dataplaneTargets := make(map[string]map[string]clusterIdentity)
	dataplaneTargetsByPool := make(map[string]map[string][]clusterIdentity)
	for _, cluster := range clusters {
		if cluster.GetCapacityMode() != apiv1beta.CapacityMode_TOPOLOGY_DRIVEN {
			continue
		}
		sp := poolByID[cluster.GetSharedPoolId()]
		if sp == nil {
			continue
		}
		if sp.GetCapacityMode() != apiv1beta.CapacityMode_TOPOLOGY_DRIVEN {
			continue
		}
		dataplaneName := sp.GetDataplaneName()
		if dataplaneName == "" {
			continue
		}
		if dataplaneTargets[dataplaneName] == nil {
			dataplaneTargets[dataplaneName] = map[string]clusterIdentity{}
		}
		if dataplaneTargetsByPool[dataplaneName] == nil {
			dataplaneTargetsByPool[dataplaneName] = map[string][]clusterIdentity{}
		}
		identity := clusterIdentity{
			tenantID:     cluster.GetTenantId(),
			sharedPoolID: cluster.GetSharedPoolId(),
			clusterID:    cluster.GetId(),
		}
		dataplaneTargets[dataplaneName][cluster.GetId()] = clusterIdentity{
			tenantID:     identity.tenantID,
			sharedPoolID: identity.sharedPoolID,
			clusterID:    identity.clusterID,
		}
		dataplaneTargetsByPool[dataplaneName][cluster.GetSharedPoolId()] = append(dataplaneTargetsByPool[dataplaneName][cluster.GetSharedPoolId()], identity)
	}

	if len(dataplaneTargets) == 0 {
		return nil
	}

	currentMinute := time.Now().Unix()/60*60 - 60
	checkpointSaved := false
	var errs []error

	for dataplaneName, topologyClusters := range dataplaneTargets {
		if len(topologyClusters) == 0 {
			continue
		}

		lastSuccessfulMinute, err := j.checkpointRepo.LoadLastSuccessfulMinute(ctx, j.region, j.cloudProvider, dataplaneName)
		if err != nil {
			reservedNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			errs = append(errs, fmt.Errorf("failed to load checkpoint for dataplane %s: %w", dataplaneName, err))
			continue
		}

		minutes := reservedMinutesToProcess(lastSuccessfulMinute, currentMinute, j.compensationWindow)
		if len(minutes) == 0 {
			continue
		}

		k8sClient, err := j.dataplaneManager.GetClient(dataplaneName)
		if err != nil {
			reservedNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			reservedNodeMeteringDataplaneAccessTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			errs = append(errs, fmt.Errorf("failed to get k8s client for dataplane %s: %w", dataplaneName, err))
			continue
		}

		nodes, pods, err := j.fetchK8sResources(ctx, k8sClient)
		if err != nil {
			reservedNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			reservedNodeMeteringDataplaneAccessTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			errs = append(errs, fmt.Errorf("failed to fetch k8s resources for dataplane %s: %w", dataplaneName, err))
			continue
		}
		reservedNodeMeteringDataplaneAccessTotal.WithLabelValues(workerNodeMeteringResultOK).Inc()

		lastProcessedMinute, err := j.processDataplaneMinutes(ctx, minutes, nodes, pods, topologyClusters, dataplaneTargetsByPool[dataplaneName])
		if err != nil {
			reservedNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			errs = append(errs, fmt.Errorf("failed to process metering for dataplane %s: %w", dataplaneName, err))
			continue
		}
		if lastProcessedMinute == 0 {
			continue
		}

		if err := j.checkpointRepo.SaveLastSuccessfulMinute(ctx, j.region, j.cloudProvider, dataplaneName, lastProcessedMinute); err != nil {
			reservedNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			errs = append(errs, fmt.Errorf("failed to save checkpoint for dataplane %s: %w", dataplaneName, err))
			continue
		}
		checkpointSaved = true
	}

	if checkpointSaved {
		reservedNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultOK).Inc()
	}
	if len(errs) > 0 {
		return errors.Join(errs...)
	}
	return nil
}

type clusterIdentity struct {
	tenantID     string
	sharedPoolID string
	clusterID    string
}

type reservedNodeOwnership struct {
	tenantID     string
	sharedPoolID string
	clusterID    string
	nodeType     string
	conflicts    bool
}

func (j *ReservedNodeMeteringJob) processDataplaneMinutes(
	ctx context.Context,
	minutes []int64,
	nodes map[string]*corev1.Node,
	pods []*corev1.Pod,
	topologyClusters map[string]clusterIdentity,
	topologyClustersByPool map[string][]clusterIdentity,
) (int64, error) {
	var lastProcessedMinute int64

	for _, minuteTs := range minutes {
		reservedNodeMeteringCompensationSecondsTotal.Add(60)
		aggregated := make(map[string]map[string]float64)

		visibleNodes := j.filterVisibleNodes(nodes, minuteTs)
		visiblePods := j.filterVisiblePods(pods, minuteTs)
		nodeOwnership := j.determineNodeOwnership(ctx, visibleNodes, visiblePods, topologyClusters, topologyClustersByPool)

		for nodeName, ownership := range nodeOwnership {
			if ownership.conflicts {
				reservedNodeMeteringNodeDropTotal.WithLabelValues("conflicting_ownership").Inc()
				continue
			}
			if ownership.tenantID == "" || ownership.sharedPoolID == "" || ownership.clusterID == "" {
				reservedNodeMeteringNodeDropTotal.WithLabelValues("missing_identity").Inc()
				continue
			}
			if ownership.nodeType == "" {
				reservedNodeMeteringNodeDropTotal.WithLabelValues("missing_node_type").Inc()
				continue
			}

			vcpu := j.parseNodeSpecCPU(ctx, visibleNodes[nodeName])
			if vcpu == 0 {
				reservedNodeMeteringNodeDropTotal.WithLabelValues("invalid_node_spec").Inc()
				continue
			}

			key := fmt.Sprintf("%s:%s:%s:%s", ownership.tenantID, ownership.sharedPoolID, ownership.clusterID, ownership.nodeType)
			if aggregated[key] == nil {
				aggregated[key] = make(map[string]float64)
			}
			aggregated[key]["vcpu"] += float64(vcpu)
		}

		for key, metrics := range aggregated {
			if _, ok := metrics["vcpu"]; !ok {
				continue
			}

			parts := strings.Split(key, ":")
			if len(parts) != 4 {
				continue
			}
			tenantID := parts[0]
			sharedPoolID := parts[1]
			clusterID := parts[2]
			nodeType := parts[3]

			meteringData := &meteringsdkcommon.MeteringData{
				Timestamp:    minuteTs,
				Category:     reservedNodeMeteringCategory,
				SelfID:       strings.ReplaceAll(uuid.NewString(), "-", ""),
				SharedPoolID: sharedPoolID,
				Data: []map[string]any{{
					"tenant_id":       tenantID,
					"resourcepool_id": sharedPoolID,
					"cluster_id":      clusterID,
					"node_type":       nodeType,
					"region":          j.region,
					"meter_name":      reservedNodeMeteringMeter,
					"json_version":    "1.0.0",
					"usage": map[string]any{
						"value": metrics["vcpu"],
						"unit":  "vCPU",
					},
				}},
			}

			if err := j.meteringWriter.Write(ctx, meteringData); err != nil {
				reservedNodeMeteringWriteTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
				return 0, fmt.Errorf("failed to write metering data: %w", err)
			}
			reservedNodeMeteringWriteTotal.WithLabelValues(workerNodeMeteringResultOK).Inc()
		}

		lastProcessedMinute = minuteTs
	}

	return lastProcessedMinute, nil
}

func (j *ReservedNodeMeteringJob) lock(ctx context.Context) (func() error, error) {
	key := "reserved_node_metering"
	lock := &entity.Lock{Key: key, TimeoutSecs: 10}
	unlockFunc, err := j.lockRepo.Lock(ctx, lock)
	if err != nil {
		return nil, err
	}
	return unlockFunc, nil
}

func (j *ReservedNodeMeteringJob) fetchK8sResources(ctx context.Context, k8sClient client.Client) (map[string]*corev1.Node, []*corev1.Pod, error) {
	var nodeList corev1.NodeList
	if err := k8sClient.List(ctx, &nodeList); err != nil {
		return nil, nil, fmt.Errorf("failed to list nodes: %w", err)
	}

	var podList corev1.PodList
	if err := k8sClient.List(ctx, &podList); err != nil {
		return nil, nil, fmt.Errorf("failed to list pods: %w", err)
	}

	workerNodes := make(map[string]*corev1.Node)
	for i := range nodeList.Items {
		node := &nodeList.Items[i]
		if j.isWorkerNode(node) {
			workerNodes[node.Name] = node
		}
	}

	billablePods := make([]*corev1.Pod, 0)
	for i := range podList.Items {
		pod := &podList.Items[i]
		if j.isBillablePod(pod) {
			billablePods = append(billablePods, pod)
		}
	}

	return workerNodes, billablePods, nil
}

func (j *ReservedNodeMeteringJob) isWorkerNode(node *corev1.Node) bool {
	for _, taint := range node.Spec.Taints {
		if taint.Key == reservedNodeMeteringDedicatedTaintKey && taint.Value == reservedNodeMeteringWorkerTaintValue {
			return true
		}
	}
	return false
}

func (j *ReservedNodeMeteringJob) isBillablePod(pod *corev1.Pod) bool {
	tenantLabel := pod.Labels[reservedNodeMeteringTenantLabelKey]
	poolLabel := j.poolLabelValue(pod)
	componentLabel := pod.Labels[reservedNodeMeteringComponentLabelKey]
	return tenantLabel != "" && poolLabel != "" && componentLabel != "" && j.isNonDaemonSetPod(pod) && !j.isInactivePod(pod)
}

func (j *ReservedNodeMeteringJob) isNonDaemonSetPod(pod *corev1.Pod) bool {
	for _, owner := range pod.OwnerReferences {
		if owner.Kind == "DaemonSet" {
			return false
		}
	}
	return true
}

func (j *ReservedNodeMeteringJob) isInactivePod(pod *corev1.Pod) bool {
	return pod.Status.Phase == corev1.PodSucceeded || pod.Status.Phase == corev1.PodFailed
}

func (j *ReservedNodeMeteringJob) poolLabelValue(pod *corev1.Pod) string {
	if poolID := pod.Labels[reservedNodeMeteringResourcePoolLabelKey]; poolID != "" {
		return poolID
	}
	return pod.Labels[reservedNodeMeteringSharedPoolLabelKey]
}

func (j *ReservedNodeMeteringJob) determineNodeOwnership(
	ctx context.Context,
	nodes map[string]*corev1.Node,
	pods []*corev1.Pod,
	topologyClusters map[string]clusterIdentity,
	topologyClustersByPool map[string][]clusterIdentity,
) map[string]reservedNodeOwnership {
	ownership := make(map[string]reservedNodeOwnership)
	for nodeName := range nodes {
		ownership[nodeName] = reservedNodeOwnership{}
	}

	for i := range pods {
		pod := pods[i]
		nodeName := pod.Spec.NodeName
		if nodeName == "" {
			continue
		}
		current, exists := ownership[nodeName]
		if !exists {
			continue
		}

		tenantID := pod.Labels[reservedNodeMeteringTenantLabelKey]
		poolID := j.poolLabelValue(pod)
		nodeType := normalizeReservedNodeType(pod.Labels[reservedNodeMeteringComponentLabelKey])
		if tenantID == "" || poolID == "" || nodeType == "" {
			continue
		}

		clusterID := pod.Labels[reservedNodeMeteringClusterLabelKey]
		target, ok := topologyClusters[clusterID]
		if !ok {
			poolTargets := topologyClustersByPool[poolID]
			if len(poolTargets) == 1 {
				target = poolTargets[0]
				clusterID = target.clusterID
				ok = true
			}
		}
		if !ok {
			continue
		}
		if target.tenantID != "" && target.tenantID != tenantID {
			continue
		}
		if target.sharedPoolID != "" && target.sharedPoolID != poolID {
			continue
		}

		if current.tenantID == "" {
			current.tenantID = tenantID
			current.sharedPoolID = poolID
			current.clusterID = clusterID
			current.nodeType = nodeType
		} else if current.tenantID != tenantID || current.sharedPoolID != poolID || current.clusterID != clusterID || current.nodeType != nodeType {
			tilog.Warn(ctx, "conflicting ownership detected on reserved node",
				zap.String("node", nodeName),
				zap.String("existing_tenant", current.tenantID),
				zap.String("existing_pool", current.sharedPoolID),
				zap.String("existing_cluster", current.clusterID),
				zap.String("existing_node_type", current.nodeType),
				zap.String("conflicting_tenant", tenantID),
				zap.String("conflicting_pool", poolID),
				zap.String("conflicting_cluster", clusterID),
				zap.String("conflicting_node_type", nodeType),
			)
			current.conflicts = true
		}

		ownership[nodeName] = current
	}

	return ownership
}

func normalizeReservedNodeType(raw string) string {
	v := strings.ToLower(strings.TrimSpace(raw))
	if v == "" {
		return ""
	}
	v = strings.ReplaceAll(v, "_", "-")
	v = strings.ReplaceAll(v, " ", "-")
	if _, ok := reservedNodeMeteringAllowedComponents[v]; !ok {
		return ""
	}
	return v
}

func (j *ReservedNodeMeteringJob) parseNodeSpecCPU(ctx context.Context, node *corev1.Node) int {
	nodeSpec := node.Labels[reservedNodeMeteringNodeSpecLabelKey]
	if nodeSpec == "" {
		tilog.Warn(ctx, "reserved node missing node-spec label", zap.String("node", node.Name))
		return 0
	}

	idx := strings.Index(nodeSpec, "C")
	if idx <= 0 {
		tilog.Warn(ctx, "invalid reserved node-spec format", zap.String("node", node.Name), zap.String("node-spec", nodeSpec))
		return 0
	}

	vcpu, err := strconv.Atoi(nodeSpec[:idx])
	if err != nil {
		tilog.Warn(ctx, "failed to parse vCPU from reserved node-spec", zap.String("node", node.Name), zap.String("node-spec", nodeSpec), zap.Error(err))
		return 0
	}

	capacityCPU := node.Status.Capacity.Cpu()
	if capacityCPU != nil {
		capacityCPUInt := int(capacityCPU.Value())
		if capacityCPUInt != vcpu {
			reservedNodeMeteringCPUMismatchTotal.Inc()
		}
	}

	return vcpu
}

func reservedMinutesToProcess(lastSuccessfulMinute, currentMinute int64, compensationWindow time.Duration) []int64 {
	if currentMinute <= 0 {
		return []int64{}
	}
	if lastSuccessfulMinute == 0 {
		return []int64{currentMinute}
	}

	startMinute := lastSuccessfulMinute + 60
	earliestAllowed := currentMinute - int64(compensationWindow.Seconds()) + 60
	if startMinute < earliestAllowed {
		startMinute = earliestAllowed
	}
	if startMinute > currentMinute {
		return []int64{}
	}

	minutes := make([]int64, 0)
	for m := startMinute; m <= currentMinute; m += 60 {
		minutes = append(minutes, m)
	}
	return minutes
}

func (j *ReservedNodeMeteringJob) filterVisibleNodes(nodes map[string]*corev1.Node, minuteTs int64) map[string]*corev1.Node {
	visible := make(map[string]*corev1.Node)
	for name, node := range nodes {
		if node.GetCreationTimestamp().Unix() <= minuteTs+59 {
			visible[name] = node
		}
	}
	return visible
}

func (j *ReservedNodeMeteringJob) filterVisiblePods(pods []*corev1.Pod, minuteTs int64) []*corev1.Pod {
	visible := make([]*corev1.Pod, 0, len(pods))
	for i := range pods {
		pod := pods[i]
		if pod.GetCreationTimestamp().Unix() > minuteTs+59 {
			continue
		}
		if pod.DeletionTimestamp != nil && pod.DeletionTimestamp.Unix() <= minuteTs+59 {
			continue
		}
		if j.isInactivePod(pod) {
			continue
		}
		visible = append(visible, pod)
	}
	return visible
}

func ShouldEnableReservedNodeMetering(cfg *config.MeteringConfig) bool {
	return cfg != nil && cfg.ReservedNodeEnabled()
}