package core

import (
	"strings"
	"sync"

	"github.com/pingcap/log"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/util/yaml"
)

const managerConfigKey = "tiflash-manager-config.yaml"

type PoolSpec struct {
	Create       bool   `yaml:"create,omitempty"`
	TemplatePool string `yaml:"templatePool,omitempty"`
}

func (p PoolSpec) EffectiveTemplatePool(defaultPool string) string {
	if p.TemplatePool != "" {
		return p.TemplatePool
	}
	return defaultPool
}

type ManagerConfig struct {
	Keyspaces  map[string]string   `yaml:"keyspaces"`
	Specialize map[string]string   `yaml:"specialize"` // Deprecated: use keyspaces.
	Pools      map[string]PoolSpec `yaml:"pools"`
}

func parseManagerConfig(data map[string]string) (*ManagerConfig, error) {
	content := strings.TrimSpace(data[managerConfigKey])
	if content == "" {
		return &ManagerConfig{
			Keyspaces:  map[string]string{},
			Specialize: map[string]string{},
			Pools:      map[string]PoolSpec{},
		}, nil
	}

	var cfg ManagerConfig
	if err := yaml.Unmarshal([]byte(content), &cfg); err != nil {
		return nil, err
	}
	if cfg.Specialize == nil {
		cfg.Specialize = map[string]string{}
	}
	if cfg.Keyspaces == nil {
		cfg.Keyspaces = map[string]string{}
	}
	if cfg.Pools == nil {
		cfg.Pools = map[string]PoolSpec{}
	}
	return &cfg, nil
}

func (c *ManagerConfig) keyspacePools() map[string]string {
	mappings := make(map[string]string, len(c.Specialize)+len(c.Keyspaces))
	for keyspaceID, poolName := range c.Specialize {
		mappings[keyspaceID] = poolName
	}
	for keyspaceID, poolName := range c.Keyspaces {
		mappings[keyspaceID] = poolName
	}
	return mappings
}

func (c *ManagerConfig) DesiredPools(defaultPool string) map[string]PoolSpec {
	desired := make(map[string]PoolSpec)
	if defaultPool != "" {
		desired[defaultPool] = c.Pools[defaultPool]
	}

	for poolName, spec := range c.Pools {
		if poolName == "" {
			continue
		}
		desired[poolName] = spec
	}

	for _, poolName := range c.keyspacePools() {
		if poolName == "" {
			continue
		}
		if _, ok := desired[poolName]; !ok {
			desired[poolName] = PoolSpec{}
		}
	}
	return desired
}

func storeTenantsPool(cfg *ManagerConfig, pools *sync.Map) {
	keyspacePools := cfg.keyspacePools()
	for keyspaceID, pool := range keyspacePools {
		if pool == "" {
			log.Warn("Keyspace pool is empty", zap.String("keyspaceID", keyspaceID))
			continue
		}

		log.Info("Storing keyspace in pool", zap.String("keyspaceID", keyspaceID), zap.String("pool", pool))
		pools.Store(keyspaceID, pool)
	}

	pools.Range(func(key, value any) bool {
		keyspaceID := key.(string)
		pool := value.(string)

		if _, ok := keyspacePools[keyspaceID]; !ok {
			log.Info("Removing keyspace from pool", zap.String("keyspaceID", keyspaceID), zap.String("pool", pool))
			pools.Delete(keyspaceID)
		}
		return true
	})
}