// Copyright 2020 newtbig Author. All Rights Reserved.
//
// 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 hash

import (
	"errors"
	"hash/crc32"
	"strconv"
	"sync"

	"gitee.com/lakertt/newtbig/dispatcher/hash/maps"
)

var ErrEmptyRing = errors.New("empty ring")

type (
	HashRing struct {
		m_RingMap    map[uint32]string
		m_MemberMap  map[string]bool
		m_SortedKeys *maps.Map
		rRep         uint32
		rLen         uint32
		sync.RWMutex
	}

	IHashRing interface {
		Add(elt string) []uint32
		Remove(elt string)
		HasMember(elt string) bool
		Members() []string
		Get(name string) (error, string)
		Get64(val int64) (error, uint32)
	}
)

func NewHashRing(rep uint32, rLen uint32) *HashRing {
	pRing := new(HashRing)
	pRing.m_RingMap = make(map[uint32]string)
	pRing.m_MemberMap = make(map[string]bool)
	pRing.m_SortedKeys = maps.NewWithUInt32Comparator()
	pRing.rLen = rLen
	pRing.rRep = rep
	return pRing
}
func (this *HashRing) addInt(id uint32, elt string) {
	this.m_RingMap[id] = elt
	this.m_SortedKeys.Put(id, true)
	this.m_MemberMap[elt] = true
	return
}

func (this *HashRing) removeInt(id uint32) {
	elt := this.m_RingMap[id]
	delete(this.m_RingMap, id)
	this.m_SortedKeys.Remove(id)
	delete(this.m_MemberMap, elt)
}

func (this *HashRing) eltKey(elt string, idx int) string {
	return strconv.Itoa(idx) + elt
}

func (this *HashRing) add(elt string) []uint32 {
	rst := make([]uint32, 0)
	for i := 0; i < int(this.rRep); i++ {
		Id := this.hashKey(this.eltKey(elt, i))
		this.m_RingMap[Id] = elt
		this.m_SortedKeys.Put(Id, true)
		rst = append(rst, Id)
	}
	this.m_MemberMap[elt] = true
	return rst
}

func (this *HashRing) remove(elt string) {
	for i := 0; i < int(this.rRep); i++ {
		Id := this.hashKey(this.eltKey(elt, i))
		delete(this.m_RingMap, Id)
		this.m_SortedKeys.Remove(Id)
	}
	delete(this.m_MemberMap, elt)
}

func (this *HashRing) hashKey(key string) uint32 {
	if len(key) < 64 {
		var scratch [64]byte
		copy(scratch[:], key)
		return crc32.ChecksumIEEE(scratch[:len(key)])
	}
	return crc32.ChecksumIEEE([]byte(key))
}

func (this *HashRing) Add(elt string) []uint32 {
	this.Lock()
	defer this.Unlock()
	return this.add(elt)
}

func (this *HashRing) Remove(elt string) {
	this.Lock()
	defer this.Unlock()
	this.remove(elt)
}

func (this *HashRing) HasMember(elt string) bool {
	this.RLock()
	defer this.RUnlock()
	_, bEx := this.m_MemberMap[elt]
	return bEx
}

func (this *HashRing) Members() []string {
	this.RLock()
	defer this.RUnlock()
	var m []string
	for k := range this.m_MemberMap {
		m = append(m, k)
	}
	return m
}

func (this *HashRing) Get(name string) (error, string) {
	this.RLock()
	defer this.RUnlock()
	if len(this.m_RingMap) == 0 {
		return ErrEmptyRing, ""
	}
	key := this.hashKey(name)
	node, bOk := this.m_SortedKeys.Ceiling(key)
	if !bOk {
		itr := this.m_SortedKeys.Iterator()
		if itr.First() {
			return nil, this.m_RingMap[itr.Key().(uint32)]
		}
		return ErrEmptyRing, ""
	}
	return nil, this.m_RingMap[node.Key.(uint32)]
}

func (this *HashRing) Get64(val int64) (error, uint32) {
	this.RLock()
	defer this.RUnlock()
	if len(this.m_RingMap) == 0 {
		return ErrEmptyRing, 0
	}
	key := this.hashKey(strconv.FormatInt(val, 10))
	node, bOk := this.m_SortedKeys.Ceiling(key)
	if !bOk {
		itr := this.m_SortedKeys.Iterator()
		if itr.First() {
			return nil, this.hashKey(this.m_RingMap[itr.Key().(uint32)])
		}
		return ErrEmptyRing, 0
	}
	return nil, this.hashKey(this.m_RingMap[node.Key.(uint32)])
}