diff --git a/internal/storage/persistence.go b/internal/storage/persistence.go deleted file mode 100644 index 1786529..0000000 --- a/internal/storage/persistence.go +++ /dev/null @@ -1,466 +0,0 @@ -/* - * 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)) - } - } -}