futriix/internal/cluster/self_healing.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/self_healing.go
// Назначение: Механизмы самоисцеления для автоматического восстановления узлов

package cluster

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"futriis/internal/log"
	"futriis/internal/storage"
)

// =============================================================================
// ТИПЫ ДЛЯ SELF-HEALING
// =============================================================================

// HealingAction представляет действие по восстановлению
type HealingAction int

const (
	HealingActionNone HealingAction = iota
	HealingActionRestartNode          // Перезапуск узла
	HealingActionRejoinCluster        // Переподключение к кластеру
	HealingActionResharding           // Перебалансировка шардов
	HealingActionDataRecovery         // Восстановление данных
	HealingActionStateRestore         // Восстановление состояния из снэпшота
)

// HealingState представляет состояние процесса восстановления
type HealingState string

const (
	HealingStateIdle      HealingState = "idle"
	HealingStateDetecting HealingState = "detecting"
	HealingStateAnalyzing HealingState = "analyzing"
	HealingStateHealing   HealingState = "healing"
	HealingStateVerifying HealingState = "verifying"
	HealingStateComplete  HealingState = "complete"
	HealingStateFailed    HealingState = "failed"
)

// HealingEvent представляет событие восстановления
type HealingEvent struct {
	ID          string                 `json:"id"`
	NodeID      string                 `json:"node_id"`
	Action      HealingAction          `json:"action"`
	State       HealingState           `json:"state"`
	Timestamp   int64                  `json:"timestamp"`
	Description string                 `json:"description"`
	Error       string                 `json:"error,omitempty"`
	Data        map[string]interface{} `json:"data,omitempty"`
}

// HealingConfig содержит настройки самоисцеления
type HealingConfig struct {
	// Включено ли самоисцеление
	Enabled bool `json:"enabled"`
	// Интервал проверки состояния (сек)
	CheckIntervalSec int `json:"check_interval_sec"`
	// Количество последовательных сбоев до активации
	FailureThreshold int `json:"failure_threshold"`
	// Задержка перед началом восстановления (сек)
	HealingDelaySec int `json:"healing_delay_sec"`
	// Максимальное время восстановления (сек)
	HealingTimeoutSec int `json:"healing_timeout_sec"`
	// Автоматический перезапуск узла
	AutoRestart bool `json:"auto_restart"`
	// Автоматическая перебалансировка
	AutoResharding bool `json:"auto_resharding"`
	// Автоматическое восстановление данных
	AutoDataRecovery bool `json:"auto_data_recovery"`
	// Максимальное количество одновременных восстановлений
	MaxConcurrentHealings int `json:"max_concurrent_healings"`
}

// DefaultHealingConfig возвращает конфигурацию по умолчанию
func DefaultHealingConfig() *HealingConfig {
	return &HealingConfig{
		Enabled:               true,
		CheckIntervalSec:      10,
		FailureThreshold:      3,
		HealingDelaySec:       5,
		HealingTimeoutSec:     300,
		AutoRestart:           true,
		AutoResharding:        true,
		AutoDataRecovery:      true,
		MaxConcurrentHealings: 3,
	}
}

// =============================================================================
// SELF-HEALING MANAGER
// =============================================================================

// SelfHealingManager управляет механизмами самоисцеления
type SelfHealingManager struct {
	config          *HealingConfig
	logger          *log.Logger
	coordinator     *RaftCoordinator
	gossipManager   *GossipManager
	store           *storage.Storage
	healingEvents   sync.Map // map[string]*HealingEvent
	activeHealings  atomic.Int32
	stopChan        chan struct{}
	wg              sync.WaitGroup
	mu              sync.RWMutex
	nodeFailures    map[string]int
	healingHistory  []*HealingEvent
	metrics         *HealingMetrics
	eventHandlers   map[string]func(*HealingEvent)
	handlerMu       sync.RWMutex
}

// HealingMetrics хранит метрики восстановления
type HealingMetrics struct {
	TotalHealings       atomic.Uint64
	SuccessfulHealings  atomic.Uint64
	FailedHealings      atomic.Uint64
	ReshardingsTriggered atomic.Uint64
	DataRecoveries      atomic.Uint64
	RestartsTriggered   atomic.Uint64
	AvgHealingDuration  atomic.Int64
}

// NewSelfHealingManager создаёт новый менеджер самоисцеления
func NewSelfHealingManager(
	config *HealingConfig,
	coordinator *RaftCoordinator,
	gossipManager *GossipManager,
	store *storage.Storage,
	logger *log.Logger,
) *SelfHealingManager {
	if config == nil {
		config = DefaultHealingConfig()
	}

	shm := &SelfHealingManager{
		config:         config,
		logger:         logger,
		coordinator:    coordinator,
		gossipManager:  gossipManager,
		store:          store,
		stopChan:       make(chan struct{}),
		nodeFailures:   make(map[string]int),
		healingHistory: make([]*HealingEvent, 0),
		metrics:        &HealingMetrics{},
		eventHandlers:  make(map[string]func(*HealingEvent)),
	}

	return shm
}

// Start запускает менеджер самоисцеления
func (shm *SelfHealingManager) Start() {
	if !shm.config.Enabled {
		shm.logger.Info("Self-healing is disabled")
		return
	}

	shm.wg.Add(2)
	go shm.monitorLoop()
	go shm.healingLoop()

	shm.logger.Info("Self-healing manager started")
}

// Stop останавливает менеджер самоисцеления
func (shm *SelfHealingManager) Stop() {
	close(shm.stopChan)
	shm.wg.Wait()
	shm.logger.Info("Self-healing manager stopped")
}

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

	ticker := time.NewTicker(time.Duration(shm.config.CheckIntervalSec) * time.Second)
	defer ticker.Stop()

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

// monitorNodes проверяет состояние всех узлов в кластере
func (shm *SelfHealingManager) monitorNodes() {
	// Получаем информацию о узлах из gossip
	membership := shm.gossipManager.GetMembership()

	for _, node := range membership {
		if node.ID == shm.coordinator.localNodeInfo.ID {
			continue
		}

		// Проверяем состояние узла
		if !node.IsAlive && !node.IsSuspect {
			shm.handleNodeFailure(node)
		}
	}

	// Проверяем также через координатор
	nodes := shm.coordinator.GetAllNodes()
	for _, node := range nodes {
		if node.ID == shm.coordinator.localNodeInfo.ID {
			continue
		}

		// Проверяем, не пропал ли узел из gossip
		if gossipNode := shm.gossipManager.GetNodeByID(node.ID); gossipNode == nil {
			// Узел известен координатору, но не в gossip - возможно проблема
			if shm.shouldHealNode(node.ID) {
				go shm.initiateHealing(node.ID, "Node missing from gossip")
			}
		}
	}
}

// handleNodeFailure обрабатывает сбой узла
func (shm *SelfHealingManager) handleNodeFailure(node *NodeState) {
	shm.mu.Lock()
	defer shm.mu.Unlock()

	// Увеличиваем счётчик сбоев
	shm.nodeFailures[node.ID]++

	if shm.logger != nil {
		shm.logger.Warn(fmt.Sprintf("Node %s failure detected (count: %d)", node.ID, shm.nodeFailures[node.ID]))
	}

	// Проверяем, превышен ли порог
	if shm.nodeFailures[node.ID] >= shm.config.FailureThreshold {
		if shm.shouldHealNode(node.ID) {
			go shm.initiateHealing(node.ID, fmt.Sprintf("Node failure threshold exceeded: %d", shm.nodeFailures[node.ID]))
		}
	}
}

// shouldHealNode проверяет, нужно ли восстанавливать узел
func (shm *SelfHealingManager) shouldHealNode(nodeID string) bool {
	// Проверяем, не идёт ли уже восстановление
	if _, ok := shm.healingEvents.Load(nodeID); ok {
		return false
	}

	// Проверяем количество активных восстановлений
	if shm.activeHealings.Load() >= int32(shm.config.MaxConcurrentHealings) {
		return false
	}

	return true
}

// initiateHealing запускает процесс восстановления узла
func (shm *SelfHealingManager) initiateHealing(nodeID string, reason string) {
	// Проверяем, не идёт ли уже восстановление
	if _, ok := shm.healingEvents.Load(nodeID); ok {
		return
	}

	// Увеличиваем счётчик активных восстановлений
	if !shm.activeHealings.CompareAndSwap(shm.activeHealings.Load(), shm.activeHealings.Load()+1) {
		return
	}
	defer shm.activeHealings.Add(-1)

	// Создаём событие восстановления
	eventID := fmt.Sprintf("healing_%s_%d", nodeID, time.Now().UnixNano())
	event := &HealingEvent{
		ID:          eventID,
		NodeID:      nodeID,
		Action:      HealingActionNone,
		State:       HealingStateDetecting,
		Timestamp:   time.Now().UnixMilli(),
		Description: reason,
		Data:        make(map[string]interface{}),
	}

	shm.healingEvents.Store(nodeID, event)
	shm.metrics.TotalHealings.Add(1)

	if shm.logger != nil {
		shm.logger.Info(fmt.Sprintf("Initiating healing for node %s: %s", nodeID, reason))
	}

	// Выполняем восстановление
	shm.executeHealing(nodeID, event)
}

// executeHealing выполняет процесс восстановления
func (shm *SelfHealingManager) executeHealing(nodeID string, event *HealingEvent) {
	// Создаём контекст с таймаутом для всего процесса восстановления
	healingTimeout := time.Duration(shm.config.HealingTimeoutSec) * time.Second
	ctx, cancel := context.WithTimeout(context.Background(), healingTimeout)
	defer cancel()

	// Фаза 1: Анализ состояния
	event.State = HealingStateAnalyzing
	event.Timestamp = time.Now().UnixMilli()
	shm.healingEvents.Store(nodeID, event)

	if shm.logger != nil {
		shm.logger.Info(fmt.Sprintf("Analyzing node %s...", nodeID))
	}

	// Проверяем контекст
	select {
	case <-ctx.Done():
		event.State = HealingStateFailed
		event.Error = "healing timeout during analysis"
		shm.healingEvents.Store(nodeID, event)
		shm.failHealing(nodeID)
		return
	default:
	}

	// Проверяем, жив ли узел
	nodeInfo := shm.coordinator.GetNodeByID(nodeID)
	if nodeInfo == nil {
		// Узел не найден - возможно, он уже удалён
		event.State = HealingStateComplete
		event.Description = "Node not found, assuming removed"
		shm.healingEvents.Store(nodeID, event)
		shm.completeHealing(nodeID)
		return
	}

	// Проверяем, не восстановился ли узел сам
	if shm.isNodeAlive(nodeID) {
		event.State = HealingStateComplete
		event.Description = "Node recovered automatically"
		shm.healingEvents.Store(nodeID, event)
		shm.completeHealing(nodeID)
		return
	}

	// Проверяем контекст перед выполнением действия
	select {
	case <-ctx.Done():
		event.State = HealingStateFailed
		event.Error = "healing timeout before action"
		shm.healingEvents.Store(nodeID, event)
		shm.failHealing(nodeID)
		return
	default:
	}

	// Фаза 2: Выбор действия
	action := shm.determineHealingAction(nodeID, nodeInfo)
	event.Action = action
	event.State = HealingStateHealing
	event.Timestamp = time.Now().UnixMilli()
	shm.healingEvents.Store(nodeID, event)

	if shm.logger != nil {
		shm.logger.Info(fmt.Sprintf("Selected healing action %v for node %s", action, nodeID))
	}

	// Фаза 3: Выполнение действия с учётом контекста
	var err error
	done := make(chan struct{})
	go func() {
		defer close(done)
		switch action {
		case HealingActionRestartNode:
			err = shm.restartNode(nodeID, event)
		case HealingActionRejoinCluster:
			err = shm.rejoinCluster(nodeID, event)
		case HealingActionResharding:
			err = shm.triggerResharding(nodeID, event)
		case HealingActionDataRecovery:
			err = shm.recoverData(nodeID, event)
		case HealingActionStateRestore:
			err = shm.restoreState(nodeID, event)
		default:
			err = fmt.Errorf("no suitable healing action found")
		}
	}()

	select {
	case <-ctx.Done():
		event.State = HealingStateFailed
		event.Error = "healing timeout during action execution"
		shm.healingEvents.Store(nodeID, event)
		shm.failHealing(nodeID)
		return
	case <-done:
		// Действие выполнено, продолжаем
	}

	// Фаза 4: Проверка результата
	event.State = HealingStateVerifying
	event.Timestamp = time.Now().UnixMilli()
	shm.healingEvents.Store(nodeID, event)

	if err != nil {
		event.State = HealingStateFailed
		event.Error = err.Error()
		shm.healingEvents.Store(nodeID, event)
		shm.failHealing(nodeID)
		if shm.logger != nil {
			shm.logger.Error(fmt.Sprintf("Healing for node %s failed: %v", nodeID, err))
		}
		return
	}

	// Проверяем контекст перед финальной проверкой
	select {
	case <-ctx.Done():
		event.State = HealingStateFailed
		event.Error = "healing timeout during verification"
		shm.healingEvents.Store(nodeID, event)
		shm.failHealing(nodeID)
		return
	default:
	}

	// Проверяем, восстановился ли узел
	if shm.isNodeAlive(nodeID) {
		event.State = HealingStateComplete
		event.Description = fmt.Sprintf("Healing completed successfully with action %v", action)
		shm.healingEvents.Store(nodeID, event)
		shm.completeHealing(nodeID)
		shm.metrics.SuccessfulHealings.Add(1)

		if shm.logger != nil {
			shm.logger.Info(fmt.Sprintf("Healing for node %s completed successfully", nodeID))
		}
	} else {
		event.State = HealingStateFailed
		event.Error = "Node did not recover after healing action"
		shm.healingEvents.Store(nodeID, event)
		shm.failHealing(nodeID)
		shm.metrics.FailedHealings.Add(1)
	}
}

// determineHealingAction определяет действие для восстановления
func (shm *SelfHealingManager) determineHealingAction(nodeID string, nodeInfo *NodeInfo) HealingAction {
	// Проверяем статус узла
	switch nodeInfo.Status {
	case "offline":
		if shm.config.AutoRestart {
			return HealingActionRestartNode
		}
		return HealingActionRejoinCluster
	case "failed":
		if shm.config.AutoDataRecovery {
			return HealingActionDataRecovery
		}
		return HealingActionRestartNode
	case "syncing":
		return HealingActionStateRestore
	default:
		return HealingActionRejoinCluster
	}
}

// restartNode перезапускает узел
func (shm *SelfHealingManager) restartNode(nodeID string, event *HealingEvent) error {
	shm.metrics.RestartsTriggered.Add(1)

	if shm.logger != nil {
		shm.logger.Info(fmt.Sprintf("Attempting to restart node %s", nodeID))
	}

	// Получаем информацию об узле
	nodeInfo := shm.coordinator.GetNodeByID(nodeID)
	if nodeInfo == nil {
		return fmt.Errorf("node %s not found", nodeID)
	}

	// Пробуем переподключиться к узлу
	address := fmt.Sprintf("%s:%d", nodeInfo.IP, nodeInfo.Port)
	if err := shm.coordinator.UpdateNodeStatus(nodeID, StatusActive); err != nil {
		return fmt.Errorf("failed to update node status: %v", err)
	}

	event.Data["address"] = address
	event.Data["method"] = "restart"

	return nil
}

// rejoinCluster переподключает узел к кластеру
func (shm *SelfHealingManager) rejoinCluster(nodeID string, event *HealingEvent) error {
	if shm.logger != nil {
		shm.logger.Info(fmt.Sprintf("Attempting to rejoin node %s to cluster", nodeID))
	}

	// Пробуем восстановить соединение
	if err := shm.coordinator.UpdateNodeStatus(nodeID, StatusActive); err != nil {
		return fmt.Errorf("failed to update node status: %v", err)
	}

	event.Data["method"] = "rejoin"

	return nil
}

// triggerResharding запускает перебалансировку шардов
func (shm *SelfHealingManager) triggerResharding(nodeID string, event *HealingEvent) error {
	if !shm.config.AutoResharding {
		return fmt.Errorf("auto resharding is disabled")
	}

	shm.metrics.ReshardingsTriggered.Add(1)

	if shm.logger != nil {
		shm.logger.Info(fmt.Sprintf("Triggering resharding due to node failure: %s", nodeID))
	}

	// Запускаем перебалансировку через координатор
	if err := shm.coordinator.TriggerResharding("self_healing_node_failure"); err != nil {
		return fmt.Errorf("failed to trigger resharding: %v", err)
	}

	event.Data["method"] = "resharding"

	return nil
}

// recoverData восстанавливает данные узла
func (shm *SelfHealingManager) recoverData(nodeID string, event *HealingEvent) error {
	if !shm.config.AutoDataRecovery {
		return fmt.Errorf("auto data recovery is disabled")
	}

	shm.metrics.DataRecoveries.Add(1)

	if shm.logger != nil {
		shm.logger.Info(fmt.Sprintf("Initiating data recovery for node %s", nodeID))
	}

	// Проверяем целостность данных
	if err := shm.verifyDataIntegrity(); err != nil {
		return fmt.Errorf("data integrity check failed: %v", err)
	}

	// Восстанавливаем данные из бэкапа если необходимо
	// TODO: Реализовать восстановление из бэкапа

	event.Data["method"] = "data_recovery"

	return nil
}

// restoreState восстанавливает состояние узла
func (shm *SelfHealingManager) restoreState(nodeID string, event *HealingEvent) error {
	if shm.logger != nil {
		shm.logger.Info(fmt.Sprintf("Restoring state for node %s", nodeID))
	}

	// Восстанавливаем состояние из снэпшота Raft
	// TODO: Реализовать восстановление состояния

	event.Data["method"] = "state_restore"

	return nil
}

// verifyDataIntegrity проверяет целостность данных
func (shm *SelfHealingManager) verifyDataIntegrity() error {
	// Проверяем целостность хранилища
	stats := shm.store.GetStats()
	if stats == nil {
		return fmt.Errorf("failed to get storage stats")
	}

	// Проверяем, что данные не повреждены
	// TODO: Реализовать полноценную проверку целостности

	return nil
}

// isNodeAlive проверяет, жив ли узел
func (shm *SelfHealingManager) isNodeAlive(nodeID string) bool {
	// Проверяем через gossip
	gossipNode := shm.gossipManager.GetNodeByID(nodeID)
	if gossipNode != nil && gossipNode.IsAlive {
		return true
	}

	// Проверяем через координатор
	nodeInfo := shm.coordinator.GetNodeByID(nodeID)
	if nodeInfo != nil && nodeInfo.Status == "active" {
		return true
	}

	return false
}

// completeHealing завершает процесс восстановления
func (shm *SelfHealingManager) completeHealing(nodeID string) {
	shm.mu.Lock()
	defer shm.mu.Unlock()

	// Удаляем из списка сбоев
	delete(shm.nodeFailures, nodeID)

	// Удаляем событие
	if event, ok := shm.healingEvents.Load(nodeID); ok {
		event.(*HealingEvent).State = HealingStateComplete
		shm.healingHistory = append(shm.healingHistory, event.(*HealingEvent))
		shm.healingEvents.Delete(nodeID)
	}
}

// failHealing отмечает неудачное восстановление
func (shm *SelfHealingManager) failHealing(nodeID string) {
	shm.mu.Lock()
	defer shm.mu.Unlock()

	if event, ok := shm.healingEvents.Load(nodeID); ok {
		shm.healingHistory = append(shm.healingHistory, event.(*HealingEvent))
		shm.healingEvents.Delete(nodeID)
	}
}

// healingLoop периодически проверяет зависшие восстановления
func (shm *SelfHealingManager) healingLoop() {
	defer shm.wg.Done()

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

	for {
		select {
		case <-shm.stopChan:
			return
		case <-ticker.C:
			shm.checkStalledHealings()
		}
	}
}

// checkStalledHealings проверяет зависшие восстановления
func (shm *SelfHealingManager) checkStalledHealings() {
	now := time.Now().UnixMilli()

	shm.healingEvents.Range(func(key, value interface{}) bool {
		nodeID := key.(string)
		event := value.(*HealingEvent)

		if event.State == HealingStateHealing || event.State == HealingStateVerifying {
			if now-event.Timestamp > int64(shm.config.HealingTimeoutSec*1000) {
				// Восстановление зависло
				event.State = HealingStateFailed
				event.Error = "Healing stalled"
				shm.healingEvents.Store(nodeID, event)
				shm.failHealing(nodeID)

				if shm.logger != nil {
					shm.logger.Warn(fmt.Sprintf("Healing for node %s stalled, marking as failed", nodeID))
				}
			}
		}

		return true
	})
}

// GetHealingStatus возвращает статус восстановления для узла
func (shm *SelfHealingManager) GetHealingStatus(nodeID string) *HealingEvent {
	if event, ok := shm.healingEvents.Load(nodeID); ok {
		return event.(*HealingEvent)
	}
	return nil
}

// GetHealingHistory возвращает историю восстановлений
func (shm *SelfHealingManager) GetHealingHistory() []*HealingEvent {
	shm.mu.RLock()
	defer shm.mu.RUnlock()

	history := make([]*HealingEvent, len(shm.healingHistory))
	copy(history, shm.healingHistory)
	return history
}

// GetMetrics возвращает метрики восстановления
func (shm *SelfHealingManager) GetMetrics() map[string]interface{} {
	return map[string]interface{}{
		"total_healings":        shm.metrics.TotalHealings.Load(),
		"successful_healings":   shm.metrics.SuccessfulHealings.Load(),
		"failed_healings":       shm.metrics.FailedHealings.Load(),
		"reshardings_triggered": shm.metrics.ReshardingsTriggered.Load(),
		"data_recoveries":       shm.metrics.DataRecoveries.Load(),
		"restarts_triggered":    shm.metrics.RestartsTriggered.Load(),
		"active_healings":       shm.activeHealings.Load(),
		"history_count":         len(shm.healingHistory),
	}
}

// RegisterEventHandler регистрирует обработчик событий
func (shm *SelfHealingManager) RegisterEventHandler(name string, handler func(*HealingEvent)) {
	shm.handlerMu.Lock()
	defer shm.handlerMu.Unlock()
	shm.eventHandlers[name] = handler
}

// emitEvent отправляет событие всем обработчикам
func (shm *SelfHealingManager) emitEvent(event *HealingEvent) {
	shm.handlerMu.RLock()
	defer shm.handlerMu.RUnlock()

	for _, handler := range shm.eventHandlers {
		go handler(event)
	}
}