// Copyright 2021 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package label

import (
	"encoding/hex"
	"encoding/json"
	"fmt"
	"slices"
	"strconv"

	"github.com/pingcap/tidb/pkg/parser/ast"
	"github.com/pingcap/tidb/pkg/tablecodec"
	"github.com/tikv/client-go/v2/tikv"
	pd "github.com/tikv/pd/client/http"
	"gopkg.in/yaml.v2"
)

const (
	// IDPrefix is the prefix for label rule ID.
	IDPrefix = "schema"
	// KeyspacePrefix is the prefix for keyspace in label rule ID.
	KeyspacePrefix = "keyspace"
	ruleType       = "key-range"
)

const (
	// RuleIndexDefault is the default index for a rule.
	RuleIndexDefault int = iota
	// RuleIndexDatabase is the index for a rule of database.
	RuleIndexDatabase
	// RuleIndexTable is the index for a rule of table.
	RuleIndexTable
	// RuleIndexPartition is the index for a rule of partition.
	RuleIndexPartition
)

var (
	// TableIDFormat is the format of the label rule ID for a table.
	// The format follows "schema/database_name/table_name".
	TableIDFormat = "%s/%s/%s"
	// PartitionIDFormat is the format of the label rule ID for a partition.
	// The format follows "schema/database_name/table_name/partition_name".
	PartitionIDFormat = "%s/%s/%s/%s"
)

// Rule is used to establish the relationship between labels and a key range.
type Rule pd.LabelRule

// NewRule creates a rule.
func NewRule() *Rule {
	return &Rule{}
}

// ApplyAttributesSpec will transfer attributes defined in AttributesSpec to the labels.
func (r *Rule) ApplyAttributesSpec(spec *ast.AttributesSpec) error {
	if spec.Default {
		r.Labels = []pd.RegionLabel{}
		return nil
	}
	// construct a string list
	attrBytes := []byte("[" + spec.Attributes + "]")
	attributes := []string{}
	err := yaml.UnmarshalStrict(attrBytes, &attributes)
	if err != nil {
		return err
	}
	r.Labels, err = NewLabels(attributes)
	if err != nil {
		return err
	}
	return ValidateRegionPolicy(r.Labels)
}

// String implements fmt.Stringer.
func (r *Rule) String() string {
	t, err := json.Marshal(r)
	if err != nil {
		return ""
	}
	return string(t)
}

// Clone clones a rule.
func (r *Rule) Clone() *Rule {
	newRule := NewRule()
	*newRule = *r
	return newRule
}

// NewRuleID generates a new rule ID for a table or partition.
func NewRuleID(tikvCodec tikv.Codec, dbName, tableName, partName string) string {
	var id string
	isPartition := partName != ""
	if isPartition {
		id = fmt.Sprintf(PartitionIDFormat, IDPrefix, dbName, tableName, partName)
	} else {
		id = fmt.Sprintf(TableIDFormat, IDPrefix, dbName, tableName)
	}
	if tikvCodec.GetKeyspaceMeta() != nil {
		id = fmt.Sprintf("%s/%d/%s", KeyspacePrefix, tikvCodec.GetKeyspaceID(), id)
	}
	return id
}

// Reset will reset the label rule for a table/partition with a given ID and names.
func (r *Rule) Reset(tikvCodec tikv.Codec, dbName, tableName, partName string, ids ...int64) *Rule {
	isPartition := partName != ""
	r.ID = NewRuleID(tikvCodec, dbName, tableName, partName)
	if len(r.Labels) == 0 {
		return r
	}
	var hasKeyspaceKey, hasDBKey, hasTableKey, hasPartitionKey bool
	for i := range r.Labels {
		switch r.Labels[i].Key {
		case keyspaceKey:
			r.Labels[i].Value = strconv.FormatInt(int64(tikvCodec.GetKeyspaceID()), 10)
			hasKeyspaceKey = true
		case dbKey:
			r.Labels[i].Value = dbName
			hasDBKey = true
		case tableKey:
			r.Labels[i].Value = tableName
			hasTableKey = true
		case partitionKey:
			if partName != "" {
				r.Labels[i].Value = partName
				hasPartitionKey = true
			}
		default:
		}
	}

	if tikvCodec.GetKeyspaceMeta() != nil && !hasKeyspaceKey {
		r.Labels = append(r.Labels, pd.RegionLabel{Key: keyspaceKey, Value: strconv.FormatInt(int64(tikvCodec.GetKeyspaceID()), 10)})
	}

	if !hasDBKey {
		r.Labels = append(r.Labels, pd.RegionLabel{Key: dbKey, Value: dbName})
	}

	if !hasTableKey {
		r.Labels = append(r.Labels, pd.RegionLabel{Key: tableKey, Value: tableName})
	}

	if isPartition && !hasPartitionKey {
		r.Labels = append(r.Labels, pd.RegionLabel{Key: partitionKey, Value: partName})
	}
	r.RuleType = ruleType
	dataSlice := make([]any, 0, len(ids))
	slices.Sort(ids)
	for i := 0; i < len(ids); i++ {
		// Label rules are consumed as region boundary keys, so V2 must mem-encode the
		// whole outer key instead of prefixing a mem-encoded table key.
		startKey, endKey := tikvCodec.EncodeRegionRange(tablecodec.GenTablePrefix(ids[i]), tablecodec.GenTablePrefix(ids[i]+1))
		data := map[string]string{
			"start_key": hex.EncodeToString(startKey),
			"end_key":   hex.EncodeToString(endKey),
		}
		dataSlice = append(dataSlice, data)
	}
	r.Data = dataSlice
	// We may support more types later.
	r.Index = RuleIndexTable
	if isPartition {
		r.Index = RuleIndexPartition
	}
	return r
}

// NewRulePatch returns a patch of rules which need to be set or deleted.
func NewRulePatch(setRules []*Rule, deleteRules []string) *pd.LabelRulePatch {
	labelRules := make([]*pd.LabelRule, 0, len(setRules))
	for _, rule := range setRules {
		labelRules = append(labelRules, (*pd.LabelRule)(rule))
	}
	return &pd.LabelRulePatch{
		SetRules:    labelRules,
		DeleteRules: deleteRules,
	}
}