package globalapi

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

	"github.com/patrickmn/go-cache"
	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/pkg/common/errors"
	globalstructs "github.com/tidbcloud/observability/pkg/globalservice/structs"
	tidbmgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	// ServerlessParentApplicationID is the fixed applicationID for all serverless
	// parent (infrastructure) clusters. When a request carries this value it is
	// already at parent-cluster granularity and does not need child→parent resolution.
	ServerlessParentApplicationID = "devtier-infra"

	// serverlessBizTypeSuffix is the bizType suffix that identifies serverless traffic.
	serverlessBizTypeSuffix = "-devtier"

	// serverlessResolverCacheTTL is the TTL for cached child→parent mappings.
	serverlessResolverCacheTTL = 30 * time.Minute
)

// ErrMgmtCallFailed is a sentinel error indicating the mgmt ListClusters RPC
// itself failed (network, timeout, service unavailable). Callers should
// propagate this error to allow external retry, rather than falling back.
var ErrMgmtCallFailed = fmt.Errorf("mgmt ListClusters call failed")

// serverlessResolver resolves serverless child-cluster routing dimensions to
// parent-cluster dimensions by querying the management service.
//
// WebConsole/Clinic send child-cluster identifiers (projectID/applicationID
// from the URL path), but the o11y 4-segment route table matches on
// parent-cluster granularity. This resolver bridges that gap transparently.
type serverlessResolver struct {
	cache *cache.Cache
}

var (
	resolverOnce sync.Once
	resolverInst *serverlessResolver
)

func getServerlessResolver() *serverlessResolver {
	resolverOnce.Do(func() {
		resolverInst = &serverlessResolver{
			cache: cache.New(serverlessResolverCacheTTL, serverlessResolverCacheTTL),
		}
	})
	return resolverInst
}

// buildServerlessCacheKey builds a string cache key from the child-cluster routing dimensions.
func buildServerlessCacheKey(bizType, tenantID, applicationID string, showDeleted bool) string {
	return fmt.Sprintf("%s:%s:%s:%t", bizType, tenantID, applicationID, showDeleted)
}

// mgmtListClustersFunc lists a single cluster from the management service.
// It is a package-level variable so tests can override it.
var mgmtListClustersFunc = defaultMgmtListClusters

func defaultMgmtListClusters(ctx context.Context, env string, clusterID uint64, showDeleted bool) (*tidbmgmt.ListClustersResponse, error) {
	cli, err := getMgmtSvcClient(env)
	if err != nil {
		return nil, err
	}
	deleted := showDeleted
	return cli.ListClusters(ctx, &tidbmgmt.ListClustersRequest{
		Filter: &tidbmgmt.ClusterFilter{
			ClusterIds: []uint64{clusterID},
			Deleted:    &deleted,
		},
		PageSize: 1,
	})
}

// shouldResolveServerlessChild reports whether the request targets a serverless
// child cluster that needs parent-cluster resolution.
//
// Trigger conditions (per migration plan §3.7):
//   - bizType is a devtier variant (ends with "-devtier")
//   - applicationID is present and is NOT the fixed parent-cluster applicationID
//
// All other requests (Dedicated, NextGen, BYOC, or already-parent serverless)
// incur zero cost — they short-circuit here.
func shouldResolveServerlessChild(routingCtx globalstructs.RoutingContext) bool {
	if !strings.HasSuffix(routingCtx.BizType, serverlessBizTypeSuffix) {
		return false
	}
	return routingCtx.ApplicationID != "" &&
		routingCtx.ApplicationID != ServerlessParentApplicationID
}

// resolveServerlessChild resolves a child cluster's routing context to its
// parent-cluster dimensions by querying mgmt for the k8s name.
//
// On success the returned RoutingContext has:
//   - ProjectID set to "{region}-{f0x}" (e.g. "us-east-1-f02")
//   - ApplicationID set to ServerlessParentApplicationID ("devtier-infra")
//
// On failure an error is returned — the caller must NOT fall back to physical
// routing, otherwise child-cluster requests would leak into the Dedicated pool.
func resolveServerlessChild(ctx context.Context, routingCtx globalstructs.RoutingContext) (globalstructs.RoutingContext, error) {
	cacheKey := buildServerlessCacheKey(routingCtx.BizType, routingCtx.TenantID, routingCtx.ApplicationID, routingCtx.ShowDeleted)

	if cached, ok := getServerlessResolver().cache.Get(cacheKey); ok {
		projectID := cached.(string)
		log.Debug("serverless child→parent cache hit",
			zap.String("applicationID", routingCtx.ApplicationID),
			zap.String("projectID", projectID))
		return applyParentDims(routingCtx, projectID), nil
	}

	// Determine the k8s name: use X-Kubernetes-Id header if provided,
	// otherwise fall back to mgmt lookup.
	k8sName := routingCtx.KubernetesID
	if k8sName == "" {
		// Derive env from bizType (e.g. "dev-devtier" → "dev", "prod-devtier" → "prod").
		env := strings.TrimSuffix(routingCtx.BizType, serverlessBizTypeSuffix)

		// The child cluster's applicationID is its numeric cluster ID.
		clusterID, err := strconv.ParseUint(routingCtx.ApplicationID, 10, 64)
		if err != nil {
			return routingCtx, errors.NewErrorf(errors.ParameterInvalid,
				"invalid serverless child cluster applicationID %q: %s", routingCtx.ApplicationID, err.Error())
		}

		resp, err := mgmtListClustersFunc(ctx, env, clusterID, routingCtx.ShowDeleted)
		if err != nil {
			return routingCtx, fmt.Errorf("%w: cluster %s: %s", ErrMgmtCallFailed, routingCtx.ApplicationID, err.Error())
		}

		clusters := resp.GetClusters()
		if len(clusters) == 0 {
			return routingCtx, errors.NewErrorf(errors.ClusterNotFound,
				"serverless child cluster %s not found in mgmt", routingCtx.ApplicationID)
		}

		k8sName = clusters[0].GetServerlessCluster().GetCluster().GetKubernetes()
		if k8sName == "" {
			return routingCtx, errors.NewErrorf(errors.UnrecognizedError,
				"empty kubernetes name for serverless cluster %s", routingCtx.ApplicationID)
		}
	}

	projectID, err := extractProjectIDFromK8sName(k8sName, routingCtx.Region)
	if err != nil {
		return routingCtx, errors.NewErrorf(errors.UnrecognizedError,
			"failed to extract parent projectID from k8s name %q: %s", k8sName, err.Error())
	}

	log.Info("resolved serverless child→parent",
		zap.String("childApplicationID", routingCtx.ApplicationID),
		zap.String("k8sName", k8sName),
		zap.String("projectID", projectID),
		zap.String("KubernetesID", routingCtx.KubernetesID),
		zap.Bool("fromHeader", routingCtx.KubernetesID != ""))

	getServerlessResolver().cache.SetDefault(cacheKey, projectID)
	return applyParentDims(routingCtx, projectID), nil
}

// extractProjectIDFromK8sName derives the o11y projectID from the mgmt
// kubernetes name.
//
// The k8s name follows the pattern "{env}-{region}-{suffix}" where suffix is
// the parent-cluster identifier (e.g. "f01", "f02", or "a01" for Alibaba).
// The o11y projectID is "{region}-{suffix}".
//
// Examples:
//
//	staging-us-east-1-f01, region us-east-1 → us-east-1-f01
//	prod-ap-southeast-1-a02, region ap-southeast-1 → ap-southeast-1-a02
func extractProjectIDFromK8sName(k8sName, region string) (string, error) {
	parts := strings.Split(k8sName, "-")
	if len(parts) < 2 {
		return "", fmt.Errorf("k8s name %q has too few segments", k8sName)
	}
	suffix := parts[len(parts)-1]
	if suffix == "" {
		return "", fmt.Errorf("empty parent cluster suffix in k8s name %q", k8sName)
	}
	return fmt.Sprintf("%s-%s", region, suffix), nil
}

// applyParentDims returns a copy of routingCtx with parent-cluster dimensions applied.
func applyParentDims(routingCtx globalstructs.RoutingContext, projectID string) globalstructs.RoutingContext {
	routingCtx.ProjectID = projectID
	routingCtx.ApplicationID = ServerlessParentApplicationID
	return routingCtx
}