// Package partition implements the data-distribution primitives an
// architect reasons about: consistent hashing with virtual nodes
// (minimal key movement under membership change) vs naive mod-N hashing,
// and a quorum read/write model (the R+W>N overlap guarantee). Stdlib-only.
package partition

import "sort"

// Ring is a consistent-hashing ring with virtual nodes. Each physical
// node is placed at `vnodes` positions on the hash circle, so load is
// spread evenly and adding/removing a node only moves the keys near that
// node's positions — ~1/N of keys, not all of them.
type Ring struct {
	vnodes int
	pos    []uint32          // sorted vnode positions on the circle
	owner  map[uint32]string // position -> physical node
	nodes  map[string]bool
}

func NewRing(vnodes int) *Ring {
	if vnodes < 1 {
		vnodes = 1
	}
	return &Ring{vnodes: vnodes, owner: map[uint32]string{}, nodes: map[string]bool{}}
}

// AddNode places a node at `vnodes` positions and re-sorts the circle.
func (r *Ring) AddNode(node string) {
	if r.nodes[node] {
		return
	}
	r.nodes[node] = true
	for i := 0; i < r.vnodes; i++ {
		p := fnv1a32(node + "#" + itoa(i))
		r.pos = append(r.pos, p)
		r.owner[p] = node
	}
	sort.Slice(r.pos, func(a, b int) bool { return r.pos[a] < r.pos[b] })
}

// RemoveNode removes a node and its vnodes.
func (r *Ring) RemoveNode(node string) {
	if !r.nodes[node] {
		return
	}
	delete(r.nodes, node)
	kept := r.pos[:0]
	for _, p := range r.pos {
		if r.owner[p] == node {
			delete(r.owner, p)
			continue
		}
		kept = append(kept, p)
	}
	r.pos = append([]uint32(nil), kept...)
}

// Get returns the node that owns a key: the first vnode position
// clockwise from the key's hash (wrapping around the circle).
func (r *Ring) Get(key string) string {
	if len(r.pos) == 0 {
		return ""
	}
	h := fnv1a32(key)
	i := sort.Search(len(r.pos), func(i int) bool { return r.pos[i] >= h })
	if i == len(r.pos) {
		i = 0 // wrap
	}
	return r.owner[r.pos[i]]
}

// Nodes returns the physical node count.
func (r *Ring) NodeCount() int { return len(r.nodes) }

// ModHashGet is the NAIVE alternative: node = sortedNodes[hash(key) %
// len]. Simple, but when `len` changes (a node joins/leaves) the modulus
// changes and almost EVERY key remaps — a catastrophic reshuffle.
func ModHashGet(nodes []string, key string) string {
	if len(nodes) == 0 {
		return ""
	}
	s := append([]string(nil), nodes...)
	sort.Strings(s)
	return s[fnv1a32(key)%uint32(len(s))]
}

func fnv1a32(s string) uint32 {
	var h uint32 = 2166136261
	for i := 0; i < len(s); i++ {
		h ^= uint32(s[i])
		h *= 16777619
	}
	return h
}

func itoa(i int) string {
	if i == 0 {
		return "0"
	}
	var b [12]byte
	p := len(b)
	for i > 0 {
		p--
		b[p] = byte('0' + i%10)
		i /= 10
	}
	return string(b[p:])
}