// Copyright 2024 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,
// See the License for the specific language governing permissions and
// limitations under the License.

package placement

import (
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strconv"
	"strings"

	"github.com/pingcap/kvproto/pkg/metapb"
)

// LabelConstraintOp defines how a LabelConstraint matches a store. It can be one of
// 'in', 'notIn', 'exists', or 'notExists'.
type LabelConstraintOp string

const (
	// In restricts the store label value should in the value list.
	// If label does not exist, `in` is always false.
	In LabelConstraintOp = "in"
	// NotIn restricts the store label value should not in the value list.
	// If label does not exist, `notIn` is always true.
	NotIn LabelConstraintOp = "notIn"
	// Exists restricts the store should have the label.
	Exists LabelConstraintOp = "exists"
	// NotExists restricts the store should not have the label.
	NotExists LabelConstraintOp = "notExists"
)

const (
	// PartitionLabel is label key of partition.
	PartitionLabel = "serverless.tidbcloud.com/partition"

	// TierLabel is label key of tier.
	TierLabel = "serverless.tidbcloud.com/tier"
	// HotTier represents hot tier.
	HotTier = "hot"
	// StandardTier represents standard tier.
	StandardTier = "standard"
)

var HotTierLabelConstraint = LabelConstraint{
	Key:    TierLabel,
	Op:     In,
	Values: []string{HotTier},
}

var StandardTierLabelConstraint = LabelConstraint{
	Key:    TierLabel,
	Op:     In,
	Values: []string{StandardTier},
}

func TierLabelConstraint(tier string) LabelConstraint {
	return LabelConstraint{
		Key:    TierLabel,
		Op:     In,
		Values: []string{tier},
	}
}

const (
	// ZoneLabel is label key of zone.
	ZoneLabel = "serverless.tidbcloud.com/zone"
	// PrimaryZone represents primary zone.
	PrimaryZone = "primary"
	// StandbyZone represents standby zone.
	StandbyZone = "standby"
	// WitnessZone represents witness zone.
	WitnessZone = "witness"
)

var PrimaryZoneLabelConstraint = LabelConstraint{
	Key:    ZoneLabel,
	Op:     In,
	Values: []string{PrimaryZone},
}

var StandbyZoneLabelConstraint = LabelConstraint{
	Key:    ZoneLabel,
	Op:     In,
	Values: []string{StandbyZone},
}

var WitnessZoneLabelConstraint = LabelConstraint{
	Key:    ZoneLabel,
	Op:     NotIn,
	Values: []string{PrimaryZone, StandbyZone},
}

const (
	DefaultMultiAZRuleGroup   = "default-multi-az"
	SingleAZKeyspaceRuleGroup = "single-az-keyspace"
	MultiAZKeyspaceRuleGroup  = "multi-az-keyspace"
	HotRegionRuleGroup        = "hot-region"
	IsolatedRuleGroup         = "isolated"
)

var MultiAZRuleGroups = map[string]int{
	DefaultMultiAZRuleGroup:   10,
	SingleAZKeyspaceRuleGroup: 30,
	MultiAZKeyspaceRuleGroup:  50,
}

const (
	HotRegionRuleGroupIndex = 80
	IsolatedRuleGroupIndex  = 90
)

// LabelConstraint is used to filter store when trying to place peer of a region.
type LabelConstraint struct {
	Key    string            `json:"key,omitempty"`
	Op     LabelConstraintOp `json:"op,omitempty"`
	Values []string          `json:"values,omitempty"`
}

func (l LabelConstraint) String() string {
	b, _ := json.Marshal(l)
	return string(b)
}

// RuleOpType indicates the operation type
type RuleOpType string

const (
	// RuleOpAdd a placement rule, only need to specify the field *Rule
	RuleOpAdd RuleOpType = "add"
	// RuleOpDel a placement rule, only need to specify the field `GroupID`, `ID`, `MatchID`
	RuleOpDel RuleOpType = "del"
)

// RuleOp is for batching placement rule actions. The action type is
// distinguished by the field `Action`.
type RuleOp struct {
	*Rule                       // information of the placement rule to add/delete the operation type
	Action           RuleOpType `json:"action"`
	DeleteByIDPrefix bool       `json:"delete_by_id_prefix"` // if action == delete, delete by the prefix of id
}

func (r RuleOp) String() string {
	b, _ := json.Marshal(r)
	return string(b)
}

// PeerRoleType is the expected peer type of the placement rule.
type PeerRoleType string

const (
	// Voter can either match a leader peer or follower peer
	Voter PeerRoleType = "voter"
	// Leader matches a leader.
	Leader PeerRoleType = "leader"
	// Follower matches a follower.
	Follower PeerRoleType = "follower"
	// Learner matches a learner.
	Learner PeerRoleType = "learner"
)

// MetaPeerRole converts placement.PeerRoleType to metapb.PeerRole.
func (s PeerRoleType) MetaPeerRole() metapb.PeerRole {
	if s == Learner {
		return metapb.PeerRole_Learner
	}
	return metapb.PeerRole_Voter
}

// Rule is the placement rule that can be checked against a region. When
// applying rules (apply means schedule regions to match selected rules), the
// apply order is defined by the tuple [GroupIndex, GroupID, Index, ID].
//
// NOTE: This type is exported by HTTP API. Please pay more attention when modifying it.
type Rule struct {
	GroupID          string            `json:"group_id"`                    // mark the source that add the rule
	ID               string            `json:"id"`                          // unique ID within a group
	Index            int               `json:"index,omitempty"`             // rule apply order in a group, rule with less ID is applied first when indexes are equal
	Override         bool              `json:"override,omitempty"`          // when it is true, all rules with less indexes are disabled
	StartKey         string            `json:"start_key"`                   // hex format start key, for marshal/unmarshal
	EndKey           string            `json:"end_key"`                     // hex format end key, for marshal/unmarshal
	Role             PeerRoleType      `json:"role"`                        // expected role of the peers
	IsWitness        bool              `json:"is_witness"`                  // when it is true, it means the role is also a witness
	Count            int               `json:"count"`                       // expected count of the peers
	LabelConstraints []LabelConstraint `json:"label_constraints,omitempty"` // used to select stores to place peers
	LocationLabels   []string          `json:"location_labels,omitempty"`   // used to make peers isolated physically
	IsolationLevel   string            `json:"isolation_level,omitempty"`   // used to isolate replicas explicitly and forcibly
	Version          uint64            `json:"version,omitempty"`           // only set at runtime, add 1 each time rules updated, begin from 0.
}

func (r *Rule) String() string {
	b, _ := json.Marshal(r)
	return string(b)
}

// Clone returns a copy of Rule.
func (r *Rule) Clone() *Rule {
	var clone Rule
	json.Unmarshal([]byte(r.String()), &clone)
	return &clone
}

// RuleGroup defines properties of a rule group.
type RuleGroup struct {
	ID       string `json:"id,omitempty"`
	Index    int    `json:"index,omitempty"`
	Override bool   `json:"override,omitempty"`
}

func (g *RuleGroup) isDefault() bool {
	return g.Index == 0 && !g.Override
}

func (g *RuleGroup) String() string {
	b, _ := json.Marshal(g)
	return string(b)
}

// Clone returns a copy of RuleGroup.
func (g *RuleGroup) Clone() *RuleGroup {
	return &RuleGroup{
		ID:       g.ID,
		Index:    g.Index,
		Override: g.Override,
	}
}

func HasNewLabelConstraint(labelConstraints []LabelConstraint, rule *Rule, ignoreLabelKeys ...string) bool {
	var ignoreKeys = map[string]struct{}{}
	for _, ignoreLabelKey := range ignoreLabelKeys {
		ignoreKeys[ignoreLabelKey] = struct{}{}
	}
	for _, labelConstraint := range labelConstraints {
		if _, ok := ignoreKeys[labelConstraint.Key]; ok {
			continue
		}
		if UpdateRuleLabelConstraint(rule, labelConstraint) {
			return true
		}
	}
	return false
}

func UpdateRuleLabelConstraint(rule *Rule, labelConstraint LabelConstraint) (updated bool) {
	idx, lc, ok := GetLabelConstraintFromSlice(rule.LabelConstraints, labelConstraint.Key)
	var equal = false
	if ok && lc.Op == labelConstraint.Op && len(lc.Values) == len(labelConstraint.Values) {
		sort.Strings(lc.Values)
		sort.Strings(labelConstraint.Values)
		equal = reflect.DeepEqual(lc.Values, labelConstraint.Values)
	}
	if !equal {
		if ok {
			rule.LabelConstraints[idx] = labelConstraint
		} else {
			rule.LabelConstraints = append(rule.LabelConstraints, labelConstraint)
		}
	}
	updated = !equal
	return
}

func GetLabelConstraintFromSlice(labelConstraints []LabelConstraint, labelKey string) (index int, labelConstraint LabelConstraint, ok bool) {
	for i, labelConstraint := range labelConstraints {
		if labelConstraint.Key == labelKey {
			return i, labelConstraint, true
		}
	}
	return 0, LabelConstraint{}, false
}

const spaceIDMax = ^uint32(0) >> 8 // 16777215 (Uint24Max) is the maximum value of spaceID.

// FullKeyspaceTxnRange constructs the txn range of the all keyspaces.
func FullKeyspaceTxnRange() (string, string) {
	firstkeyspaceIDBytes := make([]byte, 4)
	lastKeyspaceIDBytes := make([]byte, 4)
	binary.BigEndian.PutUint32(firstkeyspaceIDBytes, 1)
	binary.BigEndian.PutUint32(lastKeyspaceIDBytes, spaceIDMax)
	startKey := EncodeBytes(append([]byte{'x'}, firstkeyspaceIDBytes[1:]...))
	endKey := EncodeBytes(append(append([]byte{'x'}, lastKeyspaceIDBytes[1:]...), []byte{255, 255, 255}...))
	return strings.ToUpper(hex.EncodeToString(startKey)), strings.ToUpper(hex.EncodeToString(endKey))
}

// KeyspaceTxnRange constructs the txn range of the given keyspace.
func KeyspaceTxnRange(id uint32) (string, string) {
	keyspaceIDBytes := make([]byte, 4)
	nextKeyspaceIDBytes := make([]byte, 4)
	binary.BigEndian.PutUint32(keyspaceIDBytes, id)
	binary.BigEndian.PutUint32(nextKeyspaceIDBytes, id+1)
	startKey := EncodeBytes(append([]byte{'x'}, keyspaceIDBytes[1:]...))
	endKey := EncodeBytes(append([]byte{'x'}, nextKeyspaceIDBytes[1:]...))
	return strings.ToUpper(hex.EncodeToString(startKey)), strings.ToUpper(hex.EncodeToString(endKey))
}

// KeyspaceTableTxnRange constructs the txn range of the given keyspace and table.
func KeyspaceTableTxnRange(keyspaceID uint32, tableID uint64) (string, string) {
	startKeyBytes := make([]byte, 5, 4+9)
	binary.BigEndian.PutUint32(startKeyBytes, keyspaceID)
	startKeyBytes[0] = 'x'
	startKeyBytes[4] = 't'
	endKeyBytes := make([]byte, 5, 4+9)
	copy(endKeyBytes, startKeyBytes)
	startKeyBytes = EncodeInt(startKeyBytes, int64(tableID))
	endKeyBytes = EncodeInt(endKeyBytes, int64(tableID+1))
	startKey := EncodeBytes(startKeyBytes)
	endKey := EncodeBytes(endKeyBytes)
	return strings.ToUpper(hex.EncodeToString(startKey)), strings.ToUpper(hex.EncodeToString(endKey))
}

func FormatKeyspaceID(keyspaceID uint32) string {
	keyspaceIDBytes := make([]byte, 4)
	binary.BigEndian.PutUint32(keyspaceIDBytes, keyspaceID)
	return hex.EncodeToString(append([]byte{'x'}, keyspaceIDBytes[1:]...))
}

func ParseKeyspaceID(key string) (keyspaceID uint32, err error) {
	if len(key) < 8 || key[:2] != "78" {
		return 0, fmt.Errorf("invalid key [%s]", key)
	}
	idHex := key[2:8]
	d, err := hex.DecodeString(idHex)
	if err != nil {
		return 0, err
	}
	keyspaceID = uint32(d[0])<<16 + uint32(d[1])<<8 + uint32(d[2])
	return keyspaceID, nil
}

func FormatKeyspaceRuleID(keyspaceID uint32) string {
	return fmt.Sprintf("keyspace-%s", FormatKeyspaceID(keyspaceID))
}

func ParseKeyspaceRuleID(name string) (leftover string, keyspaceID uint32, err error) {
	if !strings.HasPrefix(name, "keyspace-") {
		return "", 0, fmt.Errorf("invalid format [%s]", name)
	}
	name = strings.TrimPrefix(name, "keyspace-")
	keyspaceID, err = ParseKeyspaceID(name[:8])
	if len(name) > 8 && name[8] == '-' {
		leftover = name[9:]
	} else {
		leftover = name[8:]
	}
	return leftover, keyspaceID, err
}

func FormatTableRuleID(tableID uint64) string {
	return fmt.Sprintf("table-%d", tableID)
}

func ParseTableRuleID(name string) (leftover string, tableID uint64, err error) {
	if !strings.HasPrefix(name, "table-") {
		return "", 0, fmt.Errorf("invalid format [%s]", name)
	}
	name = strings.TrimPrefix(name, "table-")
	idx := strings.Index(name, "-")
	var idStr = name
	if idx != -1 {
		idStr = name[0:idx]
		leftover = name[idx+1:]
	}
	tableID, err = strconv.ParseUint(idStr, 10, 64)
	if err != nil {
		return "", 0, err
	}
	return leftover, tableID, nil
}

func FormatRegionRuleID(regionID uint64) string {
	return fmt.Sprintf("region-%d", regionID)
}

func ParseRegionRuleID(name string) (leftover string, regionID uint64, err error) {
	if !strings.HasPrefix(name, "region-") {
		return "", 0, fmt.Errorf("invalid format [%s]", name)
	}
	name = strings.TrimPrefix(name, "region-")
	idx := strings.Index(name, "-")
	var idStr = name
	if idx != -1 {
		idStr = name[0:idx]
		leftover = name[idx+1:]
	}
	regionID, err = strconv.ParseUint(idStr, 10, 64)
	if err != nil {
		return "", 0, err
	}
	return leftover, regionID, nil
}

func FormatKeyspaceTableRuleID(keyspaceID uint32, tableID uint64) string {
	return fmt.Sprintf("%s-%s", FormatKeyspaceRuleID(keyspaceID), FormatTableRuleID(tableID))
}

func ParseKeyspaceTableRuleID(name string) (leftover string, keyspaceID uint32, tableID uint64, err error) {
	leftover, keyspaceID, err = ParseKeyspaceRuleID(name)
	if err != nil {
		return "", 0, 0, err
	}
	leftover, tableID, err = ParseTableRuleID(leftover)
	if err != nil {
		return "", 0, 0, err
	}
	return leftover, keyspaceID, tableID, nil
}

func FormatKeyspaceZoneRuleID(keyspaceID uint32, zoneType string) string {
	if len(zoneType) > 0 {
		zoneType = "-" + zoneType
	}
	return fmt.Sprintf("%s%s", FormatKeyspaceRuleID(keyspaceID), zoneType)
}

func ParseKeyspaceZoneRuleID(name string) (keyspaceID uint32, zoneType string, err error) {
	zoneType, keyspaceID, err = ParseKeyspaceRuleID(name)
	return
}

func IsMultiAZKeyspaceRuleID(name string) bool {
	return strings.Contains(name, "multi-az-") && strings.Contains(name, "keyspace-") && !strings.Contains(name, "table-")
}

func FormatMultiAZKeyspaceRuleID(keyspaceID uint32, zoneType string) string {
	if len(zoneType) > 0 {
		zoneType = "-" + zoneType
	}
	return fmt.Sprintf("multi-az-%s%s", FormatKeyspaceRuleID(keyspaceID), zoneType)
}

func ParseMultiAZKeyspaceRuleID(name string) (keyspaceID uint32, zoneType string, err error) {
	if !strings.HasPrefix(name, "multi-az-") {
		return 0, "", fmt.Errorf("invalid format [%s]", name)
	}
	name = strings.TrimPrefix(name, "multi-az-")
	zoneType, keyspaceID, err = ParseKeyspaceRuleID(name)
	return
}

func IsSingleAZKeyspaceRuleID(name string) bool {
	return strings.Contains(name, "single-az-") && strings.Contains(name, "keyspace-") && !strings.Contains(name, "table-")
}

func FormatSingleAZKeyspaceRuleID(keyspaceID uint32, roleType string) string {
	if len(roleType) > 0 {
		roleType = "-" + roleType
	}
	return fmt.Sprintf("single-az-%s%s", FormatKeyspaceRuleID(keyspaceID), roleType)
}

func ParseSingleAZKeyspaceRuleID(name string) (keyspaceID uint32, roleType string, err error) {
	if !strings.HasPrefix(name, "single-az-") {
		return 0, "", fmt.Errorf("invalid format [%s]", name)
	}
	name = strings.TrimPrefix(name, "single-az-")
	roleType, keyspaceID, err = ParseKeyspaceRuleID(name)
	return
}

func IsMultiAZKeyspaceTableRuleID(name string) bool {
	return strings.Contains(name, "multi-az-") && strings.Contains(name, "keyspace-") && strings.Contains(name, "table-")
}

func FormatMultiAZKeyspaceTableRuleID(keyspaceID uint32, tableID uint64, zoneType string) string {
	if len(zoneType) > 0 {
		zoneType = "-" + zoneType
	}
	return fmt.Sprintf("multi-az-%s%s", FormatKeyspaceTableRuleID(keyspaceID, tableID), zoneType)
}

func ParseMultiAZKeyspaceTableRuleID(name string) (keyspaceID uint32, tableID uint64, zoneType string, err error) {
	if !strings.HasPrefix(name, "multi-az-") {
		return 0, 0, "", fmt.Errorf("invalid format [%s]", name)
	}
	name = strings.TrimPrefix(name, "multi-az-")
	zoneType, keyspaceID, tableID, err = ParseKeyspaceTableRuleID(name)
	return
}

func IsSingleAZKeyspaceTableRuleID(name string) bool {
	return strings.Contains(name, "single-az-") && strings.Contains(name, "keyspace-") && strings.Contains(name, "table-")
}

func FormatSingleAZKeyspaceTableRuleID(keyspaceID uint32, tableID uint64, roleType string) string {
	if len(roleType) > 0 {
		roleType = "-" + roleType
	}
	return fmt.Sprintf("single-az-%s%s", FormatKeyspaceTableRuleID(keyspaceID, tableID), roleType)
}

func ParseSingleAZKeyspaceTableRuleID(name string) (keyspaceID uint32, tableID uint64, roleType string, err error) {
	if !strings.HasPrefix(name, "single-az-") {
		return 0, 0, "", fmt.Errorf("invalid format [%s]", name)
	}
	name = strings.TrimPrefix(name, "single-az-")
	roleType, keyspaceID, tableID, err = ParseKeyspaceTableRuleID(name)
	return
}

func IsMultiAZRegionRuleID(name string) bool {
	return strings.Contains(name, "multi-az-") && strings.Contains(name, "region-")
}

func FormatMultiAZRegionRuleID(regionID uint64, zoneType string) string {
	if len(zoneType) > 0 {
		zoneType = "-" + zoneType
	}
	return fmt.Sprintf("multi-az-%s%s", FormatRegionRuleID(regionID), zoneType)
}

func ParseMultiAZRegionRuleID(name string) (regionID uint64, zoneType string, err error) {
	if !strings.HasPrefix(name, "multi-az-") {
		return 0, "", fmt.Errorf("invalid format [%s]", name)
	}
	name = strings.TrimPrefix(name, "multi-az-")
	zoneType, regionID, err = ParseRegionRuleID(name)
	return
}

func IsSingleAZRegionRuleID(name string) bool {
	return strings.Contains(name, "single-az-") && strings.Contains(name, "region-")
}

func FormatSingleAZRegionRuleID(regionID uint64, roleType string) string {
	if len(roleType) > 0 {
		roleType = "-" + roleType
	}
	return fmt.Sprintf("single-az-%s%s", FormatRegionRuleID(regionID), roleType)
}

func ParseSingleAZRegionRuleID(name string) (regionID uint64, roleType string, err error) {
	if !strings.HasPrefix(name, "single-az-") {
		return 0, "", fmt.Errorf("invalid format [%s]", name)
	}
	name = strings.TrimPrefix(name, "single-az-")
	roleType, regionID, err = ParseRegionRuleID(name)
	return
}

func FormatRangeKey(start, end string) string {
	return fmt.Sprintf("%s-%s", start, end)
}

func ParseRangeKey(key string) (start, end string, err error) {
	idx := strings.Index(key, "-")
	if idx == -1 {
		return "", "", fmt.Errorf("invalid format [%s]", key)
	}
	start = key[0:idx]
	end = key[idx+1:]
	return
}

func IsMultiAZRuleID(name string) bool {
	return strings.HasPrefix(name, "multi-az-") || strings.Contains(name, "-multi-az-")
}

func IsSingleAZRuleID(name string) bool {
	return strings.HasPrefix(name, "single-az-") || strings.Contains(name, "-single-az-")
}

func IsSingleAZOnlyLeaderRuleID(name string) bool {
	return strings.HasSuffix(name, "-leader") || strings.HasSuffix(name, "-follower")
}

func IsLeaderRuleID(name string) bool {
	return strings.HasSuffix(name, "-leader")
}

func IsPrimaryRuleID(name string) bool {
	return strings.HasSuffix(name, "-"+PrimaryZone)
}

type MultiAZRuleBatch struct {
	PrimaryRuleID string
	StandbyRuleID string
	WitnessRuleID string
	StartKey      string
	EndKey        string
}

type SingleAZRuleBatch struct {
	LeaderRuleID   string
	FollowerRuleID string
	VoterRuleID    string
	StartKey       string
	EndKey         string
}

func GetMultiAZKeyspaceRuleBatch(keyspaceID uint32) *MultiAZRuleBatch {
	primaryRuleID := FormatMultiAZKeyspaceRuleID(keyspaceID, PrimaryZone)
	standbyRuleID := FormatMultiAZKeyspaceRuleID(keyspaceID, StandbyZone)
	witnessRuleID := FormatMultiAZKeyspaceRuleID(keyspaceID, WitnessZone)
	startKey, endKey := KeyspaceTxnRange(keyspaceID)
	return &MultiAZRuleBatch{
		PrimaryRuleID: primaryRuleID,
		StandbyRuleID: standbyRuleID,
		WitnessRuleID: witnessRuleID,
		StartKey:      startKey,
		EndKey:        endKey,
	}
}

func GetSingleAZKeyspaceRuleBatch(keyspaceID uint32) *SingleAZRuleBatch {
	leaderRuleID := FormatSingleAZKeyspaceRuleID(keyspaceID, "leader")
	followerRuleID := FormatSingleAZKeyspaceRuleID(keyspaceID, "follower")
	voterRuleID := FormatSingleAZKeyspaceRuleID(keyspaceID, "")
	startKey, endKey := KeyspaceTxnRange(keyspaceID)
	return &SingleAZRuleBatch{
		LeaderRuleID:   leaderRuleID,
		FollowerRuleID: followerRuleID,
		VoterRuleID:    voterRuleID,
		StartKey:       startKey,
		EndKey:         endKey,
	}
}

func GetMultiAZKeyspaceTableRuleBatch(keyspaceID uint32, tableID uint64) *MultiAZRuleBatch {
	startKey, endKey := KeyspaceTableTxnRange(keyspaceID, tableID)
	return &MultiAZRuleBatch{
		PrimaryRuleID: FormatMultiAZKeyspaceTableRuleID(keyspaceID, tableID, PrimaryZone),
		StandbyRuleID: FormatMultiAZKeyspaceTableRuleID(keyspaceID, tableID, StandbyZone),
		WitnessRuleID: FormatMultiAZKeyspaceTableRuleID(keyspaceID, tableID, WitnessZone),
		StartKey:      startKey,
		EndKey:        endKey,
	}
}

func GetSingleAZKeyspaceTableRuleBatch(keyspaceID uint32, tableID uint64) *SingleAZRuleBatch {
	startKey, endKey := KeyspaceTableTxnRange(keyspaceID, tableID)
	return &SingleAZRuleBatch{
		LeaderRuleID:   FormatSingleAZKeyspaceTableRuleID(keyspaceID, tableID, "leader"),
		FollowerRuleID: FormatSingleAZKeyspaceTableRuleID(keyspaceID, tableID, "follower"),
		VoterRuleID:    FormatSingleAZKeyspaceTableRuleID(keyspaceID, tableID, ""),
		StartKey:       startKey,
		EndKey:         endKey,
	}
}

func GetMultiAZRegionRuleBatch(regionID uint64, startKey, endKey string) *MultiAZRuleBatch {
	return &MultiAZRuleBatch{
		PrimaryRuleID: FormatMultiAZRegionRuleID(regionID, PrimaryZone),
		StandbyRuleID: FormatMultiAZRegionRuleID(regionID, StandbyZone),
		WitnessRuleID: FormatMultiAZRegionRuleID(regionID, WitnessZone),
		StartKey:      startKey,
		EndKey:        endKey,
	}
}

func GetSingleAZRegionRuleBatch(regionID uint64, startKey, endKey string) *SingleAZRuleBatch {
	return &SingleAZRuleBatch{
		LeaderRuleID:   FormatSingleAZRegionRuleID(regionID, "leader"),
		FollowerRuleID: FormatSingleAZRegionRuleID(regionID, "follower"),
		VoterRuleID:    FormatSingleAZRegionRuleID(regionID, ""),
		StartKey:       startKey,
		EndKey:         endKey,
	}
}

const (
	encGroupSize = 8
	encMarker    = byte(0xFF)
)

var pads = make([]byte, encGroupSize)

// EncodeBytes guarantees the encoded value is in ascending order for comparison,
// encoding with the following rule:
//
//	[group1][marker1]...[groupN][markerN]
//	group is 8 bytes slice which is padding with 0.
//	marker is `0xFF - padding 0 count`
//
// For example:
//
//	[] -> [0, 0, 0, 0, 0, 0, 0, 0, 247]
//	[1, 2, 3] -> [1, 2, 3, 0, 0, 0, 0, 0, 250]
//	[1, 2, 3, 0] -> [1, 2, 3, 0, 0, 0, 0, 0, 251]
//	[1, 2, 3, 4, 5, 6, 7, 8] -> [1, 2, 3, 4, 5, 6, 7, 8, 255, 0, 0, 0, 0, 0, 0, 0, 0, 247]
//
// Refer: https://github.com/facebook/mysql-5.6/wiki/MyRocks-record-format#memcomparable-format
func EncodeBytes(data []byte) []byte {
	// Allocate more space to avoid unnecessary slice growing.
	// Assume that the byte slice size is about `(len(data) / encGroupSize + 1) * (encGroupSize + 1)` bytes,
	// that is `(len(data) / 8 + 1) * 9` in our implement.
	dLen := len(data)
	result := make([]byte, 0, (dLen/encGroupSize+1)*(encGroupSize+1))
	for idx := 0; idx <= dLen; idx += encGroupSize {
		remain := dLen - idx
		padCount := 0
		if remain >= encGroupSize {
			result = append(result, data[idx:idx+encGroupSize]...)
		} else {
			padCount = encGroupSize - remain
			result = append(result, data[idx:]...)
			result = append(result, pads[:padCount]...)
		}

		marker := encMarker - byte(padCount)
		result = append(result, marker)
	}
	return result
}

const signMask uint64 = 0x8000000000000000

// EncodeIntToCmpUint make int v to comparable uint type
func EncodeIntToCmpUint(v int64) uint64 {
	return uint64(v) ^ signMask
}

// DecodeCmpUintToInt decodes the u that encoded by EncodeIntToCmpUint
func DecodeCmpUintToInt(u uint64) int64 {
	return int64(u ^ signMask)
}

// EncodeInt appends the encoded value to slice b and returns the appended slice.
// EncodeInt guarantees that the encoded value is in ascending order for comparison.
func EncodeInt(b []byte, v int64) []byte {
	var data [8]byte
	u := EncodeIntToCmpUint(v)
	binary.BigEndian.PutUint64(data[:], u)
	return append(b, data[:]...)
}

// DecodeInt decodes value encoded by EncodeInt before.
// It returns the leftover un-decoded slice, decoded value if no error.
func DecodeInt(b []byte) ([]byte, int64, error) {
	if len(b) < 8 {
		return nil, 0, errors.New("insufficient bytes to decode value")
	}

	u := binary.BigEndian.Uint64(b[:8])
	v := DecodeCmpUintToInt(u)
	b = b[8:]
	return b, v, nil
}