package main

import (
	"encoding/binary"
	"fmt"
	"hash/fnv"
	"sort"
	"strings"
	"sync"
)

// ring.go
//
// Consistent hashing ring skeleton for future integration.
// --------------------------------------------------------
// This file introduces a minimal, allocation‑light consistent hashing structure
// intended to replace per-cart ownership negotiation. It focuses on:
//   * Deterministic lookup: O(log V) via binary search
//   * Even(ish) distribution using virtual nodes (vnodes)
//   * Epoch / fingerprint tracking to detect membership drift
//
// NOT YET WIRED:
//   * SyncedPool integration (ownerForCart, lazy migration)
//   * Replication factor > 1
//   * Persistent state migration
//
// Safe to import now; unused until explicit integration code is added.
//
// Design Notes
// ------------
// - Hosts contribute `vnodesPerHost` virtual nodes. Higher counts smooth
//   distribution at cost of memory (V = hosts * vnodesPerHost).
// - Hash of vnode = FNV1a64(host + "#" + index). For improved quality you
//   can swap in xxhash or siphash later without changing API (but doing so
//   will reshuffle ownership).
// - Cart ownership lookup uses either cartID.Raw() when provided (uniform
//   64-bit space) or falls back to hashing string forms (legacy).
// - Epoch is monotonically increasing; consumers can fence stale results.
//
// Future Extensions
// -----------------
// - Weighted hosts (proportionally more vnodes).
// - Replication: LookupN(h, n) to return primary + replicas.
// - Streaming / diff-based ring updates (gossip).
// - Hash function injection for deterministic test scenarios.
//
// ---------------------------------------------------------------------------

// Vnode represents a single virtual node position on the ring.
type Vnode struct {
	Hash  uint64 // position on the ring
	Host  string // physical host owning this vnode
	Index int    // per-host vnode index (0..vnodesPerHost-1)
}

// Ring is an immutable consistent hash ring snapshot.
type Ring struct {
	Epoch       uint64
	Vnodes      []Vnode // sorted by Hash
	hosts       []string
	fingerprint uint64 // membership fingerprint (order-independent)
}

// RingBuilder accumulates parameters to construct a Ring.
type RingBuilder struct {
	epoch         uint64
	vnodesPerHost int
	hosts         []string
}

// NewRingBuilder creates a builder with defaults.
func NewRingBuilder() *RingBuilder {
	return &RingBuilder{
		vnodesPerHost: 64, // a reasonable default for small clusters
	}
}

func (b *RingBuilder) WithEpoch(e uint64) *RingBuilder {
	b.epoch = e
	return b
}

func (b *RingBuilder) WithVnodesPerHost(n int) *RingBuilder {
	if n > 0 {
		b.vnodesPerHost = n
	}
	return b
}

func (b *RingBuilder) WithHosts(hosts []string) *RingBuilder {
	uniq := make(map[string]struct{}, len(hosts))
	out := make([]string, 0, len(hosts))
	for _, h := range hosts {
		h = strings.TrimSpace(h)
		if h == "" {
			continue
		}
		if _, ok := uniq[h]; ok {
			continue
		}
		uniq[h] = struct{}{}
		out = append(out, h)
	}
	sort.Strings(out)
	b.hosts = out
	return b
}

func (b *RingBuilder) Build() *Ring {
	if len(b.hosts) == 0 {
		return &Ring{
			Epoch:       b.epoch,
			Vnodes:      nil,
			hosts:       nil,
			fingerprint: 0,
		}
	}

	totalVnodes := len(b.hosts) * b.vnodesPerHost
	vnodes := make([]Vnode, 0, totalVnodes)

	for _, host := range b.hosts {
		for i := 0; i < b.vnodesPerHost; i++ {
			h := hashVnode(host, i)
			vnodes = append(vnodes, Vnode{
				Hash:  h,
				Host:  host,
				Index: i,
			})
		}
	}
	sort.Slice(vnodes, func(i, j int) bool {
		if vnodes[i].Hash == vnodes[j].Hash {
			// Tie-break deterministically by host then index to avoid instability
			if vnodes[i].Host == vnodes[j].Host {
				return vnodes[i].Index < vnodes[j].Index
			}
			return vnodes[i].Host < vnodes[j].Host
		}
		return vnodes[i].Hash < vnodes[j].Hash
	})

	fp := fingerprintHosts(b.hosts)

	return &Ring{
		Epoch:       b.epoch,
		Vnodes:      vnodes,
		hosts:       append([]string(nil), b.hosts...),
		fingerprint: fp,
	}
}

// Hosts returns a copy of the host list (sorted).
func (r *Ring) Hosts() []string {
	if len(r.hosts) == 0 {
		return nil
	}
	cp := make([]string, len(r.hosts))
	copy(cp, r.hosts)
	return cp
}

// Fingerprint returns a hash representing the unordered membership set.
func (r *Ring) Fingerprint() uint64 {
	return r.fingerprint
}

// Empty indicates ring has no vnodes.
func (r *Ring) Empty() bool {
	return len(r.Vnodes) == 0
}

// Lookup returns the vnode owning a given hash value.
func (r *Ring) Lookup(h uint64) Vnode {
	if len(r.Vnodes) == 0 {
		return Vnode{}
	}
	// Binary search: first position with Hash >= h
	i := sort.Search(len(r.Vnodes), func(i int) bool {
		return r.Vnodes[i].Hash >= h
	})
	if i == len(r.Vnodes) {
		return r.Vnodes[0]
	}
	return r.Vnodes[i]
}

// LookupID selects owner vnode for a CartID (fast path).
func (r *Ring) LookupID(id CartID) Vnode {
	return r.Lookup(id.Raw())
}

// LookupString hashes an arbitrary string and looks up owner.
func (r *Ring) LookupString(s string) Vnode {
	return r.Lookup(hashKeyString(s))
}

// LookupN returns up to n distinct host vnodes in ring order
// starting from the primary owner of hash h (for replication).
func (r *Ring) LookupN(h uint64, n int) []Vnode {
	if n <= 0 || len(r.Vnodes) == 0 {
		return nil
	}
	if n > len(r.hosts) {
		n = len(r.hosts)
	}
	owners := make([]Vnode, 0, n)
	seen := make(map[string]struct{}, n)

	start := r.Lookup(h)

	// Find index of start (can binary search again or linear scan; since we
	// already have start.Hash we do another search for clarity)
	i := sort.Search(len(r.Vnodes), func(i int) bool {
		return r.Vnodes[i].Hash >= start.Hash
	})
	if i == len(r.Vnodes) {
		i = 0
	}

	for idx := 0; len(owners) < n && idx < len(r.Vnodes); idx++ {
		v := r.Vnodes[(i+idx)%len(r.Vnodes)]
		if _, ok := seen[v.Host]; ok {
			continue
		}
		seen[v.Host] = struct{}{}
		owners = append(owners, v)
	}
	return owners
}

// DiffHosts compares this ring's membership to another.
func (r *Ring) DiffHosts(other *Ring) (added []string, removed []string) {
	if other == nil {
		return r.Hosts(), nil
	}
	cur := make(map[string]struct{}, len(r.hosts))
	for _, h := range r.hosts {
		cur[h] = struct{}{}
	}
	oth := make(map[string]struct{}, len(other.hosts))
	for _, h := range other.hosts {
		oth[h] = struct{}{}
	}
	for h := range cur {
		if _, ok := oth[h]; !ok {
			removed = append(removed, h)
		}
	}
	for h := range oth {
		if _, ok := cur[h]; !ok {
			added = append(added, h)
		}
	}
	sort.Strings(added)
	sort.Strings(removed)
	return
}

// ---------------------------- Hash Functions ---------------------------------

func hashVnode(host string, idx int) uint64 {
	h := fnv.New64a()
	_, _ = h.Write([]byte(host))
	_, _ = h.Write([]byte{'#'})
	var buf [8]byte
	binary.BigEndian.PutUint64(buf[:], uint64(idx))
	_, _ = h.Write(buf[:])
	return h.Sum64()
}

// hashKeyString provides a stable hash for arbitrary string keys (legacy IDs).
func hashKeyString(s string) uint64 {
	h := fnv.New64a()
	_, _ = h.Write([]byte(s))
	return h.Sum64()
}

// fingerprintHosts produces an order-insensitive hash over the host set.
func fingerprintHosts(hosts []string) uint64 {
	if len(hosts) == 0 {
		return 0
	}
	h := fnv.New64a()
	for _, host := range hosts {
		_, _ = h.Write([]byte(host))
		_, _ = h.Write([]byte{0})
	}
	return h.Sum64()
}

// --------------------------- Thread-Safe Wrapper -----------------------------
//
// RingRef offers atomic swap + read semantics. SyncedPool can embed or hold
// one of these to manage live ring updates safely.

type RingRef struct {
	mu   sync.RWMutex
	ring *Ring
}

func NewRingRef(r *Ring) *RingRef {
	return &RingRef{ring: r}
}

func (rr *RingRef) Get() *Ring {
	rr.mu.RLock()
	r := rr.ring
	rr.mu.RUnlock()
	return r
}

func (rr *RingRef) Set(r *Ring) {
	rr.mu.Lock()
	rr.ring = r
	rr.mu.Unlock()
}

func (rr *RingRef) LookupID(id CartID) Vnode {
	r := rr.Get()
	if r == nil {
		return Vnode{}
	}
	return r.LookupID(id)
}

// ----------------------------- Debug Utilities -------------------------------

func (r *Ring) String() string {
	var b strings.Builder
	fmt.Fprintf(&b, "Ring{epoch=%d vnodes=%d hosts=%d}\n", r.Epoch, len(r.Vnodes), len(r.hosts))
	limit := len(r.Vnodes)
	if limit > 16 {
		limit = 16
	}
	for i := 0; i < limit; i++ {
		v := r.Vnodes[i]
		fmt.Fprintf(&b, "  %02d hash=%016x host=%s idx=%d\n", i, v.Hash, v.Host, v.Index)
	}
	if len(r.Vnodes) > limit {
		fmt.Fprintf(&b, "  ... (%d more)\n", len(r.Vnodes)-limit)
	}
	return b.String()
}