gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Upload files to "internal/storage"

Browse Source
This commit is contained in:
Григорий Сафронов
2026-05-24 21:49:35 +00:00
parent a387756a6e
commit fbe3c3ecfe
4 changed files with 2478 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

184
internal/storage/audit.go Normal file
View File

@@ -0,0 +1,184 @@
/*
* 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/audit.go
// Назначение: Аудит всех операций создания, изменения, удаления данных
// с записью временной метки с точностью до миллисекунды
package storage
import (
"fmt"
"sync"
"time"
)
// AuditEntry представляет запись аудита
type AuditEntry struct {
ID string `msgpack:"id"`
Timestamp int64 `msgpack:"timestamp"` // Unix миллисекунды
TimestampStr string `msgpack:"timestamp_str"` // Человекочитаемая строка
Operation string `msgpack:"operation"` // CREATE, UPDATE, DELETE, START, COMMIT, ABORT, CLUSTER, SOFT_DELETE, RESTORE, PERMANENT_DELETE
DataType string `msgpack:"data_type"` // DATABASE, COLLECTION, DOCUMENT, FIELD, TUPLE, SESSION, TRANSACTION, CLUSTER, INDEX
Name string `msgpack:"name"` // Имя объекта
Details map[string]interface{} `msgpack:"details"` // Детали операции
}
// AuditLogger управляет аудитом
type AuditLogger struct {
entries []AuditEntry
mu sync.RWMutex
}
var globalAuditLogger = &AuditLogger{
entries: make([]AuditEntry, 0),
}
// GetCurrentTimestamp возвращает текущую временную метку с миллисекундами
func GetCurrentTimestamp() (int64, string) {
now := time.Now()
timestampMs := now.UnixMilli()
timestampStr := now.Format("2006-01-02 15:04:05.000")
return timestampMs, timestampStr
}
// LogAudit записывает событие в аудит
func LogAudit(operation, dataType, name string, details map[string]interface{}) {
timestampMs, timestampStr := GetCurrentTimestamp()
// Если details не содержат timestamp, добавляем его
if details == nil {
details = make(map[string]interface{})
}
if _, ok := details["audit_timestamp"]; !ok {
details["audit_timestamp"] = timestampMs
details["audit_timestamp_str"] = timestampStr
}
entry := AuditEntry{
ID: fmt.Sprintf("%d", timestampMs),
Timestamp: timestampMs,
TimestampStr: timestampStr,
Operation: operation,
DataType: dataType,
Name: name,
Details: details,
}
globalAuditLogger.mu.Lock()
globalAuditLogger.entries = append(globalAuditLogger.entries, entry)
globalAuditLogger.mu.Unlock()
}
// GetAuditLog возвращает копию лога аудита
func GetAuditLog() []AuditEntry {
globalAuditLogger.mu.RLock()
defer globalAuditLogger.mu.RUnlock()
result := make([]AuditEntry, len(globalAuditLogger.entries))
copy(result, globalAuditLogger.entries)
return result
}
// GetAuditLogFiltered возвращает отфильтрованный лог аудита
func GetAuditLogFiltered(dataType, operation string, fromTime, toTime int64) []AuditEntry {
globalAuditLogger.mu.RLock()
defer globalAuditLogger.mu.RUnlock()
result := make([]AuditEntry, 0)
for _, entry := range globalAuditLogger.entries {
if dataType != "" && entry.DataType != dataType {
continue
}
if operation != "" && entry.Operation != operation {
continue
}
if fromTime > 0 && entry.Timestamp < fromTime {
continue
}
if toTime > 0 && entry.Timestamp > toTime {
continue
}
result = append(result, entry)
}
return result
}
// ClearAuditLog очищает лог аудита (только для отладки)
func ClearAuditLog() {
globalAuditLogger.mu.Lock()
defer globalAuditLogger.mu.Unlock()
globalAuditLogger.entries = make([]AuditEntry, 0)
}
// GetAuditLogSize возвращает количество записей в логе аудита
func GetAuditLogSize() int {
globalAuditLogger.mu.RLock()
defer globalAuditLogger.mu.RUnlock()
return len(globalAuditLogger.entries)
}
// AuditDatabaseOperation логирует операцию с базой данных
func AuditDatabaseOperation(operation, dbName string) {
LogAudit(operation, "DATABASE", dbName, map[string]interface{}{
"database": dbName,
})
}
// AuditCollectionOperation логирует операцию с коллекцией
func AuditCollectionOperation(operation, dbName, collName string, settings interface{}) {
LogAudit(operation, "COLLECTION", fmt.Sprintf("%s.%s", dbName, collName), map[string]interface{}{
"database": dbName,
"collection": collName,
"settings": settings,
})
}
// AuditDocumentOperation логирует операцию с документом
func AuditDocumentOperation(operation, dbName, collName, docID string, fields map[string]interface{}) {
LogAudit(operation, "DOCUMENT", fmt.Sprintf("%s.%s.%s", dbName, collName, docID), map[string]interface{}{
"database": dbName,
"collection": collName,
"document_id": docID,
"fields": fields,
})
}
// AuditFieldOperation логирует операцию с полем
func AuditFieldOperation(operation, dbName, collName, docID, fieldName string, value interface{}) {
LogAudit(operation, "FIELD", fmt.Sprintf("%s.%s.%s.%s", dbName, collName, docID, fieldName), map[string]interface{}{
"database": dbName,
"collection": collName,
"document_id": docID,
"field": fieldName,
"value": value,
})
}
// AuditTupleOperation логирует операцию с кортежем
func AuditTupleOperation(operation, dbName, collName, docID, tuplePath string) {
LogAudit(operation, "TUPLE", fmt.Sprintf("%s.%s.%s.%s", dbName, collName, docID, tuplePath), map[string]interface{}{
"database": dbName,
"collection": collName,
"document_id": docID,
"tuple_path": tuplePath,
})
}
// AuditIndexOperation логирует операцию с индексом
func AuditIndexOperation(operation, dbName, collName, indexName string, fields []string, unique bool) {
LogAudit(operation, "INDEX", fmt.Sprintf("%s.%s.%s", dbName, collName, indexName), map[string]interface{}{
"database": dbName,
"collection": collName,
"index_name": indexName,
"fields": fields,
"unique": unique,
})
}

1136
internal/storage/collection.go Normal file
View File

File diff suppressed because it is too large Load Diff

584
internal/storage/document.go Normal file
View File

@@ -0,0 +1,584 @@
/*
* 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.
package storage
import (
"fmt"
"strings"
"sync"
"time"
"futriis/internal/compression"
"futriis/internal/serializer"
"github.com/google/uuid"
)
// Document представляет документ в коллекции (аналог строки в реляционной СУБД)
type Document struct {
ID string `msgpack:"_id"` // Уникальный идентификатор документа
Fields map[string]interface{} `msgpack:"fields"` // Поля документа (аналог колонок)
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"` // Оригинальный размер до сжатия
mu sync.RWMutex `msgpack:"-"` // Блокировка для wait-free операций
}
// 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()
return &Document{
ID: uuid.New().String(),
Fields: make(map[string]interface{}),
CreatedAt: now,
UpdatedAt: now,
DeletedAt: 0,
Version: 1,
Compressed: false,
OriginalSize: 0,
}
}
// NewDocumentWithID создаёт документ с указанным ID
func NewDocumentWithID(id string) *Document {
now := time.Now().UnixMilli()
return &Document{
ID: id,
Fields: make(map[string]interface{}),
CreatedAt: now,
UpdatedAt: now,
DeletedAt: 0,
Version: 1,
Compressed: false,
OriginalSize: 0,
}
}
// SetField устанавливает значение поля документа (wait-free)
func (d *Document) SetField(name string, value interface{}) {
d.mu.Lock()
defer d.mu.Unlock()
d.Fields[name] = value
d.UpdatedAt = time.Now().UnixMilli()
d.Version++
d.Compressed = false // При изменении документа снимаем флаг сжатия
// Аудит изменения поля
AuditFieldOperation("UPDATE", "", "", d.ID, name, value)
}
// GetField возвращает значение поля документа
func (d *Document) GetField(name string) (interface{}, error) {
d.mu.RLock()
defer d.mu.RUnlock()
if val, ok := d.Fields[name]; ok {
return val, nil
}
return nil, fmt.Errorf("field not found: %s", name)
}
// DeleteField удаляет поле из документа
func (d *Document) DeleteField(name string) {
d.mu.Lock()
defer d.mu.Unlock()
delete(d.Fields, name)
d.UpdatedAt = time.Now().UnixMilli()
d.Version++
d.Compressed = false
// Аудит удаления поля
AuditFieldOperation("DELETE", "", "", d.ID, name, nil)
}
// HasField проверяет наличие поля в документе
func (d *Document) HasField(name string) bool {
d.mu.RLock()
defer d.mu.RUnlock()
_, ok := d.Fields[name]
return ok
}
// GetFields возвращает копию всех полей документа
func (d *Document) GetFields() map[string]interface{} {
d.mu.RLock()
defer d.mu.RUnlock()
copy := make(map[string]interface{})
for k, v := range d.Fields {
copy[k] = v
}
return copy
}
// 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) {
d.mu.RLock()
defer d.mu.RUnlock()
data, err := serializer.Marshal(d)
if err != nil {
return nil, err
}
return data, nil
}
// SerializeCompressed сериализует и сжимает документ
func (d *Document) SerializeCompressed(compressionConfig *compression.Config) ([]byte, error) {
d.mu.RLock()
defer d.mu.RUnlock()
// Сериализуем документ
data, err := serializer.Marshal(d)
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 {
d.mu.Lock()
defer d.mu.Unlock()
// Пытаемся определить, сжаты ли данные
// Для этого пробуем распаковать, если не получается - данные несжатые
decompressed, err := compression.DecompressAuto(data)
if err == nil && len(decompressed) < len(data) {
// Данные были сжаты, используем распакованную версию
if err := serializer.Unmarshal(decompressed, d); err != nil {
return err
}
d.Compressed = true
d.OriginalSize = int64(len(decompressed))
} else {
// Данные не сжаты или не удалось распаковать
if err := serializer.Unmarshal(data, d); err != nil {
return err
}
d.Compressed = false
d.OriginalSize = 0
}
// Обновляем временные метки при десериализации
d.UpdatedAt = time.Now().UnixMilli()
return nil
}
// Clone создаёт глубокую копию документа
func (d *Document) Clone() *Document {
d.mu.RLock()
defer d.mu.RUnlock()
clone := &Document{
ID: d.ID,
Fields: make(map[string]interface{}),
CreatedAt: d.CreatedAt,
UpdatedAt: d.UpdatedAt,
DeletedAt: d.DeletedAt,
Version: d.Version,
Compressed: d.Compressed,
OriginalSize: d.OriginalSize,
}
// Глубокое копирование полей
for k, v := range d.Fields {
clone.Fields[k] = deepCopyValue(v)
}
return clone
}
// Update применяет обновление к документу (атомарно)
func (d *Document) Update(updates map[string]interface{}) error {
d.mu.Lock()
defer d.mu.Unlock()
for k, v := range updates {
d.Fields[k] = v
}
d.UpdatedAt = time.Now().UnixMilli()
d.Version++
d.Compressed = false // После обновления документ больше не сжат
return nil
}
// SoftDelete мягко удаляет документ (устанавливает метку времени удаления)
func (d *Document) SoftDelete() {
d.mu.Lock()
defer d.mu.Unlock()
d.DeletedAt = time.Now().UnixMilli()
d.UpdatedAt = d.DeletedAt
d.Version++
}
// 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()
defer d.mu.Unlock()
d.DeletedAt = 0
d.UpdatedAt = time.Now().UnixMilli()
d.Version++
}
// GetDeletedAtStr возвращает человекочитаемую строку времени удаления
func (d *Document) GetDeletedAtStr() string {
d.mu.RLock()
defer d.mu.RUnlock()
if d.DeletedAt == 0 {
return ""
}
return time.UnixMilli(d.DeletedAt).Format("2006-01-02 15:04:05.000")
}
// GetCreatedAtStr возвращает человекочитаемую строку времени создания
func (d *Document) GetCreatedAtStr() string {
d.mu.RLock()
defer d.mu.RUnlock()
return time.UnixMilli(d.CreatedAt).Format("2006-01-02 15:04:05.000")
}
// GetUpdatedAtStr возвращает человекочитаемую строку времени обновления
func (d *Document) GetUpdatedAtStr() string {
d.mu.RLock()
defer d.mu.RUnlock()
return time.UnixMilli(d.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
}
// Сохраняем текущее состояние
originalSize := len(d.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()
defer d.mu.RUnlock()
if !d.Compressed || d.OriginalSize == 0 {
return 1.0
}
currentSize := len(d.Fields)
return float64(currentSize) / float64(d.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
}
// Иначе нужно создать промежуточные структуры
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.UpdatedAt = time.Now().UnixMilli()
d.Compressed = false
return nil
}

574
internal/storage/engine.go Normal file
View File

@@ -0,0 +1,574 @@
/*
* 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/engine.go
// Назначение: In-memory движок хранения документов с поддержкой коллекций,
// слайсов (аналог БД), тапплов (аналог таблиц), полей и кортежей.
// Полностью wait-free с использованием sync.Map и атомарных операций.
package storage
import (
"fmt"
"os"
"sync"
"sync/atomic"
"time"
"futriis/internal/log"
"futriis/internal/serializer"
)
// Storage представляет основное хранилище баз данных
type Storage struct {
databases sync.Map // map[string]*Database
pageSize int64
logger *log.Logger
totalDocs atomic.Int64
createdAt int64
}
// Database представляет базу данных (аналог слайса в реляционных СУБД)
type Database struct {
name string
collections sync.Map // map[string]*Collection
createdAt int64
updatedAt int64
deletedAt int64
mu sync.RWMutex
}
// NewStorage создаёт новый экземпляр хранилища
func NewStorage(pageSizeMB int, logger *log.Logger) *Storage {
return &Storage{
pageSize: int64(pageSizeMB) * 1024 * 1024,
logger: logger,
createdAt: time.Now().UnixMilli(),
}
}
// CreateDatabase создаёт новую базу данных
func (s *Storage) CreateDatabase(name string) error {
now := time.Now().UnixMilli()
db := &Database{
name: name,
createdAt: now,
updatedAt: now,
deletedAt: 0,
}
if _, exists := s.databases.LoadOrStore(name, db); exists {
return fmt.Errorf("database already exists")
}
AuditDatabaseOperation("CREATE", name)
if s.logger != nil {
s.logger.Info("Database created: " + name)
}
return nil
}
// GetDatabase возвращает базу данных по имени
func (s *Storage) GetDatabase(name string) (*Database, error) {
if val, ok := s.databases.Load(name); ok {
return val.(*Database), nil
}
return nil, fmt.Errorf("database not found")
}
// DropDatabase удаляет базу данных
func (s *Storage) DropDatabase(name string) error {
if val, ok := s.databases.Load(name); ok {
db := val.(*Database)
db.mu.Lock()
db.deletedAt = time.Now().UnixMilli()
db.mu.Unlock()
}
if _, ok := s.databases.LoadAndDelete(name); !ok {
return fmt.Errorf("database not found")
}
AuditDatabaseOperation("DROP", name)
if s.logger != nil {
s.logger.Info("Database dropped: " + name)
}
return nil
}
// ListDatabases возвращает список всех баз данных
func (s *Storage) ListDatabases() []string {
databases := make([]string, 0)
s.databases.Range(func(key, value interface{}) bool {
databases = append(databases, key.(string))
return true
})
return databases
}
// Name возвращает имя базы данных
func (db *Database) Name() string {
return db.name
}
// GetCreatedAt возвращает время создания базы данных
func (db *Database) GetCreatedAt() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.createdAt
}
// GetUpdatedAt возвращает время последнего обновления базы данных
func (db *Database) GetUpdatedAt() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.updatedAt
}
// GetDeletedAt возвращает время удаления базы данных (0 = не удалена)
func (db *Database) GetDeletedAt() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.deletedAt
}
// UpdateTimestamp обновляет временную метку базы данных
func (db *Database) UpdateTimestamp() {
db.mu.Lock()
defer db.mu.Unlock()
db.updatedAt = time.Now().UnixMilli()
}
// CreateCollection создаёт новую коллекцию в базе данных
func (db *Database) CreateCollection(name string) error {
if _, exists := db.collections.LoadOrStore(name, NewCollection(db.name, name, nil)); exists {
return fmt.Errorf("collection already exists")
}
db.UpdateTimestamp()
AuditCollectionOperation("CREATE", db.name, name, nil)
return nil
}
// CreateCollectionWithSettings создаёт коллекцию с настройками
func (db *Database) CreateCollectionWithSettings(name string, settings *CollectionSettings) error {
if _, exists := db.collections.LoadOrStore(name, NewCollection(db.name, name, settings)); exists {
return fmt.Errorf("collection already exists")
}
db.UpdateTimestamp()
AuditCollectionOperation("CREATE", db.name, name, settings)
return nil
}
// GetCollection возвращает коллекцию по имени
func (db *Database) GetCollection(name string) (*Collection, error) {
if val, ok := db.collections.Load(name); ok {
return val.(*Collection), nil
}
return nil, fmt.Errorf("collection not found")
}
// DropCollection удаляет коллекцию
func (db *Database) DropCollection(name string) error {
if _, ok := db.collections.LoadAndDelete(name); !ok {
return fmt.Errorf("collection not found")
}
db.UpdateTimestamp()
AuditCollectionOperation("DROP", db.name, name, nil)
return nil
}
// ListCollections возвращает список всех коллекций в базе данных
func (db *Database) ListCollections() []string {
collections := make([]string, 0)
db.collections.Range(func(key, value interface{}) bool {
collections = append(collections, key.(string))
return true
})
return collections
}
// GetTotalDocuments возвращает общее количество документов во всех коллекциях
func (s *Storage) GetTotalDocuments() int64 {
return s.totalDocs.Load()
}
// GetPageSize возвращает размер страницы памяти
func (s *Storage) GetPageSize() int64 {
return s.pageSize
}
// GetCreatedAt возвращает время создания хранилища
func (s *Storage) GetCreatedAt() int64 {
return s.createdAt
}
// SerializeDatabase сериализует всю базу данных в MessagePack
func (db *Database) SerializeDatabase() ([]byte, error) {
dbData := make(map[string]interface{})
db.collections.Range(func(key, value interface{}) bool {
coll := value.(*Collection)
collData := make(map[string]interface{})
docs := coll.GetAllDocuments()
collDocs := make([]*Document, 0, len(docs))
for _, doc := range docs {
collDocs = append(collDocs, doc)
}
collData["documents"] = collDocs
collData["metadata"] = coll.GetMetadata()
collData["timestamps"] = coll.GetTimestamps()
dbData[key.(string)] = collData
return true
})
// Добавляем метаданные базы данных
dbData["_metadata"] = map[string]interface{}{
"name": db.name,
"created_at": db.GetCreatedAt(),
"updated_at": db.GetUpdatedAt(),
"exported_at": time.Now().UnixMilli(),
}
return serializer.Marshal(dbData)
}
// DeserializeDatabase десериализует базу данных из MessagePack
func (db *Database) DeserializeDatabase(data []byte) error {
var dbData map[string]interface{}
if err := serializer.Unmarshal(data, &dbData); err != nil {
return err
}
// Извлекаем метаданные
if metaRaw, ok := dbData["_metadata"]; ok {
if meta, ok := metaRaw.(map[string]interface{}); ok {
if createdAt, ok := meta["created_at"]; ok {
if v, ok := createdAt.(int64); ok {
db.createdAt = v
}
}
}
delete(dbData, "_metadata")
}
for collName, collDataRaw := range dbData {
collData, ok := collDataRaw.(map[string]interface{})
if !ok {
continue
}
settings := &CollectionSettings{
MaxDocuments: 0,
ValidateSchema: false,
AutoIndexID: true,
TTLSeconds: 0,
SoftDelete: false,
}
if metaRaw, ok := collData["metadata"]; ok {
if meta, ok := metaRaw.(map[string]interface{}); ok {
if settingsRaw, ok := meta["settings"]; ok {
if settingsMap, ok := settingsRaw.(map[string]interface{}); ok {
if maxDocs, ok := settingsMap["max_documents"]; ok {
if v, ok := maxDocs.(int); ok {
settings.MaxDocuments = v
} else if v, ok := maxDocs.(int64); ok {
settings.MaxDocuments = int(v)
} else if v, ok := maxDocs.(float64); ok {
settings.MaxDocuments = int(v)
}
}
if validateSchema, ok := settingsMap["validate_schema"]; ok {
if v, ok := validateSchema.(bool); ok {
settings.ValidateSchema = v
}
}
if autoIndexID, ok := settingsMap["auto_index_id"]; ok {
if v, ok := autoIndexID.(bool); ok {
settings.AutoIndexID = v
}
}
if ttlSeconds, ok := settingsMap["ttl_seconds"]; ok {
if v, ok := ttlSeconds.(int); ok {
settings.TTLSeconds = v
} else if v, ok := ttlSeconds.(int64); ok {
settings.TTLSeconds = int(v)
} else if v, ok := ttlSeconds.(float64); ok {
settings.TTLSeconds = int(v)
}
}
if softDelete, ok := settingsMap["soft_delete"]; ok {
if v, ok := softDelete.(bool); ok {
settings.SoftDelete = v
}
}
}
}
} else if meta, ok := metaRaw.(*CollectionMetadata); ok {
if meta.Settings != nil {
settings = meta.Settings
}
}
}
coll := NewCollection(db.name, collName, settings)
// Восстанавливаем временные метки коллекции если они есть
if timestampsRaw, ok := collData["timestamps"]; ok {
if timestamps, ok := timestampsRaw.(map[string]interface{}); ok {
if createdAt, ok := timestamps["created_at"]; ok {
if v, ok := createdAt.(int64); ok {
coll.metadata.CreatedAt = v
}
}
if updatedAt, ok := timestamps["updated_at"]; ok {
if v, ok := updatedAt.(int64); ok {
coll.metadata.UpdatedAt = v
}
}
}
}
if docsRaw, ok := collData["documents"]; ok {
if docs, ok := docsRaw.([]interface{}); ok {
for _, docRaw := range docs {
if doc, ok := docRaw.(*Document); ok {
coll.Insert(doc)
} else if docMap, ok := docRaw.(map[string]interface{}); ok {
doc := NewDocument()
if id, ok := docMap["ID"].(string); ok {
doc.ID = id
} else if id, ok := docMap["id"].(string); ok {
doc.ID = id
}
if fields, ok := docMap["fields"]; ok {
if fieldsMap, ok := fields.(map[string]interface{}); ok {
for k, v := range fieldsMap {
doc.SetField(k, v)
}
}
}
if createdAt, ok := docMap["created_at"]; ok {
if v, ok := createdAt.(int64); ok {
doc.CreatedAt = v
} else if v, ok := createdAt.(float64); ok {
doc.CreatedAt = int64(v)
}
}
if updatedAt, ok := docMap["updated_at"]; ok {
if v, ok := updatedAt.(int64); ok {
doc.UpdatedAt = v
} else if v, ok := updatedAt.(float64); ok {
doc.UpdatedAt = int64(v)
}
}
if deletedAt, ok := docMap["deleted_at"]; ok {
if v, ok := deletedAt.(int64); ok {
doc.DeletedAt = v
} else if v, ok := deletedAt.(float64); ok {
doc.DeletedAt = int64(v)
}
}
if version, ok := docMap["version"]; ok {
if v, ok := version.(uint64); ok {
doc.Version = v
} else if v, ok := version.(int64); ok {
doc.Version = uint64(v)
} else if v, ok := version.(float64); ok {
doc.Version = uint64(v)
}
}
coll.Insert(doc)
}
}
} else if docs, ok := docsRaw.([]*Document); ok {
for _, doc := range docs {
coll.Insert(doc)
}
}
}
db.collections.Store(collName, coll)
AuditCollectionOperation("RESTORE", db.name, collName, settings)
}
db.UpdateTimestamp()
return nil
}
// GetDatabaseNames возвращает имена всех баз данных
func (s *Storage) GetDatabaseNames() []string {
return s.ListDatabases()
}
// ExistsDatabase проверяет существование базы данных
func (s *Storage) ExistsDatabase(name string) bool {
_, ok := s.databases.Load(name)
return ok
}
// GetDatabaseCount возвращает количество баз данных
func (s *Storage) GetDatabaseCount() int {
count := 0
s.databases.Range(func(key, value interface{}) bool {
count++
return true
})
return count
}
// ========== Дополнительные методы для управления хранилищем ==========
// Backup создаёт резервную копию всех данных
func (s *Storage) Backup(backupPath string) error {
if s.logger != nil {
s.logger.Info(fmt.Sprintf("Starting backup to %s", backupPath))
}
backup := make(map[string]interface{})
backup["created_at"] = s.createdAt
backup["backup_time"] = time.Now().UnixMilli()
databases := make(map[string][]byte)
s.databases.Range(func(key, value interface{}) bool {
dbName := key.(string)
db := value.(*Database)
dbData, err := db.SerializeDatabase()
if err != nil {
if s.logger != nil {
s.logger.Error(fmt.Sprintf("Failed to serialize database %s: %v", dbName, err))
}
return false
}
databases[dbName] = dbData
return true
})
backup["databases"] = databases
data, err := serializer.Marshal(backup)
if err != nil {
return fmt.Errorf("failed to marshal backup: %v", err)
}
if err := os.WriteFile(backupPath, data, 0644); err != nil {
return fmt.Errorf("failed to write backup: %v", err)
}
if s.logger != nil {
s.logger.Info(fmt.Sprintf("Backup completed: %s", backupPath))
}
return nil
}
// Restore восстанавливает данные из резервной копии
func (s *Storage) Restore(backupPath string) error {
if s.logger != nil {
s.logger.Info(fmt.Sprintf("Starting restore from %s", backupPath))
}
data, err := os.ReadFile(backupPath)
if err != nil {
return fmt.Errorf("failed to read backup file: %v", err)
}
var backup map[string]interface{}
if err := serializer.Unmarshal(data, &backup); err != nil {
return fmt.Errorf("failed to unmarshal backup: %v", err)
}
// Восстанавливаем временную метку хранилища
if createdAt, ok := backup["created_at"]; ok {
if v, ok := createdAt.(int64); ok {
s.createdAt = v
}
}
databasesRaw, ok := backup["databases"]
if !ok {
return fmt.Errorf("invalid backup format: missing databases")
}
databases, ok := databasesRaw.(map[string]interface{})
if !ok {
return fmt.Errorf("invalid backup format: databases is not a map")
}
for dbName, dbDataRaw := range databases {
dbData, ok := dbDataRaw.([]byte)
if !ok {
continue
}
if !s.ExistsDatabase(dbName) {
if err := s.CreateDatabase(dbName); err != nil {
return fmt.Errorf("failed to create database %s: %v", dbName, err)
}
}
db, err := s.GetDatabase(dbName)
if err != nil {
return fmt.Errorf("failed to get database %s: %v", dbName, err)
}
if err := db.DeserializeDatabase(dbData); err != nil {
return fmt.Errorf("failed to restore database %s: %v", dbName, err)
}
}
if s.logger != nil {
s.logger.Info("Restore completed")
}
return nil
}
// GetStats возвращает статистику хранилища
func (s *Storage) GetStats() map[string]interface{} {
stats := map[string]interface{}{
"total_databases": s.GetDatabaseCount(),
"total_documents": s.GetTotalDocuments(),
"page_size_bytes": s.pageSize,
"created_at": s.createdAt,
"created_at_str": time.UnixMilli(s.createdAt).Format("2006-01-02 15:04:05.000"),
}
databases := make([]map[string]interface{}, 0)
s.databases.Range(func(key, value interface{}) bool {
db := value.(*Database)
dbStats := map[string]interface{}{
"name": db.name,
"collections": len(db.ListCollections()),
"created_at": db.GetCreatedAt(),
"updated_at": db.GetUpdatedAt(),
"deleted_at": db.GetDeletedAt(),
}
totalDocs := int64(0)
totalSize := int64(0)
db.collections.Range(func(k, v interface{}) bool {
coll := v.(*Collection)
totalDocs += coll.Count()
totalSize += coll.Size()
return true
})
dbStats["documents"] = totalDocs
dbStats["size_bytes"] = totalSize
databases = append(databases, dbStats)
return true
})
stats["databases"] = databases
return stats
}