package systemtables

import (
	"context"
	"fmt"
	"math/big"
	"net"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"

	"github.com/everything-on-mysql/cassandra-on-mysql/internal/config"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

const (
	defaultNativePort  = 9042
	defaultStoragePort = 7000
)

type capabilityProfile struct {
	name                  string
	nativeProtocolVersion string
	releaseVersion        string
}

var capabilityProfiles = map[string]capabilityProfile{
	"compat_v4": {name: "compat_v4", nativeProtocolVersion: "4", releaseVersion: "4.0"},
	"compat_v5": {name: "compat_v5", nativeProtocolVersion: "5", releaseVersion: "4.0"},
}

// SystemTableState captures computed values used by system table providers.
type SystemTableState struct {
	clusterName           string
	datacenter            string
	rack                  string
	partitioner           string
	cqlVersion            string
	releaseVersion        string
	nativeProtocolVersion string

	advertisedAddress string
	listenIP          string
	listenPort        int
	nativePort        int
	storagePort       int

	peers []peerState

	tokensConfigured bool
	localTokens      []string
	peerTokens       [][]string

	autoTokensCached bool
	autoLocalTokens  []string
	autoPeerTokens   [][]string

	localIPFixed    bool
	localHostFixed  bool
	localPortFixed  bool
	localFixedIP    string
	localFixedPort  int
	localFixedHost  string
	localFixedIPHex string

	schemaProvider SchemaVersionProvider
	schemaVersion  atomic.Value
	schemaMu       sync.Mutex
}

type peerState struct {
	address     string
	ip16hex     string
	nativePort  int
	storagePort int
	hostID      string
	tokens      []string
}

type nodeIdentity struct {
	ip          string
	ip16hex     string
	nativePort  int
	storagePort int
	hostID      string
	tokens      []string
}

type nodeOrder struct {
	ip16hex string
	port    int
}

// NewSystemTableState builds a state snapshot from config and schema metadata.
func NewSystemTableState(cfg *config.Config, schemaManager *mysqlbackend.SchemaManager) (*SystemTableState, error) {
	if cfg == nil {
		cfg = config.DefaultConfig()
	}
	sysCfg := cfg.SystemTables

	profile, ok := capabilityProfiles[sysCfg.CapabilityProfile]
	if !ok {
		return nil, fmt.Errorf("unknown system_tables capability_profile %q", sysCfg.CapabilityProfile)
	}

	nativeProtocolVersion := sysCfg.NativeProtocolVersion
	if nativeProtocolVersion == "" {
		nativeProtocolVersion = profile.nativeProtocolVersion
	}
	if nativeProtocolVersion != profile.nativeProtocolVersion {
		return nil, fmt.Errorf("native_protocol_version %q incompatible with capability_profile %q", nativeProtocolVersion, profile.name)
	}
	if err := validateProtocolVersion(nativeProtocolVersion); err != nil {
		return nil, err
	}

	releaseVersion := sysCfg.ReleaseVersion
	if releaseVersion == "" {
		releaseVersion = profile.releaseVersion
	}
	if releaseVersion != profile.releaseVersion {
		return nil, fmt.Errorf("release_version %q incompatible with capability_profile %q", releaseVersion, profile.name)
	}

	listenAddr := ""
	if len(cfg.Server.Listeners) > 0 {
		listenAddr = cfg.Server.Listeners[0].Address
	}
	if listenAddr == "" {
		listenAddr = "0.0.0.0:9042"
	}
	listenIP, listenPort := parseAddress(listenAddr)
	defaultNative := sysCfg.NativePort
	if defaultNative == 0 {
		if listenPort > 0 {
			defaultNative = listenPort
		} else {
			defaultNative = defaultNativePort
		}
	}
	defaultStorage := sysCfg.StoragePort
	if defaultStorage == 0 {
		defaultStorage = defaultStoragePort
	}

	state := &SystemTableState{
		clusterName:           sysCfg.ClusterName,
		datacenter:            sysCfg.Datacenter,
		rack:                  sysCfg.Rack,
		partitioner:           sysCfg.Partitioner,
		cqlVersion:            sysCfg.CQLVersion,
		releaseVersion:        releaseVersion,
		nativeProtocolVersion: nativeProtocolVersion,
		advertisedAddress:     sysCfg.AdvertisedAddress,
		listenIP:              listenIP,
		listenPort:            listenPort,
		nativePort:            sysCfg.NativePort,
		storagePort:           defaultStorage,
	}

	if sysCfg.AdvertisedAddress != "" {
		if _, err := ip16Hex(sysCfg.AdvertisedAddress); err != nil {
			return nil, fmt.Errorf("invalid system_tables.advertised_address %q: %w", sysCfg.AdvertisedAddress, err)
		}
	}
	if strings.TrimSpace(sysCfg.HostID) != "" {
		if _, err := parseUUIDBytes(sysCfg.HostID); err != nil {
			return nil, fmt.Errorf("invalid system_tables.host_id %q: %w", sysCfg.HostID, err)
		}
		state.localHostFixed = true
		state.localFixedHost = sysCfg.HostID
	}

	localTokens, err := normalizeTokens(sysCfg.Tokens)
	if err != nil {
		return nil, fmt.Errorf("invalid system_tables.tokens: %w", err)
	}
	state.localTokens = localTokens

	if listenIP != "" {
		if ip := net.ParseIP(listenIP); ip != nil && !ip.IsUnspecified() {
			state.localIPFixed = true
			state.localFixedIP = listenIP
			state.localFixedIPHex, _ = ip16Hex(listenIP)
		}
	}
	if sysCfg.AdvertisedAddress != "" {
		state.localIPFixed = true
		state.localFixedIP = sysCfg.AdvertisedAddress
		state.localFixedIPHex, _ = ip16Hex(sysCfg.AdvertisedAddress)
	}

	if sysCfg.NativePort > 0 {
		state.localPortFixed = true
		state.localFixedPort = sysCfg.NativePort
	} else if listenPort > 0 {
		state.localPortFixed = true
		state.localFixedPort = listenPort
	}

	if state.localIPFixed && state.localPortFixed {
		hostID, err := hostIDFor(state.localFixedIPHex, state.localFixedPort)
		if err != nil {
			return nil, err
		}
		if !state.localHostFixed {
			state.localFixedHost = hostID
		}
	}

	peerStates := make([]peerState, 0, len(sysCfg.Peers))
	for _, peer := range sysCfg.Peers {
		if strings.TrimSpace(peer.AdvertisedAddress) == "" {
			return nil, fmt.Errorf("system_tables.peers advertised_address is required")
		}
		ipHex, err := ip16Hex(peer.AdvertisedAddress)
		if err != nil {
			return nil, fmt.Errorf("invalid peer advertised_address %q: %w", peer.AdvertisedAddress, err)
		}
		nativePort := peer.NativePort
		if nativePort == 0 {
			nativePort = defaultNative
		}
		if nativePort <= 0 {
			return nil, fmt.Errorf("invalid peer native_port for %q", peer.AdvertisedAddress)
		}
		storagePort := peer.StoragePort
		if storagePort == 0 {
			storagePort = defaultStorage
		}
		if storagePort <= 0 {
			return nil, fmt.Errorf("invalid peer storage_port for %q", peer.AdvertisedAddress)
		}
		peerTokens, err := normalizeTokens(peer.Tokens)
		if err != nil {
			return nil, fmt.Errorf("invalid peer tokens for %q: %w", peer.AdvertisedAddress, err)
		}
		hostID, err := hostIDFor(ipHex, nativePort)
		if err != nil {
			return nil, err
		}
		peerStates = append(peerStates, peerState{
			address:     peer.AdvertisedAddress,
			ip16hex:     ipHex,
			nativePort:  nativePort,
			storagePort: storagePort,
			hostID:      hostID,
			tokens:      peerTokens,
		})
	}
	sort.Slice(peerStates, func(i, j int) bool {
		if peerStates[i].ip16hex == peerStates[j].ip16hex {
			return peerStates[i].nativePort < peerStates[j].nativePort
		}
		return peerStates[i].ip16hex < peerStates[j].ip16hex
	})
	state.peers = peerStates

	state.tokensConfigured = len(state.localTokens) > 0
	for _, peer := range state.peers {
		if len(peer.tokens) > 0 {
			state.tokensConfigured = true
			break
		}
	}
	if state.tokensConfigured {
		if err := validateExplicitTokens(state.localTokens, state.peers); err != nil {
			return nil, err
		}
		peerTokens := make([][]string, len(state.peers))
		for i, peer := range state.peers {
			peerTokens[i] = peer.tokens
		}
		state.peerTokens = peerTokens
	}

	if err := validateHostIDUniqueness(state); err != nil {
		return nil, err
	}
	if err := validatePeerOverlap(state); err != nil {
		return nil, err
	}

	schemaProvider, err := newSchemaVersionProvider(sysCfg.SchemaVersion, schemaManager)
	if err != nil {
		return nil, err
	}
	state.schemaProvider = schemaProvider

	if !state.tokensConfigured && state.localIPFixed && state.localPortFixed {
		if err := state.cacheAutoTokens(state.localFixedIPHex, state.localFixedPort); err != nil {
			return nil, err
		}
	}

	return state, nil
}

// EnsureSchemaVersion initializes schema_version if missing.
func (s *SystemTableState) EnsureSchemaVersion(ctx context.Context, user, password string) error {
	if s == nil || s.schemaProvider == nil {
		return nil
	}
	if value := s.schemaVersion.Load(); value != nil {
		return nil
	}
	s.schemaMu.Lock()
	defer s.schemaMu.Unlock()
	if value := s.schemaVersion.Load(); value != nil {
		return nil
	}
	current, err := s.schemaProvider.Current(ctx, user, password)
	if err != nil {
		return err
	}
	s.schemaVersion.Store(current)
	return nil
}

// SchemaVersion returns the cached schema_version string.
func (s *SystemTableState) SchemaVersion() string {
	if s == nil {
		return ""
	}
	if value := s.schemaVersion.Load(); value != nil {
		return value.(string)
	}
	return ""
}

// RefreshSchemaVersion reloads schema_version and updates the cached value if it changed.
// It returns the current value, whether it changed from a previously cached value, and any error.
func (s *SystemTableState) RefreshSchemaVersion(ctx context.Context, user, password string) (string, bool, error) {
	if s == nil || s.schemaProvider == nil {
		return "", false, nil
	}
	s.schemaMu.Lock()
	defer s.schemaMu.Unlock()

	previous := ""
	if value := s.schemaVersion.Load(); value != nil {
		previous = value.(string)
	}

	current, err := s.schemaProvider.Current(ctx, user, password)
	if err != nil {
		return "", false, err
	}
	if previous == "" {
		s.schemaVersion.Store(current)
		return current, false, nil
	}
	if current != previous {
		s.schemaVersion.Store(current)
		return current, true, nil
	}
	return previous, false, nil
}

// BumpSchemaVersion updates schema_version using the configured strategy.
func (s *SystemTableState) BumpSchemaVersion(ctx context.Context, user, password string) (string, error) {
	if s == nil || s.schemaProvider == nil {
		return "", nil
	}
	s.schemaMu.Lock()
	defer s.schemaMu.Unlock()
	next, err := s.schemaProvider.Bump(ctx, user, password)
	if err != nil {
		return "", err
	}
	s.schemaVersion.Store(next)
	return next, nil
}

func (s *SystemTableState) localNode(ctx context.Context) (nodeIdentity, error) {
	ipStr, ipHex, err := s.resolveLocalIP(ctx)
	if err != nil {
		return nodeIdentity{}, err
	}
	nativePort := s.resolveLocalNativePort(ctx)
	hostID := s.localFixedHost
	if !s.localHostFixed && (hostID == "" || ipHex != s.localFixedIPHex || nativePort != s.localFixedPort) {
		hostID, err = hostIDFor(ipHex, nativePort)
		if err != nil {
			return nodeIdentity{}, err
		}
	}

	local := nodeIdentity{
		ip:          ipStr,
		ip16hex:     ipHex,
		nativePort:  nativePort,
		storagePort: s.storagePort,
		hostID:      hostID,
	}
	local.tokens, _, err = s.resolveTokens(ctx, local)
	if err != nil {
		return nodeIdentity{}, err
	}
	return local, nil
}

func (s *SystemTableState) peerNodes(ctx context.Context, local nodeIdentity) ([]nodeIdentity, error) {
	if s == nil || len(s.peers) == 0 {
		return nil, nil
	}
	_, peerTokens, err := s.resolveTokens(ctx, local)
	if err != nil {
		return nil, err
	}
	peers := make([]nodeIdentity, 0, len(s.peers))
	for i, peer := range s.peers {
		node := nodeIdentity{
			ip:          peer.address,
			ip16hex:     peer.ip16hex,
			nativePort:  peer.nativePort,
			storagePort: peer.storagePort,
			hostID:      peer.hostID,
			tokens:      peerTokens[i],
		}
		peers = append(peers, node)
	}
	return peers, nil
}

func (s *SystemTableState) resolveLocalIP(ctx context.Context) (string, string, error) {
	if s == nil {
		ipHex, _ := ip16Hex("127.0.0.1")
		return "127.0.0.1", ipHex, nil
	}
	if s.advertisedAddress != "" {
		ipHex, err := ip16Hex(s.advertisedAddress)
		return s.advertisedAddress, ipHex, err
	}
	if s.listenIP != "" {
		ip := net.ParseIP(s.listenIP)
		if ip != nil && !ip.IsUnspecified() {
			ipHex, err := ip16Hex(s.listenIP)
			return s.listenIP, ipHex, err
		}
	}
	if connIP := localConnIP(ctx); connIP != "" {
		ipHex, err := ip16Hex(connIP)
		return connIP, ipHex, err
	}
	ipHex, err := ip16Hex("127.0.0.1")
	return "127.0.0.1", ipHex, err
}

func (s *SystemTableState) resolveLocalNativePort(ctx context.Context) int {
	if s == nil {
		return defaultNativePort
	}
	if s.nativePort > 0 {
		return s.nativePort
	}
	if s.listenPort > 0 {
		return s.listenPort
	}
	if connPort := localConnPort(ctx); connPort > 0 {
		return connPort
	}
	return defaultNativePort
}

func (s *SystemTableState) resolveTokens(ctx context.Context, local nodeIdentity) ([]string, [][]string, error) {
	if s == nil {
		return []string{strconv.FormatInt(minToken(), 10)}, nil, nil
	}
	if s.tokensConfigured {
		return s.localTokens, s.peerTokens, nil
	}
	if s.autoTokensCached && s.localIPFixed && s.localPortFixed &&
		local.ip16hex == s.localFixedIPHex && local.nativePort == s.localFixedPort {
		return s.autoLocalTokens, s.autoPeerTokens, nil
	}
	return s.computeAutoTokens(local)
}

func (s *SystemTableState) computeAutoTokens(local nodeIdentity) ([]string, [][]string, error) {
	totalNodes := len(s.peers) + 1
	if totalNodes <= 0 {
		return nil, nil, fmt.Errorf("invalid node count")
	}

	peerTokens := make([][]string, len(s.peers))
	nodes := make([]nodeIdentity, 0, totalNodes)
	nodes = append(nodes, local)
	for _, peer := range s.peers {
		nodes = append(nodes, nodeIdentity{
			ip:         peer.address,
			ip16hex:    peer.ip16hex,
			nativePort: peer.nativePort,
		})
	}

	sorted := make([]nodeIdentity, len(nodes))
	copy(sorted, nodes)
	sort.Slice(sorted, func(i, j int) bool {
		if sorted[i].ip16hex == sorted[j].ip16hex {
			return sorted[i].nativePort < sorted[j].nativePort
		}
		return sorted[i].ip16hex < sorted[j].ip16hex
	})

	ring, err := autoTokenRing(len(sorted))
	if err != nil {
		return nil, nil, err
	}

	localTokens := []string{}
	for i, node := range sorted {
		token := ring[i]
		if node.ip16hex == local.ip16hex && node.nativePort == local.nativePort {
			localTokens = []string{token}
			continue
		}
		for peerIndex, peer := range s.peers {
			if peer.ip16hex == node.ip16hex && peer.nativePort == node.nativePort {
				peerTokens[peerIndex] = []string{token}
				break
			}
		}
	}
	return localTokens, peerTokens, nil
}

func (s *SystemTableState) cacheAutoTokens(ipHex string, port int) error {
	local := nodeIdentity{ip16hex: ipHex, nativePort: port}
	localTokens, peerTokens, err := s.computeAutoTokens(local)
	if err != nil {
		return err
	}
	s.autoTokensCached = true
	s.autoLocalTokens = localTokens
	s.autoPeerTokens = peerTokens
	return nil
}

func validateProtocolVersion(value string) error {
	parsed, err := strconv.Atoi(value)
	if err != nil || parsed <= 0 {
		return fmt.Errorf("invalid native_protocol_version %q", value)
	}
	if parsed > 5 {
		return fmt.Errorf("native_protocol_version %q exceeds supported protocol", value)
	}
	return nil
}

func validateExplicitTokens(local []string, peers []peerState) error {
	if len(local) == 0 {
		return fmt.Errorf("tokens are configured but local tokens are missing")
	}
	expected := len(local)
	seen := make(map[string]struct{})
	for _, token := range local {
		seen[token] = struct{}{}
	}
	for _, peer := range peers {
		if len(peer.tokens) == 0 {
			return fmt.Errorf("tokens are configured but peer tokens are missing")
		}
		if len(peer.tokens) != expected {
			return fmt.Errorf("token count mismatch: expected %d tokens per node", expected)
		}
		for _, token := range peer.tokens {
			if _, ok := seen[token]; ok {
				return fmt.Errorf("duplicate token %q across nodes", token)
			}
			seen[token] = struct{}{}
		}
	}
	return nil
}

func validateHostIDUniqueness(state *SystemTableState) error {
	if state == nil {
		return nil
	}
	seen := make(map[string]struct{})
	if state.localFixedHost != "" {
		seen[state.localFixedHost] = struct{}{}
	}
	for _, peer := range state.peers {
		if _, ok := seen[peer.hostID]; ok {
			return fmt.Errorf("duplicate host_id %q across nodes", peer.hostID)
		}
		seen[peer.hostID] = struct{}{}
	}
	return nil
}

func validatePeerOverlap(state *SystemTableState) error {
	if state == nil || len(state.peers) == 0 {
		return nil
	}
	if !state.localIPFixed || !state.localPortFixed {
		return nil
	}
	for _, peer := range state.peers {
		if peer.ip16hex == state.localFixedIPHex && peer.nativePort == state.localFixedPort {
			return fmt.Errorf("peer %q overlaps with local node", peer.address)
		}
	}
	return nil
}

func normalizeTokens(tokens []string) ([]string, error) {
	if len(tokens) == 0 {
		return nil, nil
	}
	parsed := make([]int64, 0, len(tokens))
	seen := make(map[int64]struct{})
	for _, token := range tokens {
		trimmed := strings.TrimSpace(token)
		if trimmed == "" {
			return nil, fmt.Errorf("token value is empty")
		}
		value, err := strconv.ParseInt(trimmed, 10, 64)
		if err != nil {
			return nil, fmt.Errorf("invalid token %q", token)
		}
		if _, ok := seen[value]; ok {
			return nil, fmt.Errorf("duplicate token %q", token)
		}
		seen[value] = struct{}{}
		parsed = append(parsed, value)
	}
	sort.Slice(parsed, func(i, j int) bool { return parsed[i] < parsed[j] })
	out := make([]string, len(parsed))
	for i, value := range parsed {
		out[i] = strconv.FormatInt(value, 10)
	}
	return out, nil
}

func autoTokenRing(size int) ([]string, error) {
	if size <= 0 {
		return nil, fmt.Errorf("invalid ring size")
	}
	if size == 1 {
		return []string{strconv.FormatInt(minToken(), 10)}, nil
	}
	total := new(big.Int).Lsh(big.NewInt(1), 64)
	step := new(big.Int).Div(total, big.NewInt(int64(size)))
	current := big.NewInt(minToken())

	ring := make([]string, size)
	for i := 0; i < size; i++ {
		ring[i] = current.String()
		current.Add(current, step)
	}
	return ring, nil
}

func minToken() int64 {
	return -1 << 63
}

func hostIDFor(ipHex string, port int) (string, error) {
	name := fmt.Sprintf("host_id|ip=%s|port=%d", ipHex, port)
	return uuidV5(hostIDNamespaceUUID, name)
}

func ip16Hex(ipStr string) (string, error) {
	ip := net.ParseIP(ipStr)
	if ip == nil {
		return "", fmt.Errorf("invalid IP %q", ipStr)
	}
	ip = ip.To16()
	if ip == nil {
		return "", fmt.Errorf("invalid IP %q", ipStr)
	}
	return fmt.Sprintf("%032x", ip), nil
}

func parseAddress(addr string) (string, int) {
	if addr == "" {
		return "", 0
	}
	host, port, err := net.SplitHostPort(addr)
	if err == nil {
		if parsed, err := strconv.Atoi(port); err == nil {
			return host, parsed
		}
		return host, 0
	}
	if strings.Contains(addr, ":") {
		return addr, 0
	}
	return addr, 0
}

func localConnIP(ctx context.Context) string {
	info, ok := connInfoFromContext(ctx)
	if !ok {
		return ""
	}
	tcpAddr, ok := info.LocalAddr.(*net.TCPAddr)
	if !ok || tcpAddr == nil || tcpAddr.IP == nil || tcpAddr.IP.IsUnspecified() {
		return ""
	}
	return tcpAddr.IP.String()
}

func localConnPort(ctx context.Context) int {
	info, ok := connInfoFromContext(ctx)
	if !ok {
		return 0
	}
	tcpAddr, ok := info.LocalAddr.(*net.TCPAddr)
	if !ok || tcpAddr == nil {
		return 0
	}
	return tcpAddr.Port
}