futriix/internal/storage/transaction.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/storage/transaction.go
// Назначение: Реализация транзакций с поддержкой MVCC (Multi-Version Concurrency Control)
// и WAL (Write-Ahead Log) без блокировок. Использует атомарные операции и версионирование.
// ОПТИМИЗИРОВАНО: пакетная запись WAL, буферизация, асинхронная обработка.
// УЛУЧШЕНИЯ: Segmented WAL, Parallel Recovery, Visibility Map, Version Pruning,
// Distributed Transactions, Deadlock Detection, Transaction Timeout, Savepoints,
// Асинхронное восстановление WAL, реальный fsync.

package storage

import (
    "bufio"
    "encoding/binary"
    "encoding/json"
    "fmt"
    "os"
    "path/filepath"
    "sort"
    "sync"
    "sync/atomic"
    "time"
)

// ========== Базовые типы ==========

// TransactionID представляет уникальный идентификатор транзакции
type TransactionID uint64

// TransactionState представляет состояние транзакции
type TransactionState int32

const (
    TransactionActive TransactionState = iota
    TransactionCommitted
    TransactionAborted
)

// TransactionRecord представляет запись в WAL
type TransactionRecord struct {
    ID         TransactionID     `json:"id"`
    State      TransactionState  `json:"state"`
    Timestamp  int64             `json:"timestamp"`
    Operations []Operation       `json:"operations"`
}

// Operation представляет одну операцию в транзакции
type Operation struct {
    Type       string                 `json:"type"` // "insert", "update", "delete"
    Database   string                 `json:"database"`
    Collection string                 `json:"collection"`
    DocumentID string                 `json:"document_id"`
    Data       map[string]interface{} `json:"data"`
    Version    uint64                 `json:"version"`
    OldData    map[string]interface{} `json:"old_data"` // для отката
}

// DocumentVersion представляет версию документа для MVCC
type DocumentVersion struct {
    Document  *Document      `json:"document"`
    Timestamp int64          `json:"timestamp"`
    TxID      TransactionID  `json:"tx_id"`
}

// TransactionInfo представляет информацию о транзакции для API
type TransactionInfo struct {
    ID             string          `json:"id"`
    Status         string          `json:"status"`
    StartTime      int64           `json:"start_time"`
    OperationCount int             `json:"operation_count"`
    Operations     []OperationInfo `json:"operations,omitempty"`
}

// OperationInfo представляет информацию об операции для API
type OperationInfo struct {
    Type       string `json:"type"`
    Database   string `json:"database"`
    Collection string `json:"collection"`
    DocumentID string `json:"document_id"`
}

// WALRecord представляет запись в WAL файле
type WALRecord struct {
    CRC        uint32 `json:"crc"`
    Length     uint32 `json:"length"`
    Type       byte   `json:"type"` // 1=Transaction, 2=Checkpoint
    Data       []byte `json:"data"`
    Timestamp  int64  `json:"timestamp"`
    LSN        uint64 `json:"lsn"` // Log Sequence Number
}

// LoggerInterface определяет интерфейс для логирования
type LoggerInterface interface {
    Debug(msg string)
    Info(msg string)
    Error(msg string)
    Warn(msg string)
}

// ========== Константы для улучшений ==========

const (
    // WAL Segment
    WALSegmentSize      = 64 * 1024 * 1024 // 64MB per segment
    WALSegmentPrefix    = "wal_segment_"
    WALIndexPrefix      = "wal_index_"

    // MVCC
    VisibilityMapSize    = 1024 * 1024
    VersionPruneInterval = 5 * time.Minute
    MaxVersionsPerDoc    = 100
    VersionRetentionDays = 7

    // Transactions
    DefaultTxTimeout        = 30 * time.Second
    DeadlockCheckInterval   = 1 * time.Second
    MaxSavepointsPerTx      = 100

    // Distributed Transactions
    TwoPhaseCommitTimeout   = 10 * time.Second

    // Асинхронное восстановление WAL
    AsyncRecoveryBufferSize = 10000
    AsyncRecoveryWorkers    = 4
    AsyncRecoveryTimeout    = 30 * time.Second
    
    // Fsync
    FsyncMaxRetries         = 3
    FsyncRetryDelay         = 100 * time.Millisecond
)

// ========== CRC32 таблица ==========

var crc32Table = [256]uint32{
    0x00000000, 0x77073096, 0xee0e612c, 0x990951ba, 0x076dc419, 0x706af48f,
    0xe963a535, 0x9e6495a3, 0x0edb8832, 0x79dcb8a4, 0xe0d5e91e, 0x97d2d988,
    0x09b64c2b, 0x7eb17cbd, 0xe7b82d07, 0x90bf1d91, 0x1db71064, 0x6ab020f2,
    0xf3b97148, 0x84be41de, 0x1adad47d, 0x6ddde4eb, 0xf4d4b551, 0x83d385c7,
    0x136c9856, 0x646ba8c0, 0xfd62f97a, 0x8a65c9ec, 0x14015c4f, 0x63066cd9,
    0xfa0f3d63, 0x8d080df5, 0x3b6e20c8, 0x4c69105e, 0xd56041e4, 0xa2677172,
    0x3c03e4d1, 0x4b04d447, 0xd20d85fd, 0xa50ab56b, 0x35b5a8fa, 0x42b2986c,
    0xdbbbc9d6, 0xacbcf940, 0x32d86ce3, 0x45df5c75, 0xdcd60dcf, 0xabd13d59,
    0x26d930ac, 0x51de003a, 0xc8d75180, 0xbfd06116, 0x21b4f4b5, 0x56b3c423,
    0xcfba9599, 0xb8bda50f, 0x2802b89e, 0x5f058808, 0xc60cd9b2, 0xb10be924,
    0x2f6f7c87, 0x58684c11, 0xc1611dab, 0xb6662d3d, 0x76dc4190, 0x01db7106,
    0x98d220bc, 0xefd5102a, 0x71b18589, 0x06b6b51f, 0x9fbfe4a5, 0xe8b8d433,
    0x7807c9a2, 0x0f00f934, 0x9609a88e, 0xe10e9818, 0x7f6a0dbb, 0x086d3d2d,
    0x91646c97, 0xe6635c01, 0x6b6b51f4, 0x1c6c6162, 0x856530d8, 0xf262004e,
    0x6c0695ed, 0x1b01a57b, 0x8208f4c1, 0xf50fc457, 0x65b0d9c6, 0x12b7e950,
    0x8bbeb8ea, 0xfcb9887c, 0x62dd1ddf, 0x15da2d49, 0x8cd37cf3, 0xfbd44c65,
    0x4db26158, 0x3ab551ce, 0xa3bc0074, 0xd4bb30e2, 0x4adfa541, 0x3dd895d7,
    0xa4d1c46d, 0xd3d6f4fb, 0x4369e96a, 0x346ed9fc, 0xad678846, 0xda60b8d0,
    0x44042d73, 0x33031de5, 0xaa0a4c5f, 0xdd0d7cc9, 0x5005713c, 0x270241aa,
    0xbe0b1010, 0xc90c2086, 0x5768b525, 0x206f85b3, 0xb966d409, 0xce61e49f,
    0x5edef90e, 0x29d9c998, 0xb0d09822, 0xc7d7a8b4, 0x59b33d17, 0x2eb40d81,
    0xb7bd5c3b, 0xc0ba6cad, 0xedb88320, 0x9abfb3b6, 0x03b6e20c, 0x74b1d29a,
    0xead54739, 0x9dd277af, 0x04db2615, 0x73dc1683, 0xe3630b12, 0x94643b84,
    0x0d6d6a3e, 0x7a6a5aa8, 0xe40ecf0b, 0x9309ff9d, 0x0a00ae27, 0x7d079eb1,
    0xf00f9344, 0x8708a3d2, 0x1e01f268, 0x6906c2fe, 0xf762575d, 0x806567cb,
    0x196c3671, 0x6e6b06e7, 0xfed41b76, 0x89d32be0, 0x10da7a5a, 0x67dd4acc,
    0xf9b9df6f, 0x8ebeeff9, 0x17b7be43, 0x60b08ed5, 0xd6d6a3e8, 0xa1d1937e,
    0x38d8c2c4, 0x4fdff252, 0xd1bb67f1, 0xa6bc5767, 0x3fb506dd, 0x48b2364b,
    0xd80d2bda, 0xaf0a1a4c, 0x36034af6, 0x41047a60, 0xdf60efc3, 0xa867df55,
    0x316e8eef, 0x4669be79, 0xcb61b38c, 0xbc66831a, 0x256fd2a0, 0x5268e236,
    0xcc0c7795, 0xbb0b4703, 0x220216b9, 0x5505262f, 0xc5ba3bbe, 0xb2bd0b28,
    0x2bb45a92, 0x5cb36a04, 0xc2d7ffa7, 0xb5d0cf31, 0x2cd99e8b, 0x5bdeae1d,
    0x9b64c2b0, 0xec63f226, 0x756aa39c, 0x026d930a, 0x9c0906a9, 0xeb0e363f,
    0x72076785, 0x05005713, 0x95bf4a82, 0xe2b87a14, 0x7bb12bae, 0x0cb61b38,
    0x92d28e9b, 0xe5d5be0d, 0x7cdcefb7, 0x0bdbdf21, 0x86d3d2d4, 0xf1d4e242,
    0x68ddb3f8, 0x1fda836e, 0x81be16cd, 0xf6b9265b, 0x6fb077e1, 0x18b74777,
    0x88085ae6, 0xff0f6a70, 0x66063bca, 0x11010b5c, 0x8f659eff, 0xf862ae69,
    0x616bffd3, 0x166ccf45, 0xa00ae278, 0xd70dd2ee, 0x4e048354, 0x3903b3c2,
    0xa7672661, 0xd06016f7, 0x4969474d, 0x3e6e77db, 0xaed16a4a, 0xd9d65adc,
    0x40df0b66, 0x37d83bf0, 0xa9bcae53, 0xdebb9ec5, 0x47b2cf7f, 0x30b5ffe9,
    0xbdbdf21c, 0xcabac28a, 0x53b39330, 0x24b4a3a6, 0xbad03605, 0xcdd70693,
    0x54de5729, 0x23d967bf, 0xb3667a2e, 0xc4614ab8, 0x5d681b02, 0x2a6f2b94,
    0xb40bbe37, 0xc30c8ea1, 0x5a05df1b, 0x2d02ef8d,
}

// crc32 вычисляет CRC32 хеш
func crc32(data []byte) uint32 {
    crc := uint32(0xFFFFFFFF)
    for _, b := range data {
        crc = (crc >> 8) ^ crc32Table[(crc^uint32(b))&0xFF]
    }
    return crc ^ 0xFFFFFFFF
}

// ========== WALManager (оригинальный) ==========

// WALManager управляет Write-Ahead Log с оптимизациями для высокой нагрузки
type WALManager struct {
    mu           sync.RWMutex
    file         *os.File
    writer       *bufio.Writer
    path         string
    currentLSN   uint64
    lastSync     time.Time
    syncInterval time.Duration
    bufferSize   int
    closed       bool
    writeChan    chan *WALRecord
    stopChan     chan struct{}
    wg           sync.WaitGroup
    batchSize    int
}

// NewWALManager создаёт новый WAL менеджер с оптимизациями
func NewWALManager(path string) (*WALManager, error) {
    dir := filepath.Dir(path)
    if err := os.MkdirAll(dir, 0755); err != nil {
        return nil, fmt.Errorf("failed to create WAL directory: %v", err)
    }

    file, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_RDWR, 0644)
    if err != nil {
        return nil, fmt.Errorf("failed to open WAL file: %v", err)
    }

    var currentLSN uint64 = 1
    stat, err := file.Stat()
    if err == nil && stat.Size() > 0 {
        currentLSN = uint64(stat.Size()) / 100
        if currentLSN < 1 {
            currentLSN = 1
        }
    }

    wm := &WALManager{
        file:         file,
        writer:       bufio.NewWriterSize(file, 64*1024),
        path:         path,
        currentLSN:   currentLSN,
        lastSync:     time.Now(),
        syncInterval: 5 * time.Second,
        bufferSize:   64 * 1024,
        writeChan:    make(chan *WALRecord, 10000),
        stopChan:     make(chan struct{}),
        batchSize:    100,
    }

    wm.wg.Add(1)
    go wm.writerLoop()

    return wm, nil
}

// writerLoop асинхронно записывает записи в WAL пакетами
func (wm *WALManager) writerLoop() {
    defer wm.wg.Done()

    batch := make([]*WALRecord, 0, wm.batchSize)
    ticker := time.NewTicker(wm.syncInterval)
    defer ticker.Stop()

    for {
        select {
        case record, ok := <-wm.writeChan:
            if !ok {
                if len(batch) > 0 {
                    wm.flushBatch(batch)
                }
                return
            }
            batch = append(batch, record)
            if len(batch) >= wm.batchSize {
                wm.flushBatch(batch)
                batch = batch[:0]
            }

        case <-ticker.C:
            if len(batch) > 0 {
                wm.flushBatch(batch)
                batch = batch[:0]
            }
            if time.Since(wm.lastSync) >= wm.syncInterval {
                wm.sync()
            }

        case <-wm.stopChan:
            if len(batch) > 0 {
                wm.flushBatch(batch)
            }
            wm.sync()
            return
        }
    }
}

// flushBatch записывает пакет записей
func (wm *WALManager) flushBatch(batch []*WALRecord) {
    wm.mu.Lock()
    defer wm.mu.Unlock()

    for _, record := range batch {
        data, err := json.Marshal(record)
        if err != nil {
            continue
        }

        lenBuf := make([]byte, 4)
        binary.BigEndian.PutUint32(lenBuf, uint32(len(data)))
        if _, err := wm.writer.Write(lenBuf); err != nil {
            continue
        }

        if _, err := wm.writer.Write(data); err != nil {
            continue
        }

        record.LSN = wm.currentLSN
        wm.currentLSN++
    }
}

// sync синхронизирует буфер с диском с реальным fsync
func (wm *WALManager) sync() {
    wm.mu.Lock()
    defer wm.mu.Unlock()

    if err := wm.writer.Flush(); err == nil {
        // Реальный fsync с повторными попытками
        if err := RealFsyncWithRetry(wm.file, FsyncMaxRetries, FsyncRetryDelay); err == nil {
            wm.lastSync = time.Now()
        }
    }
}

// Write записывает запись в WAL
func (wm *WALManager) Write(record *WALRecord) error {
    if wm.closed {
        return fmt.Errorf("WAL is closed")
    }

    record.Timestamp = time.Now().UnixMilli()

    data, err := json.Marshal(record.Data)
    if err != nil {
        return err
    }
    record.CRC = crc32(data)

    select {
    case wm.writeChan <- record:
        return nil
    case <-time.After(100 * time.Millisecond):
        return fmt.Errorf("WAL write timeout")
    }
}

// ReadAll читает все записи из WAL
func (wm *WALManager) ReadAll() ([]*WALRecord, error) {
    wm.mu.RLock()
    defer wm.mu.RUnlock()

    wm.writer.Flush()

    file, err := os.Open(wm.path)
    if err != nil {
        return nil, err
    }
    defer file.Close()

    records := make([]*WALRecord, 0)
    reader := bufio.NewReader(file)
    lenBuf := make([]byte, 4)

    for {
        _, err := reader.Read(lenBuf)
        if err != nil {
            break
        }

        recordLen := binary.BigEndian.Uint32(lenBuf)
        recordData := make([]byte, recordLen)
        _, err = reader.Read(recordData)
        if err != nil {
            break
        }

        var record WALRecord
        if err := json.Unmarshal(recordData, &record); err != nil {
            continue
        }

        data, _ := json.Marshal(record.Data)
        if crc32(data) != record.CRC {
            continue
        }

        records = append(records, &record)
    }

    return records, nil
}

// Close закрывает WAL с финальной синхронизацией
func (wm *WALManager) Close() error {
    wm.mu.Lock()
    wm.closed = true
    wm.mu.Unlock()

    close(wm.stopChan)
    close(wm.writeChan)
    wm.wg.Wait()

    wm.mu.Lock()
    defer wm.mu.Unlock()

    if err := wm.writer.Flush(); err != nil {
        return err
    }
    // Финальный fsync
    RealFsync(wm.file)
    return wm.file.Close()
}

// ========== Segmented WAL Manager ==========

// WALSegment представляет сегмент WAL файла
type WALSegment struct {
    ID       uint32
    File     *os.File
    Writer   *bufio.Writer
    Path     string
    StartLSN uint64
    EndLSN   uint64
    Size     int64
    mu       sync.Mutex
}

// WALIndexEntry представляет запись в индексе WAL
type WALIndexEntry struct {
    LSN       uint64
    SegmentID uint32
    Offset    int64
    Length    uint32
    Checksum  uint32
}

// WALIndexManager управляет индексом для быстрого поиска LSN
type WALIndexManager struct {
    index     map[uint64]*WALIndexEntry
    segments  map[uint32]*WALSegment
    mu        sync.RWMutex
    indexPath string
}

// SegmentedWALManager расширяет WALManager с поддержкой сегментов
type SegmentedWALManager struct {
    segmentsDir      string
    segments         map[uint32]*WALSegment
    currentSegment   *WALSegment
    currentSegmentID uint32
    index            *WALIndexManager
    mu               sync.RWMutex
    writeChan        chan *WALRecord
    stopChan         chan struct{}
    wg               sync.WaitGroup
    batchSize        int
    logger           LoggerInterface
    recoveryManager  *AsyncRecoveryManager
    recoveryComplete atomic.Bool
}

// NewSegmentedWALManager создаёт новый сегментированный WAL менеджер
func NewSegmentedWALManager(segmentsDir string, logger LoggerInterface) (*SegmentedWALManager, error) {
    if err := os.MkdirAll(segmentsDir, 0755); err != nil {
        return nil, fmt.Errorf("failed to create segments dir: %v", err)
    }

    wm := &SegmentedWALManager{
        segmentsDir: segmentsDir,
        segments:    make(map[uint32]*WALSegment),
        index: &WALIndexManager{
            index:     make(map[uint64]*WALIndexEntry),
            segments:  make(map[uint32]*WALSegment),
            indexPath: filepath.Join(segmentsDir, WALIndexPrefix+"index.json"),
        },
        writeChan: make(chan *WALRecord, 10000),
        stopChan:  make(chan struct{}),
        batchSize: 100,
        logger:    logger,
    }

    if err := wm.loadExistingSegments(); err != nil {
        return nil, err
    }

    if err := wm.index.load(); err != nil {
        if logger != nil {
            logger.Warn(fmt.Sprintf("Failed to load WAL index: %v", err))
        }
    }

    if wm.currentSegment == nil {
        if err := wm.rotateSegment(); err != nil {
            return nil, err
        }
    }

    wm.wg.Add(1)
    go wm.writerLoop()

    return wm, nil
}

// loadExistingSegments загружает существующие сегменты WAL
func (wm *SegmentedWALManager) loadExistingSegments() error {
    files, err := filepath.Glob(filepath.Join(wm.segmentsDir, WALSegmentPrefix+"*"))
    if err != nil {
        return err
    }

    for _, filePath := range files {
        var segmentID uint32
        if _, err := fmt.Sscanf(filepath.Base(filePath), WALSegmentPrefix+"%d.log", &segmentID); err != nil {
            continue
        }

        file, err := os.OpenFile(filePath, os.O_RDWR, 0644)
        if err != nil {
            continue
        }

        stat, _ := file.Stat()
        segment := &WALSegment{
            ID:       segmentID,
            File:     file,
            Writer:   bufio.NewWriterSize(file, 64*1024),
            Path:     filePath,
            Size:     stat.Size(),
            StartLSN: uint64(segmentID) * WALSegmentSize / 100,
        }
        wm.segments[segmentID] = segment

        if segmentID > wm.currentSegmentID {
            wm.currentSegmentID = segmentID
            wm.currentSegment = segment
        }
    }
    return nil
}

// rotateSegment создаёт новый сегмент WAL
func (wm *SegmentedWALManager) rotateSegment() error {
    wm.mu.Lock()
    defer wm.mu.Unlock()

    newSegmentID := wm.currentSegmentID + 1
    segmentPath := filepath.Join(wm.segmentsDir, fmt.Sprintf(WALSegmentPrefix+"%d.log", newSegmentID))

    file, err := os.OpenFile(segmentPath, os.O_CREATE|os.O_APPEND|os.O_RDWR, 0644)
    if err != nil {
        return fmt.Errorf("failed to create segment: %v", err)
    }

    newSegment := &WALSegment{
        ID:       newSegmentID,
        File:     file,
        Writer:   bufio.NewWriterSize(file, 64*1024),
        Path:     segmentPath,
        StartLSN: wm.getCurrentLSN(),
    }

    if wm.currentSegment != nil {
        wm.currentSegment.Writer.Flush()
        RealFsync(wm.currentSegment.File)
        wm.currentSegment.File.Close()
    }

    wm.currentSegment = newSegment
    wm.currentSegmentID = newSegmentID
    wm.segments[newSegmentID] = newSegment

    if wm.logger != nil {
        wm.logger.Info(fmt.Sprintf("Created new WAL segment: %d", newSegmentID))
    }

    return nil
}

// getCurrentLSN возвращает текущий LSN
func (wm *SegmentedWALManager) getCurrentLSN() uint64 {
    wm.mu.RLock()
    defer wm.mu.RUnlock()
    if wm.currentSegment == nil {
        return 1
    }
    return wm.currentSegment.StartLSN + uint64(wm.currentSegment.Size/100)
}

// Write записывает запись в WAL
func (wm *SegmentedWALManager) Write(record *WALRecord) error {
    record.Timestamp = time.Now().UnixMilli()
    wm.writeChan <- record
    return nil
}

// writerLoop асинхронно записывает записи
func (wm *SegmentedWALManager) writerLoop() {
    defer wm.wg.Done()

    batch := make([]*WALRecord, 0, wm.batchSize)
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case record, ok := <-wm.writeChan:
            if !ok {
                wm.flushBatch(batch)
                return
            }
            batch = append(batch, record)
            if len(batch) >= wm.batchSize {
                wm.flushBatch(batch)
                batch = batch[:0]
            }

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

        case <-wm.stopChan:
            wm.flushBatch(batch)
            return
        }
    }
}

// flushBatch записывает пакет записей с реальным fsync
func (wm *SegmentedWALManager) flushBatch(batch []*WALRecord) {
    wm.mu.Lock()
    defer wm.mu.Unlock()

    for _, record := range batch {
        if wm.currentSegment.Size >= WALSegmentSize {
            wm.mu.Unlock()
            wm.rotateSegment()
            wm.mu.Lock()
        }

        data, err := json.Marshal(record)
        if err != nil {
            continue
        }

        lsnBytes := make([]byte, 8)
        binary.BigEndian.PutUint64(lsnBytes, record.LSN)
        crcData := append(lsnBytes, data...)
        record.CRC = crc32(crcData)

        lenBuf := make([]byte, 4)
        binary.BigEndian.PutUint32(lenBuf, uint32(len(data)))
        if _, err := wm.currentSegment.Writer.Write(lenBuf); err != nil {
            continue
        }

        if _, err := wm.currentSegment.Writer.Write(data); err != nil {
            continue
        }

        wm.index.addEntry(&WALIndexEntry{
            LSN:       record.LSN,
            SegmentID: wm.currentSegment.ID,
            Offset:    wm.currentSegment.Size,
            Length:    uint32(len(data)),
            Checksum:  record.CRC,
        })

        wm.currentSegment.Size += int64(4 + len(data))
        wm.currentSegment.EndLSN = record.LSN
    }

    wm.currentSegment.Writer.Flush()
    // Реальный fsync после записи пакета
    RealFsync(wm.currentSegment.File)
    wm.index.save()
}

// ReadAll читает все записи из всех сегментов WAL
func (wm *SegmentedWALManager) ReadAll() ([]*WALRecord, error) {
    wm.mu.RLock()
    segments := make([]*WALSegment, 0, len(wm.segments))
    for _, seg := range wm.segments {
        segments = append(segments, seg)
    }
    wm.mu.RUnlock()

    sort.Slice(segments, func(i, j int) bool {
        return segments[i].ID < segments[j].ID
    })

    records := make([]*WALRecord, 0)

    for _, seg := range segments {
        segRecords, err := wm.readSegmentRecords(seg)
        if err != nil {
            return nil, err
        }
        records = append(records, segRecords...)
    }

    return records, nil
}

// readSegmentRecords читает все записи из одного сегмента
func (wm *SegmentedWALManager) readSegmentRecords(seg *WALSegment) ([]*WALRecord, error) {
    seg.mu.Lock()
    defer seg.mu.Unlock()

    if seg.File == nil {
        return nil, nil
    }

    seg.Writer.Flush()
    seg.File.Seek(0, 0)

    records := make([]*WALRecord, 0)
    reader := bufio.NewReader(seg.File)
    lenBuf := make([]byte, 4)

    for {
        _, err := reader.Read(lenBuf)
        if err != nil {
            break
        }

        recordLen := binary.BigEndian.Uint32(lenBuf)
        recordData := make([]byte, recordLen)
        _, err = reader.Read(recordData)
        if err != nil {
            break
        }

        var record WALRecord
        if err := json.Unmarshal(recordData, &record); err != nil {
            continue
        }

        lsnBytes := make([]byte, 8)
        binary.BigEndian.PutUint64(lsnBytes, record.LSN)
        crcData := append(lsnBytes, recordData...)
        if crc32(crcData) != record.CRC {
            continue
        }

        records = append(records, &record)
    }

    return records, nil
}

// RecoverParallel восстанавливает транзакции параллельно
func (wm *SegmentedWALManager) RecoverParallel(callback func(*WALRecord) error) error {
    wm.mu.RLock()
    segments := make([]*WALSegment, 0, len(wm.segments))
    for _, seg := range wm.segments {
        segments = append(segments, seg)
    }
    wm.mu.RUnlock()

    sort.Slice(segments, func(i, j int) bool {
        return segments[i].ID < segments[j].ID
    })

    const numWorkers = 4
    recordChan := make(chan *WALRecord, 1000)
    errChan := make(chan error, numWorkers)
    var wg sync.WaitGroup

    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for record := range recordChan {
                if err := callback(record); err != nil {
                    select {
                    case errChan <- err:
                    default:
                    }
                }
            }
        }()
    }

    go func() {
        defer close(recordChan)
        for _, seg := range segments {
            if err := wm.readSegment(seg, recordChan); err != nil {
                select {
                case errChan <- err:
                default:
                }
                return
            }
        }
    }()

    wg.Wait()
    close(errChan)

    if len(errChan) > 0 {
        return <-errChan
    }
    return nil
}

// readSegment читает сегмент и отправляет записи в канал
func (wm *SegmentedWALManager) readSegment(seg *WALSegment, recordChan chan<- *WALRecord) error {
    seg.mu.Lock()
    defer seg.mu.Unlock()

    if seg.File == nil {
        return nil
    }

    seg.Writer.Flush()
    seg.File.Seek(0, 0)

    reader := bufio.NewReader(seg.File)
    lenBuf := make([]byte, 4)

    for {
        _, err := reader.Read(lenBuf)
        if err != nil {
            break
        }

        recordLen := binary.BigEndian.Uint32(lenBuf)
        recordData := make([]byte, recordLen)
        _, err = reader.Read(recordData)
        if err != nil {
            break
        }

        var record WALRecord
        if err := json.Unmarshal(recordData, &record); err != nil {
            continue
        }

        lsnBytes := make([]byte, 8)
        binary.BigEndian.PutUint64(lsnBytes, record.LSN)
        crcData := append(lsnBytes, recordData...)
        if crc32(crcData) != record.CRC {
            continue
        }

        select {
        case recordChan <- &record:
        default:
        }
    }

    return nil
}

// Close закрывает SegmentedWALManager
func (wm *SegmentedWALManager) Close() error {
    close(wm.stopChan)
    close(wm.writeChan)
    wm.wg.Wait()

    wm.mu.Lock()
    defer wm.mu.Unlock()

    if wm.currentSegment != nil {
        wm.currentSegment.Writer.Flush()
        RealFsyncWithRetry(wm.currentSegment.File, FsyncMaxRetries, FsyncRetryDelay)
        wm.currentSegment.File.Close()
    }

    wm.index.save()
    return nil
}

// addEntry добавляет запись в индекс
func (im *WALIndexManager) addEntry(entry *WALIndexEntry) {
    im.mu.Lock()
    defer im.mu.Unlock()
    im.index[entry.LSN] = entry
}

// FindByLSN ищет запись по LSN
func (im *WALIndexManager) FindByLSN(lsn uint64) *WALIndexEntry {
    im.mu.RLock()
    defer im.mu.RUnlock()
    return im.index[lsn]
}

// save сохраняет индекс на диск
func (im *WALIndexManager) save() error {
    im.mu.RLock()
    defer im.mu.RUnlock()

    data, err := json.Marshal(im.index)
    if err != nil {
        return err
    }
    return os.WriteFile(im.indexPath, data, 0644)
}

// load загружает индекс с диска
func (im *WALIndexManager) load() error {
    data, err := os.ReadFile(im.indexPath)
    if err != nil {
        if os.IsNotExist(err) {
            return nil
        }
        return err
    }
    return json.Unmarshal(data, &im.index)
}

// ========== Async Recovery Manager ==========

// AsyncRecoveryManager управляет асинхронным восстановлением из WAL
type AsyncRecoveryManager struct {
    recordChan   chan *WALRecord
    errChan      chan error
    doneChan     chan struct{}
    wg           sync.WaitGroup
    callback     func(*WALRecord) error
    mu           sync.RWMutex
    isRunning    bool
    recoveredCnt atomic.Uint64
    errorCnt     atomic.Uint64
    startTime    time.Time
}

// NewAsyncRecoveryManager создаёт менеджер асинхронного восстановления
func NewAsyncRecoveryManager(callback func(*WALRecord) error, workers int) *AsyncRecoveryManager {
    arm := &AsyncRecoveryManager{
        recordChan: make(chan *WALRecord, AsyncRecoveryBufferSize),
        errChan:    make(chan error, workers),
        doneChan:   make(chan struct{}),
        callback:   callback,
        startTime:  time.Now(),
        isRunning:  true,
    }

    for i := 0; i < workers; i++ {
        arm.wg.Add(1)
        go arm.worker()
    }

    go arm.errorMonitor()

    return arm
}

// worker обрабатывает записи из канала
func (arm *AsyncRecoveryManager) worker() {
    defer arm.wg.Done()

    for record := range arm.recordChan {
        if err := arm.callback(record); err != nil {
            select {
            case arm.errChan <- err:
            default:
            }
            arm.errorCnt.Add(1)
        } else {
            arm.recoveredCnt.Add(1)
        }
    }
}

// errorMonitor отслеживает ошибки и может остановить процесс
func (arm *AsyncRecoveryManager) errorMonitor() {
    criticalErrors := 0
    for range arm.errChan {
        criticalErrors++
        if criticalErrors > 10 {
            arm.Stop()
            return
        }
    }
}

// Push добавляет запись для восстановления
func (arm *AsyncRecoveryManager) Push(record *WALRecord) bool {
    arm.mu.RLock()
    if !arm.isRunning {
        arm.mu.RUnlock()
        return false
    }
    arm.mu.RUnlock()

    select {
    case arm.recordChan <- record:
        return true
    case <-time.After(100 * time.Millisecond):
        return false
    }
}

// Wait ожидает завершения всех воркеров
func (arm *AsyncRecoveryManager) Wait() {
    close(arm.recordChan)
    arm.wg.Wait()
    close(arm.doneChan)
}

// WaitWithTimeout ожидает завершения с таймаутом
func (arm *AsyncRecoveryManager) WaitWithTimeout(timeout time.Duration) bool {
    done := make(chan struct{})
    go func() {
        arm.Wait()
        close(done)
    }()

    select {
    case <-done:
        return true
    case <-time.After(timeout):
        return false
    }
}

// Stop останавливает восстановление
func (arm *AsyncRecoveryManager) Stop() {
    arm.mu.Lock()
    if !arm.isRunning {
        arm.mu.Unlock()
        return
    }
    arm.isRunning = false
    arm.mu.Unlock()

    close(arm.recordChan)
}

// GetStats возвращает статистику восстановления
func (arm *AsyncRecoveryManager) GetStats() map[string]interface{} {
    return map[string]interface{}{
        "recovered":   arm.recoveredCnt.Load(),
        "errors":      arm.errorCnt.Load(),
        "is_running":  arm.isRunning,
        "elapsed_ms":  time.Since(arm.startTime).Milliseconds(),
    }
}

// IsComplete возвращает статус завершения
func (arm *AsyncRecoveryManager) IsComplete() bool {
    return !arm.isRunning && arm.recoveredCnt.Load() > 0
}

// ========== MVCC Improvements ==========

// VisibilityMapEntry представляет запись в карте видимости
type VisibilityMapEntry struct {
    DocID       string
    VisibleFrom uint64
    VisibleTo   uint64
    IsVisible   bool
    LastAccess  int64
}

// VisibilityMap управляет битовой картой видимости версий
type VisibilityMap struct {
    entries   sync.Map
    maxSize   int
    hitCount  atomic.Uint64
    missCount atomic.Uint64
}

// NewVisibilityMap создаёт новую карту видимости
func NewVisibilityMap(maxSize int) *VisibilityMap {
    if maxSize <= 0 {
        maxSize = VisibilityMapSize
    }
    return &VisibilityMap{maxSize: maxSize}
}

// MarkVisible отмечает версию как видимую
func (vm *VisibilityMap) MarkVisible(docID string, version uint64, visible bool) {
    key := fmt.Sprintf("%s@%d", docID, version)
    vm.entries.Store(key, &VisibilityMapEntry{
        DocID:       docID,
        VisibleFrom: version,
        VisibleTo:   version,
        IsVisible:   visible,
        LastAccess:  time.Now().Unix(),
    })
}

// IsVisible проверяет видимость версии
func (vm *VisibilityMap) IsVisible(docID string, version uint64) bool {
    key := fmt.Sprintf("%s@%d", docID, version)
    _, ok := vm.entries.Load(key)
    if !ok {
        vm.missCount.Add(1)
        return false
    }
    vm.hitCount.Add(1)
    return true
}

// GetStats возвращает статистику
func (vm *VisibilityMap) GetStats() map[string]interface{} {
    return map[string]interface{}{
        "hits":   vm.hitCount.Load(),
        "misses": vm.missCount.Load(),
    }
}

// VersionPrunePolicy определяет политику очистки версий
type VersionPrunePolicy struct {
    MaxVersions   int
    RetentionDays int
    MinVersionAge time.Duration
    PruneInterval time.Duration
}

// VersionPruner управляет очисткой старых версий
type VersionPruner struct {
    policy   *VersionPrunePolicy
    txMgr    *TransactionManager
    logger   LoggerInterface
    stopChan chan struct{}
    wg       sync.WaitGroup
}

// NewVersionPruner создаёт новый очиститель версий
func NewVersionPruner(policy *VersionPrunePolicy, txMgr *TransactionManager, logger LoggerInterface) *VersionPruner {
    if policy == nil {
        policy = &VersionPrunePolicy{
            MaxVersions:   MaxVersionsPerDoc,
            RetentionDays: VersionRetentionDays,
            MinVersionAge: time.Hour,
            PruneInterval: VersionPruneInterval,
        }
    }
    return &VersionPruner{
        policy:   policy,
        txMgr:    txMgr,
        logger:   logger,
        stopChan: make(chan struct{}),
    }
}

// Start запускает фоновую очистку
func (vp *VersionPruner) Start() {
    vp.wg.Add(1)
    go vp.pruneLoop()
}

// Stop останавливает очистку
func (vp *VersionPruner) Stop() {
    close(vp.stopChan)
    vp.wg.Wait()
}

// pruneLoop периодически очищает старые версии
func (vp *VersionPruner) pruneLoop() {
    defer vp.wg.Done()
    ticker := time.NewTicker(vp.policy.PruneInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            vp.prune()
        case <-vp.stopChan:
            return
        }
    }
}

// prune выполняет очистку старых версий
func (vp *VersionPruner) prune() {
    cutoffTime := time.Now().AddDate(0, 0, -vp.policy.RetentionDays).UnixMilli()

    vp.txMgr.documentVersions.Range(func(key, value interface{}) bool {
        versions := value.([]*DocumentVersion)
        if len(versions) <= vp.policy.MaxVersions {
            return true
        }

        newVersions := make([]*DocumentVersion, 0, vp.policy.MaxVersions)
        for _, v := range versions {
            if v.Timestamp >= cutoffTime && len(newVersions) < vp.policy.MaxVersions {
                newVersions = append(newVersions, v)
            }
        }

        if len(newVersions) < len(versions) {
            vp.txMgr.documentVersions.Store(key, newVersions)
            if vp.logger != nil {
                vp.logger.Debug(fmt.Sprintf("Pruned %d versions for doc %v", len(versions)-len(newVersions), key))
            }
        }
        return true
    })
}

// ReadTimestampCache кэширует версии документов по timestamp
type ReadTimestampCache struct {
    cache   sync.Map
    maxSize int
    ttl     time.Duration
    hits    atomic.Uint64
    misses  atomic.Uint64
}

type cachedEntry struct {
    doc      *Document
    cachedAt time.Time
}

// NewReadTimestampCache создаёт новый кэш
func NewReadTimestampCache(maxSize int, ttl time.Duration) *ReadTimestampCache {
    if maxSize <= 0 {
        maxSize = 10000
    }
    if ttl <= 0 {
        ttl = 5 * time.Minute
    }
    return &ReadTimestampCache{maxSize: maxSize, ttl: ttl}
}

// Get возвращает кэшированную версию документа
func (rtc *ReadTimestampCache) Get(docID string, timestamp int64) *Document {
    key := fmt.Sprintf("%s@%d", docID, timestamp)
    val, ok := rtc.cache.Load(key)
    if !ok {
        rtc.misses.Add(1)
        return nil
    }
    entry := val.(*cachedEntry)
    if time.Since(entry.cachedAt) > rtc.ttl {
        rtc.cache.Delete(key)
        rtc.misses.Add(1)
        return nil
    }
    rtc.hits.Add(1)
    return entry.doc
}

// Set добавляет версию в кэш
func (rtc *ReadTimestampCache) Set(docID string, timestamp int64, doc *Document) {
    key := fmt.Sprintf("%s@%d", docID, timestamp)
    rtc.cache.Store(key, &cachedEntry{doc: doc, cachedAt: time.Now()})
}

// GetStats возвращает статистику кэша
func (rtc *ReadTimestampCache) GetStats() map[string]interface{} {
    return map[string]interface{}{
        "hits":   rtc.hits.Load(),
        "misses": rtc.misses.Load(),
    }
}

// ========== Distributed Transaction Coordinator ==========

// DistributedTxInfo содержит информацию о распределённой транзакции
type DistributedTxInfo struct {
    TxID      TransactionID
    Nodes     []string
    Prepared  map[string]bool
    Committed map[string]bool
    Status    TxState
    StartTime int64
    Timeout   time.Duration
}

// TxState представляет состояние транзакции
type TxState int32

const (
    TxActive TxState = iota
    TxPrepared
    TxCommitted
    TxAborted
    TxTimeout
)

// DistributedTransactionCoordinator координирует распределённые транзакции
type DistributedTransactionCoordinator struct {
    pendingTxs sync.Map
    timeout    time.Duration
    mu         sync.RWMutex
}

// NewDistributedTransactionCoordinator создаёт новый координатор
func NewDistributedTransactionCoordinator(timeout time.Duration) *DistributedTransactionCoordinator {
    if timeout <= 0 {
        timeout = TwoPhaseCommitTimeout
    }
    return &DistributedTransactionCoordinator{timeout: timeout}
}

// Prepare подготавливает распределённую транзакцию
func (dtc *DistributedTransactionCoordinator) Prepare(txID TransactionID, nodes []string) error {
    info := &DistributedTxInfo{
        TxID:      txID,
        Nodes:     nodes,
        Prepared:  make(map[string]bool),
        Committed: make(map[string]bool),
        Status:    TxActive,
        StartTime: time.Now().UnixMilli(),
        Timeout:   dtc.timeout,
    }
    dtc.pendingTxs.Store(txID, info)
    return nil
}

// VoteCommit получает голоса за коммит от узлов
func (dtc *DistributedTransactionCoordinator) VoteCommit(txID TransactionID, node string) error {
    val, ok := dtc.pendingTxs.Load(txID)
    if !ok {
        return fmt.Errorf("transaction not found: %d", txID)
    }
    info := val.(*DistributedTxInfo)

    dtc.mu.Lock()
    info.Prepared[node] = true
    dtc.mu.Unlock()

    if len(info.Prepared) == len(info.Nodes) {
        return dtc.Commit(txID)
    }
    return nil
}

// Commit коммитит распределённую транзакцию
func (dtc *DistributedTransactionCoordinator) Commit(txID TransactionID) error {
    val, ok := dtc.pendingTxs.Load(txID)
    if !ok {
        return fmt.Errorf("transaction not found: %d", txID)
    }
    info := val.(*DistributedTxInfo)
    info.Status = TxCommitted
    return nil
}

// Abort откатывает распределённую транзакцию
func (dtc *DistributedTransactionCoordinator) Abort(txID TransactionID) error {
    dtc.pendingTxs.Delete(txID)
    return nil
}

// ========== Deadlock Detector ==========

// DeadlockDetector обнаруживает циклические зависимости между транзакциями
type DeadlockDetector struct {
    waitForGraph  sync.Map
    checkInterval time.Duration
    timeout       time.Duration
    mu            sync.RWMutex
    stopChan      chan struct{}
    wg            sync.WaitGroup
    logger        LoggerInterface
}

// NewDeadlockDetector создаёт новый детектор дедлоков
func NewDeadlockDetector(checkInterval, timeout time.Duration) *DeadlockDetector {
    if checkInterval <= 0 {
        checkInterval = DeadlockCheckInterval
    }
    if timeout <= 0 {
        timeout = DefaultTxTimeout
    }
    d := &DeadlockDetector{
        checkInterval: checkInterval,
        timeout:       timeout,
        stopChan:      make(chan struct{}),
    }
    d.wg.Add(1)
    go d.detectLoop()
    return d
}

// SetLogger устанавливает логгер для детектора дедлоков
func (dd *DeadlockDetector) SetLogger(logger LoggerInterface) {
    dd.logger = logger
}

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

// AddWaiting добавляет ожидающую транзакцию
func (dd *DeadlockDetector) AddWaiting(waiting, waitingFor TransactionID) {
    var list []TransactionID
    if val, ok := dd.waitForGraph.Load(waiting); ok {
        list = val.([]TransactionID)
    }
    list = append(list, waitingFor)
    dd.waitForGraph.Store(waiting, list)
}

// RemoveWaiting удаляет ожидающую транзакцию
func (dd *DeadlockDetector) RemoveWaiting(txID TransactionID) {
    dd.waitForGraph.Delete(txID)
}

// detectLoop обнаруживает циклы в графе ожидания
func (dd *DeadlockDetector) detectLoop() {
    defer dd.wg.Done()
    ticker := time.NewTicker(dd.checkInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            dd.detect()
        case <-dd.stopChan:
            return
        }
    }
}

// detect выполняет обнаружение дедлоков
func (dd *DeadlockDetector) detect() {
    visited := make(map[TransactionID]bool)
    stack := make(map[TransactionID]bool)

    var dfs func(txID TransactionID) bool
    dfs = func(txID TransactionID) bool {
        visited[txID] = true
        stack[txID] = true

        val, ok := dd.waitForGraph.Load(txID)
        if ok {
            for _, next := range val.([]TransactionID) {
                if !visited[next] {
                    if dfs(next) {
                        return true
                    }
                } else if stack[next] {
                    dd.resolveDeadlock(txID, next)
                    return true
                }
            }
        }
        stack[txID] = false
        return false
    }

    dd.waitForGraph.Range(func(key, value interface{}) bool {
        txID := key.(TransactionID)
        if !visited[txID] {
            dfs(txID)
        }
        return true
    })
}

// resolveDeadlock разрешает дедлок, откатывая одну из транзакций
func (dd *DeadlockDetector) resolveDeadlock(txID1, txID2 TransactionID) {
    if globalTxManager != nil {
        if val, ok := globalTxManager.activeTransactions.Load(txID1); ok {
            tx := val.(*Transaction)
            tx.State.Store(int32(TransactionAborted))
            globalTxManager.activeTransactions.Delete(txID1)
            if dd.logger != nil {
                dd.logger.Warn(fmt.Sprintf("Deadlock resolved: aborted transaction %d due to conflict with %d", txID1, txID2))
            }
        }
    }
}

// ========== Transaction With Timeout and Savepoints ==========

// Savepoint представляет точку сохранения в транзакции
type Savepoint struct {
    ID         string
    Timestamp  int64
    Operations []Operation
    Version    uint64
}

// TransactionWithTimeout расширяет Transaction с поддержкой таймаута и savepoints
type TransactionWithTimeout struct {
    *Transaction
    Timeout    time.Duration
    StartTime  time.Time
    Timer      *time.Timer
    Savepoints []*Savepoint
    mu         sync.RWMutex
}

// NewTransactionWithTimeout создаёт транзакцию с таймаутом
func NewTransactionWithTimeout(timeout time.Duration) *TransactionWithTimeout {
    tx := BeginTransaction()
    if tx == nil {
        return nil
    }

    txw := &TransactionWithTimeout{
        Transaction: tx,
        Timeout:     timeout,
        StartTime:   time.Now(),
        Savepoints:  make([]*Savepoint, 0, MaxSavepointsPerTx),
    }

    txw.Timer = time.AfterFunc(timeout, func() {
        txw.State.Store(int32(TransactionAborted))
        if globalTxManager != nil {
            globalTxManager.activeTransactions.Delete(txw.ID)
        }
        AuditLog("TIMEOUT", "TRANSACTION", fmt.Sprintf("%d", txw.ID), map[string]interface{}{
            "timeout_ms": timeout.Milliseconds(),
        })
    })

    return txw
}

// CreateSavepoint создаёт точку сохранения
func (txw *TransactionWithTimeout) CreateSavepoint(id string) error {
    txw.mu.Lock()
    defer txw.mu.Unlock()

    if len(txw.Savepoints) >= MaxSavepointsPerTx {
        return fmt.Errorf("too many savepoints, max %d", MaxSavepointsPerTx)
    }

    opsCopy := make([]Operation, len(txw.Operations))
    copy(opsCopy, txw.Operations)

    savepoint := &Savepoint{
        ID:         id,
        Timestamp:  time.Now().UnixMilli(),
        Operations: opsCopy,
        Version:    txw.Version,
    }
    txw.Savepoints = append(txw.Savepoints, savepoint)

    AuditLog("SAVEPOINT", "TRANSACTION", fmt.Sprintf("%d.%s", txw.ID, id), map[string]interface{}{
        "operation_count": len(opsCopy),
    })

    return nil
}

// RollbackToSavepoint откатывает до точки сохранения
func (txw *TransactionWithTimeout) RollbackToSavepoint(id string) error {
    txw.mu.Lock()
    defer txw.mu.Unlock()

    var targetSavepoint *Savepoint
    var savepointIndex int

    for i, sp := range txw.Savepoints {
        if sp.ID == id {
            targetSavepoint = sp
            savepointIndex = i
            break
        }
    }

    if targetSavepoint == nil {
        return fmt.Errorf("savepoint not found: %s", id)
    }

    txw.Operations = make([]Operation, len(targetSavepoint.Operations))
    copy(txw.Operations, targetSavepoint.Operations)

    txw.Savepoints = txw.Savepoints[:savepointIndex+1]
    txw.Version = targetSavepoint.Version

    AuditLog("ROLLBACK_SAVEPOINT", "TRANSACTION", fmt.Sprintf("%d.%s", txw.ID, id), map[string]interface{}{
        "rolled_back_to": targetSavepoint.Timestamp,
    })

    return nil
}

// ReleaseSavepoint освобождает точку сохранения
func (txw *TransactionWithTimeout) ReleaseSavepoint(id string) error {
    txw.mu.Lock()
    defer txw.mu.Unlock()

    for i, sp := range txw.Savepoints {
        if sp.ID == id {
            txw.Savepoints = append(txw.Savepoints[:i], txw.Savepoints[i+1:]...)
            return nil
        }
    }
    return fmt.Errorf("savepoint not found: %s", id)
}

// ========== Transaction (основная структура) ==========

// Transaction представляет одну транзакцию
type Transaction struct {
    ID         TransactionID
    State      atomic.Int32
    Operations []Operation
    StartTime  int64
    Version    uint64
    mu         sync.RWMutex
}

// TransactionManager управляет транзакциями с поддержкой MVCC
type TransactionManager struct {
    activeTransactions sync.Map // map[TransactionID]*Transaction
    nextTxID           atomic.Uint64
    wal                *SegmentedWALManager
    logger             LoggerInterface
    mu                 sync.RWMutex
    walPath            string
    checkpointInterval int64
    lastCheckpoint     int64
    checkpointFile     *os.File
    documentVersions   sync.Map // map[string][]*DocumentVersion
    maxVersions        int
    visibilityMap      *VisibilityMap
    readCache          *ReadTimestampCache
    distCoord          *DistributedTransactionCoordinator
    deadlockDetector   *DeadlockDetector
    recoveryManager    *AsyncRecoveryManager
    recoveryComplete   atomic.Bool
}

var (
    globalTxManager *TransactionManager
    txManagerOnce   sync.Once
    currentTx       atomic.Value // *Transaction
    globalStorage   *Storage
)

// InitTransactionManager инициализирует глобальный менеджер транзакций
func InitTransactionManager(walPath string) error {
    return InitTransactionManagerWithConfig(walPath, nil)
}

// InitTransactionManagerWithConfig инициализирует менеджер транзакций с конфигурацией
func InitTransactionManagerWithConfig(walPath string, config map[string]interface{}) error {
    var err error
    txManagerOnce.Do(func() {
        maxVersions := 10
        if config != nil {
            if v, ok := config["max_versions"].(int); ok && v > 0 {
                maxVersions = v
            }
        }

        globalTxManager = &TransactionManager{
            nextTxID:           atomic.Uint64{},
            walPath:            walPath,
            checkpointInterval: 300,
            lastCheckpoint:     time.Now().Unix(),
            maxVersions:        maxVersions,
            visibilityMap:      NewVisibilityMap(VisibilityMapSize),
            readCache:          NewReadTimestampCache(10000, 5*time.Minute),
            distCoord:          NewDistributedTransactionCoordinator(TwoPhaseCommitTimeout),
            deadlockDetector:   NewDeadlockDetector(DeadlockCheckInterval, DefaultTxTimeout),
        }
        globalTxManager.nextTxID.Store(1)

        var walErr error
        globalTxManager.wal, walErr = NewSegmentedWALManager(filepath.Dir(walPath), nil)
        if walErr != nil {
            err = walErr
            return
        }

        globalTxManager.startAsyncRecovery()

        go globalTxManager.checkpointLoop()
        go globalTxManager.versionCleanupLoop()
    })
    return err
}

// startAsyncRecovery запускает асинхронное восстановление из WAL
func (tm *TransactionManager) startAsyncRecovery() {
    if tm.wal == nil {
        tm.recoveryComplete.Store(true)
        return
    }

    if tm.logger != nil {
        tm.logger.Info("Starting asynchronous WAL recovery...")
    }

    records, err := tm.wal.ReadAll()
    if err != nil {
        if tm.logger != nil {
            tm.logger.Error(fmt.Sprintf("Failed to read WAL: %v", err))
        }
        tm.recoveryComplete.Store(true)
        return
    }

    if len(records) == 0 {
        if tm.logger != nil {
            tm.logger.Info("No records to recover")
        }
        tm.recoveryComplete.Store(true)
        return
    }

    tm.recoveryManager = NewAsyncRecoveryManager(func(record *WALRecord) error {
        if record.Type == 1 {
            var txRecord TransactionRecord
            if err := json.Unmarshal(record.Data, &txRecord); err != nil {
                return err
            }

            if txRecord.State == TransactionCommitted {
                for _, op := range txRecord.Operations {
                    if err := applyOperation(op); err != nil {
                        return err
                    }
                }
            }
        }
        return nil
    }, AsyncRecoveryWorkers)

    for _, record := range records {
        if !tm.recoveryManager.Push(record) {
            if tm.logger != nil {
                tm.logger.Warn("Recovery buffer full, some records may be delayed")
            }
        }
    }

    go func() {
        ticker := time.NewTicker(5 * time.Second)
        defer ticker.Stop()

        for {
            select {
            case <-ticker.C:
                stats := tm.recoveryManager.GetStats()
                if tm.logger != nil {
                    tm.logger.Debug(fmt.Sprintf("Recovery progress: %d/%d records recovered",
                        stats["recovered"], len(records)))
                }
            case <-tm.recoveryManager.doneChan:
                stats := tm.recoveryManager.GetStats()
                if tm.logger != nil {
                    tm.logger.Info(fmt.Sprintf("WAL recovery completed: %d records recovered, %d errors",
                        stats["recovered"], stats["errors"]))
                }
                tm.recoveryComplete.Store(true)
                return
            }
        }
    }()
}

// IsRecoveryComplete возвращает статус восстановления
func (tm *TransactionManager) IsRecoveryComplete() bool {
    return tm.recoveryComplete.Load()
}

// GetRecoveryProgress возвращает прогресс восстановления
func (tm *TransactionManager) GetRecoveryProgress() map[string]interface{} {
    if tm.recoveryManager == nil {
        return map[string]interface{}{
            "is_recovering": false,
            "recovered":     0,
            "complete":      true,
        }
    }

    stats := tm.recoveryManager.GetStats()
    return map[string]interface{}{
        "is_recovering": !tm.recoveryComplete.Load(),
        "recovered":     stats["recovered"],
        "complete":      tm.recoveryComplete.Load(),
        "elapsed_ms":    stats["elapsed_ms"],
    }
}

// SetTransactionLogger устанавливает логгер для транзакций
func SetTransactionLogger(logger LoggerInterface) {
    if globalTxManager != nil {
        globalTxManager.logger = logger
        if globalTxManager.wal != nil {
            globalTxManager.wal.logger = logger
        }
        if globalTxManager.deadlockDetector != nil {
            globalTxManager.deadlockDetector.SetLogger(logger)
        }
    }
}

// BeginTransaction начинает новую транзакцию
func BeginTransaction() *Transaction {
    if globalTxManager == nil {
        InitTransactionManager("futriis.wal")
    }

    tx := &Transaction{
        ID:         TransactionID(globalTxManager.nextTxID.Add(1) - 1),
        StartTime:  time.Now().UnixMilli(),
        Operations: make([]Operation, 0, 100),
        Version:    1,
    }
    tx.State.Store(int32(TransactionActive))

    globalTxManager.activeTransactions.Store(tx.ID, tx)
    currentTx.Store(tx)

    AuditLog("START", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{
        "start_time": tx.StartTime,
    })

    return tx
}

// BeginTransactionWithTimeout начинает транзакцию с таймаутом
func BeginTransactionWithTimeout(timeout time.Duration) *TransactionWithTimeout {
    return NewTransactionWithTimeout(timeout)
}

// BeginTransactionOnCollection начинает транзакцию для конкретной коллекции
func BeginTransactionOnCollection(coll *Collection) error {
    if globalTxManager == nil {
        if err := InitTransactionManager("futriis.wal"); err != nil {
            return err
        }
    }

    tx := BeginTransaction()
    if tx == nil {
        return fmt.Errorf("failed to create transaction")
    }

    if globalTxManager.logger != nil {
        globalTxManager.logger.Debug(fmt.Sprintf("Transaction %d started on collection %s.%s", tx.ID, coll.dbName, coll.name))
    }

    return nil
}

// CommitCurrentTransaction коммитит текущую транзакцию
func CommitCurrentTransaction() error {
    txVal := currentTx.Load()
    if txVal == nil {
        return fmt.Errorf("no active transaction")
    }

    tx := txVal.(*Transaction)
    if TransactionState(tx.State.Load()) != TransactionActive {
        return fmt.Errorf("transaction is not active")
    }

    for _, op := range tx.Operations {
        if err := applyOperation(op); err != nil {
            AbortCurrentTransaction()
            return fmt.Errorf("transaction commit failed at operation %s: %v", op.Type, err)
        }
        if globalTxManager != nil && op.DocumentID != "" {
            if globalStorage != nil {
                db, _ := globalStorage.GetDatabase(op.Database)
                if db != nil {
                    coll, _ := db.GetCollection(op.Collection)
                    if coll != nil {
                        if doc, err := coll.Find(op.DocumentID); err == nil {
                            globalTxManager.AddDocumentVersion(op.DocumentID, &DocumentVersion{
                                Document:  doc.Clone(),
                                Timestamp: time.Now().UnixMilli(),
                                TxID:      tx.ID,
                            })
                        }
                    }
                }
            }
        }
    }

    tx.State.Store(int32(TransactionCommitted))

    AuditLog("COMMIT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{
        "operations": len(tx.Operations),
    })

    currentTx.Store(nil)
    globalTxManager.activeTransactions.Delete(tx.ID)

    return nil
}

// AbortCurrentTransaction откатывает текущую транзакцию
func AbortCurrentTransaction() error {
    txVal := currentTx.Load()
    if txVal == nil {
        return fmt.Errorf("no active transaction")
    }

    tx := txVal.(*Transaction)
    tx.State.Store(int32(TransactionAborted))

    AuditLog("ABORT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{
        "operations": len(tx.Operations),
    })

    currentTx.Store(nil)
    globalTxManager.activeTransactions.Delete(tx.ID)

    return nil
}

// HasActiveTransaction проверяет наличие активной транзакции
func HasActiveTransaction() bool {
    return currentTx.Load() != nil
}

// GetCurrentTransactionID возвращает ID текущей транзакции
func GetCurrentTransactionID() string {
    txVal := currentTx.Load()
    if txVal == nil {
        return ""
    }
    tx := txVal.(*Transaction)
    return fmt.Sprintf("%d", tx.ID)
}

// GetActiveTransactions возвращает список активных транзакций для API
func GetActiveTransactions() []TransactionInfo {
    if globalTxManager == nil {
        return []TransactionInfo{}
    }

    transactions := make([]TransactionInfo, 0)

    globalTxManager.activeTransactions.Range(func(key, value interface{}) bool {
        tx := value.(*Transaction)
        status := "active"
        if TransactionState(tx.State.Load()) == TransactionCommitted {
            status = "committed"
        } else if TransactionState(tx.State.Load()) == TransactionAborted {
            status = "aborted"
        }

        tx.mu.RLock()
        opCount := len(tx.Operations)
        operations := make([]OperationInfo, 0, opCount)
        for _, op := range tx.Operations {
            operations = append(operations, OperationInfo{
                Type:       op.Type,
                Database:   op.Database,
                Collection: op.Collection,
                DocumentID: op.DocumentID,
            })
        }
        tx.mu.RUnlock()

        transactions = append(transactions, TransactionInfo{
            ID:             fmt.Sprintf("%d", tx.ID),
            Status:         status,
            StartTime:      tx.StartTime,
            OperationCount: opCount,
            Operations:     operations,
        })
        return true
    })

    return transactions
}

// GetTransactionByID возвращает транзакцию по ID
func GetTransactionByID(id string) (*Transaction, error) {
    if globalTxManager == nil {
        return nil, fmt.Errorf("transaction manager not initialized")
    }

    var txID TransactionID
    fmt.Sscanf(id, "%d", &txID)

    if val, ok := globalTxManager.activeTransactions.Load(txID); ok {
        return val.(*Transaction), nil
    }

    return nil, fmt.Errorf("transaction not found")
}

// AddToTransaction добавляет операцию в текущую транзакцию
func AddToTransaction(coll *Collection, opType string, docID string, newData map[string]interface{}, oldData map[string]interface{}, version uint64) error {
    txVal := currentTx.Load()
    if txVal == nil {
        return fmt.Errorf("no active transaction")
    }

    tx := txVal.(*Transaction)
    if TransactionState(tx.State.Load()) != TransactionActive {
        return fmt.Errorf("transaction is not active")
    }

    op := Operation{
        Type:       opType,
        Database:   coll.dbName,
        Collection: coll.name,
        DocumentID: docID,
        Data:       newData,
        OldData:    oldData,
        Version:    version,
    }

    tx.mu.Lock()
    tx.Operations = append(tx.Operations, op)
    tx.mu.Unlock()

    return nil
}

// FindInTransaction ищет документ в контексте транзакции с поддержкой MVCC
func FindInTransaction(coll *Collection, id string) (*Document, error) {
    txVal := currentTx.Load()
    if txVal == nil {
        return coll.Find(id)
    }

    tx := txVal.(*Transaction)

    tx.mu.RLock()
    defer tx.mu.RUnlock()

    for i := len(tx.Operations) - 1; i >= 0; i-- {
        op := tx.Operations[i]
        if op.DocumentID == id {
            if op.Type == "delete" {
                return nil, fmt.Errorf("document deleted in transaction")
            }
            if op.Type == "insert" || op.Type == "update" {
                doc := NewDocumentWithID(op.DocumentID)
                for k, v := range op.Data {
                    doc.SetField(k, v)
                }
                doc.Version = op.Version
                return doc, nil
            }
        }
    }

    if globalTxManager != nil {
        if versionDoc := globalTxManager.GetDocumentVersion(id, tx.StartTime); versionDoc != nil {
            return versionDoc, nil
        }
    }

    return coll.Find(id)
}

// applyOperation применяет операцию к хранилищу
func applyOperation(op Operation) error {
    if globalStorage == nil {
        return fmt.Errorf("storage not initialized")
    }

    db, err := globalStorage.GetDatabase(op.Database)
    if err != nil {
        return fmt.Errorf("database not found: %s", op.Database)
    }

    coll, err := db.GetCollection(op.Collection)
    if err != nil {
        return fmt.Errorf("collection not found: %s", op.Collection)
    }

    switch op.Type {
    case "insert":
        doc := NewDocumentWithID(op.DocumentID)
        for k, v := range op.Data {
            doc.SetField(k, v)
        }
        doc.Version = op.Version
        return coll.Insert(doc)

    case "update":
        return coll.Update(op.DocumentID, op.Data)

    case "delete":
        return coll.Delete(op.DocumentID)
    }

    return nil
}

// SetGlobalStorage устанавливает глобальное хранилище
func SetGlobalStorage(s *Storage) {
    globalStorage = s
}

// GetGlobalStorage возвращает глобальное хранилище
func GetGlobalStorage() *Storage {
    return globalStorage
}

// AuditLog записывает событие в аудит
func AuditLog(operation, dataType, name string, details map[string]interface{}) {
    LogAudit(operation, dataType, name, details)
}

// AddDocumentVersion добавляет версию документа для MVCC
func (tm *TransactionManager) AddDocumentVersion(docID string, version *DocumentVersion) {
    val, _ := tm.documentVersions.LoadOrStore(docID, make([]*DocumentVersion, 0))
    versions := val.([]*DocumentVersion)
    versions = append(versions, version)
    if len(versions) > tm.maxVersions && tm.maxVersions > 0 {
        versions = versions[len(versions)-tm.maxVersions:]
    }
    tm.documentVersions.Store(docID, versions)

    if tm.visibilityMap != nil {
        tm.visibilityMap.MarkVisible(docID, uint64(version.TxID), true)
    }
}

// GetDocumentVersion возвращает версию документа на определённый момент времени
func (tm *TransactionManager) GetDocumentVersion(docID string, timestamp int64) *Document {
    if tm.readCache != nil {
        if cached := tm.readCache.Get(docID, timestamp); cached != nil {
            return cached
        }
    }

    val, ok := tm.documentVersions.Load(docID)
    if !ok {
        return nil
    }

    versions := val.([]*DocumentVersion)
    for i := len(versions) - 1; i >= 0; i-- {
        if versions[i].Timestamp <= timestamp {
            doc := versions[i].Document.Clone()
            if tm.readCache != nil {
                tm.readCache.Set(docID, timestamp, doc)
            }
            return doc
        }
    }
    return nil
}

// checkpointLoop периодически создаёт чекпоинты
func (tm *TransactionManager) checkpointLoop() {
    ticker := time.NewTicker(time.Duration(tm.checkpointInterval) * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        tm.createCheckpoint()
    }
}

// createCheckpoint создаёт чекпоинт текущего состояния
func (tm *TransactionManager) createCheckpoint() {
    if tm.wal == nil {
        return
    }

    now := time.Now().Unix()
    if now-tm.lastCheckpoint < tm.checkpointInterval {
        return
    }

    checkpointPath := fmt.Sprintf("%s.checkpoint.%d", tm.walPath, now)

    checkpoint := make(map[string]interface{})
    checkpoint["timestamp"] = now
    checkpoint["last_lsn"] = tm.wal.currentSegment.EndLSN

    data, err := json.Marshal(checkpoint)
    if err != nil {
        if tm.logger != nil {
            tm.logger.Error(fmt.Sprintf("Failed to marshal checkpoint: %v", err))
        }
        return
    }

    if err := os.WriteFile(checkpointPath, data, 0644); err != nil {
        if tm.logger != nil {
            tm.logger.Error(fmt.Sprintf("Failed to write checkpoint: %v", err))
        }
        return
    }

    tm.lastCheckpoint = now
    if tm.logger != nil {
        tm.logger.Info(fmt.Sprintf("Checkpoint created: %s", checkpointPath))
    }
}

// versionCleanupLoop периодически очищает старые версии документов
func (tm *TransactionManager) versionCleanupLoop() {
    if tm.maxVersions <= 0 {
        return
    }

    pruner := NewVersionPruner(&VersionPrunePolicy{
        MaxVersions:   tm.maxVersions,
        RetentionDays: VersionRetentionDays,
        MinVersionAge: time.Hour,
        PruneInterval: VersionPruneInterval,
    }, tm, tm.logger)
    pruner.Start()
}

// MVCCSnapshot создаёт снапшот текущего состояния для MVCC
func MVCCSnapshot() uint64 {
    return uint64(time.Now().UnixNano())
}

// CreateDocumentVersion создаёт новую версию документа для MVCC
func CreateDocumentVersion(doc *Document, txID TransactionID) *DocumentVersion {
    return &DocumentVersion{
        Document:  doc.Clone(),
        Timestamp: time.Now().UnixMilli(),
        TxID:      txID,
    }
}

// StopTransactionManager останавливает менеджер транзакций
func StopTransactionManager() error {
    if globalTxManager == nil {
        return nil
    }

    if globalTxManager.deadlockDetector != nil {
        globalTxManager.deadlockDetector.Stop()
    }

    if globalTxManager.wal != nil {
        return globalTxManager.wal.Close()
    }

    return nil
}