futriix/internal/backup/backup_scheduler.go

/*
 * Copyright 2026 Safronov Grigorii
 *
 * Licensed under the CDDL, Version 1.0 (the "License");
 * you may not use this file except in compliance with the License.
 *
 * You may obtain a copy of the License at
 * https://opensource.org/licenses/CDDL-1.0
 */

package backup

import (
    "bytes"
    "compress/gzip"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "fmt"
    "io"
    "os"
    "path/filepath"
    "sort"
    "sync"
    "sync/atomic"
    "time"

    "futriis/internal/config"
    "futriis/internal/storage"
)

type BackupType string

const (
    FullBackup        BackupType = "full"
    IncrementalBackup BackupType = "incremental"
)

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"`
}

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"`
    SagaCount  int        `json:"saga_count,omitempty"`
    SagaID     string     `json:"saga_id,omitempty"`
}

type WALEntry struct {
    Index     uint64 `json:"index"`
    Term      uint64 `json:"term"`
    Type      string `json:"type"`
    Data      []byte `json:"data"`
    Timestamp int64  `json:"timestamp"`
}

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)
}

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
}

type LoggerInterface interface {
    Info(msg string)
    Warn(msg string)
    Error(msg string)
    Debug(msg string)
}

type FileBackupStorage struct {
    backupDir     string
    compressLevel int
    mu            sync.Mutex
}

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,
        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
    }

    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)
    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) {
    entries, err := os.ReadDir(fbs.backupDir)
    if err != nil {
        return nil, err
    }
    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)
    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)
    compressedPath := fullPath + ".gz"
    _, err1 := os.Stat(fullPath)
    _, err2 := os.Stat(compressedPath)
    return err1 == nil || err2 == nil
}

func (fbs *FileBackupStorage) GetTransactionManager() *storage.TransactionManager {
    return storage.GetTransactionManager()
}

type WALReaderImpl struct {
    walManager      interface{}
    lastBackupIndex atomic.Uint64
    segmentsDir     string
    mu              sync.RWMutex
    logger          LoggerInterface
}

func NewWALReaderImpl(walManager interface{}, segmentsDir string, logger LoggerInterface) *WALReaderImpl {
    wr := &WALReaderImpl{
        walManager:  walManager,
        segmentsDir: segmentsDir,
        logger:      logger,
    }
    wr.loadLastBackupIndex()
    return wr
}

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
}

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
}

func (wr *WALReaderImpl) GetLastBackupIndex() uint64 {
    return wr.lastBackupIndex.Load()
}

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)
}

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
}

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
}

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)
    }
}

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        *config.BackupConfig
    sagaManager   *storage.SagaManager
}

func NewBackupScheduler(cfg *config.BackupConfig, walReader WALReader, logger LoggerInterface) (*BackupScheduler, error) {
    if cfg == nil {
        return nil, fmt.Errorf("backup config is required")
    }

    compressLevel := 6
    if cfg.CompressEnabled {
        compressLevel = 6
    } else {
        compressLevel = 0
    }

    storage, err := NewFileBackupStorage(cfg.BackupDir, compressLevel)
    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{}),
        config:      cfg,
        sagaManager: storage.NewSagaManager(logger),
    }

    bs.loadSchedules()
    bs.loadBackupHistory()

    return bs, nil
}

func (bs *BackupScheduler) GetSagaManager() *storage.SagaManager {
    return bs.sagaManager
}

func (bs *BackupScheduler) ReloadConfig(cfg *config.BackupConfig) {
    bs.mu.Lock()
    defer bs.mu.Unlock()

    oldBackupDir := bs.config.BackupDir
    oldCompressEnabled := bs.config.CompressEnabled
    oldIncludeSagaState := bs.config.IncludeSagaState
    oldEnableIncremental := bs.config.EnableIncremental

    bs.config = cfg

    // Если изменилась директория или сжатие - пересоздаём хранилище
    if oldBackupDir != cfg.BackupDir || oldCompressEnabled != cfg.CompressEnabled {
        compressLevel := 6
        if cfg.CompressEnabled {
            compressLevel = 6
        } else {
            compressLevel = 0
        }
        newStorage, err := NewFileBackupStorage(cfg.BackupDir, compressLevel)
        if err != nil {
            if bs.logger != nil {
                bs.logger.Error(fmt.Sprintf("Failed to reload backup storage: %v", err))
            }
            return
        }
        bs.storage = newStorage
        // Перезагружаем данные из нового хранилища
        bs.loadSchedules()
        bs.loadBackupHistory()
    }

    if bs.logger != nil {
        bs.logger.Info(fmt.Sprintf("Backup scheduler configuration reloaded: enabled=%v, backup_dir=%s, compress=%v, incremental=%v, saga_state=%v",
            cfg.Enabled, cfg.BackupDir, cfg.CompressEnabled, cfg.EnableIncremental, cfg.IncludeSagaState))
    }
}

func (bs *BackupScheduler) loadSchedules() {
    data, err := bs.storage.LoadBackup("schedules.json")
    if err != nil {
        if !os.IsNotExist(err) && bs.logger != nil {
            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 {
        if bs.logger != nil {
            bs.logger.Warn(fmt.Sprintf("Failed to parse schedules: %v", err))
        }
        return
    }

    bs.mu.Lock()
    bs.schedules = schedules
    bs.mu.Unlock()
}

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)
}

func (bs *BackupScheduler) loadBackupHistory() {
    data, err := bs.storage.LoadBackup("backups.json")
    if err != nil {
        if !os.IsNotExist(err) && bs.logger != nil {
            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 {
        if bs.logger != nil {
            bs.logger.Warn(fmt.Sprintf("Failed to parse backup history: %v", err))
        }
        return
    }

    bs.mu.Lock()
    bs.backups = backups
    bs.mu.Unlock()
}

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)
}

func (bs *BackupScheduler) AddSchedule(name, cronExpr string, backupType BackupType, retentionDays int) (*BackupSchedule, error) {
    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,
        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, 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
}

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)

    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
}

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 !bs.config.EnableIncremental && schedule.Type == IncrementalBackup {
        return fmt.Errorf("incremental backups are disabled in configuration")
    }

    if schedule.Type == FullBackup {
        return bs.performFullBackup(scheduleID)
    }
    return bs.performIncrementalBackup(scheduleID)
}

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_%d", startTime, bs.backupCounter.Add(1))

    if bs.logger != nil {
        bs.logger.Info(fmt.Sprintf("Starting full backup %s", backupID))
    }

    walIndex, err := bs.walReader.GetCurrentIndex()
    if err != nil {
        return fmt.Errorf("failed to get WAL index: %v", err)
    }

    var activeSagas []*storage.SagaTransaction
    var sagaCount int
    if bs.sagaManager != nil && bs.config.IncludeSagaState {
        activeSagas = bs.sagaManager.GetActiveSagas()
        sagaCount = len(activeSagas)
        if sagaCount > 0 && bs.logger != nil {
            bs.logger.Info(fmt.Sprintf("Found %d active SAGA transactions, including in backup", sagaCount))
        }
    }

    if err := bs.walReader.SetLastBackupIndex(walIndex); err != nil {
        if bs.logger != nil {
            bs.logger.Warn(fmt.Sprintf("Failed to set last backup index: %v", err))
        }
    }

    walEntries, err := bs.walReader.ReadSince(0)
    if err != nil {
        if bs.logger != nil {
            bs.logger.Warn(fmt.Sprintf("Failed to read WAL for full backup: %v", err))
        }
        walEntries = []WALEntry{}
    }

    sagaState := make(map[string]interface{})
    if sagaCount > 0 && bs.config.IncludeSagaState {
        sagaState["active_count"] = sagaCount
        sagaState["timestamp"] = time.Now().UnixMilli()
        sagaIDs := make([]string, 0, sagaCount)
        sagaStatuses := make([]string, 0, sagaCount)
        for _, saga := range activeSagas {
            sagaIDs = append(sagaIDs, saga.ID)
            sagaStatuses = append(sagaStatuses, saga.Status)
        }
        sagaState["saga_ids"] = sagaIDs
        sagaState["saga_statuses"] = sagaStatuses
    }

    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"`
        SagaState  interface{} `json:"saga_state,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),
            "saga_active_count":  sagaCount,
            "include_saga_state": bs.config.IncludeSagaState,
            "backup_config": map[string]interface{}{
                "backup_dir":         bs.config.BackupDir,
                "compress_enabled":   bs.config.CompressEnabled,
                "enable_incremental": bs.config.EnableIncremental,
                "retention_days":     bs.config.RetentionDays,
            },
        },
        SagaState: sagaState,
    }

    data, err := json.Marshal(backupData)
    if err != nil {
        return fmt.Errorf("failed to marshal backup data: %v", err)
    }

    backupPath := fmt.Sprintf("full/%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 backup: %v", err)
    }

    hash := sha256.Sum256(data)
    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(data)),
        Path:       backupPath,
        WALStart:   0,
        WALEnd:     walIndex,
        Checksum:   checksum,
        SagaCount:  sagaCount,
    }

    bs.mu.Lock()
    bs.backups[backupID] = backupInfo
    bs.mu.Unlock()

    if err := bs.saveBackupHistory(); err != nil {
        if bs.logger != nil {
            bs.logger.Warn(fmt.Sprintf("Failed to save backup history: %v", err))
        }
    }

    bs.updateScheduleLastRun(scheduleID, startTime)
    bs.cleanupOldBackups(scheduleID)

    duration := (endTime - startTime) / 1000
    if bs.logger != nil {
        bs.logger.Info(fmt.Sprintf("Full backup %s completed in %d seconds, size=%d bytes, entries=%d, active_sagas=%d",
            backupID, duration, backupInfo.SizeBytes, len(walEntries), sagaCount))
    }

    return nil
}

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_%d", startTime, bs.backupCounter.Add(1))

    sagaID := fmt.Sprintf("backup_saga_%s", backupID)
    saga := bs.sagaManager.BeginSaga(sagaID)

    saga.AddStep("validate_parent", func() error {
        lastFull := bs.findLastFullBackup()
        if lastFull == nil {
            return fmt.Errorf("no full backup found")
        }
        if !bs.storage.BackupExists(lastFull.Path) {
            return fmt.Errorf("parent backup file missing: %s", lastFull.ID)
        }
        return nil
    }, func() error {
        if bs.logger != nil {
            bs.logger.Warn("Compensating: parent backup validation failed")
        }
        return nil
    }, map[string]interface{}{
        "step": "validate_parent",
    })

    saga.AddStep("read_wal", func() error {
        lastIndex := bs.walReader.GetLastBackupIndex()
        if lastIndex == 0 {
            lastFull := bs.findLastFullBackup()
            if lastFull != nil {
                lastIndex = lastFull.WALEnd
            }
        }

        entries, err := bs.walReader.ReadSince(lastIndex)
        if err != nil {
            return fmt.Errorf("failed to read WAL: %v", err)
        }

        if len(entries) == 0 {
            if bs.logger != nil {
                bs.logger.Info("No new WAL entries since last backup")
            }
            return nil
        }

        saga.SetData("wal_entries", entries)
        saga.SetData("last_index", lastIndex)
        return nil
    }, func() error {
        if bs.logger != nil {
            bs.logger.Warn("Compensating: WAL read failed")
        }
        return nil
    }, map[string]interface{}{
        "step": "read_wal",
    })

    saga.AddStep("get_saga_state", func() error {
        if bs.sagaManager != nil && bs.config.IncludeSagaState {
            activeSagas := bs.sagaManager.GetActiveSagas()
            saga.SetData("active_sagas", activeSagas)
            saga.SetData("saga_count", len(activeSagas))
            if len(activeSagas) > 0 && bs.logger != nil {
                bs.logger.Info(fmt.Sprintf("Found %d active SAGA transactions in incremental backup", len(activeSagas)))
            }
        }
        return nil
    }, func() error {
        if bs.logger != nil {
            bs.logger.Warn("Compensating: SAGA state retrieval failed")
        }
        return nil
    }, map[string]interface{}{
        "step": "get_saga_state",
    })

    saga.AddStep("create_backup_file", func() error {
        entriesVal, _ := saga.GetData("wal_entries")
        entries, _ := entriesVal.([]WALEntry)
        lastIndexVal, _ := saga.GetData("last_index")
        lastIndex, _ := lastIndexVal.(uint64)
        sagaCountVal, _ := saga.GetData("saga_count")
        sagaCount, _ := sagaCountVal.(int)
        activeSagasVal, _ := saga.GetData("active_sagas")
        activeSagas, _ := activeSagasVal.([]*storage.SagaTransaction)

        currentIndex, err := bs.walReader.GetCurrentIndex()
        if err != nil {
            return fmt.Errorf("failed to get current WAL index: %v", err)
        }

        lastFull := bs.findLastFullBackup()
        if lastFull == nil {
            return fmt.Errorf("no full backup found")
        }

        sagaState := make(map[string]interface{})
        if sagaCount > 0 && bs.config.IncludeSagaState {
            sagaState["active_count"] = sagaCount
            sagaState["timestamp"] = time.Now().UnixMilli()
            if activeSagas != nil {
                sagaIDs := make([]string, 0, len(activeSagas))
                sagaStatuses := make([]string, 0, len(activeSagas))
                for _, s := range activeSagas {
                    sagaIDs = append(sagaIDs, s.ID)
                    sagaStatuses = append(sagaStatuses, s.Status)
                }
                sagaState["saga_ids"] = sagaIDs
                sagaState["saga_statuses"] = sagaStatuses
            }
        }

        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"`
            Metadata   interface{} `json:"metadata,omitempty"`
            SagaState  interface{} `json:"saga_state,omitempty"`
            SagaID     string      `json:"saga_id"`
        }{
            BackupID:  backupID,
            Type:      "incremental",
            StartTime: startTime,
            WALStart:  lastIndex,
            WALEnd:    currentIndex,
            ParentID:  lastFull.ID,
            Entries:   entries,
            Metadata: map[string]interface{}{
                "version":            "1.0",
                "created_by":         "backup_scheduler",
                "entries_count":      len(entries),
                "parent_backup":      lastFull.ID,
                "parent_time":        lastFull.StartTime,
                "saga_active_count":  sagaCount,
                "include_saga_state": bs.config.IncludeSagaState,
                "saga_id":            sagaID,
                "backup_config": map[string]interface{}{
                    "backup_dir":         bs.config.BackupDir,
                    "compress_enabled":   bs.config.CompressEnabled,
                    "enable_incremental": bs.config.EnableIncremental,
                    "retention_days":     bs.config.RetentionDays,
                },
            },
            SagaState: sagaState,
            SagaID:    sagaID,
        }

        data, err := json.Marshal(backupData)
        if err != nil {
            return fmt.Errorf("failed to marshal backup data: %v", err)
        }

        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)
        }

        saga.SetData("backup_path", backupPath)
        saga.SetData("backup_data", data)
        saga.SetData("backup_size", int64(len(data)))
        return nil
    }, func() error {
        backupPathVal, _ := saga.GetData("backup_path")
        if backupPath, ok := backupPathVal.(string); ok && backupPath != "" {
            bs.storage.DeleteBackup(backupPath)
            if bs.logger != nil {
                bs.logger.Warn(fmt.Sprintf("Compensating: deleted backup file %s", backupPath))
            }
        }
        return nil
    }, map[string]interface{}{
        "step": "create_backup_file",
    })

    saga.AddStep("update_metadata", func() error {
        backupPathVal, _ := saga.GetData("backup_path")
        backupPath, _ := backupPathVal.(string)
        backupDataVal, _ := saga.GetData("backup_data")
        backupDataBytes, _ := backupDataVal.([]byte)
        backupSizeVal, _ := saga.GetData("backup_size")
        backupSize, _ := backupSizeVal.(int64)
        sagaCountVal, _ := saga.GetData("saga_count")
        sagaCount, _ := sagaCountVal.(int)

        if backupPath == "" {
            return fmt.Errorf("backup path not found")
        }

        currentIndex, err := bs.walReader.GetCurrentIndex()
        if err != nil {
            if bs.logger != nil {
                bs.logger.Warn(fmt.Sprintf("Failed to get current WAL index: %v", err))
            }
        } else {
            if err := bs.walReader.SetLastBackupIndex(currentIndex); err != nil {
                if bs.logger != nil {
                    bs.logger.Warn(fmt.Sprintf("Failed to set last backup index: %v", err))
                }
            }
        }

        hash := sha256.Sum256(backupDataBytes)
        checksum := hex.EncodeToString(hash[:])

        endTime := time.Now().UnixMilli()
        lastFull := bs.findLastFullBackup()
        lastIndex := bs.walReader.GetLastBackupIndex()

        backupInfo := &BackupInfo{
            ID:         backupID,
            ScheduleID: scheduleID,
            Type:       IncrementalBackup,
            StartTime:  startTime,
            EndTime:    endTime,
            Status:     "completed",
            SizeBytes:  backupSize,
            Path:       backupPath,
            WALStart:   lastIndex,
            WALEnd:     currentIndex,
            ParentID:   lastFull.ID,
            Checksum:   checksum,
            SagaCount:  sagaCount,
            SagaID:     sagaID,
        }

        bs.mu.Lock()
        bs.backups[backupID] = backupInfo
        bs.mu.Unlock()

        if err := bs.saveBackupHistory(); err != nil {
            if bs.logger != nil {
                bs.logger.Warn(fmt.Sprintf("Failed to save backup history: %v", err))
            }
        }

        bs.updateScheduleLastRun(scheduleID, startTime)

        duration := (endTime - startTime) / 1000
        if bs.logger != nil {
            bs.logger.Info(fmt.Sprintf("Incremental backup %s completed in %d seconds, entries=%d, size=%d bytes, active_sagas=%d",
                backupID, duration, len(entries), backupSize, sagaCount))
        }

        return nil
    }, func() error {
        if bs.logger != nil {
            bs.logger.Warn(fmt.Sprintf("Compensating: rolling back metadata for backup %s", backupID))
        }
        bs.mu.Lock()
        delete(bs.backups, backupID)
        bs.mu.Unlock()
        bs.saveBackupHistory()
        return nil
    }, map[string]interface{}{
        "step": "update_metadata",
    })

    if err := bs.sagaManager.Execute(saga); err != nil {
        if bs.logger != nil {
            bs.logger.Error(fmt.Sprintf("SAGA backup transaction failed: %v", err))
        }
        return err
    }

    return nil
}

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
}

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()
        if err := bs.saveSchedules(); err != nil {
            if bs.logger != nil {
                bs.logger.Warn(fmt.Sprintf("Failed to save schedules after update: %v", err))
            }
        }
    }
}

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)
            if err := bs.storage.DeleteBackup(backup.Path); err != nil {
                if bs.logger != nil {
                    bs.logger.Warn(fmt.Sprintf("Failed to delete backup file %s: %v", backup.Path, err))
                }
            }
        }
    }

    for _, id := range toDelete {
        delete(bs.backups, id)
    }

    if len(toDelete) > 0 {
        if err := bs.saveBackupHistory(); err != nil {
            if bs.logger != nil {
                bs.logger.Warn(fmt.Sprintf("Failed to save backup history after cleanup: %v", err))
            }
        }
        if bs.logger != nil {
            bs.logger.Info(fmt.Sprintf("Cleaned up %d old backups for schedule %s", len(toDelete), scheduleID))
        }
    }
}

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
}

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
}

func (bs *BackupScheduler) GetBackup(backupID string) *BackupInfo {
    bs.mu.RLock()
    defer bs.mu.RUnlock()
    return bs.backups[backupID]
}

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)
    }

    if bs.logger != nil {
        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)
    }

    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 {
        if bs.logger != nil {
            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"`
            SagaState  interface{} `json:"saga_state,omitempty"`
        }

        if err := json.Unmarshal(data, &fullData); err != nil {
            return fmt.Errorf("failed to parse full backup data: %v", err)
        }

        if bs.logger != nil {
            bs.logger.Info(fmt.Sprintf("Full backup contains %d WAL entries", len(fullData.Entries)))
        }

        if fullData.SagaState != nil && bs.config.IncludeSagaState {
            if bs.logger != nil {
                bs.logger.Info("Restoring SAGA state from backup...")
            }
            if sagaState, ok := fullData.SagaState.(map[string]interface{}); ok {
                if activeCount, ok := sagaState["active_count"].(float64); ok && activeCount > 0 {
                    if bs.logger != nil {
                        bs.logger.Info(fmt.Sprintf("Found %d active SAGA transactions in backup", int(activeCount)))
                    }
                }
            }
        }

    } else {
        if bs.logger != nil {
            bs.logger.Info("Restoring from incremental backup...")
        }

        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"`
            SagaState  interface{} `json:"saga_state,omitempty"`
            SagaID     string     `json:"saga_id,omitempty"`
        }

        if err := json.Unmarshal(data, &incData); err != nil {
            return fmt.Errorf("failed to parse incremental backup data: %v", err)
        }

        if bs.logger != nil {
            bs.logger.Info(fmt.Sprintf("Applying %d incremental changes (WAL %d -> %d)",
                len(incData.Entries), incData.WALStart, incData.WALEnd))
        }

        if incData.SagaState != nil && bs.config.IncludeSagaState {
            if bs.logger != nil {
                bs.logger.Info("Restoring SAGA state from incremental backup...")
            }
            if sagaState, ok := incData.SagaState.(map[string]interface{}); ok {
                if activeCount, ok := sagaState["active_count"].(float64); ok && activeCount > 0 {
                    if bs.logger != nil {
                        bs.logger.Info(fmt.Sprintf("Found %d active SAGA transactions in incremental backup", int(activeCount)))
                    }
                    if incData.SagaID != "" && bs.sagaManager != nil {
                        if bs.logger != nil {
                            bs.logger.Info(fmt.Sprintf("Restoring SAGA transaction %s", incData.SagaID))
                        }
                    }
                }
            }
        }
    }

    if bs.logger != nil {
        bs.logger.Info(fmt.Sprintf("Restore from backup %s completed", backupID))
    }
    return nil
}

func (bs *BackupScheduler) Start() {
    bs.wg.Add(1)
    go bs.schedulerLoop()
    if bs.logger != nil {
        bs.logger.Info("Backup scheduler started")
    }
}

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()
        }
    }
}

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 {
        if schedule.NextRun == 0 || schedule.NextRun <= now.UnixMilli() {
            schedule.NextRun = now.Add(24 * time.Hour).UnixMilli()

            bs.mu.Lock()
            bs.schedules[schedule.ID] = schedule
            bs.mu.Unlock()

            if err := bs.saveSchedules(); err != nil {
                if bs.logger != nil {
                    bs.logger.Warn(fmt.Sprintf("Failed to save schedule after update: %v", err))
                }
            }

            scheduleID := schedule.ID
            backupType := schedule.Type
            go func() {
                if backupType == FullBackup {
                    bs.performFullBackup(scheduleID)
                } else {
                    bs.performIncrementalBackup(scheduleID)
                }
            }()
        }
    }
}

func (bs *BackupScheduler) Stop() {
    close(bs.stopChan)
    bs.wg.Wait()
    if bs.logger != nil {
        bs.logger.Info("Backup scheduler stopped")
    }
}

func (bs *BackupScheduler) GetStats() map[string]interface{} {
    bs.mu.RLock()
    defer bs.mu.RUnlock()

    fullCount := 0
    incCount := 0
    totalSize := int64(0)
    totalSaga := 0

    for _, b := range bs.backups {
        if b.Type == FullBackup {
            fullCount++
        } else {
            incCount++
        }
        totalSize += b.SizeBytes
        totalSaga += b.SagaCount
    }

    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,
        "include_saga_state":     bs.config.IncludeSagaState,
        "total_saga_transactions": totalSaga,
        "retention_days":         bs.config.RetentionDays,
        "max_concurrent":         bs.config.MaxConcurrent,
    }
}