futriix/internal/cluster/gossip.go

/*
 * Copyright 2026 Safronov Grigorii
 *
 * Licensed under the CDDL, Version 1.0 (the "License");
 * you may not use this file except in compliance with the License.
 *
 * You may obtain a copy of the License at
 * https://opensource.org/licenses/CDDL-1.0
 */

// Файл: internal/cluster/gossip.go
// Назначение: Gossip Protocol для автоматического обнаружения узлов
// и обмена информацией о состоянии кластера.

package cluster

import (
	"encoding/json"
	"fmt"
	"math/rand"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"futriis/internal/log"
)

// =============================================================================
// ТИПЫ ДЛЯ GOSSIP PROTOCOL
// =============================================================================

// GossipMessageType представляет тип сообщения протокола сплетен
type GossipMessageType int

const (
	GossipTypeMembership GossipMessageType = iota // Обновление членства
	GossipTypeHealth                              // Информация о здоровье
	GossipTypeLoad                                // Информация о нагрузке
	GossipTypeConfig                              // Обновление конфигурации
	GossipTypeSync                                // Синхронизация состояния
)

// GossipMessage представляет сообщение протокола сплетен
type GossipMessage struct {
	Type      GossipMessageType       `json:"type"`
	SenderID  string                  `json:"sender_id"`
	SenderIP  string                  `json:"sender_ip"`
	SenderPort int                    `json:"sender_port"`
	Timestamp int64                   `json:"timestamp"`
	Payload   json.RawMessage         `json:"payload"`
	TTL       int                     `json:"ttl"`
	Version   uint64                  `json:"version"`
}

// NodeState представляет состояние узла в протоколе сплетен
type NodeState struct {
	ID           string    `json:"id"`
	IP           string    `json:"ip"`
	Port         int       `json:"port"`
	RaftPort     int       `json:"raft_port"`
	Status       string    `json:"status"`
	LastSeen     time.Time `json:"last_seen"`
	LastHeartbeat time.Time `json:"last_heartbeat"`
	Incarnation  uint64    `json:"incarnation"`
	LoadCPU      float64   `json:"load_cpu"`
	LoadMemory   float64   `json:"load_memory"`
	LoadQueue    int64     `json:"load_queue"`
	Connections  int       `json:"connections"`
	Version      string    `json:"version"`
	IsAlive      bool      `json:"is_alive"`
	IsSuspect    bool      `json:"is_suspect"`
}

// GossipConfig содержит настройки протокола сплетен
type GossipConfig struct {
	// Интервал отправки gossip сообщений (сек)
	BroadcastIntervalSec int `json:"broadcast_interval_sec"`
	// Интервал проверки состояния узлов (сек)
	MonitorIntervalSec int `json:"monitor_interval_sec"`
	// Количество узлов для случайного распространения
	Fanout int `json:"fanout"`
	// Максимальное TTL сообщений
	MaxTTL int `json:"max_ttl"`
	// Время ожидания подтверждения (сек)
	TimeoutSec int `json:"timeout_sec"`
	// Время до перевода в подозрительные (сек)
	SuspectTimeoutSec int `json:"suspect_timeout_sec"`
	// Время до удаления узла (сек)
	DeadTimeoutSec int `json:"dead_timeout_sec"`
	// Максимальный размер буфера сообщений
	BufferSize int `json:"buffer_size"`
}

// DefaultGossipConfig возвращает конфигурацию gossip по умолчанию
func DefaultGossipConfig() *GossipConfig {
	return &GossipConfig{
		BroadcastIntervalSec: 5,
		MonitorIntervalSec:   2,
		Fanout:               3,
		MaxTTL:               3,
		TimeoutSec:           10,
		SuspectTimeoutSec:    15,
		DeadTimeoutSec:       30,
		BufferSize:           10000,
	}
}

// =============================================================================
// GOSSIP MANAGER
// =============================================================================

// GossipManager управляет протоколом сплетен для обнаружения узлов
type GossipManager struct {
	config       *GossipConfig
	logger       *log.Logger
	coordinator  *RaftCoordinator
	localNode    *NodeState
	nodes        sync.Map // map[string]*NodeState
	membership   atomic.Value // []*NodeState
	subscribers  map[string]chan *GossipMessage
	subscriberMu sync.RWMutex
	stopChan     chan struct{}
	wg           sync.WaitGroup
	udpConn      *net.UDPConn
	msgCounter   atomic.Uint64
	incarnation  atomic.Uint64
	mu           sync.RWMutex
	seeds        []string
	metrics      *GossipMetrics
}

// GossipMetrics хранит метрики gossip протокола
type GossipMetrics struct {
	MessagesSent     atomic.Uint64
	MessagesReceived atomic.Uint64
	MessagesDropped  atomic.Uint64
	NodesDiscovered  atomic.Uint64
	NodesRemoved     atomic.Uint64
	LastBroadcast    atomic.Int64
	LastSync         atomic.Int64
}

// NewGossipManager создаёт новый менеджер gossip протокола
func NewGossipManager(config *GossipConfig, coordinator *RaftCoordinator, logger *log.Logger) *GossipManager {
	if config == nil {
		config = DefaultGossipConfig()
	}

	gm := &GossipManager{
		config:      config,
		logger:      logger,
		coordinator: coordinator,
		subscribers: make(map[string]chan *GossipMessage),
		stopChan:    make(chan struct{}),
		metrics:     &GossipMetrics{},
		seeds:       make([]string, 0),
	}

	gm.membership.Store(make([]*NodeState, 0))

	return gm
}

// Start запускает gossip протокол
func (gm *GossipManager) Start() error {
	gm.localNode = &NodeState{
		ID:           gm.coordinator.localNodeInfo.ID,
		IP:           gm.coordinator.localNodeInfo.IP,
		Port:         gm.coordinator.localNodeInfo.Port,
		RaftPort:     gm.coordinator.config.Cluster.RaftPort,
		Status:       "active",
		LastSeen:     time.Now(),
		LastHeartbeat: time.Now(),
		Incarnation:  0,
		IsAlive:      true,
		IsSuspect:    false,
		Version:      "1.0",
	}

	// Запускаем UDP сервер для приёма gossip сообщений
	if err := gm.startUDPServer(); err != nil {
		return fmt.Errorf("failed to start UDP server: %v", err)
	}

	// Запускаем горутины
	gm.wg.Add(4)
	go gm.broadcastLoop()
	go gm.monitorLoop()
	go gm.receiveLoop()
	go gm.syncLoop()

	gm.logger.Info(fmt.Sprintf("Gossip protocol started on port %d", gm.coordinator.config.Cluster.RaftPort+1))

	return nil
}

// Stop останавливает gossip протокол
func (gm *GossipManager) Stop() {
	close(gm.stopChan)
	gm.wg.Wait()

	if gm.udpConn != nil {
		gm.udpConn.Close()
	}

	gm.logger.Info("Gossip protocol stopped")
}

// startUDPServer запускает UDP сервер для приёма сообщений
func (gm *GossipManager) startUDPServer() error {
	port := gm.coordinator.config.Cluster.RaftPort + 1
	addr := fmt.Sprintf("%s:%d", gm.localNode.IP, port)

	udpAddr, err := net.ResolveUDPAddr("udp", addr)
	if err != nil {
		return err
	}

	conn, err := net.ListenUDP("udp", udpAddr)
	if err != nil {
		return err
	}

	gm.udpConn = conn
	return nil
}

// broadcastLoop периодически рассылает gossip сообщения
func (gm *GossipManager) broadcastLoop() {
	defer gm.wg.Done()

	ticker := time.NewTicker(time.Duration(gm.config.BroadcastIntervalSec) * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-gm.stopChan:
			return
		case <-ticker.C:
			gm.broadcast()
		}
	}
}

// broadcast рассылает сообщение случайным узлам
func (gm *GossipManager) broadcast() {
	// Формируем сообщение с состоянием узла
	payload, _ := json.Marshal(gm.localNode)
	msg := &GossipMessage{
		Type:      GossipTypeMembership,
		SenderID:  gm.localNode.ID,
		SenderIP:  gm.localNode.IP,
		SenderPort: gm.localNode.Port,
		Timestamp: time.Now().UnixMilli(),
		Payload:   payload,
		TTL:       gm.config.MaxTTL,
		Version:   gm.incarnation.Add(1),
	}

	// Получаем список активных узлов
	nodes := gm.getActiveNodes()

	if len(nodes) == 0 {
		// Если нет известных узлов, пробуем связаться с seed узлами
		gm.broadcastToSeeds(msg)
		return
	}

	// Выбираем случайные узлы для рассылки
	fanout := gm.config.Fanout
	if fanout > len(nodes) {
		fanout = len(nodes)
	}

	selected := gm.selectRandomNodes(nodes, fanout)

	// Рассылаем сообщения
	for _, node := range selected {
		gm.sendMessage(msg, node.IP, node.Port)
	}

	gm.metrics.MessagesSent.Add(uint64(len(selected)))
	gm.metrics.LastBroadcast.Store(time.Now().UnixMilli())
}

// broadcastToSeeds рассылает сообщения seed узлам
func (gm *GossipManager) broadcastToSeeds(msg *GossipMessage) {
	gm.mu.RLock()
	seeds := make([]string, len(gm.seeds))
	copy(seeds, gm.seeds)
	gm.mu.RUnlock()

	for _, seed := range seeds {
		// Парсим seed адрес
		host, port, err := net.SplitHostPort(seed)
		if err != nil {
			continue
		}
		var portInt int
		fmt.Sscanf(port, "%d", &portInt)
		gm.sendMessage(msg, host, portInt)
	}
}

// monitorLoop периодически проверяет состояние узлов
func (gm *GossipManager) monitorLoop() {
	defer gm.wg.Done()

	ticker := time.NewTicker(time.Duration(gm.config.MonitorIntervalSec) * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-gm.stopChan:
			return
		case <-ticker.C:
			gm.monitorNodes()
		}
	}
}

// monitorNodes проверяет состояние всех известных узлов
func (gm *GossipManager) monitorNodes() {
	now := time.Now()
	deadTimeout := time.Duration(gm.config.DeadTimeoutSec) * time.Second
	suspectTimeout := time.Duration(gm.config.SuspectTimeoutSec) * time.Second

	var nodesToRemove []string

	gm.nodes.Range(func(key, value interface{}) bool {
		nodeID := key.(string)
		node := value.(*NodeState)

		if nodeID == gm.localNode.ID {
			return true
		}

		if !node.IsAlive {
			return true
		}

		lastSeen := now.Sub(node.LastSeen)

		if lastSeen > deadTimeout {
			// Узел считается мёртвым
			node.IsAlive = false
			node.IsSuspect = false
			gm.nodes.Store(nodeID, node)
			nodesToRemove = append(nodesToRemove, nodeID)
			gm.metrics.NodesRemoved.Add(1)

			if gm.logger != nil {
				gm.logger.Warn(fmt.Sprintf("Node %s considered dead (last seen %v ago)", nodeID, lastSeen))
			}

			// Уведомляем координатор об удалении узла
			if gm.coordinator != nil {
				gm.coordinator.RemoveNode(nodeID)
			}
		} else if lastSeen > suspectTimeout {
			// Узел под подозрением
			if !node.IsSuspect {
				node.IsSuspect = true
				gm.nodes.Store(nodeID, node)

				if gm.logger != nil {
					gm.logger.Warn(fmt.Sprintf("Node %s marked as suspect (last seen %v ago)", nodeID, lastSeen))
				}

				// Отправляем запрос на подтверждение
				gm.requestConfirmation(node)
			}
		}

		return true
	})

	// Удаляем мёртвые узлы
	for _, nodeID := range nodesToRemove {
		gm.nodes.Delete(nodeID)
	}

	// Обновляем список членства
	gm.updateMembership()
}

// requestConfirmation запрашивает подтверждение от узла
func (gm *GossipManager) requestConfirmation(node *NodeState) {
	msg := &GossipMessage{
		Type:      GossipTypeHealth,
		SenderID:  gm.localNode.ID,
		SenderIP:  gm.localNode.IP,
		SenderPort: gm.localNode.Port,
		Timestamp: time.Now().UnixMilli(),
		Payload:   json.RawMessage(`{"action":"ping"}`),
		TTL:       1,
	}

	gm.sendMessage(msg, node.IP, node.Port)
}

// receiveLoop принимает входящие gossip сообщения
func (gm *GossipManager) receiveLoop() {
	defer gm.wg.Done()

	buffer := make([]byte, 65536)

	for {
		select {
		case <-gm.stopChan:
			return
		default:
			n, addr, err := gm.udpConn.ReadFromUDP(buffer)
			if err != nil {
				continue
			}

			if n > 0 {
				go gm.handleMessage(buffer[:n], addr)
			}
		}
	}
}

// handleMessage обрабатывает полученное gossip сообщение
func (gm *GossipManager) handleMessage(data []byte, addr *net.UDPAddr) {
	var msg GossipMessage
	if err := json.Unmarshal(data, &msg); err != nil {
		return
	}

	gm.metrics.MessagesReceived.Add(1)

	// Обновляем информацию об отправителе
	gm.updateNodeFromMessage(&msg)

	// Обрабатываем сообщение в зависимости от типа
	switch msg.Type {
	case GossipTypeMembership:
		gm.handleMembershipMessage(&msg)
	case GossipTypeHealth:
		gm.handleHealthMessage(&msg)
	case GossipTypeLoad:
		gm.handleLoadMessage(&msg)
	case GossipTypeConfig:
		gm.handleConfigMessage(&msg)
	case GossipTypeSync:
		gm.handleSyncMessage(&msg)
	}

	// Пересылаем сообщение дальше (если TTL > 0)
	if msg.TTL > 0 && msg.SenderID != gm.localNode.ID {
		msg.TTL--
		gm.forwardMessage(&msg)
	}
}

// updateNodeFromMessage обновляет информацию об узле из сообщения
func (gm *GossipManager) updateNodeFromMessage(msg *GossipMessage) {
	// Проверяем, не от себя ли сообщение
	if msg.SenderID == gm.localNode.ID {
		return
	}

	var nodeState NodeState
	if msg.Type == GossipTypeMembership {
		if err := json.Unmarshal(msg.Payload, &nodeState); err != nil {
			return
		}
	} else {
		// Для других типов создаём базовое состояние
		nodeState = NodeState{
			ID:       msg.SenderID,
			IP:       msg.SenderIP,
			Port:     msg.SenderPort,
			LastSeen: time.Now(),
			IsAlive:  true,
		}
	}

	// Обновляем или создаём узел
	if existing, ok := gm.nodes.Load(msg.SenderID); ok {
		existingNode := existing.(*NodeState)

		// Обновляем только если новее
		if nodeState.Incarnation > existingNode.Incarnation {
			existingNode.IP = nodeState.IP
			existingNode.Port = nodeState.Port
			existingNode.LastSeen = time.Now()
			existingNode.Incarnation = nodeState.Incarnation
			existingNode.IsAlive = true
			existingNode.IsSuspect = false
			if nodeState.Status != "" {
				existingNode.Status = nodeState.Status
			}
			gm.nodes.Store(msg.SenderID, existingNode)
		} else {
			// Обновляем только время последнего контакта
			existingNode.LastSeen = time.Now()
			if existingNode.IsSuspect {
				existingNode.IsSuspect = false
			}
			gm.nodes.Store(msg.SenderID, existingNode)
		}
	} else {
		// Новый узел
		gm.nodes.Store(msg.SenderID, &nodeState)
		gm.metrics.NodesDiscovered.Add(1)

		if gm.logger != nil {
			gm.logger.Info(fmt.Sprintf("Discovered new node: %s (%s:%d)", msg.SenderID, msg.SenderIP, msg.SenderPort))
		}
	}

	gm.updateMembership()
}

// handleMembershipMessage обрабатывает сообщение о членстве
func (gm *GossipManager) handleMembershipMessage(msg *GossipMessage) {
	// Обновление уже выполнено в updateNodeFromMessage
}

// handleHealthMessage обрабатывает сообщение о здоровье
func (gm *GossipManager) handleHealthMessage(msg *GossipMessage) {
	// Отвечаем на ping запросы
	var payload map[string]string
	if err := json.Unmarshal(msg.Payload, &payload); err == nil {
		if action, ok := payload["action"]; ok && action == "ping" {
			// Отправляем pong
			response := &GossipMessage{
				Type:      GossipTypeHealth,
				SenderID:  gm.localNode.ID,
				SenderIP:  gm.localNode.IP,
				SenderPort: gm.localNode.Port,
				Timestamp: time.Now().UnixMilli(),
				Payload:   json.RawMessage(`{"action":"pong"}`),
				TTL:       1,
			}
			// Отправляем обратно отправителю
			host := msg.SenderIP
			port := msg.SenderPort
			gm.sendMessage(response, host, port)
		}
	}
}

// handleLoadMessage обрабатывает сообщение о нагрузке
func (gm *GossipManager) handleLoadMessage(msg *GossipMessage) {
	// TODO: Обновляем информацию о нагрузке узла
}

// handleConfigMessage обрабатывает сообщение об изменении конфигурации
func (gm *GossipManager) handleConfigMessage(msg *GossipMessage) {
	// TODO: Применяем изменения конфигурации
}

// handleSyncMessage обрабатывает сообщение синхронизации
func (gm *GossipManager) handleSyncMessage(msg *GossipMessage) {
	// Отправляем полное состояние кластера
	if msg.SenderID != gm.localNode.ID {
		gm.sendFullState(msg.SenderIP, msg.SenderPort)
	}
}

// sendMessage отправляет gossip сообщение указанному узлу
func (gm *GossipManager) sendMessage(msg *GossipMessage, ip string, port int) {
	if gm.udpConn == nil {
		return
	}

	// Сериализуем сообщение
	data, err := json.Marshal(msg)
	if err != nil {
		return
	}

	// Отправляем
	addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", ip, port))
	if err != nil {
		return
	}

	gm.udpConn.WriteToUDP(data, addr)
}

// forwardMessage пересылает сообщение дальше
func (gm *GossipManager) forwardMessage(msg *GossipMessage) {
	// Если TTL истёк, не пересылаем
	if msg.TTL <= 0 {
		return
	}

	// Получаем активные узлы
	nodes := gm.getActiveNodes()
	if len(nodes) == 0 {
		return
	}

	// Выбираем случайные узлы для пересылки
	selected := gm.selectRandomNodes(nodes, gm.config.Fanout)
	for _, node := range selected {
		if node.ID != msg.SenderID && node.ID != gm.localNode.ID {
			gm.sendMessage(msg, node.IP, node.Port)
		}
	}
}

// sendFullState отправляет полное состояние кластера
func (gm *GossipManager) sendFullState(ip string, port int) {
	nodes := gm.getAllNodes()
	payload, _ := json.Marshal(nodes)

	msg := &GossipMessage{
		Type:      GossipTypeSync,
		SenderID:  gm.localNode.ID,
		SenderIP:  gm.localNode.IP,
		SenderPort: gm.localNode.Port,
		Timestamp: time.Now().UnixMilli(),
		Payload:   payload,
		TTL:       1,
	}

	gm.sendMessage(msg, ip, port)
}

// syncLoop периодически синхронизирует состояние с другими узлами
func (gm *GossipManager) syncLoop() {
	defer gm.wg.Done()

	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-gm.stopChan:
			return
		case <-ticker.C:
			gm.syncWithCluster()
		}
	}
}

// syncWithCluster синхронизирует состояние с кластером
func (gm *GossipManager) syncWithCluster() {
	// Выбираем случайный узел для синхронизации
	nodes := gm.getActiveNodes()
	if len(nodes) == 0 {
		return
	}

	// Выбираем случайный узел, кроме себя
	var target *NodeState
	for _, node := range nodes {
		if node.ID != gm.localNode.ID {
			target = node
			break
		}
	}

	if target == nil {
		return
	}

	// Запрашиваем полное состояние
	gm.sendFullState(target.IP, target.Port)
	gm.metrics.LastSync.Store(time.Now().UnixMilli())
}

// getActiveNodes возвращает список активных узлов
func (gm *GossipManager) getActiveNodes() []*NodeState {
	nodes := make([]*NodeState, 0)
	gm.nodes.Range(func(key, value interface{}) bool {
		node := value.(*NodeState)
		if node.IsAlive && node.ID != gm.localNode.ID {
			nodes = append(nodes, node)
		}
		return true
	})
	return nodes
}

// getAllNodes возвращает список всех узлов
func (gm *GossipManager) getAllNodes() []*NodeState {
	nodes := make([]*NodeState, 0)
	gm.nodes.Range(func(key, value interface{}) bool {
		nodes = append(nodes, value.(*NodeState))
		return true
	})
	return nodes
}

// selectRandomNodes выбирает случайные узлы из списка
func (gm *GossipManager) selectRandomNodes(nodes []*NodeState, count int) []*NodeState {
	if len(nodes) == 0 {
		return nil
	}

	if count > len(nodes) {
		count = len(nodes)
	}

	// Создаём копию и перемешиваем
	shuffled := make([]*NodeState, len(nodes))
	copy(shuffled, nodes)

	// Перемешиваем с использованием rand
	for i := len(shuffled) - 1; i > 0; i-- {
		j := rand.Intn(i + 1)
		shuffled[i], shuffled[j] = shuffled[j], shuffled[i]
	}

	return shuffled[:count]
}

// updateMembership обновляет список членства
func (gm *GossipManager) updateMembership() {
	nodes := gm.getAllNodes()
	gm.membership.Store(nodes)
}

// GetMembership возвращает текущий список членства
func (gm *GossipManager) GetMembership() []*NodeState {
	return gm.membership.Load().([]*NodeState)
}

// GetNodeByID возвращает узел по ID
func (gm *GossipManager) GetNodeByID(id string) *NodeState {
	if val, ok := gm.nodes.Load(id); ok {
		return val.(*NodeState)
	}
	return nil
}

// AddSeed добавляет seed узел
func (gm *GossipManager) AddSeed(addr string) {
	gm.mu.Lock()
	defer gm.mu.Unlock()

	// Проверяем, что адрес не дублируется
	for _, existing := range gm.seeds {
		if existing == addr {
			return
		}
	}
	gm.seeds = append(gm.seeds, addr)
}

// Subscribe подписывается на события gossip
func (gm *GossipManager) Subscribe(id string) <-chan *GossipMessage {
	gm.subscriberMu.Lock()
	defer gm.subscriberMu.Unlock()

	ch := make(chan *GossipMessage, gm.config.BufferSize)
	gm.subscribers[id] = ch
	return ch
}

// Unsubscribe отписывается от событий gossip
func (gm *GossipManager) Unsubscribe(id string) {
	gm.subscriberMu.Lock()
	defer gm.subscriberMu.Unlock()

	if ch, ok := gm.subscribers[id]; ok {
		close(ch)
		delete(gm.subscribers, id)
	}
}

// GetMetrics возвращает метрики gossip протокола
func (gm *GossipManager) GetMetrics() map[string]interface{} {
	return map[string]interface{}{
		"messages_sent":      gm.metrics.MessagesSent.Load(),
		"messages_received":  gm.metrics.MessagesReceived.Load(),
		"messages_dropped":   gm.metrics.MessagesDropped.Load(),
		"nodes_discovered":   gm.metrics.NodesDiscovered.Load(),
		"nodes_removed":      gm.metrics.NodesRemoved.Load(),
		"last_broadcast":     gm.metrics.LastBroadcast.Load(),
		"last_sync":          gm.metrics.LastSync.Load(),
		"known_nodes":        gm.getNodeCount(),
		"active_nodes":       len(gm.getActiveNodes()),
	}
}

// getNodeCount возвращает количество известных узлов
func (gm *GossipManager) getNodeCount() int {
	count := 0
	gm.nodes.Range(func(key, value interface{}) bool {
		count++
		return true
	})
	return count
}

// UpdateLocalNode обновляет локальное состояние узла
func (gm *GossipManager) UpdateLocalNode(status string, loadCPU float64, loadMemory float64) {
	gm.mu.Lock()
	defer gm.mu.Unlock()

	if gm.localNode != nil {
		gm.localNode.Status = status
		gm.localNode.LoadCPU = loadCPU
		gm.localNode.LoadMemory = loadMemory
		gm.localNode.LastHeartbeat = time.Now()
		gm.localNode.Incarnation++
	}
}