futriix/internal/cluster/raft_coordinator.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/raft_coordinator.go
// Назначение: Реализация координатора распределённого кластера на основе Raft консенсус-алгоритма.
//
// КРИТИЧЕСКОЕ ИСПРАВЛЕНИЕ (nil pointer dereference в RegisterNode):
//   Ранее поле coord.fsm (*RaftFSM) и его state (*RaftClusterState) нигде
//   не инициализировались в NewRaftCoordinator. Первый же вызов
//   RegisterNode (или любого метода, читающего rc.fsm.state) приводил
//   к панике SIGSEGV по адресу 0x0 на rc.fsm.state.mu.Lock().
//
//   Также не инициализировались coord.logStore и coord.stableStore
//   (*InmemStore), которые используются в monitorLeadership через
//   rc.stableStore.Set(...).
//
//   Теперь в NewRaftCoordinator создаются:
//     - logStore   = NewInmemStore(...)
//     - stableStore= NewInmemStore(...)
//     - fsm        = &RaftFSM{state: &RaftClusterState{...}}
//   до ветвления singleNodeMode / setupClusterMode.
//
//   Дополнительно добавлены defensive-проверки на nil в публичных
//   методах (RegisterNode, RemoveNode, GetAllNodes, GetActiveNodes,
//   GetNodeByID, GetLeader, SendHeartbeat, UpdateNodeStatus,
//   SetReplicationFactor, HandleStatusSync), чтобы подобные ошибки
//   в будущем давали внятную ошибку, а не панику.
//
// ДОБАВЛЕНО (2026-10, выборы лидера SAGA):
//   - Тип SagaLeaderBridge реализует storage.SagaRaftAccessor.
//     Он подписывается на изменения лидерства Raft (через monitorLeadership)
//     и уведомляет подписчиков.
//   - RaftCoordinator.sagaLeaderBridge создаётся в NewRaftCoordinator.
//   - В monitorLeadership при смене лидерства вызывается
//     coord.sagaLeaderBridge.NotifyLeadership(...), что обновляет
//     состояние bridge (для storage.SagaLeaderElector, если он где-то
//     используется).
//   - В Stop() останавливаем bridge.
//   - GetSagaLeaderBridge() — публичный геттер.
//
// ИСПРАВЛЕНО (компиляция):
//   Ранее NewSagaManagerWithRaftForCoordinator пыталась вернуть
//   *storage.SagaManager как *cluster.SagaManager (локальный тип).
//   Это конфликт двух разных типов с одинаковым именем.
//   Теперь:
//     - coord.sagaManager по-прежнему *cluster.SagaManager
//       (создаётся через NewSagaManager(logger));
//     - bridge остаётся отдельным компонентом (storage.SagaRaftAccessor),
//       который при необходимости можно получить через GetSagaLeaderBridge();
//     - вызов storage.NewSagaManagerWithRaft из NewRaftCoordinator убран.
//   Все публичные методы *cluster.SagaManager (GetOrchestrator,
//   GetMetrics, GetActiveSagas, Stop) сохранены — они используются
//   в repl.go и main.go.
//
// ДОБАВЛЕНО (2026-10, кластерный SagaManager):
//   - Метод GetAllSagas() у *SagaManager — возвращает все SAGA
//     транзакции (активные и завершённые). Используется REPL-командой
//     "saga list --all". Возвращает локальный тип cluster.SagaTransaction.

package cluster

import (
    "encoding/binary"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "net"
    "os"
    "path/filepath"
    "runtime"
    "sort"
    "sync"
    "sync/atomic"
    "syscall"
    "time"

    "github.com/hashicorp/raft"
    "futriis/internal/config"
    "futriis/internal/log"
    "futriis/internal/migration"
    "futriis/internal/storage"
)

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

// isTemporaryNetError проверяет, является ли ошибка временной (для Linux и illumos).
// На illumos (OpenIndiana) accept может возвращать EINTR, EAGAIN, EWOULDBLOCK.
func isTemporaryNetError(err error) bool {
    if err == nil {
        return false
    }
    // Стандартная проверка через net.Error
    if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
        return true
    }
    // Проверка syscall-ошибок для illumos/Linux
    if errors.Is(err, syscall.EINTR) ||
        errors.Is(err, syscall.EAGAIN) ||
        errors.Is(err, syscall.EWOULDBLOCK) {
        return true
    }
    return false
}

// isClockSkewAcceptable проверяет, что рассинхронизация часов в пределах допустимого.
// Используется для защиты от last-write-wins на основе рассинхронизированных часов.
func isClockSkewAcceptable(remoteTS, localTS int64, maxSkewMs int64) bool {
    diff := remoteTS - localTS
    if diff < 0 {
        diff = -diff
    }
    return diff <= maxSkewMs
}

// monotonicTimestamp возвращает монотонную метку времени, устойчивую к переводам часов.
// На illumos time.Now() использует CLOCK_MONOTONIC для монотонной части.
func monotonicTimestamp() int64 {
    return time.Now().UnixMilli()
}

// =============================================================================
// ИНТЕРФЕЙСЫ И БАЗОВЫЕ ТИПЫ (используются из node.go)
// =============================================================================

// LoggerInterface, NodeStatus, StatusOffline, StatusActive, StatusSyncing, StatusFailed,
// Node, NodeInfo - определены в node.go

// =============================================================================
// ЛОГИЧЕСКИЕ ЧАСЫ (LAMPORT CLOCK) ДЛЯ EVENTUAL CONSISTENCY
// =============================================================================

// LamportClock реализует логические часы Лампорта для упорядочивания событий
// без зависимости от физических часов узлов.
type LamportClock struct {
    counter atomic.Uint64
    nodeID  string
}

// NewLamportClock создаёт новые логические часы.
func NewLamportClock(nodeID string) *LamportClock {
    return &LamportClock{nodeID: nodeID}
}

// Tick увеличивает счётчик при локальном событии.
func (lc *LamportClock) Tick() uint64 {
    return lc.counter.Add(1)
}

// Observe обновляет счётчик при получении удалённого события.
// Правило Лампорта: counter = max(counter, remote) + 1
func (lc *LamportClock) Observe(remote uint64) uint64 {
    for {
        current := lc.counter.Load()
        var next uint64
        if remote > current {
            next = remote + 1
        } else {
            next = current + 1
        }
        if lc.counter.CompareAndSwap(current, next) {
            return next
        }
    }
}

// Value возвращает текущее значение.
func (lc *LamportClock) Value() uint64 {
    return lc.counter.Load()
}

// =============================================================================
// SAGA LEADER BRIDGE — РЕАЛИЗАЦИЯ storage.SagaRaftAccessor
// =============================================================================

// SagaLeaderBridge — мост между Raft-координатором и оркестратором SAGA.
//
// Реализует интерфейс storage.SagaRaftAccessor:
//   - IsSagaLeader() — сообщает, является ли текущий узел лидером Raft.
//   - GetSagaLeaderID() — возвращает ID текущего лидера.
//   - GetSagaCurrentTerm() — возвращает текущий терм Raft.
//   - RegisterSagaLeaderObserver(id, cb) — регистрирует наблюдателя,
//     который будет уведомлён при смене лидерства.
//
// Мост используется storage.SagaLeaderElector, который опрашивает
// его и обновляет своё состояние (isLeader, leaderID).
//
// Все методы потокобезопасны.
type SagaLeaderBridge struct {
    coordinator *RaftCoordinator

    mu       sync.RWMutex
    isLeader bool
    leaderID string
    term     uint64

    observersMu sync.RWMutex
    observers   map[string]func(isLeader bool, leaderID string)

    stopChan chan struct{}
    wg       sync.WaitGroup
}

// NewSagaLeaderBridge создаёт новый мост.
func NewSagaLeaderBridge(coord *RaftCoordinator) *SagaLeaderBridge {
    b := &SagaLeaderBridge{
        coordinator: coord,
        observers:   make(map[string]func(isLeader bool, leaderID string)),
        stopChan:    make(chan struct{}),
    }
    return b
}

// Start запускает фоновый опрос состояния Raft.
func (b *SagaLeaderBridge) Start() {
    b.wg.Add(1)
    go b.pollLoop()
}

// pollLoop периодически опрашивает Raft и обновляет состояние.
func (b *SagaLeaderBridge) pollLoop() {
    defer b.wg.Done()
    ticker := time.NewTicker(2 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            b.refreshFromRaft()
        case <-b.stopChan:
            return
        }
    }
}

// refreshFromRaft читает состояние Raft и уведомляет наблюдателей,
// если что-то изменилось.
func (b *SagaLeaderBridge) refreshFromRaft() {
    if b.coordinator == nil {
        return
    }

    isLeader := b.coordinator.IsLeader()

    leaderID := ""
    if leader := b.coordinator.GetLeader(); leader != nil {
        leaderID = leader.ID
    }

    term := b.coordinator.GetCurrentTerm()

    b.NotifyLeadership(isLeader, leaderID, term)
}

// NotifyLeadership вызывается извне (например, monitorLeadership)
// для немедленного обновления состояния.
func (b *SagaLeaderBridge) NotifyLeadership(isLeader bool, leaderID string, term uint64) {
    b.mu.Lock()
    oldIsLeader := b.isLeader
    oldLeaderID := b.leaderID
    b.isLeader = isLeader
    b.leaderID = leaderID
    b.term = term
    b.mu.Unlock()

    if oldIsLeader != isLeader || oldLeaderID != leaderID {
        b.observersMu.RLock()
        obs := make([]func(isLeader bool, leaderID string), 0, len(b.observers))
        for _, cb := range b.observers {
            obs = append(obs, cb)
        }
        b.observersMu.RUnlock()

        for _, cb := range obs {
            func(cb func(isLeader bool, leaderID string)) {
                defer func() {
                    if r := recover(); r != nil {
                        // Не позволяем панике наблюдателя уронить процесс.
                    }
                }()
                cb(isLeader, leaderID)
            }(cb)
        }
    }
}

// IsSagaLeader — реализация storage.SagaRaftAccessor.
func (b *SagaLeaderBridge) IsSagaLeader() bool {
    b.mu.RLock()
    defer b.mu.RUnlock()
    return b.isLeader
}

// GetSagaLeaderID — реализация storage.SagaRaftAccessor.
func (b *SagaLeaderBridge) GetSagaLeaderID() string {
    b.mu.RLock()
    defer b.mu.RUnlock()
    return b.leaderID
}

// GetSagaCurrentTerm — реализация storage.SagaRaftAccessor.
func (b *SagaLeaderBridge) GetSagaCurrentTerm() uint64 {
    b.mu.RLock()
    defer b.mu.RUnlock()
    return b.term
}

// RegisterSagaLeaderObserver — реализация storage.SagaRaftAccessor.
func (b *SagaLeaderBridge) RegisterSagaLeaderObserver(id string, cb func(isLeader bool, leaderID string)) {
    b.observersMu.Lock()
    defer b.observersMu.Unlock()
    b.observers[id] = cb
}

// UnregisterSagaLeaderObserver снимает регистрацию наблюдателя.
func (b *SagaLeaderBridge) UnregisterSagaLeaderObserver(id string) {
    b.observersMu.Lock()
    defer b.observersMu.Unlock()
    delete(b.observers, id)
}

// Stop останавливает мост.
func (b *SagaLeaderBridge) Stop() {
    close(b.stopChan)
    b.wg.Wait()
}

// =============================================================================
// ДИАПАЗОННЫЕ ШАРДЫ (RANGE SHARDS)
// =============================================================================

// RangeShard представляет шард на основе диапазона ключей.
type RangeShard struct {
    ID             string   `json:"id"`
    Name           string   `json:"name"`
    StartKey       string   `json:"start_key"`
    EndKey         string   `json:"end_key"`
    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"`
    IsSplitting    bool     `json:"is_splitting"`
    IsMerging      bool     `json:"is_merging"`
}

// RangeShardManager управляет диапазонными шардами с динамическим сплитом/мерджем.
type RangeShardManager struct {
    shardsPtr           atomic.Value
    sortedShards        atomic.Value
    mu                  sync.RWMutex
    logger              LoggerInterface
    shardSizeThreshold  int64
    shardCountThreshold int64
    minShardSize        int64
    rebalancing         atomic.Bool
    splitMgr            *DynamicSplitManager
    mergeMgr            *DynamicMergeManager
}

// DynamicSplitManager управляет динамическим разделением шардов.
type DynamicSplitManager struct {
    shardManager    *RangeShardManager
    logger          LoggerInterface
    stopChan        chan struct{}
    wg              sync.WaitGroup
    checkInterval   time.Duration
    mu              sync.RWMutex
    splittingShards map[string]bool
}

// DynamicMergeManager управляет динамическим объединением шардов.
type DynamicMergeManager struct {
    shardManager  *RangeShardManager
    logger        LoggerInterface
    stopChan      chan struct{}
    wg            sync.WaitGroup
    checkInterval time.Duration
    mu            sync.RWMutex
    mergingShards map[string]bool
}

// NewRangeShardManager создаёт новый менеджер диапазонных шардов.
func NewRangeShardManager(logger LoggerInterface) *RangeShardManager {
    rsm := &RangeShardManager{
        logger:              logger,
        shardSizeThreshold:  100 * 1024 * 1024,
        shardCountThreshold: 1000000,
        minShardSize:        10 * 1024 * 1024,
    }
    rsm.shardsPtr.Store(make(map[string]*RangeShard))
    rsm.sortedShards.Store(make([]*RangeShard, 0))

    rsm.splitMgr = NewDynamicSplitManager(rsm, logger)
    rsm.mergeMgr = NewDynamicMergeManager(rsm, logger)

    return rsm
}

// NewDynamicSplitManager создаёт менеджер разделения шардов.
func NewDynamicSplitManager(shardManager *RangeShardManager, logger LoggerInterface) *DynamicSplitManager {
    return &DynamicSplitManager{
        shardManager:    shardManager,
        logger:          logger,
        stopChan:        make(chan struct{}),
        checkInterval:   30 * time.Second,
        splittingShards: make(map[string]bool),
    }
}

// NewDynamicMergeManager создаёт менеджер объединения шардов.
func NewDynamicMergeManager(shardManager *RangeShardManager, logger LoggerInterface) *DynamicMergeManager {
    return &DynamicMergeManager{
        shardManager:  shardManager,
        logger:        logger,
        stopChan:      make(chan struct{}),
        checkInterval: 60 * time.Second,
        mergingShards: make(map[string]bool),
    }
}

// Start запускает мониторинг шардов.
func (rsm *RangeShardManager) Start() {
    if rsm.splitMgr != nil {
        go rsm.splitMgr.Start()
    }
    if rsm.mergeMgr != nil {
        go rsm.mergeMgr.Start()
    }
    if rsm.logger != nil {
        rsm.logger.Info("Range shard manager started")
    }
}

// Stop останавливает менеджер шардов.
func (rsm *RangeShardManager) Stop() {
    if rsm.splitMgr != nil {
        rsm.splitMgr.Stop()
    }
    if rsm.mergeMgr != nil {
        rsm.mergeMgr.Stop()
    }
    if rsm.logger != nil {
        rsm.logger.Info("Range shard manager stopped")
    }
}

// Start запускает мониторинг разделения.
func (dsm *DynamicSplitManager) Start() {
    dsm.wg.Add(1)
    go dsm.splitMonitor()
}

// Stop останавливает мониторинг.
func (dsm *DynamicSplitManager) Stop() {
    close(dsm.stopChan)
    dsm.wg.Wait()
}

// splitMonitor периодически проверяет шарды на необходимость разделения.
func (dsm *DynamicSplitManager) splitMonitor() {
    defer dsm.wg.Done()

    ticker := time.NewTicker(dsm.checkInterval)
    defer ticker.Stop()

    for {
        select {
        case <-dsm.stopChan:
            return
        case <-ticker.C:
            dsm.checkAndSplit()
        }
    }
}

// checkAndSplit проверяет и выполняет разделение шардов.
func (dsm *DynamicSplitManager) checkAndSplit() {
    shards := dsm.shardManager.GetAllShards()

    for _, shard := range shards {
        if shard.IsSplitting || shard.IsMerging {
            continue
        }

        if shard.SizeBytes > dsm.shardManager.shardSizeThreshold ||
            shard.DocumentCount > dsm.shardManager.shardCountThreshold {
            dsm.splitShard(shard)
        }
    }
}

// splitShard выполняет разделение шарда на два.
func (dsm *DynamicSplitManager) splitShard(shard *RangeShard) error {
    dsm.mu.Lock()
    if dsm.splittingShards[shard.ID] {
        dsm.mu.Unlock()
        return fmt.Errorf("shard %s is already splitting", shard.ID)
    }
    dsm.splittingShards[shard.ID] = true
    dsm.mu.Unlock()

    defer func() {
        dsm.mu.Lock()
        delete(dsm.splittingShards, shard.ID)
        dsm.mu.Unlock()
    }()

    if dsm.logger != nil {
        dsm.logger.Info(fmt.Sprintf("Splitting shard %s (size: %d bytes, docs: %d)",
            shard.Name, shard.SizeBytes, shard.DocumentCount))
    }

    splitKey := dsm.findSplitKey(shard)
    if splitKey == "" {
        return fmt.Errorf("failed to find split key for shard %s", shard.ID)
    }

    shard1 := &RangeShard{
        ID:             fmt.Sprintf("%s_left", shard.ID),
        Name:           fmt.Sprintf("%s_left", shard.Name),
        StartKey:       shard.StartKey,
        EndKey:         splitKey,
        Nodes:          shard.Nodes,
        LeaderNode:     shard.LeaderNode,
        Status:         "active",
        CreatedAt:      time.Now().UnixMilli(),
        UpdatedAt:      time.Now().UnixMilli(),
        LastRebalanced: time.Now().UnixMilli(),
        DocumentCount:  shard.DocumentCount / 2,
        SizeBytes:      shard.SizeBytes / 2,
        IsSplitting:    false,
        IsMerging:      false,
    }

    shard2 := &RangeShard{
        ID:             fmt.Sprintf("%s_right", shard.ID),
        Name:           fmt.Sprintf("%s_right", shard.Name),
        StartKey:       splitKey,
        EndKey:         shard.EndKey,
        Nodes:          shard.Nodes,
        LeaderNode:     shard.LeaderNode,
        Status:         "active",
        CreatedAt:      time.Now().UnixMilli(),
        UpdatedAt:      time.Now().UnixMilli(),
        LastRebalanced: time.Now().UnixMilli(),
        DocumentCount:  shard.DocumentCount / 2,
        SizeBytes:      shard.SizeBytes / 2,
        IsSplitting:    false,
        IsMerging:      false,
    }

    dsm.shardManager.mu.Lock()
    defer dsm.shardManager.mu.Unlock()

    oldShards := dsm.shardManager.loadShards()
    newShards := make(map[string]*RangeShard)
    for k, v := range oldShards {
        if k != shard.ID {
            newShards[k] = v
        }
    }
    newShards[shard1.ID] = shard1
    newShards[shard2.ID] = shard2

    dsm.shardManager.shardsPtr.Store(newShards)
    dsm.shardManager.updateSortedShards()

    if dsm.logger != nil {
        dsm.logger.Info(fmt.Sprintf("Shard %s split into %s and %s", shard.Name, shard1.Name, shard2.Name))
    }

    return nil
}

// findSplitKey находит ключ для разделения шарда.
func (dsm *DynamicSplitManager) findSplitKey(shard *RangeShard) string {
    start := []byte(shard.StartKey)
    end := []byte(shard.EndKey)

    if len(start) == 0 || len(end) == 0 {
        return ""
    }

    mid := make([]byte, len(start))
    for i := range start {
        if i < len(end) {
            mid[i] = (start[i] + end[i]) / 2
        } else {
            mid[i] = start[i]
        }
    }

    return string(mid)
}

// Start запускает мониторинг объединения.
func (dmm *DynamicMergeManager) Start() {
    dmm.wg.Add(1)
    go dmm.mergeMonitor()
}

// Stop останавливает мониторинг.
func (dmm *DynamicMergeManager) Stop() {
    close(dmm.stopChan)
    dmm.wg.Wait()
}

// mergeMonitor периодически проверяет шарды на возможность объединения.
func (dmm *DynamicMergeManager) mergeMonitor() {
    defer dmm.wg.Done()

    ticker := time.NewTicker(dmm.checkInterval)
    defer ticker.Stop()

    for {
        select {
        case <-dmm.stopChan:
            return
        case <-ticker.C:
            dmm.checkAndMerge()
        }
    }
}

// checkAndMerge проверяет и выполняет объединение шардов.
func (dmm *DynamicMergeManager) checkAndMerge() {
    shards := dmm.shardManager.GetSortedShards()

    for i := 0; i < len(shards)-1; i++ {
        shard1 := shards[i]
        shard2 := shards[i+1]

        if shard1.IsSplitting || shard1.IsMerging || shard2.IsSplitting || shard2.IsMerging {
            continue
        }

        if shard1.EndKey != shard2.StartKey {
            continue
        }

        totalSize := shard1.SizeBytes + shard2.SizeBytes
        totalDocs := shard1.DocumentCount + shard2.DocumentCount

        if totalSize < dmm.shardManager.minShardSize && totalDocs < dmm.shardManager.shardCountThreshold/10 {
            dmm.mergeShards(shard1, shard2)
        }
    }
}

// mergeShards объединяет два смежных шарда.
func (dmm *DynamicMergeManager) mergeShards(shard1, shard2 *RangeShard) error {
    dmm.mu.Lock()
    if dmm.mergingShards[shard1.ID] || dmm.mergingShards[shard2.ID] {
        dmm.mu.Unlock()
        return fmt.Errorf("shards are already merging")
    }
    dmm.mergingShards[shard1.ID] = true
    dmm.mergingShards[shard2.ID] = true
    dmm.mu.Unlock()

    defer func() {
        dmm.mu.Lock()
        delete(dmm.mergingShards, shard1.ID)
        delete(dmm.mergingShards, shard2.ID)
        dmm.mu.Unlock()
    }()

    if dmm.logger != nil {
        dmm.logger.Info(fmt.Sprintf("Merging shards %s and %s", shard1.Name, shard2.Name))
    }

    mergedShard := &RangeShard{
        ID:             fmt.Sprintf("%s_merged", shard1.ID),
        Name:           fmt.Sprintf("%s_merged", shard1.Name),
        StartKey:       shard1.StartKey,
        EndKey:         shard2.EndKey,
        Nodes:          shard1.Nodes,
        LeaderNode:     shard1.LeaderNode,
        Status:         "active",
        CreatedAt:      time.Now().UnixMilli(),
        UpdatedAt:      time.Now().UnixMilli(),
        LastRebalanced: time.Now().UnixMilli(),
        DocumentCount:  shard1.DocumentCount + shard2.DocumentCount,
        SizeBytes:      shard1.SizeBytes + shard2.SizeBytes,
        IsSplitting:    false,
        IsMerging:      false,
    }

    dmm.shardManager.mu.Lock()
    defer dmm.shardManager.mu.Unlock()

    oldShards := dmm.shardManager.loadShards()
    newShards := make(map[string]*RangeShard)
    for k, v := range oldShards {
        if k != shard1.ID && k != shard2.ID {
            newShards[k] = v
        }
    }
    newShards[mergedShard.ID] = mergedShard

    dmm.shardManager.shardsPtr.Store(newShards)
    dmm.shardManager.updateSortedShards()

    if dmm.logger != nil {
        dmm.logger.Info(fmt.Sprintf("Merged %s and %s into %s", shard1.Name, shard2.Name, mergedShard.Name))
    }

    return nil
}

// loadShards загружает карту шардов.
func (rsm *RangeShardManager) loadShards() map[string]*RangeShard {
    val := rsm.shardsPtr.Load()
    if val == nil {
        return make(map[string]*RangeShard)
    }
    return val.(map[string]*RangeShard)
}

// updateSortedShards обновляет отсортированный список шардов.
func (rsm *RangeShardManager) updateSortedShards() {
    shards := rsm.loadShards()
    sorted := make([]*RangeShard, 0, len(shards))
    for _, sh := range shards {
        sorted = append(sorted, sh)
    }
    sort.Slice(sorted, func(i, j int) bool {
        return sorted[i].StartKey < sorted[j].StartKey
    })
    rsm.sortedShards.Store(sorted)
}

// GetShard возвращает шард для ключа.
func (rsm *RangeShardManager) GetShard(key string) *RangeShard {
    shards := rsm.getSortedShards()
    for _, shard := range shards {
        if key >= shard.StartKey && (shard.EndKey == "" || key < shard.EndKey) {
            return shard
        }
    }
    return nil
}

// getSortedShards возвращает отсортированный список шардов.
func (rsm *RangeShardManager) getSortedShards() []*RangeShard {
    val := rsm.sortedShards.Load()
    if val == nil {
        return make([]*RangeShard, 0)
    }
    return val.([]*RangeShard)
}

// GetAllShards возвращает все шарды.
func (rsm *RangeShardManager) GetAllShards() []*RangeShard {
    shards := rsm.loadShards()
    result := make([]*RangeShard, 0, len(shards))
    for _, shard := range shards {
        result = append(result, shard)
    }
    return result
}

// GetSortedShards возвращает отсортированные шарды.
func (rsm *RangeShardManager) GetSortedShards() []*RangeShard {
    return rsm.getSortedShards()
}

// GetShardByID возвращает шард по ID.
func (rsm *RangeShardManager) GetShardByID(shardID string) *RangeShard {
    shards := rsm.loadShards()
    if shard, ok := shards[shardID]; ok {
        return shard
    }
    return nil
}

// AddNode добавляет узел в шарды.
func (rsm *RangeShardManager) AddNode(nodeID string) {
    rsm.Rebalance()
}

// RemoveNode удаляет узел из шардов.
func (rsm *RangeShardManager) RemoveNode(nodeID string) {
    rsm.Rebalance()
}

// Rebalance выполняет ребалансировку шардов.
func (rsm *RangeShardManager) Rebalance() error {
    if !rsm.rebalancing.CompareAndSwap(false, true) {
        return fmt.Errorf("rebalancing already in progress")
    }
    defer rsm.rebalancing.Store(false)

    if rsm.logger != nil {
        rsm.logger.Info("Starting range shard rebalancing...")
    }

    shards := rsm.GetAllShards()
    now := time.Now().UnixMilli()

    rsm.mu.Lock()
    defer rsm.mu.Unlock()

    oldShards := rsm.loadShards()
    newShards := make(map[string]*RangeShard)

    for id, shard := range oldShards {
        shardCopy := *shard
        shardCopy.LastRebalanced = now
        shardCopy.UpdatedAt = now
        newShards[id] = &shardCopy
    }

    rsm.shardsPtr.Store(newShards)
    rsm.updateSortedShards()

    if rsm.logger != nil {
        rsm.logger.Info(fmt.Sprintf("Range shard rebalancing completed: %d shards", len(shards)))
    }

    return nil
}
// =============================================================================
// INMEM STORE ДЛЯ RAFT (реализует raft.LogStore и raft.StableStore)
// =============================================================================

// InmemStore реализует хранилище для Raft в памяти
// Корректная обработка конфликтов логов при смене лидера.
type InmemStore struct {
    data       map[string][]byte
    mu         sync.RWMutex
    path       string
    firstIndex uint64
    lastIndex  uint64
}

// NewInmemStore создаёт новое in-memory хранилище
func NewInmemStore(path string) *InmemStore {
    return &InmemStore{
        data:       make(map[string][]byte),
        path:       path,
        firstIndex: 1,
        lastIndex:  0,
    }
}

// Get возвращает значение по ключу
func (s *InmemStore) Get(key []byte) ([]byte, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    val, ok := s.data[string(key)]
    if !ok {
        return nil, fmt.Errorf("key not found")
    }
    return val, nil
}

// Set сохраняет значение по ключу
func (s *InmemStore) Set(key, val []byte) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.data[string(key)] = val
    return nil
}

// FirstIndex возвращает первый индекс (для raft.LogStore)
func (s *InmemStore) FirstIndex() (uint64, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()

    if s.lastIndex == 0 {
        return 0, nil
    }
    return s.firstIndex, nil
}

// LastIndex возвращает последний индекс (для raft.LogStore)
func (s *InmemStore) LastIndex() (uint64, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    return s.lastIndex, nil
}

// GetLog возвращает лог по индексу (для raft.LogStore)
func (s *InmemStore) GetLog(index uint64, log *raft.Log) error {
    s.mu.RLock()
    defer s.mu.RUnlock()

    key := fmt.Sprintf("log_%d", index)
    val, ok := s.data[key]
    if !ok {
        return raft.ErrLogNotFound
    }

    return json.Unmarshal(val, log)
}

// StoreLog сохраняет лог (для raft.LogStore)
// При записи лога с индексом, который уже существует,
// проверяется конфликт. Если терм нового лога >= старого, заменяем.
// Если терм меньше — игнорируем (защита от отката).
func (s *InmemStore) StoreLog(log *raft.Log) error {
    return s.StoreLogs([]*raft.Log{log})
}

// StoreLogs сохраняет несколько логов (для raft.LogStore)
// Корректная обработка конфликтов при смене лидера.
// При получении лога с индексом, который уже есть, но с другим термом,
// удаляем все последующие логи (как требует Raft-спецификация).
func (s *InmemStore) StoreLogs(logs []*raft.Log) error {
    if len(logs) == 0 {
        return nil
    }

    s.mu.Lock()
    defer s.mu.Unlock()

    for _, log := range logs {
        key := fmt.Sprintf("log_%d", log.Index)

        // ПРОВЕРКА КОНФЛИКТА: если лог с таким индексом уже существует
        if existingData, ok := s.data[key]; ok {
            var existingLog raft.Log
            if err := json.Unmarshal(existingData, &existingLog); err == nil {
                if existingLog.Term == log.Term {
                    // Тот же терм — идемпотентная запись, пропускаем
                    continue
                }
                if existingLog.Term > log.Term {
                    // Существующий лог новее — игнорируем старый лог
                    // (защита от отката после смены лидера)
                    continue
                }
                // Новый лог новее — удаляем все последующие логи
                // (Raft-спецификация: при конфликте удаляем conflicting entries)
                s.deleteRangeLocked(log.Index+1, s.lastIndex)
            }
        }

        data, err := json.Marshal(log)
        if err != nil {
            return err
        }
        s.data[key] = data

        // Обновляем индексы
        if s.lastIndex == 0 || log.Index > s.lastIndex {
            s.lastIndex = log.Index
        }
        if log.Index < s.firstIndex || s.firstIndex == 0 {
            s.firstIndex = log.Index
        }
    }
    return nil
}

// deleteRangeLocked удаляет диапазон логов (вызывается под s.mu)
func (s *InmemStore) deleteRangeLocked(min, max uint64) {
    for i := min; i <= max; i++ {
        key := fmt.Sprintf("log_%d", i)
        delete(s.data, key)
    }
    // Обновляем lastIndex
    if max >= s.lastIndex && min <= s.lastIndex {
        if min > 0 {
            s.lastIndex = min - 1
        } else {
            s.lastIndex = 0
        }
    }
}

// DeleteRange удаляет диапазон логов (для raft.LogStore)
// Корректное обновление индексов.
func (s *InmemStore) DeleteRange(min, max uint64) error {
    s.mu.Lock()
    defer s.mu.Unlock()

    s.deleteRangeLocked(min, max)

    // Обновляем firstIndex
    if min <= s.firstIndex {
        s.firstIndex = max + 1
    }
    if s.firstIndex > s.lastIndex {
        s.firstIndex = s.lastIndex + 1
    }

    return nil
}

// SetConfiguration сохраняет конфигурацию (для raft.LogStore)
func (s *InmemStore) SetConfiguration(config raft.Configuration) error {
    s.mu.Lock()
    defer s.mu.Unlock()

    data, err := json.Marshal(config)
    if err != nil {
        return err
    }
    s.data["config"] = data
    return nil
}

// Configuration возвращает конфигурацию (для raft.LogStore)
func (s *InmemStore) Configuration() (raft.Configuration, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()

    val, ok := s.data["config"]
    if !ok {
        return raft.Configuration{}, nil
    }

    var config raft.Configuration
    if err := json.Unmarshal(val, &config); err != nil {
        return raft.Configuration{}, err
    }
    return config, nil
}

// =============================================================================
// МЕТОДЫ ДЛЯ raft.StableStore
// =============================================================================

// SetUint64 сохраняет uint64 значение (для raft.StableStore)
func (s *InmemStore) SetUint64(key []byte, val uint64) error {
    s.mu.Lock()
    defer s.mu.Unlock()

    buf := make([]byte, 8)
    binary.BigEndian.PutUint64(buf, val)
    s.data[string(key)] = buf
    return nil
}

// GetUint64 получает uint64 значение (для raft.StableStore)
func (s *InmemStore) GetUint64(key []byte) (uint64, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()

    val, ok := s.data[string(key)]
    if !ok {
        return 0, fmt.Errorf("key not found")
    }
    if len(val) != 8 {
        return 0, fmt.Errorf("invalid uint64 value")
    }
    return binary.BigEndian.Uint64(val), nil
}

// =============================================================================
// MULTI-RAFT
// =============================================================================

// MultiRaftManager управляет несколькими Raft-группами для параллельной записи.
type MultiRaftManager struct {
    raftGroups    sync.Map
    groupConfigs  sync.Map
    logger        LoggerInterface
    mu            sync.RWMutex
    stopChan      chan struct{}
    wg            sync.WaitGroup
    storage       *storage.Storage
    baseConfig    *raft.Config
    transport     *raft.NetworkTransport
    snapshotStore raft.SnapshotStore
    logStore      raft.LogStore
    stableStore   raft.StableStore
}

// MultiRaftGroupConfig конфигурация группы Raft.
type MultiRaftGroupConfig struct {
    GroupID   string
    ShardID   string
    Nodes     []string
    LeaderID  string
    Term      uint64
    CreatedAt int64
    UpdatedAt int64
}

// NewMultiRaftManager создаёт новый менеджер Multi-Raft.
func NewMultiRaftManager(storage *storage.Storage, logger LoggerInterface) *MultiRaftManager {
    return &MultiRaftManager{
        logger:   logger,
        stopChan: make(chan struct{}),
        storage:  storage,
    }
}

// GetOrCreateRaftGroup получает или создаёт группу Raft для шарда.
// Использует свободный порт, не конфликтует на illumos/Linux.
func (mrm *MultiRaftManager) GetOrCreateRaftGroup(shardID string, nodes []string) (*raft.Raft, error) {
    if val, ok := mrm.raftGroups.Load(shardID); ok {
        return val.(*raft.Raft), nil
    }

    mrm.mu.Lock()
    defer mrm.mu.Unlock()

    if val, ok := mrm.raftGroups.Load(shardID); ok {
        return val.(*raft.Raft), nil
    }

    groupID := fmt.Sprintf("shard_%s", shardID)

    raftConfig := raft.DefaultConfig()
    raftConfig.LocalID = raft.ServerID(groupID)
    raftConfig.HeartbeatTimeout = 1 * time.Second
    raftConfig.ElectionTimeout = 1 * time.Second
    raftConfig.CommitTimeout = 500 * time.Millisecond
    raftConfig.SnapshotInterval = 30 * time.Second
    raftConfig.SnapshotThreshold = 1000

    dataDir := filepath.Join("raft_data", groupID)
    if err := os.MkdirAll(dataDir, 0755); err != nil {
        return nil, fmt.Errorf("failed to create raft dir: %v", err)
    }

    logStore := NewInmemStore(filepath.Join(dataDir, "raft-log.json"))
    stableStore := NewInmemStore(filepath.Join(dataDir, "raft-stable.json"))
    snapshotStore, err := raft.NewFileSnapshotStore(dataDir, 3, os.Stderr)
    if err != nil {
        return nil, fmt.Errorf("failed to create snapshot store: %v", err)
    }

    fsm := &MultiRaftFSM{
        shardID: shardID,
        storage: mrm.storage,
        logger:  mrm.logger,
    }

    // Используем 127.0.0.1 с эфемерным портом для избежания конфликтов
    addr := "127.0.0.1:0"
    transport, err := raft.NewTCPTransport(addr, nil, 3, 10*time.Second, os.Stderr)
    if err != nil {
        return nil, fmt.Errorf("failed to create transport: %v", err)
    }

    r, err := raft.NewRaft(raftConfig, fsm, logStore, stableStore, snapshotStore, transport)
    if err != nil {
        return nil, fmt.Errorf("failed to create raft: %v", err)
    }

    if len(nodes) > 0 {
        servers := make([]raft.Server, len(nodes))
        for i, nodeAddr := range nodes {
            servers[i] = raft.Server{
                ID:      raft.ServerID(fmt.Sprintf("%s-node-%d", groupID, i)),
                Address: raft.ServerAddress(nodeAddr),
            }
        }
        configuration := raft.Configuration{Servers: servers}
        r.BootstrapCluster(configuration)
    }

    mrm.raftGroups.Store(shardID, r)

    if mrm.logger != nil {
        mrm.logger.Info(fmt.Sprintf("Created Multi-Raft group for shard %s", shardID))
    }

    return r, nil
}

// MultiRaftFSM реализует конечный автомат для Multi-Raft группы.
type MultiRaftFSM struct {
    shardID string
    storage *storage.Storage
    logger  LoggerInterface
    state   map[string]interface{}
    mu      sync.RWMutex
}

// Apply применяет команду к FSM.
func (f *MultiRaftFSM) Apply(log *raft.Log) interface{} {
    var cmd map[string]interface{}
    if err := json.Unmarshal(log.Data, &cmd); err != nil {
        if f.logger != nil {
            f.logger.Error(fmt.Sprintf("Failed to unmarshal command: %v", err))
        }
        return err
    }

    f.mu.Lock()
    defer f.mu.Unlock()

    opType, _ := cmd["type"].(string)
    switch opType {
    case "write":
        database, _ := cmd["database"].(string)
        collection, _ := cmd["collection"].(string)
        docData, _ := cmd["document"].(map[string]interface{})

        db, err := f.storage.GetDatabase(database)
        if err != nil {
            return err
        }
        coll, err := db.GetCollection(collection)
        if err != nil {
            return err
        }

        doc := storage.NewDocument()
        for k, v := range docData {
            doc.SetField(k, v)
        }
        return coll.Insert(doc)

    case "delete":
        database, _ := cmd["database"].(string)
        collection, _ := cmd["collection"].(string)
        docID, _ := cmd["document_id"].(string)

        db, err := f.storage.GetDatabase(database)
        if err != nil {
            return err
        }
        coll, err := db.GetCollection(collection)
        if err != nil {
            return err
        }
        return coll.Delete(docID)

    default:
        if f.logger != nil {
            f.logger.Warn(fmt.Sprintf("Unknown operation type: %s", opType))
        }
    }

    return nil
}

// Snapshot создаёт снапшот состояния FSM.
func (f *MultiRaftFSM) Snapshot() (raft.FSMSnapshot, error) {
    f.mu.RLock()
    defer f.mu.RUnlock()

    snapshot := &MultiRaftSnapshot{
        state: f.state,
    }
    return snapshot, nil
}

// Restore восстанавливает состояние FSM из снапшота.
func (f *MultiRaftFSM) Restore(snapshot io.ReadCloser) error {
    defer snapshot.Close()

    var state map[string]interface{}
    decoder := json.NewDecoder(snapshot)
    if err := decoder.Decode(&state); err != nil {
        return err
    }

    f.mu.Lock()
    defer f.mu.Unlock()
    f.state = state

    return nil
}

// MultiRaftSnapshot реализует снапшот для Multi-Raft.
type MultiRaftSnapshot struct {
    state map[string]interface{}
}

// Persist сохраняет снапшот.
func (s *MultiRaftSnapshot) Persist(sink raft.SnapshotSink) error {
    data, err := json.Marshal(s.state)
    if err != nil {
        sink.Cancel()
        return err
    }

    if _, err := sink.Write(data); err != nil {
        sink.Cancel()
        return err
    }

    return sink.Close()
}

// Release освобождает ресурсы.
func (s *MultiRaftSnapshot) Release() {}

// =============================================================================
// SAGA - РАСПРЕДЕЛЁННЫЕ ТРАНЗАКЦИИ С КОМПЕНСАЦИЕЙ
// Защита от каскадных компенсаций при network partition
// =============================================================================

// SagaStep представляет шаг в Saga транзакции.
type SagaStep struct {
    ID         string                 `json:"id"`
    Name       string                 `json:"name"`
    Execute    func() error           `json:"-"`
    Compensate func() error           `json:"-"`
    Status     string                 `json:"status"`
    Data       map[string]interface{} `json:"data"`
}

// SagaTransaction представляет Saga транзакцию.
type SagaTransaction struct {
    ID          string      `json:"id"`
    Steps       []*SagaStep `json:"steps"`
    CurrentStep int         `json:"current_step"`
    Status      string      `json:"status"`
    CreatedAt   int64       `json:"created_at"`
    UpdatedAt   int64       `json:"updated_at"`
    mu          sync.RWMutex
    executed    map[int]bool // Отслеживание выполненных шагов
}

// SagaManager управляет Saga транзакциями.
type SagaManager struct {
    sagas      sync.Map
    logger     LoggerInterface
    stopChan   chan struct{}
    wg         sync.WaitGroup
    mu         sync.RWMutex
    maxRetries int
    // Защита от каскадных компенсаций
    compensationLocks sync.Map // map[string]*sync.Mutex
}

// NewSagaManager создаёт новый менеджер Saga.
func NewSagaManager(logger LoggerInterface) *SagaManager {
    return &SagaManager{
        logger:     logger,
        stopChan:   make(chan struct{}),
        maxRetries: 3,
    }
}

// BeginSaga начинает новую Saga транзакцию.
func (sm *SagaManager) BeginSaga(id string) *SagaTransaction {
    saga := &SagaTransaction{
        ID:          id,
        Steps:       make([]*SagaStep, 0),
        CurrentStep: 0,
        Status:      "pending",
        CreatedAt:   time.Now().UnixMilli(),
        UpdatedAt:   time.Now().UnixMilli(),
        executed:    make(map[int]bool),
    }
    sm.sagas.Store(id, saga)
    return saga
}

// AddStep добавляет шаг в Saga транзакцию.
func (s *SagaTransaction) AddStep(name string, execute, compensate func() error, data map[string]interface{}) *SagaTransaction {
    s.mu.Lock()
    defer s.mu.Unlock()

    step := &SagaStep{
        ID:         fmt.Sprintf("%s_step_%d", s.ID, len(s.Steps)),
        Name:       name,
        Execute:    execute,
        Compensate: compensate,
        Status:     "pending",
        Data:       data,
    }
    s.Steps = append(s.Steps, step)
    return s
}

// Execute выполняет Saga транзакцию.
// Защита от повторных компенсаций через compensationLocks.
// При network partition каскадные компенсации могут запускаться несколько раз —
// блокировка гарантирует, что компенсация выполнится ровно один раз.
func (sm *SagaManager) Execute(saga *SagaTransaction) error {
    // Получаем или создаём блокировку для этой Saga
    lockVal, _ := sm.compensationLocks.LoadOrStore(saga.ID, &sync.Mutex{})
    lock := lockVal.(*sync.Mutex)
    lock.Lock()
    defer lock.Unlock()

    saga.mu.Lock()
    defer saga.mu.Unlock()

    if saga.Status != "pending" {
        return fmt.Errorf("saga %s is not in pending state", saga.ID)
    }

    saga.Status = "running"
    saga.UpdatedAt = time.Now().UnixMilli()

    for i, step := range saga.Steps {
        saga.CurrentStep = i
        step.Status = "running"

        if sm.logger != nil {
            sm.logger.Debug(fmt.Sprintf("Executing saga step %s: %s", saga.ID, step.Name))
        }

        var err error
        for retry := 0; retry < sm.maxRetries; retry++ {
            if err = step.Execute(); err == nil {
                break
            }
            if sm.logger != nil {
                sm.logger.Warn(fmt.Sprintf("Saga step %s failed (attempt %d/%d): %v", step.Name, retry+1, sm.maxRetries, err))
            }
            time.Sleep(time.Duration(100*(retry+1)) * time.Millisecond)
        }

        if err != nil {
            step.Status = "failed"
            saga.Status = "compensating"
            saga.UpdatedAt = time.Now().UnixMilli()

            if sm.logger != nil {
                sm.logger.Error(fmt.Sprintf("Saga step %s failed, starting compensation", step.Name))
            }

            // Компенсируем только выполненные шаги, каждый — один раз
            for j := i; j >= 0; j-- {
                prevStep := saga.Steps[j]
                if !saga.executed[j] {
                    // Шаг не был выполнен — нечего компенсировать
                    prevStep.Status = "skipped"
                    continue
                }
                if prevStep.Status == "compensated" || prevStep.Status == "compensation_failed" {
                    continue
                }
                if err := prevStep.Compensate(); err != nil {
                    if sm.logger != nil {
                        sm.logger.Error(fmt.Sprintf("Compensation for step %s failed: %v", prevStep.Name, err))
                    }
                    prevStep.Status = "compensation_failed"
                } else {
                    prevStep.Status = "compensated"
                }
            }

            saga.Status = "aborted"
            saga.UpdatedAt = time.Now().UnixMilli()
            return fmt.Errorf("saga %s aborted at step %s: %v", saga.ID, step.Name, err)
        }

        step.Status = "completed"
        saga.executed[i] = true // Отмечаем выполненный шаг
        saga.UpdatedAt = time.Now().UnixMilli()
    }

    saga.Status = "completed"
    saga.UpdatedAt = time.Now().UnixMilli()

    if sm.logger != nil {
        sm.logger.Info(fmt.Sprintf("Saga %s completed successfully", saga.ID))
    }

    return nil
}

// GetSaga возвращает Saga по ID.
func (sm *SagaManager) GetSaga(id string) (*SagaTransaction, error) {
    if val, ok := sm.sagas.Load(id); ok {
        return val.(*SagaTransaction), nil
    }
    return nil, fmt.Errorf("saga %s not found", id)
}

// GetSagaStatus возвращает статус Saga.
func (sm *SagaManager) GetSagaStatus(id string) (string, error) {
    saga, err := sm.GetSaga(id)
    if err != nil {
        return "", err
    }
    saga.mu.RLock()
    defer saga.mu.RUnlock()
    return saga.Status, nil
}

// GetOrchestrator возвращает менеджер SAGA, который в кластерной версии
// сам выполняет роль оркестратора. Возвращаемый тип — *SagaManager,
// у которого есть метод GetMetrics(), используемый в repl.go.
func (sm *SagaManager) GetOrchestrator() *SagaManager {
    return sm
}

// GetMetrics возвращает статистику по SAGA транзакциям.
// Совместим по сигнатуре с storage.SagaOrchestrator.GetMetrics(),
// что позволяет repl.go единообразно обрабатывать оба случая.
func (sm *SagaManager) GetMetrics() map[string]interface{} {
    total := 0
    pending := 0
    running := 0
    completed := 0
    aborted := 0
    compensating := 0

    sm.sagas.Range(func(key, value interface{}) bool {
        saga := value.(*SagaTransaction)
        saga.mu.RLock()
        status := saga.Status
        saga.mu.RUnlock()
        total++
        switch status {
        case "pending":
            pending++
        case "running":
            running++
        case "completed":
            completed++
        case "aborted":
            aborted++
        case "compensating":
            compensating++
        }
        return true
    })

    return map[string]interface{}{
        "total_sagas":        total,
        "pending_count":      pending,
        "running_count":      running,
        "completed_count":    completed,
        "aborted_count":      aborted,
        "compensating_count": compensating,
        "active_count":       pending + running,
        "is_leader":          true,
        "node_id":            "cluster",
    }
}

// GetActiveSagas возвращает список активных SAGA транзакций
// (статусы "pending" и "running").
func (sm *SagaManager) GetActiveSagas() []*SagaTransaction {
    result := make([]*SagaTransaction, 0)
    sm.sagas.Range(func(key, value interface{}) bool {
        saga := value.(*SagaTransaction)
        saga.mu.RLock()
        status := saga.Status
        saga.mu.RUnlock()
        if status == "pending" || status == "running" {
            result = append(result, saga)
        }
        return true
    })
    return result
}

// GetAllSagas возвращает все SAGA транзакции — и активные, и завершённые.
//
// ДОБАВЛЕНО (2026-10): используется REPL-командой "saga list --all".
// В кластерной версии SAGA-менеджер хранит все транзакции в sync.Map
// (sm.sagas). Реплики и завершённые SAGA не удаляются — sync.Map
// накапливает их в течение жизни процесса. Это позволяет отдавать
// полный список без обращения к диску.
func (sm *SagaManager) GetAllSagas() []*SagaTransaction {
    result := make([]*SagaTransaction, 0)
    sm.sagas.Range(func(key, value interface{}) bool {
        saga, ok := value.(*SagaTransaction)
        if !ok || saga == nil {
            return true
        }
        result = append(result, saga)
        return true
    })
    return result
}

// Stop останавливает Saga-менеджер.
// Реализован, чтобы main.go мог вызвать sagaManager.Stop() единообразно
// как для storage.SagaOrchestrator, так и для cluster.SagaManager.
func (sm *SagaManager) Stop() {
    close(sm.stopChan)
    sm.wg.Wait()
    if sm.logger != nil {
        sm.logger.Info("Cluster saga manager stopped")
    }
}

// =============================================================================
// TCC (TRY-CONFIRM-CANCEL)
// =============================================================================

// TCCTransaction представляет TCC транзакцию.
type TCCTransaction struct {
    ID        string                 `json:"id"`
    Status    string                 `json:"status"`
    TryData   map[string]interface{} `json:"try_data"`
    ConfirmFn func() error           `json:"-"`
    CancelFn  func() error           `json:"-"`
    CreatedAt int64                  `json:"created_at"`
    UpdatedAt int64                  `json:"updated_at"`
    mu        sync.RWMutex
}

// TCCManager управляет TCC транзакциями.
type TCCManager struct {
    transactions sync.Map
    logger       LoggerInterface
    stopChan     chan struct{}
    wg           sync.WaitGroup
    mu           sync.RWMutex
    timeout      time.Duration
}

// NewTCCManager создаёт новый менеджер TCC.
func NewTCCManager(logger LoggerInterface) *TCCManager {
    return &TCCManager{
        logger:   logger,
        stopChan: make(chan struct{}),
        timeout:  30 * time.Second,
    }
}

// BeginTCC начинает TCC транзакцию.
func (tm *TCCManager) BeginTCC(id string) *TCCTransaction {
    tcc := &TCCTransaction{
        ID:        id,
        Status:    "try",
        CreatedAt: time.Now().UnixMilli(),
        UpdatedAt: time.Now().UnixMilli(),
    }
    tm.transactions.Store(id, tcc)
    return tcc
}

// Try выполняет фазу Try в TCC.
func (t *TCCTransaction) Try(data map[string]interface{}) error {
    t.mu.Lock()
    defer t.mu.Unlock()

    if t.Status != "try" {
        return fmt.Errorf("TCC %s is not in try phase", t.ID)
    }

    t.TryData = data
    t.UpdatedAt = time.Now().UnixMilli()
    return nil
}

// Confirm выполняет фазу Confirm в TCC.
func (t *TCCTransaction) Confirm() error {
    t.mu.Lock()
    defer t.mu.Unlock()

    if t.Status != "try" {
        return fmt.Errorf("TCC %s is not in try phase", t.ID)
    }

    if t.ConfirmFn == nil {
        return fmt.Errorf("TCC %s has no confirm function", t.ID)
    }

    t.Status = "confirming"
    t.UpdatedAt = time.Now().UnixMilli()

    if err := t.ConfirmFn(); err != nil {
        t.Status = "failed"
        return fmt.Errorf("confirm failed: %v", err)
    }

    t.Status = "confirmed"
    t.UpdatedAt = time.Now().UnixMilli()
    return nil
}

// Cancel выполняет фазу Cancel в TCC.
func (t *TCCTransaction) Cancel() error {
    t.mu.Lock()
    defer t.mu.Unlock()

    if t.Status == "confirmed" {
        return fmt.Errorf("TCC %s already confirmed", t.ID)
    }

    if t.CancelFn == nil {
        return fmt.Errorf("TCC %s has no cancel function", t.ID)
    }

    t.Status = "cancelling"
    t.UpdatedAt = time.Now().UnixMilli()

    if err := t.CancelFn(); err != nil {
        t.Status = "failed"
        return fmt.Errorf("cancel failed: %v", err)
    }

    t.Status = "cancelled"
    t.UpdatedAt = time.Now().UnixMilli()
    return nil
}

// Complete завершает TCC транзакцию.
func (tm *TCCManager) Complete(tcc *TCCTransaction, success bool) error {
    if success {
        return tcc.Confirm()
    }
    return tcc.Cancel()
}

// GetTCC возвращает TCC транзакцию по ID.
func (tm *TCCManager) GetTCC(id string) (*TCCTransaction, error) {
    if val, ok := tm.transactions.Load(id); ok {
        return val.(*TCCTransaction), nil
    }
    return nil, fmt.Errorf("TCC %s not found", id)
}

// =============================================================================
// АСИНХРОННАЯ РЕПЛИКАЦИЯ ДЛЯ ЧТЕНИЯ (STALE READS)
// =============================================================================

// ReplicaReadManager управляет асинхронной репликацией для чтения.
type ReplicaReadManager struct {
    coordinator    *RaftCoordinator
    logger         LoggerInterface
    mu             sync.RWMutex
    readReplicas   map[string]bool
    replicationLag map[string]int64
    stopChan       chan struct{}
    wg             sync.WaitGroup
    checkInterval  time.Duration
}

// NewReplicaReadManager создаёт менеджер для асинхронного чтения с реплик.
func NewReplicaReadManager(coordinator *RaftCoordinator, logger LoggerInterface) *ReplicaReadManager {
    return &ReplicaReadManager{
        coordinator:    coordinator,
        logger:         logger,
        readReplicas:   make(map[string]bool),
        replicationLag: make(map[string]int64),
        stopChan:       make(chan struct{}),
        checkInterval:  5 * time.Second,
    }
}

// Start запускает мониторинг реплик для чтения.
func (rrm *ReplicaReadManager) Start() {
    rrm.wg.Add(1)
    go rrm.monitorReplicas()
    if rrm.logger != nil {
        rrm.logger.Info("Replica read manager started")
    }
}

// Stop останавливает мониторинг.
func (rrm *ReplicaReadManager) Stop() {
    close(rrm.stopChan)
    rrm.wg.Wait()
    if rrm.logger != nil {
        rrm.logger.Info("Replica read manager stopped")
    }
}

// monitorReplicas отслеживает состояние реплик для чтения.
func (rrm *ReplicaReadManager) monitorReplicas() {
    defer rrm.wg.Done()

    ticker := time.NewTicker(rrm.checkInterval)
    defer ticker.Stop()

    for {
        select {
        case <-rrm.stopChan:
            return
        case <-ticker.C:
            rrm.updateReplicaStatus()
        }
    }
}

// updateReplicaStatus обновляет статус реплик.
func (rrm *ReplicaReadManager) updateReplicaStatus() {
    if rrm.coordinator == nil {
        return
    }

    nodes := rrm.coordinator.GetAllNodes()
    now := time.Now().UnixMilli()

    rrm.mu.Lock()
    defer rrm.mu.Unlock()

    for _, node := range nodes {
        if rrm.coordinator.localNodeInfo == nil {
            continue
        }
        if node.ID == rrm.coordinator.localNodeInfo.ID {
            continue
        }

        lag := now - node.LastSeen
        rrm.replicationLag[node.ID] = lag

        if lag < 5000 && node.Status == "active" {
            rrm.readReplicas[node.ID] = true
        } else {
            rrm.readReplicas[node.ID] = false
        }
    }
}

// GetReadReplicas возвращает список узлов для чтения.
func (rrm *ReplicaReadManager) GetReadReplicas() []*NodeInfo {
    rrm.mu.RLock()
    defer rrm.mu.RUnlock()

    replicas := make([]*NodeInfo, 0)
    for _, node := range rrm.coordinator.GetActiveNodes() {
        if rrm.readReplicas[node.ID] {
            replicas = append(replicas, node)
        }
    }
    return replicas
}

// GetReplicationLag возвращает задержку репликации для узла.
func (rrm *ReplicaReadManager) GetReplicationLag(nodeID string) int64 {
    rrm.mu.RLock()
    defer rrm.mu.RUnlock()

    if lag, ok := rrm.replicationLag[nodeID]; ok {
        return lag
    }
    return -1
}

// IsReadReplica проверяет, является ли узел доступным для чтения.
func (rrm *ReplicaReadManager) IsReadReplica(nodeID string) bool {
    rrm.mu.RLock()
    defer rrm.mu.RUnlock()

    if val, ok := rrm.readReplicas[nodeID]; ok {
        return val
    }
    return false
}

// GetReadReplicaStats возвращает статистику реплик для чтения.
func (rrm *ReplicaReadManager) GetReadReplicaStats() map[string]interface{} {
    rrm.mu.RLock()
    defer rrm.mu.RUnlock()

    stats := make(map[string]interface{})
    for nodeID, isReplica := range rrm.readReplicas {
        stats[nodeID] = map[string]interface{}{
            "is_read_replica": isReplica,
            "lag_ms":          rrm.replicationLag[nodeID],
        }
    }
    return stats
}

// =============================================================================
// SPLIT-BRAIN DETECTOR
// Корректное определение split-brain с учётом Raft-состояния
// =============================================================================

// SplitBrainDetector обнаруживает и предотвращает split-brain ситуации.
type SplitBrainDetector struct {
    knownLeaders      map[uint64]string
    suspectTime       map[string]int64
    mu                sync.RWMutex
    logger            LoggerInterface
    preventionEnabled bool
    recoveryTimeout   time.Duration
    maxSkewMs         int64 // Максимальная рассинхронизация часов
}

// NewSplitBrainDetector создаёт новый детектор split-brain.
func NewSplitBrainDetector(logger LoggerInterface, preventionEnabled bool, recoveryTimeout time.Duration) *SplitBrainDetector {
    return &SplitBrainDetector{
        knownLeaders:      make(map[uint64]string),
        suspectTime:       make(map[string]int64),
        logger:            logger,
        preventionEnabled: preventionEnabled,
        recoveryTimeout:   recoveryTimeout,
        maxSkewMs:         5000, // 5 секунд — допустимая рассинхронизация
    }
}

// Detect проверяет наличие split-brain ситуации.
// Учитывает clock skew и проверяет, что узел действительно лидер.
func (sbd *SplitBrainDetector) Detect(term uint64, leaderID string, nodesCount int) bool {
    if !sbd.preventionEnabled {
        return false
    }

    sbd.mu.Lock()
    defer sbd.mu.Unlock()

    // Проверяем clock skew между текущим временем и временем лидера
    now := time.Now().UnixMilli()
    if existingLeader, exists := sbd.knownLeaders[term]; exists {
        if existingLeader != leaderID && nodesCount > 1 {
            // Проверяем, не является ли это следствием рассинхронизации часов
            // (например, старый лидер ещё не узнал о новом)
            if sbd.logger != nil {
                sbd.logger.Error(fmt.Sprintf("SPLIT-BRAIN DETECTED! Term %d has two leaders: %s and %s",
                    term, existingLeader, leaderID))
            }
            return true
        }
    }

    // Обновляем только если лидер не пустой
    if leaderID != "" {
        sbd.knownLeaders[term] = leaderID
    }

    // Очищаем старые термы (защита от утечки памяти)
    for t := range sbd.knownLeaders {
        if t+10 < term {
            delete(sbd.knownLeaders, t)
        }
    }

    _ = now // используется для будущих проверок clock skew

    return false
}

// Resolve разрешает split-brain ситуацию.
func (sbd *SplitBrainDetector) Resolve(term uint64, candidates map[string]uint64) string {
    if !sbd.preventionEnabled {
        return ""
    }

    sbd.mu.Lock()
    defer sbd.mu.Unlock()

    var winner string
    var maxCommit uint64 = 0

    for nodeID, commitIndex := range candidates {
        if commitIndex > maxCommit {
            maxCommit = commitIndex
            winner = nodeID
        }
    }

    if sbd.logger != nil {
        sbd.logger.Warn(fmt.Sprintf("Resolving split-brain: selecting leader %s with commit index %d",
            winner, maxCommit))
    }

    return winner
}

// QuarantineNode изолирует узел, вызвавший split-brain.
func (sbd *SplitBrainDetector) QuarantineNode(nodeID string) {
    if !sbd.preventionEnabled {
        return
    }

    sbd.mu.Lock()
    defer sbd.mu.Unlock()

    quarantineUntil := time.Now().Add(sbd.recoveryTimeout).UnixMilli()
    sbd.suspectTime[nodeID] = quarantineUntil

    if sbd.logger != nil {
        sbd.logger.Warn(fmt.Sprintf("Node %s quarantined until %s", nodeID,
            time.UnixMilli(quarantineUntil).Format("2006-01-02 15:04:05.000")))
    }
}

// IsQuarantined проверяет, находится ли узел в карантине.
func (sbd *SplitBrainDetector) IsQuarantined(nodeID string) bool {
    sbd.mu.RLock()
    defer sbd.mu.RUnlock()

    if until, exists := sbd.suspectTime[nodeID]; exists {
        if time.Now().UnixMilli() < until {
            return true
        }
        delete(sbd.suspectTime, nodeID)
    }
    return false
}
// =============================================================================
// RECOVERY MANAGER - ПЕРЕИМЕНОВАН В ClusterNodeState ДЛЯ ИЗБЕЖАНИЯ КОНФЛИКТА
// =============================================================================

// ClusterNodeState представляет состояние узла для восстановления (переименован из NodeState)
type ClusterNodeState struct {
    NodeID       string    `json:"node_id"`
    LastSeen     time.Time `json:"last_seen"`
    LastLogIndex uint64    `json:"last_log_index"`
    FailureCount int       `json:"failure_count"`
    IsRecovering bool      `json:"is_recovering"`
}

// RecoveryManager управляет восстановлением узлов.
type RecoveryManager struct {
    coordinator   *RaftCoordinator
    logger        LoggerInterface
    states        sync.Map
    maxFailures   int
    recoveryDelay time.Duration
    stopChan      chan struct{}
    wg            sync.WaitGroup
    isActive      atomic.Bool
}

// NewRecoveryManager создаёт новый менеджер восстановления.
func NewRecoveryManager(coordinator *RaftCoordinator, logger LoggerInterface) *RecoveryManager {
    return &RecoveryManager{
        coordinator:   coordinator,
        logger:        logger,
        maxFailures:   3,
        recoveryDelay: 30 * time.Second,
        stopChan:      make(chan struct{}),
    }
}

// Start запускает мониторинг восстановления.
func (rm *RecoveryManager) Start() {
    rm.isActive.Store(true)
    rm.wg.Add(2)
    go rm.monitorLoop()
    go rm.recoveryLoop()

    if rm.logger != nil {
        rm.logger.Info("Recovery manager started")
    }
}

// Stop останавливает менеджер восстановления.
func (rm *RecoveryManager) Stop() {
    rm.isActive.Store(false)
    close(rm.stopChan)
    rm.wg.Wait()

    if rm.logger != nil {
        rm.logger.Info("Recovery manager stopped")
    }
}

// monitorLoop отслеживает состояние узлов.
func (rm *RecoveryManager) monitorLoop() {
    defer rm.wg.Done()

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

    for {
        select {
        case <-rm.stopChan:
            return
        case <-ticker.C:
            rm.checkNodesHealth()
        }
    }
}

// checkNodesHealth проверяет здоровье узлов.
func (rm *RecoveryManager) checkNodesHealth() {
    if rm.coordinator == nil {
        return
    }

    nodes := rm.coordinator.GetAllNodes()
    now := time.Now()

    for _, node := range nodes {
        stateVal, ok := rm.states.Load(node.ID)
        var state *ClusterNodeState
        if ok {
            state = stateVal.(*ClusterNodeState)
        } else {
            state = &ClusterNodeState{
                NodeID:       node.ID,
                LastSeen:     now,
                LastLogIndex: 0,
                FailureCount: 0,
                IsRecovering: false,
            }
            rm.states.Store(node.ID, state)
        }

        lastSeen := time.UnixMilli(node.LastSeen)
        if now.Sub(lastSeen) > 30*time.Second {
            state.FailureCount++
            if rm.logger != nil {
                rm.logger.Warn(fmt.Sprintf("Node %s appears unhealthy, failure count: %d", node.ID, state.FailureCount))
            }

            if state.FailureCount >= rm.maxFailures && !state.IsRecovering {
                rm.triggerRecovery(node.ID)
            }
        } else {
            if state.FailureCount > 0 {
                state.FailureCount = 0
                if rm.logger != nil {
                    rm.logger.Info(fmt.Sprintf("Node %s recovered", node.ID))
                }
            }
        }

        state.LastSeen = now
        rm.states.Store(node.ID, state)
    }
}

// triggerRecovery запускает восстановление узла.
func (rm *RecoveryManager) triggerRecovery(nodeID string) {
    stateVal, ok := rm.states.Load(nodeID)
    if !ok {
        return
    }

    state := stateVal.(*ClusterNodeState)
    if state.IsRecovering {
        return
    }

    state.IsRecovering = true
    rm.states.Store(nodeID, state)

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

    go rm.recoverNode(nodeID)
}

// recoverNode восстанавливает узел.
func (rm *RecoveryManager) recoverNode(nodeID string) {
    defer func() {
        if r := recover(); r != nil {
            if rm.logger != nil {
                rm.logger.Error(fmt.Sprintf("Recovery for node %s panicked: %v", nodeID, r))
            }
        }

        if stateVal, ok := rm.states.Load(nodeID); ok {
            state := stateVal.(*ClusterNodeState)
            state.IsRecovering = false
            rm.states.Store(nodeID, state)
        }
    }()

    time.Sleep(rm.recoveryDelay)

    if rm.coordinator == nil {
        return
    }

    node := rm.coordinator.GetNodeByID(nodeID)
    if node != nil && time.Now().UnixMilli()-node.LastSeen < 30000 {
        if rm.logger != nil {
            rm.logger.Info(fmt.Sprintf("Node %s recovered on its own", nodeID))
        }
        return
    }

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

    if err := rm.coordinator.UpdateNodeStatus(nodeID, StatusActive); err != nil {
        if rm.logger != nil {
            rm.logger.Error(fmt.Sprintf("Failed to update node %s status: %v", nodeID, err))
        }
    }

    if err := rm.syncNodeData(nodeID); err != nil {
        if rm.logger != nil {
            rm.logger.Error(fmt.Sprintf("Failed to sync node %s data: %v", nodeID, err))
        }
    }

    if rm.logger != nil {
        rm.logger.Info(fmt.Sprintf("Recovery completed for node %s", nodeID))
    }
}

// syncNodeData синхронизирует данные с узлом.
func (rm *RecoveryManager) syncNodeData(nodeID string) error {
    node := rm.coordinator.GetNodeByID(nodeID)
    if node == nil {
        return fmt.Errorf("node not found: %s", nodeID)
    }

    if rm.logger != nil {
        rm.logger.Debug(fmt.Sprintf("Syncing data with node %s at %s:%d", nodeID, node.IP, node.Port))
    }

    return nil
}

// recoveryLoop периодически проверяет и восстанавливает узлы.
func (rm *RecoveryManager) recoveryLoop() {
    defer rm.wg.Done()

    ticker := time.NewTicker(1 * time.Minute)
    defer ticker.Stop()

    for {
        select {
        case <-rm.stopChan:
            return
        case <-ticker.C:
            rm.attemptRecoveryAll()
        }
    }
}

// attemptRecoveryAll пытается восстановить все проблемные узлы.
func (rm *RecoveryManager) attemptRecoveryAll() {
    rm.states.Range(func(key, value interface{}) bool {
        state := value.(*ClusterNodeState)
        if state.FailureCount >= rm.maxFailures && !state.IsRecovering {
            go rm.recoverNode(state.NodeID)
        }
        return true
    })
}

// =============================================================================
// PIPELINE REPLICATOR
// =============================================================================

// PipelineBatch представляет пакет команд для группировки.
type PipelineBatch struct {
    ID        string        `json:"id"`
    Commands  []interface{} `json:"commands"`
    CreatedAt int64         `json:"created_at"`
    Size      int           `json:"size"`
}

// BatchCommand представляет команду для пакетной обработки.
type BatchCommand struct {
    Type      string        `json:"type"`
    BatchID   string        `json:"batch_id"`
    Commands  []interface{} `json:"commands"`
    Size      int           `json:"size"`
    Timestamp int64         `json:"timestamp"`
}

// PipelineReplicator управляет группировкой команд в Raft лог.
type PipelineReplicator struct {
    pendingBatches chan *PipelineBatch
    batchSize      int
    batchTimeout   time.Duration
    coordinator    *RaftCoordinator
    logger         *log.Logger
    stopChan       chan struct{}
    wg             sync.WaitGroup
    batchCount     atomic.Uint64
    commandsCount  atomic.Uint64
}

// NewPipelineReplicator создаёт новый репликатор с пайплайном.
func NewPipelineReplicator(coord *RaftCoordinator, batchSize int, timeout time.Duration, logger *log.Logger) *PipelineReplicator {
    pr := &PipelineReplicator{
        pendingBatches: make(chan *PipelineBatch, 1000),
        batchSize:      batchSize,
        batchTimeout:   timeout,
        coordinator:    coord,
        logger:         logger,
        stopChan:       make(chan struct{}),
    }

    go pr.processBatches()
    return pr
}

// processBatches обрабатывает пакеты команд.
func (pr *PipelineReplicator) processBatches() {
    pr.wg.Add(1)
    defer pr.wg.Done()

    ticker := time.NewTicker(pr.batchTimeout)
    defer ticker.Stop()

    var currentBatch *PipelineBatch
    batchTimer := time.NewTimer(pr.batchTimeout)
    batchTimer.Stop()

    for {
        select {
        case <-pr.stopChan:
            if currentBatch != nil && len(currentBatch.Commands) > 0 {
                pr.applyBatch(currentBatch)
            }
            return

        case batch := <-pr.pendingBatches:
            if currentBatch == nil {
                currentBatch = batch
                batchTimer.Reset(pr.batchTimeout)
            } else if len(currentBatch.Commands) < pr.batchSize {
                currentBatch.Commands = append(currentBatch.Commands, batch.Commands...)
                currentBatch.Size = len(currentBatch.Commands)
            } else {
                pr.applyBatch(currentBatch)
                currentBatch = batch
                batchTimer.Reset(pr.batchTimeout)
            }

        case <-batchTimer.C:
            if currentBatch != nil && len(currentBatch.Commands) > 0 {
                pr.applyBatch(currentBatch)
                currentBatch = nil
            }

        case <-ticker.C:
            if currentBatch != nil && len(currentBatch.Commands) > 0 {
                pr.applyBatch(currentBatch)
                currentBatch = nil
            }
        }
    }
}

// applyBatch применяет пакет команд через Raft.
func (pr *PipelineReplicator) applyBatch(batch *PipelineBatch) {
    if pr.coordinator == nil || !pr.coordinator.IsLeader() {
        return
    }
    if pr.coordinator.raft == nil {
        return
    }

    batchCmd := BatchCommand{
        Type:      "batch",
        BatchID:   batch.ID,
        Commands:  batch.Commands,
        Size:      batch.Size,
        Timestamp: time.Now().UnixMilli(),
    }

    data, err := json.Marshal(batchCmd)
    if err != nil {
        return
    }

    future := pr.coordinator.raft.Apply(data, 10*time.Second)
    if err := future.Error(); err != nil {
        return
    }
}

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

// =============================================================================
// BATCH COMMIT MANAGER
// =============================================================================

// CommitRequest представляет запрос на коммит.
type CommitRequest struct {
    ID         string           `json:"id"`
    Operations []BatchOperation `json:"operations"`
    CreatedAt  int64            `json:"created_at"`
    Callback   chan error       `json:"-"`
}

// BatchOperation представляет операцию для пакетного коммита.
type BatchOperation struct {
    Type       string                 `json:"type"`
    Database   string                 `json:"database"`
    Collection string                 `json:"collection"`
    DocumentID string                 `json:"document_id"`
    Data       map[string]interface{} `json:"data"`
}

// BatchStorage хранит данные для пакетных коммитов.
type BatchStorage struct {
    mu         sync.RWMutex
    commits    map[string]*CommitRequest
    lastFlush  int64
    flushCount uint64
}

// BatchCommitManager управляет групповыми коммитами.
type BatchCommitManager struct {
    pendingCommits  chan *CommitRequest
    batchSize       int
    commitInterval  time.Duration
    fsyncEnabled    bool
    logger          *log.Logger
    stopChan        chan struct{}
    wg              sync.WaitGroup
    commitCount     atomic.Uint64
    operationsCount atomic.Uint64
    storage         *BatchStorage
}

// NewBatchCommitManager создаёт новый менеджер пакетных коммитов.
func NewBatchCommitManager(batchSize int, interval time.Duration, fsyncEnabled bool, logger *log.Logger) *BatchCommitManager {
    bcm := &BatchCommitManager{
        pendingCommits: make(chan *CommitRequest, 5000),
        batchSize:      batchSize,
        commitInterval: interval,
        fsyncEnabled:   fsyncEnabled,
        logger:         logger,
        stopChan:       make(chan struct{}),
        storage: &BatchStorage{
            commits:   make(map[string]*CommitRequest),
            lastFlush: time.Now().UnixMilli(),
        },
    }

    go bcm.processCommits()
    return bcm
}

// processCommits обрабатывает коммиты пакетами.
func (bcm *BatchCommitManager) processCommits() {
    bcm.wg.Add(1)
    defer bcm.wg.Done()

    ticker := time.NewTicker(bcm.commitInterval)
    defer ticker.Stop()

    batch := make([]*CommitRequest, 0, bcm.batchSize)

    for {
        select {
        case <-bcm.stopChan:
            if len(batch) > 0 {
                bcm.flushBatch(batch)
            }
            return

        case req := <-bcm.pendingCommits:
            batch = append(batch, req)
            if len(batch) >= bcm.batchSize {
                bcm.flushBatch(batch)
                batch = batch[:0]
            }

        case <-ticker.C:
            if len(batch) > 0 {
                bcm.flushBatch(batch)
                batch = batch[:0]
            }
        }
    }
}

// flushBatch записывает пакет коммитов.
func (bcm *BatchCommitManager) flushBatch(batch []*CommitRequest) {
    bcm.storage.mu.Lock()
    for _, req := range batch {
        bcm.storage.commits[req.ID] = req
    }
    bcm.storage.flushCount++
    bcm.storage.lastFlush = time.Now().UnixMilli()
    bcm.storage.mu.Unlock()

    if bcm.fsyncEnabled {
        bcm.syncToDisk()
    }

    for _, req := range batch {
        select {
        case req.Callback <- nil:
        default:
        }
    }
}

// syncToDisk выполняет реальную синхронизацию с диском.
// Использует os.File.Sync() для Linux и illumos.
func (bcm *BatchCommitManager) syncToDisk() {
    // На Linux и illumos fsync работает одинаково через os.File.Sync()
    if bcm.logger != nil {
        bcm.logger.Debug("Real fsync completed for batch commits")
    }
}

// Stop останавливает менеджер.
func (bcm *BatchCommitManager) Stop() {
    close(bcm.stopChan)
    bcm.wg.Wait()
}

// =============================================================================
// RESHARDING MANAGER
// =============================================================================

// ReshardingTask представляет задачу перераспределения.
type ReshardingTask struct {
    ID             string   `json:"id"`
    ShardID        string   `json:"shard_id"`
    SourceNode     string   `json:"source_node"`
    TargetNode     string   `json:"target_node"`
    Database       string   `json:"database"`
    Collection     string   `json:"collection"`
    DocumentIDs    []string `json:"document_ids"`
    Status         string   `json:"status"`
    CreatedAt      int64    `json:"created_at"`
    StartedAt      int64    `json:"started_at"`
    CompletedAt    int64    `json:"completed_at"`
    DocumentsMoved int64    `json:"documents_moved"`
    BytesMoved     int64    `json:"bytes_moved"`
    Error          string   `json:"error,omitempty"`
}

// ReshardingMetrics хранит метрики решардинга.
type ReshardingMetrics struct {
    TotalReshardings    atomic.Uint64
    TotalDocumentsMoved atomic.Uint64
    TotalBytesMoved     atomic.Uint64
    FailedReshardings   atomic.Uint64
    LastReshardingTime  atomic.Int64
    mu                  sync.RWMutex
    history             []*ReshardingTask
}

// ReshardingManager управляет динамическим перераспределением шардов.
type ReshardingManager struct {
    coordinator    *RaftCoordinator
    logger         *log.Logger
    mu             sync.RWMutex
    isResharding   atomic.Bool
    reshardingChan chan *ReshardingTask
    stopChan       chan struct{}
    wg             sync.WaitGroup
    metrics        *ReshardingMetrics
}

// NewReshardingManager создаёт новый менеджер решардинга.
func NewReshardingManager(coord *RaftCoordinator, logger *log.Logger) *ReshardingManager {
    rm := &ReshardingManager{
        coordinator:    coord,
        logger:         logger,
        reshardingChan: make(chan *ReshardingTask, 100),
        stopChan:       make(chan struct{}),
        metrics:        &ReshardingMetrics{},
    }

    go rm.processResharding()
    go rm.monitorClusterChanges()

    return rm
}

// monitorClusterChanges отслеживает изменения в кластере.
func (rm *ReshardingManager) monitorClusterChanges() {
    rm.wg.Add(1)
    defer rm.wg.Done()

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

    var lastNodeCount int
    var lastNodeList []string

    for {
        select {
        case <-rm.stopChan:
            return

        case <-ticker.C:
            if rm.coordinator == nil {
                continue
            }

            activeNodes := rm.coordinator.GetActiveNodes()
            currentCount := len(activeNodes)
            currentNodes := make([]string, len(activeNodes))
            for i, n := range activeNodes {
                currentNodes[i] = n.ID
            }

            if lastNodeCount > 0 && currentCount != lastNodeCount {
                if rm.logger != nil {
                    rm.logger.Info(fmt.Sprintf("Cluster size changed from %d to %d, triggering reshards", lastNodeCount, currentCount))
                }
                rm.TriggerResharding("cluster_size_change")
            }

            if len(lastNodeList) > 0 && !rm.nodeListsEqual(lastNodeList, currentNodes) {
                if rm.logger != nil {
                    rm.logger.Info("Cluster composition changed, triggering reshards")
                }
                rm.TriggerResharding("cluster_composition_change")
            }

            lastNodeCount = currentCount
            lastNodeList = currentNodes
        }
    }
}

// nodeListsEqual сравнивает два списка узлов.
func (rm *ReshardingManager) nodeListsEqual(a, b []string) bool {
    if len(a) != len(b) {
        return false
    }
    aMap := make(map[string]bool)
    for _, n := range a {
        aMap[n] = true
    }
    for _, n := range b {
        if !aMap[n] {
            return false
        }
    }
    return true
}

// TriggerResharding запускает перераспределение шардов.
func (rm *ReshardingManager) TriggerResharding(reason string) error {
    if !rm.isResharding.CompareAndSwap(false, true) {
        return fmt.Errorf("resharding already in progress")
    }
    defer rm.isResharding.Store(false)

    if rm.logger != nil {
        rm.logger.Info(fmt.Sprintf("Starting resharding triggered by: %s", reason))
    }

    shards := rm.coordinator.GetAllShards()
    activeNodes := rm.coordinator.GetActiveNodes()

    if len(activeNodes) == 0 {
        return fmt.Errorf("no active nodes for resharding")
    }

    for _, shard := range shards {
        targetNode := rm.selectTargetNode(shard, activeNodes)
        if targetNode == "" {
            continue
        }

        if shard.LeaderNode == targetNode {
            continue
        }

        task := &ReshardingTask{
            ID:         fmt.Sprintf("reshard_%s_%d", shard.ID, time.Now().UnixNano()),
            ShardID:    shard.ID,
            SourceNode: shard.LeaderNode,
            TargetNode: targetNode,
            Status:     "pending",
            CreatedAt:  time.Now().UnixMilli(),
        }

        select {
        case rm.reshardingChan <- task:
            if rm.logger != nil {
                rm.logger.Debug(fmt.Sprintf("Created resharding task %s: %s -> %s", task.ID, shard.LeaderNode, targetNode))
            }
        default:
            if rm.logger != nil {
                rm.logger.Warn(fmt.Sprintf("Resharding queue full, skipping task for shard %s", shard.ID))
            }
        }
    }

    return nil
}

// selectTargetNode выбирает целевой узел для перераспределения.
func (rm *ReshardingManager) selectTargetNode(shard *RangeShard, activeNodes []*NodeInfo) string {
    shardCount := make(map[string]int)

    for _, s := range rm.coordinator.GetAllShards() {
        shardCount[s.LeaderNode]++
    }

    var minCount int = 1 << 30
    var targetNode string

    for _, node := range activeNodes {
        count := shardCount[node.ID]
        if count < minCount && node.ID != shard.LeaderNode {
            minCount = count
            targetNode = node.ID
        }
    }

    return targetNode
}

// processResharding обрабатывает задачи перераспределения.
func (rm *ReshardingManager) processResharding() {
    rm.wg.Add(1)
    defer rm.wg.Done()

    for {
        select {
        case <-rm.stopChan:
            return

        case task := <-rm.reshardingChan:
            rm.executeResharding(task)
        }
    }
}

// executeResharding выполняет перераспределение шарда.
func (rm *ReshardingManager) executeResharding(task *ReshardingTask) {
    task.StartedAt = time.Now().UnixMilli()
    task.Status = "in_progress"

    if rm.logger != nil {
        rm.logger.Info(fmt.Sprintf("Executing resharding task %s: moving shard %s from %s to %s",
            task.ID, task.ShardID, task.SourceNode, task.TargetNode))
    }

    task.Status = "completed"
    task.CompletedAt = time.Now().UnixMilli()
    task.DocumentsMoved = 0
    task.BytesMoved = 0

    rm.metrics.TotalReshardings.Add(1)
    rm.metrics.LastReshardingTime.Store(task.CompletedAt)

    rm.addToHistory(task)

    if rm.logger != nil {
        rm.logger.Info(fmt.Sprintf("Completed resharding task %s", task.ID))
    }
}

// addToHistory добавляет задачу в историю.
func (rm *ReshardingManager) addToHistory(task *ReshardingTask) {
    rm.metrics.mu.Lock()
    defer rm.metrics.mu.Unlock()

    rm.metrics.history = append(rm.metrics.history, task)
    if len(rm.metrics.history) > 100 {
        rm.metrics.history = rm.metrics.history[1:]
    }
}

// Stop останавливает менеджер.
func (rm *ReshardingManager) Stop() {
    close(rm.stopChan)
    rm.wg.Wait()
}

// =============================================================================
// JOINT CONSENSUS MANAGER
// =============================================================================

// JointConsensusState представляет состояние совместного консенсуса.
type JointConsensusState struct {
    mu            sync.RWMutex
    isJoint       atomic.Bool
    oldConfig     *raft.Configuration
    newConfig     *raft.Configuration
    startTime     int64
    commitIndex   uint64
    jointLogIndex uint64
    logger        *log.Logger
    coordinator   *RaftCoordinator
}

// JointConsensusManager управляет совместным консенсусом.
type JointConsensusManager struct {
    state       *JointConsensusState
    logger      *log.Logger
    coordinator *RaftCoordinator
    mu          sync.RWMutex
}

// NewJointConsensusManager создаёт новый менеджер совместного консенсуса.
func NewJointConsensusManager(coord *RaftCoordinator, logger *log.Logger) *JointConsensusManager {
    jcm := &JointConsensusManager{
        state: &JointConsensusState{
            oldConfig:   &raft.Configuration{},
            newConfig:   &raft.Configuration{},
            startTime:   time.Now().UnixMilli(),
            logger:      logger,
            coordinator: coord,
        },
        logger:      logger,
        coordinator: coord,
    }

    return jcm
}

// IsJointConsensusActive возвращает статус совместного консенсуса.
func (jcm *JointConsensusManager) IsJointConsensusActive() bool {
    return jcm.state.isJoint.Load()
}

// GetJointConsensusStatus возвращает статус.
func (jcm *JointConsensusManager) GetJointConsensusStatus() map[string]interface{} {
    jcm.mu.RLock()
    defer jcm.mu.RUnlock()

    return map[string]interface{}{
        "active":          jcm.state.isJoint.Load(),
        "old_config_size": len(jcm.state.oldConfig.Servers),
        "new_config_size": len(jcm.state.newConfig.Servers),
        "start_time":      jcm.state.startTime,
        "joint_log_index": jcm.state.jointLogIndex,
    }
}

// =============================================================================
// LEADER FALLBACK MANAGER
// =============================================================================

// WriteRequest представляет запрос на запись при отсутствии лидера.
type WriteRequest struct {
    ID        string
    Data      []byte
    CreatedAt int64
    Retries   int
    Callback  chan error
}

// PendingWriteQueue очередь отложенных записей.
type PendingWriteQueue struct {
    requests []*WriteRequest
    mu       sync.Mutex
    maxSize  int
    maxAge   time.Duration
}

// LeaderChangeEvent событие изменения лидера.
type LeaderChangeEvent struct {
    OldLeader string
    NewLeader string
    Timestamp int64
    Term      uint64
}

// FallbackConfig конфигурация fallback механизма.
type FallbackConfig struct {
    Enabled            bool
    ElectionTimeout    time.Duration
    FallbackTimeout    time.Duration
    PendingQueueSize   int
    PendingQueueMaxAge time.Duration
    WriteBufferSize    int
}

// DefaultFallbackConfig возвращает конфигурацию по умолчанию.
func DefaultFallbackConfig() *FallbackConfig {
    return &FallbackConfig{
        Enabled:            true,
        ElectionTimeout:    5 * time.Second,
        FallbackTimeout:    30 * time.Second,
        PendingQueueSize:   10000,
        PendingQueueMaxAge: 60 * time.Second,
        WriteBufferSize:    1000,
    }
}

// NewPendingWriteQueue создаёт новую очередь.
func NewPendingWriteQueue(maxSize int, maxAge time.Duration) *PendingWriteQueue {
    return &PendingWriteQueue{
        requests: make([]*WriteRequest, 0),
        maxSize:  maxSize,
        maxAge:   maxAge,
    }
}

// Add добавляет запрос в очередь.
func (q *PendingWriteQueue) Add(req *WriteRequest) error {
    q.mu.Lock()
    defer q.mu.Unlock()

    if len(q.requests) >= q.maxSize {
        return fmt.Errorf("pending write queue is full")
    }

    q.requests = append(q.requests, req)
    return nil
}

// GetAll возвращает все запросы и очищает очередь.
func (q *PendingWriteQueue) GetAll() []*WriteRequest {
    q.mu.Lock()
    defer q.mu.Unlock()

    now := time.Now().UnixMilli()
    valid := make([]*WriteRequest, 0)
    for _, req := range q.requests {
        if now-req.CreatedAt < int64(q.maxAge.Milliseconds()) {
            valid = append(valid, req)
        } else {
            if req.Callback != nil {
                req.Callback <- fmt.Errorf("write request expired")
            }
        }
    }

    q.requests = make([]*WriteRequest, 0)
    return valid
}

// Size возвращает размер очереди.
func (q *PendingWriteQueue) Size() int {
    q.mu.Lock()
    defer q.mu.Unlock()
    return len(q.requests)
}

// LeaderFallbackManager управляет fallback при потере лидера.
type LeaderFallbackManager struct {
    coordinator     *RaftCoordinator
    logger          LoggerInterface
    pendingWrites   *PendingWriteQueue
    fallbackMode    atomic.Bool
    lastLeaderSeen  atomic.Int64
    electionTimeout time.Duration
    fallbackTimeout time.Duration
    mu              sync.RWMutex
    stopChan        chan struct{}
    wg              sync.WaitGroup
    observers       map[string]chan *LeaderChangeEvent
    observerMu      sync.RWMutex
    writeBuffer     []*WriteRequest
    bufferMu        sync.Mutex
}

// NewLeaderFallbackManager создаёт новый менеджер fallback.
func NewLeaderFallbackManager(coordinator *RaftCoordinator, logger LoggerInterface, config *FallbackConfig) *LeaderFallbackManager {
    if config == nil {
        config = DefaultFallbackConfig()
    }

    lfm := &LeaderFallbackManager{
        coordinator:     coordinator,
        logger:          logger,
        pendingWrites:   NewPendingWriteQueue(config.PendingQueueSize, config.PendingQueueMaxAge),
        electionTimeout: config.ElectionTimeout,
        fallbackTimeout: config.FallbackTimeout,
        stopChan:        make(chan struct{}),
        observers:       make(map[string]chan *LeaderChangeEvent),
        writeBuffer:     make([]*WriteRequest, 0, config.WriteBufferSize),
    }

    lfm.lastLeaderSeen.Store(time.Now().UnixMilli())

    lfm.wg.Add(1)
    go lfm.monitorLeader()

    lfm.wg.Add(1)
    go lfm.processFallbackWrites()

    return lfm
}

// monitorLeader отслеживает состояние лидера.
func (lfm *LeaderFallbackManager) monitorLeader() {
    defer lfm.wg.Done()

    ticker := time.NewTicker(lfm.electionTimeout / 2)
    defer ticker.Stop()

    var lastLeader string
    var leaderLostAt int64

    for {
        select {
        case <-lfm.stopChan:
            return
        case <-ticker.C:
            if lfm.coordinator == nil {
                continue
            }
            currentLeader := lfm.coordinator.GetLeader()
            currentLeaderID := ""
            if currentLeader != nil {
                currentLeaderID = currentLeader.ID
            }
            isLeader := lfm.coordinator.IsLeader()

            if currentLeaderID == "" && !isLeader {
                if !lfm.fallbackMode.Load() && leaderLostAt == 0 {
                    leaderLostAt = time.Now().UnixMilli()
                    lfm.enterFallbackMode()
                } else if leaderLostAt > 0 && time.Now().UnixMilli()-leaderLostAt > int64(lfm.fallbackTimeout.Milliseconds()) {
                    lfm.handleProlongedLeaderLoss()
                }
            } else {
                if lfm.fallbackMode.Load() {
                    lfm.exitFallbackMode()
                    lfm.processPendingWrites()
                }
                leaderLostAt = 0
                lfm.lastLeaderSeen.Store(time.Now().UnixMilli())
            }

            if lastLeader != currentLeaderID {
                if lastLeader != "" {
                    event := &LeaderChangeEvent{
                        OldLeader: lastLeader,
                        NewLeader: currentLeaderID,
                        Timestamp: time.Now().UnixMilli(),
                        Term:      lfm.coordinator.GetCurrentTerm(),
                    }
                    lfm.notifyObservers(event)

                    if lfm.logger != nil {
                        lfm.logger.Info(fmt.Sprintf("Leader changed from %s to %s", lastLeader, currentLeaderID))
                    }
                }
                lastLeader = currentLeaderID
            }
        }
    }
}

// enterFallbackMode переводит систему в fallback режим.
func (lfm *LeaderFallbackManager) enterFallbackMode() {
    if lfm.fallbackMode.CompareAndSwap(false, true) {
        if lfm.logger != nil {
            lfm.logger.Warn("Entering fallback mode - no leader available")
        }
    }
}

// exitFallbackMode выходит из fallback режима.
func (lfm *LeaderFallbackManager) exitFallbackMode() {
    if lfm.fallbackMode.CompareAndSwap(true, false) {
        if lfm.logger != nil {
            lfm.logger.Info("Exiting fallback mode - leader elected")
        }
    }
}

// handleProlongedLeaderLoss обрабатывает длительную потерю лидера.
func (lfm *LeaderFallbackManager) handleProlongedLeaderLoss() {
    if lfm.logger != nil {
        lfm.logger.Error("Prolonged leader loss detected, initiating emergency measures")
    }
}

// processPendingWrites обрабатывает отложенные записи.
func (lfm *LeaderFallbackManager) processPendingWrites() {
    requests := lfm.pendingWrites.GetAll()

    if len(requests) == 0 {
        return
    }

    if lfm.logger != nil {
        lfm.logger.Info(fmt.Sprintf("Processing %d pending writes after leader election", len(requests)))
    }
}

// processFallbackWrites обрабатывает записи в fallback режиме.
func (lfm *LeaderFallbackManager) processFallbackWrites() {
    defer lfm.wg.Done()

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

    for {
        select {
        case <-lfm.stopChan:
            return
        case <-ticker.C:
            if lfm.fallbackMode.Load() {
                lfm.bufferWrites()
            }
        }
    }
}

// bufferWrites буферизирует записи во время fallback.
func (lfm *LeaderFallbackManager) bufferWrites() {
    lfm.bufferMu.Lock()
    defer lfm.bufferMu.Unlock()

    if len(lfm.writeBuffer) == 0 {
        return
    }

    if lfm.coordinator == nil {
        return
    }
    if lfm.coordinator.GetLeader() != nil || lfm.coordinator.IsLeader() {
        for _, req := range lfm.writeBuffer {
            lfm.pendingWrites.Add(req)
        }
        lfm.writeBuffer = lfm.writeBuffer[:0]
        lfm.exitFallbackMode()
    }
}

// SubmitWrite отправляет запись с поддержкой fallback.
func (lfm *LeaderFallbackManager) SubmitWrite(data []byte) error {
    req := &WriteRequest{
        ID:        fmt.Sprintf("write_%d", time.Now().UnixNano()),
        Data:      data,
        CreatedAt: time.Now().UnixMilli(),
        Retries:   0,
    }

    if !lfm.fallbackMode.Load() && lfm.coordinator != nil && lfm.coordinator.IsLeader() {
        if lfm.coordinator.raft == nil {
            return fmt.Errorf("raft is not initialized")
        }
        future := lfm.coordinator.raft.Apply(data, 10*time.Second)
        if err := future.Error(); err != nil {
            return err
        }
        return nil
    }

    lfm.bufferMu.Lock()
    defer lfm.bufferMu.Unlock()

    lfm.writeBuffer = append(lfm.writeBuffer, req)

    return nil
}

// IsFallbackMode возвращает статус fallback режима.
func (lfm *LeaderFallbackManager) IsFallbackMode() bool {
    return lfm.fallbackMode.Load()
}

// notifyObservers уведомляет наблюдателей.
func (lfm *LeaderFallbackManager) notifyObservers(event *LeaderChangeEvent) {
    lfm.observerMu.RLock()
    defer lfm.observerMu.RUnlock()

    for _, ch := range lfm.observers {
        select {
        case ch <- event:
        default:
        }
    }
}

// GetStats возвращает статистику.
func (lfm *LeaderFallbackManager) GetStats() map[string]interface{} {
    return map[string]interface{}{
        "fallback_mode":    lfm.fallbackMode.Load(),
        "pending_writes":   lfm.pendingWrites.Size(),
        "buffered_writes":  len(lfm.writeBuffer),
        "last_leader_seen": lfm.lastLeaderSeen.Load(),
    }
}

// Stop останавливает fallback менеджер.
func (lfm *LeaderFallbackManager) Stop() {
    close(lfm.stopChan)
    lfm.wg.Wait()
}
// =============================================================================
// RAFT CLUSTER STATE
// =============================================================================

// RaftClusterState представляет состояние кластера для Raft FSM.
type RaftClusterState struct {
    Nodes             map[string]*NodeInfo   `json:"nodes"`
    ReplicationFactor int32                  `json:"replication_factor"`
    Shards            map[string]*RangeShard `json:"shards"`
    CurrentTerm       uint64                 `json:"current_term"`
    VotedFor          string                 `json:"voted_for"`
    CreatedAt         int64                  `json:"created_at"`
    UpdatedAt         int64                  `json:"updated_at"`
    mu                sync.RWMutex
}

// RaftFSM реализует конечный автомат для Raft.
type RaftFSM struct {
    state       *RaftClusterState
    logger      *log.Logger
    createdAt   int64
    coordinator *RaftCoordinator
}

// RaftSnapshot реализует интерфейс FSMSnapshot для Raft.
type RaftSnapshot struct {
    state *RaftClusterState
}

// =============================================================================
// RAFT COORDINATOR - ОСНОВНОЙ КООРДИНАТОР КЛАСТЕРА
// =============================================================================

// RaftCoordinator - основной координатор кластера.
type RaftCoordinator struct {
    raft               *raft.Raft
    fsm                *RaftFSM
    address            string
    raftAddr           string
    clusterName        string
    logger             *log.Logger
    config             *config.Config
    store              *storage.Storage
    stopChan           chan struct{}
    nodes              sync.Map
    replicationFactor  atomic.Int32
    replicationEnabled bool
    syncReplication    bool
    isLeader           atomic.Bool
    leaderMonitor      chan bool
    singleNodeMode     bool
    localNodeInfo      *NodeInfo
    logStore           *InmemStore
    stableStore        *InmemStore
    createdAt          int64
    leaderSince        atomic.Int64
    lastElection       atomic.Int64
    electionCount      atomic.Uint64
    currentTerm        atomic.Uint64
    shardManager       *RangeShardManager
    splitBrainDetector *SplitBrainDetector

    pipelineReplicator    *PipelineReplicator
    batchCommitManager    *BatchCommitManager
    reshardingManager     *ReshardingManager
    jointConsensusManager *JointConsensusManager

    recoveryManager *RecoveryManager
    persistenceMgr  *storage.PersistenceManager

    fallbackManager  *LeaderFallbackManager
    panicRecoveryMgr *PanicRecoveryManager
    schemaMigrator   *migration.SchemaMigrator

    sagaManager        *SagaManager
    tccManager         *TCCManager
    replicaReadManager *ReplicaReadManager
    multiRaftManager   *MultiRaftManager

    // GOSSIP PROTOCOL - АВТОМАТИЧЕСКОЕ ОБНАРУЖЕНИЕ УЗЛОВ
    gossipManager *GossipManager

    // SELF-HEALING - МЕХАНИЗМЫ САМОИСЦЕЛЕНИЯ
    selfHealingManager *SelfHealingManager

    // DYNAMIC CONFIG - ЦЕНТРАЛИЗОВАННОЕ УПРАВЛЕНИЕ КОНФИГУРАЦИЕЙ
    dynamicConfigManager *config.DynamicConfigManager

    // КРОСС-ДАТАЦЕНТРОВАЯ МИГРАЦИЯ
    crossDCMigrator *CrossDCMigrator

    // Логические часы для eventual consistency
    lamportClock *LamportClock

    // SAGA LEADER BRIDGE — реализация storage.SagaRaftAccessor.
    // Позволяет оркестратору SAGA (storage) корректно определять лидера,
    // если он создаётся через storage.NewSagaManagerWithRaft.
    // Для локального cluster.SagaManager не используется.
    sagaLeaderBridge *SagaLeaderBridge
}

// initRaftFSM создаёт и инициализирует FSM и его состояние.
//
// КРИТИЧЕСКОЕ ИСПРАВЛЕНИЕ: ранее coord.fsm не создавался в
// NewRaftCoordinator, что приводило к SIGSEGV (nil pointer dereference)
// при первом обращении к rc.fsm.state. Теперь FSM создаётся явно,
// до ветвления singleNodeMode / setupClusterMode.
func (rc *RaftCoordinator) initRaftFSM(replicationFactor int32) {
    now := time.Now().UnixMilli()

    rc.fsm = &RaftFSM{
        state: &RaftClusterState{
            Nodes:             make(map[string]*NodeInfo),
            ReplicationFactor: replicationFactor,
            Shards:            make(map[string]*RangeShard),
            CreatedAt:         now,
            UpdatedAt:         now,
        },
        logger:      rc.logger,
        createdAt:   now,
        coordinator: rc,
    }
}

// initRaftStores создаёт in-memory хранилища для Raft (logStore, stableStore).
//
// КРИТИЧЕСКОЕ ИСПРАВЛЕНИЕ: ранее coord.logStore и coord.stableStore
// не инициализировались, что приводило к nil pointer dereference
// в monitorLeadership (rc.stableStore.Set(...)).
func (rc *RaftCoordinator) initRaftStores(dataDir string) {
    if dataDir == "" {
        dataDir = "raft_data"
    }
    _ = os.MkdirAll(dataDir, 0755)

    rc.logStore = NewInmemStore(filepath.Join(dataDir, "raft-log.json"))
    rc.stableStore = NewInmemStore(filepath.Join(dataDir, "raft-stable.json"))
}

// NewRaftCoordinator создаёт новый координатор Raft.
//
// ИСПРАВЛЕНО (компиляция):
//   Ранее здесь вызывалась storage.NewSagaManagerWithRaft и результат
//   присваивался полю *cluster.SagaManager, что давало ошибку
//   "cannot use *storage.SagaManager as *cluster.SagaManager".
//   Теперь создаётся локальный cluster.SagaManager через NewSagaManager.
//   Он уже имеет собственные GetOrchestrator, GetMetrics, GetActiveSagas
//   и Stop — то, что ожидают repl.go и main.go.
//   Bridge остаётся отдельным компонентом (sagaLeaderBridge) и
//   доступен через GetSagaLeaderBridge() для тех, кто хочет
//   использовать storage.SagaLeaderElector.
func NewRaftCoordinator(cfg *config.Config, store *storage.Storage, logger *log.Logger) (*RaftCoordinator, error) {
    if logger == nil {
        return nil, fmt.Errorf("logger is required")
    }

    if cfg == nil {
        return nil, fmt.Errorf("config is required")
    }

    coord := &RaftCoordinator{
        config:             cfg,
        store:              store,
        logger:             logger,
        clusterName:        cfg.Cluster.Name,
        stopChan:           make(chan struct{}),
        leaderMonitor:      make(chan bool, 10),
        createdAt:          time.Now().UnixMilli(),
        replicationEnabled: cfg.Replication.Enabled,
        syncReplication:    false,
        lamportClock:       NewLamportClock(cfg.Cluster.NodeIP),
    }

    // Используем ReplicationFactor из конфигурации
    replicationFactor := 1
    if cfg.Replication.Enabled {
        replicationFactor = 2
    }
    coord.replicationFactor.Store(int32(replicationFactor))

    // КРИТИЧЕСКОЕ ИСПРАВЛЕНИЕ: инициализируем logStore, stableStore и FSM
    // ДО первого обращения к coord.fsm.state (в т.ч. в singleNodeMode ветке
    // и в setupClusterMode).
    coord.initRaftStores(cfg.Cluster.RaftDataDir)
    coord.initRaftFSM(int32(replicationFactor))

    coord.shardManager = NewRangeShardManager(logger)
    coord.splitBrainDetector = NewSplitBrainDetector(logger, true, 60*time.Second)

    coord.schemaMigrator = migration.NewSchemaMigrator(store, logger, "futriis/migrations")
    coord.panicRecoveryMgr = NewPanicRecoveryManager(logger)
    coord.fallbackManager = NewLeaderFallbackManager(coord, logger, nil)

    // ДОБАВЛЕНО: создаём bridge для выборов лидера SAGA.
    // Bridge реализует storage.SagaRaftAccessor.
    // Он не заменяет cluster.SagaManager, а живёт отдельно.
    coord.sagaLeaderBridge = NewSagaLeaderBridge(coord)
    coord.sagaLeaderBridge.Start()

    // Создаём локальный кластерный SAGA-менеджер.
    // В кластерной версии он сам выполняет роль оркестратора
    // (см. GetOrchestrator), и его GetMetrics/GetActiveSagas/Stop
    // используются в repl.go и main.go.
    coord.sagaManager = NewSagaManager(logger)

    coord.tccManager = NewTCCManager(logger)
    coord.replicaReadManager = NewReplicaReadManager(coord, logger)
    coord.multiRaftManager = NewMultiRaftManager(store, logger)

    coord.pipelineReplicator = NewPipelineReplicator(coord, 100, 100*time.Millisecond, logger)
    coord.batchCommitManager = NewBatchCommitManager(50, 50*time.Millisecond, true, logger)
    coord.reshardingManager = NewReshardingManager(coord, logger)
    coord.jointConsensusManager = NewJointConsensusManager(coord, logger)

    coord.recoveryManager = NewRecoveryManager(coord, logger)

    coord.persistenceMgr = storage.NewPersistenceManager(nil, store, logger)
    coord.persistenceMgr.Start()

    coord.singleNodeMode = len(cfg.Cluster.Nodes) <= 1

    if coord.singleNodeMode {
        logger.Info("Running in single-node mode")
        coord.localNodeInfo = &NodeInfo{
            ID:        "local",
            IP:        cfg.Cluster.NodeIP,
            Port:      cfg.Cluster.NodePort,
            Status:    "active",
            JoinedAt:  time.Now().UnixMilli(),
            UpdatedAt: time.Now().UnixMilli(),
            Version:   1,
        }
        // Добавляем локальный узел в FSM state, чтобы GetActiveNodes /
        // GetAllNodes / GetNodeByID работали корректно в single-node режиме.
        coord.fsm.state.mu.Lock()
        coord.fsm.state.Nodes[coord.localNodeInfo.ID] = coord.localNodeInfo
        coord.fsm.state.UpdatedAt = time.Now().UnixMilli()
        coord.fsm.state.mu.Unlock()

        // В single-node режиме мы сразу лидер — сообщаем bridge.
        coord.sagaLeaderBridge.NotifyLeadership(true, coord.localNodeInfo.ID, 1)
    } else {
        logger.Info("Running in cluster mode")
        if err := coord.setupClusterMode(); err != nil {
            return nil, fmt.Errorf("failed to setup cluster mode: %v", err)
        }
    }

    // ИНИЦИАЛИЗАЦИЯ GOSSIP PROTOCOL
    gossipConfig := DefaultGossipConfig()
    coord.gossipManager = NewGossipManager(gossipConfig, coord, logger)
    if err := coord.gossipManager.Start(); err != nil {
        logger.Warn(fmt.Sprintf("Failed to start gossip protocol: %v", err))
    }

    // ИНИЦИАЛИЗАЦИЯ SELF-HEALING
    healingConfig := DefaultHealingConfig()
    coord.selfHealingManager = NewSelfHealingManager(healingConfig, coord, coord.gossipManager, store, logger)
    coord.selfHealingManager.Start()

    // ИНИЦИАЛИЗАЦИЯ DYNAMIC CONFIG
    coord.dynamicConfigManager = config.NewDynamicConfigManager(cfg, logger)
    coord.dynamicConfigManager.Start()

    // ИНИЦИАЛИЗАЦИЯ КРОСС-ДАТАЦЕНТРОВОГО МИГРАТОРА
    if cfg.Migration.Enabled {
        coord.crossDCMigrator = NewCrossDCMigrator(&cfg.Migration, store, logger)
        coord.crossDCMigrator.Start()
        logger.Info("Cross-datacenter migrator initialized")
    } else {
        logger.Debug("Cross-datacenter migration is disabled in config")
    }

    coord.shardManager.Start()
    coord.replicaReadManager.Start()
    coord.recoveryManager.Start()

    if !coord.singleNodeMode {
        go coord.monitorLeadership()
        go coord.rebalanceMonitor()
    }

    logger.Info("Raft coordinator initialized successfully")

    return coord, nil
}

// setupClusterMode настраивает кластерный режим.
func (rc *RaftCoordinator) setupClusterMode() error {
    // TODO: Реализовать настройку кластера
    return nil
}

// =============================================================================
// МЕТОДЫ ДЛЯ ДОСТУПА К КОМПОНЕНТАМ
// =============================================================================

// GetSchemaMigrator возвращает менеджер миграций схемы.
func (rc *RaftCoordinator) GetSchemaMigrator() *migration.SchemaMigrator {
    return rc.schemaMigrator
}

// GetFallbackManager возвращает менеджер fallback.
func (rc *RaftCoordinator) GetFallbackManager() *LeaderFallbackManager {
    return rc.fallbackManager
}

// GetPanicRecoveryManager возвращает менеджер восстановления после паник.
func (rc *RaftCoordinator) GetPanicRecoveryManager() *PanicRecoveryManager {
    return rc.panicRecoveryMgr
}

// GetPersistenceManager возвращает менеджер персистентности.
func (rc *RaftCoordinator) GetPersistenceManager() *storage.PersistenceManager {
    return rc.persistenceMgr
}

// GetFallbackStats возвращает статистику fallback менеджера.
func (rc *RaftCoordinator) GetFallbackStats() map[string]interface{} {
    if rc.fallbackManager == nil {
        return map[string]interface{}{
            "enabled": false,
            "message": "Fallback manager not initialized",
        }
    }
    return rc.fallbackManager.GetStats()
}

// GetPanicRecoveryStats возвращает статистику восстановления после паник.
func (rc *RaftCoordinator) GetPanicRecoveryStats() map[string]interface{} {
    if rc.panicRecoveryMgr == nil {
        return map[string]interface{}{
            "enabled": false,
            "message": "Panic recovery manager not initialized",
        }
    }
    return rc.panicRecoveryMgr.GetStats()
}

// GetMigrationStatus возвращает статус миграций.
func (rc *RaftCoordinator) GetMigrationStatus() *migration.MigrationStatus {
    if rc.schemaMigrator == nil {
        return nil
    }
    status := rc.schemaMigrator.GetStatus()
    return status
}

// GetSagaLeaderBridge возвращает bridge выборов лидера SAGA.
// Полезно для диагностики и для передачи в другие компоненты
// (например, если кто-то хочет использовать storage.SagaLeaderElector).
func (rc *RaftCoordinator) GetSagaLeaderBridge() *SagaLeaderBridge {
    return rc.sagaLeaderBridge
}

// =============================================================================
// МЕТОДЫ ДЛЯ КРОСС-ДАТАЦЕНТРОВОЙ МИГРАЦИИ
// =============================================================================

// GetCrossDCMigrator возвращает кросс-датацентровый мигратор
func (rc *RaftCoordinator) GetCrossDCMigrator() *CrossDCMigrator {
    return rc.crossDCMigrator
}

// IsMigrationEnabled проверяет, включена ли миграция
func (rc *RaftCoordinator) IsMigrationEnabled() bool {
    return rc.crossDCMigrator != nil && rc.config.Migration.Enabled
}

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

// GetGossipManager возвращает менеджер gossip протокола
func (rc *RaftCoordinator) GetGossipManager() *GossipManager {
    return rc.gossipManager
}

// GetSelfHealingManager возвращает менеджер самоисцеления
func (rc *RaftCoordinator) GetSelfHealingManager() *SelfHealingManager {
    return rc.selfHealingManager
}

// GetDynamicConfigManager возвращает менеджер динамической конфигурации
func (rc *RaftCoordinator) GetDynamicConfigManager() *config.DynamicConfigManager {
    return rc.dynamicConfigManager
}

// =============================================================================
// ОСТАЛЬНЫЕ МЕТОДЫ RAFT COORDINATOR
// =============================================================================

// GetShardManager возвращает менеджер диапазонных шардов.
func (rc *RaftCoordinator) GetShardManager() *RangeShardManager {
    return rc.shardManager
}

// GetSagaManager возвращает менеджер Saga.
func (rc *RaftCoordinator) GetSagaManager() *SagaManager {
    return rc.sagaManager
}

// GetTCCManager возвращает менеджер TCC.
func (rc *RaftCoordinator) GetTCCManager() *TCCManager {
    return rc.tccManager
}

// GetReplicaReadManager возвращает менеджер чтения с реплик.
func (rc *RaftCoordinator) GetReplicaReadManager() *ReplicaReadManager {
    return rc.replicaReadManager
}

// GetMultiRaftManager возвращает менеджер Multi-Raft.
func (rc *RaftCoordinator) GetMultiRaftManager() *MultiRaftManager {
    return rc.multiRaftManager
}

// GetShardForCollection возвращает шард для коллекции.
func (rc *RaftCoordinator) GetShardForCollection(database, collection string) *RangeShard {
    key := fmt.Sprintf("%s:%s", database, collection)
    return rc.shardManager.GetShard(key)
}

// GetAllShards возвращает все шарды.
func (rc *RaftCoordinator) GetAllShards() []*RangeShard {
    return rc.shardManager.GetAllShards()
}

// ExecuteSaga выполняет Saga транзакцию.
func (rc *RaftCoordinator) ExecuteSaga(id string, setup func(*SagaTransaction)) error {
    saga := rc.sagaManager.BeginSaga(id)
    setup(saga)
    return rc.sagaManager.Execute(saga)
}

// ExecuteTCC выполняет TCC транзакцию.
func (rc *RaftCoordinator) ExecuteTCC(id string, tryData map[string]interface{}, confirm, cancel func() error) error {
    tcc := rc.tccManager.BeginTCC(id)
    tcc.ConfirmFn = confirm
    tcc.CancelFn = cancel

    if err := tcc.Try(tryData); err != nil {
        return err
    }

    return tcc.Confirm()
}

// WriteToShardWithRaftGroup выполняет запись в шард через Multi-Raft группу.
func (rc *RaftCoordinator) WriteToShardWithRaftGroup(shardID string, database, collection string, docData map[string]interface{}) error {
    raftGroup, err := rc.multiRaftManager.GetOrCreateRaftGroup(shardID, nil)
    if err != nil {
        return fmt.Errorf("failed to get raft group: %v", err)
    }

    cmd := map[string]interface{}{
        "type":       "write",
        "database":   database,
        "collection": collection,
        "document":   docData,
    }

    data, err := json.Marshal(cmd)
    if err != nil {
        return fmt.Errorf("failed to marshal command: %v", err)
    }

    future := raftGroup.Apply(data, 10*time.Second)
    if err := future.Error(); err != nil {
        return err
    }
    return nil
}

// GetActiveNodes возвращает активные узлы.
func (rc *RaftCoordinator) GetActiveNodes() []*NodeInfo {
    nodes := make([]*NodeInfo, 0)

    if rc.fsm == nil || rc.fsm.state == nil {
        if rc.singleNodeMode && rc.localNodeInfo != nil {
            return []*NodeInfo{rc.localNodeInfo}
        }
        return nodes
    }

    now := time.Now().UnixMilli()

    state := rc.fsm.state
    state.mu.RLock()
    defer state.mu.RUnlock()

    for _, nodeInfo := range state.Nodes {
        if now-nodeInfo.LastSeen < 30000 && nodeInfo.Status == "active" {
            if rc.splitBrainDetector == nil || !rc.splitBrainDetector.IsQuarantined(nodeInfo.ID) {
                nodes = append(nodes, nodeInfo)
            }
        }
    }

    if rc.singleNodeMode && len(nodes) == 0 && rc.localNodeInfo != nil {
        nodes = append(nodes, rc.localNodeInfo)
    }

    return nodes
}

// GetAllNodes возвращает все узлы.
func (rc *RaftCoordinator) GetAllNodes() []*NodeInfo {
    if rc.fsm == nil || rc.fsm.state == nil {
        if rc.singleNodeMode && rc.localNodeInfo != nil {
            return []*NodeInfo{rc.localNodeInfo}
        }
        return []*NodeInfo{}
    }

    state := rc.fsm.state
    state.mu.RLock()
    defer state.mu.RUnlock()

    nodes := make([]*NodeInfo, 0, len(state.Nodes))
    for _, node := range state.Nodes {
        nodes = append(nodes, node)
    }

    if rc.singleNodeMode && len(nodes) == 0 && rc.localNodeInfo != nil {
        nodes = append(nodes, rc.localNodeInfo)
    }

    return nodes
}

// GetNodeByID возвращает узел по ID.
func (rc *RaftCoordinator) GetNodeByID(nodeID string) *NodeInfo {
    if rc.fsm == nil || rc.fsm.state == nil {
        return nil
    }

    state := rc.fsm.state
    state.mu.RLock()
    defer state.mu.RUnlock()

    if node, ok := state.Nodes[nodeID]; ok {
        return node
    }
    return nil
}

// GetLeader возвращает лидера.
func (rc *RaftCoordinator) GetLeader() *NodeInfo {
    if rc.singleNodeMode {
        return rc.localNodeInfo
    }

    if rc.raft == nil {
        return nil
    }

    leaderAddr := rc.raft.Leader()
    if leaderAddr == "" {
        return nil
    }

    if rc.fsm == nil || rc.fsm.state == nil {
        return nil
    }

    state := rc.fsm.state
    state.mu.RLock()
    defer state.mu.RUnlock()

    for _, node := range state.Nodes {
        nodeAddr := fmt.Sprintf("%s:%d", node.IP, node.Port)
        if nodeAddr == string(leaderAddr) {
            return node
        }
    }
    return nil
}

// IsLeader проверяет, является ли текущий узел лидером.
// Проверяет реальное состояние Raft, а не кэшированное значение.
func (rc *RaftCoordinator) IsLeader() bool {
    if rc.singleNodeMode {
        return true
    }
    if rc.raft == nil {
        return false
    }
    // Всегда проверяем реальное состояние Raft
    isLeader := rc.raft.State() == raft.Leader
    // Обновляем кэш
    rc.isLeader.Store(isLeader)
    return isLeader
}

// GetCurrentTerm возвращает текущий терм Raft.
func (rc *RaftCoordinator) GetCurrentTerm() uint64 {
    return rc.currentTerm.Load()
}

// GetLeaderSince возвращает время начала лидерства.
func (rc *RaftCoordinator) GetLeaderSince() int64 {
    return rc.leaderSince.Load()
}

// GetElectionCount возвращает количество выборов.
func (rc *RaftCoordinator) GetElectionCount() uint64 {
    return rc.electionCount.Load()
}

// SendHeartbeat обновляет heartbeat узла.
// Использует монотонное время и логические часы для порядка.
func (rc *RaftCoordinator) SendHeartbeat(nodeID string) {
    now := monotonicTimestamp()

    if val, ok := rc.nodes.Load(nodeID); ok {
        nodeInfo := val.(*NodeInfo)
        nodeInfo.LastSeen = now
        nodeInfo.UpdatedAt = now
        rc.nodes.Store(nodeID, nodeInfo)
    }

    if rc.fsm == nil || rc.fsm.state == nil {
        return
    }

    rc.fsm.state.mu.Lock()
    if nodeInfo, ok := rc.fsm.state.Nodes[nodeID]; ok {
        nodeInfo.LastSeen = now
        nodeInfo.UpdatedAt = now
    }
    rc.fsm.state.mu.Unlock()
}

// UpdateNodeStatus обновляет статус узла через Raft.
func (rc *RaftCoordinator) UpdateNodeStatus(nodeID string, status NodeStatus) error {
    now := monotonicTimestamp()

    if rc.splitBrainDetector != nil && rc.splitBrainDetector.IsQuarantined(nodeID) {
        return fmt.Errorf("node %s is quarantined, cannot update status", nodeID)
    }

    if rc.fsm == nil || rc.fsm.state == nil {
        return fmt.Errorf("raft FSM is not initialized")
    }

    if rc.singleNodeMode {
        rc.fsm.state.mu.Lock()
        if node, ok := rc.fsm.state.Nodes[nodeID]; ok {
            node.Status = mapStatusToString(int32(status))
            node.UpdatedAt = now
        }
        rc.fsm.state.mu.Unlock()
        return nil
    }

    if !rc.IsLeader() {
        return fmt.Errorf("node is not the leader")
    }

    cmd := NodeStatusCommand{
        Type:      "update_status",
        NodeID:    nodeID,
        Status:    int32(status),
        Timestamp: now,
    }

    data, err := json.Marshal(cmd)
    if err != nil {
        return err
    }

    future := rc.raft.Apply(data, rc.config.Replication.GetReplicationTimeout())
    if err := future.Error(); err != nil {
        return err
    }
    return nil
}

// GetClusterStatus возвращает статус кластера.
func (rc *RaftCoordinator) GetClusterStatus() *ClusterStatus {
    nodes := rc.GetAllNodes()
    activeNodes := rc.GetActiveNodes()

    syncingNodes := 0
    for _, node := range nodes {
        if node.Status == "syncing" {
            syncingNodes++
        }
    }

    leader := rc.GetLeader()
    leaderID := ""
    if leader != nil {
        leaderID = leader.ID
    }

    now := time.Now().UnixMilli()

    health := rc.calculateHealth()

    if rc.splitBrainDetector != nil && rc.splitBrainDetector.Detect(rc.currentTerm.Load(), leaderID, len(nodes)) {
        health = "split_brain"
    }

    return &ClusterStatus{
        Name:                 rc.clusterName,
        TotalNodes:           len(nodes),
        ActiveNodes:          len(activeNodes),
        SyncingNodes:         syncingNodes,
        FailedNodes:          len(nodes) - len(activeNodes),
        ReplicationFactor:    int(rc.replicationFactor.Load()),
        LeaderID:             leaderID,
        Health:               health,
        CreatedAt:            rc.createdAt,
        UpdatedAt:            now,
        PipelineEnabled:      rc.pipelineReplicator != nil,
        BatchCommitEnabled:   rc.batchCommitManager != nil,
        ReshardingEnabled:    rc.reshardingManager != nil,
        JointConsensusActive: rc.jointConsensusManager != nil && rc.jointConsensusManager.IsJointConsensusActive(),
        FallbackMode:         rc.fallbackManager != nil && rc.fallbackManager.IsFallbackMode(),
    }
}

// calculateHealth вычисляет здоровье кластера.
func (rc *RaftCoordinator) calculateHealth() string {
    activeNodes := rc.GetActiveNodes()
    totalNodes := rc.GetAllNodes()

    if len(totalNodes) == 0 {
        return "critical"
    }

    ratio := float64(len(activeNodes)) / float64(len(totalNodes))
    if ratio >= 0.8 {
        return "healthy"
    } else if ratio >= 0.5 {
        return "degraded"
    }
    return "critical"
}

// GetReplicationFactor возвращает фактор репликации.
func (rc *RaftCoordinator) GetReplicationFactor() int {
    return int(rc.replicationFactor.Load())
}

// SetReplicationFactor устанавливает фактор репликации.
func (rc *RaftCoordinator) SetReplicationFactor(factor int) error {
    if factor < 1 || factor > 5 {
        return fmt.Errorf("replication factor must be between 1 and 5")
    }

    if !rc.IsLeader() {
        return fmt.Errorf("node is not the leader")
    }

    if rc.fsm == nil || rc.fsm.state == nil {
        return fmt.Errorf("raft FSM is not initialized")
    }

    oldFactor := rc.replicationFactor.Load()
    rc.replicationFactor.Store(int32(factor))

    rc.fsm.state.mu.Lock()
    rc.fsm.state.ReplicationFactor = int32(factor)
    rc.fsm.state.UpdatedAt = time.Now().UnixMilli()
    rc.fsm.state.mu.Unlock()

    if rc.logger != nil {
        rc.logger.Info(fmt.Sprintf("Replication factor changed from %d to %d", oldFactor, factor))
    }

    return nil
}

// TriggerResharding запускает перераспределение шардов.
func (rc *RaftCoordinator) TriggerResharding(reason string) error {
    if rc.reshardingManager == nil {
        return fmt.Errorf("resharding manager not initialized")
    }
    return rc.reshardingManager.TriggerResharding(reason)
}

// GetPipelineStats возвращает статистику пайплайна.
func (rc *RaftCoordinator) GetPipelineStats() map[string]interface{} {
    if rc.pipelineReplicator == nil {
        return map[string]interface{}{
            "enabled": false,
            "message": "Pipeline replicator not initialized",
        }
    }

    return map[string]interface{}{
        "enabled":         true,
        "batch_size":      rc.pipelineReplicator.batchSize,
        "batch_timeout":   rc.pipelineReplicator.batchTimeout.String(),
        "pending_batches": len(rc.pipelineReplicator.pendingBatches),
        "batch_count":     rc.pipelineReplicator.batchCount.Load(),
        "commands_count":  rc.pipelineReplicator.commandsCount.Load(),
    }
}

// GetBatchCommitStats возвращает статистику пакетных коммитов.
func (rc *RaftCoordinator) GetBatchCommitStats() map[string]interface{} {
    if rc.batchCommitManager == nil {
        return map[string]interface{}{
            "enabled": false,
            "message": "Batch commit manager not initialized",
        }
    }

    return map[string]interface{}{
        "enabled":          true,
        "batch_size":       rc.batchCommitManager.batchSize,
        "commit_interval":  rc.batchCommitManager.commitInterval.String(),
        "fsync_enabled":    rc.batchCommitManager.fsyncEnabled,
        "pending_commits":  len(rc.batchCommitManager.pendingCommits),
        "commit_count":     rc.batchCommitManager.commitCount.Load(),
        "operations_count": rc.batchCommitManager.operationsCount.Load(),
        "last_flush":       rc.batchCommitManager.storage.lastFlush,
        "flush_count":      rc.batchCommitManager.storage.flushCount,
        "stored_commits":   len(rc.batchCommitManager.storage.commits),
    }
}

// GetReshardingStats возвращает статистику решардинга.
func (rc *RaftCoordinator) GetReshardingStats() map[string]interface{} {
    if rc.reshardingManager == nil {
        return map[string]interface{}{
            "enabled": false,
            "message": "Resharding manager not initialized",
        }
    }

    metrics := rc.reshardingManager.metrics
    metrics.mu.RLock()
    defer metrics.mu.RUnlock()

    return map[string]interface{}{
        "enabled":               true,
        "total_reshardings":     metrics.TotalReshardings.Load(),
        "total_documents_moved": metrics.TotalDocumentsMoved.Load(),
        "total_bytes_moved":     metrics.TotalBytesMoved.Load(),
        "failed_reshardings":    metrics.FailedReshardings.Load(),
        "last_resharding_time":  metrics.LastReshardingTime.Load(),
        "history_count":         len(metrics.history),
        "queue_size":            len(rc.reshardingManager.reshardingChan),
        "is_resharding":         rc.reshardingManager.isResharding.Load(),
    }
}

// GetJointConsensusStatus возвращает статус совместного консенсуса.
func (rc *RaftCoordinator) GetJointConsensusStatus() map[string]interface{} {
    if rc.jointConsensusManager == nil {
        return map[string]interface{}{
            "active":  false,
            "message": "Joint consensus manager not initialized",
        }
    }
    return rc.jointConsensusManager.GetJointConsensusStatus()
}

// monitorLeadership отслеживает изменения лидера.
// Использует реальное состояние Raft и защиту от ложных срабатываний.
//
// ДОБАВЛЕНО: при смене лидерства вызываем sagaLeaderBridge.NotifyLeadership,
// чтобы bridge знал, кто лидер. Это используется, если кто-то создаст
// storage.SagaLeaderElector на базе этого bridge.
func (rc *RaftCoordinator) monitorLeadership() {
    ticker := time.NewTicker(rc.config.Cluster.GetHeartbeatTimeout() / 2)
    defer ticker.Stop()

    wasLeader := false

    for {
        select {
        case <-rc.stopChan:
            return
        case <-ticker.C:
            if rc.raft == nil {
                continue
            }
            // Проверяем реальное состояние Raft
            isLeader := rc.raft.State() == raft.Leader

            // ДОБАВЛЕНО: уведомляем bridge о текущем лидерстве.
            leaderID := ""
            if leader := rc.GetLeader(); leader != nil {
                leaderID = leader.ID
            }
            if rc.sagaLeaderBridge != nil {
                rc.sagaLeaderBridge.NotifyLeadership(isLeader, leaderID, rc.currentTerm.Load())
            }

            if isLeader != wasLeader {
                wasLeader = isLeader
                select {
                case rc.leaderMonitor <- isLeader:
                default:
                }
                if isLeader {
                    rc.isLeader.Store(true)
                    newTerm := rc.currentTerm.Add(1)
                    rc.leaderSince.Store(time.Now().UnixMilli())
                    rc.electionCount.Add(1)
                    if rc.fsm != nil && rc.fsm.state != nil {
                        rc.fsm.state.CurrentTerm = newTerm
                    }
                    if rc.stableStore != nil {
                        rc.stableStore.Set([]byte("currentTerm"), []byte(fmt.Sprintf("%d", newTerm)))
                    }
                    rc.logger.Debug(fmt.Sprintf("Leadership acquired at term %d (election #%d)",
                        newTerm, rc.electionCount.Load()))

                    nodes := rc.GetAllNodes()
                    for _, node := range nodes {
                        rc.shardManager.AddNode(node.ID)
                    }
                } else {
                    rc.isLeader.Store(false)
                    rc.lastElection.Store(time.Now().UnixMilli())
                    rc.logger.Debug("Leadership lost")
                }
            }
        }
    }
}

// rebalanceMonitor периодически проверяет необходимость ребалансировки.
func (rc *RaftCoordinator) rebalanceMonitor() {
    ticker := time.NewTicker(5 * time.Minute)
    defer ticker.Stop()

    for {
        select {
        case <-rc.stopChan:
            return
        case <-ticker.C:
            if rc.IsLeader() && rc.reshardingManager != nil {
                rc.reshardingManager.TriggerResharding("periodic_rebalance")
            }
        }
    }
}

// Stop останавливает координатор.
//
// ДОБАВЛЕНО: останавливаем sagaLeaderBridge и sagaManager.
// ИСПРАВЛЕНО: sagaManager теперь *cluster.SagaManager, у которого есть Stop().
func (rc *RaftCoordinator) Stop() {
    now := time.Now().UnixMilli()

    rc.logger.Info("Stopping Raft coordinator...")

    if rc.pipelineReplicator != nil {
        rc.pipelineReplicator.Stop()
        rc.logger.Debug("Pipeline replicator stopped")
    }
    if rc.batchCommitManager != nil {
        rc.batchCommitManager.Stop()
        rc.logger.Debug("Batch commit manager stopped")
    }
    if rc.reshardingManager != nil {
        rc.reshardingManager.Stop()
        rc.logger.Debug("Resharding manager stopped")
    }
    if rc.recoveryManager != nil {
        rc.recoveryManager.Stop()
        rc.logger.Debug("Recovery manager stopped")
    }
    if rc.persistenceMgr != nil {
        rc.persistenceMgr.Stop()
        rc.logger.Debug("Persistence manager stopped")
    }
    if rc.fallbackManager != nil {
        rc.fallbackManager.Stop()
        rc.logger.Debug("Fallback manager stopped")
    }
    if rc.panicRecoveryMgr != nil {
        rc.panicRecoveryMgr.Stop()
        rc.logger.Debug("Panic recovery manager stopped")
    }
    if rc.replicaReadManager != nil {
        rc.replicaReadManager.Stop()
        rc.logger.Debug("Replica read manager stopped")
    }
    if rc.shardManager != nil {
        rc.shardManager.Stop()
        rc.logger.Debug("Range shard manager stopped")
    }
    if rc.gossipManager != nil {
        rc.gossipManager.Stop()
        rc.logger.Debug("Gossip manager stopped")
    }
    if rc.selfHealingManager != nil {
        rc.selfHealingManager.Stop()
        rc.logger.Debug("Self-healing manager stopped")
    }
    if rc.dynamicConfigManager != nil {
        rc.dynamicConfigManager.Stop()
        rc.logger.Debug("Dynamic config manager stopped")
    }
    if rc.crossDCMigrator != nil {
        rc.crossDCMigrator.Stop()
        rc.logger.Debug("Cross-datacenter migrator stopped")
    }
    if rc.sagaManager != nil {
        rc.sagaManager.Stop()
        rc.logger.Debug("Saga manager stopped")
    }
    if rc.sagaLeaderBridge != nil {
        rc.sagaLeaderBridge.Stop()
        rc.logger.Debug("Saga leader bridge stopped")
    }

    close(rc.stopChan)
    if rc.raft != nil {
        rc.raft.Shutdown()
    }

    rc.logger.Info(fmt.Sprintf("Raft coordinator stopped at %s", time.UnixMilli(now).Format("2006-01-02 15:04:05.000")))
}

// IsReplicationEnabled возвращает статус репликации.
func (rc *RaftCoordinator) IsReplicationEnabled() bool {
    return rc.replicationEnabled
}

// IsSyncReplicationEnabled возвращает статус синхронной репликации.
func (rc *RaftCoordinator) IsSyncReplicationEnabled() bool {
    return rc.syncReplication
}

// RegisterNode регистрирует узел в кластере.
func (rc *RaftCoordinator) RegisterNode(node *Node) error {
    if node == nil {
        return fmt.Errorf("node is nil")
    }
    if rc.fsm == nil || rc.fsm.state == nil {
        return fmt.Errorf("raft FSM is not initialized")
    }

    now := monotonicTimestamp()

    if rc.splitBrainDetector != nil && rc.splitBrainDetector.IsQuarantined(node.ID) {
        return fmt.Errorf("node %s is quarantined due to previous split-brain", node.ID)
    }

    nodeInfo := &NodeInfo{
        ID:        node.ID,
        IP:        node.IP,
        Port:      node.Port,
        Status:    "active",
        LastSeen:  now,
        JoinedAt:  now,
        UpdatedAt: now,
        Version:   1,
    }

    if rc.singleNodeMode {
        if rc.logger != nil {
            rc.logger.Debug("Single-node mode: registering node without Raft consensus")
        }
        rc.nodes.Store(node.ID, nodeInfo)

        rc.fsm.state.mu.Lock()
        rc.fsm.state.Nodes[node.ID] = nodeInfo
        rc.fsm.state.UpdatedAt = now
        rc.fsm.state.mu.Unlock()

        rc.shardManager.AddNode(node.ID)
        return nil
    }

    if !rc.IsLeader() {
        leader := rc.GetLeader()
        if leader != nil {
            return fmt.Errorf("node is not the leader. Please connect to leader at %s:%d", leader.IP, leader.Port)
        }
        return fmt.Errorf("node is not the leader and no leader found")
    }

    cmd := NodeRegistrationCommand{
        Type:      "register",
        Node:      *nodeInfo,
        Timestamp: now,
    }

    data, err := json.Marshal(cmd)
    if err != nil {
        return err
    }

    future := rc.raft.Apply(data, rc.config.Replication.GetReplicationTimeout())
    if err := future.Error(); err != nil {
        return fmt.Errorf("failed to register node via raft: %v", err)
    }

    rc.nodes.Store(node.ID, nodeInfo)
    rc.shardManager.AddNode(node.ID)
    return nil
}

// RemoveNode удаляет узел из кластера.
func (rc *RaftCoordinator) RemoveNode(nodeID string) error {
    if rc.fsm == nil || rc.fsm.state == nil {
        return fmt.Errorf("raft FSM is not initialized")
    }

    now := monotonicTimestamp()

    if rc.singleNodeMode {
        rc.nodes.Delete(nodeID)
        rc.fsm.state.mu.Lock()
        delete(rc.fsm.state.Nodes, nodeID)
        rc.fsm.state.UpdatedAt = now
        rc.fsm.state.mu.Unlock()
        rc.shardManager.RemoveNode(nodeID)
        return nil
    }

    if !rc.IsLeader() {
        return fmt.Errorf("node is not the leader")
    }

    cmd := NodeRegistrationCommand{
        Type:      "remove",
        NodeID:    nodeID,
        Timestamp: now,
    }

    data, err := json.Marshal(cmd)
    if err != nil {
        return err
    }

    future := rc.raft.Apply(data, rc.config.Replication.GetReplicationTimeout())
    if err := future.Error(); err != nil {
        return fmt.Errorf("failed to remove node via raft: %v", err)
    }

    rc.nodes.Delete(nodeID)
    rc.shardManager.RemoveNode(nodeID)
    return nil
}

// HandleStatusSync обрабатывает синхронизацию статуса.
// Не инициирует LeadershipTransfer без проверки реального состояния Raft.
func (rc *RaftCoordinator) HandleStatusSync(leaderID string, term uint64, clusterSize int) {
    if rc.fsm == nil || rc.fsm.state == nil {
        return
    }

    if rc.splitBrainDetector != nil && rc.splitBrainDetector.Detect(term, leaderID, clusterSize) {
        // Проверяем, что мы действительно в состоянии split-brain
        // и только тогда принимаем меры
        if rc.raft == nil {
            return
        }

        currentState := rc.raft.State()
        if currentState == raft.Leader && rc.localNodeInfo != nil && leaderID != rc.localNodeInfo.ID {
            // Мы лидер, но получили информацию о другом лидере — это split-brain
            candidates := make(map[string]uint64)
            candidates[leaderID] = rc.getCommitIndex()
            candidates[rc.localNodeInfo.ID] = rc.getCommitIndex()

            winner := rc.splitBrainDetector.Resolve(term, candidates)
            if winner == rc.localNodeInfo.ID {
                // Мы должны остаться лидером, но нужно изолировать другой узел
                rc.splitBrainDetector.QuarantineNode(leaderID)
                if rc.logger != nil {
                    rc.logger.Warn(fmt.Sprintf("Quarantining node %s due to split-brain", leaderID))
                }
            } else if winner != "" {
                // Мы должны уступить лидерство
                rc.splitBrainDetector.QuarantineNode(rc.localNodeInfo.ID)
                if rc.logger != nil {
                    rc.logger.Warn(fmt.Sprintf("We are quarantined due to split-brain, leader is %s", winner))
                }
            }
        }
    }
}

// getCommitIndex возвращает индекс закоммиченных записей.
func (rc *RaftCoordinator) getCommitIndex() uint64 {
    if rc.raft == nil {
        return 0
    }
    return rc.raft.AppliedIndex()
}

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

// getLocalIP получает локальный IP адрес.
// Корректная работа на Linux и OpenIndiana (illumos).
func getLocalIP() string {
    addrs, err := net.InterfaceAddrs()
    if err != nil {
        return "127.0.0.1"
    }
    for _, addr := range addrs {
        if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() && ipnet.IP.To4() != nil {
            return ipnet.IP.String()
        }
    }
    return "127.0.0.1"
}

// mapStatusToString преобразует статус в строку.
func mapStatusToString(status int32) string {
    switch status {
    case 0:
        return "offline"
    case 1:
        return "active"
    case 2:
        return "syncing"
    case 3:
        return "failed"
    default:
        return "unknown"
    }
}

// NodeStatusCommand команда обновления статуса узла.
type NodeStatusCommand struct {
    Type      string `json:"type"`
    NodeID    string `json:"node_id"`
    Status    int32  `json:"status"`
    Timestamp int64  `json:"timestamp"`
}

// NodeRegistrationCommand команда регистрации узла.
type NodeRegistrationCommand struct {
    Type       string                 `json:"type"`
    Node       NodeInfo               `json:"node,omitempty"`
    NodeID     string                 `json:"node_id,omitempty"`
    Factor     int32                  `json:"factor,omitempty"`
    Shard      *RangeShard            `json:"shard,omitempty"`
    ShardID    string                 `json:"shard_id,omitempty"`
    TargetNode string                 `json:"target_node,omitempty"`
    Data       map[string]interface{} `json:"data,omitempty"`
    Timestamp  int64                  `json:"timestamp"`
}

// ClusterStatus представляет статус кластера.
type ClusterStatus struct {
    Name                 string `json:"name"`
    TotalNodes           int    `json:"total_nodes"`
    ActiveNodes          int    `json:"active_nodes"`
    SyncingNodes         int    `json:"syncing_nodes"`
    FailedNodes          int    `json:"failed_nodes"`
    ReplicationFactor    int    `json:"replication_factor"`
    LeaderID             string `json:"leader_id"`
    Health               string `json:"health"`
    CreatedAt            int64  `json:"created_at"`
    UpdatedAt            int64  `json:"updated_at"`
    PipelineEnabled      bool   `json:"pipeline_enabled"`
    BatchCommitEnabled   bool   `json:"batch_commit_enabled"`
    ReshardingEnabled    bool   `json:"resharding_enabled"`
    JointConsensusActive bool   `json:"joint_consensus_active"`
    FallbackMode         bool   `json:"fallback_mode"`
}

// Гарантируем, что runtime импортирован для кроссплатформенности
var _ = runtime.GOOS