Delete internal/storage/document.go

2026-10-03 21:14:12 +00:00
1 parent 11e79007cc
commit d0613d6aa1
@@ -1,777 +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/document.go
 // Назначение: Определение структуры документа, его методов для работы
 // с полями, кортежами (вложенными документами) и сериализации в MessagePack.
 // Документ является основной единицей хранения в СУБД futriis.
 // Lock-free: Document.fields переведён на atomic.Value для wait-free доступа.
 //
 // ИСПРАВЛЕНО (2026): Устранён race condition в Document.
 // Все методы, читающие/записывающие метаданные (ID, CreatedAt, UpdatedAt,
 // DeletedAt, Version, Compressed, OriginalSize), теперь используют
 // sync.RWMutex для защиты от гонок.
 //
 // ИЗМЕНЕНО (2026): Убраны вызовы AuditFieldOperation из SetField/DeleteField
 // (они были неинформативными — db/coll пустые). Аудит выполняется на уровне
 // Collection (AuditDocumentOperation).
 //
 // ИЗМЕНЕНО (2026): Удалены поля Version, Compressed, OriginalSize из
 // MVCC-контекста — они остались, но больше не используются для версионирования.
 // Версия документа увеличивается при каждом изменении, но time-travel
 // реализован через WAL (см. transactions.go: GetDocumentAtTimestamp).
 package storage
 import (
 	"fmt"
 	"math/rand"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 	"futriis/internal/compression"
 	"futriis/internal/serializer"
 	"github.com/google/uuid"
 )
 // =============================================================================
 // КОНСТАНТЫ ДЛЯ ЗАЩИТЫ ОТ LIVELOCK
 // =============================================================================
 const (
 	// Максимальное количество попыток CAS перед применением backoff
 	maxCASRetriesBeforeBackoff = 10
 	// Базовая задержка backoff в наносекундах
 	baseBackoffNanos = 100
 	// Максимальная задержка backoff в наносекундах
 	maxBackoffNanos = 10000
 )
 // =============================================================================
 // FIELD VERSION - ЗАЩИТА ОТ ABA
 // =============================================================================
 // fieldVersionWrapper оборачивает map полей с версией для защиты от ABA.
 type fieldVersionWrapper struct {
 	fields  map[string]interface{}
 	version uint64
 }
 // Document представляет документ в коллекции (аналог строки в реляционной СУБД).
 //
 // ИСПРАВЛЕНО: добавлен sync.RWMutex для защиты метаданных документа
 // (ID, CreatedAt, UpdatedAt, DeletedAt, Version, Compressed, OriginalSize)
 // от гонок при конкурентном чтении/записи.
 type Document struct {
 	mu sync.RWMutex // защита метаданных документа
 	ID           string       `msgpack:"_id"`           // Уникальный идентификатор документа
 	fieldsPtr    atomic.Value // *fieldVersionWrapper - lock-free хранилище полей с версией
 	CreatedAt    int64        `msgpack:"created_at"`    // Время создания (Unix миллисекунды)
 	UpdatedAt    int64        `msgpack:"updated_at"`    // Время последнего обновления
 	DeletedAt    int64        `msgpack:"deleted_at"`    // Время удаления (Unix миллисекунды, 0 = не удалён)
 	Version      uint64       `msgpack:"version"`       // Версия документа
 	Compressed   bool         `msgpack:"compressed"`    // Флаг, сжат ли документ
 	OriginalSize int64        `msgpack:"original_size"` // Оригинальный размер до сжатия
 }
 // Tuple представляет вложенный документ (аналог кортежа).
 type Tuple struct {
 	Fields    map[string]interface{} `msgpack:"fields"`
 	CreatedAt int64                  `msgpack:"created_at"`
 	UpdatedAt int64                  `msgpack:"updated_at"`
 	mu        sync.RWMutex
 }
 // Field представляет отдельное поле документа.
 type Field struct {
 	Name      string      `msgpack:"name"`
 	Type      FieldType   `msgpack:"type"`
 	Value     interface{} `msgpack:"value"`
 	UpdatedAt int64       `msgpack:"updated_at"`
 }
 // FieldType определяет тип поля документа.
 type FieldType int
 const (
 	TypeString FieldType = iota
 	TypeNumber
 	TypeBoolean
 	TypeTuple
 	TypeArray
 	TypeNull
 )
 // NewDocument создаёт новый документ с автоматической генерацией ID.
 func NewDocument() *Document {
 	now := time.Now().UnixMilli()
 	d := &Document{
 		ID:           uuid.New().String(),
 		CreatedAt:    now,
 		UpdatedAt:    now,
 		DeletedAt:    0,
 		Version:      1,
 		Compressed:   false,
 		OriginalSize: 0,
 	}
 	d.fieldsPtr.Store(&fieldVersionWrapper{
 		fields:  make(map[string]interface{}),
 		version: 0,
 	})
 	return d
 }
 // NewDocumentWithID создаёт документ с указанным ID.
 func NewDocumentWithID(id string) *Document {
 	now := time.Now().UnixMilli()
 	d := &Document{
 		ID:           id,
 		CreatedAt:    now,
 		UpdatedAt:    now,
 		DeletedAt:    0,
 		Version:      1,
 		Compressed:   false,
 		OriginalSize: 0,
 	}
 	d.fieldsPtr.Store(&fieldVersionWrapper{
 		fields:  make(map[string]interface{}),
 		version: 0,
 	})
 	return d
 }
 // loadFieldsWrapper загружает обёртку полей с версией (lock-free).
 func (d *Document) loadFieldsWrapper() *fieldVersionWrapper {
 	val := d.fieldsPtr.Load()
 	if val == nil {
 		return &fieldVersionWrapper{
 			fields:  make(map[string]interface{}),
 			version: 0,
 		}
 	}
 	return val.(*fieldVersionWrapper)
 }
 // loadFields загружает карту полей (lock-free).
 // ВНИМАНИЕ: возвращает ссылку на внутреннюю map. Не мутировать!
 func (d *Document) loadFields() map[string]interface{} {
 	wrapper := d.loadFieldsWrapper()
 	if wrapper == nil || wrapper.fields == nil {
 		return make(map[string]interface{})
 	}
 	return wrapper.fields
 }
 // storeFields сохраняет карту полей (lock-free).
 func (d *Document) storeFields(newMap map[string]interface{}) {
 	oldWrapper := d.loadFieldsWrapper()
 	newVersion := uint64(0)
 	if oldWrapper != nil {
 		newVersion = oldWrapper.version + 1
 	}
 	d.fieldsPtr.Store(&fieldVersionWrapper{
 		fields:  newMap,
 		version: newVersion,
 	})
 }
 // casFields выполняет CAS операцию для полей с защитой от ABA.
 func (d *Document) casFields(oldWrapper, newWrapper *fieldVersionWrapper) bool {
 	if oldWrapper == nil || newWrapper == nil {
 		return false
 	}
 	if newWrapper.version <= oldWrapper.version {
 		newWrapper.version = oldWrapper.version + 1
 	}
 	return d.fieldsPtr.CompareAndSwap(oldWrapper, newWrapper)
 }
 // casFieldsWithBackoff выполняет CAS с backoff для предотвращения livelock.
 func (d *Document) casFieldsWithBackoff(updateFunc func(map[string]interface{}) map[string]interface{}) bool {
 	attempt := 0
 	for {
 		oldWrapper := d.loadFieldsWrapper()
 		newFields := make(map[string]interface{})
 		for k, v := range oldWrapper.fields {
 			newFields[k] = v
 		}
 		newFields = updateFunc(newFields)
 		newWrapper := &fieldVersionWrapper{
 			fields:  newFields,
 			version: oldWrapper.version + 1,
 		}
 		if d.casFields(oldWrapper, newWrapper) {
 			return true
 		}
 		attempt++
 		if attempt > maxCASRetriesBeforeBackoff {
 			backoff := baseBackoffNanos * (1 << min(attempt-maxCASRetriesBeforeBackoff, 10))
 			if backoff > maxBackoffNanos {
 				backoff = maxBackoffNanos
 			}
 			jitter := rand.Int63n(int64(backoff)/2 + 1)
 			totalBackoff := backoff + int(jitter)
 			time.Sleep(time.Duration(totalBackoff))
 		}
 		if attempt > 1000 {
 			return false
 		}
 	}
 }
 // min возвращает минимальное из двух чисел.
 func min(a, b int) int {
 	if a < b {
 		return a
 	}
 	return b
 }
 // SetField устанавливает значение поля документа (lock-free).
 //
 // ИЗМЕНЕНО: убран вызов AuditFieldOperation (был неинформативным).
 func (d *Document) SetField(name string, value interface{}) {
 	d.casFieldsWithBackoff(func(fields map[string]interface{}) map[string]interface{} {
 		fields[name] = value
 		return fields
 	})
 	d.mu.Lock()
 	d.UpdatedAt = time.Now().UnixMilli()
 	d.Version++
 	d.Compressed = false
 	d.mu.Unlock()
 }
 // GetField возвращает значение поля документа (lock-free).
 func (d *Document) GetField(name string) (interface{}, error) {
 	fields := d.loadFields()
 	if val, ok := fields[name]; ok {
 		return val, nil
 	}
 	return nil, fmt.Errorf("field not found: %s", name)
 }
 // DeleteField удаляет поле из документа (lock-free).
 //
 // ИЗМЕНЕНО: убран вызов AuditFieldOperation.
 func (d *Document) DeleteField(name string) {
 	d.casFieldsWithBackoff(func(fields map[string]interface{}) map[string]interface{} {
 		delete(fields, name)
 		return fields
 	})
 	d.mu.Lock()
 	d.UpdatedAt = time.Now().UnixMilli()
 	d.Version++
 	d.Compressed = false
 	d.mu.Unlock()
 }
 // HasField проверяет наличие поля в документе (lock-free).
 func (d *Document) HasField(name string) bool {
 	fields := d.loadFields()
 	_, ok := fields[name]
 	return ok
 }
 // GetFields возвращает копию всех полей документа (lock-free).
 func (d *Document) GetFields() map[string]interface{} {
 	fields := d.loadFields()
 	copy := make(map[string]interface{})
 	for k, v := range fields {
 		copy[k] = v
 	}
 	return copy
 }
 // ToMap возвращает полное представление документа в виде map.
 func (d *Document) ToMap() map[string]interface{} {
 	fields := d.GetFields()
 	result := make(map[string]interface{})
 	for k, v := range fields {
 		result[k] = v
 	}
 	d.mu.RLock()
 	result["_id"] = d.ID
 	result["_created_at"] = d.CreatedAt
 	result["_updated_at"] = d.UpdatedAt
 	result["_deleted_at"] = d.DeletedAt
 	result["_version"] = d.Version
 	d.mu.RUnlock()
 	return result
 }
 // SetTuple устанавливает вложенный документ (кортеж) в поле.
 func (d *Document) SetTuple(fieldName string, tuple *Tuple) {
 	d.SetField(fieldName, tuple)
 }
 // GetTuple возвращает вложенный документ из поля.
 func (d *Document) GetTuple(fieldName string) (*Tuple, error) {
 	val, err := d.GetField(fieldName)
 	if err != nil {
 		return nil, err
 	}
 	if tuple, ok := val.(*Tuple); ok {
 		return tuple, nil
 	}
 	return nil, fmt.Errorf("field %s is not a tuple", fieldName)
 }
 // Serialize сериализует документ в MessagePack с поддержкой сжатия.
 func (d *Document) Serialize() ([]byte, error) {
 	fields := d.GetFields()
 	d.mu.RLock()
 	docCopy := &Document{
 		ID:           d.ID,
 		CreatedAt:    d.CreatedAt,
 		UpdatedAt:    d.UpdatedAt,
 		DeletedAt:    d.DeletedAt,
 		Version:      d.Version,
 		Compressed:   d.Compressed,
 		OriginalSize: d.OriginalSize,
 	}
 	d.mu.RUnlock()
 	docCopy.storeFields(fields)
 	data, err := serializer.Marshal(docCopy)
 	if err != nil {
 		return nil, err
 	}
 	return data, nil
 }
 // SerializeCompressed сериализует и сжимает документ.
 func (d *Document) SerializeCompressed(compressionConfig *compression.Config) ([]byte, error) {
 	data, err := d.Serialize()
 	if err != nil {
 		return nil, err
 	}
 	if compressionConfig != nil && compressionConfig.Enabled && len(data) >= compressionConfig.MinSize {
 		compressed, err := compression.Compress(data, compressionConfig)
 		if err != nil {
 			return data, nil
 		}
 		return compressed, nil
 	}
 	return data, nil
 }
 // Deserialize десериализует документ из MessagePack.
 func (d *Document) Deserialize(data []byte) error {
 	decompressed, err := compression.DecompressAuto(data)
 	if err == nil && len(decompressed) < len(data) {
 		var doc Document
 		if err := serializer.Unmarshal(decompressed, &doc); err != nil {
 			return err
 		}
 		d.mu.Lock()
 		d.ID = doc.ID
 		d.CreatedAt = doc.CreatedAt
 		d.UpdatedAt = doc.UpdatedAt
 		d.DeletedAt = doc.DeletedAt
 		d.Version = doc.Version
 		d.Compressed = true
 		d.OriginalSize = int64(len(decompressed))
 		d.mu.Unlock()
 		d.storeFields(doc.loadFields())
 	} else {
 		var doc Document
 		if err := serializer.Unmarshal(data, &doc); err != nil {
 			return err
 		}
 		d.mu.Lock()
 		d.ID = doc.ID
 		d.CreatedAt = doc.CreatedAt
 		d.UpdatedAt = doc.UpdatedAt
 		d.DeletedAt = doc.DeletedAt
 		d.Version = doc.Version
 		d.Compressed = false
 		d.OriginalSize = 0
 		d.mu.Unlock()
 		d.storeFields(doc.loadFields())
 	}
 	d.mu.Lock()
 	d.UpdatedAt = time.Now().UnixMilli()
 	d.mu.Unlock()
 	return nil
 }
 // Clone создаёт глубокую копию документа (lock-free).
 func (d *Document) Clone() *Document {
 	fields := d.GetFields()
 	d.mu.RLock()
 	clone := &Document{
 		ID:           d.ID,
 		CreatedAt:    d.CreatedAt,
 		UpdatedAt:    d.UpdatedAt,
 		DeletedAt:    d.DeletedAt,
 		Version:      d.Version,
 		Compressed:   d.Compressed,
 		OriginalSize: d.OriginalSize,
 	}
 	d.mu.RUnlock()
 	copiedFields := make(map[string]interface{})
 	for k, v := range fields {
 		copiedFields[k] = deepCopyValue(v)
 	}
 	clone.storeFields(copiedFields)
 	return clone
 }
 // Update применяет обновление к документу (атомарно, lock-free).
 func (d *Document) Update(updates map[string]interface{}) error {
 	success := d.casFieldsWithBackoff(func(fields map[string]interface{}) map[string]interface{} {
 		for k, v := range updates {
 			fields[k] = v
 		}
 		return fields
 	})
 	if success {
 		d.mu.Lock()
 		d.UpdatedAt = time.Now().UnixMilli()
 		d.Version++
 		d.Compressed = false
 		d.mu.Unlock()
 		return nil
 	}
 	return fmt.Errorf("failed to update document after multiple attempts (possible high contention)")
 }
 // SoftDelete мягко удаляет документ.
 func (d *Document) SoftDelete() {
 	d.mu.Lock()
 	d.DeletedAt = time.Now().UnixMilli()
 	d.UpdatedAt = d.DeletedAt
 	d.Version++
 	d.mu.Unlock()
 }
 // IsDeleted проверяет, удалён ли документ (мягкое удаление).
 func (d *Document) IsDeleted() bool {
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 	return d.DeletedAt > 0
 }
 // Restore восстанавливает мягко удалённый документ.
 func (d *Document) Restore() {
 	d.mu.Lock()
 	d.DeletedAt = 0
 	d.UpdatedAt = time.Now().UnixMilli()
 	d.Version++
 	d.mu.Unlock()
 }
 // GetDeletedAtStr возвращает человекочитаемую строку времени удаления.
 func (d *Document) GetDeletedAtStr() string {
 	d.mu.RLock()
 	deletedAt := d.DeletedAt
 	d.mu.RUnlock()
 	if deletedAt == 0 {
 		return ""
 	}
 	return time.UnixMilli(deletedAt).Format("2006-01-02 15:04:05.000")
 }
 // GetCreatedAtStr возвращает человекочитаемую строку времени создания.
 func (d *Document) GetCreatedAtStr() string {
 	d.mu.RLock()
 	createdAt := d.CreatedAt
 	d.mu.RUnlock()
 	return time.UnixMilli(createdAt).Format("2006-01-02 15:04:05.000")
 }
 // GetUpdatedAtStr возвращает человекочитаемую строку времени обновления.
 func (d *Document) GetUpdatedAtStr() string {
 	d.mu.RLock()
 	updatedAt := d.UpdatedAt
 	d.mu.RUnlock()
 	return time.UnixMilli(updatedAt).Format("2006-01-02 15:04:05.000")
 }
 // Compress сжимает документ в памяти.
 func (d *Document) Compress(config *compression.Config) error {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	if d.Compressed {
 		return nil
 	}
 	fields := d.loadFields()
 	originalSize := len(fields)
 	if originalSize < config.MinSize {
 		return nil
 	}
 	d.Compressed = true
 	d.OriginalSize = int64(originalSize)
 	return nil
 }
 // Decompress распаковывает документ в памяти.
 func (d *Document) Decompress() error {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	if !d.Compressed {
 		return nil
 	}
 	d.Compressed = false
 	d.OriginalSize = 0
 	return nil
 }
 // GetCompressionRatio возвращает коэффициент сжатия.
 func (d *Document) GetCompressionRatio() float64 {
 	d.mu.RLock()
 	compressed := d.Compressed
 	originalSize := d.OriginalSize
 	d.mu.RUnlock()
 	if !compressed || originalSize == 0 {
 		return 1.0
 	}
 	fields := d.loadFields()
 	currentSize := len(fields)
 	return float64(currentSize) / float64(originalSize)
 }
 // GetMetadata возвращает метаданные документа (временные метки).
 func (d *Document) GetMetadata() map[string]int64 {
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 	return map[string]int64{
 		"created_at": d.CreatedAt,
 		"updated_at": d.UpdatedAt,
 		"deleted_at": d.DeletedAt,
 		"version":    int64(d.Version),
 	}
 }
 // deepCopyValue выполняет глубокое копирование значения.
 func deepCopyValue(val interface{}) interface{} {
 	switch v := val.(type) {
 	case *Tuple:
 		return v.Clone()
 	case map[string]interface{}:
 		copy := make(map[string]interface{})
 		for k, val := range v {
 			copy[k] = deepCopyValue(val)
 		}
 		return copy
 	case []interface{}:
 		copy := make([]interface{}, len(v))
 		for i, val := range v {
 			copy[i] = deepCopyValue(val)
 		}
 		return copy
 	default:
 		return v
 	}
 }
 // NewTuple создаёт новый вложенный документ (кортеж).
 func NewTuple() *Tuple {
 	now := time.Now().UnixMilli()
 	return &Tuple{
 		Fields:    make(map[string]interface{}),
 		CreatedAt: now,
 		UpdatedAt: now,
 	}
 }
 // Set устанавливает поле во вложенном документе.
 func (t *Tuple) Set(name string, value interface{}) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	t.Fields[name] = value
 	t.UpdatedAt = time.Now().UnixMilli()
 }
 // Get возвращает поле из вложенного документа.
 func (t *Tuple) Get(name string) (interface{}, error) {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 	if val, ok := t.Fields[name]; ok {
 		return val, nil
 	}
 	return nil, fmt.Errorf("tuple field not found: %s", name)
 }
 // Clone создаёт копию кортежа.
 func (t *Tuple) Clone() *Tuple {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 	clone := NewTuple()
 	for k, v := range t.Fields {
 		clone.Fields[k] = deepCopyValue(v)
 	}
 	clone.CreatedAt = t.CreatedAt
 	clone.UpdatedAt = t.UpdatedAt
 	return clone
 }
 // ToMap конвертирует кортеж в map.
 func (t *Tuple) ToMap() map[string]interface{} {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 	copy := make(map[string]interface{})
 	for k, v := range t.Fields {
 		copy[k] = v
 	}
 	return copy
 }
 // GetTupleMetadata возвращает метаданные кортежа.
 func (t *Tuple) GetTupleMetadata() map[string]int64 {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 	return map[string]int64{
 		"created_at": t.CreatedAt,
 		"updated_at": t.UpdatedAt,
 	}
 }
 // GetNestedField получает значение по точечному пути (например, "user.address.city").
 func (d *Document) GetNestedField(path string) (interface{}, error) {
 	parts := strings.Split(path, ".")
 	if len(parts) == 0 {
 		return nil, fmt.Errorf("empty path")
 	}
 	current := interface{}(d)
 	for _, part := range parts {
 		switch v := current.(type) {
 		case *Document:
 			val, err := v.GetField(part)
 			if err != nil {
 				return nil, err
 			}
 			current = val
 		case *Tuple:
 			val, err := v.Get(part)
 			if err != nil {
 				return nil, err
 			}
 			current = val
 		case map[string]interface{}:
 			if val, ok := v[part]; ok {
 				current = val
 			} else {
 				return nil, fmt.Errorf("field not found: %s", part)
 			}
 		default:
 			return nil, fmt.Errorf("cannot navigate into non-document value at %s", part)
 		}
 	}
 	return current, nil
 }
 // SetNestedField устанавливает значение по точечному пути.
 func (d *Document) SetNestedField(path string, value interface{}) error {
 	parts := strings.Split(path, ".")
 	if len(parts) == 0 {
 		return fmt.Errorf("empty path")
 	}
 	if len(parts) == 1 {
 		d.SetField(parts[0], value)
 		return nil
 	}
 	var current interface{} = d
 	for i := 0; i < len(parts)-1; i++ {
 		part := parts[i]
 		switch v := current.(type) {
 		case *Document:
 			if !v.HasField(part) {
 				newTuple := NewTuple()
 				v.SetField(part, newTuple)
 				current = newTuple
 			} else {
 				field, _ := v.GetField(part)
 				if tuple, ok := field.(*Tuple); ok {
 					current = tuple
 				} else {
 					return fmt.Errorf("field %s is not a tuple", part)
 				}
 			}
 		case *Tuple:
 			if val, err := v.Get(part); err == nil {
 				if tuple, ok := val.(*Tuple); ok {
 					current = tuple
 				} else {
 					return fmt.Errorf("field %s is not a tuple", part)
 				}
 			} else {
 				newTuple := NewTuple()
 				v.Set(part, newTuple)
 				current = newTuple
 			}
 		default:
 			return fmt.Errorf("cannot set nested field on non-document value")
 		}
 	}
 	lastPart := parts[len(parts)-1]
 	switch v := current.(type) {
 	case *Document:
 		v.SetField(lastPart, value)
 	case *Tuple:
 		v.Set(lastPart, value)
 	default:
 		return fmt.Errorf("cannot set field on non-document value")
 	}
 	d.mu.Lock()
 	d.UpdatedAt = time.Now().UnixMilli()
 	d.Compressed = false
 	d.mu.Unlock()
 	return nil
 }