diff --git a/internal/migration/11schema_migrator.go b/internal/migration/11schema_migrator.go deleted file mode 100644 index b88e12e..0000000 --- a/internal/migration/11schema_migrator.go +++ /dev/null @@ -1,1180 +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/migration/schema_migrator.go -// Назначение: Миграция схемы данных при обновлении версии СУБД -// Новая функциональность: Прозрачное обновление схемы данных (автоматическая миграция документов) - -package migration - -import ( - "encoding/json" - "fmt" - "os" - "path/filepath" - "sort" - "strings" - "sync" - "time" - - "futriis/internal/log" - "futriis/internal/storage" -) - -// SchemaDefinition определяет схему коллекции -type SchemaDefinition struct { - Version string `json:"version"` - Collections map[string]*CollectionSchema `json:"collections"` - UpdatedAt int64 `json:"updated_at"` -} - -// CollectionSchema определяет схему одной коллекции -type CollectionSchema struct { - Fields map[string]*FieldDefinition `json:"fields"` - Required []string `json:"required"` - Indexes []IndexDefinition `json:"indexes"` - UpdatedAt int64 `json:"updated_at"` -} - -// FieldDefinition определяет поле в схеме -type FieldDefinition struct { - Type string `json:"type"` // string, int, float, bool, array, object - Required bool `json:"required"` - Default interface{} `json:"default"` - Validate string `json:"validate"` // regex, min, max, enum - ValidateValue interface{} `json:"validate_value"` -} - -// IndexDefinition определяет индекс -type IndexDefinition struct { - Name string `json:"name"` - Fields []string `json:"fields"` - Unique bool `json:"unique"` -} - -// SchemaMigration представляет миграцию схемы -type SchemaMigration struct { - ID string - Version string - Description string - Up func(tx *storage.Transaction, schema *SchemaDefinition) error - Down func(tx *storage.Transaction, schema *SchemaDefinition) error - CreatedAt int64 - AppliedAt int64 -} - -// SchemaMigrationRecord запись о применённой миграции схемы -type SchemaMigrationRecord struct { - ID string `json:"id"` - Version string `json:"version"` - Description string `json:"description"` - AppliedAt int64 `json:"applied_at"` - Success bool `json:"success"` - Error string `json:"error,omitempty"` -} - -// DocumentMigrationStrategy стратегия миграции документа -type DocumentMigrationStrategy int - -const ( - StrategyInPlace DocumentMigrationStrategy = iota // Обновление на месте - StrategyCopyNew // Копирование в новую коллекцию - StrategyLazy // Ленивая миграция при доступе -) - -// SchemaMigrator управляет миграциями схемы данных -type SchemaMigrator struct { - store *storage.Storage - logger *log.Logger - migrations map[string]*SchemaMigration - applied map[string]*SchemaMigrationRecord - mu sync.RWMutex - migrationDir string - currentVersion string - targetVersion string - schema *SchemaDefinition - strategy DocumentMigrationStrategy - migrationStats *MigrationStats -} - -// MigrationStats статистика миграции -type MigrationStats struct { - TotalDocuments int64 - MigratedDocuments int64 - FailedDocuments int64 - SkippedDocuments int64 - StartTime int64 - EndTime int64 - mu sync.RWMutex -} - -// NewSchemaMigrator создаёт новый мигратор схемы -func NewSchemaMigrator(store *storage.Storage, logger *log.Logger, migrationDir string) *SchemaMigrator { - sm := &SchemaMigrator{ - store: store, - logger: logger, - migrations: make(map[string]*SchemaMigration), - applied: make(map[string]*SchemaMigrationRecord), - migrationDir: migrationDir, - strategy: StrategyLazy, // Ленивая миграция по умолчанию - migrationStats: &MigrationStats{}, - schema: &SchemaDefinition{ - Version: "1.0.0", - Collections: make(map[string]*CollectionSchema), - UpdatedAt: time.Now().UnixMilli(), - }, - } - - // Создаём директорию для миграций - os.MkdirAll(migrationDir, 0755) - - // Загружаем существующую схему - sm.loadSchema() - - // Загружаем применённые миграции - sm.loadAppliedMigrations() - - // Регистрируем встроенные миграции схемы - sm.registerBuiltinSchemaMigrations() - - return sm -} - -// loadSchema загружает определение схемы из файла -func (sm *SchemaMigrator) loadSchema() { - path := filepath.Join(sm.migrationDir, "schema.json") - - data, err := os.ReadFile(path) - if err != nil { - if sm.logger != nil { - sm.logger.Debug("No existing schema found, creating default schema") - } - sm.createDefaultSchema() - return - } - - var schema SchemaDefinition - if err := json.Unmarshal(data, &schema); err != nil { - if sm.logger != nil { - sm.logger.Error(fmt.Sprintf("Failed to unmarshal schema: %v", err)) - } - sm.createDefaultSchema() - return - } - - sm.mu.Lock() - sm.schema = &schema - sm.currentVersion = schema.Version - sm.mu.Unlock() - - if sm.logger != nil { - sm.logger.Info(fmt.Sprintf("Loaded schema version %s with %d collections", schema.Version, len(schema.Collections))) - } -} - -// saveSchema сохраняет определение схемы -func (sm *SchemaMigrator) saveSchema() error { - sm.mu.RLock() - defer sm.mu.RUnlock() - - data, err := json.MarshalIndent(sm.schema, "", " ") - if err != nil { - return err - } - - path := filepath.Join(sm.migrationDir, "schema.json") - return os.WriteFile(path, data, 0644) -} - -// createDefaultSchema создаёт схему по умолчанию -func (sm *SchemaMigrator) createDefaultSchema() { - sm.mu.Lock() - defer sm.mu.Unlock() - - sm.schema = &SchemaDefinition{ - Version: "1.0.0", - Collections: make(map[string]*CollectionSchema), - UpdatedAt: time.Now().UnixMilli(), - } - - // Добавляем базовую схему для всех коллекций - sm.schema.Collections["_default"] = &CollectionSchema{ - Fields: map[string]*FieldDefinition{ - "_id": { - Type: "string", - Required: true, - }, - "created_at": { - Type: "int64", - Default: int64(0), - }, - "updated_at": { - Type: "int64", - Default: int64(0), - }, - }, - Required: []string{"_id"}, - Indexes: []IndexDefinition{}, - UpdatedAt: time.Now().UnixMilli(), - } - - sm.saveSchema() -} - -// registerBuiltinSchemaMigrations регистрирует встроенные миграции схемы -func (sm *SchemaMigrator) registerBuiltinSchemaMigrations() { - // Миграция v1.0.0 -> v1.1.0: Добавление обязательных полей created_at/updated_at - sm.RegisterSchemaMigration(&SchemaMigration{ - ID: "schema_001_add_timestamps", - Version: "1.1.0", - Description: "Add created_at and updated_at timestamps to all collections", - Up: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaAddTimestamps(tx, schema) - }, - Down: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaRemoveTimestamps(tx, schema) - }, - CreatedAt: time.Now().UnixMilli(), - }) - - // Миграция v1.1.0 -> v1.2.0: Добавление мягкого удаления - sm.RegisterSchemaMigration(&SchemaMigration{ - ID: "schema_002_add_soft_delete", - Version: "1.2.0", - Description: "Add soft delete support (deleted_at, deleted fields)", - Up: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaAddSoftDelete(tx, schema) - }, - Down: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaRemoveSoftDelete(tx, schema) - }, - CreatedAt: time.Now().UnixMilli(), - }) - - // Миграция v1.2.0 -> v1.3.0: Добавление версионирования - sm.RegisterSchemaMigration(&SchemaMigration{ - ID: "schema_003_add_versioning", - Version: "1.3.0", - Description: "Add document versioning (_version field)", - Up: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaAddVersioning(tx, schema) - }, - Down: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaRemoveVersioning(tx, schema) - }, - CreatedAt: time.Now().UnixMilli(), - }) - - // Миграция v1.3.0 -> v2.0.0: Добавление индексов - sm.RegisterSchemaMigration(&SchemaMigration{ - ID: "schema_004_add_indexes", - Version: "2.0.0", - Description: "Add secondary indexes for performance", - Up: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaAddIndexes(tx, schema) - }, - Down: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaRemoveIndexes(tx, schema) - }, - CreatedAt: time.Now().UnixMilli(), - }) - - // Миграция v2.0.0 -> v2.1.0: Добавление валидации полей - sm.RegisterSchemaMigration(&SchemaMigration{ - ID: "schema_005_add_validation", - Version: "2.1.0", - Description: "Add field validation rules", - Up: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaAddValidation(tx, schema) - }, - Down: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return sm.migrateSchemaRemoveValidation(tx, schema) - }, - CreatedAt: time.Now().UnixMilli(), - }) -} - -// RegisterSchemaMigration регистрирует миграцию схемы -func (sm *SchemaMigrator) RegisterSchemaMigration(m *SchemaMigration) { - sm.mu.Lock() - defer sm.mu.Unlock() - sm.migrations[m.ID] = m -} - -// loadAppliedMigrations загружает применённые миграции -func (sm *SchemaMigrator) loadAppliedMigrations() { - path := filepath.Join(sm.migrationDir, "schema_migrations.json") - - data, err := os.ReadFile(path) - if err != nil { - return - } - - var records []SchemaMigrationRecord - if err := json.Unmarshal(data, &records); err != nil { - if sm.logger != nil { - sm.logger.Error(fmt.Sprintf("Failed to unmarshal schema migrations: %v", err)) - } - return - } - - sm.mu.Lock() - defer sm.mu.Unlock() - - for _, record := range records { - sm.applied[record.ID] = &record - } -} - -// saveAppliedMigrations сохраняет применённые миграции -func (sm *SchemaMigrator) saveAppliedMigrations() error { - sm.mu.RLock() - defer sm.mu.RUnlock() - - records := make([]SchemaMigrationRecord, 0, len(sm.applied)) - for _, record := range sm.applied { - records = append(records, *record) - } - - data, err := json.MarshalIndent(records, "", " ") - if err != nil { - return err - } - - path := filepath.Join(sm.migrationDir, "schema_migrations.json") - return os.WriteFile(path, data, 0644) -} - -// MigrateSchema выполняет миграцию схемы до указанной версии -func (sm *SchemaMigrator) MigrateSchema(targetVersion string) error { - sm.mu.Lock() - defer sm.mu.Unlock() - - if sm.logger != nil { - sm.logger.Info(fmt.Sprintf("Starting schema migration to version %s", targetVersion)) - } - - // Получаем отсортированный список миграций - migrations := sm.getSortedSchemaMigrations() - - currentVersion := sm.schema.Version - - if currentVersion == targetVersion { - if sm.logger != nil { - sm.logger.Info("Already at target version, no migration needed") - } - return nil - } - - var toApply []*SchemaMigration - - for _, m := range migrations { - if sm.isVersionGreater(m.Version, currentVersion) && !sm.isVersionGreater(m.Version, targetVersion) { - if _, applied := sm.applied[m.ID]; !applied { - toApply = append(toApply, m) - } - } - } - - if sm.logger != nil { - sm.logger.Info(fmt.Sprintf("Applying %d schema migrations", len(toApply))) - } - - // Начинаем транзакцию для миграции - tx := storage.BeginTransaction() - defer func() { - if r := recover(); r != nil { - storage.AbortCurrentTransaction() - if sm.logger != nil { - sm.logger.Error(fmt.Sprintf("Panic during schema migration: %v", r)) - } - } - }() - - for _, m := range toApply { - if sm.logger != nil { - sm.logger.Info(fmt.Sprintf("Applying schema migration: %s (%s)", m.ID, m.Description)) - } - - startTime := time.Now() - - // Применяем миграцию - if err := m.Up(tx, sm.schema); err != nil { - storage.AbortCurrentTransaction() - return fmt.Errorf("schema migration %s failed: %v", m.ID, err) - } - - // Обновляем версию схемы - sm.schema.Version = m.Version - sm.schema.UpdatedAt = time.Now().UnixMilli() - - // Сохраняем запись - record := &SchemaMigrationRecord{ - ID: m.ID, - Version: m.Version, - Description: m.Description, - AppliedAt: time.Now().UnixMilli(), - Success: true, - } - sm.applied[m.ID] = record - - duration := time.Since(startTime) - if sm.logger != nil { - sm.logger.Info(fmt.Sprintf("Schema migration %s completed in %v", m.ID, duration)) - } - } - - // Сохраняем схему - if err := sm.saveSchema(); err != nil { - storage.AbortCurrentTransaction() - return fmt.Errorf("failed to save schema: %v", err) - } - - // Коммитим транзакцию - if err := storage.CommitCurrentTransaction(); err != nil { - return fmt.Errorf("failed to commit schema migration: %v", err) - } - - sm.saveAppliedMigrations() - sm.currentVersion = targetVersion - - if sm.logger != nil { - sm.logger.Info(fmt.Sprintf("Schema migration to version %s completed successfully", targetVersion)) - } - - return nil -} - -// MigrateDocument прозрачно мигрирует отдельный документ к актуальной схеме -func (sm *SchemaMigrator) MigrateDocument(doc *storage.Document, collectionName string) (*storage.Document, error) { - sm.mu.RLock() - defer sm.mu.RUnlock() - - // Получаем схему для коллекции - collectionSchema, exists := sm.schema.Collections[collectionName] - if !exists { - // Используем схему по умолчанию - collectionSchema = sm.schema.Collections["_default"] - if collectionSchema == nil { - return doc, nil - } - } - - // Клонируем документ - migratedDoc := doc.Clone() - changed := false - - // Проверяем и добавляем недостающие поля с значениями по умолчанию - for fieldName, fieldDef := range collectionSchema.Fields { - if _, err := migratedDoc.GetField(fieldName); err != nil { - if fieldDef.Default != nil { - migratedDoc.SetField(fieldName, fieldDef.Default) - changed = true - if sm.logger != nil { - sm.logger.Debug(fmt.Sprintf("Added default value for field %s in document %s", fieldName, doc.ID)) - } - } - } - } - - // Проверяем типы полей и преобразуем при необходимости - for fieldName, fieldDef := range collectionSchema.Fields { - if value, err := migratedDoc.GetField(fieldName); err == nil { - converted, needsConversion := sm.convertFieldType(value, fieldDef.Type) - if needsConversion { - migratedDoc.SetField(fieldName, converted) - changed = true - if sm.logger != nil { - sm.logger.Debug(fmt.Sprintf("Converted field %s type for document %s", fieldName, doc.ID)) - } - } - } - } - - if changed { - migratedDoc.UpdatedAt = time.Now().UnixMilli() - migratedDoc.Version++ - sm.migrationStats.mu.Lock() - sm.migrationStats.MigratedDocuments++ - sm.migrationStats.mu.Unlock() - } - - return migratedDoc, nil -} - -// MigrateCollectionDocuments прозрачно мигрирует все документы коллекции -func (sm *SchemaMigrator) MigrateCollectionDocuments(collection *storage.Collection) error { - startTime := time.Now() - - sm.migrationStats.mu.Lock() - sm.migrationStats.StartTime = startTime.UnixMilli() - sm.migrationStats.TotalDocuments = collection.Count() - sm.migrationStats.MigratedDocuments = 0 - sm.migrationStats.FailedDocuments = 0 - sm.migrationStats.SkippedDocuments = 0 - sm.migrationStats.mu.Unlock() - - if sm.logger != nil { - sm.logger.Info(fmt.Sprintf("Starting migration of collection %s with %d documents", - collection.Name(), collection.Count())) - } - - docs := collection.GetAllDocuments() - - for _, doc := range docs { - migratedDoc, err := sm.MigrateDocument(doc, collection.Name()) - if err != nil { - sm.migrationStats.mu.Lock() - sm.migrationStats.FailedDocuments++ - sm.migrationStats.mu.Unlock() - if sm.logger != nil { - sm.logger.Error(fmt.Sprintf("Failed to migrate document %s: %v", doc.ID, err)) - } - continue - } - - if migratedDoc != doc { - // Обновляем документ в коллекции - updates := migratedDoc.GetFields() - if err := collection.Update(doc.ID, updates); err != nil { - sm.migrationStats.mu.Lock() - sm.migrationStats.FailedDocuments++ - sm.migrationStats.mu.Unlock() - if sm.logger != nil { - sm.logger.Error(fmt.Sprintf("Failed to update migrated document %s: %v", doc.ID, err)) - } - } - } else { - sm.migrationStats.mu.Lock() - sm.migrationStats.SkippedDocuments++ - sm.migrationStats.mu.Unlock() - } - } - - sm.migrationStats.mu.Lock() - sm.migrationStats.EndTime = time.Now().UnixMilli() - sm.migrationStats.mu.Unlock() - - duration := time.Since(startTime) - if sm.logger != nil { - sm.logger.Info(fmt.Sprintf("Migration of collection %s completed in %v: %d migrated, %d failed, %d skipped", - collection.Name(), duration, - sm.migrationStats.MigratedDocuments, - sm.migrationStats.FailedDocuments, - sm.migrationStats.SkippedDocuments)) - } - - return nil -} - -// convertFieldType преобразует значение поля к нужному типу -func (sm *SchemaMigrator) convertFieldType(value interface{}, targetType string) (interface{}, bool) { - switch targetType { - case "string": - if _, ok := value.(string); !ok { - return fmt.Sprintf("%v", value), true - } - case "int64": - switch v := value.(type) { - case int: - return int64(v), true - case float64: - return int64(v), true - case string: - var i int64 - if _, err := fmt.Sscanf(v, "%d", &i); err == nil { - return i, true - } - } - case "float64": - switch v := value.(type) { - case int: - return float64(v), true - case int64: - return float64(v), true - case string: - var f float64 - if _, err := fmt.Sscanf(v, "%f", &f); err == nil { - return f, true - } - } - case "bool": - if _, ok := value.(bool); !ok { - if str, ok := value.(string); ok { - return strings.ToLower(str) == "true" || str == "1", true - } - return false, true - } - } - return value, false -} - -// getSortedSchemaMigrations возвращает отсортированный список миграций -func (sm *SchemaMigrator) getSortedSchemaMigrations() []*SchemaMigration { - migrations := make([]*SchemaMigration, 0, len(sm.migrations)) - for _, m := range sm.migrations { - migrations = append(migrations, m) - } - - sort.Slice(migrations, func(i, j int) bool { - return sm.isVersionLess(migrations[i].Version, migrations[j].Version) - }) - - return migrations -} - -// isVersionLess сравнивает версии -func (sm *SchemaMigrator) isVersionLess(v1, v2 string) bool { - return sm.versionToInt(v1) < sm.versionToInt(v2) -} - -// isVersionGreater сравнивает версии -func (sm *SchemaMigrator) isVersionGreater(v1, v2 string) bool { - return sm.versionToInt(v1) > sm.versionToInt(v2) -} - -// versionToInt преобразует версию в число -func (sm *SchemaMigrator) versionToInt(version string) int64 { - var major, minor, patch int - fmt.Sscanf(version, "%d.%d.%d", &major, &minor, &patch) - return int64(major*1000000 + minor*1000 + patch) -} - -// ========== Конкретные реализации миграций схемы ========== - -// migrateSchemaAddTimestamps добавляет поля created_at и updated_at -func (sm *SchemaMigrator) migrateSchemaAddTimestamps(tx *storage.Transaction, schema *SchemaDefinition) error { - for _, collSchema := range schema.Collections { - if _, exists := collSchema.Fields["created_at"]; !exists { - collSchema.Fields["created_at"] = &FieldDefinition{ - Type: "int64", - Required: false, - Default: int64(0), - } - } - if _, exists := collSchema.Fields["updated_at"]; !exists { - collSchema.Fields["updated_at"] = &FieldDefinition{ - Type: "int64", - Required: false, - Default: int64(0), - } - } - collSchema.UpdatedAt = time.Now().UnixMilli() - } - return nil -} - -// migrateSchemaRemoveTimestamps удаляет поля created_at и updated_at -func (sm *SchemaMigrator) migrateSchemaRemoveTimestamps(tx *storage.Transaction, schema *SchemaDefinition) error { - for _, collSchema := range schema.Collections { - delete(collSchema.Fields, "created_at") - delete(collSchema.Fields, "updated_at") - collSchema.UpdatedAt = time.Now().UnixMilli() - } - return nil -} - -// migrateSchemaAddSoftDelete добавляет поддержку мягкого удаления -func (sm *SchemaMigrator) migrateSchemaAddSoftDelete(tx *storage.Transaction, schema *SchemaDefinition) error { - for _, collSchema := range schema.Collections { - if _, exists := collSchema.Fields["deleted_at"]; !exists { - collSchema.Fields["deleted_at"] = &FieldDefinition{ - Type: "int64", - Required: false, - Default: int64(0), - } - } - if _, exists := collSchema.Fields["deleted"]; !exists { - collSchema.Fields["deleted"] = &FieldDefinition{ - Type: "bool", - Required: false, - Default: false, - } - } - collSchema.UpdatedAt = time.Now().UnixMilli() - } - return nil -} - -// migrateSchemaRemoveSoftDelete удаляет поддержку мягкого удаления -func (sm *SchemaMigrator) migrateSchemaRemoveSoftDelete(tx *storage.Transaction, schema *SchemaDefinition) error { - for _, collSchema := range schema.Collections { - delete(collSchema.Fields, "deleted_at") - delete(collSchema.Fields, "deleted") - collSchema.UpdatedAt = time.Now().UnixMilli() - } - return nil -} - -// migrateSchemaAddVersioning добавляет версионирование -func (sm *SchemaMigrator) migrateSchemaAddVersioning(tx *storage.Transaction, schema *SchemaDefinition) error { - for _, collSchema := range schema.Collections { - if _, exists := collSchema.Fields["_version"]; !exists { - collSchema.Fields["_version"] = &FieldDefinition{ - Type: "int64", - Required: false, - Default: int64(1), - } - } - collSchema.UpdatedAt = time.Now().UnixMilli() - } - return nil -} - -// migrateSchemaRemoveVersioning удаляет версионирование -func (sm *SchemaMigrator) migrateSchemaRemoveVersioning(tx *storage.Transaction, schema *SchemaDefinition) error { - for _, collSchema := range schema.Collections { - delete(collSchema.Fields, "_version") - collSchema.UpdatedAt = time.Now().UnixMilli() - } - return nil -} - -// migrateSchemaAddIndexes добавляет индексы -func (sm *SchemaMigrator) migrateSchemaAddIndexes(tx *storage.Transaction, schema *SchemaDefinition) error { - for collectionName, collSchema := range schema.Collections { - // Добавляем индексы для часто используемых полей - defaultIndexes := []IndexDefinition{ - {Name: "idx_created_at", Fields: []string{"created_at"}, Unique: false}, - {Name: "idx_updated_at", Fields: []string{"updated_at"}, Unique: false}, - } - - for _, idx := range defaultIndexes { - found := false - for _, existing := range collSchema.Indexes { - if existing.Name == idx.Name { - found = true - break - } - } - if !found { - collSchema.Indexes = append(collSchema.Indexes, idx) - } - } - collSchema.UpdatedAt = time.Now().UnixMilli() - - if sm.logger != nil { - sm.logger.Debug(fmt.Sprintf("Added indexes to collection %s", collectionName)) - } - } - return nil -} - -// migrateSchemaRemoveIndexes удаляет индексы -func (sm *SchemaMigrator) migrateSchemaRemoveIndexes(tx *storage.Transaction, schema *SchemaDefinition) error { - for _, collSchema := range schema.Collections { - // Оставляем только пользовательские индексы, удаляем автоматические - newIndexes := make([]IndexDefinition, 0) - for _, idx := range collSchema.Indexes { - if !strings.HasPrefix(idx.Name, "idx_") { - newIndexes = append(newIndexes, idx) - } - } - collSchema.Indexes = newIndexes - collSchema.UpdatedAt = time.Now().UnixMilli() - } - return nil -} - -// migrateSchemaAddValidation добавляет валидацию полей -func (sm *SchemaMigrator) migrateSchemaAddValidation(tx *storage.Transaction, schema *SchemaDefinition) error { - for _, collSchema := range schema.Collections { - for fieldName, fieldDef := range collSchema.Fields { - // Добавляем базовую валидацию для обязательных полей - if fieldName == "_id" { - fieldDef.Required = true - } - } - collSchema.UpdatedAt = time.Now().UnixMilli() - } - return nil -} - -// migrateSchemaRemoveValidation удаляет валидацию -func (sm *SchemaMigrator) migrateSchemaRemoveValidation(tx *storage.Transaction, schema *SchemaDefinition) error { - for _, collSchema := range schema.Collections { - for _, fieldDef := range collSchema.Fields { - fieldDef.Required = false - fieldDef.Validate = "" - fieldDef.ValidateValue = nil - } - collSchema.UpdatedAt = time.Now().UnixMilli() - } - return nil -} - -// ========== Публичные методы для управления схемой ========== - -// AddCollectionSchema добавляет схему для новой коллекции -func (sm *SchemaMigrator) AddCollectionSchema(collectionName string, schema *CollectionSchema) error { - sm.mu.Lock() - defer sm.mu.Unlock() - - sm.schema.Collections[collectionName] = schema - sm.schema.UpdatedAt = time.Now().UnixMilli() - - return sm.saveSchema() -} - -// GetCollectionSchema возвращает схему коллекции -func (sm *SchemaMigrator) GetCollectionSchema(collectionName string) *CollectionSchema { - sm.mu.RLock() - defer sm.mu.RUnlock() - - if schema, exists := sm.schema.Collections[collectionName]; exists { - return schema - } - return sm.schema.Collections["_default"] -} - -// UpdateFieldSchema обновляет схему поля -func (sm *SchemaMigrator) UpdateFieldSchema(collectionName, fieldName string, fieldDef *FieldDefinition) error { - sm.mu.Lock() - defer sm.mu.Unlock() - - collSchema, exists := sm.schema.Collections[collectionName] - if !exists { - collSchema = &CollectionSchema{ - Fields: make(map[string]*FieldDefinition), - Required: make([]string, 0), - Indexes: make([]IndexDefinition, 0), - UpdatedAt: time.Now().UnixMilli(), - } - sm.schema.Collections[collectionName] = collSchema - } - - collSchema.Fields[fieldName] = fieldDef - collSchema.UpdatedAt = time.Now().UnixMilli() - sm.schema.UpdatedAt = time.Now().UnixMilli() - - return sm.saveSchema() -} - -// GetSchema возвращает текущую схему -func (sm *SchemaMigrator) GetSchema() *SchemaDefinition { - sm.mu.RLock() - defer sm.mu.RUnlock() - - // Возвращаем копию - data, _ := json.Marshal(sm.schema) - var copy SchemaDefinition - json.Unmarshal(data, ©) - return © -} - -// GetSchemaVersion возвращает текущую версию схемы -func (sm *SchemaMigrator) GetSchemaVersion() string { - sm.mu.RLock() - defer sm.mu.RUnlock() - return sm.schema.Version -} - -// GetMigrationStats возвращает статистику миграции -func (sm *SchemaMigrator) GetMigrationStats() *MigrationStats { - sm.migrationStats.mu.RLock() - defer sm.migrationStats.mu.RUnlock() - - stats := &MigrationStats{ - TotalDocuments: sm.migrationStats.TotalDocuments, - MigratedDocuments: sm.migrationStats.MigratedDocuments, - FailedDocuments: sm.migrationStats.FailedDocuments, - SkippedDocuments: sm.migrationStats.SkippedDocuments, - StartTime: sm.migrationStats.StartTime, - EndTime: sm.migrationStats.EndTime, - } - return stats -} - -// SetMigrationStrategy устанавливает стратегию миграции -func (sm *SchemaMigrator) SetMigrationStrategy(strategy DocumentMigrationStrategy) { - sm.mu.Lock() - defer sm.mu.Unlock() - sm.strategy = strategy - - if sm.logger != nil { - strategyName := "InPlace" - if strategy == StrategyCopyNew { - strategyName = "CopyNew" - } else if strategy == StrategyLazy { - strategyName = "Lazy" - } - sm.logger.Info(fmt.Sprintf("Migration strategy set to %s", strategyName)) - } -} - -// ValidateDocumentAgainstSchema проверяет документ на соответствие схеме -func (sm *SchemaMigrator) ValidateDocumentAgainstSchema(doc *storage.Document, collectionName string) error { - sm.mu.RLock() - defer sm.mu.RUnlock() - - collSchema, exists := sm.schema.Collections[collectionName] - if !exists { - collSchema = sm.schema.Collections["_default"] - } - - // Проверяем обязательные поля - for _, requiredField := range collSchema.Required { - if _, err := doc.GetField(requiredField); err != nil { - return fmt.Errorf("required field '%s' is missing", requiredField) - } - } - - // Проверяем типы полей - for fieldName, fieldDef := range collSchema.Fields { - if value, err := doc.GetField(fieldName); err == nil { - if err := sm.validateFieldType(value, fieldDef.Type); err != nil { - return fmt.Errorf("field '%s' validation failed: %v", fieldName, err) - } - - // Дополнительная валидация - if fieldDef.Validate != "" { - if err := sm.validateFieldValue(value, fieldDef); err != nil { - return fmt.Errorf("field '%s' value validation failed: %v", fieldName, err) - } - } - } - } - - return nil -} - -// validateFieldType проверяет тип поля -func (sm *SchemaMigrator) validateFieldType(value interface{}, expectedType string) error { - switch expectedType { - case "string": - if _, ok := value.(string); !ok { - return fmt.Errorf("expected string, got %T", value) - } - case "int64": - switch value.(type) { - case int, int64, float64: - // допустимые типы - default: - return fmt.Errorf("expected int64, got %T", value) - } - case "float64": - switch value.(type) { - case float64, int, int64: - // допустимые типы - default: - return fmt.Errorf("expected float64, got %T", value) - } - case "bool": - if _, ok := value.(bool); !ok { - return fmt.Errorf("expected bool, got %T", value) - } - case "array": - // проверка на слайс - case "object": - // проверка на map - } - return nil -} - -// validateFieldValue проверяет значение поля по правилам валидации -func (sm *SchemaMigrator) validateFieldValue(value interface{}, fieldDef *FieldDefinition) error { - switch fieldDef.Validate { - case "min": - if min, ok := fieldDef.ValidateValue.(float64); ok { - switch v := value.(type) { - case float64: - if v < min { - return fmt.Errorf("value %.2f is less than minimum %.2f", v, min) - } - case int64: - if float64(v) < min { - return fmt.Errorf("value %d is less than minimum %.2f", v, min) - } - } - } - case "max": - if max, ok := fieldDef.ValidateValue.(float64); ok { - switch v := value.(type) { - case float64: - if v > max { - return fmt.Errorf("value %.2f exceeds maximum %.2f", v, max) - } - case int64: - if float64(v) > max { - return fmt.Errorf("value %d exceeds maximum %.2f", v, max) - } - } - } - case "enum": - if enumVals, ok := fieldDef.ValidateValue.([]interface{}); ok { - found := false - for _, ev := range enumVals { - if fmt.Sprintf("%v", value) == fmt.Sprintf("%v", ev) { - found = true - break - } - } - if !found { - return fmt.Errorf("value %v not in enum list", value) - } - } - case "regex": - if pattern, ok := fieldDef.ValidateValue.(string); ok { - if str, ok := value.(string); ok { - // Простая проверка (в реальности использовать regexp) - if !strings.Contains(str, pattern) { - return fmt.Errorf("value '%s' does not match pattern '%s'", str, pattern) - } - } - } - } - return nil -} - -// GetSchemaStatus возвращает статус схемы -func (sm *SchemaMigrator) GetSchemaStatus() *SchemaStatus { - sm.mu.RLock() - defer sm.mu.RUnlock() - - status := &SchemaStatus{ - CurrentVersion: sm.schema.Version, - TotalCollections: len(sm.schema.Collections), - TotalMigrations: len(sm.migrations), - AppliedMigrations: len(sm.applied), - PendingMigrations: len(sm.migrations) - len(sm.applied), - UpdatedAt: sm.schema.UpdatedAt, - } - - for name, collSchema := range sm.schema.Collections { - status.Collections = append(status.Collections, CollectionSchemaInfo{ - Name: name, - FieldsCount: len(collSchema.Fields), - IndexesCount: len(collSchema.Indexes), - UpdatedAt: collSchema.UpdatedAt, - }) - } - - return status -} - -// SchemaStatus статус схемы -type SchemaStatus struct { - CurrentVersion string `json:"current_version"` - TotalCollections int `json:"total_collections"` - TotalMigrations int `json:"total_migrations"` - AppliedMigrations int `json:"applied_migrations"` - PendingMigrations int `json:"pending_migrations"` - UpdatedAt int64 `json:"updated_at"` - Collections []CollectionSchemaInfo `json:"collections"` -} - -// CollectionSchemaInfo информация о схеме коллекции -type CollectionSchemaInfo struct { - Name string `json:"name"` - FieldsCount int `json:"fields_count"` - IndexesCount int `json:"indexes_count"` - UpdatedAt int64 `json:"updated_at"` -} - -// MigrationStatusItem элемент статуса миграции (совместимость со старым кодом) -type MigrationStatusItem struct { - ID string `json:"id"` - Version string `json:"version"` - Description string `json:"description"` - Applied bool `json:"applied"` - AppliedAt int64 `json:"applied_at,omitempty"` - CreatedAt int64 `json:"created_at"` - Success bool `json:"success"` -} - -// MigrationStatus статус миграций (совместимость со старым кодом) -type MigrationStatus struct { - CurrentVersion string `json:"current_version"` - TotalMigrations int `json:"total_migrations"` - AppliedMigrations int `json:"applied_migrations"` - PendingMigrations int `json:"pending_migrations"` - Migrations []MigrationStatusItem `json:"migrations"` -} - -// GetStatus возвращает статус миграций (для совместимости со старым кодом) -func (sm *SchemaMigrator) GetStatus() *MigrationStatus { - sm.mu.RLock() - defer sm.mu.RUnlock() - - status := &MigrationStatus{ - CurrentVersion: sm.schema.Version, - TotalMigrations: len(sm.migrations), - AppliedMigrations: len(sm.applied), - PendingMigrations: len(sm.migrations) - len(sm.applied), - Migrations: make([]MigrationStatusItem, 0), - } - - for _, m := range sm.getSortedSchemaMigrations() { - item := MigrationStatusItem{ - ID: m.ID, - Version: m.Version, - Description: m.Description, - CreatedAt: m.CreatedAt, - } - - if record, ok := sm.applied[m.ID]; ok { - item.Applied = true - item.AppliedAt = record.AppliedAt - item.Success = record.Success - } else { - item.Applied = false - } - - status.Migrations = append(status.Migrations, item) - } - - return status -} - -// ========== Совместимость со старым кодом ========== - -// Migration (старая структура для совместимости) -type Migration struct { - ID string - Version string - Description string - Up func(tx *storage.Transaction) error - Down func(tx *storage.Transaction) error - CreatedAt int64 - AppliedAt int64 -} - -// MigrationRecord (старая структура для совместимости) -type MigrationRecord struct { - ID string `json:"id"` - Version string `json:"version"` - Description string `json:"description"` - AppliedAt int64 `json:"applied_at"` - Success bool `json:"success"` - Error string `json:"error,omitempty"` -} - -// RegisterMigration регистрирует миграцию (старый интерфейс) -func (sm *SchemaMigrator) RegisterMigration(m *Migration) { - // Конвертируем в новую структуру - schemaMig := &SchemaMigration{ - ID: m.ID, - Version: m.Version, - Description: m.Description, - Up: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return m.Up(tx) - }, - Down: func(tx *storage.Transaction, schema *SchemaDefinition) error { - return m.Down(tx) - }, - CreatedAt: m.CreatedAt, - } - sm.RegisterSchemaMigration(schemaMig) -} - -// Migrate выполняет миграцию (старый интерфейс) -func (sm *SchemaMigrator) Migrate(targetVersion string) error { - return sm.MigrateSchema(targetVersion) -}