futriix/cmd/futriis/main.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
 *
 * Файл: cmd/futriis/main.go
 * Назначение: Точка входа в приложение СУБД futriis.
 *
 * ИСПРАВЛЕНО:
 *   - Валидация пути к конфигурации и проверка прав доступа к файлу
 *   - Валидация конфигурации после загрузки (ValidateConfig)
 *   - WAL и data-dir привязаны к директории futriis, а не CWD
 *   - Graceful shutdown: убраны os.Exit(1) до завершения defer'ов,
 *     используется переменная exitCode и os.Exit в самом конце
 *   - Ожидание запуска HTTP-сервера с таймаутом
 *   - SAGA оркестратор останавливается всегда (даже если disabled)
 *   - Проверка allow-list для плагинов перед StartPlugin
 *   - Заменён SIGHUP на SIGUSR1 (reload) + SIGTERM/SIGINT (shutdown)
 *   - Проверка nil для raftCoordinator перед использованием
 *   - repl.NewRepl теперь возвращает (*Repl, error)
 *   - Интеграция с Prometheus: /metrics endpoint
 *   - ДОБАВЛЕНО: поддержка TLS на HTTP API (вариант A — TLS внутри futriiX).
 *     Теперь используется api.NewHTTPServerWithTLS(..., &cfg.Security).
 *   - ИСПРАВЛЕНО (UI): убраны лишние пустые строки перед баннером, чтобы
 *     строка "futriis 3i²(by 29.09.2026)" шла сразу после сообщения ACL
 *     о созданном admin-пользователе.
 *   - ИСПРАВЛЕНО (UI): текст баннера окрашивается в точный #00bfff
 *     (Deep Sky Blue) через utils.PrintlnDeepSkyBlueReset, чтобы совпадать
 *     с цветом надписи "futriis 3i²(by 29.09.2026)".
 *   - ИСПРАВЛЕНО: acl.NewACLManager() теперь возвращает (*ACLManager, error).
 *     Обрабатываем ошибку и завершаемся через os.Exit(1) с логом,
 *     без panic и stack trace.
 *
 * ИСПРАВЛЕНО (2026-10): после замены ACID-транзакций на batch-операции
 * функции storage.InitTransactionManager, storage.SetTransactionLogger,
 * storage.StopTransactionManager были удалены. Заменены на:
 *   - storage.InitBatchManager(walPath)  — инициализирует batch-менеджер,
 *     WAL и AOF (append-only file для мутаций вне batch);
 *   - storage.SetBatchLogger(logger)     — устанавливает логгер;
 *   - storage.StopBatchManager()         — останавливает batch-менеджер,
 *     WAL и AOF.
 * Дополнительно: при shutdown логируется состояние batch-статистики
 * (количество выполненных batch'ей, операций, ошибок) для диагностики.
 *
 * ДОБАВЛЕНО (2026-10, выборы лидера SAGA через Raft):
 *   - SAGA-оркестратор теперь создаётся через
 *     storage.NewSagaOrchestratorWithRaft и получает bridge из
 *     raftCoordinator (raftCoordinator.GetSagaLeaderBridge()).
 *     Это позволяет в кластерном режиме выбирать SAGA-лидера через Raft
 *     и выполнять компенсации только на одном узле.
 *   - В single-node режиме bridge также возвращает корректные значения
 *     (IsLeader() == true), так что поведение не меняется.
 *   - Раньше оркестратор создавался через NewSagaOrchestratorWithConfig
 *     с raftAccessor == nil, что в кластерном режиме приводило к тому,
 *     что каждый узел считал себя SAGA-лидером. Идемпотентность шагов
 *     через saga_executed_keys.json защищала от дублей операций, но не
 *     от одновременной компенсации одной SAGA с двух узлов.
 *   - Порядок shutdown безопасен: sagaOrchestrator.Stop() вызывается
 *     раньше raftCoordinator.Stop(), поэтому bridge ещё жив, когда
 *     оркестратор отписывается от него.
 */

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"syscall"
	"time"

	"futriis/internal/acl"
	"futriis/internal/api"
	"futriis/internal/cluster"
	"futriis/internal/config"
	"futriis/internal/log"
	"futriis/internal/metrics"
	"futriis/internal/plugin"
	"futriis/internal/repl"
	"futriis/internal/storage"
	"futriis/pkg/utils"
)

const (
	defaultDataDir    = "futriis"
	maxConfigFileSize = 1 * 1024 * 1024
)

var exitCode = 0

func main() {
	utils.SetColorEnabled(true)

	logLevel := os.Getenv("LOG_LEVEL")
	if logLevel == "" {
		logLevel = "info"
	}

	dataDir, err := ensureDataDir()
	if err != nil {
		fmt.Fprintf(os.Stderr, "Failed to prepare data directory: %v\n", err)
		os.Exit(1)
	}

	logFile := filepath.Join(dataDir, "futriis.log")
	if err := log.InitDefaultLogger(logFile, logLevel); err != nil {
		fmt.Printf("Failed to initialize logger: %v\n", err)
		os.Exit(1)
	}

	log.Info("Futriis DB starting...")
	log.Infof("Log level: %s", logLevel)
	log.Infof("Data directory: %s", dataDir)

	configPath := os.Getenv("FUTRIIS_CONFIG")
	if configPath == "" {
		configPath = "config.toml"
	}

	cfg, err := loadAndValidateConfig(configPath)
	if err != nil {
		log.Error("Failed to load config: " + err.Error())
		utils.PrintError("Failed to load config: " + err.Error())
		os.Exit(1)
	}

	logger, err := log.NewLogger(cfg.Log.LogFile, cfg.Log.LogLevel)
	if err != nil {
		log.Error("Failed to initialize logger: " + err.Error())
		utils.PrintError("Failed to initialize logger: " + err.Error())
		os.Exit(1)
	}
	defer logger.Close()
	logger.Info("futriis database starting...")

	store := storage.NewStorage(cfg.Storage.PageSizeMB, logger)
	storage.SetGlobalStorage(store)

	// ========== BATCH MANAGER (замена TransactionManager) ==========
	//
	// ИСПРАВЛЕНО (2026-10): InitTransactionManager заменён на InitBatchManager.
	// InitBatchManager инициализирует:
	//   - WAL (сегментированный журнал для batch-операций);
	//   - AOF (append-only file для мутаций вне batch).
	// Директория AOF создаётся автоматически как filepath.Join(filepath.Dir(walPath), "aof").
	walPath := filepath.Join(dataDir, "futriis.wal")
	if err := storage.InitBatchManager(walPath); err != nil {
		logger.Warn("Failed to initialize batch manager: " + err.Error())
	} else {
		storage.SetBatchLogger(logger)
		logger.Info("Batch manager initialized (WAL + AOF)")
		logger.Info(fmt.Sprintf("  WAL path: %s", walPath))
		logger.Info(fmt.Sprintf("  AOF path: %s", filepath.Join(dataDir, "aof")))
	}

	storage.InitTriggerManager(logger)
	logger.Info("Trigger manager initialized")

	// ИСПРАВЛЕНО: NewACLManager теперь возвращает (*ACLManager, error).
	// Обрабатываем ошибку управляемо (без panic) и завершаемся с кодом 1.
	aclManager, err := acl.NewACLManager()
	if err != nil {
		logger.Error("Failed to initialize ACL manager: " + err.Error())
		utils.PrintError("Failed to initialize ACL manager: " + err.Error())
		os.Exit(1)
	}
	logger.Info("ACL manager initialized")

	raftCoordinator, err := cluster.NewRaftCoordinator(cfg, store, logger)
	if err != nil {
		logger.Error("Failed to start Raft coordinator: " + err.Error())
		utils.PrintError("Failed to start Raft coordinator: " + err.Error())
		os.Exit(1)
	}

	// ========== SAGA ORCHESTRATOR ==========
	//
	// ДОБАВЛЕНО (2026-10): используем NewSagaOrchestratorWithRaft и
	// передаём bridge из raftCoordinator. Это позволяет в кластерном
	// режиме выбирать SAGA-лидера через Raft, чтобы компенсации
	// выполнялись только на одном узле.
	//
	// В single-node режиме bridge возвращает IsLeader() == true, так что
	// поведение не меняется. В кластерном режиме только лидер Raft
	// будет IsSagaLeader() == true, и только он будет выполнять SAGA.
	//
	// Если raftCoordinator == nil (не должно случаться — выше os.Exit(1)
	// при ошибке), создаём оркестратор без Raft (single-node fallback).
	var sagaOrchestrator *storage.SagaOrchestrator
	if cfg.Saga.Enabled {
		sagaConfig := &cfg.Saga
		nodeID := cfg.Cluster.NodeIP + ":" + fmt.Sprint(cfg.Cluster.NodePort)

		// Передаём bridge из Raft в качестве storage.SagaRaftAccessor.
		// Тип bridge реализует интерфейс storage.SagaRaftAccessor
		// (IsSagaLeader, GetSagaLeaderID, GetSagaCurrentTerm,
		// RegisterSagaLeaderObserver).
		var raftAccessor storage.SagaRaftAccessor
		if raftCoordinator != nil {
			raftAccessor = raftCoordinator.GetSagaLeaderBridge()
		}

		sagaOrchestrator, err = storage.NewSagaOrchestratorWithRaft(
			sagaConfig,
			nodeID,
			raftAccessor,
			logger,
		)
		if err != nil {
			logger.Warn(fmt.Sprintf("Failed to initialize saga orchestrator: %v", err))
		} else {
			logger.Info(fmt.Sprintf("Saga orchestrator initialized with %d coordinators (raft leader election: %v)",
				sagaConfig.GetCoordinatorCount(), raftAccessor != nil))
			storage.SetGlobalSagaOrchestrator(sagaOrchestrator)
		}
	} else {
		logger.Info("Saga orchestrator disabled in configuration")
	}

	if raftCoordinator != nil {
		schemaMigrator := raftCoordinator.GetSchemaMigrator()
		if schemaMigrator != nil {
			logger.Info("Checking for schema migrations...")
			status := schemaMigrator.GetStatus()
			logger.Infof("Current schema version: %s, total migrations: %d, applied: %d, pending: %d",
				status.CurrentVersion, status.TotalMigrations, status.AppliedMigrations, status.PendingMigrations)
			if status.PendingMigrations > 0 {
				logger.Info("Applying pending schema migrations...")
				if err := schemaMigrator.Migrate("2.1.0"); err != nil {
					logger.Warn(fmt.Sprintf("Schema migration warning: %v", err))
				}
			}
		}
	}

	if cfg.Cluster.Bootstrap || len(cfg.Cluster.Nodes) <= 1 {
		for i := 0; i < 10; i++ {
			if raftCoordinator.IsLeader() {
				break
			}
			time.Sleep(1 * time.Second)
		}
	}

	node := cluster.NewNode(cfg.Cluster.NodeIP, cfg.Cluster.NodePort, store, logger)

	var registerErr error
	for i := 0; i < 5; i++ {
		registerErr = raftCoordinator.RegisterNode(node)
		if registerErr == nil {
			break
		}
		if i < 4 {
			logger.Warn(fmt.Sprintf("Failed to register node (attempt %d/5): %v, retrying...", i+1, registerErr))
			time.Sleep(2 * time.Second)
		}
	}
	if registerErr != nil {
		logger.Error("Failed to register node: " + registerErr.Error())
		utils.PrintError("Failed to register node: " + registerErr.Error())
		os.Exit(1)
	}

	pluginManager := plugin.NewPluginManager(
		cfg.Plugins.ScriptDir,
		logger,
		store,
		cfg.Plugins.Enabled,
	)

	if cfg.Plugins.Enabled {
		logger.Info(fmt.Sprintf("Plugin manager initialized, directory: %s", cfg.Plugins.ScriptDir))
		allowList := make(map[string]bool)
		for _, name := range cfg.Plugins.AllowList {
			allowList[name] = true
		}
		for _, p := range pluginManager.ListPlugins() {
			if len(allowList) > 0 && !allowList[p.Name] {
				logger.Warn(fmt.Sprintf("Plugin %s is not in allow-list, skipping start", p.Name))
				continue
			}
			if err := pluginManager.StartPlugin(p.Name); err != nil {
				logger.Warn(fmt.Sprintf("Failed to start plugin %s: %v", p.Name, err))
			}
		}
	}

	// ========== PROMETHEUS METRICS ==========
	var metricsCollector *metrics.Collector
	if cfg.Metrics.Enabled {
		metricsCollector = metrics.NewCollector(metrics.CollectorDeps{
			Storage:     newStorageStatsAdapter(store),
			Coordinator: newCoordinatorStatsAdapter(raftCoordinator),
			Logger:      logger,
			Interval:    time.Duration(cfg.Metrics.CollectIntervalSec) * time.Second,
		})
		metricsCollector.Start()
		logger.Info("Prometheus metrics collector enabled")
	} else {
		logger.Info("Prometheus metrics collector disabled in configuration")
	}

	// ========== HTTP API SERVER ==========
	// ДОБАВЛЕНО: используем NewHTTPServerWithTLS и передаём cfg.Security.
	// Если в config.toml [security].enable_tls = true, сервер поднимется по HTTPS.
	httpPort := cfg.API.Port
	httpServer := api.NewHTTPServerWithTLS(httpPort, store, raftCoordinator, aclManager, logger, &cfg.Security)
	if metricsCollector != nil {
		httpServer.SetMetricsCollector(metricsCollector)
	}

	httpErrChan := make(chan error, 1)
	go func() {
		if err := httpServer.Start(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			httpErrChan <- fmt.Errorf("HTTP server error: %w", err)
		}
	}()

	httpStarted := false
	for i := 0; i < 50; i++ {
		select {
		case err := <-httpErrChan:
			logger.Error(err.Error())
			utils.PrintError(err.Error())
			os.Exit(1)
		default:
		}
		if isPortListening(httpPort) {
			httpStarted = true
			break
		}
		time.Sleep(100 * time.Millisecond)
	}
	if !httpStarted {
		logger.Warn("HTTP server did not start within timeout, continuing anyway")
	}
	scheme := "http"
	if httpServer.IsTLSEnabled() {
		scheme = "https"
	}
	logger.Info(fmt.Sprintf("HTTP API server started on %s://0.0.0.0:%d", scheme, httpPort))
	if metricsCollector != nil {
		logger.Info(fmt.Sprintf("Prometheus metrics available at %s://localhost:%d/metrics", scheme, httpPort))
	}

	if raftCoordinator != nil {
		logger.Info("Cluster features enabled: Pipeline Replication, Batch Commit, Dynamic Resharding, Joint Consensus")
		logger.Info("  - Single Point of Failure protection (Leader Fallback) - ENABLED")
		logger.Info("  - Automatic Panic Recovery for goroutines - ENABLED")
		logger.Info("  - Schema Migration support - ENABLED")
		logger.Info("  - Full configuration validation - ENABLED")
		logger.Info("  - Automatic data persistence - ENABLED")
	}

	// ИСПРАВЛЕНО (UI): убраны лишние пустые строки перед баннером.
	// Внутри displayBanner уже есть utils.Println("") первой строкой,
	// чтобы отделить баннер от ACL-сообщения. Строка "futriis 3i²(by
	// 29.09.2026)" идёт сразу после сообщения ACL.
	displayBanner(cfg.Cluster.Name, httpPort, raftCoordinator, cfg.Saga.Enabled, cfg.Metrics.Enabled)

	replInstance, err := repl.NewRepl(store, raftCoordinator, logger, cfg, aclManager, pluginManager)
	if err != nil {
		logger.Error("Failed to initialize REPL: " + err.Error())
		utils.PrintError("Failed to initialize REPL: " + err.Error())
		os.Exit(1)
	}
	defer replInstance.Close()

	// ========== GRACEFUL SHUTDOWN ==========
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGUSR1)

	go func() {
		for sig := range sigChan {
			switch sig {
			case syscall.SIGUSR1:
				logger.Info("Received SIGUSR1 (reload signal), reloading configuration...")
				continue
			case syscall.SIGINT, syscall.SIGTERM:
				logger.Info(fmt.Sprintf("Received signal %v, starting graceful shutdown...", sig))
			}
			break
		}

		utils.Println("\nReceived shutdown signal, starting graceful shutdown...")

		logger.Info("Stopping REPL...")
		replInstance.Close()
		logger.Info("REPL stopped")

		logger.Info("Stopping HTTP server...")
		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer shutdownCancel()

		httpShutdownDone := make(chan struct{})
		go func() {
			if err := httpServer.Stop(); err != nil {
				logger.Error(fmt.Sprintf("HTTP server stop error: %v", err))
			}
			close(httpShutdownDone)
		}()
		select {
		case <-httpShutdownDone:
			logger.Info("HTTP server stopped")
		case <-shutdownCtx.Done():
			logger.Warn("HTTP server shutdown timeout")
		}

		if metricsCollector != nil {
			logger.Info("Stopping metrics collector...")
			metricsCollector.Stop()
			logger.Info("Metrics collector stopped")
		}

		if cfg.Plugins.Enabled && pluginManager != nil {
			logger.Info("Stopping plugins...")
			for _, p := range pluginManager.ListPlugins() {
				if err := pluginManager.StopPlugin(p.Name); err != nil {
					logger.Warn(fmt.Sprintf("Failed to stop plugin %s: %v", p.Name, err))
				}
			}
		}

		// ВАЖНО: sagaOrchestrator останавливается ДО raftCoordinator.
		// Это гарантирует, что оркестратор отпишется от sagaLeaderBridge
		// (и остановит leaderElector) до того, как bridge будет остановлен
		// вместе с координатором. Двойной остановки bridge не будет.
		if sagaOrchestrator != nil {
			logger.Info("Stopping SAGA orchestrator...")
			sagaStopDone := make(chan struct{})
			go func() {
				sagaOrchestrator.Stop()
				close(sagaStopDone)
			}()
			select {
			case <-sagaStopDone:
				logger.Info("SAGA orchestrator stopped")
			case <-time.After(10 * time.Second):
				logger.Warn("SAGA orchestrator shutdown timeout")
			}
		}

		if raftCoordinator != nil {
			logger.Info("Persisting data to disk...")
			persistDone := make(chan struct{})
			go func() {
				if pm := raftCoordinator.GetPersistenceManager(); pm != nil {
					if err := pm.SaveAll(); err != nil {
						logger.Error(fmt.Sprintf("Failed to save data: %v", err))
					}
				}
				close(persistDone)
			}()
			select {
			case <-persistDone:
				logger.Info("Persistence completed")
			case <-time.After(15 * time.Second):
				logger.Warn("Persistence timeout")
			}
		}

		if raftCoordinator != nil {
			logger.Info("Stopping Raft coordinator...")
			raftStopDone := make(chan struct{})
			go func() {
				raftCoordinator.Stop()
				close(raftStopDone)
			}()
			select {
			case <-raftStopDone:
				logger.Info("Raft coordinator stopped")
			case <-time.After(15 * time.Second):
				logger.Warn("Raft coordinator shutdown timeout")
			}
		}

		logger.Info("Stopping cluster node...")
		nodeStopDone := make(chan struct{})
		go func() {
			node.Stop()
			close(nodeStopDone)
		}()
		select {
		case <-nodeStopDone:
			logger.Info("Cluster node stopped")
		case <-time.After(10 * time.Second):
			logger.Warn("Node shutdown timeout")
		}

		// ИСПРАВЛЕНО (2026-10): StopTransactionManager заменён на StopBatchManager.
		// StopBatchManager синхронизирует WAL, закрывает его и закрывает AOF.
		logger.Info("Stopping batch manager (WAL + AOF)...")
		if stats := storage.GetBatchStats(); stats != nil {
			logger.Info(fmt.Sprintf("  Batch stats: total_batches=%v, total_operations=%v, total_failed=%v, active_batches=%v",
				stats["total_batches"], stats["total_operations"], stats["total_failed"], stats["active_batches"]))
		}
		batchStopDone := make(chan struct{})
		go func() {
			if err := storage.StopBatchManager(); err != nil {
				logger.Error(fmt.Sprintf("Failed to stop batch manager: %v", err))
			} else {
				logger.Info("Batch manager stopped")
			}
			close(batchStopDone)
		}()
		select {
		case <-batchStopDone:
			// уже залогировано внутри горутины
		case <-time.After(10 * time.Second):
			logger.Warn("Batch manager shutdown timeout")
		}

		logger.Info("Finalizing logger...")
		if err := logger.Sync(); err != nil {
			fmt.Printf("Failed to sync logger: %v\n", err)
		}
		logger.Close()

		if aclManager != nil {
			aclManager.Stop()
		}

		utils.DisableColorMode()
		fmt.Println("Futriis DB shutdown complete")

		exitCode = 0
		close(sigChan)
	}()

	if err := replInstance.Run(); err != nil {
		logger.Error("REPL error: " + err.Error())
		utils.PrintError("REPL error: " + err.Error())
		exitCode = 1
	}

	time.Sleep(500 * time.Millisecond)

	if exitCode != 0 {
		os.Exit(exitCode)
	}
}

// ==================== Адаптеры для metrics ====================

type storageStatsAdapter struct {
	s *storage.Storage
}

func newStorageStatsAdapter(s *storage.Storage) *storageStatsAdapter {
	return &storageStatsAdapter{s: s}
}

func (a *storageStatsAdapter) GetStats() map[string]interface{} { return a.s.GetStats() }
func (a *storageStatsAdapter) GetDatabaseCount() int            { return a.s.GetDatabaseCount() }
func (a *storageStatsAdapter) GetTotalDocuments() int64         { return a.s.GetTotalDocuments() }

type coordinatorStatsAdapter struct {
	c *cluster.RaftCoordinator
}

func newCoordinatorStatsAdapter(c *cluster.RaftCoordinator) *coordinatorStatsAdapter {
	return &coordinatorStatsAdapter{c: c}
}

func (a *coordinatorStatsAdapter) GetClusterStatus() metrics.ClusterStatusLike {
	st := a.c.GetClusterStatus()
	return &clusterStatusAdapter{st: st}
}

func (a *coordinatorStatsAdapter) GetActiveNodes() []metrics.NodeInfoLike {
	nodes := a.c.GetActiveNodes()
	out := make([]metrics.NodeInfoLike, 0, len(nodes))
	for _, n := range nodes {
		out = append(out, &nodeInfoAdapter{n: n})
	}
	return out
}

func (a *coordinatorStatsAdapter) GetAllNodes() []metrics.NodeInfoLike {
	nodes := a.c.GetAllNodes()
	out := make([]metrics.NodeInfoLike, 0, len(nodes))
	for _, n := range nodes {
		out = append(out, &nodeInfoAdapter{n: n})
	}
	return out
}

func (a *coordinatorStatsAdapter) IsLeader() bool         { return a.c.IsLeader() }
func (a *coordinatorStatsAdapter) GetCurrentTerm() uint64 { return a.c.GetCurrentTerm() }

type clusterStatusAdapter struct {
	st *cluster.ClusterStatus
}

func (a *clusterStatusAdapter) GetTotalNodes() int  { return a.st.TotalNodes }
func (a *clusterStatusAdapter) GetActiveNodes() int { return a.st.ActiveNodes }
func (a *clusterStatusAdapter) GetFailedNodes() int { return a.st.FailedNodes }
func (a *clusterStatusAdapter) GetLeaderID() string { return a.st.LeaderID }
func (a *clusterStatusAdapter) GetHealth() string   { return a.st.Health }

type nodeInfoAdapter struct {
	n *cluster.NodeInfo
}

func (a *nodeInfoAdapter) GetID() string      { return a.n.ID }
func (a *nodeInfoAdapter) GetIP() string      { return a.n.IP }
func (a *nodeInfoAdapter) GetPort() int       { return a.n.Port }
func (a *nodeInfoAdapter) GetStatus() string  { return a.n.Status }
func (a *nodeInfoAdapter) GetLastSeen() int64 { return a.n.LastSeen }

// ==================== Утилиты ====================

func ensureDataDir() (string, error) {
	dataDir := os.Getenv("FUTRIIS_DATA_DIR")
	if dataDir == "" {
		dataDir = defaultDataDir
	}
	absPath, err := filepath.Abs(dataDir)
	if err != nil {
		return "", fmt.Errorf("failed to resolve absolute path for %s: %w", dataDir, err)
	}
	if err := os.MkdirAll(absPath, 0755); err != nil {
		return "", fmt.Errorf("failed to create data directory %s: %w", absPath, err)
	}
	testFile := filepath.Join(absPath, ".write_test")
	if err := os.WriteFile(testFile, []byte("test"), 0600); err != nil {
		return "", fmt.Errorf("data directory %s is not writable: %w", absPath, err)
	}
	_ = os.Remove(testFile)
	return absPath, nil
}

func loadAndValidateConfig(path string) (*config.Config, error) {
	info, err := os.Stat(path)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, fmt.Errorf("config file not found: %s", path)
		}
		return nil, fmt.Errorf("failed to stat config file: %w", err)
	}
	if info.Size() > maxConfigFileSize {
		return nil, fmt.Errorf("config file too large: %d bytes (max %d)", info.Size(), maxConfigFileSize)
	}
	if !info.Mode().IsRegular() {
		return nil, fmt.Errorf("config path is not a regular file: %s", path)
	}
	cfg, err := config.LoadConfig(path)
	if err != nil {
		return nil, fmt.Errorf("failed to parse config: %w", err)
	}
	if cfg == nil {
		return nil, fmt.Errorf("config is nil after load")
	}
	if err := config.ValidateConfig(cfg); err != nil {
		return nil, fmt.Errorf("config validation failed: %w", err)
	}
	return cfg, nil
}

func isPortListening(port int) bool {
	if port <= 0 {
		return false
	}
	conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 200*time.Millisecond)
	if err != nil {
		return false
	}
	conn.Close()
	return true
}

// displayBanner выводит стартовый баннер.
//
// ИСПРАВЛЕНО (UI): текст баннера окрашивается точным цветом
// #00bfff (Deep Sky Blue) через utils.PrintlnDeepSkyBlueReset, чтобы
// совпадать с цветом надписи "futriis 3i²(by 29.09.2026)".
//
// Внутри уже есть utils.Println("") первой строкой, чтобы отделить
// баннер от ACL-сообщения. Дополнительные пустые строки в main() не
// добавляются (см. комментарий перед вызовом displayBanner).
func displayBanner(clusterName string, httpPort int, coordinator *cluster.RaftCoordinator, sagaEnabled, metricsEnabled bool) {
	utils.Println("")
	bannerLines := []string{
		"                futriis 3i²(by 29.09.2026)                 ",
		"                Distributed Document-Store in-memory database with support lua plugins   ",
		"                Cluster status: enable (Raft consensus)",
		"                Cluster features: Pipeline Replication, Batch Commit, Dynamic Resharding",
		"                Cluster name: " + clusterName,
		"                HTTP API (for curl or wget utils only): http://localhost:" + fmt.Sprintf("%d", httpPort) + "/api/",
	}
	if metricsEnabled {
		bannerLines = append(bannerLines,
			fmt.Sprintf("                Prometheus metrics: http://localhost:%d/metrics", httpPort),
			"                Grafana: add Prometheus datasource pointing to /metrics URL",
		)
	}
	bannerLines = append(bannerLines, "                Additional features:")
	bannerLines = append(bannerLines, "                  - Single Point of Failure protection (Leader Fallback)")
	bannerLines = append(bannerLines, "                  - Automatic Panic Recovery for goroutines")
	bannerLines = append(bannerLines, "                  - Schema Migration support")
	bannerLines = append(bannerLines, "                  - Full configuration validation")
	bannerLines = append(bannerLines, "                  - Automatic data persistence")
	if sagaEnabled {
		bannerLines = append(bannerLines, "                  - SAGA Orchestrator: ENABLED (distributed transactions)")
	} else {
		bannerLines = append(bannerLines, "                  - SAGA Orchestrator: DISABLED")
	}
	if metricsEnabled {
		bannerLines = append(bannerLines, "                  - Prometheus metrics: ENABLED")
	} else {
		bannerLines = append(bannerLines, "                  - Prometheus metrics: DISABLED")
	}

	bannerLines = append(bannerLines, []string{
		"",
		"                Type 'quit' or 'exit' to quit",
		"                Type 'status' to see cluster status",
		"                Type 'acl login <user> <pass>' to authenticate",
		"                Type 'plugin list' to see loaded plugins",
		"                Type 'cluster pipeline' to see pipeline stats",
		"                Type 'cluster reshard' to trigger manual resharding",
		"                Type 'migrate status' to see schema migration status",
		"                Type 'fallback status' to see SPoF protection status",
		"                Type 'panic stats' to see panic recovery statistics",
		"                Type 'persist status' to see data persistence status",
		"                Type 'saga status' to see SAGA orchestrator status",
		"                Type 'saga list' to list active SAGA transactions",
	}...)

	for _, line := range bannerLines {
		// ИСПРАВЛЕНО: используем точный цвет #00bfff (Deep Sky Blue),
		// совпадающий с цветом заголовка баннера.
		utils.PrintlnDeepSkyBlueReset(line)
	}
}

func printJSON(data interface{}) {
	jsonData, err := json.MarshalIndent(data, "", "  ")
	if err != nil {
		utils.PrintError("Failed to marshal JSON: " + err.Error())
		return
	}
	fmt.Println(string(jsonData))
}