package ops_svc

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/go-resty/resty/v2"
	"github.com/pingcap/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/utils"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
)

const (
	requestTimeout = 5 * time.Second
	retryCount     = 3
	retryWaitTime  = 200 * time.Millisecond

	opListGatewayHPAInfos            = "ListGatewayHPAInfos"
	opUpdateGatewayHPAMinReplicas    = "UpdateGatewayHPAMinReplicas"
	opListTiDBAutoscalingInfos       = "ListTiDBAutoscalingInfos"
	opUpdateTiDBAutoscalingThreshold = "UpdateTiDBAutoscalingThreshold"
	opExecuteCommand                 = "ExecuteCommand"
	opRestartTiDBPodsByNamespace     = "RestartTiDBPodsByNamespace"
	opRestartTiDBPodsByCluster       = "RestartTiDBPodsByCluster"

	updateK8SMessageHeader = "Provider: %s\nRegion: %s\nK8S Cluster: %s\nNamespace: %s\n"

	allowedGatewayHPANamespace = "tidb-admin"
)

var operationPaths = map[string]string{
	opListGatewayHPAInfos:            "/ops/api/v1/gateway_hpa_min_replicas",
	opUpdateGatewayHPAMinReplicas:    "/ops/api/v1/update_gateway_hpa_min_replicas",
	opListTiDBAutoscalingInfos:       "/ops/api/v1/tidb_autoscaling_threshold",
	opUpdateTiDBAutoscalingThreshold: "/ops/api/v1/update_tidb_autoscaling_threshold",
	opExecuteCommand:                 "/ops/api/v1/execute_command",
	opRestartTiDBPodsByNamespace:     "/ops/api/v1/restart_tidb_pods_by_namespace",
	opRestartTiDBPodsByCluster:       "/ops/api/v1/restart_tidb_pods_by_cluster",
}

// clusterInfo represents a Kubernetes cluster with its URL
type clusterInfo struct {
	url string
}

// regionInfo represents a region with its clusters
type regionInfo struct {
	clusters map[string]*clusterInfo // map[clusterName]*clusterInfo
}

// providerInfo represents a provider with its regions
type providerInfo struct {
	regions map[string]*regionInfo // map[regionName]*regionInfo
}

// regionalClusterURLs represents a hierarchical structure of service URLs
// organized by provider, region, and k8s cluster
type regionalClusterURLs struct {
	providers map[string]*providerInfo // map[providerName]*providerInfo
}

type k8sRpcService struct {
	client         *resty.Client
	k8sIngressUrls regionalClusterURLs
	rpc.UnimplementedRegionalK8SOperationServiceServer
}

func NewK8sRpcService() rpc.RegionalK8SOperationServiceServer {
	client := resty.New().
		SetTimeout(requestTimeout).
		SetRetryCount(retryCount).
		SetRetryWaitTime(retryWaitTime)

	// Initialize the structured hierarchy
	urls := regionalClusterURLs{
		providers: make(map[string]*providerInfo),
	}

	for _, regionCfg := range config.Global().WorkerConfig.RegionalConfig {
		provider := regionCfg.Provider
		region := regionCfg.Region

		// Initialize provider if not exists
		if _, ok := urls.providers[provider]; !ok {
			urls.providers[provider] = &providerInfo{
				regions: make(map[string]*regionInfo),
			}
		}

		// Initialize region if not exists
		if _, ok := urls.providers[provider].regions[region]; !ok {
			urls.providers[provider].regions[region] = &regionInfo{
				clusters: make(map[string]*clusterInfo),
			}
		}

		for _, k8sCfg := range regionCfg.K8sConfig {
			cluster := k8sCfg.Kubernetes
			ingressUrl := k8sCfg.K8sIngressUrl

			if !strings.HasPrefix(ingressUrl, "http://") &&
				!strings.HasPrefix(ingressUrl, "https://") {
				ingressUrl = "https://" + ingressUrl
			}

			// Store URL in the structured hierarchy
			urls.providers[provider].regions[region].clusters[cluster] = &clusterInfo{
				url: ingressUrl,
			}
		}
	}

	return &k8sRpcService{
		client:         client,
		k8sIngressUrls: urls,
	}
}

type k8sRequest interface {
	GetProvider() string
	GetRegion() string
	GetK8SCluster() string
	ValidateAll() error
}

func (svc *k8sRpcService) validateRequestHeader(request k8sRequest) (string, error) {
	if request == nil {
		return "", errors.New("request is nil")
	}
	if err := request.ValidateAll(); err != nil {
		return "", err
	}

	provider := request.GetProvider()
	region := request.GetRegion()
	k8sCluster := request.GetK8SCluster()

	// Look up provider in the structured hierarchy
	providerInfo, ok := svc.k8sIngressUrls.providers[provider]
	if !ok {
		return "", fmt.Errorf("provider %s not found", provider)
	}

	// Look up region
	regionInfo, ok := providerInfo.regions[region]
	if !ok {
		return "", fmt.Errorf("region %s not found for provider %s", region, provider)
	}

	// Look up cluster
	clusterInfo, ok := regionInfo.clusters[k8sCluster]
	if !ok {
		return "", fmt.Errorf("k8s cluster %s not found for provider %s, region %s",
			k8sCluster, provider, region)
	}

	return clusterInfo.url, nil
}

func (svc *k8sRpcService) UpdateGatewayHPAMinReplicas(
	ctx context.Context,
	request *rpc.UpdateGatewayHPAMinReplicasRequest,
) (*emptypb.Empty, error) {
	serviceUrl, err := svc.validateRequestHeader(request)
	if err != nil {
		return nil, fmt.Errorf("failed to validate request header, err: %w", err)
	}
	if request.GetNamespace() != allowedGatewayHPANamespace {
		return nil, fmt.Errorf("unsupported namespace to update: %s, only %s is currently supported",
			request.GetNamespace(), allowedGatewayHPANamespace)
	}

	minReplicas := request.GetMinReplicas()
	apiPath := operationPaths[opUpdateGatewayHPAMinReplicas]
	err = svc.forwardToRegion(ctx, serviceUrl, http.MethodPost, apiPath,
		func(req *resty.Request) *resty.Request {
			return req.SetHeader("Content-Type", "application/json").
				SetBody(UpdateMinReplicasRequest{
					MinReplicas: minReplicas,
				})
		})
	if err != nil {
		return nil, err
	}

	// Send notification message.
	title := fmt.Sprintf("[%s] Update Gateway HPA Min Replicas", strings.ToUpper(config.Global().ServerConfig.Stage))
	message := fmt.Sprintf(
		updateK8SMessageHeader,
		request.GetProvider(),
		request.GetRegion(),
		request.GetK8SCluster(),
		request.GetNamespace(),
	)
	message += fmt.Sprintf("Min Replicas: %d", minReplicas)
	go func() {
		utils.SendLarkMessage(title, message, utils.LarkMsgLevelInfo)
	}()
	return &emptypb.Empty{}, nil
}

func (svc *k8sRpcService) UpdateTiDBAutoscalingThreshold(
	ctx context.Context,
	request *rpc.UpdateTiDBAutoscalingThresholdRequest,
) (*emptypb.Empty, error) {
	serviceUrl, err := svc.validateRequestHeader(request)
	if err != nil {
		return nil, fmt.Errorf("failed to validate request header, err: %w", err)
	}

	apiPath := operationPaths[opUpdateTiDBAutoscalingThreshold]
	err = svc.forwardToRegion(ctx, serviceUrl, http.MethodPost, apiPath,
		func(req *resty.Request) *resty.Request {
			return req.SetHeader("Content-Type", "application/json").
				SetBody(UpdateTiDBAutoscalingThresholdRequest{
					Namespace: request.GetNamespace(),
					Metric:    request.GetMetric(),
					AvgValue:  request.GetAvgValue(),
				})
		})
	if err != nil {
		return nil, err
	}

	// Send notification message.
	title := fmt.Sprintf("[%s] Update TiDB Autoscaling Threshold", strings.ToUpper(config.Global().ServerConfig.Stage))
	message := fmt.Sprintf(
		updateK8SMessageHeader,
		request.GetProvider(),
		request.GetRegion(),
		request.GetK8SCluster(),
		request.GetNamespace(),
	)
	message += fmt.Sprintf("Metric: %s\nAverage Value: %f\n",
		request.GetMetric(),
		request.GetAvgValue(),
	)
	go func() {
		utils.SendLarkMessage(title, message, utils.LarkMsgLevelInfo)
	}()

	return &emptypb.Empty{}, nil
}

func (svc *k8sRpcService) ListGatewayHPAInfos(
	ctx context.Context,
	_ *emptypb.Empty,
) (*rpc.ListGatewayHPAInfosResponse, error) {
	regionalInfos := make([]*rpc.RegionalGatewayHPAInfos, 0)

	// Iterate through the structured hierarchy
	for providerName, providerInfo := range svc.k8sIngressUrls.providers {
		for regionName, regionInfo := range providerInfo.regions {
			regionalInfo := &rpc.RegionalGatewayHPAInfos{
				Provider:                  providerName,
				Region:                    regionName,
				K8SClusterGatewayHpaInfos: make([]*rpc.RegionalGatewayHPAInfos_K8SClusterGatewayHPAInfos, 0),
			}
			for clusterName, clusterInfo := range regionInfo.clusters {
				var hpaInfos []GatewayHPAInfo
				apiPath := operationPaths[opListGatewayHPAInfos]
				err := svc.forwardToRegion(ctx, clusterInfo.url, http.MethodGet, apiPath,
					func(req *resty.Request) *resty.Request {
						return req.SetHeader("Accept", "application/json").
							SetResult(&hpaInfos)
					})
				if err != nil {
					// Log error but continue with other clusters
					log.Error("Failed to get Gateway HPA info",
						zap.String("provider", providerName),
						zap.String("region", regionName),
						zap.String("cluster", clusterName),
						zap.Error(err))
					continue
				}
				clusterProtoInfo := &rpc.RegionalGatewayHPAInfos_K8SClusterGatewayHPAInfos{
					K8SCluster:      clusterName,
					GatewayHpaInfos: make([]*rpc.RegionalGatewayHPAInfos_K8SClusterGatewayHPAInfos_GatewayHPAInfo, 0),
				}
				for _, info := range hpaInfos {
					if info.Namespace != allowedGatewayHPANamespace {
						continue // Skip unsupported namespaces
					}
					clusterProtoInfo.GatewayHpaInfos = append(clusterProtoInfo.GatewayHpaInfos,
						&rpc.RegionalGatewayHPAInfos_K8SClusterGatewayHPAInfos_GatewayHPAInfo{
							Namespace:       info.Namespace,
							Name:            info.Name,
							MinReplicas:     info.MinReplicas,
							MaxReplicas:     info.MaxReplicas,
							DesiredReplicas: info.DesiredReplicas,
						})
				}
				regionalInfo.K8SClusterGatewayHpaInfos = append(
					regionalInfo.K8SClusterGatewayHpaInfos,
					clusterProtoInfo,
				)
			}
			regionalInfos = append(regionalInfos, regionalInfo)
		}
	}

	return &rpc.ListGatewayHPAInfosResponse{
		Infos: regionalInfos,
	}, nil
}

func (svc *k8sRpcService) ListTiDBAutoscalingInfos(
	ctx context.Context,
	_ *emptypb.Empty,
) (*rpc.ListTiDBAutoscalingInfosResponse, error) {
	regionalInfos := make([]*rpc.RegionalTiDBAutoscalingInfos, 0)

	// Iterate through the structured hierarchy
	for providerName, providerInfo := range svc.k8sIngressUrls.providers {
		for regionName, regionInfo := range providerInfo.regions {
			regionalInfo := &rpc.RegionalTiDBAutoscalingInfos{
				Provider: providerName,
				Region:   regionName,
				K8SClusterTidbAutoscalingInfos: make(
					[]*rpc.RegionalTiDBAutoscalingInfos_K8SClusterTiDBAutoscalingInfos,
					0,
				),
			}
			for clusterName, clusterInfo := range regionInfo.clusters {
				var tidbInfos []TiDBAutoscalingThresholdInfo
				apiPath := operationPaths[opListTiDBAutoscalingInfos]
				err := svc.forwardToRegion(ctx, clusterInfo.url, http.MethodGet, apiPath,
					func(req *resty.Request) *resty.Request {
						return req.SetHeader("Accept", "application/json").
							SetResult(&tidbInfos)
					})
				if err != nil {
					// Log error but continue with other clusters
					log.Error("Failed to get TiDB autoscaling info",
						zap.String("provider", providerName),
						zap.String("region", regionName),
						zap.String("cluster", clusterName),
						zap.Error(err))
					continue
				}

				// Convert to proto format
				clusterProtoInfo := &rpc.RegionalTiDBAutoscalingInfos_K8SClusterTiDBAutoscalingInfos{
					K8SCluster: clusterName,
					TidbAutoscalingInfos: make(
						[]*rpc.RegionalTiDBAutoscalingInfos_K8SClusterTiDBAutoscalingInfos_TiDBAutoscalingInfo,
						0,
					),
				}

				for _, info := range tidbInfos {
					tidbInfo := &rpc.RegionalTiDBAutoscalingInfos_K8SClusterTiDBAutoscalingInfos_TiDBAutoscalingInfo{
						Namespace: info.Namespace,
						Name:      info.Name,
						Metrics: make(
							[]*rpc.RegionalTiDBAutoscalingInfos_K8SClusterTiDBAutoscalingInfos_TiDBAutoscalingInfo_Metrics,
							0,
						),
					}

					for _, metric := range info.Metrics {
						tidbInfo.Metrics = append(
							tidbInfo.Metrics,
							&rpc.RegionalTiDBAutoscalingInfos_K8SClusterTiDBAutoscalingInfos_TiDBAutoscalingInfo_Metrics{
								Name:          metric.Name,
								PodLabel:      metric.PodLabel,
								PromSql:       metric.PromSql,
								ScaleFlapping: metric.ScaleFlapping,
								AvgValue:      metric.AvgValue,
							},
						)
					}

					clusterProtoInfo.TidbAutoscalingInfos = append(clusterProtoInfo.TidbAutoscalingInfos, tidbInfo)
				}

				regionalInfo.K8SClusterTidbAutoscalingInfos = append(
					regionalInfo.K8SClusterTidbAutoscalingInfos,
					clusterProtoInfo,
				)
			}

			regionalInfos = append(regionalInfos, regionalInfo)
		}
	}

	return &rpc.ListTiDBAutoscalingInfosResponse{
		Infos: regionalInfos,
	}, nil
}

// ExecuteCommand forwards a command (e.g. pd-ctl) to the ops-service on the
// specified k8s cluster. Command validation (allowlist, shell metachar blocking)
// is enforced by ops-service itself; this layer is a pure forwarder.
func (svc *k8sRpcService) ExecuteCommand(
	ctx context.Context,
	request *rpc.OpsExecuteCommandRequest,
) (*rpc.OpsExecuteCommandResponse, error) {
	serviceUrl, err := svc.validateRequestHeader(request)
	if err != nil {
		return nil, fmt.Errorf("failed to validate request header, err: %w", err)
	}

	var opsResp ExecuteCommandResponse
	err = svc.forwardToRegion(ctx, serviceUrl, http.MethodPost,
		operationPaths[opExecuteCommand],
		func(req *resty.Request) *resty.Request {
			return req.SetHeader("Content-Type", "application/json").
				SetBody(ExecuteCommandRequest{Command: request.GetCommand()}).
				SetResult(&opsResp)
		})
	if err != nil {
		return nil, err
	}

	return &rpc.OpsExecuteCommandResponse{
		Stdout:   opsResp.Stdout,
		Stderr:   opsResp.Stderr,
		ExitCode: opsResp.ExitCode,
	}, nil
}

func (svc *k8sRpcService) RestartTiDBPodsByNamespace(
	ctx context.Context,
	request *rpc.OpsRestartByNamespaceRequest,
) (*rpc.OpsRestartByNamespaceResponse, error) {
	serviceUrl, err := svc.validateRequestHeader(request)
	if err != nil {
		return nil, fmt.Errorf("failed to validate request header, err: %w", err)
	}

	err = svc.forwardToRegion(ctx, serviceUrl, http.MethodPost,
		operationPaths[opRestartTiDBPodsByNamespace],
		func(req *resty.Request) *resty.Request {
			return req.SetHeader("Content-Type", "application/json").
				SetBody(RestartByNamespaceRequest{
					Namespace: request.GetNamespace(),
					Force:     request.GetForce(),
				})
		})
	if err != nil {
		return nil, err
	}

	return &rpc.OpsRestartByNamespaceResponse{Message: "ok"}, nil
}

func (svc *k8sRpcService) RestartTiDBPodsByCluster(
	ctx context.Context,
	request *rpc.OpsRestartByClusterRequest,
) (*rpc.OpsRestartByClusterResponse, error) {
	serviceUrl, err := svc.validateRequestHeader(request)
	if err != nil {
		return nil, fmt.Errorf("failed to validate request header, err: %w", err)
	}

	err = svc.forwardToRegion(ctx, serviceUrl, http.MethodPost,
		operationPaths[opRestartTiDBPodsByCluster],
		func(req *resty.Request) *resty.Request {
			return req.SetHeader("Content-Type", "application/json").
				SetBody(RestartByClusterRequest{
					KeyspaceToken: request.GetKeyspaceToken(),
					Force:         request.GetForce(),
				})
		})
	if err != nil {
		return nil, err
	}

	return &rpc.OpsRestartByClusterResponse{Message: "ok"}, nil
}

func (svc *k8sRpcService) forwardToRegion(
	ctx context.Context,
	address, method, apiPath string,
	configureRequest func(req *resty.Request) *resty.Request,
) error {
	request := svc.client.R().SetContext(ctx)

	// Apply request configuration function
	request = configureRequest(request)

	resp, err := request.Execute(method, address+apiPath)
	if err != nil {
		return fmt.Errorf("failed to forward request to region, err: %w", err)
	}
	if resp.StatusCode() != http.StatusOK {
		return fmt.Errorf(
			"failed to forward request to region, status code: %d, status: %s, body: %s",
			resp.StatusCode(), resp.Status(), resp.String(),
		)
	}
	return nil
}