Update internal/storage/persistence.go

README.md

@@ -54,7 +54,7 @@
     <li><a href="#триггеры">Триггеры</a></li>
     <li><a href="#сжатие-данных">Сжатие данных</a></li>
     <li><a href="#графический-интерфейс">Графический интерфейс</a></li>
-    <li><a href="#дорожная-карта">Дорожная карта</a></li>
+    <li><a href="#план-развития">План развития</a></li>
     <li><a href="#контакты">Контакты</a></li>
   </ol>
 <!-- </details> -->
@@ -1662,7 +1662,7 @@ futriix:~> doc compression employees 550e8400-e29b-41d4-a716-446655440000
 <p align="right">(<a href="#readme-top">К началу</a>)</p>
 
 
-## Дорожная карта
+## План развития
 
 - [x] Реализовать поддержку хранимых процедур
 - [x] Реализовать поддержку триггеров (обратных вызовов)

internal/backup/backup_scheduler.go

@@ -14,10 +14,13 @@
 package backup
 
 import (
+    "bytes"
+    "compress/gzip"
     "crypto/sha256"
     "encoding/hex"
     "encoding/json"
     "fmt"
+    "io"
     "os"
     "path/filepath"
     "sort"
@@ -65,18 +68,13 @@ type BackupInfo struct {
     Error      string     `json:"error,omitempty"`
 }
 
-// BackupScheduler планировщик бэкапов
-type BackupScheduler struct {
-    schedules      map[string]*BackupSchedule
-    backups        map[string]*BackupInfo
-    walReader      WALReader
-    storage        BackupStorage
-    logger         LoggerInterface
-    mu             sync.RWMutex
-    stopChan       chan struct{}
-    wg             sync.WaitGroup
-    running        atomic.Bool
-    backupCounter  atomic.Uint64
+// WALEntry запись WAL
+type WALEntry struct {
+    Index     uint64 `json:"index"`
+    Term      uint64 `json:"term"`
+    Type      string `json:"type"`
+    Data      []byte `json:"data"`
+    Timestamp int64  `json:"timestamp"`
 }
 
 // WALReader интерфейс для чтения WAL
@@ -85,15 +83,8 @@ type WALReader interface {
     GetCurrentIndex() (uint64, error)
     GetLastBackupIndex() uint64
     SetLastBackupIndex(index uint64) error
-}
-
-// WALEntry запись WAL
-type WALEntry struct {
-    Index     uint64      `json:"index"`
-    Term      uint64      `json:"term"`
-    Type      string      `json:"type"`
-    Data      []byte      `json:"data"`
-    Timestamp int64       `json:"timestamp"`
+    GetSegments() ([]string, error)
+    ReadSegment(segmentPath string) ([]WALEntry, error)
 }
 
 // BackupStorage интерфейс для хранения бэкапов
@@ -116,27 +107,84 @@ type LoggerInterface interface {
 // FileBackupStorage реализация BackupStorage для файловой системы
 type FileBackupStorage struct {
     backupDir     string
+    compressLevel int
+    mu            sync.Mutex
 }
 
 // NewFileBackupStorage создаёт новое файловое хранилище бэкапов
-func NewFileBackupStorage(backupDir string) (*FileBackupStorage, error) {
+func NewFileBackupStorage(backupDir string, compressLevel int) (*FileBackupStorage, error) {
+    if compressLevel < 1 || compressLevel > 9 {
+        compressLevel = 6
+    }
     if err := os.MkdirAll(backupDir, 0755); err != nil {
         return nil, fmt.Errorf("failed to create backup dir: %v", err)
     }
-    return &FileBackupStorage{backupDir: backupDir}, nil
+    return &FileBackupStorage{
+        backupDir:     backupDir,
+        compressLevel: compressLevel,
+    }, nil
 }
 
 func (fbs *FileBackupStorage) SaveBackup(path string, data []byte) error {
+    fbs.mu.Lock()
+    defer fbs.mu.Unlock()
+
     fullPath := filepath.Join(fbs.backupDir, path)
     if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil {
         return err
     }
-    return os.WriteFile(fullPath, data, 0644)
+
+    // Сжимаем данные
+    var buf bytes.Buffer
+    gzWriter, err := gzip.NewWriterLevel(&buf, fbs.compressLevel)
+    if err != nil {
+        return err
+    }
+    if _, err := gzWriter.Write(data); err != nil {
+        gzWriter.Close()
+        return err
+    }
+    if err := gzWriter.Close(); err != nil {
+        return err
+    }
+
+    compressedPath := fullPath + ".gz"
+    return os.WriteFile(compressedPath, buf.Bytes(), 0644)
 }
 
 func (fbs *FileBackupStorage) LoadBackup(path string) ([]byte, error) {
+    fbs.mu.Lock()
+    defer fbs.mu.Unlock()
+
     fullPath := filepath.Join(fbs.backupDir, path)
-    return os.ReadFile(fullPath)
+    compressedPath := fullPath + ".gz"
+
+    var data []byte
+    var err error
+
+    // Пробуем сначала сжатый файл
+    if _, err := os.Stat(compressedPath); err == nil {
+        compressedData, err := os.ReadFile(compressedPath)
+        if err != nil {
+            return nil, err
+        }
+        gzReader, err := gzip.NewReader(bytes.NewReader(compressedData))
+        if err != nil {
+            return nil, err
+        }
+        defer gzReader.Close()
+        data, err = io.ReadAll(gzReader)
+        if err != nil {
+            return nil, err
+        }
+    } else {
+        data, err = os.ReadFile(fullPath)
+        if err != nil {
+            return nil, err
+        }
+    }
+
+    return data, nil
 }
 
 func (fbs *FileBackupStorage) ListBackups() ([]string, error) {
@@ -146,20 +194,243 @@ func (fbs *FileBackupStorage) ListBackups() ([]string, error) {
     }
     names := make([]string, 0, len(entries))
     for _, e := range entries {
+        if e.IsDir() {
+            // Рекурсивно собираем файлы из поддиректорий
+            subEntries, err := os.ReadDir(filepath.Join(fbs.backupDir, e.Name()))
+            if err == nil {
+                for _, sub := range subEntries {
+                    if !sub.IsDir() {
+                        names = append(names, filepath.Join(e.Name(), sub.Name()))
+                    }
+                }
+            }
+        } else {
             names = append(names, e.Name())
         }
+    }
     return names, nil
 }
 
 func (fbs *FileBackupStorage) DeleteBackup(path string) error {
+    fbs.mu.Lock()
+    defer fbs.mu.Unlock()
+
     fullPath := filepath.Join(fbs.backupDir, path)
-    return os.RemoveAll(fullPath)
+    compressedPath := fullPath + ".gz"
+
+    var err error
+    if _, statErr := os.Stat(compressedPath); statErr == nil {
+        err = os.Remove(compressedPath)
+    } else {
+        err = os.Remove(fullPath)
+    }
+
+    // Пробуем удалить пустые директории
+    dirPath := filepath.Dir(fullPath)
+    os.Remove(dirPath)
+
+    return err
 }
 
 func (fbs *FileBackupStorage) BackupExists(path string) bool {
     fullPath := filepath.Join(fbs.backupDir, path)
-    _, err := os.Stat(fullPath)
-    return err == nil
+    compressedPath := fullPath + ".gz"
+    _, err1 := os.Stat(fullPath)
+    _, err2 := os.Stat(compressedPath)
+    return err1 == nil || err2 == nil
+}
+
+// WALReaderImpl реализация WALReader для SegmentedWALManager
+type WALReaderImpl struct {
+    walManager      interface{} // *SegmentedWALManager или *WALManager
+    lastBackupIndex atomic.Uint64
+    segmentsDir     string
+    mu              sync.RWMutex
+    logger          LoggerInterface
+}
+
+// NewWALReaderImpl создаёт новый WALReader
+func NewWALReaderImpl(walManager interface{}, segmentsDir string, logger LoggerInterface) *WALReaderImpl {
+    return &WALReaderImpl{
+        walManager:  walManager,
+        segmentsDir: segmentsDir,
+        logger:      logger,
+    }
+}
+
+// ReadSince читает записи WAL с указанного индекса
+func (wr *WALReaderImpl) ReadSince(index uint64) ([]WALEntry, error) {
+    wr.mu.RLock()
+    defer wr.mu.RUnlock()
+
+    entries := make([]WALEntry, 0)
+
+    // Получаем список сегментов
+    segments, err := wr.GetSegments()
+    if err != nil {
+        return nil, err
+    }
+
+    for _, segmentPath := range segments {
+        segEntries, err := wr.ReadSegment(segmentPath)
+        if err != nil {
+            if wr.logger != nil {
+                wr.logger.Warn(fmt.Sprintf("Failed to read segment %s: %v", segmentPath, err))
+            }
+            continue
+        }
+
+        for _, entry := range segEntries {
+            if entry.Index > index {
+                entries = append(entries, entry)
+            }
+        }
+    }
+
+    return entries, nil
+}
+
+// GetCurrentIndex возвращает текущий индекс WAL
+func (wr *WALReaderImpl) GetCurrentIndex() (uint64, error) {
+    segments, err := wr.GetSegments()
+    if err != nil {
+        return 0, err
+    }
+
+    if len(segments) == 0 {
+        return 0, nil
+    }
+
+    // Читаем последний сегмент для получения последнего индекса
+    lastSegment := segments[len(segments)-1]
+    entries, err := wr.ReadSegment(lastSegment)
+    if err != nil {
+        return 0, err
+    }
+
+    if len(entries) == 0 {
+        return 0, nil
+    }
+
+    return entries[len(entries)-1].Index, nil
+}
+
+// GetLastBackupIndex возвращает последний сохранённый индекс бэкапа
+func (wr *WALReaderImpl) GetLastBackupIndex() uint64 {
+    return wr.lastBackupIndex.Load()
+}
+
+// SetLastBackupIndex устанавливает последний индекс бэкапа
+func (wr *WALReaderImpl) SetLastBackupIndex(index uint64) error {
+    wr.lastBackupIndex.Store(index)
+
+    // Сохраняем в файл для персистентности
+    indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json")
+    data, err := json.Marshal(map[string]uint64{"last_index": index})
+    if err != nil {
+        return err
+    }
+    return os.WriteFile(indexPath, data, 0644)
+}
+
+// GetSegments возвращает список сегментов WAL
+func (wr *WALReaderImpl) GetSegments() ([]string, error) {
+    if wr.segmentsDir == "" {
+        return nil, fmt.Errorf("segments directory not set")
+    }
+
+    pattern := filepath.Join(wr.segmentsDir, "wal_segment_*.log")
+    files, err := filepath.Glob(pattern)
+    if err != nil {
+        return nil, err
+    }
+
+    sort.Strings(files)
+    return files, nil
+}
+
+// ReadSegment читает сегмент WAL
+func (wr *WALReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) {
+    data, err := os.ReadFile(segmentPath)
+    if err != nil {
+        return nil, err
+    }
+
+    entries := make([]WALEntry, 0)
+
+    // Парсим WAL записи (формат: [length][data] повторяется)
+    pos := 0
+    for pos < len(data) {
+        if pos+4 > len(data) {
+            break
+        }
+
+        // Читаем длину записи (4 байта, big-endian)
+        length := int(data[pos])<<24 | int(data[pos+1])<<16 | int(data[pos+2])<<8 | int(data[pos+3])
+        pos += 4
+
+        if pos+length > len(data) {
+            break
+        }
+
+        recordData := data[pos : pos+length]
+        pos += length
+
+        var record struct {
+            LSN       uint64 `json:"lsn"`
+            Timestamp int64  `json:"timestamp"`
+            Type      byte   `json:"type"`
+            Data      []byte `json:"data"`
+        }
+
+        if err := json.Unmarshal(recordData, &record); err != nil {
+            continue
+        }
+
+        entry := WALEntry{
+            Index:     record.LSN,
+            Timestamp: record.Timestamp,
+            Type:      fmt.Sprintf("%d", record.Type),
+            Data:      record.Data,
+        }
+
+        entries = append(entries, entry)
+    }
+
+    return entries, nil
+}
+
+// loadLastBackupIndex загружает последний индекс из файла
+func (wr *WALReaderImpl) loadLastBackupIndex() {
+    indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json")
+    data, err := os.ReadFile(indexPath)
+    if err != nil {
+        return
+    }
+
+    var meta map[string]uint64
+    if err := json.Unmarshal(data, &meta); err != nil {
+        return
+    }
+
+    if lastIdx, ok := meta["last_index"]; ok {
+        wr.lastBackupIndex.Store(lastIdx)
+    }
+}
+
+// BackupScheduler планировщик бэкапов
+type BackupScheduler struct {
+    schedules     map[string]*BackupSchedule
+    backups       map[string]*BackupInfo
+    walReader     WALReader
+    storage       BackupStorage
+    logger        LoggerInterface
+    mu            sync.RWMutex
+    stopChan      chan struct{}
+    wg            sync.WaitGroup
+    running       atomic.Bool
+    backupCounter atomic.Uint64
+    config        *BackupSchedulerConfig
 }
 
 // BackupSchedulerConfig конфигурация планировщика
@@ -168,6 +439,8 @@ type BackupSchedulerConfig struct {
     MaxConcurrent    int    `json:"max_concurrent"`
     CompressEnabled  bool   `json:"compress_enabled"`
     CompressionLevel int    `json:"compression_level"`
+    MaxBackupSizeMB  int64  `json:"max_backup_size_mb"`
+    EnableIncremental bool  `json:"enable_incremental"`
 }
 
 // DefaultBackupSchedulerConfig возвращает конфигурацию по умолчанию
@@ -177,6 +450,8 @@ func DefaultBackupSchedulerConfig() *BackupSchedulerConfig {
         MaxConcurrent:     1,
         CompressEnabled:   true,
         CompressionLevel:  6,
+        MaxBackupSizeMB:   10240, // 10GB
+        EnableIncremental: true,
     }
 }
 
@@ -186,7 +461,12 @@ func NewBackupScheduler(cfg *BackupSchedulerConfig, walReader WALReader, logger
         cfg = DefaultBackupSchedulerConfig()
     }
 
-    storage, err := NewFileBackupStorage(cfg.BackupDir)
+    compressLevel := cfg.CompressionLevel
+    if !cfg.CompressEnabled {
+        compressLevel = 0
+    }
+
+    storage, err := NewFileBackupStorage(cfg.BackupDir, compressLevel)
     if err != nil {
         return nil, err
     }
@@ -198,6 +478,7 @@ func NewBackupScheduler(cfg *BackupSchedulerConfig, walReader WALReader, logger
         storage:   storage,
         logger:    logger,
         stopChan:  make(chan struct{}),
+        config:    cfg,
     }
 
     // Загружаем существующие расписания
@@ -211,10 +492,9 @@ func NewBackupScheduler(cfg *BackupSchedulerConfig, walReader WALReader, logger
 
 // loadSchedules загружает расписания из хранилища
 func (bs *BackupScheduler) loadSchedules() {
-    schedulesPath := filepath.Join(bs.storage.(*FileBackupStorage).backupDir, "schedules.json")
     data, err := bs.storage.LoadBackup("schedules.json")
     if err != nil {
-        if !os.IsNotExist(err) {
+        if !os.IsNotExist(err) && bs.logger != nil {
             bs.logger.Warn(fmt.Sprintf("Failed to load schedules: %v", err))
         }
         return
@@ -222,7 +502,9 @@ func (bs *BackupScheduler) loadSchedules() {
 
     var schedules map[string]*BackupSchedule
     if err := json.Unmarshal(data, &schedules); err != nil {
+        if bs.logger != nil {
             bs.logger.Warn(fmt.Sprintf("Failed to parse schedules: %v", err))
+        }
         return
     }
 
@@ -246,10 +528,9 @@ func (bs *BackupScheduler) saveSchedules() error {
 
 // loadBackupHistory загружает историю бэкапов
 func (bs *BackupScheduler) loadBackupHistory() {
-    historyPath := filepath.Join(bs.storage.(*FileBackupStorage).backupDir, "backups.json")
     data, err := bs.storage.LoadBackup("backups.json")
     if err != nil {
-        if !os.IsNotExist(err) {
+        if !os.IsNotExist(err) && bs.logger != nil {
             bs.logger.Warn(fmt.Sprintf("Failed to load backup history: %v", err))
         }
         return
@@ -257,7 +538,9 @@ func (bs *BackupScheduler) loadBackupHistory() {
 
     var backups map[string]*BackupInfo
     if err := json.Unmarshal(data, &backups); err != nil {
+        if bs.logger != nil {
             bs.logger.Warn(fmt.Sprintf("Failed to parse backup history: %v", err))
+        }
         return
     }
 
@@ -284,6 +567,13 @@ func (bs *BackupScheduler) AddSchedule(name, cronExpr string, backupType BackupT
     bs.mu.Lock()
     defer bs.mu.Unlock()
 
+    // Проверяем, что расписание с таким именем не существует
+    for _, s := range bs.schedules {
+        if s.Name == name {
+            return nil, fmt.Errorf("schedule with name '%s' already exists", name)
+        }
+    }
+
     schedule := &BackupSchedule{
         ID:            fmt.Sprintf("sch_%d", time.Now().UnixNano()),
         Name:          name,
@@ -298,10 +588,12 @@ func (bs *BackupScheduler) AddSchedule(name, cronExpr string, backupType BackupT
     bs.schedules[schedule.ID] = schedule
 
     if err := bs.saveSchedules(); err != nil {
-        return nil, err
+        return nil, fmt.Errorf("failed to save schedules: %v", err)
     }
 
+    if bs.logger != nil {
         bs.logger.Info(fmt.Sprintf("Added backup schedule: %s (%s) with cron %s", name, backupType, cronExpr))
+    }
     return schedule, nil
 }
 
@@ -315,7 +607,15 @@ func (bs *BackupScheduler) RemoveSchedule(scheduleID string) error {
     }
 
     delete(bs.schedules, scheduleID)
-    return bs.saveSchedules()
+
+    if err := bs.saveSchedules(); err != nil {
+        return fmt.Errorf("failed to save schedules: %v", err)
+    }
+
+    if bs.logger != nil {
+        bs.logger.Info(fmt.Sprintf("Removed schedule: %s", scheduleID))
+    }
+    return nil
 }
 
 // RunNow запускает бэкап немедленно
@@ -328,6 +628,10 @@ func (bs *BackupScheduler) RunNow(scheduleID string) error {
         return fmt.Errorf("schedule not found: %s", scheduleID)
     }
 
+    if !bs.config.EnableIncremental && schedule.Type == IncrementalBackup {
+        return fmt.Errorf("incremental backups are disabled in configuration")
+    }
+
     if schedule.Type == FullBackup {
         return bs.performFullBackup(scheduleID)
     }
@@ -342,7 +646,7 @@ func (bs *BackupScheduler) performFullBackup(scheduleID string) error {
     defer bs.running.Store(false)
 
     startTime := time.Now().UnixMilli()
-    backupID := fmt.Sprintf("full_%d", bs.backupCounter.Add(1))
+    backupID := fmt.Sprintf("full_%d_%d", startTime, bs.backupCounter.Add(1))
 
     bs.logger.Info(fmt.Sprintf("Starting full backup %s", backupID))
 
@@ -357,20 +661,55 @@ func (bs *BackupScheduler) performFullBackup(scheduleID string) error {
         bs.logger.Warn(fmt.Sprintf("Failed to set last backup index: %v", err))
     }
 
-    // Здесь выполняется реальный бэкап данных
+    // Читаем все WAL записи для полного бэкапа
+    walEntries, err := bs.walReader.ReadSince(0)
+    if err != nil {
+        bs.logger.Warn(fmt.Sprintf("Failed to read WAL for full backup: %v", err))
+        walEntries = []WALEntry{}
+    }
+
+    // Создаём структуру бэкапа
+    backupData := struct {
+        BackupID   string      `json:"backup_id"`
+        Type       string      `json:"type"`
+        StartTime  int64       `json:"start_time"`
+        WALIndex   uint64      `json:"wal_index"`
+        Entries    []WALEntry  `json:"entries,omitempty"`
+        Metadata   interface{} `json:"metadata,omitempty"`
+    }{
+        BackupID:  backupID,
+        Type:      "full",
+        StartTime: startTime,
+        WALIndex:  walIndex,
+        Entries:   walEntries,
+        Metadata: map[string]interface{}{
+            "version":     "1.0",
+            "created_by":  "backup_scheduler",
+            "wal_entries_count": len(walEntries),
+        },
+    }
+
+    data, err := json.Marshal(backupData)
+    if err != nil {
+        return fmt.Errorf("failed to marshal backup data: %v", err)
+    }
+
+    // Проверяем размер бэкапа
+    if bs.config.MaxBackupSizeMB > 0 && int64(len(data)) > bs.config.MaxBackupSizeMB*1024*1024 {
+        return fmt.Errorf("backup size %d exceeds maximum allowed %d MB",
+            len(data)/(1024*1024), bs.config.MaxBackupSizeMB)
+    }
+
+    // Сохраняем бэкап
     backupPath := fmt.Sprintf("full/%s/%s.backup",
         time.Now().Format("2006-01-02"), backupID)
 
-    // Симуляция бэкапа (в реальности здесь будет копирование данных)
-    backupData := []byte(fmt.Sprintf(`{"backup_id":"%s","type":"full","timestamp":%d,"wal_index":%d}`, 
-        backupID, startTime, walIndex))
-    
-    if err := bs.storage.SaveBackup(backupPath, backupData); err != nil {
+    if err := bs.storage.SaveBackup(backupPath, data); err != nil {
         return fmt.Errorf("failed to save backup: %v", err)
     }
 
     // Вычисляем контрольную сумму
-    hash := sha256.Sum256(backupData)
+    hash := sha256.Sum256(data)
     checksum := hex.EncodeToString(hash[:])
 
     endTime := time.Now().UnixMilli()
@@ -382,7 +721,7 @@ func (bs *BackupScheduler) performFullBackup(scheduleID string) error {
         StartTime:  startTime,
         EndTime:    endTime,
         Status:     "completed",
-        SizeBytes:  int64(len(backupData)),
+        SizeBytes:  int64(len(data)),
         Path:       backupPath,
         WALStart:   0,
         WALEnd:     walIndex,
@@ -393,13 +732,16 @@ func (bs *BackupScheduler) performFullBackup(scheduleID string) error {
     bs.backups[backupID] = backupInfo
     bs.mu.Unlock()
 
-    bs.saveBackupHistory()
+    if err := bs.saveBackupHistory(); err != nil {
+        bs.logger.Warn(fmt.Sprintf("Failed to save backup history: %v", err))
+    }
+
     bs.updateScheduleLastRun(scheduleID, startTime)
     bs.cleanupOldBackups(scheduleID)
 
     duration := (endTime - startTime) / 1000
-    bs.logger.Info(fmt.Sprintf("Full backup %s completed in %d seconds, size=%d bytes", 
-        backupID, duration, backupInfo.SizeBytes))
+    bs.logger.Info(fmt.Sprintf("Full backup %s completed in %d seconds, size=%d bytes, entries=%d",
+        backupID, duration, backupInfo.SizeBytes, len(walEntries)))
 
     return nil
 }
@@ -412,7 +754,7 @@ func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error {
     defer bs.running.Store(false)
 
     startTime := time.Now().UnixMilli()
-    backupID := fmt.Sprintf("inc_%d", bs.backupCounter.Add(1))
+    backupID := fmt.Sprintf("inc_%d_%d", startTime, bs.backupCounter.Add(1))
 
     // Находим последний полный бэкап
     lastFullBackup := bs.findLastFullBackup()
@@ -421,6 +763,12 @@ func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error {
         return bs.performFullBackup(scheduleID)
     }
 
+    // Проверяем, что родительский бэкап существует
+    if !bs.storage.BackupExists(lastFullBackup.Path) {
+        bs.logger.Warn(fmt.Sprintf("Parent backup %s file missing, performing full backup instead", lastFullBackup.ID))
+        return bs.performFullBackup(scheduleID)
+    }
+
     // Получаем WAL записи с момента последнего бэкапа
     lastIndex := bs.walReader.GetLastBackupIndex()
     if lastIndex == 0 {
@@ -437,12 +785,12 @@ func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error {
         return nil
     }
 
-    currentIndex, _ := bs.walReader.GetCurrentIndex()
-    
-    // Сохраняем инкрементальный бэкап
-    backupPath := fmt.Sprintf("inc/%s/%s.backup",
-        time.Now().Format("2006-01-02"), backupID)
+    currentIndex, err := bs.walReader.GetCurrentIndex()
+    if err != nil {
+        return fmt.Errorf("failed to get current WAL index: %v", err)
+    }
 
+    // Создаём структуру инкрементального бэкапа
     backupData := struct {
         BackupID   string      `json:"backup_id"`
         Type       string      `json:"type"`
@@ -451,6 +799,7 @@ func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error {
         WALEnd     uint64      `json:"wal_end"`
         ParentID   string      `json:"parent_id"`
         Entries    []WALEntry  `json:"entries"`
+        Metadata   interface{} `json:"metadata,omitempty"`
     }{
         BackupID:  backupID,
         Type:      "incremental",
@@ -459,6 +808,13 @@ func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error {
         WALEnd:    currentIndex,
         ParentID:  lastFullBackup.ID,
         Entries:   walEntries,
+        Metadata: map[string]interface{}{
+            "version":         "1.0",
+            "created_by":      "backup_scheduler",
+            "entries_count":   len(walEntries),
+            "parent_backup":   lastFullBackup.ID,
+            "parent_time":     lastFullBackup.StartTime,
+        },
     }
 
     data, err := json.Marshal(backupData)
@@ -466,6 +822,17 @@ func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error {
         return fmt.Errorf("failed to marshal backup data: %v", err)
     }
 
+    // Проверяем размер бэкапа
+    if bs.config.MaxBackupSizeMB > 0 && int64(len(data)) > bs.config.MaxBackupSizeMB*1024*1024 {
+        // Если инкрементальный бэкап слишком большой, делаем полный
+        bs.logger.Info("Incremental backup too large, performing full backup instead")
+        return bs.performFullBackup(scheduleID)
+    }
+
+    // Сохраняем инкрементальный бэкап
+    backupPath := fmt.Sprintf("inc/%s/%s.backup",
+        time.Now().Format("2006-01-02"), backupID)
+
     if err := bs.storage.SaveBackup(backupPath, data); err != nil {
         return fmt.Errorf("failed to save incremental backup: %v", err)
     }
@@ -499,7 +866,10 @@ func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error {
     bs.backups[backupID] = backupInfo
     bs.mu.Unlock()
 
-    bs.saveBackupHistory()
+    if err := bs.saveBackupHistory(); err != nil {
+        bs.logger.Warn(fmt.Sprintf("Failed to save backup history: %v", err))
+    }
+
     bs.updateScheduleLastRun(scheduleID, startTime)
 
     duration := (endTime - startTime) / 1000
@@ -533,7 +903,9 @@ func (bs *BackupScheduler) updateScheduleLastRun(scheduleID string, timestamp in
     if schedule, exists := bs.schedules[scheduleID]; exists {
         schedule.LastRun = timestamp
         schedule.UpdatedAt = time.Now().UnixMilli()
-        bs.saveSchedules()
+        if err := bs.saveSchedules(); err != nil {
+            bs.logger.Warn(fmt.Sprintf("Failed to save schedules after update: %v", err))
+        }
     }
 }
 
@@ -554,7 +926,9 @@ func (bs *BackupScheduler) cleanupOldBackups(scheduleID string) {
         if backup.ScheduleID == scheduleID && backup.StartTime < cutoffTime {
             toDelete = append(toDelete, id)
             // Удаляем файл бэкапа
-            bs.storage.DeleteBackup(backup.Path)
+            if err := bs.storage.DeleteBackup(backup.Path); err != nil {
+                bs.logger.Warn(fmt.Sprintf("Failed to delete backup file %s: %v", backup.Path, err))
+            }
         }
     }
 
@@ -563,7 +937,9 @@ func (bs *BackupScheduler) cleanupOldBackups(scheduleID string) {
     }
 
     if len(toDelete) > 0 {
-        bs.saveBackupHistory()
+        if err := bs.saveBackupHistory(); err != nil {
+            bs.logger.Warn(fmt.Sprintf("Failed to save backup history after cleanup: %v", err))
+        }
         bs.logger.Info(fmt.Sprintf("Cleaned up %d old backups for schedule %s", len(toDelete), scheduleID))
     }
 }
@@ -578,7 +954,7 @@ func (bs *BackupScheduler) ListBackups() []*BackupInfo {
         backups = append(backups, b)
     }
 
-    // Сортируем по времени
+    // Сортируем по времени (новейшие первыми)
     sort.Slice(backups, func(i, j int) bool {
         return backups[i].StartTime > backups[j].StartTime
     })
@@ -607,6 +983,11 @@ func (bs *BackupScheduler) GetBackup(backupID string) *BackupInfo {
 
 // Restore восстанавливает данные из бэкапа
 func (bs *BackupScheduler) Restore(backupID string) error {
+    if !bs.running.CompareAndSwap(false, true) {
+        return fmt.Errorf("restore already in progress")
+    }
+    defer bs.running.Store(false)
+
     backup := bs.GetBackup(backupID)
     if backup == nil {
         return fmt.Errorf("backup not found: %s", backupID)
@@ -614,6 +995,11 @@ func (bs *BackupScheduler) Restore(backupID string) error {
 
     bs.logger.Info(fmt.Sprintf("Starting restore from backup %s", backupID))
 
+    // Проверяем существование файла бэкапа
+    if !bs.storage.BackupExists(backup.Path) {
+        return fmt.Errorf("backup file not found: %s", backup.Path)
+    }
+
     data, err := bs.storage.LoadBackup(backup.Path)
     if err != nil {
         return fmt.Errorf("failed to load backup: %v", err)
@@ -629,16 +1015,54 @@ func (bs *BackupScheduler) Restore(backupID string) error {
     if backup.Type == FullBackup {
         // Восстановление из полного бэкапа
         bs.logger.Info("Restoring from full backup...")
-        // Здесь реальное восстановление
+
+        var fullData struct {
+            BackupID string      `json:"backup_id"`
+            Type     string      `json:"type"`
+            Entries  []WALEntry  `json:"entries"`
+            Metadata interface{} `json:"metadata"`
+        }
+
+        if err := json.Unmarshal(data, &fullData); err != nil {
+            return fmt.Errorf("failed to parse full backup data: %v", err)
+        }
+
+        bs.logger.Info(fmt.Sprintf("Full backup contains %d WAL entries", len(fullData.Entries)))
+        // Здесь реальное восстановление данных
+
     } else {
         // Восстановление из инкрементального бэкапа
+        bs.logger.Info("Restoring from incremental backup...")
+
         // Сначала восстанавливаем полный бэкап-родитель
-        if backup.ParentID != "" {
+        if backup.ParentID == "" {
+            return fmt.Errorf("incremental backup has no parent ID")
+        }
+
+        // Проверяем существование родительского бэкапа
+        parentBackup := bs.GetBackup(backup.ParentID)
+        if parentBackup == nil {
+            return fmt.Errorf("parent backup %s not found", backup.ParentID)
+        }
+
         if err := bs.Restore(backup.ParentID); err != nil {
             return fmt.Errorf("failed to restore parent backup: %v", err)
         }
+
+        var incData struct {
+            BackupID string     `json:"backup_id"`
+            Type     string     `json:"type"`
+            WALStart uint64     `json:"wal_start"`
+            WALEnd   uint64     `json:"wal_end"`
+            Entries  []WALEntry `json:"entries"`
         }
-        bs.logger.Info("Applying incremental changes...")
+
+        if err := json.Unmarshal(data, &incData); err != nil {
+            return fmt.Errorf("failed to parse incremental backup data: %v", err)
+        }
+
+        bs.logger.Info(fmt.Sprintf("Applying %d incremental changes (WAL %d -> %d)",
+            len(incData.Entries), incData.WALStart, incData.WALEnd))
         // Здесь применяем инкрементальные изменения
     }
 
@@ -650,8 +1074,10 @@ func (bs *BackupScheduler) Restore(backupID string) error {
 func (bs *BackupScheduler) Start() {
     bs.wg.Add(1)
     go bs.schedulerLoop()
+    if bs.logger != nil {
         bs.logger.Info("Backup scheduler started")
     }
+}
 
 // schedulerLoop основной цикл планировщика
 func (bs *BackupScheduler) schedulerLoop() {
@@ -691,7 +1117,10 @@ func (bs *BackupScheduler) checkSchedules() {
             bs.mu.Lock()
             bs.schedules[schedule.ID] = schedule
             bs.mu.Unlock()
-            bs.saveSchedules()
+
+            if err := bs.saveSchedules(); err != nil {
+                bs.logger.Warn(fmt.Sprintf("Failed to save schedule after update: %v", err))
+            }
 
             // Запускаем бэкап в отдельной горутине
             scheduleID := schedule.ID
@@ -711,5 +1140,38 @@ func (bs *BackupScheduler) checkSchedules() {
 func (bs *BackupScheduler) Stop() {
     close(bs.stopChan)
     bs.wg.Wait()
+    if bs.logger != nil {
         bs.logger.Info("Backup scheduler stopped")
     }
+}
+
+// GetStats возвращает статистику планировщика
+func (bs *BackupScheduler) GetStats() map[string]interface{} {
+    bs.mu.RLock()
+    defer bs.mu.RUnlock()
+
+    fullCount := 0
+    incCount := 0
+    totalSize := int64(0)
+
+    for _, b := range bs.backups {
+        if b.Type == FullBackup {
+            fullCount++
+        } else {
+            incCount++
+        }
+        totalSize += b.SizeBytes
+    }
+
+    return map[string]interface{}{
+        "total_backups":      len(bs.backups),
+        "full_backups":       fullCount,
+        "incremental_backups": incCount,
+        "total_size_bytes":   totalSize,
+        "schedules":          len(bs.schedules),
+        "is_running":         bs.running.Load(),
+        "backup_dir":         bs.config.BackupDir,
+        "compress_enabled":   bs.config.CompressEnabled,
+        "incremental_enabled": bs.config.EnableIncremental,
+    }
+}

internal/storage/persistence.go

@@ -16,6 +16,8 @@ package storage
 import (
     "bytes"
     "compress/gzip"
+    "crypto/sha256"
+    "encoding/hex"
     "encoding/json"
     "fmt"
     "os"
@@ -27,6 +29,202 @@ import (
     "time"
 )
 
+// ========== WALReader интерфейс и реализация ==========
+
+// WALEntry представляет запись WAL
+type WALEntry struct {
+    Index     uint64 `json:"index"`
+    Term      uint64 `json:"term"`
+    Type      string `json:"type"`
+    Data      []byte `json:"data"`
+    Timestamp int64  `json:"timestamp"`
+}
+
+// WALReader интерфейс для чтения WAL
+type WALReader interface {
+    ReadSince(index uint64) ([]WALEntry, error)
+    GetCurrentIndex() (uint64, error)
+    GetLastBackupIndex() uint64
+    SetLastBackupIndex(index uint64) error
+    GetSegments() ([]string, error)
+    ReadSegment(segmentPath string) ([]WALEntry, error)
+}
+
+// walReaderImpl реализация WALReader для работы с файлами WAL
+type walReaderImpl struct {
+    segmentsDir     string
+    lastBackupIndex atomic.Uint64
+    mu              sync.RWMutex
+    logger          LoggerInterface
+}
+
+// NewWALReaderImpl создаёт новый WALReader
+func NewWALReaderImpl(segmentsDir string, logger LoggerInterface) WALReader {
+    wr := &walReaderImpl{
+        segmentsDir: segmentsDir,
+        logger:      logger,
+    }
+    wr.loadLastBackupIndex()
+    return wr
+}
+
+// loadLastBackupIndex загружает последний индекс из файла
+func (wr *walReaderImpl) loadLastBackupIndex() {
+    indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json")
+    data, err := os.ReadFile(indexPath)
+    if err != nil {
+        return
+    }
+
+    var meta map[string]uint64
+    if err := json.Unmarshal(data, &meta); err != nil {
+        return
+    }
+
+    if lastIdx, ok := meta["last_index"]; ok {
+        wr.lastBackupIndex.Store(lastIdx)
+    }
+}
+
+// ReadSince читает записи WAL с указанного индекса
+func (wr *walReaderImpl) ReadSince(index uint64) ([]WALEntry, error) {
+    wr.mu.RLock()
+    defer wr.mu.RUnlock()
+
+    entries := make([]WALEntry, 0)
+
+    segments, err := wr.GetSegments()
+    if err != nil {
+        return nil, err
+    }
+
+    for _, segmentPath := range segments {
+        segEntries, err := wr.ReadSegment(segmentPath)
+        if err != nil {
+            if wr.logger != nil {
+                wr.logger.Warn(fmt.Sprintf("Failed to read segment %s: %v", segmentPath, err))
+            }
+            continue
+        }
+
+        for _, entry := range segEntries {
+            if entry.Index > index {
+                entries = append(entries, entry)
+            }
+        }
+    }
+
+    return entries, nil
+}
+
+// GetCurrentIndex возвращает текущий индекс WAL
+func (wr *walReaderImpl) GetCurrentIndex() (uint64, error) {
+    segments, err := wr.GetSegments()
+    if err != nil {
+        return 0, err
+    }
+
+    if len(segments) == 0 {
+        return 0, nil
+    }
+
+    lastSegment := segments[len(segments)-1]
+    entries, err := wr.ReadSegment(lastSegment)
+    if err != nil {
+        return 0, err
+    }
+
+    if len(entries) == 0 {
+        return 0, nil
+    }
+
+    return entries[len(entries)-1].Index, nil
+}
+
+// GetLastBackupIndex возвращает последний сохранённый индекс бэкапа
+func (wr *walReaderImpl) GetLastBackupIndex() uint64 {
+    return wr.lastBackupIndex.Load()
+}
+
+// SetLastBackupIndex устанавливает последний индекс бэкапа
+func (wr *walReaderImpl) SetLastBackupIndex(index uint64) error {
+    wr.lastBackupIndex.Store(index)
+
+    indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json")
+    data, err := json.Marshal(map[string]uint64{"last_index": index})
+    if err != nil {
+        return err
+    }
+    return os.WriteFile(indexPath, data, 0644)
+}
+
+// GetSegments возвращает список сегментов WAL
+func (wr *walReaderImpl) GetSegments() ([]string, error) {
+    if wr.segmentsDir == "" {
+        return nil, fmt.Errorf("segments directory not set")
+    }
+
+    pattern := filepath.Join(wr.segmentsDir, "wal_segment_*.log")
+    files, err := filepath.Glob(pattern)
+    if err != nil {
+        return nil, err
+    }
+
+    sort.Strings(files)
+    return files, nil
+}
+
+// ReadSegment читает сегмент WAL
+func (wr *walReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) {
+    data, err := os.ReadFile(segmentPath)
+    if err != nil {
+        return nil, err
+    }
+
+    entries := make([]WALEntry, 0)
+
+    pos := 0
+    for pos < len(data) {
+        if pos+4 > len(data) {
+            break
+        }
+
+        length := int(data[pos])<<24 | int(data[pos+1])<<16 | int(data[pos+2])<<8 | int(data[pos+3])
+        pos += 4
+
+        if pos+length > len(data) {
+            break
+        }
+
+        recordData := data[pos : pos+length]
+        pos += length
+
+        var record struct {
+            LSN       uint64 `json:"lsn"`
+            Timestamp int64  `json:"timestamp"`
+            Type      byte   `json:"type"`
+            Data      []byte `json:"data"`
+        }
+
+        if err := json.Unmarshal(recordData, &record); err != nil {
+            continue
+        }
+
+        entry := WALEntry{
+            Index:     record.LSN,
+            Timestamp: record.Timestamp,
+            Type:      fmt.Sprintf("%d", record.Type),
+            Data:      record.Data,
+        }
+
+        entries = append(entries, entry)
+    }
+
+    return entries, nil
+}
+
+// ========== PersistenceConfig ==========
+
 // PersistenceConfig конфигурация персистентного хранения
 type PersistenceConfig struct {
     DataDir             string        `json:"data_dir"`
@@ -34,6 +232,9 @@ type PersistenceConfig struct {
     MaxCheckpoints      int           `json:"max_checkpoints"`
     CompressEnabled     bool          `json:"compress_enabled"`
     SyncWrites          bool          `json:"sync_writes"`
+    WalPath             string        `json:"wal_path"`
+    UseWalForCheckpoint bool          `json:"use_wal_for_checkpoint"`
+    AtomicWrites        bool          `json:"atomic_writes"`
 }
 
 // DefaultPersistenceConfig возвращает конфигурацию по умолчанию
@@ -44,6 +245,9 @@ func DefaultPersistenceConfig() *PersistenceConfig {
         MaxCheckpoints:      10,
         CompressEnabled:     true,
         SyncWrites:          true,
+        WalPath:             "futriis.wal",
+        UseWalForCheckpoint: true,
+        AtomicWrites:        true,
     }
 }
 
@@ -54,6 +258,21 @@ type DatabaseSnapshot struct {
     CreatedAt   int64                  `json:"created_at"`
     Version     uint64                 `json:"version"`
     Checksum    string                 `json:"checksum"`
+    WalLSN      uint64                 `json:"wal_lsn,omitempty"`
+    WalSegment  string                 `json:"wal_segment,omitempty"`
+}
+
+// CheckpointRecoveryInfo содержит информацию о восстановлении из чекпоинта
+type CheckpointRecoveryInfo struct {
+    CheckpointTime   int64   `json:"checkpoint_time"`
+    CheckpointVersion uint64  `json:"checkpoint_version"`
+    WalLSN           uint64  `json:"wal_lsn"`
+    AppliedWalCount  int     `json:"applied_wal_count"`
+    RestoredDocs     int64   `json:"restored_docs"`
+    RestoredColls    int     `json:"restored_collections"`
+    DurationMs       int64   `json:"duration_ms"`
+    Success          bool    `json:"success"`
+    Error            string  `json:"error,omitempty"`
 }
 
 // PersistenceManager управляет персистентным хранением
@@ -66,6 +285,10 @@ type PersistenceManager struct {
     wg              sync.WaitGroup
     lastCheckpoint  int64
     checkpointID    atomic.Uint64
+    walManager      interface{}
+    walReader       WALReader
+    recoveryInfo    *CheckpointRecoveryInfo
+    isRestoring     atomic.Bool
 }
 
 // NewPersistenceManager создаёт новый менеджер персистентности
@@ -82,16 +305,30 @@ func NewPersistenceManager(config *PersistenceConfig, storage *Storage, logger L
         lastCheckpoint: time.Now().UnixMilli(),
     }
 
-    // Создаём директорию для данных
     if err := os.MkdirAll(config.DataDir, 0755); err != nil {
         if logger != nil {
             logger.Error(fmt.Sprintf("Failed to create data directory: %v", err))
         }
     }
 
+    if config.UseWalForCheckpoint && config.WalPath != "" {
+        pm.walReader = NewWALReaderImpl(filepath.Dir(config.WalPath), logger)
+    }
+
     return pm
 }
 
+// SetWALManager устанавливает WAL менеджер для синхронизации чекпоинтов
+func (pm *PersistenceManager) SetWALManager(walManager interface{}) {
+    pm.mu.Lock()
+    defer pm.mu.Unlock()
+    pm.walManager = walManager
+
+    if walManager != nil && pm.config.UseWalForCheckpoint {
+        pm.walReader = NewWALReaderImpl(filepath.Dir(pm.config.WalPath), pm.logger)
+    }
+}
+
 // Start запускает фоновое сохранение чекпоинтов
 func (pm *PersistenceManager) Start() {
     pm.wg.Add(1)
@@ -107,7 +344,6 @@ func (pm *PersistenceManager) Stop() {
     close(pm.stopChan)
     pm.wg.Wait()
 
-    // Финальное сохранение
     pm.SaveAll()
 
     if pm.logger != nil {
@@ -127,6 +363,12 @@ func (pm *PersistenceManager) checkpointLoop() {
         case <-pm.stopChan:
             return
         case <-ticker.C:
+            if pm.isRestoring.Load() {
+                if pm.logger != nil {
+                    pm.logger.Info("Skipping checkpoint during restore")
+                }
+                continue
+            }
             if err := pm.SaveAll(); err != nil {
                 if pm.logger != nil {
                     pm.logger.Error(fmt.Sprintf("Failed to save checkpoint: %v", err))
@@ -138,8 +380,12 @@ func (pm *PersistenceManager) checkpointLoop() {
     }
 }
 
-// SaveDatabase сохраняет базу данных на диск
+// SaveDatabase сохраняет базу данных на диск с атомарной записью
 func (pm *PersistenceManager) SaveDatabase(dbName string) error {
+    if pm.isRestoring.Load() {
+        return fmt.Errorf("cannot save checkpoint during restore")
+    }
+
     pm.mu.Lock()
     defer pm.mu.Unlock()
 
@@ -148,12 +394,27 @@ func (pm *PersistenceManager) SaveDatabase(dbName string) error {
         return fmt.Errorf("database not found: %s", dbName)
     }
 
-    // Собираем данные всех коллекций
+    var walLSN uint64
+    var walSegment string
+
+    if pm.config.UseWalForCheckpoint && pm.walManager != nil {
+        if segWal, ok := pm.walManager.(*SegmentedWALManager); ok {
+            if segWal.currentSegment != nil {
+                segWal.currentSegment.Writer.Flush()
+                RealFsync(segWal.currentSegment.File)
+                walLSN = segWal.currentSegment.EndLSN
+                walSegment = fmt.Sprintf("%d", segWal.currentSegment.ID)
+            }
+        }
+    }
+
     snapshot := &DatabaseSnapshot{
         Name:        dbName,
         Collections: make(map[string]interface{}),
         CreatedAt:   time.Now().UnixMilli(),
         Version:     pm.checkpointID.Add(1),
+        WalLSN:      walLSN,
+        WalSegment:  walSegment,
     }
 
     for _, collName := range db.ListCollections() {
@@ -170,19 +431,28 @@ func (pm *PersistenceManager) SaveDatabase(dbName string) error {
                 "fields":     doc.GetFields(),
                 "created_at": doc.CreatedAt,
                 "updated_at": doc.UpdatedAt,
+                "deleted_at": doc.DeletedAt,
                 "version":    doc.Version,
             })
         }
         snapshot.Collections[collName] = collData
     }
 
-    // Сериализуем и сохраняем
     data, err := json.Marshal(snapshot)
     if err != nil {
         return err
     }
 
-    // Опционально сжимаем
+    // Вычисляем контрольную сумму
+    hash := sha256.Sum256(data)
+    snapshot.Checksum = hex.EncodeToString(hash[:])
+
+    // Пере-сериализуем с контрольной суммой
+    data, err = json.Marshal(snapshot)
+    if err != nil {
+        return err
+    }
+
     if pm.config.CompressEnabled {
         data, err = pm.compress(data)
         if err != nil {
@@ -190,29 +460,61 @@ func (pm *PersistenceManager) SaveDatabase(dbName string) error {
         }
     }
 
-    // Сохраняем в файл
+    // Атомарная запись с использованием временного файла
     filename := pm.getSnapshotFilename(dbName, snapshot.Version)
-    if err := os.WriteFile(filename, data, 0644); err != nil {
+    if err := pm.atomicWriteFile(filename, data); err != nil {
         return err
     }
 
-    // Синхронизируем с диском если нужно
-    if pm.config.SyncWrites {
-        if f, err := os.OpenFile(filename, os.O_RDWR, 0644); err == nil {
-            RealFsync(f)
-            f.Close()
-        }
-    }
+    pm.lastCheckpoint = snapshot.CreatedAt
 
     if pm.logger != nil {
-        pm.logger.Info(fmt.Sprintf("Saved database %s checkpoint %d (%d bytes)", dbName, snapshot.Version, len(data)))
+        pm.logger.Info(fmt.Sprintf("Saved database %s checkpoint %d (%d bytes, checksum: %s, WAL LSN: %d)",
+            dbName, snapshot.Version, len(data), snapshot.Checksum[:8], walLSN))
     }
 
     return nil
 }
 
+// atomicWriteFile выполняет атомарную запись файла
+func (pm *PersistenceManager) atomicWriteFile(filename string, data []byte) error {
+    if pm.config.AtomicWrites {
+        // Создаём временный файл в той же директории
+        tempFile := filename + ".tmp"
+        if err := os.WriteFile(tempFile, data, 0644); err != nil {
+            return fmt.Errorf("failed to write temp file: %v", err)
+        }
+
+        // Синхронизируем временный файл
+        if pm.config.SyncWrites {
+            if f, err := os.OpenFile(tempFile, os.O_RDWR, 0644); err == nil {
+                RealFsync(f)
+                f.Close()
+            }
+        }
+
+        // Атомарное переименование
+        if err := os.Rename(tempFile, filename); err != nil {
+            os.Remove(tempFile)
+            return fmt.Errorf("failed to rename temp file: %v", err)
+        }
+
+        // Синхронизируем директорию
+        FsyncDir(pm.config.DataDir)
+
+        return nil
+    }
+
+    // Обычная запись (не атомарная)
+    return os.WriteFile(filename, data, 0644)
+}
+
 // SaveAll сохраняет все базы данных
 func (pm *PersistenceManager) SaveAll() error {
+    if pm.isRestoring.Load() {
+        return fmt.Errorf("cannot save checkpoints during restore")
+    }
+
     databases := pm.storage.ListDatabases()
 
     var lastErr error
@@ -228,44 +530,80 @@ func (pm *PersistenceManager) SaveAll() error {
     return lastErr
 }
 
-// LoadDatabase загружает базу данных с диска
+// LoadDatabase загружает базу данных с диска с применением WAL
 func (pm *PersistenceManager) LoadDatabase(dbName string) error {
+    pm.isRestoring.Store(true)
+    defer pm.isRestoring.Store(false)
+
     pm.mu.Lock()
     defer pm.mu.Unlock()
 
-    // Находим последний чекпоинт
+    startTime := time.Now()
+
+    recoveryInfo := &CheckpointRecoveryInfo{
+        CheckpointTime: time.Now().UnixMilli(),
+        Success:        false,
+    }
+
     snapshot, err := pm.findLatestSnapshot(dbName)
     if err != nil {
+        recoveryInfo.Error = err.Error()
+        pm.recoveryInfo = recoveryInfo
         return err
     }
 
     if snapshot == nil {
+        recoveryInfo.Error = "no snapshot found"
+        pm.recoveryInfo = recoveryInfo
         return fmt.Errorf("no snapshot found for database: %s", dbName)
     }
 
-    // Восстанавливаем базу данных
+    recoveryInfo.CheckpointVersion = snapshot.Version
+    recoveryInfo.WalLSN = snapshot.WalLSN
+
+    // Проверяем целостность чекпоинта
+    if err := pm.verifySnapshot(snapshot); err != nil {
+        recoveryInfo.Error = err.Error()
+        pm.recoveryInfo = recoveryInfo
+        return fmt.Errorf("snapshot integrity check failed: %v", err)
+    }
+
+    // Создаём базу данных если не существует
     if !pm.storage.ExistsDatabase(dbName) {
         if err := pm.storage.CreateDatabase(dbName); err != nil {
+            recoveryInfo.Error = err.Error()
+            pm.recoveryInfo = recoveryInfo
             return err
         }
     }
 
     db, err := pm.storage.GetDatabase(dbName)
     if err != nil {
+        recoveryInfo.Error = err.Error()
+        pm.recoveryInfo = recoveryInfo
         return err
     }
 
     // Восстанавливаем коллекции
+    restoredColls := 0
+    restoredDocs := int64(0)
+
     for collName, collDataRaw := range snapshot.Collections {
         collData, ok := collDataRaw.([]interface{})
         if !ok {
             continue
         }
 
-        if _, err := db.GetCollection(collName); err != nil {
-            if err := db.CreateCollection(collName); err != nil {
-                continue
+        // Удаляем существующую коллекцию если есть
+        if _, err := db.GetCollection(collName); err == nil {
+            db.DropCollection(collName)
         }
+
+        if err := db.CreateCollection(collName); err != nil {
+            if pm.logger != nil {
+                pm.logger.Error(fmt.Sprintf("Failed to create collection %s: %v", collName, err))
+            }
+            continue
         }
 
         coll, err := db.GetCollection(collName)
@@ -273,6 +611,8 @@ func (pm *PersistenceManager) LoadDatabase(dbName string) error {
             continue
         }
 
+        restoredColls++
+
         for _, docRaw := range collData {
             docMap, ok := docRaw.(map[string]interface{})
             if !ok {
@@ -294,26 +634,172 @@ func (pm *PersistenceManager) LoadDatabase(dbName string) error {
 
             if createdAt, ok := docMap["created_at"].(float64); ok {
                 doc.CreatedAt = int64(createdAt)
+            } else if createdAt, ok := docMap["created_at"].(int64); ok {
+                doc.CreatedAt = createdAt
             }
 
             if updatedAt, ok := docMap["updated_at"].(float64); ok {
                 doc.UpdatedAt = int64(updatedAt)
+            } else if updatedAt, ok := docMap["updated_at"].(int64); ok {
+                doc.UpdatedAt = updatedAt
+            }
+
+            if deletedAt, ok := docMap["deleted_at"].(float64); ok {
+                doc.DeletedAt = int64(deletedAt)
+            } else if deletedAt, ok := docMap["deleted_at"].(int64); ok {
+                doc.DeletedAt = deletedAt
             }
 
             if version, ok := docMap["version"].(float64); ok {
                 doc.Version = uint64(version)
+            } else if version, ok := docMap["version"].(uint64); ok {
+                doc.Version = version
             }
 
             if err := coll.Insert(doc); err != nil {
                 if pm.logger != nil {
-                    pm.logger.Error(fmt.Sprintf("Failed to restore document %s: %v", docID, err))
+                    pm.logger.Warn(fmt.Sprintf("Failed to restore document %s: %v", docID, err))
+                }
+                continue
+            }
+            restoredDocs++
+        }
+    }
+
+    recoveryInfo.RestoredDocs = restoredDocs
+    recoveryInfo.RestoredColls = restoredColls
+
+    // Применяем WAL записи после чекпоинта
+    appliedWalCount := 0
+    if pm.config.UseWalForCheckpoint && snapshot.WalLSN > 0 && pm.walReader != nil {
+        currentLSN, err := pm.walReader.GetCurrentIndex()
+        if err == nil && currentLSN > snapshot.WalLSN {
+            if pm.logger != nil {
+                pm.logger.Info(fmt.Sprintf("Database %s: Applying %d WAL entries after checkpoint (LSN %d -> %d)",
+                    dbName, currentLSN-snapshot.WalLSN, snapshot.WalLSN, currentLSN))
+            }
+
+            entries, err := pm.walReader.ReadSince(snapshot.WalLSN)
+            if err != nil {
+                if pm.logger != nil {
+                    pm.logger.Error(fmt.Sprintf("Failed to read WAL entries: %v", err))
+                }
+            } else {
+                for _, entry := range entries {
+                    if err := pm.applyWALEntry(db, entry); err != nil {
+                        if pm.logger != nil {
+                            pm.logger.Error(fmt.Sprintf("Failed to apply WAL entry %d: %v", entry.Index, err))
+                        }
+                        continue
+                    }
+                    appliedWalCount++
                 }
             }
         }
     }
 
+    recoveryInfo.AppliedWalCount = appliedWalCount
+    recoveryInfo.DurationMs = time.Since(startTime).Milliseconds()
+    recoveryInfo.Success = true
+
+    pm.recoveryInfo = recoveryInfo
+
     if pm.logger != nil {
-        pm.logger.Info(fmt.Sprintf("Loaded database %s from snapshot (version %d)", dbName, snapshot.Version))
+        pm.logger.Info(fmt.Sprintf("Loaded database %s from snapshot (version %d, WAL LSN: %d, applied %d WAL entries, %d docs, %d colls, duration: %dms)",
+            dbName, snapshot.Version, snapshot.WalLSN, appliedWalCount, restoredDocs, restoredColls, recoveryInfo.DurationMs))
+    }
+
+    return nil
+}
+
+// applyWALEntry применяет одну запись WAL к базе данных
+func (pm *PersistenceManager) applyWALEntry(db *Database, entry WALEntry) error {
+    var txRecord struct {
+        ID         uint64 `json:"id"`
+        State      int32  `json:"state"`
+        Timestamp  int64  `json:"timestamp"`
+        Operations []struct {
+            Type       string                 `json:"type"`
+            Database   string                 `json:"database"`
+            Collection string                 `json:"collection"`
+            DocumentID string                 `json:"document_id"`
+            Data       map[string]interface{} `json:"data"`
+            Version    uint64                 `json:"version"`
+        } `json:"operations"`
+    }
+
+    if err := json.Unmarshal(entry.Data, &txRecord); err != nil {
+        return err
+    }
+
+    // Применяем только закоммиченные транзакции
+    if txRecord.State != 1 { // TransactionCommitted
+        return nil
+    }
+
+    for _, op := range txRecord.Operations {
+        // Пропускаем операции для других баз данных
+        if op.Database != db.name {
+            continue
+        }
+
+        coll, err := db.GetCollection(op.Collection)
+        if err != nil {
+            continue
+        }
+
+        switch op.Type {
+        case "insert":
+            doc := NewDocumentWithID(op.DocumentID)
+            for k, v := range op.Data {
+                doc.SetField(k, v)
+            }
+            doc.Version = op.Version
+            if err := coll.Insert(doc); err != nil {
+                return err
+            }
+
+        case "update":
+            if err := coll.Update(op.DocumentID, op.Data); err != nil {
+                return err
+            }
+
+        case "delete":
+            if err := coll.Delete(op.DocumentID); err != nil {
+                return err
+            }
+        }
+    }
+
+    return nil
+}
+
+// verifySnapshot проверяет целостность снапшота
+func (pm *PersistenceManager) verifySnapshot(snapshot *DatabaseSnapshot) error {
+    if snapshot.Checksum == "" {
+        return fmt.Errorf("snapshot has no checksum")
+    }
+
+    // Пересоздаём данные без контрольной суммы для проверки
+    tempSnapshot := &DatabaseSnapshot{
+        Name:        snapshot.Name,
+        Collections: snapshot.Collections,
+        CreatedAt:   snapshot.CreatedAt,
+        Version:     snapshot.Version,
+        WalLSN:      snapshot.WalLSN,
+        WalSegment:  snapshot.WalSegment,
+    }
+
+    data, err := json.Marshal(tempSnapshot)
+    if err != nil {
+        return fmt.Errorf("failed to marshal for checksum: %v", err)
+    }
+
+    hash := sha256.Sum256(data)
+    checksum := hex.EncodeToString(hash[:])
+
+    if checksum != snapshot.Checksum {
+        return fmt.Errorf("checksum mismatch: expected %s, got %s", snapshot.Checksum[:8], checksum[:8])
     }
 
     return nil
@@ -329,8 +815,6 @@ func (pm *PersistenceManager) LoadAll() error {
     databases := make(map[string]bool)
     for _, file := range files {
         base := filepath.Base(file)
-        // Извлекаем имя базы данных из имени файла
-        // Формат: snapshot_{dbname}_{version}.json.gz
         parts := strings.Split(base, "_")
         if len(parts) >= 2 {
             dbName := parts[1]
@@ -370,7 +854,6 @@ func (pm *PersistenceManager) findLatestSnapshot(dbName string) (*DatabaseSnapsh
         return nil, nil
     }
 
-    // Сортируем по времени создания (новейшие первыми)
     sort.Slice(files, func(i, j int) bool {
         infoI, _ := os.Stat(files[i])
         infoJ, _ := os.Stat(files[j])
@@ -380,7 +863,6 @@ func (pm *PersistenceManager) findLatestSnapshot(dbName string) (*DatabaseSnapsh
         return infoI.ModTime().After(infoJ.ModTime())
     })
 
-    // Читаем последний снапшот
     data, err := os.ReadFile(files[0])
     if err != nil {
         return nil, err
@@ -445,7 +927,6 @@ func (pm *PersistenceManager) cleanupOldCheckpoints() {
         return
     }
 
-    // Сортируем по времени модификации
     sort.Slice(files, func(i, j int) bool {
         infoI, _ := os.Stat(files[i])
         infoJ, _ := os.Stat(files[j])
@@ -455,7 +936,6 @@ func (pm *PersistenceManager) cleanupOldCheckpoints() {
         return infoI.ModTime().Before(infoJ.ModTime())
     })
 
-    // Удаляем старые
     toDelete := files[:len(files)-pm.config.MaxCheckpoints]
     for _, f := range toDelete {
         os.Remove(f)
@@ -464,3 +944,37 @@ func (pm *PersistenceManager) cleanupOldCheckpoints() {
         }
     }
 }
+
+// GetLastCheckpointInfo возвращает информацию о последнем чекпоинте
+func (pm *PersistenceManager) GetLastCheckpointInfo() map[string]interface{} {
+    pm.mu.RLock()
+    defer pm.mu.RUnlock()
+
+    info := map[string]interface{}{
+        "last_checkpoint_time":      pm.lastCheckpoint,
+        "last_checkpoint_time_str":  time.UnixMilli(pm.lastCheckpoint).Format("2006-01-02 15:04:05.000"),
+        "checkpoint_id":             pm.checkpointID.Load(),
+        "wal_path":                  pm.config.WalPath,
+        "use_wal":                   pm.config.UseWalForCheckpoint,
+        "atomic_writes":             pm.config.AtomicWrites,
+        "is_restoring":              pm.isRestoring.Load(),
+    }
+
+    if pm.recoveryInfo != nil {
+        info["last_recovery"] = pm.recoveryInfo
+    }
+
+    return info
+}
+
+// GetRecoveryInfo возвращает информацию о последнем восстановлении
+func (pm *PersistenceManager) GetRecoveryInfo() *CheckpointRecoveryInfo {
+    pm.mu.RLock()
+    defer pm.mu.RUnlock()
+    return pm.recoveryInfo
+}
+
+// IsRestoring возвращает статус восстановления
+func (pm *PersistenceManager) IsRestoring() bool {
+    return pm.isRestoring.Load()
+}