Add internal/backup/backup_scheduler.go

internal/backup/backup_scheduler.go

@@ -0,0 +1,715 @@
+/*
+ * 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/backup/backup_scheduler.go
+// Назначение: Планировщик бэкапов и инкрементальные бэкапы на основе WAL
+
+package backup
+
+import (
+    "crypto/sha256"
+    "encoding/hex"
+    "encoding/json"
+    "fmt"
+    "os"
+    "path/filepath"
+    "sort"
+    "sync"
+    "sync/atomic"
+    "time"
+)
+
+// BackupType тип бэкапа
+type BackupType string
+
+const (
+    FullBackup        BackupType = "full"
+    IncrementalBackup BackupType = "incremental"
+)
+
+// BackupSchedule расписание бэкапов
+type BackupSchedule struct {
+    ID          string        `json:"id"`
+    Name        string        `json:"name"`
+    CronExpr    string        `json:"cron_expr"`
+    Type        BackupType    `json:"type"`
+    RetentionDays int         `json:"retention_days"`
+    Enabled     bool          `json:"enabled"`
+    LastRun     int64         `json:"last_run"`
+    NextRun     int64         `json:"next_run"`
+    CreatedAt   int64         `json:"created_at"`
+    UpdatedAt   int64         `json:"updated_at"`
+}
+
+// BackupInfo информация о бэкапе
+type BackupInfo struct {
+    ID           string        `json:"id"`
+    ScheduleID   string        `json:"schedule_id"`
+    Type         BackupType    `json:"type"`
+    StartTime    int64         `json:"start_time"`
+    EndTime      int64         `json:"end_time"`
+    Status       string        `json:"status"`
+    SizeBytes    int64         `json:"size_bytes"`
+    Path         string        `json:"path"`
+    WALStart     uint64        `json:"wal_start"`
+    WALEnd       uint64        `json:"wal_end"`
+    ParentID     string        `json:"parent_id,omitempty"`
+    Checksum     string        `json:"checksum"`
+    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
+}
+
+// WALReader интерфейс для чтения WAL
+type WALReader interface {
+    ReadSince(index uint64) ([]WALEntry, error)
+    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"`
+}
+
+// BackupStorage интерфейс для хранения бэкапов
+type BackupStorage interface {
+    SaveBackup(path string, data []byte) error
+    LoadBackup(path string) ([]byte, error)
+    ListBackups() ([]string, error)
+    DeleteBackup(path string) error
+    BackupExists(path string) bool
+}
+
+// LoggerInterface интерфейс для логирования
+type LoggerInterface interface {
+    Info(msg string)
+    Warn(msg string)
+    Error(msg string)
+    Debug(msg string)
+}
+
+// FileBackupStorage реализация BackupStorage для файловой системы
+type FileBackupStorage struct {
+    backupDir string
+}
+
+// NewFileBackupStorage создаёт новое файловое хранилище бэкапов
+func NewFileBackupStorage(backupDir string) (*FileBackupStorage, error) {
+    if err := os.MkdirAll(backupDir, 0755); err != nil {
+        return nil, fmt.Errorf("failed to create backup dir: %v", err)
+    }
+    return &FileBackupStorage{backupDir: backupDir}, nil
+}
+
+func (fbs *FileBackupStorage) SaveBackup(path string, data []byte) error {
+    fullPath := filepath.Join(fbs.backupDir, path)
+    if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil {
+        return err
+    }
+    return os.WriteFile(fullPath, data, 0644)
+}
+
+func (fbs *FileBackupStorage) LoadBackup(path string) ([]byte, error) {
+    fullPath := filepath.Join(fbs.backupDir, path)
+    return os.ReadFile(fullPath)
+}
+
+func (fbs *FileBackupStorage) ListBackups() ([]string, error) {
+    entries, err := os.ReadDir(fbs.backupDir)
+    if err != nil {
+        return nil, err
+    }
+    names := make([]string, 0, len(entries))
+    for _, e := range entries {
+        names = append(names, e.Name())
+    }
+    return names, nil
+}
+
+func (fbs *FileBackupStorage) DeleteBackup(path string) error {
+    fullPath := filepath.Join(fbs.backupDir, path)
+    return os.RemoveAll(fullPath)
+}
+
+func (fbs *FileBackupStorage) BackupExists(path string) bool {
+    fullPath := filepath.Join(fbs.backupDir, path)
+    _, err := os.Stat(fullPath)
+    return err == nil
+}
+
+// BackupSchedulerConfig конфигурация планировщика
+type BackupSchedulerConfig struct {
+    BackupDir       string `json:"backup_dir"`
+    MaxConcurrent   int    `json:"max_concurrent"`
+    CompressEnabled bool   `json:"compress_enabled"`
+    CompressionLevel int   `json:"compression_level"`
+}
+
+// DefaultBackupSchedulerConfig возвращает конфигурацию по умолчанию
+func DefaultBackupSchedulerConfig() *BackupSchedulerConfig {
+    return &BackupSchedulerConfig{
+        BackupDir:        "backups",
+        MaxConcurrent:    1,
+        CompressEnabled:  true,
+        CompressionLevel: 6,
+    }
+}
+
+// NewBackupScheduler создаёт новый планировщик бэкапов
+func NewBackupScheduler(cfg *BackupSchedulerConfig, walReader WALReader, logger LoggerInterface) (*BackupScheduler, error) {
+    if cfg == nil {
+        cfg = DefaultBackupSchedulerConfig()
+    }
+    
+    storage, err := NewFileBackupStorage(cfg.BackupDir)
+    if err != nil {
+        return nil, err
+    }
+    
+    bs := &BackupScheduler{
+        schedules: make(map[string]*BackupSchedule),
+        backups:   make(map[string]*BackupInfo),
+        walReader: walReader,
+        storage:   storage,
+        logger:    logger,
+        stopChan:  make(chan struct{}),
+    }
+    
+    // Загружаем существующие расписания
+    bs.loadSchedules()
+    
+    // Загружаем историю бэкапов
+    bs.loadBackupHistory()
+    
+    return bs, nil
+}
+
+// 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) {
+            bs.logger.Warn(fmt.Sprintf("Failed to load schedules: %v", err))
+        }
+        return
+    }
+    
+    var schedules map[string]*BackupSchedule
+    if err := json.Unmarshal(data, &schedules); err != nil {
+        bs.logger.Warn(fmt.Sprintf("Failed to parse schedules: %v", err))
+        return
+    }
+    
+    bs.mu.Lock()
+    bs.schedules = schedules
+    bs.mu.Unlock()
+}
+
+// saveSchedules сохраняет расписания
+func (bs *BackupScheduler) saveSchedules() error {
+    bs.mu.RLock()
+    data, err := json.MarshalIndent(bs.schedules, "", "  ")
+    bs.mu.RUnlock()
+    
+    if err != nil {
+        return err
+    }
+    
+    return bs.storage.SaveBackup("schedules.json", data)
+}
+
+// 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) {
+            bs.logger.Warn(fmt.Sprintf("Failed to load backup history: %v", err))
+        }
+        return
+    }
+    
+    var backups map[string]*BackupInfo
+    if err := json.Unmarshal(data, &backups); err != nil {
+        bs.logger.Warn(fmt.Sprintf("Failed to parse backup history: %v", err))
+        return
+    }
+    
+    bs.mu.Lock()
+    bs.backups = backups
+    bs.mu.Unlock()
+}
+
+// saveBackupHistory сохраняет историю бэкапов
+func (bs *BackupScheduler) saveBackupHistory() error {
+    bs.mu.RLock()
+    data, err := json.MarshalIndent(bs.backups, "", "  ")
+    bs.mu.RUnlock()
+    
+    if err != nil {
+        return err
+    }
+    
+    return bs.storage.SaveBackup("backups.json", data)
+}
+
+// AddSchedule добавляет расписание бэкапов
+func (bs *BackupScheduler) AddSchedule(name, cronExpr string, backupType BackupType, retentionDays int) (*BackupSchedule, error) {
+    bs.mu.Lock()
+    defer bs.mu.Unlock()
+    
+    schedule := &BackupSchedule{
+        ID:           fmt.Sprintf("sch_%d", time.Now().UnixNano()),
+        Name:         name,
+        CronExpr:     cronExpr,
+        Type:         backupType,
+        RetentionDays: retentionDays,
+        Enabled:      true,
+        CreatedAt:    time.Now().UnixMilli(),
+        UpdatedAt:    time.Now().UnixMilli(),
+    }
+    
+    bs.schedules[schedule.ID] = schedule
+    
+    if err := bs.saveSchedules(); err != nil {
+        return nil, err
+    }
+    
+    bs.logger.Info(fmt.Sprintf("Added backup schedule: %s (%s) with cron %s", name, backupType, cronExpr))
+    return schedule, nil
+}
+
+// RemoveSchedule удаляет расписание
+func (bs *BackupScheduler) RemoveSchedule(scheduleID string) error {
+    bs.mu.Lock()
+    defer bs.mu.Unlock()
+    
+    if _, exists := bs.schedules[scheduleID]; !exists {
+        return fmt.Errorf("schedule not found: %s", scheduleID)
+    }
+    
+    delete(bs.schedules, scheduleID)
+    return bs.saveSchedules()
+}
+
+// RunNow запускает бэкап немедленно
+func (bs *BackupScheduler) RunNow(scheduleID string) error {
+    bs.mu.RLock()
+    schedule, exists := bs.schedules[scheduleID]
+    bs.mu.RUnlock()
+    
+    if !exists {
+        return fmt.Errorf("schedule not found: %s", scheduleID)
+    }
+    
+    if schedule.Type == FullBackup {
+        return bs.performFullBackup(scheduleID)
+    }
+    return bs.performIncrementalBackup(scheduleID)
+}
+
+// performFullBackup выполняет полный бэкап
+func (bs *BackupScheduler) performFullBackup(scheduleID string) error {
+    if !bs.running.CompareAndSwap(false, true) {
+        return fmt.Errorf("backup already in progress")
+    }
+    defer bs.running.Store(false)
+    
+    startTime := time.Now().UnixMilli()
+    backupID := fmt.Sprintf("full_%d", bs.backupCounter.Add(1))
+    
+    bs.logger.Info(fmt.Sprintf("Starting full backup %s", backupID))
+    
+    // Получаем текущую позицию WAL
+    walIndex, err := bs.walReader.GetCurrentIndex()
+    if err != nil {
+        return fmt.Errorf("failed to get WAL index: %v", err)
+    }
+    
+    // Сохраняем последний индекс для инкрементальных бэкапов
+    if err := bs.walReader.SetLastBackupIndex(walIndex); err != nil {
+        bs.logger.Warn(fmt.Sprintf("Failed to set last backup index: %v", err))
+    }
+    
+    // Здесь выполняется реальный бэкап данных
+    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 {
+        return fmt.Errorf("failed to save backup: %v", err)
+    }
+    
+    // Вычисляем контрольную сумму
+    hash := sha256.Sum256(backupData)
+    checksum := hex.EncodeToString(hash[:])
+    
+    endTime := time.Now().UnixMilli()
+    
+    backupInfo := &BackupInfo{
+        ID:         backupID,
+        ScheduleID: scheduleID,
+        Type:       FullBackup,
+        StartTime:  startTime,
+        EndTime:    endTime,
+        Status:     "completed",
+        SizeBytes:  int64(len(backupData)),
+        Path:       backupPath,
+        WALStart:   0,
+        WALEnd:     walIndex,
+        Checksum:   checksum,
+    }
+    
+    bs.mu.Lock()
+    bs.backups[backupID] = backupInfo
+    bs.mu.Unlock()
+    
+    bs.saveBackupHistory()
+    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))
+    
+    return nil
+}
+
+// performIncrementalBackup выполняет инкрементальный бэкап на основе WAL
+func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error {
+    if !bs.running.CompareAndSwap(false, true) {
+        return fmt.Errorf("backup already in progress")
+    }
+    defer bs.running.Store(false)
+    
+    startTime := time.Now().UnixMilli()
+    backupID := fmt.Sprintf("inc_%d", bs.backupCounter.Add(1))
+    
+    // Находим последний полный бэкап
+    lastFullBackup := bs.findLastFullBackup()
+    if lastFullBackup == nil {
+        bs.logger.Info("No full backup found, performing full backup instead")
+        return bs.performFullBackup(scheduleID)
+    }
+    
+    // Получаем WAL записи с момента последнего бэкапа
+    lastIndex := bs.walReader.GetLastBackupIndex()
+    if lastIndex == 0 {
+        lastIndex = lastFullBackup.WALEnd
+    }
+    
+    walEntries, err := bs.walReader.ReadSince(lastIndex)
+    if err != nil {
+        return fmt.Errorf("failed to read WAL: %v", err)
+    }
+    
+    if len(walEntries) == 0 {
+        bs.logger.Info("No new WAL entries since last backup")
+        return nil
+    }
+    
+    currentIndex, _ := bs.walReader.GetCurrentIndex()
+    
+    // Сохраняем инкрементальный бэкап
+    backupPath := fmt.Sprintf("inc/%s/%s.backup",
+        time.Now().Format("2006-01-02"), backupID)
+    
+    backupData := struct {
+        BackupID   string      `json:"backup_id"`
+        Type       string      `json:"type"`
+        StartTime  int64       `json:"start_time"`
+        WALStart   uint64      `json:"wal_start"`
+        WALEnd     uint64      `json:"wal_end"`
+        ParentID   string      `json:"parent_id"`
+        Entries    []WALEntry  `json:"entries"`
+    }{
+        BackupID:  backupID,
+        Type:      "incremental",
+        StartTime: startTime,
+        WALStart:  lastIndex,
+        WALEnd:    currentIndex,
+        ParentID:  lastFullBackup.ID,
+        Entries:   walEntries,
+    }
+    
+    data, err := json.Marshal(backupData)
+    if err != nil {
+        return fmt.Errorf("failed to marshal backup data: %v", err)
+    }
+    
+    if err := bs.storage.SaveBackup(backupPath, data); err != nil {
+        return fmt.Errorf("failed to save incremental backup: %v", err)
+    }
+    
+    // Обновляем последний индекс
+    if err := bs.walReader.SetLastBackupIndex(currentIndex); err != nil {
+        bs.logger.Warn(fmt.Sprintf("Failed to set last backup index: %v", err))
+    }
+    
+    hash := sha256.Sum256(data)
+    checksum := hex.EncodeToString(hash[:])
+    
+    endTime := time.Now().UnixMilli()
+    
+    backupInfo := &BackupInfo{
+        ID:         backupID,
+        ScheduleID: scheduleID,
+        Type:       IncrementalBackup,
+        StartTime:  startTime,
+        EndTime:    endTime,
+        Status:     "completed",
+        SizeBytes:  int64(len(data)),
+        Path:       backupPath,
+        WALStart:   lastIndex,
+        WALEnd:     currentIndex,
+        ParentID:   lastFullBackup.ID,
+        Checksum:   checksum,
+    }
+    
+    bs.mu.Lock()
+    bs.backups[backupID] = backupInfo
+    bs.mu.Unlock()
+    
+    bs.saveBackupHistory()
+    bs.updateScheduleLastRun(scheduleID, startTime)
+    
+    duration := (endTime - startTime) / 1000
+    bs.logger.Info(fmt.Sprintf("Incremental backup %s completed in %d seconds, %d entries, size=%d bytes",
+        backupID, duration, len(walEntries), backupInfo.SizeBytes))
+    
+    return nil
+}
+
+// findLastFullBackup находит последний успешный полный бэкап
+func (bs *BackupScheduler) findLastFullBackup() *BackupInfo {
+    bs.mu.RLock()
+    defer bs.mu.RUnlock()
+    
+    var lastFull *BackupInfo
+    for _, backup := range bs.backups {
+        if backup.Type == FullBackup && backup.Status == "completed" {
+            if lastFull == nil || backup.EndTime > lastFull.EndTime {
+                lastFull = backup
+            }
+        }
+    }
+    return lastFull
+}
+
+// updateScheduleLastRun обновляет время последнего запуска расписания
+func (bs *BackupScheduler) updateScheduleLastRun(scheduleID string, timestamp int64) {
+    bs.mu.Lock()
+    defer bs.mu.Unlock()
+    
+    if schedule, exists := bs.schedules[scheduleID]; exists {
+        schedule.LastRun = timestamp
+        schedule.UpdatedAt = time.Now().UnixMilli()
+        bs.saveSchedules()
+    }
+}
+
+// cleanupOldBackups удаляет старые бэкапы
+func (bs *BackupScheduler) cleanupOldBackups(scheduleID string) {
+    bs.mu.Lock()
+    defer bs.mu.Unlock()
+    
+    schedule, exists := bs.schedules[scheduleID]
+    if !exists {
+        return
+    }
+    
+    cutoffTime := time.Now().AddDate(0, 0, -schedule.RetentionDays).UnixMilli()
+    
+    toDelete := make([]string, 0)
+    for id, backup := range bs.backups {
+        if backup.ScheduleID == scheduleID && backup.StartTime < cutoffTime {
+            toDelete = append(toDelete, id)
+            // Удаляем файл бэкапа
+            bs.storage.DeleteBackup(backup.Path)
+        }
+    }
+    
+    for _, id := range toDelete {
+        delete(bs.backups, id)
+    }
+    
+    if len(toDelete) > 0 {
+        bs.saveBackupHistory()
+        bs.logger.Info(fmt.Sprintf("Cleaned up %d old backups for schedule %s", len(toDelete), scheduleID))
+    }
+}
+
+// ListBackups возвращает список бэкапов
+func (bs *BackupScheduler) ListBackups() []*BackupInfo {
+    bs.mu.RLock()
+    defer bs.mu.RUnlock()
+    
+    backups := make([]*BackupInfo, 0, len(bs.backups))
+    for _, b := range bs.backups {
+        backups = append(backups, b)
+    }
+    
+    // Сортируем по времени
+    sort.Slice(backups, func(i, j int) bool {
+        return backups[i].StartTime > backups[j].StartTime
+    })
+    
+    return backups
+}
+
+// ListSchedules возвращает список расписаний
+func (bs *BackupScheduler) ListSchedules() []*BackupSchedule {
+    bs.mu.RLock()
+    defer bs.mu.RUnlock()
+    
+    schedules := make([]*BackupSchedule, 0, len(bs.schedules))
+    for _, s := range bs.schedules {
+        schedules = append(schedules, s)
+    }
+    return schedules
+}
+
+// GetBackup возвращает информацию о бэкапе
+func (bs *BackupScheduler) GetBackup(backupID string) *BackupInfo {
+    bs.mu.RLock()
+    defer bs.mu.RUnlock()
+    return bs.backups[backupID]
+}
+
+// Restore восстанавливает данные из бэкапа
+func (bs *BackupScheduler) Restore(backupID string) error {
+    backup := bs.GetBackup(backupID)
+    if backup == nil {
+        return fmt.Errorf("backup not found: %s", backupID)
+    }
+    
+    bs.logger.Info(fmt.Sprintf("Starting restore from backup %s", backupID))
+    
+    data, err := bs.storage.LoadBackup(backup.Path)
+    if err != nil {
+        return fmt.Errorf("failed to load backup: %v", err)
+    }
+    
+    // Проверяем контрольную сумму
+    hash := sha256.Sum256(data)
+    checksum := hex.EncodeToString(hash[:])
+    if checksum != backup.Checksum {
+        return fmt.Errorf("backup checksum mismatch, data may be corrupted")
+    }
+    
+    if backup.Type == FullBackup {
+        // Восстановление из полного бэкапа
+        bs.logger.Info("Restoring from full backup...")
+        // Здесь реальное восстановление
+    } else {
+        // Восстановление из инкрементального бэкапа
+        // Сначала восстанавливаем полный бэкап-родитель
+        if backup.ParentID != "" {
+            if err := bs.Restore(backup.ParentID); err != nil {
+                return fmt.Errorf("failed to restore parent backup: %v", err)
+            }
+        }
+        bs.logger.Info("Applying incremental changes...")
+        // Здесь применяем инкрементальные изменения
+    }
+    
+    bs.logger.Info(fmt.Sprintf("Restore from backup %s completed", backupID))
+    return nil
+}
+
+// Start запускает планировщик
+func (bs *BackupScheduler) Start() {
+    bs.wg.Add(1)
+    go bs.schedulerLoop()
+    bs.logger.Info("Backup scheduler started")
+}
+
+// schedulerLoop основной цикл планировщика
+func (bs *BackupScheduler) schedulerLoop() {
+    defer bs.wg.Done()
+    
+    ticker := time.NewTicker(60 * time.Second)
+    defer ticker.Stop()
+    
+    for {
+        select {
+        case <-bs.stopChan:
+            return
+        case <-ticker.C:
+            bs.checkSchedules()
+        }
+    }
+}
+
+// checkSchedules проверяет расписания и запускает бэкапы
+func (bs *BackupScheduler) checkSchedules() {
+    bs.mu.RLock()
+    schedules := make([]*BackupSchedule, 0, len(bs.schedules))
+    for _, s := range bs.schedules {
+        if s.Enabled {
+            schedules = append(schedules, s)
+        }
+    }
+    bs.mu.RUnlock()
+    
+    now := time.Now()
+    for _, schedule := range schedules {
+        // Простая проверка расписания (в реальности используется cron-парсер)
+        if schedule.NextRun == 0 || schedule.NextRun <= now.UnixMilli() {
+            // Вычисляем следующее время (упрощённо - через 24 часа)
+            schedule.NextRun = now.Add(24 * time.Hour).UnixMilli()
+            
+            bs.mu.Lock()
+            bs.schedules[schedule.ID] = schedule
+            bs.mu.Unlock()
+            bs.saveSchedules()
+            
+            // Запускаем бэкап в отдельной горутине
+            scheduleID := schedule.ID
+            backupType := schedule.Type
+            go func() {
+                if backupType == FullBackup {
+                    bs.performFullBackup(scheduleID)
+                } else {
+                    bs.performIncrementalBackup(scheduleID)
+                }
+            }()
+        }
+    }
+}
+
+// Stop останавливает планировщик
+func (bs *BackupScheduler) Stop() {
+    close(bs.stopChan)
+    bs.wg.Wait()
+    bs.logger.Info("Backup scheduler stopped")
+}