package ngoperation

import (
	"context"
	"fmt"
	"math"
	"strconv"
	"time"

	"github.com/tidbcloud/devops/pkg/infra/clients/cliniccli"
	"github.com/tidbcloud/devops/pkg/service/operation/actions"
	nextgenv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
)

const (
	resourcePoolPauseQPSThreshold = 10.0
	resourcePoolPauseQPSPromQL    = `sum(rate(tidb_executor_statement_total[5m]))`
	resourcePoolClusterPageSize   = uint32(100)
)

type resourcePoolQPSChecker func(ctx context.Context, resourcePoolID uint64) (float64, error)

type listNextGenClusters func(context.Context, *nextgenv1.ListClustersRequest) (*nextgenv1.ListClustersResponse, error)

type resourcePoolQPS struct {
	listClusters listNextGenClusters
	clinic       cliniccli.Client
	now          func() time.Time
}

func newResourcePoolQPSChecker(comps *actions.Components) resourcePoolQPSChecker {
	var listClusters listNextGenClusters
	var clinic cliniccli.Client
	if comps != nil {
		clinic = comps.ClinicCli
		if comps.CoreSvc != nil && comps.CoreSvc.Mgmt != nil && comps.CoreSvc.Mgmt.NextGenCluster != nil {
			client := comps.CoreSvc.Mgmt.NextGenCluster
			listClusters = func(ctx context.Context, req *nextgenv1.ListClustersRequest) (*nextgenv1.ListClustersResponse, error) {
				return client.ListClusters(ctx, req)
			}
		}
	}
	checker := &resourcePoolQPS{
		listClusters: listClusters,
		clinic:       clinic,
		now:          time.Now,
	}
	return checker.AverageAggregateQPS
}

func (q *resourcePoolQPS) AverageAggregateQPS(ctx context.Context, resourcePoolID uint64) (float64, error) {
	if q.listClusters == nil {
		return 0, fmt.Errorf("NextGen cluster client is not configured")
	}
	if q.clinic == nil {
		return 0, fmt.Errorf("Clinic client is not configured")
	}

	clusters, err := q.listResourcePoolClusters(ctx, resourcePoolID)
	if err != nil {
		return 0, err
	}

	observationTime := q.now()
	var aggregateQPS float64
	for _, cluster := range clusters {
		qps, err := q.clusterQPS(ctx, cluster.GetClusterId(), observationTime)
		if err != nil {
			return 0, err
		}
		aggregateQPS += qps
	}
	return aggregateQPS, nil
}

func (q *resourcePoolQPS) listResourcePoolClusters(
	ctx context.Context,
	resourcePoolID uint64,
) ([]*nextgenv1.Cluster, error) {
	clusters := make([]*nextgenv1.Cluster, 0)
	for page := uint32(1); ; page++ {
		resp, err := q.listClusters(ctx, &nextgenv1.ListClustersRequest{
			ResourcePoolIds: []uint64{resourcePoolID},
			PageSize:        resourcePoolClusterPageSize,
			Page:            page,
		})
		if err != nil {
			return nil, fmt.Errorf("list clusters in resource pool %d: %w", resourcePoolID, err)
		}
		if resp == nil {
			return nil, fmt.Errorf("list clusters in resource pool %d: empty response", resourcePoolID)
		}

		clusters = append(clusters, resp.GetClusters()...)
		reachedTotal := resp.GetTotal() > 0 && len(clusters) >= int(resp.GetTotal())
		if len(resp.GetClusters()) < int(resourcePoolClusterPageSize) || reachedTotal {
			return clusters, nil
		}
	}
}

func (q *resourcePoolQPS) clusterQPS(
	ctx context.Context,
	clusterID uint64,
	observationTime time.Time,
) (float64, error) {
	clusterIDRaw := strconv.FormatUint(clusterID, 10)
	clinicClusters, err := q.clinic.ListClusters(ctx, &cliniccli.ClusterFilter{
		ClusterIds: []string{clusterIDRaw},
	})
	if err != nil {
		return 0, fmt.Errorf("resolve cluster %d in Clinic: %w", clusterID, err)
	}

	var clinicCluster *cliniccli.Cluster
	for i := range clinicClusters {
		if clinicClusters[i].ClusterID == clusterIDRaw {
			clinicCluster = &clinicClusters[i]
			break
		}
	}
	if clinicCluster == nil {
		return 0, fmt.Errorf("cluster %d is not found in Clinic", clusterID)
	}

	resp, err := q.clinic.QueryClusterMetrics(ctx, cliniccli.QueryClusterMetricsRequest{
		OrgID:     clinicCluster.OrgID,
		ClusterID: clinicCluster.ClusterID,
		PromQL:    resourcePoolPauseQPSPromQL,
		Start:     observationTime.Add(-5 * time.Minute),
		End:       observationTime,
		Step:      "5m",
	})
	if err != nil {
		return 0, fmt.Errorf("query QPS for cluster %d: %w", clusterID, err)
	}
	if resp == nil || len(resp.Data.Result) == 0 {
		return 0, fmt.Errorf("query QPS for cluster %d: metric is unavailable", clusterID)
	}
	if resp.IsPartial {
		return 0, fmt.Errorf("query QPS for cluster %d: metric response is partial", clusterID)
	}

	var qps float64
	for _, stream := range resp.Data.Result {
		if stream == nil || len(stream.Values) == 0 {
			return 0, fmt.Errorf("query QPS for cluster %d: metric has no samples", clusterID)
		}
		value := float64(stream.Values[len(stream.Values)-1].Value)
		if math.IsNaN(value) || math.IsInf(value, 0) || value < 0 {
			return 0, fmt.Errorf("query QPS for cluster %d: invalid metric value %v", clusterID, value)
		}
		qps += value
	}
	return qps, nil
}