diff --git a/internal/backup/backup_scheduler.go b/internal/backup/backup_scheduler.go new file mode 100644 index 0000000..9870395 --- /dev/null +++ b/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") +} \ No newline at end of file