futriix/internal/storage/persistence.go

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

// Файл: internal/storage/persistence.go
// Назначение: Персистентное хранение данных на диске с поддержкой checkpoint и recovery

package storage

import (
    "bytes"
    "compress/gzip"
    "encoding/json"
    "fmt"
    "os"
    "path/filepath"
    "sort"
    "strings"
    "sync"
    "sync/atomic"
    "time"
)

// PersistenceConfig конфигурация персистентного хранения
type PersistenceConfig struct {
    DataDir           string        `json:"data_dir"`
    CheckpointInterval time.Duration `json:"checkpoint_interval"`
    MaxCheckpoints    int           `json:"max_checkpoints"`
    CompressEnabled   bool          `json:"compress_enabled"`
    SyncWrites        bool          `json:"sync_writes"`
}

// DefaultPersistenceConfig возвращает конфигурацию по умолчанию
func DefaultPersistenceConfig() *PersistenceConfig {
    return &PersistenceConfig{
        DataDir:           "futriis_data",
        CheckpointInterval: 5 * time.Minute,
        MaxCheckpoints:    10,
        CompressEnabled:   true,
        SyncWrites:        true,
    }
}

// DatabaseSnapshot представляет снапшот базы данных
type DatabaseSnapshot struct {
    Name        string                 `json:"name"`
    Collections map[string]interface{} `json:"collections"`
    CreatedAt   int64                  `json:"created_at"`
    Version     uint64                 `json:"version"`
    Checksum    string                 `json:"checksum"`
}

// PersistenceManager управляет персистентным хранением
type PersistenceManager struct {
    config         *PersistenceConfig
    storage        *Storage
    logger         LoggerInterface
    mu             sync.RWMutex
    stopChan       chan struct{}
    wg             sync.WaitGroup
    lastCheckpoint int64
    checkpointID   atomic.Uint64
}

// NewPersistenceManager создаёт новый менеджер персистентности
func NewPersistenceManager(config *PersistenceConfig, storage *Storage, logger LoggerInterface) *PersistenceManager {
    if config == nil {
        config = DefaultPersistenceConfig()
    }
    
    pm := &PersistenceManager{
        config:         config,
        storage:        storage,
        logger:         logger,
        stopChan:       make(chan struct{}),
        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))
        }
    }
    
    return pm
}

// Start запускает фоновое сохранение чекпоинтов
func (pm *PersistenceManager) Start() {
    pm.wg.Add(1)
    go pm.checkpointLoop()
    
    if pm.logger != nil {
        pm.logger.Info(fmt.Sprintf("Persistence manager started, data dir: %s", pm.config.DataDir))
    }
}

// Stop останавливает менеджер
func (pm *PersistenceManager) Stop() {
    close(pm.stopChan)
    pm.wg.Wait()
    
    // Финальное сохранение
    pm.SaveAll()
    
    if pm.logger != nil {
        pm.logger.Info("Persistence manager stopped")
    }
}

// checkpointLoop периодически создаёт чекпоинты
func (pm *PersistenceManager) checkpointLoop() {
    defer pm.wg.Done()
    
    ticker := time.NewTicker(pm.config.CheckpointInterval)
    defer ticker.Stop()
    
    for {
        select {
        case <-pm.stopChan:
            return
        case <-ticker.C:
            if err := pm.SaveAll(); err != nil {
                if pm.logger != nil {
                    pm.logger.Error(fmt.Sprintf("Failed to save checkpoint: %v", err))
                }
            } else {
                pm.cleanupOldCheckpoints()
            }
        }
    }
}

// SaveDatabase сохраняет базу данных на диск
func (pm *PersistenceManager) SaveDatabase(dbName string) error {
    pm.mu.Lock()
    defer pm.mu.Unlock()
    
    db, err := pm.storage.GetDatabase(dbName)
    if err != nil {
        return fmt.Errorf("database not found: %s", dbName)
    }
    
    // Собираем данные всех коллекций
    snapshot := &DatabaseSnapshot{
        Name:        dbName,
        Collections: make(map[string]interface{}),
        CreatedAt:   time.Now().UnixMilli(),
        Version:     pm.checkpointID.Add(1),
    }
    
    for _, collName := range db.ListCollections() {
        coll, err := db.GetCollection(collName)
        if err != nil {
            continue
        }
        
        docs := coll.GetAllDocuments()
        collData := make([]map[string]interface{}, 0, len(docs))
        for _, doc := range docs {
            collData = append(collData, map[string]interface{}{
                "_id":        doc.ID,
                "fields":     doc.GetFields(),
                "created_at": doc.CreatedAt,
                "updated_at": doc.UpdatedAt,
                "version":    doc.Version,
            })
        }
        snapshot.Collections[collName] = collData
    }
    
    // Сериализуем и сохраняем
    data, err := json.Marshal(snapshot)
    if err != nil {
        return err
    }
    
    // Опционально сжимаем
    if pm.config.CompressEnabled {
        data, err = pm.compress(data)
        if err != nil {
            return err
        }
    }
    
    // Сохраняем в файл
    filename := pm.getSnapshotFilename(dbName, snapshot.Version)
    if err := os.WriteFile(filename, data, 0644); err != nil {
        return err
    }
    
    // Синхронизируем с диском если нужно
    if pm.config.SyncWrites {
        if f, err := os.OpenFile(filename, os.O_RDWR, 0644); err == nil {
            RealFsync(f)
            f.Close()
        }
    }
    
    if pm.logger != nil {
        pm.logger.Info(fmt.Sprintf("Saved database %s checkpoint %d (%d bytes)", dbName, snapshot.Version, len(data)))
    }
    
    return nil
}

// SaveAll сохраняет все базы данных
func (pm *PersistenceManager) SaveAll() error {
    databases := pm.storage.ListDatabases()
    
    var lastErr error
    for _, dbName := range databases {
        if err := pm.SaveDatabase(dbName); err != nil {
            lastErr = err
            if pm.logger != nil {
                pm.logger.Error(fmt.Sprintf("Failed to save database %s: %v", dbName, err))
            }
        }
    }
    
    return lastErr
}

// LoadDatabase загружает базу данных с диска
func (pm *PersistenceManager) LoadDatabase(dbName string) error {
    pm.mu.Lock()
    defer pm.mu.Unlock()
    
    // Находим последний чекпоинт
    snapshot, err := pm.findLatestSnapshot(dbName)
    if err != nil {
        return err
    }
    
    if snapshot == nil {
        return fmt.Errorf("no snapshot found for database: %s", dbName)
    }
    
    // Восстанавливаем базу данных
    if !pm.storage.ExistsDatabase(dbName) {
        if err := pm.storage.CreateDatabase(dbName); err != nil {
            return err
        }
    }
    
    db, err := pm.storage.GetDatabase(dbName)
    if err != nil {
        return err
    }
    
    // Восстанавливаем коллекции
    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
            }
        }
        
        coll, err := db.GetCollection(collName)
        if err != nil {
            continue
        }
        
        for _, docRaw := range collData {
            docMap, ok := docRaw.(map[string]interface{})
            if !ok {
                continue
            }
            
            docID, ok := docMap["_id"].(string)
            if !ok {
                continue
            }
            
            doc := NewDocumentWithID(docID)
            
            if fields, ok := docMap["fields"].(map[string]interface{}); ok {
                for k, v := range fields {
                    doc.SetField(k, v)
                }
            }
            
            if createdAt, ok := docMap["created_at"].(float64); ok {
                doc.CreatedAt = int64(createdAt)
            }
            
            if updatedAt, ok := docMap["updated_at"].(float64); ok {
                doc.UpdatedAt = int64(updatedAt)
            }
            
            if version, ok := docMap["version"].(float64); ok {
                doc.Version = uint64(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))
                }
            }
        }
    }
    
    if pm.logger != nil {
        pm.logger.Info(fmt.Sprintf("Loaded database %s from snapshot (version %d)", dbName, snapshot.Version))
    }
    
    return nil
}

// LoadAll загружает все базы данных с диска
func (pm *PersistenceManager) LoadAll() error {
    files, err := filepath.Glob(filepath.Join(pm.config.DataDir, "snapshot_*.json*"))
    if err != nil {
        return err
    }
    
    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]
            databases[dbName] = true
        }
    }
    
    for dbName := range databases {
        if err := pm.LoadDatabase(dbName); err != nil {
            if pm.logger != nil {
                pm.logger.Error(fmt.Sprintf("Failed to load database %s: %v", dbName, err))
            }
        }
    }
    
    return nil
}

// getSnapshotFilename возвращает имя файла для снапшота
func (pm *PersistenceManager) getSnapshotFilename(dbName string, version uint64) string {
    filename := fmt.Sprintf("snapshot_%s_%d.json", dbName, version)
    if pm.config.CompressEnabled {
        filename += ".gz"
    }
    return filepath.Join(pm.config.DataDir, filename)
}

// findLatestSnapshot находит последний снапшот базы данных
func (pm *PersistenceManager) findLatestSnapshot(dbName string) (*DatabaseSnapshot, error) {
    pattern := filepath.Join(pm.config.DataDir, fmt.Sprintf("snapshot_%s_*.json*", dbName))
    files, err := filepath.Glob(pattern)
    if err != nil {
        return nil, err
    }
    
    if len(files) == 0 {
        return nil, nil
    }
    
    // Сортируем по времени создания (новейшие первыми)
    sort.Slice(files, func(i, j int) bool {
        infoI, _ := os.Stat(files[i])
        infoJ, _ := os.Stat(files[j])
        if infoI == nil || infoJ == nil {
            return false
        }
        return infoI.ModTime().After(infoJ.ModTime())
    })
    
    // Читаем последний снапшот
    data, err := os.ReadFile(files[0])
    if err != nil {
        return nil, err
    }
    
    if pm.config.CompressEnabled && strings.HasSuffix(files[0], ".gz") {
        data, err = pm.decompress(data)
        if err != nil {
            return nil, err
        }
    }
    
    var snapshot DatabaseSnapshot
    if err := json.Unmarshal(data, &snapshot); err != nil {
        return nil, err
    }
    
    return &snapshot, nil
}

// compress сжимает данные с помощью gzip
func (pm *PersistenceManager) compress(data []byte) ([]byte, error) {
    var buf bytes.Buffer
    w := gzip.NewWriter(&buf)
    
    if _, err := w.Write(data); err != nil {
        return nil, fmt.Errorf("failed to write compressed data: %v", err)
    }
    
    if err := w.Close(); err != nil {
        return nil, fmt.Errorf("failed to close gzip writer: %v", err)
    }
    
    return buf.Bytes(), nil
}

// decompress разжимает данные с помощью gzip
func (pm *PersistenceManager) decompress(data []byte) ([]byte, error) {
    reader, err := gzip.NewReader(bytes.NewReader(data))
    if err != nil {
        return nil, fmt.Errorf("failed to create gzip reader: %v", err)
    }
    defer reader.Close()
    
    var buf bytes.Buffer
    if _, err := buf.ReadFrom(reader); err != nil {
        return nil, fmt.Errorf("failed to decompress data: %v", err)
    }
    
    return buf.Bytes(), nil
}

// cleanupOldCheckpoints удаляет старые чекпоинты
func (pm *PersistenceManager) cleanupOldCheckpoints() {
    pattern := filepath.Join(pm.config.DataDir, "snapshot_*.json*")
    files, err := filepath.Glob(pattern)
    if err != nil {
        return
    }
    
    if len(files) <= pm.config.MaxCheckpoints {
        return
    }
    
    // Сортируем по времени модификации
    sort.Slice(files, func(i, j int) bool {
        infoI, _ := os.Stat(files[i])
        infoJ, _ := os.Stat(files[j])
        if infoI == nil || infoJ == nil {
            return false
        }
        return infoI.ModTime().Before(infoJ.ModTime())
    })
    
    // Удаляем старые
    toDelete := files[:len(files)-pm.config.MaxCheckpoints]
    for _, f := range toDelete {
        os.Remove(f)
        if pm.logger != nil {
            pm.logger.Debug(fmt.Sprintf("Removed old checkpoint: %s", f))
        }
    }
}