package cron

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

	"github.com/google/uuid"
	meteringsdkcommon "github.com/pingcap/metering_sdk/common"
	"github.com/tidbcloud/cluster-service-ng/domain/app/workernodeusage"
	"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"
)

type WorkerNodeMeteringJob struct {
	checkpointRepo     repo.WorkerNodeMeteringCheckpointRepo
	dataplaneManager   dataplane.Manager
	sharedPoolRepo     repo.SharedPoolRepo
	lockRepo           repo.LockRepo
	meteringWriter     meteringWriter
	region             string
	cloudProvider      string
	compensationWindow time.Duration
	expression         string
}

func NewWorkerNodeMeteringJob(
	checkpointRepo repo.WorkerNodeMeteringCheckpointRepo,
	dataplaneManager dataplane.Manager,
	sharedPoolRepo repo.SharedPoolRepo,
	lockRepo repo.LockRepo,
	meteringWriter meteringWriter,
	region string,
	cloudProvider string,
	compensationWindow time.Duration,
	expression string,
) *WorkerNodeMeteringJob {
	return &WorkerNodeMeteringJob{
		checkpointRepo:     checkpointRepo,
		dataplaneManager:   dataplaneManager,
		sharedPoolRepo:     sharedPoolRepo,
		lockRepo:           lockRepo,
		meteringWriter:     meteringWriter,
		region:             region,
		cloudProvider:      cloudProvider,
		compensationWindow: compensationWindow,
		expression:         expression,
	}
}

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

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

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

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

	// Only settle fully completed minutes. The job observes a point-in-time K8s snapshot,
	// so checkpointing the in-progress minute would permanently miss workloads that start later.
	currentMinute := time.Now().Unix()/60*60 - 60
	checkpointSaved := false
	var errs []error
	for _, dataplaneName := range dataplaneNamesFromPools(pools) {
		lastSuccessfulMinute, err := j.checkpointRepo.LoadLastSuccessfulMinute(ctx, j.region, j.cloudProvider, dataplaneName)
		if err != nil {
			workerNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			tilog.Error(ctx, "failed to load checkpoint",
				zap.String("dataplane", dataplaneName),
				zap.Error(err))
			errs = append(errs, fmt.Errorf("failed to load checkpoint for dataplane %s: %w", dataplaneName, err))
			continue
		}

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

		k8sClient, err := j.dataplaneManager.GetClient(dataplaneName)
		if err != nil {
			workerNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			workerNodeMeteringDataplaneAccessTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			tilog.Error(ctx, "failed to get k8s client for dataplane",
				zap.String("dataplane", dataplaneName),
				zap.Error(err))
			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 {
			workerNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			workerNodeMeteringDataplaneAccessTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			tilog.Error(ctx, "failed to fetch k8s resources",
				zap.String("dataplane", dataplaneName),
				zap.Error(err))
			errs = append(errs, fmt.Errorf("failed to fetch k8s resources for dataplane %s: %w", dataplaneName, err))
			continue
		}

		workerNodeMeteringDataplaneAccessTotal.WithLabelValues(workerNodeMeteringResultOK).Inc()

		lastProcessedMinute, err := j.processDataplaneMinutes(ctx, minutes, nodes, pods)
		if err != nil {
			workerNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			tilog.Error(ctx, "failed to process metering minutes",
				zap.String("dataplane", dataplaneName),
				zap.Error(err))
			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 {
			workerNodeMeteringRunTotal.WithLabelValues(workerNodeMeteringResultError).Inc()
			tilog.Error(ctx, "failed to save checkpoint",
				zap.String("dataplane", dataplaneName),
				zap.Error(err))
			errs = append(errs, fmt.Errorf("failed to save checkpoint for dataplane %s: %w", dataplaneName, err))
			continue
		}
		checkpointSaved = true
	}

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

func dataplaneNamesFromPools(pools []*entity.SharedPool) []string {
	seen := make(map[string]struct{}, len(pools))
	dataplanes := make([]string, 0, len(pools))
	for _, pool := range pools {
		dataplaneName := pool.GetDataplaneName()
		if dataplaneName == "" {
			continue
		}
		if _, ok := seen[dataplaneName]; ok {
			continue
		}
		seen[dataplaneName] = struct{}{}
		dataplanes = append(dataplanes, dataplaneName)
	}
	return dataplanes
}

func (j *WorkerNodeMeteringJob) processDataplaneMinutes(ctx context.Context, minutes []int64, nodes map[string]*corev1.Node, pods []*corev1.Pod) (int64, error) {
	var lastProcessedMinute int64
	for _, minuteTs := range minutes {
		workerNodeMeteringCompensationSecondsTotal.Add(60)

		visibleNodes := j.filterVisibleNodes(nodes, minuteTs)
		visiblePods := j.filterVisiblePods(pods, minuteTs)
		snapshot := workernodeusage.BuildSnapshot(ctx, visibleNodes, visiblePods)
		for _, node := range snapshot.Nodes {
			if node.HasCPUMismatch {
				workerNodeMeteringCPUMismatchTotal.Inc()
			}
			if node.DropReason != "" {
				workerNodeMeteringNodeDropTotal.WithLabelValues(node.DropReason).Inc()
			}
		}

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

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

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

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

		lastProcessedMinute = minuteTs
	}

	return lastProcessedMinute, nil
}

func (j *WorkerNodeMeteringJob) lock(ctx context.Context) (func() error, error) {
	key := "worker_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 *WorkerNodeMeteringJob) 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 workernodeusage.IsWorkerNode(node) {
			workerNodes[node.Name] = node
		}
	}

	visiblePods := make([]*corev1.Pod, 0, len(podList.Items))
	for i := range podList.Items {
		pod := &podList.Items[i]
		if workernodeusage.IsNonDaemonSetPod(pod) {
			visiblePods = append(visiblePods, pod)
			tenantLabel := pod.Labels[workernodeusage.TenantLabelKey]
			poolLabel := workernodeusage.PoolLabelValue(pod)
			if tenantLabel == "" || poolLabel == "" {
				tilog.Warn(ctx, "non-DaemonSet pod missing tenant or resource-pool identity",
					zap.String("pod", pod.Name),
					zap.String("namespace", pod.Namespace),
					zap.String("node", pod.Spec.NodeName))
			}
		}
	}

	return workerNodes, visiblePods, nil
}

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

	startMinute := lastSuccessfulMinute + 60
	compensationWindowSeconds := int64(compensationWindow.Seconds())
	earliestAllowed := currentMinute - compensationWindowSeconds + 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 *WorkerNodeMeteringJob) 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 *WorkerNodeMeteringJob) 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
		}
		// We only have a point-in-time snapshot, so once a pod is terminal or already deleting
		// for the sampled minute, stop attributing node ownership to avoid stale over-billing.
		if pod.DeletionTimestamp != nil && pod.DeletionTimestamp.Unix() <= minuteTs+59 {
			continue
		}
		if workernodeusage.IsInactivePod(pod) {
			continue
		}
		visible = append(visible, pod)
	}
	return visible
}