package proxy

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"regexp"
	"sort"
	"strings"
	"sync"
	"time"

	gocache "github.com/patrickmn/go-cache"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/log"
	"github.com/pingcap-inc/clinic/pkg/o11y"
	o11yurls "github.com/pingcap-inc/clinic/pkg/o11y/urls"
)

const (
	resourcePoolTargetTTL  = 5 * time.Minute
	logicalClusterRouteTTL = 24 * time.Hour
	accountRefreshWorkers  = 8
)

var (
	vmAccountFilterPattern = regexp.MustCompile(`^\{?\s*vm_account_id\s*=\s*"([0-9]+)"\s*\}?$`)
	vmAccountIDPattern     = regexp.MustCompile(`^[0-9]+$`)
)

type metricsTargetResolver interface {
	ResolveMetricsTarget(context.Context, *db.Cluster, o11yurls.MetricsScope, string) (*o11yurls.MetricsTarget, error)
}

// resourcePoolMetricsTargetResolver builds a single VictoriaMetrics target for
// the logical-cluster TiDB metrics and shared resource-pool metrics displayed
// together by the Essential V2 Shared Resource OnePage dashboard.
type resourcePoolMetricsTargetResolver struct {
	db                      *gorm.DB
	targetResolver          metricsTargetResolver
	resourcePoolTargetCache *gocache.Cache
	routeTTL                time.Duration
	now                     func() time.Time
	poolLocks               sync.Map
}

func newResourcePoolMetricsTargetResolver(gormDB *gorm.DB, cli *o11y.Client) *resourcePoolMetricsTargetResolver {
	return &resourcePoolMetricsTargetResolver{
		db:                      gormDB,
		targetResolver:          o11yurls.NewClusterURLsResolver(cli),
		resourcePoolTargetCache: gocache.New(resourcePoolTargetTTL, resourcePoolTargetTTL),
		routeTTL:                logicalClusterRouteTTL,
		now:                     time.Now,
	}
}

func (r *resourcePoolMetricsTargetResolver) resolve(
	ctx context.Context,
	pool *db.Cluster,
	base *o11yurls.MetricsTarget,
) (*o11yurls.MetricsTarget, error) {
	if r == nil || r.db == nil || r.targetResolver == nil {
		return nil, errors.New("shared resource metrics resolver is not initialized")
	}
	if pool == nil || pool.DeployTypeV2 != o11y.DeployTypeV2EssentialV2ResourcePool {
		return base, nil
	}
	if base == nil || base.MetricsURL == "" {
		return nil, errors.New("base multitenant metrics target is empty")
	}

	cacheKey := pool.ClusterID
	if cached, ok := r.resourcePoolTargetCache.Get(cacheKey); ok {
		if metricsURL, valid := cached.(string); valid && metricsURL != "" {
			return cloneMetricsTarget(base, metricsURL), nil
		}
	}

	lockValue, _ := r.poolLocks.LoadOrStore(cacheKey, &sync.Mutex{})
	poolLock := lockValue.(*sync.Mutex)
	poolLock.Lock()
	defer poolLock.Unlock()
	if cached, ok := r.resourcePoolTargetCache.Get(cacheKey); ok {
		if metricsURL, valid := cached.(string); valid && metricsURL != "" {
			return cloneMetricsTarget(base, metricsURL), nil
		}
	}

	metricsURL, err := r.buildMetricsURL(ctx, pool, base.MetricsURL)
	if err != nil {
		return nil, err
	}
	r.resourcePoolTargetCache.Set(cacheKey, metricsURL, resourcePoolTargetTTL)
	return cloneMetricsTarget(base, metricsURL), nil
}

func cloneMetricsTarget(base *o11yurls.MetricsTarget, metricsURL string) *o11yurls.MetricsTarget {
	return &o11yurls.MetricsTarget{
		MetricsURL:  metricsURL,
		O11yCluster: base.O11yCluster,
	}
}

func (r *resourcePoolMetricsTargetResolver) buildMetricsURL(ctx context.Context, pool *db.Cluster, baseURL string) (string, error) {
	resourcePoolAccountID, err := vmAccountIDFromURL(baseURL)
	if err != nil {
		return "", fmt.Errorf("parse resource-pool VM account: %w", err)
	}

	var logicalClusters []db.Cluster
	// Resource-pool dashboards support historical time ranges after a pool or
	// logical cluster has been deleted, so membership lookup must include GORM
	// soft-deleted logical-cluster rows.
	if err := r.db.WithContext(ctx).Unscoped().
		Where("parent_id = ? AND deploy_type_v2 = ?", pool.ClusterID, o11y.DeployTypeV2EssentialV2).
		Find(&logicalClusters).Error; err != nil {
		return "", fmt.Errorf("list resource-pool logical clusters: %w", err)
	}

	accountIDs := map[string]struct{}{resourcePoolAccountID: {}}
	if len(logicalClusters) > 0 {
		logicalClusterAccounts, err := r.resolveLogicalClusterAccounts(ctx, pool, logicalClusters)
		if err != nil {
			return "", err
		}
		for _, accountID := range logicalClusterAccounts {
			accountIDs[accountID] = struct{}{}
		}
	}
	return buildResourcePoolMetricsURL(baseURL, pool.ClusterID, accountIDs)
}

func (r *resourcePoolMetricsTargetResolver) resolveLogicalClusterAccounts(
	ctx context.Context,
	pool *db.Cluster,
	logicalClusters []db.Cluster,
) (map[string]string, error) {
	clusterIDs := make([]string, 0, len(logicalClusters))
	for i := range logicalClusters {
		clusterIDs = append(clusterIDs, logicalClusters[i].ClusterID)
	}

	var routes []db.ResourcePoolO11yRoute
	if err := r.db.WithContext(ctx).
		Where("resource_pool_id = ? AND cluster_id IN ?", pool.ClusterID, clusterIDs).
		Find(&routes).Error; err != nil {
		return nil, fmt.Errorf("load resource-pool O11y routes: %w", err)
	}
	routeByCluster := make(map[string]db.ResourcePoolO11yRoute, len(routes))
	for _, route := range routes {
		routeByCluster[route.ClusterID] = route
	}

	now := r.now()
	result := make(map[string]string, len(logicalClusters))
	stale := make([]db.Cluster, 0)
	for i := range logicalClusters {
		logicalCluster := logicalClusters[i]
		route, found := routeByCluster[logicalCluster.ClusterID]
		fresh := found && route.VMAccountID != "" &&
			route.RefreshedAt.After(now.Add(-r.routeTTL)) &&
			!logicalCluster.UpdatedAt.After(route.RefreshedAt)
		if fresh {
			result[logicalCluster.ClusterID] = route.VMAccountID
			continue
		}
		stale = append(stale, logicalCluster)
	}

	refreshed, refreshErrors := r.refreshAccounts(ctx, stale)
	for clusterID, accountID := range refreshed {
		result[clusterID] = accountID
	}
	for _, logicalCluster := range stale {
		if _, ok := refreshed[logicalCluster.ClusterID]; ok {
			continue
		}
		if route, ok := routeByCluster[logicalCluster.ClusterID]; ok && route.VMAccountID != "" {
			result[logicalCluster.ClusterID] = route.VMAccountID
			log.Warn(ctx, "use stale resource-pool O11y route after refresh failure",
				zap.String("resourcePoolID", pool.ClusterID),
				zap.String("logicalClusterID", logicalCluster.ClusterID),
				zap.Error(refreshErrors[logicalCluster.ClusterID]))
			continue
		}
		log.Warn(ctx, "skip logical cluster without an O11y VM account",
			zap.String("resourcePoolID", pool.ClusterID),
			zap.String("logicalClusterID", logicalCluster.ClusterID),
			zap.Error(refreshErrors[logicalCluster.ClusterID]))
	}

	if len(refreshed) > 0 {
		rows := make([]db.ResourcePoolO11yRoute, 0, len(refreshed))
		for clusterID, accountID := range refreshed {
			rows = append(rows, db.ResourcePoolO11yRoute{
				ResourcePoolID: pool.ClusterID,
				ClusterID:      clusterID,
				VMAccountID:    accountID,
				RefreshedAt:    now,
			})
		}
		if err := r.db.WithContext(ctx).Clauses(clause.OnConflict{
			Columns: []clause.Column{{Name: "cluster_id"}},
			DoUpdates: clause.AssignmentColumns([]string{
				"resource_pool_id", "vm_account_id", "refreshed_at", "updated_at",
			}),
		}).Create(&rows).Error; err != nil {
			return nil, fmt.Errorf("save resource-pool O11y routes: %w", err)
		}
	}
	return result, nil
}

type accountRefreshResult struct {
	clusterID string
	accountID string
	err       error
}

func (r *resourcePoolMetricsTargetResolver) refreshAccounts(
	ctx context.Context,
	clusters []db.Cluster,
) (map[string]string, map[string]error) {
	accounts := make(map[string]string, len(clusters))
	errs := make(map[string]error)
	if len(clusters) == 0 {
		return accounts, errs
	}

	jobs := make(chan db.Cluster)
	results := make(chan accountRefreshResult, len(clusters))
	workerCount := min(accountRefreshWorkers, len(clusters))
	var workers sync.WaitGroup
	for range workerCount {
		workers.Add(1)
		go func() {
			defer workers.Done()
			for cluster := range jobs {
				target, err := r.targetResolver.ResolveMetricsTarget(ctx, &cluster, o11yurls.MetricsScopeChild, "")
				if err == nil && target != nil && target.O11yCluster != nil {
					accountURL := target.O11yCluster.MultitenantQueryUri
					var accountID string
					accountID, err = vmAccountIDFromURL(accountURL)
					results <- accountRefreshResult{clusterID: cluster.ClusterID, accountID: accountID, err: err}
					continue
				}
				if err == nil {
					err = errors.New("O11y multitenant query URI is empty")
				}
				results <- accountRefreshResult{clusterID: cluster.ClusterID, err: err}
			}
		}()
	}
	go func() {
		defer close(jobs)
		for _, cluster := range clusters {
			select {
			case jobs <- cluster:
			case <-ctx.Done():
				return
			}
		}
	}()
	go func() {
		workers.Wait()
		close(results)
	}()

	for result := range results {
		if result.err != nil {
			errs[result.clusterID] = result.err
			continue
		}
		accounts[result.clusterID] = result.accountID
	}
	return accounts, errs
}

func vmAccountIDFromURL(rawURL string) (string, error) {
	u, err := url.Parse(rawURL)
	if err != nil {
		return "", err
	}
	for _, filter := range u.Query()["extra_filters[]"] {
		matches := vmAccountFilterPattern.FindStringSubmatch(strings.TrimSpace(filter))
		if len(matches) == 2 {
			return matches[1], nil
		}
	}
	return "", errors.New("vm_account_id filter is missing")
}

func buildResourcePoolMetricsURL(rawURL, resourcePoolID string, accountSet map[string]struct{}) (string, error) {
	u, err := url.Parse(rawURL)
	if err != nil || u.Scheme == "" || u.Host == "" {
		return "", errors.New("invalid multitenant metrics URL")
	}
	accountIDs := make([]string, 0, len(accountSet))
	for accountID := range accountSet {
		if !vmAccountIDPattern.MatchString(accountID) {
			return "", fmt.Errorf("invalid VM account ID %q", accountID)
		}
		accountIDs = append(accountIDs, accountID)
	}
	if len(accountIDs) == 0 {
		return "", errors.New("VM account allowlist is empty")
	}
	sort.Strings(accountIDs)

	query := u.Query()
	filters := query["extra_filters[]"]
	query.Del("extra_filters[]")
	for _, filter := range filters {
		if vmAccountFilterPattern.MatchString(strings.TrimSpace(filter)) || strings.Contains(filter, "sharedpool_id") {
			continue
		}
		query.Add("extra_filters[]", filter)
	}
	query.Add("extra_filters[]", fmt.Sprintf(`{vm_account_id=~"^(%s)$"}`, strings.Join(accountIDs, "|")))
	query.Add("extra_filters[]", fmt.Sprintf(`{sharedpool_id="%s"}`, escapePrometheusLabelValue(resourcePoolID)))
	u.RawQuery = query.Encode()
	return u.String(), nil
}

func escapePrometheusLabelValue(value string) string {
	value = strings.ReplaceAll(value, `\`, `\\`)
	return strings.ReplaceAll(value, `"`, `\"`)
}