futriix/internal/api/http.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/api/http.go
// Назначение: HTTP RESTful API для взаимодействия с СУБД через curl.
// Поддерживает CRUD операции, управление индексами, ACL и ограничениями.
// Реализован с минимальными блокировками, использует wait-free структуры.

package api

import (
    "crypto/rand"
    "encoding/base64"
    "encoding/json"
    "fmt"
    "net/http"
    "strconv"
    "strings"
    "sync"
    "sync/atomic"
    "time"
    
    "futriis/internal/acl"
    "futriis/internal/cluster"
    "futriis/internal/log"
    "futriis/internal/storage"
)

// RateLimiterConfig содержит конфигурацию rate limiter'а
type RateLimiterConfigHTTP struct {
    RequestsPerSecond int
    BurstSize         int
    CleanupInterval   time.Duration
    MaxEntries        int
}

// DefaultRateLimiterConfigHTTP возвращает конфигурацию по умолчанию
func DefaultRateLimiterConfigHTTP() *RateLimiterConfigHTTP {
    return &RateLimiterConfigHTTP{
        RequestsPerSecond: 100,
        BurstSize:         20,
        CleanupInterval:   5 * time.Minute,
        MaxEntries:        10000,
    }
}

// TokenBucketHTTP реализует алгоритм token bucket для HTTP API
type TokenBucketHTTP struct {
    tokens        float64
    maxTokens     float64
    refillRate    float64
    lastRefill    time.Time
    mu            sync.Mutex
}

// NewTokenBucketHTTP создаёт новый token bucket
func NewTokenBucketHTTP(ratePerSecond float64, burst int) *TokenBucketHTTP {
    return &TokenBucketHTTP{
        tokens:     float64(burst),
        maxTokens:  float64(burst),
        refillRate: ratePerSecond,
        lastRefill: time.Now(),
    }
}

// Allow проверяет, разрешён ли запрос
func (tb *TokenBucketHTTP) Allow() bool {
    tb.mu.Lock()
    defer tb.mu.Unlock()
    
    now := time.Now()
    elapsed := now.Sub(tb.lastRefill).Seconds()
    tb.tokens += elapsed * tb.refillRate
    if tb.tokens > tb.maxTokens {
        tb.tokens = tb.maxTokens
    }
    tb.lastRefill = now
    
    if tb.tokens >= 1.0 {
        tb.tokens -= 1.0
        return true
    }
    return false
}

// RateLimiterHTTP управляет rate limiting для HTTP API
type RateLimiterHTTP struct {
    config      *RateLimiterConfigHTTP
    limiters    sync.Map
    rejected    atomic.Uint64
    allowed     atomic.Uint64
    stopChan    chan struct{}
    wg          sync.WaitGroup
}

// NewRateLimiterHTTP создаёт новый rate limiter
func NewRateLimiterHTTP(config *RateLimiterConfigHTTP) *RateLimiterHTTP {
    if config == nil {
        config = DefaultRateLimiterConfigHTTP()
    }
    
    rl := &RateLimiterHTTP{
        config:   config,
        stopChan: make(chan struct{}),
    }
    
    rl.wg.Add(1)
    go rl.cleanupLoop()
    
    return rl
}

// getLimiter возвращает или создаёт лимитер для ключа
func (rl *RateLimiterHTTP) getLimiter(key string) *TokenBucketHTTP {
    if val, ok := rl.limiters.Load(key); ok {
        return val.(*TokenBucketHTTP)
    }
    
    limiter := NewTokenBucketHTTP(float64(rl.config.RequestsPerSecond), rl.config.BurstSize)
    actual, _ := rl.limiters.LoadOrStore(key, limiter)
    return actual.(*TokenBucketHTTP)
}

// Allow проверяет, разрешён ли запрос для ключа
func (rl *RateLimiterHTTP) Allow(key string) bool {
    limiter := rl.getLimiter(key)
    allowed := limiter.Allow()
    
    if allowed {
        rl.allowed.Add(1)
    } else {
        rl.rejected.Add(1)
    }
    
    return allowed
}

// Middleware возвращает middleware для HTTP
func (rl *RateLimiterHTTP) Middleware(next http.HandlerFunc) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        // Используем IP + User-Agent как ключ
        key := r.RemoteAddr
        if forwarded := r.Header.Get("X-Forwarded-For"); forwarded != "" {
            key = forwarded
        }
        
        if !rl.Allow(key) {
            w.Header().Set("X-RateLimit-Limit", fmt.Sprintf("%d", rl.config.RequestsPerSecond))
            w.Header().Set("X-RateLimit-Remaining", "0")
            w.Header().Set("Retry-After", "1")
            http.Error(w, "Rate limit exceeded. Please try again later.", http.StatusTooManyRequests)
            return
        }
        
        next(w, r)
    }
}

// cleanupLoop периодически очищает старые лимитеры
func (rl *RateLimiterHTTP) cleanupLoop() {
    defer rl.wg.Done()
    
    ticker := time.NewTicker(rl.config.CleanupInterval)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            rl.cleanup()
        case <-rl.stopChan:
            return
        }
    }
}

// cleanup удаляет старые лимитеры
func (rl *RateLimiterHTTP) cleanup() {
    count := 0
    rl.limiters.Range(func(key, value interface{}) bool {
        count++
        if count > rl.config.MaxEntries {
            rl.limiters.Delete(key)
        }
        return true
    })
}

// GetStats возвращает статистику rate limiter'а
func (rl *RateLimiterHTTP) GetStats() map[string]interface{} {
    count := 0
    rl.limiters.Range(func(key, value interface{}) bool {
        count++
        return true
    })
    
    return map[string]interface{}{
        "active_limiters":     count,
        "allowed_requests":    rl.allowed.Load(),
        "rejected_requests":   rl.rejected.Load(),
        "requests_per_second": rl.config.RequestsPerSecond,
        "burst_size":          rl.config.BurstSize,
    }
}

// Stop останавливает rate limiter
func (rl *RateLimiterHTTP) Stop() {
    close(rl.stopChan)
    rl.wg.Wait()
}

// generateSecureToken генерирует криптостойкий токен
func generateSecureToken() string {
    bytes := make([]byte, 32)
    _, err := rand.Read(bytes)
    if err != nil {
        return fmt.Sprintf("fallback_%d", time.Now().UnixNano())
    }
    return base64.URLEncoding.EncodeToString(bytes)
}

// HTTPServer представляет HTTP сервер API
type HTTPServer struct {
    store       *storage.Storage
    coordinator *cluster.RaftCoordinator
    aclManager  *acl.ACLManager
    logger      *log.Logger
    server      *http.Server
    port        int
    rateLimiter *RateLimiterHTTP
    sessions    sync.Map // map[string]string (sessionID -> username)
}

// APIResponse представляет стандартный ответ API
type APIResponse struct {
    Success bool        `json:"success"`
    Data    interface{} `json:"data,omitempty"`
    Error   string      `json:"error,omitempty"`
}

// NewHTTPServer создаёт новый HTTP сервер
func NewHTTPServer(port int, store *storage.Storage, coord *cluster.RaftCoordinator, aclMgr *acl.ACLManager, logger *log.Logger) *HTTPServer {
    s := &HTTPServer{
        store:       store,
        coordinator: coord,
        aclManager:  aclMgr,
        logger:      logger,
        port:        port,
        rateLimiter: NewRateLimiterHTTP(DefaultRateLimiterConfigHTTP()),
    }
    
    mux := http.NewServeMux()
    
    // CORS middleware wrapper с rate limiting
    corsHandler := func(handler http.HandlerFunc) http.HandlerFunc {
        return func(w http.ResponseWriter, r *http.Request) {
            w.Header().Set("Access-Control-Allow-Origin", "*")
            w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
            w.Header().Set("Access-Control-Allow-Headers", "Content-Type, X-Session-ID, Authorization")
            
            if r.Method == http.MethodOptions {
                w.WriteHeader(http.StatusOK)
                return
            }
            
            // Применяем rate limiting
            s.rateLimiter.Middleware(handler)(w, r)
        }
    }
    
    // Middleware для аутентификации
    authMiddleware := func(handler http.HandlerFunc) http.HandlerFunc {
        return func(w http.ResponseWriter, r *http.Request) {
            sessionID := r.Header.Get("X-Session-ID")
            if sessionID == "" {
                // Также проверяем Authorization header
                auth := r.Header.Get("Authorization")
                if strings.HasPrefix(auth, "Bearer ") {
                    sessionID = strings.TrimPrefix(auth, "Bearer ")
                }
            }
            
            if sessionID == "" {
                s.sendError(w, "Authentication required", http.StatusUnauthorized)
                return
            }
            
            if _, ok := s.sessions.Load(sessionID); !ok {
                s.sendError(w, "Invalid or expired session", http.StatusUnauthorized)
                return
            }
            
            handler(w, r)
        }
    }
    
    // Аутентификация
    mux.HandleFunc("/api/auth/login", corsHandler(s.handleLogin))
    mux.HandleFunc("/api/auth/logout", corsHandler(authMiddleware(s.handleLogout)))
    mux.HandleFunc("/api/auth/session", corsHandler(authMiddleware(s.handleSessionCheck)))
    
    // CRUD операции
    mux.HandleFunc("/api/db/", corsHandler(authMiddleware(s.handleDatabaseRequest)))
    
    // Индексы
    mux.HandleFunc("/api/index/", corsHandler(authMiddleware(s.handleIndexRequest)))
    
    // ACL
    mux.HandleFunc("/api/acl/", corsHandler(authMiddleware(s.handleACLRequest)))
    
    // Constraints
    mux.HandleFunc("/api/constraint/", corsHandler(authMiddleware(s.handleConstraintRequest)))
    
    // Cluster
    mux.HandleFunc("/api/cluster/", corsHandler(authMiddleware(s.handleClusterRequest)))
    
    // Триггеры
    mux.HandleFunc("/api/trigger/", corsHandler(authMiddleware(s.handleTriggerRequest)))
    
    // Транзакции
    mux.HandleFunc("/api/transaction/", corsHandler(authMiddleware(s.handleTransactionRequest)))
    
    // Метрики и здоровье (без аутентификации для мониторинга)
    mux.HandleFunc("/api/health", s.handleHealthCheck)
    mux.HandleFunc("/api/metrics", s.handleMetricsEndpoint)
    
    s.server = &http.Server{
        Addr:    fmt.Sprintf(":%d", port),
        Handler: mux,
    }
    
    return s
}

// Start запускает HTTP сервер
func (s *HTTPServer) Start() error {
    s.logger.Info("Starting HTTP API server on port " + strconv.Itoa(s.port))
    return s.server.ListenAndServe()
}

// Stop останавливает HTTP сервер
func (s *HTTPServer) Stop() error {
    s.rateLimiter.Stop()
    return s.server.Close()
}

// GetRateLimiterStats возвращает статистику rate limiter'а
func (s *HTTPServer) GetRateLimiterStats() map[string]interface{} {
    return s.rateLimiter.GetStats()
}

// isUserAdmin проверяет, является ли пользователь администратором через ACL
func (s *HTTPServer) isUserAdmin(username string) bool {
    if s.aclManager == nil {
        return username == "admin"
    }
    // Проверяем через ACL - пользователь с ролью admin
    roles := s.aclManager.GetUserRoles(username)
    for _, role := range roles {
        if role == "admin" {
            return true
        }
    }
    return false
}

// handleLogin обрабатывает аутентификацию
func (s *HTTPServer) handleLogin(w http.ResponseWriter, r *http.Request) {
    if r.Method != http.MethodPost {
        s.sendError(w, "Method not allowed", http.StatusMethodNotAllowed)
        return
    }
    
    var creds struct {
        Username string `json:"username"`
        Password string `json:"password"`
    }
    
    if err := json.NewDecoder(r.Body).Decode(&creds); err != nil {
        s.sendError(w, "Invalid request body", http.StatusBadRequest)
        return
    }
    
    sessionID, err := s.aclManager.Authenticate(creds.Username, creds.Password)
    if err != nil {
        s.logOperation("LOGIN", creds.Username, "error", err.Error(), nil)
        s.sendError(w, err.Error(), http.StatusUnauthorized)
        return
    }
    
    // Генерируем криптостойкий токен сессии
    token := generateSecureToken()
    s.sessions.Store(token, creds.Username)
    
    // Также сохраняем связь с оригинальным sessionID
    s.sessions.Store(sessionID, creds.Username)
    
    s.logOperation("LOGIN", creds.Username, "success", "", nil)
    
    s.sendSuccess(w, map[string]interface{}{
        "session_id": sessionID,
        "token":      token,
        "username":   creds.Username,
        "is_admin":   s.isUserAdmin(creds.Username),
    })
}

// handleLogout обрабатывает выход
func (s *HTTPServer) handleLogout(w http.ResponseWriter, r *http.Request) {
    sessionID := r.Header.Get("X-Session-ID")
    if sessionID == "" {
        auth := r.Header.Get("Authorization")
        if strings.HasPrefix(auth, "Bearer ") {
            sessionID = strings.TrimPrefix(auth, "Bearer ")
        }
    }
    
    if sessionID != "" {
        if username, ok := s.sessions.Load(sessionID); ok {
            s.logOperation("LOGOUT", username.(string), "success", "", nil)
            s.sessions.Delete(sessionID)
        }
        s.aclManager.Logout(sessionID)
    }
    
    s.sendSuccess(w, map[string]string{"status": "logged out"})
}

// handleSessionCheck проверяет активность сессии
func (s *HTTPServer) handleSessionCheck(w http.ResponseWriter, r *http.Request) {
    sessionID := r.Header.Get("X-Session-ID")
    if sessionID == "" {
        auth := r.Header.Get("Authorization")
        if strings.HasPrefix(auth, "Bearer ") {
            sessionID = strings.TrimPrefix(auth, "Bearer ")
        }
    }
    
    if sessionID == "" {
        s.sendError(w, "No session", http.StatusUnauthorized)
        return
    }
    
    username, ok := s.sessions.Load(sessionID)
    if !ok {
        s.sendError(w, "Invalid session", http.StatusUnauthorized)
        return
    }
    
    s.sendSuccess(w, map[string]interface{}{
        "authenticated": true,
        "username":      username,
        "is_admin":      s.isUserAdmin(username.(string)),
    })
}

// handleDatabaseRequest обрабатывает запросы к БД
func (s *HTTPServer) handleDatabaseRequest(w http.ResponseWriter, r *http.Request) {
    // URL: /api/db/{database}/{collection}/{document_id}
    path := strings.TrimPrefix(r.URL.Path, "/api/db/")
    parts := strings.Split(path, "/")
    
    if len(parts) < 2 {
        s.sendError(w, "Invalid path. Use /api/db/{database}/{collection}[/{id}]", http.StatusBadRequest)
        return
    }
    
    database := parts[0]
    collection := parts[1]
    docID := ""
    if len(parts) > 2 {
        docID = parts[2]
    }
    
    // Проверка аутентификации
    sessionID := s.getSessionID(r)
    if sessionID == "" {
        s.sendError(w, "Authentication required", http.StatusUnauthorized)
        return
    }
    
    switch r.Method {
    case http.MethodGet:
        s.handleGetDocument(w, r, sessionID, database, collection, docID)
    case http.MethodPost:
        s.handleInsertDocument(w, r, sessionID, database, collection)
    case http.MethodPut:
        s.handleUpdateDocument(w, r, sessionID, database, collection, docID)
    case http.MethodDelete:
        s.handleDeleteDocument(w, r, sessionID, database, collection, docID)
    default:
        s.sendError(w, "Method not allowed", http.StatusMethodNotAllowed)
    }
}

// getSessionID извлекает session ID из запроса
func (s *HTTPServer) getSessionID(r *http.Request) string {
    sessionID := r.Header.Get("X-Session-ID")
    if sessionID == "" {
        auth := r.Header.Get("Authorization")
        if strings.HasPrefix(auth, "Bearer ") {
            sessionID = strings.TrimPrefix(auth, "Bearer ")
        }
    }
    return sessionID
}

// handleGetDocument обрабатывает GET запросы
func (s *HTTPServer) handleGetDocument(w http.ResponseWriter, r *http.Request, sessionID, database, collection, docID string) {
    // Проверка прав
    if !s.aclManager.CheckPermission(sessionID, database, collection, "read") {
        s.logOperation("GET_DOCUMENT", fmt.Sprintf("%s.%s", database, collection), "error", "Access denied", nil)
        s.sendError(w, "Access denied", http.StatusForbidden)
        return
    }
    
    db, err := s.store.GetDatabase(database)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    coll, err := db.GetCollection(collection)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    // Поиск по индексу или ID
    query := r.URL.Query()
    if indexName := query.Get("index"); indexName != "" {
        indexValue := query.Get("value")
        docs, err := coll.FindByIndex(indexName, indexValue)
        if err != nil {
            s.sendError(w, err.Error(), http.StatusNotFound)
            return
        }
        s.logOperation("GET_DOCUMENT_BY_INDEX", fmt.Sprintf("%s.%s[%s=%s]", database, collection, indexName, indexValue), "success", "", nil)
        s.sendSuccess(w, docs)
        return
    }
    
    if docID == "" {
        // Возвращаем все документы (с пагинацией)
        limit := 100
        if limitStr := query.Get("limit"); limitStr != "" {
            if l, err := strconv.Atoi(limitStr); err == nil && l > 0 && l <= 1000 {
                limit = l
            }
        }
        
        offset := 0
        if offsetStr := query.Get("offset"); offsetStr != "" {
            if o, err := strconv.Atoi(offsetStr); err == nil && o >= 0 {
                offset = o
            }
        }
        
        allDocs := coll.GetAllDocuments()
        start := offset
        end := offset + limit
        if start > len(allDocs) {
            start = len(allDocs)
        }
        if end > len(allDocs) {
            end = len(allDocs)
        }
        
        result := allDocs[start:end]
        s.logOperation("LIST_DOCUMENTS", fmt.Sprintf("%s.%s", database, collection), "success", "", map[string]interface{}{
            "total":  len(allDocs),
            "limit":  limit,
            "offset": offset,
        })
        s.sendSuccess(w, map[string]interface{}{
            "documents": result,
            "total":     len(allDocs),
            "limit":     limit,
            "offset":    offset,
        })
        return
    }
    
    doc, err := coll.Find(docID)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    s.logOperation("GET_DOCUMENT", fmt.Sprintf("%s.%s.%s", database, collection, docID), "success", "", nil)
    s.sendSuccess(w, doc)
}

// handleInsertDocument обрабатывает POST запросы
func (s *HTTPServer) handleInsertDocument(w http.ResponseWriter, r *http.Request, sessionID, database, collection string) {
    if !s.aclManager.CheckPermission(sessionID, database, collection, "write") {
        s.logOperation("INSERT_DOCUMENT", fmt.Sprintf("%s.%s", database, collection), "error", "Access denied", nil)
        s.sendError(w, "Access denied", http.StatusForbidden)
        return
    }
    
    db, err := s.store.GetDatabase(database)
    if err != nil {
        // Создаём БД если не существует
        if err := s.store.CreateDatabase(database); err != nil {
            s.sendError(w, err.Error(), http.StatusInternalServerError)
            return
        }
        db, _ = s.store.GetDatabase(database)
    }
    
    coll, err := db.GetCollection(collection)
    if err != nil {
        if err := db.CreateCollection(collection); err != nil {
            s.sendError(w, err.Error(), http.StatusInternalServerError)
            return
        }
        coll, _ = db.GetCollection(collection)
    }
    
    var doc map[string]interface{}
    if err := json.NewDecoder(r.Body).Decode(&doc); err != nil {
        s.sendError(w, "Invalid JSON", http.StatusBadRequest)
        return
    }
    
    docID := ""
    if id, ok := doc["_id"]; ok {
        if idStr, ok := id.(string); ok {
            docID = idStr
        }
    }
    
    if err := coll.InsertFromMap(doc); err != nil {
        s.logOperation("INSERT_DOCUMENT", fmt.Sprintf("%s.%s/%s", database, collection, docID), "error", err.Error(), nil)
        s.sendError(w, err.Error(), http.StatusBadRequest)
        return
    }
    
    s.logOperation("INSERT_DOCUMENT", fmt.Sprintf("%s.%s/%s", database, collection, docID), "success", "", nil)
    s.sendSuccess(w, map[string]interface{}{
        "status": "inserted",
        "id":     docID,
    })
}

// handleUpdateDocument обрабатывает PUT запросы
func (s *HTTPServer) handleUpdateDocument(w http.ResponseWriter, r *http.Request, sessionID, database, collection, docID string) {
    if docID == "" {
        s.sendError(w, "Document ID required", http.StatusBadRequest)
        return
    }
    
    if !s.aclManager.CheckPermission(sessionID, database, collection, "write") {
        s.logOperation("UPDATE_DOCUMENT", fmt.Sprintf("%s.%s.%s", database, collection, docID), "error", "Access denied", nil)
        s.sendError(w, "Access denied", http.StatusForbidden)
        return
    }
    
    db, err := s.store.GetDatabase(database)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    coll, err := db.GetCollection(collection)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    var updates map[string]interface{}
    if err := json.NewDecoder(r.Body).Decode(&updates); err != nil {
        s.sendError(w, "Invalid JSON", http.StatusBadRequest)
        return
    }
    
    if err := coll.Update(docID, updates); err != nil {
        s.logOperation("UPDATE_DOCUMENT", fmt.Sprintf("%s.%s.%s", database, collection, docID), "error", err.Error(), nil)
        s.sendError(w, err.Error(), http.StatusBadRequest)
        return
    }
    
    s.logOperation("UPDATE_DOCUMENT", fmt.Sprintf("%s.%s.%s", database, collection, docID), "success", "", map[string]interface{}{
        "updated_fields": len(updates),
    })
    s.sendSuccess(w, map[string]interface{}{
        "status": "updated",
        "id":     docID,
    })
}

// handleDeleteDocument обрабатывает DELETE запросы
func (s *HTTPServer) handleDeleteDocument(w http.ResponseWriter, r *http.Request, sessionID, database, collection, docID string) {
    if docID == "" {
        s.sendError(w, "Document ID required", http.StatusBadRequest)
        return
    }
    
    if !s.aclManager.CheckPermission(sessionID, database, collection, "delete") {
        s.logOperation("DELETE_DOCUMENT", fmt.Sprintf("%s.%s.%s", database, collection, docID), "error", "Access denied", nil)
        s.sendError(w, "Access denied", http.StatusForbidden)
        return
    }
    
    db, err := s.store.GetDatabase(database)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    coll, err := db.GetCollection(collection)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    // Проверяем параметр permanent для жёсткого удаления
    permanent := r.URL.Query().Get("permanent") == "true"
    
    if permanent {
        if err := coll.PermanentDelete(docID); err != nil {
            s.logOperation("PERMANENT_DELETE_DOCUMENT", fmt.Sprintf("%s.%s.%s", database, collection, docID), "error", err.Error(), nil)
            s.sendError(w, err.Error(), http.StatusNotFound)
            return
        }
        s.logOperation("PERMANENT_DELETE_DOCUMENT", fmt.Sprintf("%s.%s.%s", database, collection, docID), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status": "permanently_deleted",
            "id":     docID,
        })
        return
    }
    
    if err := coll.Delete(docID); err != nil {
        s.logOperation("DELETE_DOCUMENT", fmt.Sprintf("%s.%s.%s", database, collection, docID), "error", err.Error(), nil)
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    s.logOperation("DELETE_DOCUMENT", fmt.Sprintf("%s.%s.%s", database, collection, docID), "success", "", nil)
    s.sendSuccess(w, map[string]interface{}{
        "status": "deleted",
        "id":     docID,
    })
}

// handleIndexRequest обрабатывает запросы к индексам
func (s *HTTPServer) handleIndexRequest(w http.ResponseWriter, r *http.Request) {
    // URL: /api/index/{database}/{collection}/{action}
    path := strings.TrimPrefix(r.URL.Path, "/api/index/")
    parts := strings.Split(path, "/")
    
    if len(parts) < 3 {
        s.sendError(w, "Invalid path. Use /api/index/{database}/{collection}/{action}", http.StatusBadRequest)
        return
    }
    
    database := parts[0]
    collection := parts[1]
    action := parts[2]
    
    sessionID := s.getSessionID(r)
    if !s.aclManager.CheckPermission(sessionID, database, collection, "admin") {
        s.logOperation("INDEX_OPERATION", fmt.Sprintf("%s.%s.%s", database, collection, action), "error", "Admin access required", nil)
        s.sendError(w, "Admin access required", http.StatusForbidden)
        return
    }
    
    db, err := s.store.GetDatabase(database)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    coll, err := db.GetCollection(collection)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    switch action {
    case "list":
        indexes := coll.GetIndexesInfo()
        s.logOperation("LIST_INDEXES", fmt.Sprintf("%s.%s", database, collection), "success", "", nil)
        s.sendSuccess(w, indexes)
        
    case "create":
        var req struct {
            Name   string   `json:"name"`
            Fields []string `json:"fields"`
            Unique bool     `json:"unique"`
        }
        if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
            s.sendError(w, "Invalid request body", http.StatusBadRequest)
            return
        }
        if req.Name == "" {
            s.sendError(w, "Index name required", http.StatusBadRequest)
            return
        }
        if len(req.Fields) == 0 {
            s.sendError(w, "At least one field required", http.StatusBadRequest)
            return
        }
        if err := coll.CreateIndex(req.Name, req.Fields, req.Unique); err != nil {
            s.logOperation("CREATE_INDEX", fmt.Sprintf("%s.%s.%s", database, collection, req.Name), "error", err.Error(), nil)
            s.sendError(w, err.Error(), http.StatusBadRequest)
            return
        }
        s.logOperation("CREATE_INDEX", fmt.Sprintf("%s.%s.%s", database, collection, req.Name), "success", "", map[string]interface{}{
            "fields": req.Fields,
            "unique": req.Unique,
        })
        s.sendSuccess(w, map[string]interface{}{
            "status": "index_created",
            "name":   req.Name,
        })
        
    case "drop":
        if len(parts) < 4 {
            s.sendError(w, "Index name required", http.StatusBadRequest)
            return
        }
        indexName := parts[3]
        if err := coll.DropIndex(indexName); err != nil {
            s.logOperation("DROP_INDEX", fmt.Sprintf("%s.%s.%s", database, collection, indexName), "error", err.Error(), nil)
            s.sendError(w, err.Error(), http.StatusBadRequest)
            return
        }
        s.logOperation("DROP_INDEX", fmt.Sprintf("%s.%s.%s", database, collection, indexName), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status": "index_dropped",
            "name":   indexName,
        })
        
    default:
        s.sendError(w, "Unknown action", http.StatusBadRequest)
    }
}

// handleACLRequest обрабатывает запросы ACL
func (s *HTTPServer) handleACLRequest(w http.ResponseWriter, r *http.Request) {
    path := strings.TrimPrefix(r.URL.Path, "/api/acl/")
    parts := strings.Split(path, "/")
    
    if len(parts) < 1 {
        s.sendError(w, "Invalid path", http.StatusBadRequest)
        return
    }
    
    sessionID := s.getSessionID(r)
    if !s.aclManager.CheckPermission(sessionID, "*", "*", "admin") {
        s.logOperation("ACL_OPERATION", strings.Join(parts, "/"), "error", "Admin access required", nil)
        s.sendError(w, "Admin access required", http.StatusForbidden)
        return
    }
    
    action := parts[0]
    
    switch action {
    case "users":
        users := s.aclManager.ListUsers()
        s.sendSuccess(w, users)
        
    case "user":
        if len(parts) < 2 {
            s.sendError(w, "Username required", http.StatusBadRequest)
            return
        }
        username := parts[1]
        
        switch r.Method {
        case http.MethodGet:
            userInfo, err := s.aclManager.GetUserInfo(username)
            if err != nil {
                s.sendError(w, err.Error(), http.StatusNotFound)
                return
            }
            s.sendSuccess(w, userInfo)
            
        case http.MethodPost:
            var req struct {
                Password string   `json:"password"`
                Roles    []string `json:"roles"`
            }
            if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
                s.sendError(w, "Invalid request body", http.StatusBadRequest)
                return
            }
            if err := s.aclManager.CreateUser(username, req.Password, req.Roles); err != nil {
                s.logOperation("ACL_CREATE_USER", username, "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusBadRequest)
                return
            }
            s.logOperation("ACL_CREATE_USER", username, "success", "", map[string]interface{}{
                "roles": req.Roles,
            })
            s.sendSuccess(w, map[string]interface{}{
                "status": "user_created",
                "user":   username,
            })
            
        case http.MethodPut:
            var req struct {
                AddRole    string `json:"add_role"`
                RemoveRole string `json:"remove_role"`
                Password   string `json:"password"`
                Disable    bool   `json:"disable"`
                Enable     bool   `json:"enable"`
            }
            if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
                s.sendError(w, "Invalid request body", http.StatusBadRequest)
                return
            }
            
            if req.AddRole != "" {
                if err := s.aclManager.AddUserRole(username, req.AddRole); err != nil {
                    s.sendError(w, err.Error(), http.StatusBadRequest)
                    return
                }
            }
            if req.RemoveRole != "" {
                if err := s.aclManager.RemoveUserRole(username, req.RemoveRole); err != nil {
                    s.sendError(w, err.Error(), http.StatusBadRequest)
                    return
                }
            }
            if req.Password != "" {
                if err := s.aclManager.ChangePassword(username, req.Password); err != nil {
                    s.sendError(w, err.Error(), http.StatusBadRequest)
                    return
                }
            }
            if req.Disable {
                if err := s.aclManager.DisableUser(username); err != nil {
                    s.sendError(w, err.Error(), http.StatusBadRequest)
                    return
                }
            }
            if req.Enable {
                if err := s.aclManager.EnableUser(username); err != nil {
                    s.sendError(w, err.Error(), http.StatusBadRequest)
                    return
                }
            }
            s.logOperation("ACL_UPDATE_USER", username, "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "user_updated",
                "user":   username,
            })
            
        case http.MethodDelete:
            if err := s.aclManager.DeleteUser(username); err != nil {
                s.logOperation("ACL_DELETE_USER", username, "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusBadRequest)
                return
            }
            s.logOperation("ACL_DELETE_USER", username, "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "user_deleted",
                "user":   username,
            })
            
        default:
            s.sendError(w, "Method not allowed", http.StatusMethodNotAllowed)
        }
        
    case "roles":
        roles := s.aclManager.ListRoles()
        s.sendSuccess(w, roles)
        
    case "role":
        if len(parts) < 2 {
            s.sendError(w, "Role name required", http.StatusBadRequest)
            return
        }
        roleName := parts[1]
        
        switch r.Method {
        case http.MethodGet:
            perms, err := s.aclManager.GetRolePermissions(roleName)
            if err != nil {
                s.sendError(w, err.Error(), http.StatusNotFound)
                return
            }
            s.sendSuccess(w, map[string]interface{}{
                "name":        roleName,
                "permissions": perms,
            })
            
        case http.MethodPost:
            if err := s.aclManager.CreateRole(roleName); err != nil {
                s.logOperation("ACL_CREATE_ROLE", roleName, "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusBadRequest)
                return
            }
            s.logOperation("ACL_CREATE_ROLE", roleName, "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "role_created",
                "role":   roleName,
            })
            
        case http.MethodDelete:
            if err := s.aclManager.DeleteRole(roleName); err != nil {
                s.logOperation("ACL_DELETE_ROLE", roleName, "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusBadRequest)
                return
            }
            s.logOperation("ACL_DELETE_ROLE", roleName, "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "role_deleted",
                "role":   roleName,
            })
            
        default:
            s.sendError(w, "Method not allowed", http.StatusMethodNotAllowed)
        }
        
    case "grant":
        if len(parts) < 3 {
            s.sendError(w, "Role and permission required", http.StatusBadRequest)
            return
        }
        roleName := parts[1]
        permission := parts[2]
        if err := s.aclManager.GrantPermission(roleName, permission); err != nil {
            s.logOperation("ACL_GRANT", fmt.Sprintf("%s.%s", roleName, permission), "error", err.Error(), nil)
            s.sendError(w, err.Error(), http.StatusBadRequest)
            return
        }
        s.logOperation("ACL_GRANT", fmt.Sprintf("%s.%s", roleName, permission), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status":     "permission_granted",
            "role":       roleName,
            "permission": permission,
        })
        
    case "revoke":
        if len(parts) < 3 {
            s.sendError(w, "Role and permission required", http.StatusBadRequest)
            return
        }
        roleName := parts[1]
        permission := parts[2]
        if err := s.aclManager.RevokePermission(roleName, permission); err != nil {
            s.logOperation("ACL_REVOKE", fmt.Sprintf("%s.%s", roleName, permission), "error", err.Error(), nil)
            s.sendError(w, err.Error(), http.StatusBadRequest)
            return
        }
        s.logOperation("ACL_REVOKE", fmt.Sprintf("%s.%s", roleName, permission), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status":     "permission_revoked",
            "role":       roleName,
            "permission": permission,
        })
        
    default:
        s.sendError(w, "Unknown action", http.StatusBadRequest)
    }
}

// handleConstraintRequest обрабатывает запросы к ограничениям
func (s *HTTPServer) handleConstraintRequest(w http.ResponseWriter, r *http.Request) {
    path := strings.TrimPrefix(r.URL.Path, "/api/constraint/")
    parts := strings.Split(path, "/")
    
    if len(parts) < 3 {
        s.sendError(w, "Invalid path. Use /api/constraint/{database}/{collection}/{action}", http.StatusBadRequest)
        return
    }
    
    database := parts[0]
    collection := parts[1]
    action := parts[2]
    
    sessionID := s.getSessionID(r)
    if !s.aclManager.CheckPermission(sessionID, database, collection, "admin") {
        s.logOperation("CONSTRAINT_OPERATION", fmt.Sprintf("%s.%s.%s", database, collection, action), "error", "Admin access required", nil)
        s.sendError(w, "Admin access required", http.StatusForbidden)
        return
    }
    
    db, err := s.store.GetDatabase(database)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    coll, err := db.GetCollection(collection)
    if err != nil {
        s.sendError(w, err.Error(), http.StatusNotFound)
        return
    }
    
    switch action {
    case "list":
        constraints := make([]map[string]interface{}, 0)
        
        for _, field := range coll.GetRequiredFields() {
            constraints = append(constraints, map[string]interface{}{
                "type":  "required",
                "field": field,
            })
        }
        for _, field := range coll.GetUniqueConstraints() {
            constraints = append(constraints, map[string]interface{}{
                "type":  "unique",
                "field": field,
            })
        }
        for field, value := range coll.GetMinConstraints() {
            constraints = append(constraints, map[string]interface{}{
                "type":  "min",
                "field": field,
                "value": value,
            })
        }
        for field, value := range coll.GetMaxConstraints() {
            constraints = append(constraints, map[string]interface{}{
                "type":  "max",
                "field": field,
                "value": value,
            })
        }
        for field, values := range coll.GetEnumConstraints() {
            constraints = append(constraints, map[string]interface{}{
                "type":   "enum",
                "field":  field,
                "values": values,
            })
        }
        for field, pattern := range coll.GetRegexConstraints() {
            constraints = append(constraints, map[string]interface{}{
                "type":    "regex",
                "field":   field,
                "pattern": pattern,
            })
        }
        
        s.sendSuccess(w, constraints)
        
    case "required":
        if len(parts) < 4 {
            s.sendError(w, "Field name required", http.StatusBadRequest)
            return
        }
        field := parts[3]
        coll.AddRequiredField(field)
        s.logOperation("ADD_REQUIRED", fmt.Sprintf("%s.%s.%s", database, collection, field), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status": "required_field_added",
            "field":  field,
        })
        
    case "unique":
        if len(parts) < 4 {
            s.sendError(w, "Field name required", http.StatusBadRequest)
            return
        }
        field := parts[3]
        coll.AddUniqueConstraint(field)
        s.logOperation("ADD_UNIQUE", fmt.Sprintf("%s.%s.%s", database, collection, field), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status": "unique_constraint_added",
            "field":  field,
        })
        
    case "min":
        if len(parts) < 5 {
            s.sendError(w, "Field name and value required", http.StatusBadRequest)
            return
        }
        field := parts[3]
        minVal, err := strconv.ParseFloat(parts[4], 64)
        if err != nil {
            s.sendError(w, "Invalid minimum value", http.StatusBadRequest)
            return
        }
        coll.AddMinConstraint(field, minVal)
        s.logOperation("ADD_MIN", fmt.Sprintf("%s.%s.%s", database, collection, field), "success", "", map[string]interface{}{"value": minVal})
        s.sendSuccess(w, map[string]interface{}{
            "status": "min_constraint_added",
            "field":  field,
            "value":  minVal,
        })
        
    case "max":
        if len(parts) < 5 {
            s.sendError(w, "Field name and value required", http.StatusBadRequest)
            return
        }
        field := parts[3]
        maxVal, err := strconv.ParseFloat(parts[4], 64)
        if err != nil {
            s.sendError(w, "Invalid maximum value", http.StatusBadRequest)
            return
        }
        coll.AddMaxConstraint(field, maxVal)
        s.logOperation("ADD_MAX", fmt.Sprintf("%s.%s.%s", database, collection, field), "success", "", map[string]interface{}{"value": maxVal})
        s.sendSuccess(w, map[string]interface{}{
            "status": "max_constraint_added",
            "field":  field,
            "value":  maxVal,
        })
        
    case "enum":
        if len(parts) < 5 {
            s.sendError(w, "Field name and values required", http.StatusBadRequest)
            return
        }
        field := parts[3]
        values := make([]interface{}, len(parts)-4)
        for i := 4; i < len(parts); i++ {
            if num, err := strconv.ParseFloat(parts[i], 64); err == nil {
                values[i-4] = num
            } else {
                values[i-4] = parts[i]
            }
        }
        coll.AddEnumConstraint(field, values)
        s.logOperation("ADD_ENUM", fmt.Sprintf("%s.%s.%s", database, collection, field), "success", "", map[string]interface{}{"values": values})
        s.sendSuccess(w, map[string]interface{}{
            "status": "enum_constraint_added",
            "field":  field,
            "values": values,
        })
        
    case "regex":
        if len(parts) < 5 {
            s.sendError(w, "Field name and pattern required", http.StatusBadRequest)
            return
        }
        field := parts[3]
        pattern := parts[4]
        coll.AddRegexConstraint(field, pattern)
        s.logOperation("ADD_REGEX", fmt.Sprintf("%s.%s.%s", database, collection, field), "success", "", map[string]interface{}{"pattern": pattern})
        s.sendSuccess(w, map[string]interface{}{
            "status":  "regex_constraint_added",
            "field":   field,
            "pattern": pattern,
        })
        
    case "remove":
        if len(parts) < 5 {
            s.sendError(w, "Constraint type and field required", http.StatusBadRequest)
            return
        }
        constraintType := parts[3]
        field := parts[4]
        
        switch constraintType {
        case "required":
            coll.RemoveRequiredField(field)
        case "unique":
            coll.RemoveUniqueConstraint(field)
        case "min":
            coll.RemoveMinConstraint(field)
        case "max":
            coll.RemoveMaxConstraint(field)
        case "enum":
            coll.RemoveEnumConstraint(field)
        case "regex":
            coll.RemoveRegexConstraint(field)
        default:
            s.sendError(w, "Unknown constraint type", http.StatusBadRequest)
            return
        }
        s.logOperation("REMOVE_CONSTRAINT", fmt.Sprintf("%s.%s.%s.%s", database, collection, constraintType, field), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status": "constraint_removed",
            "type":   constraintType,
            "field":  field,
        })
        
    default:
        s.sendError(w, "Unknown action", http.StatusBadRequest)
    }
}

// handleClusterRequest обрабатывает запросы к кластеру
func (s *HTTPServer) handleClusterRequest(w http.ResponseWriter, r *http.Request) {
    sessionID := s.getSessionID(r)
    if !s.aclManager.CheckPermission(sessionID, "*", "*", "admin") {
        s.logOperation("CLUSTER_OPERATION", r.URL.Path, "error", "Admin access required", nil)
        s.sendError(w, "Admin access required", http.StatusForbidden)
        return
    }
    
    if s.coordinator == nil {
        s.sendError(w, "Cluster not available", http.StatusServiceUnavailable)
        return
    }
    
    path := strings.TrimPrefix(r.URL.Path, "/api/cluster/")
    parts := strings.Split(path, "/")
    
    switch r.Method {
    case http.MethodGet:
        if len(parts) == 0 || parts[0] == "" || parts[0] == "status" {
            status := s.coordinator.GetClusterStatus()
            s.logOperation("CLUSTER_STATUS", "", "success", "", nil)
            s.sendSuccess(w, status)
        } else if parts[0] == "health" {
            // Используем GetClusterStatus для получения информации о здоровье
            status := s.coordinator.GetClusterStatus()
            s.sendSuccess(w, map[string]interface{}{
                "status":             status.Health,
                "total_nodes":        status.TotalNodes,
                "active_nodes":       status.ActiveNodes,
                "syncing_nodes":      status.SyncingNodes,
                "failed_nodes":       status.FailedNodes,
                "replication_factor": status.ReplicationFactor,
                "leader_id":          status.LeaderID,
                "checked_at":         time.Now().UnixMilli(),
            })
        } else if parts[0] == "nodes" {
            nodes := s.coordinator.GetAllNodes()
            s.sendSuccess(w, nodes)
        } else if parts[0] == "leader" {
            leader := s.coordinator.GetLeader()
            s.sendSuccess(w, leader)
        } else if parts[0] == "shards" {
            shards := s.coordinator.GetAllShards()
            s.sendSuccess(w, shards)
        } else if parts[0] == "pipeline" {
            stats := s.coordinator.GetPipelineStats()
            s.sendSuccess(w, stats)
        } else if parts[0] == "batch" {
            stats := s.coordinator.GetBatchCommitStats()
            s.sendSuccess(w, stats)
        } else if parts[0] == "resharding" {
            stats := s.coordinator.GetReshardingStats()
            s.sendSuccess(w, stats)
        } else {
            s.sendError(w, "Unknown cluster endpoint", http.StatusNotFound)
        }
        
    case http.MethodPost:
        if len(parts) >= 2 && parts[0] == "replication" && parts[1] == "factor" {
            var req struct {
                Factor int `json:"factor"`
            }
            if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
                s.sendError(w, "Invalid request body", http.StatusBadRequest)
                return
            }
            if req.Factor < 1 || req.Factor > 5 {
                s.sendError(w, "Replication factor must be between 1 and 5", http.StatusBadRequest)
                return
            }
            if err := s.coordinator.SetReplicationFactor(req.Factor); err != nil {
                s.logOperation("SET_REPLICATION_FACTOR", fmt.Sprintf("%d", req.Factor), "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusBadRequest)
                return
            }
            s.logOperation("SET_REPLICATION_FACTOR", fmt.Sprintf("%d", req.Factor), "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "updated",
                "factor": req.Factor,
            })
        } else if len(parts) >= 1 && parts[0] == "resharding" {
            var req struct {
                Reason string `json:"reason"`
            }
            json.NewDecoder(r.Body).Decode(&req)
            if req.Reason == "" {
                req.Reason = "api_trigger"
            }
            if err := s.coordinator.TriggerResharding(req.Reason); err != nil {
                s.logOperation("TRIGGER_RESHARDING", "", "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusBadRequest)
                return
            }
            s.logOperation("TRIGGER_RESHARDING", "", "success", "", map[string]interface{}{"reason": req.Reason})
            s.sendSuccess(w, map[string]interface{}{
                "status": "resharding_triggered",
                "reason": req.Reason,
            })
        } else {
            s.sendError(w, "Unknown cluster endpoint", http.StatusNotFound)
        }
        
    default:
        s.sendError(w, "Method not allowed", http.StatusMethodNotAllowed)
    }
}

// handleTriggerRequest обрабатывает запросы к триггерам
func (s *HTTPServer) handleTriggerRequest(w http.ResponseWriter, r *http.Request) {
    path := strings.TrimPrefix(r.URL.Path, "/api/trigger/")
    parts := strings.Split(path, "/")
    
    if len(parts) < 2 {
        s.sendError(w, "Invalid path. Use /api/trigger/{collection}/{action}", http.StatusBadRequest)
        return
    }
    
    collection := parts[0]
    action := parts[1]
    
    sessionID := s.getSessionID(r)
    if !s.aclManager.CheckPermission(sessionID, "", collection, "admin") {
        s.logOperation("TRIGGER_OPERATION", fmt.Sprintf("%s.%s", collection, action), "error", "Admin access required", nil)
        s.sendError(w, "Admin access required", http.StatusForbidden)
        return
    }
    
    tm := storage.GetTriggerManager()
    if tm == nil {
        storage.InitTriggerManager(s.logger)
        tm = storage.GetTriggerManager()
    }
    
    switch action {
    case "list":
        triggers := tm.ListTriggers(collection)
        s.sendSuccess(w, triggers)
        
    case "create":
        if r.Method != http.MethodPost {
            s.sendError(w, "Method not allowed", http.StatusMethodNotAllowed)
            return
        }
        
        var req struct {
            Name        string                 `json:"name"`
            Event       string                 `json:"event"`
            Action      string                 `json:"action"`
            Condition   map[string]interface{} `json:"condition"`
            Description string                 `json:"description"`
        }
        if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
            s.sendError(w, "Invalid request body", http.StatusBadRequest)
            return
        }
        
        if req.Name == "" {
            s.sendError(w, "Trigger name required", http.StatusBadRequest)
            return
        }
        
        config := make(map[string]interface{})
        config["action"] = req.Action
        config["description"] = req.Description
        if req.Condition != nil {
            config["condition"] = req.Condition
        }
        
        event := storage.TriggerEvent(req.Event)
        if event == "" {
            event = storage.TriggerAfterInsert
        }
        
        if err := tm.CreateTrigger("", collection, req.Name, event, config); err != nil {
            s.logOperation("CREATE_TRIGGER", fmt.Sprintf("%s.%s", collection, req.Name), "error", err.Error(), nil)
            s.sendError(w, err.Error(), http.StatusBadRequest)
            return
        }
        s.logOperation("CREATE_TRIGGER", fmt.Sprintf("%s.%s", collection, req.Name), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status": "trigger_created",
            "name":   req.Name,
        })
        
    case "drop":
        if len(parts) < 3 {
            s.sendError(w, "Trigger name required", http.StatusBadRequest)
            return
        }
        triggerName := parts[2]
        if err := tm.DropTrigger(collection, "", triggerName); err != nil {
            s.logOperation("DROP_TRIGGER", fmt.Sprintf("%s.%s", collection, triggerName), "error", err.Error(), nil)
            s.sendError(w, err.Error(), http.StatusNotFound)
            return
        }
        s.logOperation("DROP_TRIGGER", fmt.Sprintf("%s.%s", collection, triggerName), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status": "trigger_dropped",
            "name":   triggerName,
        })
        
    case "enable":
        if len(parts) < 3 {
            s.sendError(w, "Trigger name required", http.StatusBadRequest)
            return
        }
        triggerName := parts[2]
        if err := tm.EnableTrigger(collection, "", triggerName); err != nil {
            s.logOperation("ENABLE_TRIGGER", fmt.Sprintf("%s.%s", collection, triggerName), "error", err.Error(), nil)
            s.sendError(w, err.Error(), http.StatusNotFound)
            return
        }
        s.logOperation("ENABLE_TRIGGER", fmt.Sprintf("%s.%s", collection, triggerName), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status": "trigger_enabled",
            "name":   triggerName,
        })
        
    case "disable":
        if len(parts) < 3 {
            s.sendError(w, "Trigger name required", http.StatusBadRequest)
            return
        }
        triggerName := parts[2]
        if err := tm.DisableTrigger(collection, "", triggerName); err != nil {
            s.logOperation("DISABLE_TRIGGER", fmt.Sprintf("%s.%s", collection, triggerName), "error", err.Error(), nil)
            s.sendError(w, err.Error(), http.StatusNotFound)
            return
        }
        s.logOperation("DISABLE_TRIGGER", fmt.Sprintf("%s.%s", collection, triggerName), "success", "", nil)
        s.sendSuccess(w, map[string]interface{}{
            "status": "trigger_disabled",
            "name":   triggerName,
        })
        
    case "log":
        logs := tm.GetTriggerExecutionLog()
        // Фильтруем по коллекции
        filtered := make([]interface{}, 0)
        for _, entry := range logs {
            if entry.Collection == collection {
                filtered = append(filtered, entry)
            }
        }
        s.sendSuccess(w, filtered)
        
    default:
        s.sendError(w, "Unknown action", http.StatusBadRequest)
    }
}

// handleTransactionRequest обрабатывает запросы к транзакциям
func (s *HTTPServer) handleTransactionRequest(w http.ResponseWriter, r *http.Request) {
    path := strings.TrimPrefix(r.URL.Path, "/api/transaction/")
    parts := strings.Split(path, "/")
    
    sessionID := s.getSessionID(r)
    if !s.aclManager.CheckPermission(sessionID, "*", "*", "write") {
        s.logOperation("TRANSACTION_OPERATION", "", "error", "Write access required", nil)
        s.sendError(w, "Write access required", http.StatusForbidden)
        return
    }
    
    switch r.Method {
    case http.MethodGet:
        transactions := storage.GetActiveTransactions()
        s.sendSuccess(w, transactions)
        
    case http.MethodPost:
        if len(parts) == 0 || parts[0] == "" {
            s.sendError(w, "Action required", http.StatusBadRequest)
            return
        }
        
        action := parts[0]
        switch action {
        case "begin":
            if err := storage.InitTransactionManager("futriis.wal"); err != nil {
                s.logOperation("BEGIN_TRANSACTION", "", "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusInternalServerError)
                return
            }
            
            databases := s.store.ListDatabases()
            if len(databases) == 0 {
                s.sendError(w, "No database available", http.StatusBadRequest)
                return
            }
            
            db, err := s.store.GetDatabase(databases[0])
            if err != nil {
                s.sendError(w, err.Error(), http.StatusBadRequest)
                return
            }
            
            collections := db.ListCollections()
            if len(collections) == 0 {
                s.sendError(w, "No collection available", http.StatusBadRequest)
                return
            }
            
            coll, err := db.GetCollection(collections[0])
            if err != nil {
                s.sendError(w, err.Error(), http.StatusBadRequest)
                return
            }
            
            if err := storage.BeginTransactionOnCollection(coll); err != nil {
                s.logOperation("BEGIN_TRANSACTION", "", "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusInternalServerError)
                return
            }
            s.logOperation("BEGIN_TRANSACTION", "", "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "transaction_started",
                "id":     storage.GetCurrentTransactionID(),
            })
            
        case "commit":
            if err := storage.CommitCurrentTransaction(); err != nil {
                s.logOperation("COMMIT_TRANSACTION", "", "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusInternalServerError)
                return
            }
            s.logOperation("COMMIT_TRANSACTION", "", "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "transaction_committed",
            })
            
        case "abort":
            if err := storage.AbortCurrentTransaction(); err != nil {
                s.logOperation("ABORT_TRANSACTION", "", "error", err.Error(), nil)
                s.sendError(w, err.Error(), http.StatusInternalServerError)
                return
            }
            s.logOperation("ABORT_TRANSACTION", "", "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "transaction_aborted",
            })
            
        default:
            s.sendError(w, "Unknown action", http.StatusBadRequest)
        }
        
    case http.MethodPut:
        if len(parts) < 2 {
            s.sendError(w, "Transaction ID and action required", http.StatusBadRequest)
            return
        }
        txID := parts[0]
        action := parts[1]
        
        switch action {
        case "commit":
            // В реальной реализации здесь был бы двухфазный коммит
            s.logOperation("COMMIT_TRANSACTION", txID, "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "transaction_committed",
                "id":     txID,
            })
            
        case "abort":
            s.logOperation("ABORT_TRANSACTION", txID, "success", "", nil)
            s.sendSuccess(w, map[string]interface{}{
                "status": "transaction_aborted",
                "id":     txID,
            })
            
        default:
            s.sendError(w, "Unknown action", http.StatusBadRequest)
        }
        
    default:
        s.sendError(w, "Method not allowed", http.StatusMethodNotAllowed)
    }
}

// handleHealthCheck обрабатывает проверку здоровья (без аутентификации)
func (s *HTTPServer) handleHealthCheck(w http.ResponseWriter, r *http.Request) {
    status := "healthy"
    httpStatus := http.StatusOK
    
    if s.store == nil {
        status = "unhealthy"
        httpStatus = http.StatusServiceUnavailable
    }
    
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(httpStatus)
    json.NewEncoder(w).Encode(map[string]interface{}{
        "status":    status,
        "timestamp": time.Now().UnixMilli(),
        "version":   "1.0.0",
    })
}

// handleMetricsEndpoint обрабатывает запрос метрик (без аутентификации)
func (s *HTTPServer) handleMetricsEndpoint(w http.ResponseWriter, r *http.Request) {
    metrics := map[string]interface{}{
        "timestamp":    time.Now().UnixMilli(),
        "rate_limiter": s.rateLimiter.GetStats(),
        "store_stats":  s.store.GetStats(),
    }
    
    if s.coordinator != nil {
        metrics["cluster"] = s.coordinator.GetClusterStatus()
        metrics["pipeline"] = s.coordinator.GetPipelineStats()
        metrics["batch_commit"] = s.coordinator.GetBatchCommitStats()
        metrics["resharding"] = s.coordinator.GetReshardingStats()
    }
    
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(http.StatusOK)
    json.NewEncoder(w).Encode(metrics)
}

// logOperation логирует операцию
func (s *HTTPServer) logOperation(operation, target, status, errMsg string, details map[string]interface{}) {
    if s.logger == nil {
        return
    }
    
    logMsg := fmt.Sprintf("[API] %s: %s - %s", operation, target, status)
    if errMsg != "" {
        logMsg += " - " + errMsg
    }
    
    if status == "error" {
        s.logger.Error(logMsg)
    } else {
        s.logger.Info(logMsg)
    }
}

// sendSuccess отправляет успешный ответ
func (s *HTTPServer) sendSuccess(w http.ResponseWriter, data interface{}) {
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(http.StatusOK)
    json.NewEncoder(w).Encode(APIResponse{
        Success: true,
        Data:    data,
    })
}

// sendError отправляет ответ с ошибкой
func (s *HTTPServer) sendError(w http.ResponseWriter, errMsg string, statusCode int) {
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(statusCode)
    json.NewEncoder(w).Encode(APIResponse{
        Success: false,
        Error:   errMsg,
    })
}