futriix/internal/cluster/node.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/node.go
// Назначение: Реализация узла кластера (node) для распределённой СУБД с поддержкой временных меток.
//
// ОСНОВНЫЕ ФУНКЦИИ:
// 1. Управление жизненным циклом узла (создание, запуск, остановка)
// 2. TCP-сервер для приёма входящих соединений от других узлов
// 3. Обработка различных типов запросов (репликация, запросы, синхронизация, heartbeat)
// 4. Управление состоянием узла (active, syncing, offline, failed)
// 5. Интеграция с Raft-координатором для управления кластером
// 6. Пул воркеров для асинхронной обработки запросов
// 7. Механизм восстановления после паники (panic recovery)
// 8. Репликация документов между узлами кластера
// 9. Сбор статистики и мониторинг состояния

package cluster

import (
    "crypto/rand"
    "encoding/base64"
    "encoding/json"
    "fmt"
    "io"
    "net"
    "runtime/debug"
    "sync"
    "sync/atomic"
    "time"
    
    "futriis/internal/log"
    "futriis/internal/storage"
    "github.com/google/uuid"
    "futriis/internal/config" 
)

// =============================================================================
// NETWORK REPLICATOR - ЗАГЛУШКА
// =============================================================================

// NetworkReplicator - заглушка для сетевой репликации.
// В реальной реализации этот компонент отвечал бы за отправку данных на другие узлы,  с поддержкой повторных попыток, backoff и джиттера.
// Полная реализация с поддержкой протокола gRPC или HTTP/2
type NetworkReplicator struct{}

// NewNetworkReplicator создаёт новый экземпляр NetworkReplicator.
// Принимает конфигурацию повторных попыток, пул воркеров и логгер.
// Возвращает заглушку, которая не выполняет реальной репликации.
func NewNetworkReplicator(config *ReplicationRetryConfig, workerPool *WorkerPool, logger *log.Logger) *NetworkReplicator {
    return &NetworkReplicator{}
}

// ReplicationRetryConfig определяет конфигурацию повторных попыток для репликации.
// Содержит параметры экспоненциальной задержки и джиттера.
type ReplicationRetryConfig struct {
    MaxRetries     int           // Максимальное количество попыток
    InitialBackoff time.Duration // Начальная задержка между попытками
    MaxBackoff     time.Duration // Максимальная задержка
    BackoffFactor  float64       // Множитель для экспоненциального увеличения задержки
    JitterEnabled  bool          // Включение случайного джиттера для предотвращения "thundering herd"
}

// DefaultReplicationRetryConfig возвращает конфигурацию по умолчанию.
// Настройки подобраны для баланса между надёжностью и производительностью.
func DefaultReplicationRetryConfig() *ReplicationRetryConfig {
    return &ReplicationRetryConfig{
        MaxRetries:     5,
        InitialBackoff: 100 * time.Millisecond,
        MaxBackoff:     10 * time.Second,
        BackoffFactor:  2.0,
        JitterEnabled:  true,
    }
}

// Close закрывает репликатор и освобождает ресурсы.
// В заглушке ничего не делает.
func (nr *NetworkReplicator) Close() error {
    return nil
}

// GetStats возвращает статистику работы репликатора.
// В заглушке возвращает информацию о том, что репликация отключена.
func (nr *NetworkReplicator) GetStats() map[string]interface{} {
    return map[string]interface{}{
        "enabled": false,
        "message": "NetworkReplicator is a stub - full implementation pending",
    }
}

// Replicate выполняет репликацию данных на целевой узел.
// В заглушке всегда возвращает nil (успех).
func (nr *NetworkReplicator) Replicate(targetNodeID, targetAddress string, data []byte) error {
    return nil
}

// =============================================================================
// ОСНОВНЫЕ ТИПЫ
// =============================================================================

// LoggerInterface определяет интерфейс для логирования.
// Используется для абстракции от конкретной реализации логгера.
type LoggerInterface interface {
    Debug(msg string)
    Info(msg string)
    Warn(msg string)
    Error(msg string)
    Debugf(format string, args ...interface{})
    Infof(format string, args ...interface{})
    Warnf(format string, args ...interface{})
    Errorf(format string, args ...interface{})
}

// NodeStatus представляет состояние узла кластера.
// Использует атомарные операции для потокобезопасного доступа.
type NodeStatus int32

const (
    StatusOffline NodeStatus = iota // Узел отключён или недоступен
    StatusActive                    // Узел активен и готов к работе
    StatusSyncing                   // Узел синхронизируется с кластером
    StatusFailed                    // Узел в состоянии ошибки
)

// NodeRequest представляет запрос между узлами кластера.
// Используется для передачи данных в формате JSON через TCP.
type NodeRequest struct {
    Type      string          `json:"type"`       // Тип запроса: replicate, query, sync, heartbeat, status_sync
    FromNode  string          `json:"from_node"`  // ID узла-отправителя
    Data      json.RawMessage `json:"data"`       // Данные запроса в формате JSON
    Timestamp int64           `json:"timestamp"`  // Временная метка запроса (Unix millis)
}

// NodeInfo представляет информацию об узле в кластере.
// Используется для обмена метаданными между узлами.
type NodeInfo struct {
    ID        string `json:"id"`         // Уникальный идентификатор узла
    IP        string `json:"ip"`         // IP-адрес узла
    Port      int    `json:"port"`       // Порт для TCP-соединений
    Status    string `json:"status"`     // Текущий статус узла
    LastSeen  int64  `json:"last_seen"`  // Время последнего контакта (Unix millis)
    JoinedAt  int64  `json:"joined_at"`  // Время присоединения к кластеру
    UpdatedAt int64  `json:"updated_at"` // Время последнего обновления информации
    Version   int    `json:"version"`    // Версия узла для обнаружения изменений
}

// ShardInfo представляет информацию о шарде.
// Используется для управления распределением данных между узлами.
type ShardInfo struct {
    ID             string   `json:"id"`               // Уникальный идентификатор шарда
    Name           string   `json:"name"`             // Имя шарда
    Nodes          []string `json:"nodes"`            // Список узлов, хранящих шард
    LeaderNode     string   `json:"leader_node"`      // Лидер шарда
    Status         string   `json:"status"`           // Статус шарда
    CreatedAt      int64    `json:"created_at"`       // Время создания
    UpdatedAt      int64    `json:"updated_at"`       // Время последнего обновления
    LastRebalanced int64    `json:"last_rebalanced"`  // Время последней перебалансировки
    DocumentCount  int64    `json:"document_count"`   // Количество документов в шарде
    SizeBytes      int64    `json:"size_bytes"`       // Размер шарда в байтах
}

// =============================================================================
// ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ
// =============================================================================

// SafeGoWithLogger запускает горутину с защитой от паники.
// При панике горутина перезапускается с задержкой 5 секунд.
// Это обеспечивает самовосстановление критических компонентов.
//
// Параметры:
//   - fn: функция для выполнения в горутине
//   - logger: логгер для записи ошибок
//   - name: имя горутины для идентификации в логах
func SafeGoWithLogger(fn func(), logger *log.Logger, name string) {
    go func() {
        defer func() {
            if r := recover(); r != nil {
                if logger != nil {
                    logger.Error(fmt.Sprintf("Goroutine %s panicked: %v\n%s", name, r, debug.Stack()))
                }
                time.Sleep(5 * time.Second)
                SafeGoWithLogger(fn, logger, name)
            }
        }()
        fn()
    }()
}

// =============================================================================
// NODE - ОСНОВНАЯ СТРУКТУРА УЗЛА
// =============================================================================

// Node представляет узел в распределённом кластере.
// Каждый узел имеет свой TCP-сервер, хранилище данных и интеграцию с Raft-координатором.
//
// ПОТОКОВАЯ БЕЗОПАСНОСТЬ:
// - Status, lastSeen, joinedAt используют atomic для потокобезопасного доступа
// - mu защищает операции, изменяющие состояние узла
// - connPool использует sync.Map для потокобезопасного хранения соединений
// - routinesMu защищает доступ к recoverableRoutines
type Node struct {
    // Основная информация об узле
    ID           string // Уникальный идентификатор узла (UUID)
    IP           string // IP-адрес для приёма соединений
    Port         int    // Порт для TCP-сервера
    
    // Состояние узла (атомарное для потокобезопасности)
    Status       atomic.Int32 // Текущий статус (NodeStatus)
    
    // Хранилище и компоненты
    Storage      *storage.Storage // Хранилище данных
    logger       *log.Logger      // Логгер для записи событий
    coordinator  *RaftCoordinator // Координатор для управления кластером (Raft)
    
    // Временные метки (атомарные)
    lastSeen     atomic.Int64 // Время последнего полученного heartbeat
    joinedAt     atomic.Int64 // Время присоединения к кластеру
    
    // Время жизненного цикла
    createdAt    int64 // Время создания узла
    startedAt    int64 // Время запуска TCP-сервера
    stoppedAt    int64 // Время остановки узла
    
    // Сетевые компоненты
    incomingConn chan net.Conn // Канал для входящих TCP-соединений (буфер 10000)
    stopChan     chan struct{} // Канал для остановки всех горутин
    
    // Статистика (атомарная)
    requestCount atomic.Uint64 // Счётчик обработанных запросов
    bytesRx      atomic.Uint64 // Количество полученных байт
    bytesTx      atomic.Uint64 // Количество отправленных байт
    
    // Защита состояния
    mu           sync.RWMutex // Блокировка для операций изменения состояния
    
    // Компоненты обработки
    workerPool   *WorkerPool         // Пул воркеров для асинхронной обработки
    replicator   *NetworkReplicator  // Репликатор для отправки данных
    connPool     sync.Map            // Пул активных TCP-соединений
    
    // Механизм восстановления после паники
    panicRecoveryMgr *PanicRecoveryManager        // Менеджер восстановления
    recoverableRoutines map[string]*RecoverableRoutine // Зарегистрированные восстанавливаемые горутины
    routinesMu         sync.RWMutex               // Защита доступа к recoverableRoutines
}

// NodeConfig представляет конфигурацию для создания узла.
// Используется для передачи зависимостей при создании.
type NodeConfig struct {
    IP          string                 // IP-адрес узла
    Port        int                    // Порт для TCP-сервера
    Storage     *storage.Storage       // Хранилище данных
    Logger      *log.Logger            // Логгер
    Coordinator *RaftCoordinator       // Raft-координатор
    PanicRecoveryMgr *PanicRecoveryManager // Менеджер восстановления
}

// =============================================================================
// СОЗДАНИЕ УЗЛА
// =============================================================================

// NewNode создаёт новый узел с механизмом восстановления по умолчанию.
// Это упрощённый конструктор для случаев, когда менеджер восстановления не требуется.
//
// Параметры:
//   - ip: IP-адрес для приёма соединений
//   - port: порт для TCP-сервера
//   - store: хранилище данных
//   - logger: логгер
//
// Возвращает: указатель на созданный узел
func NewNode(ip string, port int, store *storage.Storage, logger *log.Logger) *Node {
    return NewNodeWithRecovery(ip, port, store, logger, nil)
}

// NewNodeWithRecovery создаёт новый узел с поддержкой восстановления после паники.
// Это основной конструктор, который инициализирует все компоненты узла.
//
// ПОРЯДОК ИНИЦИАЛИЗАЦИИ:
// 1. Создание пула воркеров и репликатора
// 2. Инициализация структуры Node
// 3. Запуск восстанавливаемых горутин (если есть менеджер)
// 4. Запуск горутин с защитой от паники (если нет менеджера)
//
// Параметры:
//   - ip: IP-адрес для приёма соединений
//   - port: порт для TCP-сервера
//   - store: хранилище данных
//   - logger: логгер
//   - panicRecoveryMgr: менеджер восстановления (может быть nil)
//
// Возвращает: указатель на созданный узел
func NewNodeWithRecovery(ip string, port int, store *storage.Storage, logger *log.Logger, panicRecoveryMgr *PanicRecoveryManager) *Node {
    now := time.Now().UnixMilli()
    
    // Создаём пул воркеров с максимальным количеством 500 задач
    workerPool := NewWorkerPool(500, logger)
    // Создаём репликатор с настройками по умолчанию
    replicator := NewNetworkReplicator(DefaultReplicationRetryConfig(), workerPool, logger)
    
    // Инициализируем структуру узла
    node := &Node{
        ID:           uuid.New().String(), // Генерируем уникальный ID
        IP:           ip,
        Port:         port,
        Storage:      store,
        logger:       logger,
        incomingConn: make(chan net.Conn, 10000), // Буфер на 10000 соединений
        stopChan:     make(chan struct{}),
        createdAt:    now,
        startedAt:    now,
        workerPool:   workerPool,
        replicator:   replicator,
        panicRecoveryMgr: panicRecoveryMgr,
        recoverableRoutines: make(map[string]*RecoverableRoutine),
    }
    
    // Устанавливаем начальный статус и временные метки
    node.Status.Store(int32(StatusActive))
    node.lastSeen.Store(now)
    node.joinedAt.Store(0) // 0 означает "не присоединён"
    
    // Запускаем горутины с механизмом восстановления или без него
    if panicRecoveryMgr != nil {
        // Используем восстанавливаемые горутины для критических компонентов
        node.startRecoverableRoutine("TCPServer", node.startTCPServer)
        node.startRecoverableRoutine("IncomingConnections", node.handleIncomingConnections)
        node.startRecoverableRoutine("HeartbeatLoop", node.heartbeatLoop)
        node.startRecoverableRoutine("ConnectionHealthMonitor", node.connectionHealthMonitor)
        logger.Info(fmt.Sprintf("Node %s created with panic recovery (max workers: 500)", node.ID))
    } else {
        // Используем базовую защиту от паники с перезапуском
        SafeGoWithLogger(node.startTCPServer, logger, "TCPServer")
        SafeGoWithLogger(node.handleIncomingConnections, logger, "IncomingConnections")
        SafeGoWithLogger(node.heartbeatLoop, logger, "HeartbeatLoop")
        SafeGoWithLogger(node.connectionHealthMonitor, logger, "ConnectionHealthMonitor")
        logger.Info(fmt.Sprintf("Node %s created at %s with lock-free worker pool (max: 500)", node.ID, node.GetCreatedAtStr()))
    }
    
    return node
}

// =============================================================================
// ВОССТАНАВЛИВАЕМЫЕ ГОРУТИНЫ
// =============================================================================

// startRecoverableRoutine запускает горутину с поддержкой восстановления.
// Если panicRecoveryMgr доступен, горутина будет автоматически перезапущена при панике.
// В противном случае используется базовый механизм SafeGoWithLogger.
//
// Параметры:
//   - name: имя горутины для идентификации
//   - fn: функция для выполнения
func (n *Node) startRecoverableRoutine(name string, fn func()) {
    // Если менеджер восстановления недоступен, используем базовый механизм
    if n.panicRecoveryMgr == nil {
        SafeGoWithLogger(fn, n.logger, name)
        return
    }
    
    // Блокируем доступ к карте восстанавливаемых горутин
    n.routinesMu.Lock()
    defer n.routinesMu.Unlock()
    
    // Создаём новую восстанавливаемую горутину
    routine := NewRecoverableRoutine(name, func() error {
        fn()
        return nil
    }, n.panicRecoveryMgr, 10) // Максимум 10 перезапусков
    
    // Сохраняем и запускаем
    n.recoverableRoutines[name] = routine
    routine.Start()
    
    if n.logger != nil {
        n.logger.Debug(fmt.Sprintf("Started recoverable routine: %s", name))
    }
}

// stopRecoverableRoutine останавливает восстанавливаемую горутину.
// Используется при остановке узла для корректного завершения.
//
// Параметры:
//   - name: имя горутины для остановки
func (n *Node) stopRecoverableRoutine(name string) {
    n.routinesMu.RLock()
    routine, ok := n.recoverableRoutines[name]
    n.routinesMu.RUnlock()
    
    if ok && routine != nil {
        routine.Stop()
        if n.logger != nil {
            n.logger.Debug(fmt.Sprintf("Stopped recoverable routine: %s", name))
        }
    }
}

// =============================================================================
// TCP СЕРВЕР
// =============================================================================

// startTCPServer запускает TCP-сервер для приёма входящих соединений.
// Это основная точка входа для межузлового взаимодействия.
//
// ОСОБЕННОСТИ:
// - Использует deadline для предотвращения блокировки accept
// - Автоматически перезапускается при панике
// - Обрабатывает ошибки таймаута без остановки сервера
// - Передаёт соединения в канал incomingConn для асинхронной обработки
func (n *Node) startTCPServer() {
    // Защита от паники с автоматическим перезапуском
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("TCP server panicked: %v\n%s", r, debug.Stack()))
            }
            time.Sleep(5 * time.Second)
            if n.panicRecoveryMgr != nil {
                n.startRecoverableRoutine("TCPServer", n.startTCPServer)
            } else {
                SafeGoWithLogger(n.startTCPServer, n.logger, "TCPServer")
            }
        }
    }()
    
    // Формируем адрес и начинаем прослушивание
    addr := fmt.Sprintf("%s:%d", n.IP, n.Port)
    listener, err := net.Listen("tcp", addr)
    if err != nil {
        if n.logger != nil {
            n.logger.Error(fmt.Sprintf("Node %s failed to start TCP server: %v", n.ID, err))
        }
        n.Status.Store(int32(StatusFailed))
        return
    }
    defer listener.Close()
    
    if n.logger != nil {
        n.logger.Info(fmt.Sprintf("Node %s listening on %s (started at %s)", n.ID, addr, n.GetStartedAtStr()))
    }
    
    // Основной цикл приёма соединений
    for {
        select {
        case <-n.stopChan:
            // Сигнал остановки получен
            if n.logger != nil {
                n.logger.Info(fmt.Sprintf("Node %s TCP server stopped", n.ID))
            }
            return
        default:
            // Устанавливаем deadline для таймаута accept (5 секунд)
            // Это позволяет периодически проверять stopChan
            if err := listener.(*net.TCPListener).SetDeadline(time.Now().Add(5 * time.Second)); err != nil {
                continue
            }
            
            // Принимаем новое соединение
            conn, err := listener.Accept()
            if err != nil {
                // Проверяем, не является ли ошибка таймаутом (это нормально)
                if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
                    continue
                }
                if n.logger != nil {
                    n.logger.Error(fmt.Sprintf("Node %s accept error: %v", n.ID, err))
                }
                continue
            }
            
            // Устанавливаем таймауты для соединения
            conn.SetReadDeadline(time.Now().Add(30 * time.Second))
            conn.SetWriteDeadline(time.Now().Add(30 * time.Second))
            
            // Передаём соединение в канал для асинхронной обработки
            select {
            case n.incomingConn <- conn:
                n.bytesRx.Add(1)
            default:
                // Канал переполнен - закрываем соединение
                if n.logger != nil {
                    n.logger.Warn(fmt.Sprintf("Node %s incoming connection queue full, dropping connection", n.ID))
                }
                conn.Close()
            }
        }
    }
}

// =============================================================================
// ОБРАБОТКА ВХОДЯЩИХ СОЕДИНЕНИЙ
// =============================================================================

// handleIncomingConnections обрабатывает входящие соединения из канала.
// Каждое соединение обрабатывается асинхронно через пул воркеров.
//
// РАБОТА С ПУЛОМ ВОРКЕРОВ:
// 1. Получение соединения из канала
// 2. Создание задачи с уникальным ID
// 3. Отправка задачи в пул воркеров
// 4. При ошибке отправки - закрытие соединения
func (n *Node) handleIncomingConnections() {
    // Защита от паники с автоматическим перезапуском
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Incoming connections handler panicked: %v\n%s", r, debug.Stack()))
            }
            time.Sleep(5 * time.Second)
            if n.panicRecoveryMgr != nil {
                n.startRecoverableRoutine("IncomingConnections", n.handleIncomingConnections)
            } else {
                SafeGoWithLogger(n.handleIncomingConnections, n.logger, "IncomingConnections")
            }
        }
    }()
    
    for {
        select {
        case <-n.stopChan:
            return
        case conn := <-n.incomingConn:
            // Увеличиваем счётчик запросов
            n.requestCount.Add(1)
            
            // Создаём задачу для обработки соединения
            taskID := fmt.Sprintf("handle_conn_%d_%s", time.Now().UnixNano(), conn.RemoteAddr().String())
            err := n.workerPool.SubmitFunc(taskID, func() error {
                n.handleNodeRequest(conn)
                return nil
            })
            
            if err != nil {
                if n.logger != nil {
                    n.logger.Warn(fmt.Sprintf("Failed to submit connection task: %v", err))
                }
                conn.Close()
            }
        }
    }
}

// =============================================================================
// ОБРАБОТКА ЗАПРОСОВ
// =============================================================================

// handleNodeRequest обрабатывает один запрос от узла.
// Декодирует JSON-запрос и направляет его соответствующему обработчику.
//
// ТИПЫ ЗАПРОСОВ:
//   - replicate: репликация документа
//   - query: запрос документа
//   - sync: синхронизация коллекции
//   - heartbeat: проверка доступности узла
//   - status_sync: синхронизация статуса кластера
func (n *Node) handleNodeRequest(conn net.Conn) {
    // Защита от паники - всегда закрываем соединение
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Request handler panicked: %v\n%s", r, debug.Stack()))
            }
        }
        conn.Close()
    }()
    
    // Устанавливаем таймаут на чтение
    conn.SetReadDeadline(time.Now().Add(30 * time.Second))
    
    // Декодируем запрос из JSON
    decoder := json.NewDecoder(conn)
    var req NodeRequest
    if err := decoder.Decode(&req); err != nil {
        if err != io.EOF && n.logger != nil {
            n.logger.Error(fmt.Sprintf("Node %s failed to decode request: %v", n.ID, err))
        }
        return
    }
    
    // Обновляем время последнего контакта
    n.lastSeen.Store(time.Now().UnixMilli())
    
    if n.logger != nil {
        n.logger.Debug(fmt.Sprintf("Node %s received request type %s from %s at %s", 
            n.ID, req.Type, req.FromNode, time.UnixMilli(req.Timestamp).Format("15:04:05.000")))
    }
    
    // Маршрутизация по типу запроса
    switch req.Type {
    case "replicate":
        n.handleReplicateRequest(req.Data)
    case "query":
        n.handleQueryRequest(req.Data, conn)
    case "sync":
        n.handleSyncRequest(req.Data, conn)
    case "heartbeat":
        n.handleHeartbeatRequest(req, conn)
    case "status_sync":
        n.handleStatusSyncRequest(req, conn)
    default:
        if n.logger != nil {
            n.logger.Warn(fmt.Sprintf("Node %s unknown request type: %s", n.ID, req.Type))
        }
    }
}

// handleReplicateRequest обрабатывает запрос на репликацию документа.
// Сохраняет полученный документ в локальном хранилище.
//
// ПОРЯДОК ОБРАБОТКИ:
// 1. Декодирование данных репликации
// 2. Получение базы данных и коллекции
// 3. Создание документа с полями
// 4. Вставка документа в коллекцию
// 5. Логирование результата
func (n *Node) handleReplicateRequest(data []byte) {
    // Защита от паники
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Replicate request handler panicked: %v\n%s", r, debug.Stack()))
            }
        }
    }()
    
    startTime := time.Now().UnixMilli()
    
    // Структура данных репликации
    var repData struct {
        Database   string                 `json:"database"`    // Имя базы данных
        Collection string                 `json:"collection"`  // Имя коллекции
        Document   map[string]interface{} `json:"document"`    // Данные документа
        SourceNode string                 `json:"source_node"` // Узел-источник
        ReplicaID  string                 `json:"replica_id"`  // Уникальный ID репликации
    }
    
    if err := json.Unmarshal(data, &repData); err != nil {
        if n.logger != nil {
            n.logger.Error(fmt.Sprintf("Node %s failed to unmarshal replicate data: %v", n.ID, err))
        }
        return
    }
    
    // Получаем базу данных
    db, err := n.Storage.GetDatabase(repData.Database)
    if err != nil {
        if n.logger != nil {
            n.logger.Error(fmt.Sprintf("Node %s database not found for replication: %s", n.ID, repData.Database))
        }
        return
    }
    
    // Получаем коллекцию
    coll, err := db.GetCollection(repData.Collection)
    if err != nil {
        if n.logger != nil {
            n.logger.Error(fmt.Sprintf("Node %s collection not found for replication: %s", n.ID, repData.Collection))
        }
        return
    }
    
    // Извлекаем ID документа
    docID, ok := repData.Document["_id"].(string)
    if !ok {
        if n.logger != nil {
            n.logger.Error(fmt.Sprintf("Node %s document missing _id field", n.ID))
        }
        return
    }
    
    // Создаём документ с полученными полями
    doc := storage.NewDocumentWithID(docID)
    for k, v := range repData.Document {
        doc.SetField(k, v)
    }
    
    // Вставляем документ в коллекцию
    if err := coll.Insert(doc); err != nil {
        if n.logger != nil {
            n.logger.Error(fmt.Sprintf("Node %s failed to replicate document: %v", n.ID, err))
        }
    } else {
        // Логируем успешную репликацию
        duration := time.Now().UnixMilli() - startTime
        if n.logger != nil {
            n.logger.Debug(fmt.Sprintf("Node %s replicated document %s from %s (took %d ms)", 
                n.ID, doc.ID, repData.SourceNode, duration))
        }
    }
}

// handleQueryRequest обрабатывает запрос на получение документа.
// Находит документ по ID и возвращает его в ответе.
//
// ПОРЯДОК ОБРАБОТКИ:
// 1. Декодирование данных запроса
// 2. Получение базы данных и коллекции
// 3. Поиск документа по ID
// 4. Формирование успешного или ошибочного ответа
func (n *Node) handleQueryRequest(data []byte, conn net.Conn) {
    // Защита от паники с отправкой ошибки
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Query request handler panicked: %v\n%s", r, debug.Stack()))
            }
            n.sendErrorResponse(conn, "Internal server error")
        }
    }()
    
    startTime := time.Now().UnixMilli()
    
    // Структура данных запроса
    var queryData struct {
        Database   string `json:"database"`    // Имя базы данных
        Collection string `json:"collection"`  // Имя коллекции
        DocumentID string `json:"document_id"` // ID документа
        RequestID  string `json:"request_id"`  // ID запроса для трекинга
    }
    
    if err := json.Unmarshal(data, &queryData); err != nil {
        n.sendErrorResponse(conn, err.Error())
        return
    }
    
    // Получаем базу данных
    db, err := n.Storage.GetDatabase(queryData.Database)
    if err != nil {
        n.sendErrorResponse(conn, err.Error())
        return
    }
    
    // Получаем коллекцию
    coll, err := db.GetCollection(queryData.Collection)
    if err != nil {
        n.sendErrorResponse(conn, err.Error())
        return
    }
    
    // Ищем документ
    doc, err := coll.Find(queryData.DocumentID)
    if err != nil {
        n.sendErrorResponse(conn, err.Error())
        return
    }
    
    duration := time.Now().UnixMilli() - startTime
    
    // Формируем успешный ответ
    response := map[string]interface{}{
        "status":      "success",
        "data":        doc,
        "node_id":     n.ID,
        "request_id":  queryData.RequestID,
        "duration_ms": duration,
        "timestamp":   time.Now().UnixMilli(),
    }
    
    // Отправляем ответ
    conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
    encoder := json.NewEncoder(conn)
    if err := encoder.Encode(response); err == nil {
        responseData, _ := json.Marshal(response)
        n.bytesTx.Add(uint64(len(responseData)))
    }
    
    if n.logger != nil {
        n.logger.Debug(fmt.Sprintf("Node %s handled query for %s.%s:%s (took %d ms)", 
            n.ID, queryData.Database, queryData.Collection, queryData.DocumentID, duration))
    }
}

// handleSyncRequest обрабатывает запрос на синхронизацию коллекции.
// Возвращает все документы коллекции (или только изменённые после указанного времени).
//
// ПОРЯДОК ОБРАБОТКИ:
// 1. Декодирование данных запроса
// 2. Получение базы данных и коллекции
// 3. Фильтрация документов по времени (если указано)
// 4. Формирование ответа со всеми документами
func (n *Node) handleSyncRequest(data []byte, conn net.Conn) {
    // Защита от паники с отправкой ошибки
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Sync request handler panicked: %v\n%s", r, debug.Stack()))
            }
            n.sendErrorResponse(conn, "Internal server error")
        }
    }()
    
    startTime := time.Now().UnixMilli()
    
    // Структура данных запроса синхронизации
    var syncData struct {
        Database   string `json:"database"`    // Имя базы данных
        Collection string `json:"collection"`  // Имя коллекции
        RequestID  string `json:"request_id"`  // ID запроса
        Since      int64  `json:"since"`       // Временная метка для инкрементальной синхронизации
    }
    
    if err := json.Unmarshal(data, &syncData); err != nil {
        n.sendErrorResponse(conn, err.Error())
        return
    }
    
    // Получаем базу данных
    db, err := n.Storage.GetDatabase(syncData.Database)
    if err != nil {
        n.sendErrorResponse(conn, err.Error())
        return
    }
    
    // Получаем коллекцию
    coll, err := db.GetCollection(syncData.Collection)
    if err != nil {
        n.sendErrorResponse(conn, err.Error())
        return
    }
    
    // Получаем все документы
    docs := coll.GetAllDocuments()
    
    // Фильтруем по времени, если указано
    if syncData.Since > 0 {
        filtered := make([]*storage.Document, 0)
        for _, doc := range docs {
            if doc.UpdatedAt > syncData.Since {
                filtered = append(filtered, doc)
            }
        }
        docs = filtered
    }
    
    duration := time.Now().UnixMilli() - startTime
    
    // Формируем ответ
    response := map[string]interface{}{
        "status":           "success",
        "docs":             docs,
        "count":            len(docs),
        "node_id":          n.ID,
        "request_id":       syncData.RequestID,
        "duration_ms":      duration,
        "timestamp":        time.Now().UnixMilli(),
        "sync_duration_ms": duration,
    }
    
    // Отправляем ответ
    conn.SetWriteDeadline(time.Now().Add(30 * time.Second))
    encoder := json.NewEncoder(conn)
    if err := encoder.Encode(response); err != nil && n.logger != nil {
        n.logger.Error(fmt.Sprintf("Failed to send sync response: %v", err))
    }
    
    if n.logger != nil {
        n.logger.Info(fmt.Sprintf("Node %s synced %d documents from %s.%s (took %d ms)", 
            n.ID, len(docs), syncData.Database, syncData.Collection, duration))
    }
}

// handleHeartbeatRequest обрабатывает heartbeat-запрос.
// Подтверждает, что узел жив, и возвращает информацию о статусе.
//
// ОТВЕТ СОДЕРЖИТ:
//   - status: "alive" - подтверждение жизни
//   - node_id: ID текущего узла
//   - timestamp: время обработки
//   - uptime_ms: время работы узла в миллисекундах
func (n *Node) handleHeartbeatRequest(req NodeRequest, conn net.Conn) {
    // Защита от паники
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Heartbeat handler panicked: %v\n%s", r, debug.Stack()))
            }
        }
    }()
    
    // Обновляем время последнего контакта
    n.lastSeen.Store(time.Now().UnixMilli())
    
    // Формируем ответ
    response := map[string]interface{}{
        "status":     "alive",
        "node_id":    n.ID,
        "timestamp":  time.Now().UnixMilli(),
        "uptime_ms":  time.Now().UnixMilli() - n.startedAt,
    }
    
    // Отправляем ответ
    conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
    encoder := json.NewEncoder(conn)
    encoder.Encode(response)
}

// handleStatusSyncRequest обрабатывает запрос на синхронизацию статуса кластера.
// Обновляет информацию о лидере и термине через координатор.
//
// ПОРЯДОК ОБРАБОТКИ:
// 1. Декодирование данных статуса
// 2. Передача данных координатору
// 3. Формирование ответа с текущим статусом
func (n *Node) handleStatusSyncRequest(req NodeRequest, conn net.Conn) {
    // Защита от паники с отправкой ошибки
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Status sync handler panicked: %v\n%s", r, debug.Stack()))
            }
            n.sendErrorResponse(conn, "Internal server error")
        }
    }()
    
    // Структура данных статуса
    var syncStatus struct {
        LeaderID    string `json:"leader_id"`    // ID лидера
        Term        uint64 `json:"term"`         // Текущий термин Raft
        ClusterSize int    `json:"cluster_size"` // Размер кластера
    }
    
    if err := json.Unmarshal(req.Data, &syncStatus); err != nil {
        n.sendErrorResponse(conn, err.Error())
        return
    }
    
    // Передаём статус координатору
    if n.coordinator != nil {
        n.coordinator.HandleStatusSync(syncStatus.LeaderID, syncStatus.Term, syncStatus.ClusterSize)
    }
    
    // Формируем ответ
    response := map[string]interface{}{
        "status":    "synced",
        "node_id":   n.ID,
        "term":      n.coordinator.GetCurrentTerm(),
        "is_leader": n.coordinator.IsLeader(),
        "timestamp": time.Now().UnixMilli(),
    }
    
    // Отправляем ответ
    conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
    encoder := json.NewEncoder(conn)
    encoder.Encode(response)
}

// sendErrorResponse отправляет ответ об ошибке клиенту.
// Используется для единообразного форматирования ошибок.
//
// Параметры:
//   - conn: TCP-соединение
//   - errMsg: текст ошибки
func (n *Node) sendErrorResponse(conn net.Conn, errMsg string) {
    response := map[string]interface{}{
        "status":    "error",
        "error":     errMsg,
        "node_id":   n.ID,
        "timestamp": time.Now().UnixMilli(),
    }
    
    conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
    encoder := json.NewEncoder(conn)
    encoder.Encode(response)
}

// =============================================================================
// HEARTBEAT И МОНИТОРИНГ
// =============================================================================

// heartbeatLoop периодически отправляет heartbeat в кластер.
// Использует координатор для отправки heartbeat-сообщений.
//
// ИНТЕРВАЛ: 5 секунд
// УСЛОВИЕ: запускается только если координатор доступен
func (n *Node) heartbeatLoop() {
    // Защита от паники с автоматическим перезапуском
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Heartbeat loop panicked: %v\n%s", r, debug.Stack()))
            }
            time.Sleep(5 * time.Second)
            if n.panicRecoveryMgr != nil {
                n.startRecoverableRoutine("HeartbeatLoop", n.heartbeatLoop)
            } else {
                SafeGoWithLogger(n.heartbeatLoop, n.logger, "HeartbeatLoop")
            }
        }
    }()
    
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-n.stopChan:
            return
        case <-ticker.C:
            // Отправляем heartbeat через координатор
            if n.coordinator != nil {
                n.coordinator.SendHeartbeat(n.ID)
                n.lastSeen.Store(time.Now().UnixMilli())
                if n.logger != nil {
                    n.logger.Debug(fmt.Sprintf("Node %s sent heartbeat at %s", n.ID, n.GetLastSeenStr()))
                }
            }
        }
    }
}

// connectionHealthMonitor периодически проверяет здоровье активных соединений.
// Закрывает "мёртвые" соединения и удаляет их из пула.
//
// ИНТЕРВАЛ: 30 секунд
// МЕТОД ПРОВЕРКИ: чтение 1 байта с таймаутом 1 секунда
func (n *Node) connectionHealthMonitor() {
    // Защита от паники с перезапуском
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Connection health monitor panicked: %v\n%s", r, debug.Stack()))
            }
            SafeGoWithLogger(n.connectionHealthMonitor, n.logger, "ConnectionHealthMonitor")
        }
    }()
    
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-n.stopChan:
            return
        case <-ticker.C:
            n.cleanupStaleConnections()
        }
    }
}

// cleanupStaleConnections закрывает "зависшие" соединения.
// Проверяет каждое соединение в пуле и удаляет недоступные.
func (n *Node) cleanupStaleConnections() {
    // Защита от паники
    defer func() {
        if r := recover(); r != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Cleanup connections panicked: %v", r))
            }
        }
    }()
    
    n.connPool.Range(func(key, value interface{}) bool {
        if conn, ok := value.(net.Conn); ok {
            // Проверяем соединение с таймаутом 1 секунда
            conn.SetReadDeadline(time.Now().Add(1 * time.Second))
            buf := make([]byte, 1)
            _, err := conn.Read(buf)
            if err != nil {
                // Соединение недоступно - закрываем
                conn.Close()
                n.connPool.Delete(key)
                if n.logger != nil {
                    n.logger.Debug(fmt.Sprintf("Cleaned up stale connection for %v", key))
                }
            }
        }
        return true
    })
}

// =============================================================================
// УПРАВЛЕНИЕ СТАТУСОМ УЗЛА
// =============================================================================

// GetNodeStatus возвращает текущий статус узла.
// Потокобезопасный доступ через атомарную операцию.
func (n *Node) GetNodeStatus() NodeStatus {
    return NodeStatus(n.Status.Load())
}

// IsActive проверяет, активен ли узел.
// Возвращает true только для статуса StatusActive.
func (n *Node) IsActive() bool {
    return NodeStatus(n.Status.Load()) == StatusActive
}

// SetStatus устанавливает новый статус узла.
// При изменении статуса уведомляет координатор (если узел является лидером).
//
// ПОТОКОВАЯ БЕЗОПАСНОСТЬ:
// - Использует мьютекс для предотвращения гонок
// - Атомарно обновляет статус через Store
//
// Параметры:
//   - status: новый статус узла
//
// Возвращает: ошибку, если не удалось обновить статус через координатор
func (n *Node) SetStatus(status NodeStatus) error {
    n.mu.Lock()
    defer n.mu.Unlock()
    
    oldStatus := n.Status.Load()
    if oldStatus == int32(status) {
        return nil // Статус не изменился
    }
    
    // Если узел является лидером, обновляем статус через координатор
    if n.coordinator != nil && n.coordinator.IsLeader() {
        if err := n.coordinator.UpdateNodeStatus(n.ID, status); err != nil {
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Failed to update node status via Raft: %v", err))
            }
            return err
        }
    }
    
    // Обновляем статус
    n.Status.Store(int32(status))
    
    if n.logger != nil {
        n.logger.Info(fmt.Sprintf("Node %s status changed from %d to %d at %s", 
            n.ID, oldStatus, status, time.Now().Format("2006-01-02 15:04:05.000")))
    }
    
    return nil
}

// =============================================================================
// УПРАВЛЕНИЕ КЛАСТЕРОМ
// =============================================================================

// SetCoordinator устанавливает координатор для узла.
// Также обновляет время присоединения к кластеру.
//
// Параметры:
//   - coord: Raft-координатор
func (n *Node) SetCoordinator(coord *RaftCoordinator) {
    n.coordinator = coord
    now := time.Now().UnixMilli()
    n.joinedAt.Store(now)
    
    if n.logger != nil {
        n.logger.Info(fmt.Sprintf("Node %s joined cluster at %s", n.ID, n.GetJoinedAtStr()))
    }
}

// JoinCluster присоединяет узел к кластеру.
// Регистрирует узел в координаторе и устанавливает активный статус.
//
// ПОРЯДОК ДЕЙСТВИЙ:
// 1. Проверка, что узел ещё не присоединён
// 2. Установка координатора
// 3. Регистрация узла в координаторе
// 4. Установка статуса Active
//
// Параметры:
//   - coord: Raft-координатор
//
// Возвращает: ошибку, если присоединение не удалось
func (n *Node) JoinCluster(coord *RaftCoordinator) error {
    if n.coordinator != nil {
        return fmt.Errorf("node already joined to cluster")
    }
    
    n.SetCoordinator(coord)
    
    if err := coord.RegisterNode(n); err != nil {
        return fmt.Errorf("failed to register node: %v", err)
    }
    
    if err := n.SetStatus(StatusActive); err != nil {
        return fmt.Errorf("failed to set active status: %v", err)
    }
    
    if n.logger != nil {
        n.logger.Info(fmt.Sprintf("Node %s successfully joined cluster at %s", n.ID, n.GetJoinedAtStr()))
    }
    
    return nil
}

// LeaveCluster отключает узел от кластера.
// Устанавливает оффлайн-статус и удаляет узел из координатора.
//
// ПОРЯДОК ДЕЙСТВИЙ:
// 1. Установка статуса Offline
// 2. Удаление узла из координатора
// 3. Очистка координатора и времени присоединения
//
// Возвращает: ошибку, если отключение не удалось
func (n *Node) LeaveCluster() error {
    if n.coordinator == nil {
        return fmt.Errorf("node not in cluster")
    }
    
    if err := n.SetStatus(StatusOffline); err != nil {
        n.logger.Warn(fmt.Sprintf("Failed to set offline status: %v", err))
    }
    
    if err := n.coordinator.RemoveNode(n.ID); err != nil {
        n.logger.Warn(fmt.Sprintf("Failed to remove node from coordinator: %v", err))
    }
    
    n.coordinator = nil
    n.joinedAt.Store(0)
    
    if n.logger != nil {
        n.logger.Info(fmt.Sprintf("Node %s left cluster at %s", n.ID, time.Now().Format("2006-01-02 15:04:05.000")))
    }
    
    return nil
}

// =============================================================================
// ИНФОРМАЦИЯ О ВРЕМЕНИ
// =============================================================================

// GetLastSeen возвращает время последнего контакта в миллисекундах.
func (n *Node) GetLastSeen() int64 {
    return n.lastSeen.Load()
}

// GetLastSeenStr возвращает строковое представление времени последнего контакта.
func (n *Node) GetLastSeenStr() string {
    lastSeen := n.lastSeen.Load()
    if lastSeen == 0 {
        return "never"
    }
    return time.UnixMilli(lastSeen).Format("2006-01-02 15:04:05.000")
}

// GetJoinedAt возвращает время присоединения к кластеру в миллисекундах.
func (n *Node) GetJoinedAt() int64 {
    return n.joinedAt.Load()
}

// GetJoinedAtStr возвращает строковое представление времени присоединения.
func (n *Node) GetJoinedAtStr() string {
    joinedAt := n.joinedAt.Load()
    if joinedAt == 0 {
        return "not joined"
    }
    return time.UnixMilli(joinedAt).Format("2006-01-02 15:04:05.000")
}

// GetStartedAt возвращает время запуска TCP-сервера в миллисекундах.
func (n *Node) GetStartedAt() int64 {
    return n.startedAt
}

// GetStartedAtStr возвращает строковое представление времени запуска.
func (n *Node) GetStartedAtStr() string {
    return time.UnixMilli(n.startedAt).Format("2006-01-02 15:04:05.000")
}

// GetCreatedAt возвращает время создания узла в миллисекундах.
func (n *Node) GetCreatedAt() int64 {
    return n.createdAt
}

// GetCreatedAtStr возвращает строковое представление времени создания.
func (n *Node) GetCreatedAtStr() string {
    return time.UnixMilli(n.createdAt).Format("2006-01-02 15:04:05.000")
}

// GetUptime возвращает продолжительность работы узла.
func (n *Node) GetUptime() time.Duration {
    if n.startedAt == 0 {
        return 0
    }
    return time.Duration(time.Now().UnixMilli()-n.startedAt) * time.Millisecond
}

// GetAddress возвращает адрес узла в формате "IP:PORT".
func (n *Node) GetAddress() string {
    return fmt.Sprintf("%s:%d", n.IP, n.Port)
}

// =============================================================================
// СТАТИСТИКА
// =============================================================================

// GetStats возвращает полную статистику узла.
// Включает информацию о статусе, времени, нагрузке и компонентах.
func (n *Node) GetStats() map[string]interface{} {
    stats := map[string]interface{}{
        "id":            n.ID,
        "ip":            n.IP,
        "port":          n.Port,
        "status":        n.GetNodeStatus(),
        "created_at":    n.GetCreatedAtStr(),
        "started_at":    n.GetStartedAtStr(),
        "joined_at":     n.GetJoinedAtStr(),
        "last_seen":     n.GetLastSeenStr(),
        "uptime":        n.GetUptime().String(),
        "request_count": n.requestCount.Load(),
        "bytes_rx":      n.bytesRx.Load(),
        "bytes_tx":      n.bytesTx.Load(),
    }
    
    // Добавляем статистику пула воркеров, если доступен
    if n.workerPool != nil {
        stats["worker_pool"] = n.workerPool.GetStats()
    }
    
    // Добавляем статистику репликации, если доступна
    if n.replicator != nil {
        stats["replication"] = n.replicator.GetStats()
    }
    
    // Добавляем информацию о восстанавливаемых горутинах
    if n.panicRecoveryMgr != nil {
        n.routinesMu.RLock()
        routineStatus := make(map[string]bool)
        for name, routine := range n.recoverableRoutines {
            routineStatus[name] = routine.IsRunning()
        }
        n.routinesMu.RUnlock()
        stats["recoverable_routines"] = routineStatus
        stats["panic_recovery_stats"] = n.panicRecoveryMgr.GetStats()
    }
    
    return stats
}

// GetWorkerPoolStats возвращает статистику пула воркеров.
func (n *Node) GetWorkerPoolStats() map[string]interface{} {
    if n.workerPool == nil {
        return map[string]interface{}{"enabled": false}
    }
    return n.workerPool.GetStats()
}

// GetReplicationStats возвращает статистику репликации.
func (n *Node) GetReplicationStats() map[string]interface{} {
    if n.replicator == nil {
        return map[string]interface{}{"enabled": false}
    }
    return n.replicator.GetStats()
}

// GetPanicRecoveryStats возвращает статистику механизма восстановления.
func (n *Node) GetPanicRecoveryStats() map[string]interface{} {
    if n.panicRecoveryMgr == nil {
        return map[string]interface{}{"enabled": false}
    }
    return n.panicRecoveryMgr.GetStats()
}

// =============================================================================
// РЕПЛИКАЦИЯ ДОКУМЕНТОВ
// =============================================================================

// generateReplicationID генерирует уникальный ID для операции репликации.
// Используется для отслеживания и дедупликации репликаций.
func generateReplicationID() string {
    bytes := make([]byte, 16)
    rand.Read(bytes)
    return base64.URLEncoding.EncodeToString(bytes)
}

// ReplicateDocument реплицирует документ на все активные узлы кластера.
//
// ПОРЯДОК ДЕЙСТВИЙ:
// 1. Получение списка активных узлов от координатора
// 2. Формирование данных репликации
// 3. Асинхронная отправка на каждый узел (через пул воркеров)
// 4. Ожидание завершения всех репликаций с таймаутом
// 5. Возврат результата (успех/частичный успех/ошибка)
//
// Параметры:
//   - database: имя базы данных
//   - collection: имя коллекции
//   - doc: документ для репликации
//
// Возвращает: ошибку, если репликация не удалась
func (n *Node) ReplicateDocument(database, collection string, doc *storage.Document) error {
    if n.coordinator == nil {
        if n.logger != nil {
            n.logger.Warn("No coordinator set, skipping replication")
        }
        return fmt.Errorf("no coordinator set")
    }
    
    // Получаем список активных узлов
    nodes := n.coordinator.GetActiveNodes()
    if len(nodes) <= 1 {
        if n.logger != nil {
            n.logger.Debug("No other nodes for replication")
        }
        return nil // Только текущий узел - репликация не требуется
    }
    
    // Формируем данные для репликации
    repData := struct {
        Database   string                 `json:"database"`
        Collection string                 `json:"collection"`
        Document   map[string]interface{} `json:"document"`
        SourceNode string                 `json:"source_node"`
        ReplicaID  string                 `json:"replica_id"`
    }{
        Database:   database,
        Collection: collection,
        Document:   doc.GetFields(),
        SourceNode: n.ID,
        ReplicaID:  generateReplicationID(),
    }
    
    data, err := json.Marshal(repData)
    if err != nil {
        return fmt.Errorf("failed to marshal replication data: %v", err)
    }
    
    startTime := time.Now().UnixMilli()
    
    // Отправляем репликацию на все узлы, кроме текущего
    var wg sync.WaitGroup
    var failedCount atomic.Int32
    var successCount atomic.Int32
    
    for _, nodeInfo := range nodes {
        if nodeInfo.ID == n.ID {
            continue // Пропускаем себя
        }
        
        wg.Add(1)
        targetNodeID := nodeInfo.ID
        targetAddress := fmt.Sprintf("%s:%d", nodeInfo.IP, nodeInfo.Port)
        
        // Создаём задачу для каждого узла
        taskID := fmt.Sprintf("replicate_%s_to_%s_%s", doc.ID, targetNodeID, repData.ReplicaID)
        err := n.workerPool.SubmitFunc(taskID, func() error {
            defer wg.Done()
            
            if n.replicator == nil {
                failedCount.Add(1)
                return fmt.Errorf("replicator not initialized")
            }
            
            // Выполняем репликацию
            err := n.replicator.Replicate(targetNodeID, targetAddress, data)
            if err != nil {
                failedCount.Add(1)
                if n.logger != nil {
                    n.logger.Error(fmt.Sprintf("Failed to replicate document %s to node %s after retries: %v", 
                        doc.ID, targetNodeID, err))
                }
                return err
            }
            
            successCount.Add(1)
            if n.logger != nil {
                n.logger.Debug(fmt.Sprintf("Successfully replicated document %s to node %s", doc.ID, targetNodeID))
            }
            return nil
        })
        
        if err != nil {
            wg.Done()
            failedCount.Add(1)
            if n.logger != nil {
                n.logger.Error(fmt.Sprintf("Failed to submit replication task for %s to %s: %v", 
                    doc.ID, targetNodeID, err))
            }
        }
    }
    
    // Ожидаем завершения всех репликаций с таймаутом 30 секунд
    done := make(chan struct{})
    go func() {
        wg.Wait()
        close(done)
    }()
    
    select {
    case <-done:
        duration := time.Now().UnixMilli() - startTime
        if n.logger != nil {
            n.logger.Info(fmt.Sprintf("Replicated document %s to %d/%d nodes (took %d ms)", 
                doc.ID, successCount.Load(), len(nodes)-1, duration))
        }
    case <-time.After(30 * time.Second):
        if n.logger != nil {
            n.logger.Warn(fmt.Sprintf("Replication timeout for document %s after %d ms", doc.ID, time.Now().UnixMilli()-startTime))
        }
        return fmt.Errorf("replication timeout")
    }
    
    // Проверяем результат
    if failedCount.Load() > 0 {
        return fmt.Errorf("replication partially failed: %d of %d nodes failed", failedCount.Load(), len(nodes)-1)
    }
    
    return nil
}

// =============================================================================
// УПРАВЛЕНИЕ PANIC RECOVERY
// =============================================================================

// SetPanicRecoveryManager устанавливает менеджер восстановления после паники.
// Позволяет включить расширенный механизм восстановления после создания узла.
//
// Параметры:
//   - mgr: менеджер восстановления
func (n *Node) SetPanicRecoveryManager(mgr *PanicRecoveryManager) {
    n.panicRecoveryMgr = mgr
    if n.logger != nil {
        n.logger.Debug("Panic recovery manager set for node")
    }
}

// =============================================================================
// МЕТОДЫ ДЛЯ GOSSIP И SELF-HEALING
// =============================================================================

// GetGossipManager возвращает gossip менеджер узла
func (n *Node) GetGossipManager() *GossipManager {
    if n.coordinator != nil {
        return n.coordinator.GetGossipManager()
    }
    return nil
}

// GetSelfHealingManager возвращает менеджер самоисцеления узла
func (n *Node) GetSelfHealingManager() *SelfHealingManager {
    if n.coordinator != nil {
        return n.coordinator.GetSelfHealingManager()
    }
    return nil
}

// GetDynamicConfigManager возвращает менеджер динамической конфигурации узла
func (n *Node) GetDynamicConfigManager() *config.DynamicConfigManager {
    if n.coordinator != nil {
        return n.coordinator.GetDynamicConfigManager()
    }
    return nil
}

// =============================================================================
// ОСТАНОВКА УЗЛА
// =============================================================================

// Stop останавливает узел и все его компоненты.
// Выполняет корректное завершение всех горутин и освобождение ресурсов.
//
// ПОРЯДОК ОСТАНОВКИ:
// 1. Обновление статуса на Offline
// 2. Остановка всех восстанавливаемых горутин
// 3. Закрытие канала stopChan (сигнал для всех горутин)
// 4. Остановка пула воркеров
// 5. Закрытие репликатора
// 6. Закрытие всех активных TCP-соединений
// 7. Логирование остановки
func (n *Node) Stop() {
    // Обновляем статус перед остановкой
    if n.coordinator != nil && n.coordinator.IsLeader() {
        n.SetStatus(StatusOffline)
    }
    n.Status.Store(int32(StatusOffline))
    n.stoppedAt = time.Now().UnixMilli()
    
    // Останавливаем все восстанавливаемые горутины
    n.routinesMu.RLock()
    for name, routine := range n.recoverableRoutines {
        routine.Stop()
        if n.logger != nil {
            n.logger.Debug(fmt.Sprintf("Stopped recoverable routine: %s", name))
        }
    }
    n.routinesMu.RUnlock()
    
    // Закрываем канал остановки
    close(n.stopChan)
    
    // Останавливаем пул воркеров
    if n.workerPool != nil {
        n.workerPool.Stop()
    }
    
    // Закрываем репликатор
    if n.replicator != nil {
        n.replicator.Close()
    }
    
    // Закрываем все активные соединения
    n.connPool.Range(func(key, value interface{}) bool {
        if conn, ok := value.(net.Conn); ok {
            conn.Close()
        }
        return true
    })
    
    if n.logger != nil {
        n.logger.Info(fmt.Sprintf("Node %s stopped at %s", n.ID, time.UnixMilli(n.stoppedAt).Format("2006-01-02 15:04:05.000")))
    }
}