// Copyright 2017 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 main

import (
	"fmt"

	"github.com/juju/errors"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/pingcap/schrodinger/pkg/test_client"
)

// IsSameLocation return whehter two peers locate in same stores.
func IsSameLocation(one, other []*metapb.Peer) bool {
	set := make(map[uint64]struct{})

	for _, item := range one {
		set[item.StoreId] = struct{}{}
	}

	for _, item := range other {
		delete(set, item.StoreId)
	}

	return len(set) == 0 && len(one) == len(other)
}

// FindTableRegions get regions with specified table id.
func FindTableRegions(tableID int64) ([]*RegionInfo, error) {
	regions, err := GetRegions()
	if err != nil {
		return nil, errors.Trace(err)
	}

	var find []*RegionInfo
	for _, region := range regions {
		if GetTableID(region) == tableID {
			find = append(find, region)
		}
	}

	if len(find) == 0 {
		return nil, fmt.Errorf("no new region after creating table")
	}
	client.Infof("table %d has %d regions", tableID, len(find))
	return find, nil
}

// GetIndexRegions filters region with index item.
func GetIndexRegions(origin []*RegionInfo) []*RegionInfo {
	indexRegions := make([]*RegionInfo, 0, len(origin))
	for _, region := range origin {
		if IsIndex(region) {
			indexRegions = append(indexRegions, region)
		}
	}
	return indexRegions
}

// GetNewTableRegions get new table regions compared to old table regions.
func GetNewTableRegions(old, now []*RegionInfo) ([]*RegionInfo, error) {
	set := make(map[int64]*RegionInfo)
	for _, n := range now {
		set[GetTableID(n)] = n
	}
	for _, o := range old {
		delete(set, GetTableID(o))
	}

	var regions []*RegionInfo
	for _, s := range set {
		region, err := GetRegionInfo(s.ID)
		if err != nil {
			return nil, errors.Trace(err)
		}
		regions = append(regions, region)
	}
	return regions, nil
}

// IsRegionIDExist returns whether specified region in regions.
func IsRegionIDExist(regions []*RegionInfo, id uint64) bool {
	for _, region := range regions {
		if region.ID == id {
			return true
		}
	}
	return false
}

// CheckRegionsTableID check tableRegions has other tableIds excepet specified table id.
func CheckRegionsTableID(tableID int64, tableRegions []*RegionInfo) ([]*RegionInfo, error) {
	regions, err := GetRegions()
	if err != nil {
		return tableRegions, errors.Trace(err)
	}

	set := make(map[uint64]struct{})
	for _, region := range tableRegions {
		set[region.ID] = struct{}{}
	}

	for _, region := range regions {
		if _, ok := set[region.ID]; ok {
			continue
		}
		if GetTableID(region) == tableID {
			tableRegions = append(tableRegions, region)
			client.Infof("insert new region %d for table %v", region.ID, tableID)
		}
	}

	return tableRegions, nil
}

// GetTableID returns table id of the region.
func GetTableID(region *RegionInfo) int64 {
	key := Key{}
	fmt.Sscanf("\""+region.StartKey+"\"", "%q", &key)
	return key.TableID()
}

// IsIndex returns whether the region store index info.
func IsIndex(region *RegionInfo) bool {
	key := Key{}
	fmt.Sscanf("\""+region.StartKey+"\"", "%q", &key)
	return key.IsIndex()
}

// FindAdjacentRegion returns adjacenet region infos of specified region.
func FindAdjacentRegion(regions []*RegionInfo, region *RegionInfo) *RegionInfo {
	for _, r := range regions {
		if r.EndKey == region.StartKey || region.EndKey == r.StartKey {
			return r
		}
	}

	return nil
}