package urls

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"slices"
	"strconv"
	"strings"

	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/auth/session"
	"github.com/pingcap-inc/clinic/pkg/datasource"
	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/o11y"
	"github.com/pingcap-inc/clinic/pkg/types"
)

// ErrInvalidControlPlaneInfo identifies an explicitly malformed logical
// control-plane selector.
var ErrInvalidControlPlaneInfo = errors.New("invalid control_plane_info")

// ClusterURLsResolver is responsible for resolving o11y URLs for clusters
type ClusterURLsResolver struct {
	o11yCli *o11y.Client
}

type MetricsScope string

const (
	MetricsScopeChild  MetricsScope = "child"
	MetricsScopeParent MetricsScope = "parent"
)

type MetricsTarget struct {
	MetricsURL  string
	O11yCluster *o11y.ClusterInfo
}

type InternalReadTarget struct {
	InternalReadURL string
	O11yCluster     *o11y.ClusterInfo
}

// NodeMetricsTarget describes how the node metrics proxy should reach the
// data-plane metrics backend. BYOC targets use InternalReadURL while legacy
// project-scoped targets continue to use the O11Y cluster ID.
type NodeMetricsTarget struct {
	InternalReadURL string
	O11yCluster     *o11y.ClusterInfo
}

type LogsTarget struct {
	LogQueryURL string
	O11yCluster *o11y.ClusterInfo
}

type StorageTargetParam struct {
	Provider     string
	Region       string
	TenantID     string
	ClusterID    string
	DeployType   string
	DeployTypeV2 string
}

type StorageTarget struct {
	BucketName  string
	ObjectPath  string
	O11yCluster *o11y.ClusterInfo
}

// NewClusterURLsResolver creates a new ClusterURLsResolver
func NewClusterURLsResolver(o11yCli *o11y.Client) *ClusterURLsResolver {
	return &ClusterURLsResolver{
		o11yCli: o11yCli,
	}
}

func (r *ClusterURLsResolver) dataServiceMetricsURL(region, provider, metricsPath string) string {
	u := url.URL{Scheme: "https"}
	u.Host = r.o11yCli.GenDataServiceProxyHost(region, provider)
	u.Path = metricsPath
	return u.String()
}

func (r *ClusterURLsResolver) internalReadMetricsURL(cluster *o11y.ClusterInfo) string {
	if cluster == nil {
		return ""
	}
	return cluster.InternalReadUri
}

func (r *ClusterURLsResolver) multitenantQueryMetricsURL(cluster *o11y.ClusterInfo) string {
	if cluster == nil {
		return ""
	}
	return cluster.MultitenantQueryUri
}

func (r *ClusterURLsResolver) historicalMetricsURL(region, provider string, param *o11y.ListClusterParam) string {
	if param == nil {
		return ""
	}
	return r.dataServiceMetricsURL(
		region,
		provider,
		r.o11yCli.GenHistoricalDataProxyPath("", param.BizType, param.TenantID, param.ProjectID, param.ApplicationID),
	)
}

func (r *ClusterURLsResolver) multitenantMetricsURL(region, provider string) string {
	return r.dataServiceMetricsURL(region, provider, "/internal/v2/metrics/multitenant/api/v1/")
}

func nextGenTidbTenantID(cluster *db.Cluster) string {
	if cluster.DeployType == o11y.PremiumType &&
		cluster.Organization.DbaasOrgID == 0 &&
		cluster.TenantID != 0 {
		return strconv.FormatUint(cluster.TenantID, 10)
	}
	return fmt.Sprintf("%d", cluster.Organization.DbaasOrgID)
}

func (r *ClusterURLsResolver) logStatsS3URL(cluster *db.Cluster, objectStorageSchema, bucketSuffix string) string {
	if cluster.Provider != "aws" {
		return ""
	}
	if cluster.DeployType != o11y.DedicatedType &&
		cluster.DeployType != o11y.PremiumType &&
		cluster.DeployType != o11y.NgHostType {
		return ""
	}

	region := cluster.Region
	if cluster.DeployType == o11y.PremiumType || cluster.DeployType == o11y.NgHostType {
		if region == "ap-south-1" {
			// ap-south-1 keeps the raw region without any premium/ng suffix.
		} else if region == "ap-southeast-2" {
			region += "-ng"
		} else {
			region += "-premium"
		}
	}

	return fmt.Sprintf("%s://o11y-prod-shared-%s%s/deltalake/global/log-stats/",
		objectStorageSchema,
		region,
		bucketSuffix)
}

func parseControlPlaneInfo(info string) (projectID, provider, region, applicationID string, err error) {
	parts := strings.Split(info, "/")
	if len(parts) < 3 {
		return "", "", "", "", fmt.Errorf("invalid control_plane_info format")
	}

	middlePart := parts[1]
	firstDashIndex := strings.Index(middlePart, "-")
	if firstDashIndex == -1 {
		return "", "", "", "", fmt.Errorf("invalid control_plane_info format")
	}

	return parts[0], middlePart[:firstDashIndex], middlePart[firstDashIndex+1:], parts[2], nil
}

func normalizeLogicalControlPlaneInfo(info string) (string, error) {
	normalized := strings.TrimSpace(info)
	parts := strings.Split(normalized, "/")
	switch len(parts) {
	case 2:
		if parts[0] != o11y.SeedID || parts[1] == "" {
			return "", fmt.Errorf("%w: unsupported two-part selector", ErrInvalidControlPlaneInfo)
		}
		normalized += "/" + o11y.SeedID
	case 3:
		if parts[0] == "" || parts[1] == "" || parts[2] == "" {
			return "", fmt.Errorf("%w: selector contains an empty segment", ErrInvalidControlPlaneInfo)
		}
	default:
		return "", fmt.Errorf("%w: selector must contain two or three segments", ErrInvalidControlPlaneInfo)
	}

	projectID, provider, region, applicationID, err := parseControlPlaneInfo(normalized)
	if err != nil || projectID != o11y.SeedID || provider == "" || region == "" || applicationID == "" {
		return "", fmt.Errorf("%w: malformed logical seed selector", ErrInvalidControlPlaneInfo)
	}
	return normalized, nil
}

func (r *ClusterURLsResolver) ResolveMetricsTarget(ctx context.Context, cluster *db.Cluster, scope MetricsScope, controlPlaneInfo string) (*MetricsTarget, error) {
	if r.o11yCli == nil {
		return nil, fmt.Errorf("o11y is not enabled")
	}
	switch scope {
	case MetricsScopeChild:
		return r.resolveChildMetricsTarget(ctx, cluster)
	case MetricsScopeParent:
		return r.resolveParentMetricsTarget(ctx, cluster, controlPlaneInfo)
	default:
		return nil, fmt.Errorf("invalid metrics scope: %s", scope)
	}
}

func (r *ClusterURLsResolver) ResolveInternalReadTarget(ctx context.Context, cluster *db.Cluster, scope MetricsScope, controlPlaneInfo string) (*InternalReadTarget, error) {
	if r.o11yCli == nil {
		return nil, fmt.Errorf("o11y is not enabled")
	}
	var (
		o11yCluster *o11y.ClusterInfo
		err         error
	)
	switch scope {
	case MetricsScopeChild:
		o11yCluster, err = r.resolveChildInternalReadCluster(ctx, cluster)
	case MetricsScopeParent:
		var target *MetricsTarget
		target, err = r.resolveParentMetricsTarget(ctx, cluster, controlPlaneInfo)
		if target != nil {
			o11yCluster = target.O11yCluster
		}
	default:
		return nil, fmt.Errorf("invalid metrics scope: %s", scope)
	}
	if err != nil {
		return nil, err
	}
	if o11yCluster == nil {
		return nil, fmt.Errorf("internal read target is empty")
	}
	internalReadURL := r.internalReadMetricsURL(o11yCluster)
	if internalReadURL == "" {
		return nil, fmt.Errorf("internal read URL is empty")
	}
	return &InternalReadTarget{
		InternalReadURL: internalReadURL,
		O11yCluster:     o11yCluster,
	}, nil
}

// ResolveNodeMetricsTarget resolves the data-plane metrics backend used by the
// cloud node metrics proxy. BYOC data planes are identified by DataPlaneName,
// not by the mutable DBaaS project ID.
func (r *ClusterURLsResolver) ResolveNodeMetricsTarget(ctx context.Context, cluster *db.Cluster, region string) (*NodeMetricsTarget, error) {
	if cluster == nil {
		return nil, errors.New("cluster is nil")
	}
	if region == "" {
		region = cluster.Region
	}

	var (
		o11yCluster *o11y.ClusterInfo
		err         error
	)
	if isByocNodeMetricsTarget(cluster) {
		if cluster.DataPlaneName == "" {
			return nil, errors.New("BYOC data plane name is empty")
		}
		o11yCluster, err = r.o11yCli.GetByocK8sCluster(
			ctx,
			cluster.Provider,
			region,
			strconv.FormatUint(cluster.TenantID, 10),
			cluster.DataPlaneName,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to get BYOC node metrics target: %w", err)
		}
		if o11yCluster == nil {
			return nil, errors.New("BYOC node metrics target is nil")
		}
		if o11yCluster.InternalReadUri == "" {
			return nil, errors.New("BYOC node metrics internal read URL is empty")
		}
		internalReadURL, parseErr := url.Parse(o11yCluster.InternalReadUri)
		if parseErr != nil || internalReadURL.Host == "" || (internalReadURL.Scheme != "http" && internalReadURL.Scheme != "https") {
			return nil, errors.New("BYOC node metrics internal read URL is invalid")
		}
		return &NodeMetricsTarget{
			InternalReadURL: o11yCluster.InternalReadUri,
			O11yCluster:     o11yCluster,
		}, nil
	}

	o11yCluster, err = r.o11yCli.GetProjectK8sCluster(
		ctx,
		cluster.Provider,
		region,
		strconv.FormatUint(cluster.TenantID, 10),
		cluster.ProjectID,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to get project node metrics target: %w", err)
	}
	if o11yCluster == nil || o11yCluster.ID == "" {
		return nil, errors.New("project node metrics target is empty")
	}
	return &NodeMetricsTarget{O11yCluster: o11yCluster}, nil
}

func (r *ClusterURLsResolver) ResolveLogicalControlPlaneMetricsTarget(ctx context.Context, controlPlaneInfo string) (*MetricsTarget, error) {
	if r.o11yCli == nil {
		return nil, fmt.Errorf("o11y is not enabled")
	}
	normalizedControlPlaneInfo, err := normalizeLogicalControlPlaneInfo(controlPlaneInfo)
	if err != nil {
		return nil, err
	}
	projectID, vendor, region, applicationID, err := parseControlPlaneInfo(normalizedControlPlaneInfo)
	if err != nil {
		return nil, fmt.Errorf("%w: %v", ErrInvalidControlPlaneInfo, err)
	}
	if projectID != o11y.SeedID {
		return nil, fmt.Errorf("%w: unsupported logical control_plane_info project", ErrInvalidControlPlaneInfo)
	}

	o11yCluster, err := r.o11yCli.GetDedicatedSeedCluster(ctx, vendor, region, applicationID)
	if errors.Is(err, o11y.ErrNotFound) {
		o11yCluster, err = r.o11yCli.GetNextGenSeedCluster(ctx, vendor, region, applicationID)
	}
	if err != nil {
		return nil, fmt.Errorf("failed to get seed o11y cluster: %w", err)
	}
	if o11yCluster.InternalReadUri == "" {
		return nil, fmt.Errorf("seed metrics target internalReadUri is empty")
	}
	return &MetricsTarget{
		MetricsURL:  r.internalReadMetricsURL(o11yCluster),
		O11yCluster: o11yCluster,
	}, nil
}

func (r *ClusterURLsResolver) ResolveServerlessControlPlaneMetricsTarget(ctx context.Context, controlPlaneInfo string) (*MetricsTarget, error) {
	if r.o11yCli == nil {
		return nil, fmt.Errorf("o11y is not enabled")
	}
	projectID, vendor, region, applicationID, err := parseControlPlaneInfo(controlPlaneInfo)
	if err != nil {
		return nil, err
	}
	o11yCluster, err := r.o11yCli.GetServerlessControlPlaneCluster(ctx, vendor, region, projectID, applicationID)
	if err != nil {
		return nil, fmt.Errorf("failed to get serverless control plane o11y cluster: %w", err)
	}
	return &MetricsTarget{
		MetricsURL:  r.internalReadMetricsURL(o11yCluster),
		O11yCluster: o11yCluster,
	}, nil
}

func (r *ClusterURLsResolver) ResolveLogsTarget(ctx context.Context, cluster *db.Cluster) (*LogsTarget, error) {
	if r.o11yCli == nil {
		return nil, fmt.Errorf("o11y is not enabled")
	}
	return r.resolveChildLogsTarget(ctx, cluster)
}

func (r *ClusterURLsResolver) ResolveStorageTarget(ctx context.Context, param StorageTargetParam, category string) (*StorageTarget, error) {
	if r.o11yCli == nil {
		return nil, fmt.Errorf("o11y is not enabled")
	}
	if category == "" {
		return nil, fmt.Errorf("storage category is required")
	}
	var o11yClusters []o11y.ClusterInfo
	switch {
	case isServerlessStorageTarget(param):
		listParam := &o11y.ListClusterParam{
			BizType:  r.o11yCli.DevTierBizTypeWithEnv(),
			Region:   param.Region,
			Vendor:   param.Provider,
			TenantID: param.TenantID,
			// A DBaaS project ID is intentionally not part of this lookup because
			// user clusters can move between DBaaS projects.
			ApplicationID: param.ClusterID,
			ShowDeleted:   true,
		}
		var err error
		o11yClusters, err = r.o11yCli.ListClusterWithCache(ctx, listParam)
		if err != nil {
			return nil, fmt.Errorf("get serverless storage target failed: %w", err)
		}
	case isDedicatedStorageTarget(param):
		o11yCluster, err := r.o11yCli.GetDedicatedTidbCluster(
			ctx,
			param.Provider,
			param.Region,
			param.TenantID,
			param.ClusterID,
		)
		if err != nil {
			return nil, fmt.Errorf("get dedicated storage target failed: %w", err)
		}
		if o11yCluster == nil {
			return nil, fmt.Errorf("%w: dedicated storage target is empty", o11y.ErrNotFound)
		}
		o11yClusters = []o11y.ClusterInfo{*o11yCluster}
	default:
		return nil, fmt.Errorf("storage target is only supported for dedicated/starter/essential clusters")
	}

	var target *StorageTarget
	for i := range o11yClusters {
		location, ok := o11yClusters[i].StorageLocation(category)
		if !ok {
			continue
		}
		if target != nil {
			return nil, fmt.Errorf("more than one storage target found for category: %s", category)
		}
		target = &StorageTarget{
			BucketName:  location.BucketName,
			ObjectPath:  location.ObjectPath,
			O11yCluster: &o11yClusters[i],
		}
	}
	if target == nil {
		return nil, fmt.Errorf("%w: no storage target found for category: %s", o11y.ErrNotFound, category)
	}
	return target, nil
}

func isServerlessStorageTarget(param StorageTargetParam) bool {
	if strings.EqualFold(param.DeployType, o11y.SharedType) || strings.EqualFold(param.DeployType, o11y.ServerlessType) {
		return true
	}
	return slices.Contains(o11y.SharedTypes, param.DeployTypeV2)
}

func isByocNodeMetricsTarget(cluster *db.Cluster) bool {
	if cluster == nil {
		return false
	}
	return cluster.DeployType == o11y.ByocType ||
		cluster.DeployType == o11y.ByocHostType ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2Byoc ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2ByocResourcePool
}

func isDedicatedStorageTarget(param StorageTargetParam) bool {
	return strings.EqualFold(param.DeployType, o11y.DedicatedType) ||
		param.DeployTypeV2 == o11y.DeployTypeV2Dedicated
}

func isMultitenantSharedScopeMetricsTarget(cluster *db.Cluster) bool {
	if cluster == nil {
		return false
	}
	return cluster.DeployTypeV2 == o11y.DeployTypeV2PremiumResourcePool ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2PremiumReservedResourcePool ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2EssentialV2ResourcePool
}

func (r *ClusterURLsResolver) resolveChildMetricsTarget(ctx context.Context, cluster *db.Cluster) (*MetricsTarget, error) {
	o11yCluster, err := r.resolveChildInternalReadCluster(ctx, cluster)
	if err != nil {
		return nil, err
	}
	if isMultitenantSharedScopeMetricsTarget(cluster) {
		metricsURL := r.multitenantQueryMetricsURL(o11yCluster)
		if metricsURL == "" {
			return nil, fmt.Errorf("shared scope multitenant metrics URL is empty")
		}
		return &MetricsTarget{
			MetricsURL:  metricsURL,
			O11yCluster: o11yCluster,
		}, nil
	}
	metricsURL := r.internalReadMetricsURL(o11yCluster)
	if metricsURL == "" {
		return nil, fmt.Errorf("cluster internal read metrics URL is empty")
	}
	return &MetricsTarget{
		MetricsURL:  metricsURL,
		O11yCluster: o11yCluster,
	}, nil
}

func (r *ClusterURLsResolver) resolveChildInternalReadCluster(ctx context.Context, cluster *db.Cluster) (*o11y.ClusterInfo, error) {
	switch {
	case cluster.DeployType == o11y.SharedType ||
		cluster.DeployType == o11y.ServerlessType ||
		slices.Contains(o11y.SharedTypes, cluster.DeployTypeV2):
		o11yClusters, err := r.o11yCli.GetServerlessTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.TenantID),
			cluster.ClusterID,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to get serverless o11y cluster: %w", err)
		}
		selected := selectServerlessO11yCluster(cluster, o11yClusters)
		return &selected, nil
	case cluster.DeployType == o11y.DedicatedType ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2Dedicated:
		o11yCluster, err := r.o11yCli.GetDedicatedTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.Organization.DbaasOrgID),
			cluster.ClusterID,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to get dedicated o11y cluster: %w", err)
		}
		return o11yCluster, nil
	case cluster.DeployType == o11y.ByocHostType:
		o11yCluster, err := r.o11yCli.GetByocHostCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.Organization.DbaasOrgID),
			cluster.ClusterID,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to get byoc host o11y cluster: %w", err)
		}
		return o11yCluster, nil
	case cluster.DeployType == o11y.ByocType ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2Byoc ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2ByocResourcePool:
		o11yCluster, err := r.o11yCli.GetByocTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.Organization.DbaasOrgID),
			cluster.ClusterID,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to get byoc o11y cluster: %w", err)
		}
		return o11yCluster, nil
	case cluster.DeployType == o11y.NgHostType:
		o11yCluster, err := r.o11yCli.GetNextGenHostTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.Organization.DbaasOrgID),
			cluster.ClusterID,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to get nextgen host o11y cluster: %w", err)
		}
		return o11yCluster, nil
	case cluster.DeployType == o11y.PremiumType ||
		slices.Contains(o11y.NextgenResourcePoolTypes, cluster.DeployTypeV2):
		o11yCluster, err := r.o11yCli.GetNextGenTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			nextGenTidbTenantID(cluster),
			cluster.ClusterID,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to get nextgen o11y cluster: %w", err)
		}
		return o11yCluster, nil
	default:
		return nil, fmt.Errorf("unsupported internal read target deploy type: %s, deploy type v2: %s", cluster.DeployType, cluster.DeployTypeV2)
	}
}

func (r *ClusterURLsResolver) resolveChildLogsTarget(ctx context.Context, cluster *db.Cluster) (*LogsTarget, error) {
	var (
		o11yCluster *o11y.ClusterInfo
		err         error
	)
	switch {
	case cluster.DeployType == o11y.SharedType ||
		cluster.DeployType == o11y.ServerlessType ||
		slices.Contains(o11y.SharedTypes, cluster.DeployTypeV2):
		var o11yClusters []o11y.ClusterInfo
		o11yClusters, err = r.o11yCli.GetServerlessTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.TenantID),
			cluster.ClusterID,
		)
		if err == nil {
			selected := selectServerlessO11yCluster(cluster, o11yClusters)
			o11yCluster = &selected
		}
	case cluster.DeployType == o11y.DedicatedType ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2Dedicated:
		o11yCluster, err = r.o11yCli.GetDedicatedTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.Organization.DbaasOrgID),
			cluster.ClusterID,
		)
	case cluster.DeployType == o11y.ByocHostType:
		o11yCluster, err = r.o11yCli.GetByocHostCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.Organization.DbaasOrgID),
			cluster.ClusterID,
		)
	case cluster.DeployType == o11y.ByocType ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2Byoc ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2ByocResourcePool:
		o11yCluster, err = r.o11yCli.GetByocTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.Organization.DbaasOrgID),
			cluster.ClusterID,
		)
	case cluster.DeployType == o11y.PremiumType ||
		cluster.DeployType == o11y.NgHostType ||
		slices.Contains(o11y.NextgenResourcePoolTypes, cluster.DeployTypeV2):
		o11yCluster, err = r.o11yCli.GetNextGenTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			nextGenTidbTenantID(cluster),
			cluster.ClusterID,
		)
	default:
		return nil, fmt.Errorf("unsupported logs target deploy type: %s, deploy type v2: %s", cluster.DeployType, cluster.DeployTypeV2)
	}
	if err != nil {
		return nil, fmt.Errorf("failed to get logs o11y cluster: %w", err)
	}
	if o11yCluster == nil {
		return nil, fmt.Errorf("logs o11y cluster is empty")
	}
	if o11yCluster.LogQueryUri == "" {
		return nil, fmt.Errorf("logs target logQueryUri is empty")
	}
	return &LogsTarget{
		LogQueryURL: o11yCluster.LogQueryUri,
		O11yCluster: o11yCluster,
	}, nil
}

func (r *ClusterURLsResolver) resolveParentMetricsTarget(ctx context.Context, cluster *db.Cluster, controlPlaneInfo string) (*MetricsTarget, error) {
	if !slices.Contains(o11y.SharedTypes, cluster.DeployTypeV2) {
		return nil, fmt.Errorf("parent metrics scope is only supported for starter/essential clusters")
	}
	if controlPlaneInfo == "" {
		return nil, fmt.Errorf("control_plane_info is required for parent metrics scope")
	}

	projectID, vendor, region, applicationID, err := parseControlPlaneInfo(controlPlaneInfo)
	if err != nil {
		return nil, err
	}
	o11yCluster, err := r.o11yCli.GetServerlessControlPlaneCluster(ctx, vendor, region, projectID, applicationID)
	if err != nil {
		return nil, fmt.Errorf("failed to get parent o11y cluster: %w", err)
	}
	return &MetricsTarget{
		MetricsURL:  r.internalReadMetricsURL(o11yCluster),
		O11yCluster: o11yCluster,
	}, nil
}

func (r *ClusterURLsResolver) resolveParentLogsTarget(ctx context.Context, cluster *db.Cluster, controlPlaneInfo string) (*LogsTarget, error) {
	if !slices.Contains(o11y.SharedTypes, cluster.DeployTypeV2) {
		return nil, fmt.Errorf("parent logs scope is only supported for starter/essential clusters")
	}
	if controlPlaneInfo == "" {
		return nil, fmt.Errorf("control_plane_info is required for parent logs scope")
	}

	projectID, vendor, region, applicationID, err := parseControlPlaneInfo(controlPlaneInfo)
	if err != nil {
		return nil, err
	}
	o11yCluster, err := r.o11yCli.GetServerlessControlPlaneCluster(ctx, vendor, region, projectID, applicationID)
	if err != nil {
		return nil, fmt.Errorf("failed to get parent o11y cluster: %w", err)
	}
	if o11yCluster.LogQueryUri == "" {
		return nil, fmt.Errorf("parent logs target logQueryUri is empty")
	}
	return &LogsTarget{
		LogQueryURL: o11yCluster.LogQueryUri,
		O11yCluster: o11yCluster,
	}, nil
}

func selectServerlessO11yCluster(cluster *db.Cluster, o11yClusters []o11y.ClusterInfo) o11y.ClusterInfo {
	o11yCluster := o11yClusters[0]
	if len(o11yClusters) <= 1 {
		return o11yCluster
	}

	clusterID, err := strconv.ParseUint(cluster.ClusterID, 10, 64)
	if err != nil {
		log.Error("parse cluster id failed", zap.Error(err), zap.String("clusterID", cluster.ClusterID))
		return o11yCluster
	}
	mgmtCluster, err := datasource.RealtimeGetMgmtCluster(&session.CtxUser{
		Internal: true,
		Email:    "clinic@pingcap.com",
	}, clusterID)
	if err != nil || mgmtCluster == nil {
		log.Error("realtime get mgmt cluster failed", zap.Error(err), zap.Uint64("clusterID", clusterID))
		return o11yCluster
	}
	serverlessCluster := mgmtCluster.GetServerlessCluster()
	if serverlessCluster == nil || serverlessCluster.Cluster == nil {
		return o11yCluster
	}
	serverlessK8s := serverlessCluster.Cluster.Kubernetes
	if serverlessK8s == "" {
		return o11yCluster
	}
	for n := range o11yClusters {
		if strings.HasSuffix(serverlessK8s, o11yClusters[n].ProjectID) {
			return o11yClusters[n]
		}
	}
	log.Warn("failed to disambiguate serverless o11y cluster",
		zap.String("clusterID", cluster.ClusterID),
		zap.String("serverlessK8s", serverlessK8s),
		zap.Any("o11yClusters", o11yClusters))
	return o11yCluster
}

func legacyLokiURLFromMetricsURL(metricsURL string) (string, error) {
	if metricsURL == "" {
		return "", nil
	}
	parsedURL, err := url.Parse(metricsURL)
	if err != nil {
		return "", fmt.Errorf("failed to parse cluster metrics URL: %w", err)
	}
	return fmt.Sprintf("https://%s/loki/self-monitoring", parsedURL.Host), nil
}

func statementsS3URLFromSlowQueryURL(slowQueryURL string) string {
	const slowLogsSuffix = "/slowlogs"

	trimmedURL := strings.TrimSuffix(slowQueryURL, "/")
	if !strings.HasSuffix(trimmedURL, slowLogsSuffix) {
		return ""
	}

	return strings.TrimSuffix(trimmedURL, slowLogsSuffix) + "/persisted_statements_summary"
}

func isPremiumStatementsCluster(cluster *db.Cluster) bool {
	if cluster == nil {
		return false
	}

	switch cluster.DeployTypeV2 {
	case o11y.DeployTypeV2Premium,
		o11y.DeployTypeV2PremiumReserved,
		o11y.DeployTypeV2PremiumResourcePool,
		o11y.DeployTypeV2PremiumReservedResourcePool:
		return true
	case "":
		// Older records do not have deploy_type_v2. Preserve Premium support for
		// those records without treating newer Essential V2 clusters, whose
		// legacy deploy type can also be "premium", as Premium.
		return cluster.DeployType == o11y.PremiumType
	default:
		return false
	}
}

func useBYOCLogsTarget(cluster *db.Cluster) bool {
	if cluster == nil {
		return false
	}
	return cluster.DeployType == o11y.ByocType ||
		cluster.DeployType == o11y.ByocHostType ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2Byoc ||
		cluster.DeployTypeV2 == o11y.DeployTypeV2ByocResourcePool
}

func usesCanonicalEssentialV2StorageIdentity(cluster *db.Cluster) bool {
	return cluster != nil &&
		cluster.DeployTypeV2 == o11y.DeployTypeV2EssentialV2 &&
		cluster.Organization.DbaasOrgID == 0
}

func canonicalEssentialV2StorageTenant(cluster *db.Cluster, o11yCluster *o11y.ClusterInfo) (string, error) {
	if o11yCluster != nil {
		tenantID := strings.TrimSpace(o11yCluster.TenantID)
		if tenantID != "" && tenantID != "0" {
			return tenantID, nil
		}
	}
	if cluster != nil && cluster.TenantID != 0 {
		return strconv.FormatUint(cluster.TenantID, 10), nil
	}
	return "", errors.New("canonical Essential V2 storage tenant is empty")
}

func objectStorageURL(schema string, location *o11y.ObjectStorageLocation) string {
	if location == nil || location.BucketName == "" || location.ObjectPath == "" {
		return ""
	}
	return fmt.Sprintf("%s://%s/%s", schema, location.BucketName, strings.TrimLeft(location.ObjectPath, "/"))
}

// Resolve resolves the o11y URLs for a given cluster
func (r *ClusterURLsResolver) Resolve(ctx context.Context, cluster *db.Cluster) (*types.ClusterURLs, error) {
	if r.o11yCli == nil {
		return nil, fmt.Errorf("o11y is not enabled")
	}
	objectStorageSchema := "s3"
	switch cluster.Provider {
	case "gcp":
		objectStorageSchema = "gs"
	case "alicloud":
		objectStorageSchema = "oss"
	case "azure":
		objectStorageSchema = "abfss"
	}

	// bucket 后缀: prod 用 o11y-prod-shared-{region}，非 prod 统一用 o11y-prod-shared-{region}-staging
	bucketSuffix := ""
	if r.o11yCli.GetEnv() != "prod" {
		bucketSuffix = "-staging"
	}

	clusterURLs := &types.ClusterURLs{}
	clusterURLs.LogStatsS3URL = r.logStatsS3URL(cluster, objectStorageSchema, bucketSuffix)
	if cluster.DeployType == o11y.PremiumType || cluster.DeployType == o11y.NgHostType {
		// Premium & NG-Host Clusters
		region := cluster.Region
		if cluster.Provider == "aws" {
			if region == "ap-south-1" {
				// ap-south-1 keeps the raw region without any premium/ng suffix.
			} else if region == "ap-southeast-2" {
				region += "-ng"
			} else {
				region += "-premium"
			}
		}
		clusterURLs.ClusterMetricsURL = r.multitenantMetricsURL(region, cluster.Provider)
		clusterURLs.DataPlaneMetricsURL = clusterURLs.ClusterMetricsURL

		o11yCluster, err := r.o11yCli.GetNextGenTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			nextGenTidbTenantID(cluster),
			cluster.ClusterID,
		)
		if err != nil {
			if usesCanonicalEssentialV2StorageIdentity(cluster) {
				return nil, fmt.Errorf("failed to resolve Essential V2 storage identity: %w", err)
			}
			log.Error("failed to get premium cluster from o11y", zap.Error(err))
		} else {
			if cluster.Provider == "aws" && cluster.Region == "ap-south-1" {
				if internalReadURL := r.internalReadMetricsURL(o11yCluster); internalReadURL != "" {
					clusterURLs.ClusterMetricsURL = internalReadURL
					clusterURLs.DataPlaneMetricsURL = internalReadURL
				}
			}
			storageTenant := fmt.Sprintf("%d", cluster.Organization.DbaasOrgID)
			if usesCanonicalEssentialV2StorageIdentity(cluster) {
				storageTenant, err = canonicalEssentialV2StorageTenant(cluster, o11yCluster)
				if err != nil {
					return nil, err
				}
				if storageLocation, ok := o11yCluster.StorageLocation(o11y.StorageCategorySlowQuery); ok {
					clusterURLs.SlowQueryS3URL = objectStorageURL(objectStorageSchema, storageLocation)
				}
			}
			if clusterURLs.SlowQueryS3URL == "" {
				clusterURLs.SlowQueryS3URL = fmt.Sprintf("%s://o11y-prod-shared-%s%s/deltalake/%s/%s/slowlogs",
					objectStorageSchema,
					region,
					bucketSuffix,
					storageTenant,
					o11yCluster.ID)
			}
			clusterURLs.TopSQLS3URL = fmt.Sprintf("%s://o11y-prod-shared-%s%s/deltalake/org=%s/cluster=%s/type=topsql/",
				objectStorageSchema,
				cluster.Region,
				bucketSuffix,
				storageTenant,
				cluster.ClusterID)
		}
	} else if cluster.DeployType == o11y.DedicatedType || (cluster.DeployType == o11y.PremiumType || cluster.DeployType == o11y.NgHostType) {
		// Dedicated Clusters
		o11yCluster, err := r.o11yCli.GetDedicatedTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.Organization.DbaasOrgID),
			cluster.ClusterID,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to get cluster from o11y: %w", err)
		} else {
			clusterURLs.ClusterMetricsURL = r.internalReadMetricsURL(o11yCluster)
			clusterURLs.SlowQueryS3URL = fmt.Sprintf("%s://o11y-prod-shared-%s%s/deltalake/%d/%s/slowlogs",
				objectStorageSchema,
				cluster.Region,
				bucketSuffix,
				cluster.Organization.DbaasOrgID,
				o11yCluster.ID)
			clusterURLs.TopSQLS3URL = fmt.Sprintf("%s://o11y-prod-shared-%s%s/deltalake/org=%d/cluster=%s/type=topsql/",
				objectStorageSchema,
				cluster.Region,
				bucketSuffix,
				cluster.Organization.DbaasOrgID,
				cluster.ClusterID)
		}

		var k8sInfraO11yCluster *o11y.ClusterInfo
		switch cluster.DeployType {
		case o11y.NgHostType, o11y.PremiumType:
			k8sInfraO11yCluster, err = r.o11yCli.GetNextGenK8sCluster(
				context.Background(),
				cluster.Provider,
				cluster.Region,
				fmt.Sprintf("%d", cluster.TenantID),
				cluster.DataPlaneName,
			)
		default:
			// A dedicated cluster can retain its historical O11Y project after its
			// DBaaS project changes. Use the project returned by GS for the cluster
			// instead of the mutable DBaaS project stored in Clinic.
			k8sInfraO11yCluster, err = r.o11yCli.GetProjectK8sCluster(
				context.Background(),
				cluster.Provider,
				cluster.Region,
				fmt.Sprintf("%d", cluster.TenantID),
				o11yCluster.ProjectID,
			)
		}

		if err != nil {
			return nil, fmt.Errorf("failed to get k8s-infra cluster from o11y: %w", err)
		} else {
			clusterURLs.DataPlaneMetricsURL = r.internalReadMetricsURL(k8sInfraO11yCluster)
		}
	} else if cluster.DeployType == o11y.SharedType {
		o11yClusters, err := r.o11yCli.GetServerlessTidbCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.TenantID),
			cluster.ClusterID,
		)

		if err != nil || len(o11yClusters) == 0 {
			return nil, fmt.Errorf("failed to get cluster from o11y: %w", err)
		} else {
			o11yCluster := selectServerlessO11yCluster(cluster, o11yClusters)
			clusterURLs.ClusterMetricsURL = r.internalReadMetricsURL(&o11yCluster)
		}
	} else if cluster.DeployType == o11y.ByocHostType || cluster.DeployType == o11y.ByocType {
		o11yCluster, err := r.o11yCli.GetByocK8sCluster(
			ctx,
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.TenantID),
			cluster.ClusterID,
		)

		if err != nil {
			return nil, fmt.Errorf("failed to get cluster from o11y: %w", err)
		} else {
			clusterURLs.ClusterMetricsURL = r.internalReadMetricsURL(o11yCluster)
			clusterURLs.SlowQueryS3URL = fmt.Sprintf("%s://o11y-prod-shared-%s%s/deltalake/%d/%s/slowlogs",
				objectStorageSchema,
				cluster.Region,
				bucketSuffix,
				cluster.Organization.DbaasOrgID,
				o11yCluster.ID)
			clusterURLs.TopSQLS3URL = fmt.Sprintf("%s://o11y-prod-shared-%s%s/deltalake/org=%d/cluster=%s/type=topsql/",
				objectStorageSchema,
				cluster.Region,
				bucketSuffix,
				cluster.Organization.DbaasOrgID,
				cluster.ClusterID)
		}

		k8sInfraO11yCluster, err := r.o11yCli.GetByocK8sCluster(
			context.Background(),
			cluster.Provider,
			cluster.Region,
			fmt.Sprintf("%d", cluster.TenantID),
			cluster.DataPlaneName,
		)

		if err != nil {
			return nil, fmt.Errorf("failed to get k8s-infra cluster from o11y: %w", err)
		} else {
			// Normal Dedicated Projects
			clusterURLs.DataPlaneMetricsURL = r.internalReadMetricsURL(k8sInfraO11yCluster)
		}
	} else {
		log.Error("Not supported to get cluster metrics URL",
			zap.String("clusterID", cluster.ClusterID),
			zap.String("deployType", cluster.DeployType),
		)
	}

	if useBYOCLogsTarget(cluster) {
		logsTarget, err := r.ResolveLogsTarget(ctx, cluster)
		if err != nil {
			log.Error("failed to resolve BYOC cluster logs target",
				zap.String("clusterID", cluster.ClusterID),
				zap.String("deployType", cluster.DeployType),
				zap.String("deployTypeV2", cluster.DeployTypeV2),
				zap.String("provider", cluster.Provider),
				zap.String("region", cluster.Region),
				zap.Error(err))
		} else {
			clusterURLs.LokiURL = logsTarget.LogQueryURL
		}
	} else {
		logsTarget, err := r.ResolveLogsTarget(ctx, cluster)
		if err != nil {
			log.Error("failed to resolve cluster logs target",
				zap.String("clusterID", cluster.ClusterID),
				zap.String("deployType", cluster.DeployType),
				zap.String("deployTypeV2", cluster.DeployTypeV2),
				zap.String("provider", cluster.Provider),
				zap.String("region", cluster.Region),
				zap.Error(err))
		} else {
			clusterURLs.LokiURL = logsTarget.LogQueryURL
		}
	}

	if isPremiumStatementsCluster(cluster) {
		clusterURLs.StatementsS3URL = statementsS3URLFromSlowQueryURL(clusterURLs.SlowQueryS3URL)
	}

	return clusterURLs, nil
}

// IsEnabled returns whether the o11y client is enabled
func (r *ClusterURLsResolver) IsEnabled() bool {
	return r.o11yCli != nil
}