feat: drop in-repo smartbotic-database service and storage.proto

Remove the entire src/database/ tree and proto/storage.proto. Going forward
all storage goes through libsmartbotic-db-client against the upstream
smartbotic-database daemon, configured via webserver.json / runner.json
'database_address'.

CMakeLists.txt

@@ -103,9 +103,7 @@ target_link_libraries(smartbotic_credentials PUBLIC
 add_library(smartbotic_proto STATIC
     ${CMAKE_CURRENT_BINARY_DIR}/proto/common.pb.cc
     ${CMAKE_CURRENT_BINARY_DIR}/proto/common.grpc.pb.cc
-    ${CMAKE_CURRENT_BINARY_DIR}/proto/storage.pb.cc
-    ${CMAKE_CURRENT_BINARY_DIR}/proto/storage.grpc.pb.cc
-    ${CMAKE_CURRENT_BINARY_DIR}/proto/workflow.pb.cc
+${CMAKE_CURRENT_BINARY_DIR}/proto/workflow.pb.cc
     ${CMAKE_CURRENT_BINARY_DIR}/proto/workflow.grpc.pb.cc
     ${CMAKE_CURRENT_BINARY_DIR}/proto/runner.pb.cc
     ${CMAKE_CURRENT_BINARY_DIR}/proto/runner.grpc.pb.cc
@@ -128,7 +126,7 @@ set(PROTO_DIR ${CMAKE_CURRENT_SOURCE_DIR}/proto)
 set(PROTO_OUT_DIR ${CMAKE_CURRENT_BINARY_DIR}/proto)
 file(MAKE_DIRECTORY ${PROTO_OUT_DIR})
 
-set(PROTO_FILES common storage workflow runner credentials)
+set(PROTO_FILES common workflow runner credentials)
 foreach(PROTO ${PROTO_FILES})
     add_custom_command(
         OUTPUT
@@ -147,26 +145,6 @@ foreach(PROTO ${PROTO_FILES})
     )
 endforeach()
 
-# Database service
-add_executable(smartbotic-database
-    src/database/main.cpp
-    src/database/memory_store.cpp
-    src/database/database_service.cpp
-    src/database/persistence/wal.cpp
-    src/database/persistence/snapshot.cpp
-)
-target_include_directories(smartbotic-database PRIVATE
-    ${CMAKE_CURRENT_SOURCE_DIR}/src
-    ${CMAKE_CURRENT_BINARY_DIR}
-)
-target_link_libraries(smartbotic-database PRIVATE
-    smartbotic_common
-    smartbotic_logging
-    smartbotic_config
-    smartbotic_proto
-    gRPC::grpc++
-)
-
 # WebServer service
 add_executable(smartbotic-webserver
     src/webserver/main.cpp
@@ -283,7 +261,6 @@ target_link_libraries(smartbotic-migrate-nodes PRIVATE
 
 # Install targets
 install(TARGETS
-    smartbotic-database
     smartbotic-webserver
     smartbotic-runner
     smartbotic-migrate-nodes

proto/storage.proto

@@ -1,204 +0,0 @@
-syntax = "proto3";
-
-package smartbotic.proto;
-
-import "common.proto";
-
-option cc_enable_arenas = true;
-
-// Document with metadata
-message Document {
-    string id = 1;
-    string collection = 2;
-    string data = 3;  // JSON string
-    int64 version = 4;
-    int64 created_at = 5;
-    int64 updated_at = 6;
-    int64 expires_at = 7;  // 0 = no expiry
-}
-
-// Query filter operation
-enum FilterOp {
-    FILTER_OP_UNSPECIFIED = 0;
-    FILTER_OP_EQ = 1;      // Equal
-    FILTER_OP_NE = 2;      // Not equal
-    FILTER_OP_GT = 3;      // Greater than
-    FILTER_OP_GTE = 4;     // Greater than or equal
-    FILTER_OP_LT = 5;      // Less than
-    FILTER_OP_LTE = 6;     // Less than or equal
-    FILTER_OP_IN = 7;      // In array
-    FILTER_OP_NIN = 8;     // Not in array
-    FILTER_OP_CONTAINS = 9; // Contains substring
-    FILTER_OP_REGEX = 10;  // Regular expression match
-    FILTER_OP_EXISTS = 11; // Field exists
-}
-
-// Single filter condition
-message Filter {
-    string field = 1;
-    FilterOp op = 2;
-    string value = 3;  // JSON encoded value
-}
-
-// Sort direction
-enum SortDirection {
-    SORT_DIRECTION_UNSPECIFIED = 0;
-    SORT_DIRECTION_ASC = 1;
-    SORT_DIRECTION_DESC = 2;
-}
-
-// Sort specification
-message Sort {
-    string field = 1;
-    SortDirection direction = 2;
-}
-
-// Query request
-message QueryRequest {
-    string collection = 1;
-    repeated Filter filters = 2;
-    repeated Sort sorts = 3;
-    PaginationRequest pagination = 4;
-    repeated string fields = 5;  // Projection - empty means all fields
-}
-
-// Query response
-message QueryResponse {
-    repeated Document documents = 1;
-    PaginationResponse pagination = 2;
-}
-
-// Get document request
-message GetRequest {
-    string collection = 1;
-    string id = 2;
-}
-
-// Insert request
-message InsertRequest {
-    string collection = 1;
-    string id = 2;  // Optional - generated if empty
-    string data = 3;  // JSON string
-    int64 ttl_ms = 4;  // Time to live in milliseconds (0 = no expiry)
-}
-
-// Update request
-message UpdateRequest {
-    string collection = 1;
-    string id = 2;
-    string data = 3;  // JSON string - full document or partial update
-    int64 expected_version = 4;  // Optimistic locking (0 = no check)
-    bool partial = 5;  // If true, merge with existing document
-}
-
-// Delete request
-message DeleteRequest {
-    string collection = 1;
-    string id = 2;
-    int64 expected_version = 3;  // Optimistic locking (0 = no check)
-}
-
-// Batch operations
-message BatchInsertRequest {
-    string collection = 1;
-    repeated InsertRequest documents = 2;
-}
-
-message BatchDeleteRequest {
-    string collection = 1;
-    repeated string ids = 2;
-}
-
-// Versioning configuration
-message VersioningConfig {
-    bool enabled = 1;
-    int32 max_versions = 2;      // 0 = unlimited
-    int64 version_ttl_ms = 3;    // 0 = no expiry
-    bool keep_on_delete = 4;     // Archive deleted documents
-}
-
-// Collection management
-message CreateCollectionRequest {
-    string name = 1;
-    string schema = 2;  // JSON Schema for validation (optional)
-    int64 default_ttl_ms = 3;  // Default TTL for documents
-    VersioningConfig versioning = 4;  // Versioning configuration
-}
-
-message DropCollectionRequest {
-    string name = 1;
-}
-
-message ListCollectionsRequest {}
-
-message ListCollectionsResponse {
-    repeated string collections = 1;
-}
-
-// Mutation response
-message MutationResponse {
-    string id = 1;
-    int64 version = 2;
-    bool success = 3;
-    Error error = 4;
-}
-
-message BatchMutationResponse {
-    repeated MutationResponse results = 1;
-    int32 success_count = 2;
-    int32 failure_count = 3;
-}
-
-// Document version for version history
-message DocumentVersion {
-    string id = 1;              // "{documentId}__v{version}"
-    string document_id = 2;
-    int64 version = 3;
-    string data = 4;            // JSON string
-    int64 created_at = 5;
-    int64 expires_at = 6;
-}
-
-// Get specific document version
-message GetVersionRequest {
-    string collection = 1;
-    string id = 2;
-    int64 version = 3;
-}
-
-// List versions for a document
-message ListVersionsRequest {
-    string collection = 1;
-    string id = 2;
-    int32 limit = 3;
-    int32 offset = 4;
-}
-
-message ListVersionsResponse {
-    repeated DocumentVersion versions = 1;
-    int64 total_count = 2;
-    bool has_more = 3;
-}
-
-// Storage service definition
-service StorageService {
-    // Document operations
-    rpc Get(GetRequest) returns (Document);
-    rpc Query(QueryRequest) returns (QueryResponse);
-    rpc Insert(InsertRequest) returns (MutationResponse);
-    rpc Update(UpdateRequest) returns (MutationResponse);
-    rpc Delete(DeleteRequest) returns (MutationResponse);
-
-    // Batch operations
-    rpc BatchInsert(BatchInsertRequest) returns (BatchMutationResponse);
-    rpc BatchDelete(BatchDeleteRequest) returns (BatchMutationResponse);
-
-    // Collection management
-    rpc CreateCollection(CreateCollectionRequest) returns (Empty);
-    rpc DropCollection(DropCollectionRequest) returns (Empty);
-    rpc ListCollections(ListCollectionsRequest) returns (ListCollectionsResponse);
-
-    // Version operations
-    rpc GetVersion(GetVersionRequest) returns (Document);
-    rpc ListVersions(ListVersionsRequest) returns (ListVersionsResponse);
-}

src/database/database_service.cpp

@@ -1,609 +0,0 @@
-#include "database_service.hpp"
-#include "logging/logger.hpp"
-#include "common/time_utils.hpp"
-#include <grpcpp/health_check_service_interface.h>
-
-namespace smartbotic::database {
-
-using namespace common;
-
-// DatabaseServiceImpl implementation
-DatabaseServiceImpl::DatabaseServiceImpl(MemoryStore& store)
-    : store_(store) {}
-
-grpc::Status DatabaseServiceImpl::Get(grpc::ServerContext* context,
-                                      const proto::GetRequest* request,
-                                      proto::Document* response) {
-    auto result = store_.get(request->collection(), request->id());
-    if (result.failed()) {
-        return grpc::Status(grpc::StatusCode::NOT_FOUND, result.error().message());
-    }
-
-    documentToProto(result.value(), response);
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::Query(grpc::ServerContext* context,
-                                        const proto::QueryRequest* request,
-                                        proto::QueryResponse* response) {
-    struct Query q(request->collection());
-
-    // Build filters
-    for (const auto& f : request->filters()) {
-        Filter filter;
-        filter.field = f.field();
-        filter.op = protoToFilterOp(f.op());
-        filter.value = nlohmann::json::parse(f.value());
-        q.filters.push_back(std::move(filter));
-    }
-
-    // Build sorts
-    for (const auto& s : request->sorts()) {
-        Sort sort;
-        sort.field = s.field();
-        sort.direction = s.direction() == proto::SORT_DIRECTION_DESC ?
-                        SortDirection::Desc : SortDirection::Asc;
-        q.sorts.push_back(std::move(sort));
-    }
-
-    // Pagination
-    if (request->has_pagination()) {
-        q.offset = (request->pagination().page() - 1) * request->pagination().page_size();
-        q.limit = request->pagination().page_size();
-    }
-
-    // Projection
-    for (const auto& field : request->fields()) {
-        q.fields.push_back(field);
-    }
-
-    auto result = store_.query(q);
-
-    for (const auto& doc : result.documents) {
-        documentToProto(doc, response->add_documents());
-    }
-
-    auto* pagination = response->mutable_pagination();
-    pagination->set_total_count(result.total_count);
-    pagination->set_has_more(result.has_more);
-
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::Insert(grpc::ServerContext* context,
-                                         const proto::InsertRequest* request,
-                                         proto::MutationResponse* response) {
-    Document doc;
-    doc.id = request->id();
-    doc.collection = request->collection();
-    doc.data = nlohmann::json::parse(request->data());
-
-    if (request->ttl_ms() > 0) {
-        doc.expires_at = TimeUtils::nowMs() + request->ttl_ms();
-    }
-
-    auto result = store_.insert(request->collection(), std::move(doc));
-    if (result.failed()) {
-        response->set_success(false);
-        auto* error = response->mutable_error();
-        error->set_code(static_cast<int32_t>(result.error().code()));
-        error->set_message(result.error().message());
-        return grpc::Status::OK;
-    }
-
-    response->set_id(result.value().id);
-    response->set_version(result.value().version);
-    response->set_success(true);
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::Update(grpc::ServerContext* context,
-                                         const proto::UpdateRequest* request,
-                                         proto::MutationResponse* response) {
-    auto data = nlohmann::json::parse(request->data());
-    auto result = store_.update(request->collection(), request->id(), data,
-                               request->expected_version(), request->partial());
-
-    if (result.failed()) {
-        response->set_success(false);
-        auto* error = response->mutable_error();
-        error->set_code(static_cast<int32_t>(result.error().code()));
-        error->set_message(result.error().message());
-        return grpc::Status::OK;
-    }
-
-    response->set_id(result.value().id);
-    response->set_version(result.value().version);
-    response->set_success(true);
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::Delete(grpc::ServerContext* context,
-                                         const proto::DeleteRequest* request,
-                                         proto::MutationResponse* response) {
-    auto result = store_.remove(request->collection(), request->id(),
-                               request->expected_version());
-
-    if (result.failed()) {
-        response->set_success(false);
-        auto* error = response->mutable_error();
-        error->set_code(static_cast<int32_t>(result.error().code()));
-        error->set_message(result.error().message());
-        return grpc::Status::OK;
-    }
-
-    response->set_id(request->id());
-    response->set_success(true);
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::BatchInsert(grpc::ServerContext* context,
-                                              const proto::BatchInsertRequest* request,
-                                              proto::BatchMutationResponse* response) {
-    int32_t success_count = 0;
-    int32_t failure_count = 0;
-
-    for (const auto& req : request->documents()) {
-        Document doc;
-        doc.id = req.id();
-        doc.collection = request->collection();
-        doc.data = nlohmann::json::parse(req.data());
-
-        if (req.ttl_ms() > 0) {
-            doc.expires_at = TimeUtils::nowMs() + req.ttl_ms();
-        }
-
-        auto result = store_.insert(request->collection(), std::move(doc));
-        auto* mutation_result = response->add_results();
-
-        if (result.ok()) {
-            mutation_result->set_id(result.value().id);
-            mutation_result->set_version(result.value().version);
-            mutation_result->set_success(true);
-            ++success_count;
-        } else {
-            mutation_result->set_success(false);
-            auto* error = mutation_result->mutable_error();
-            error->set_code(static_cast<int32_t>(result.error().code()));
-            error->set_message(result.error().message());
-            ++failure_count;
-        }
-    }
-
-    response->set_success_count(success_count);
-    response->set_failure_count(failure_count);
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::BatchDelete(grpc::ServerContext* context,
-                                              const proto::BatchDeleteRequest* request,
-                                              proto::BatchMutationResponse* response) {
-    int32_t success_count = 0;
-    int32_t failure_count = 0;
-
-    for (const auto& id : request->ids()) {
-        auto result = store_.remove(request->collection(), id, 0);
-        auto* mutation_result = response->add_results();
-        mutation_result->set_id(id);
-
-        if (result.ok()) {
-            mutation_result->set_success(true);
-            ++success_count;
-        } else {
-            mutation_result->set_success(false);
-            auto* error = mutation_result->mutable_error();
-            error->set_code(static_cast<int32_t>(result.error().code()));
-            error->set_message(result.error().message());
-            ++failure_count;
-        }
-    }
-
-    response->set_success_count(success_count);
-    response->set_failure_count(failure_count);
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::CreateCollection(grpc::ServerContext* context,
-                                                    const proto::CreateCollectionRequest* request,
-                                                    proto::Empty* response) {
-    CollectionConfig config(request->name());
-    config.default_ttl_ms = request->default_ttl_ms();
-
-    if (!request->schema().empty()) {
-        config.schema = nlohmann::json::parse(request->schema());
-    }
-
-    // Handle versioning configuration
-    if (request->has_versioning()) {
-        VersioningConfig versioning;
-        versioning.enabled = request->versioning().enabled();
-        versioning.max_versions = request->versioning().max_versions();
-        versioning.version_ttl_ms = request->versioning().version_ttl_ms();
-        versioning.keep_on_delete = request->versioning().keep_on_delete();
-        config.versioning = versioning;
-    }
-
-    auto result = store_.createCollection(config);
-    if (result.failed()) {
-        return grpc::Status(grpc::StatusCode::ALREADY_EXISTS, result.error().message());
-    }
-
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::DropCollection(grpc::ServerContext* context,
-                                                  const proto::DropCollectionRequest* request,
-                                                  proto::Empty* response) {
-    auto result = store_.dropCollection(request->name());
-    if (result.failed()) {
-        return grpc::Status(grpc::StatusCode::NOT_FOUND, result.error().message());
-    }
-
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::ListCollections(grpc::ServerContext* context,
-                                                   const proto::ListCollectionsRequest* request,
-                                                   proto::ListCollectionsResponse* response) {
-    auto collections = store_.listCollections();
-    for (const auto& name : collections) {
-        response->add_collections(name);
-    }
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::GetVersion(grpc::ServerContext* context,
-                                             const proto::GetVersionRequest* request,
-                                             proto::Document* response) {
-    auto result = store_.getVersion(request->collection(), request->id(), request->version());
-    if (result.failed()) {
-        return grpc::Status(grpc::StatusCode::NOT_FOUND, result.error().message());
-    }
-
-    const auto& ver = result.value();
-    response->set_id(ver.document_id);
-    response->set_collection(request->collection());
-    response->set_data(ver.data.dump());
-    response->set_version(ver.version);
-    response->set_created_at(ver.created_at);
-    response->set_updated_at(ver.created_at);
-    response->set_expires_at(ver.expires_at);
-
-    return grpc::Status::OK;
-}
-
-grpc::Status DatabaseServiceImpl::ListVersions(grpc::ServerContext* context,
-                                               const proto::ListVersionsRequest* request,
-                                               proto::ListVersionsResponse* response) {
-    int32_t limit = request->limit() > 0 ? request->limit() : 100;
-    int32_t offset = request->offset();
-
-    auto result = store_.listVersions(request->collection(), request->id(), limit, offset);
-
-    for (const auto& doc : result.documents) {
-        auto ver = DocumentVersion::fromJson(doc.data);
-        auto* proto_ver = response->add_versions();
-        documentVersionToProto(ver, proto_ver);
-    }
-
-    response->set_total_count(result.total_count);
-    response->set_has_more(result.has_more);
-
-    return grpc::Status::OK;
-}
-
-void DatabaseServiceImpl::documentToProto(const Document& doc, proto::Document* proto) {
-    proto->set_id(doc.id);
-    proto->set_collection(doc.collection);
-    proto->set_data(doc.data.dump());
-    proto->set_version(doc.version);
-    proto->set_created_at(doc.created_at);
-    proto->set_updated_at(doc.updated_at);
-    proto->set_expires_at(doc.expires_at);
-}
-
-void DatabaseServiceImpl::documentVersionToProto(const DocumentVersion& ver, proto::DocumentVersion* proto) {
-    proto->set_id(ver.id);
-    proto->set_document_id(ver.document_id);
-    proto->set_version(ver.version);
-    proto->set_data(ver.data.dump());
-    proto->set_created_at(ver.created_at);
-    proto->set_expires_at(ver.expires_at);
-}
-
-FilterOp DatabaseServiceImpl::protoToFilterOp(proto::FilterOp op) {
-    switch (op) {
-        case proto::FILTER_OP_EQ: return FilterOp::Eq;
-        case proto::FILTER_OP_NE: return FilterOp::Ne;
-        case proto::FILTER_OP_GT: return FilterOp::Gt;
-        case proto::FILTER_OP_GTE: return FilterOp::Gte;
-        case proto::FILTER_OP_LT: return FilterOp::Lt;
-        case proto::FILTER_OP_LTE: return FilterOp::Lte;
-        case proto::FILTER_OP_IN: return FilterOp::In;
-        case proto::FILTER_OP_NIN: return FilterOp::Nin;
-        case proto::FILTER_OP_CONTAINS: return FilterOp::Contains;
-        case proto::FILTER_OP_REGEX: return FilterOp::Regex;
-        case proto::FILTER_OP_EXISTS: return FilterOp::Exists;
-        default: return FilterOp::Eq;
-    }
-}
-
-// DatabaseService implementation
-DatabaseService::DatabaseService(const Config& config)
-    : config_(config) {
-
-    // Setup WAL
-    WAL::Config wal_config;
-    wal_config.directory = std::filesystem::path(config_.data_directory) / "wal";
-    wal_config.sync_interval_ms = config_.wal.sync_interval_ms;
-    wal_config.enabled = true;
-    wal_ = std::make_unique<WAL>(wal_config);
-
-    // Setup snapshot manager
-    SnapshotManager::Config snapshot_config;
-    snapshot_config.directory = std::filesystem::path(config_.data_directory) / "snapshots";
-    snapshot_config.interval_sec = config_.snapshot.interval_sec;
-    snapshot_config.enabled = true;
-    snapshot_manager_ = std::make_unique<SnapshotManager>(snapshot_config);
-
-    // Setup mutation callback for WAL
-    store_.setMutationCallback([this](const WalEntry& entry) {
-        wal_->append(entry);
-    });
-}
-
-DatabaseService::~DatabaseService() {
-    stop();
-}
-
-DatabaseService::Config DatabaseService::loadConfig(const std::filesystem::path& path) {
-    Config config;
-
-    auto result = config::Config::fromFile(path);
-    if (result.ok()) {
-        auto& cfg = result.value();
-        config.grpc_port = cfg.getOr<int>("grpc_port", 9001);
-        config.data_directory = cfg.getOr<std::string>("data_directory", "./data/database");
-        config.max_message_size_mb = cfg.getOr<int>("max_message_size_mb", 64);
-        config.wal.sync_interval_ms = cfg.getOr<int>("persistence.wal_sync_interval_ms", 100);
-        config.snapshot.interval_sec = cfg.getOr<int>("persistence.snapshot_interval_sec", 3600);
-    }
-
-    return config;
-}
-
-void DatabaseService::start() {
-    if (running_) {
-        return;
-    }
-
-    LOG_INFO("Starting database service...");
-
-    // Create data directory
-    std::filesystem::create_directories(config_.data_directory);
-
-    // Recovery from persistence
-    recoveryFromPersistence();
-
-    // Create predefined collections
-    createPredefinedCollections();
-
-    // Start WAL
-    wal_->start();
-
-    // Start gRPC server
-    service_impl_ = std::make_unique<DatabaseServiceImpl>(store_);
-
-    // Disable health check to avoid async completion queue issues
-    // grpc::EnableDefaultHealthCheckService(true);
-    grpc::ServerBuilder builder;
-    builder.AddListeningPort("0.0.0.0:" + std::to_string(config_.grpc_port),
-                            grpc::InsecureServerCredentials());
-    builder.RegisterService(service_impl_.get());
-
-    // Set max message size to handle large execution results
-    int max_msg_size = config_.max_message_size_mb * 1024 * 1024;
-    builder.SetMaxReceiveMessageSize(max_msg_size);
-    builder.SetMaxSendMessageSize(max_msg_size);
-    LOG_INFO("Max gRPC message size: {} MB", config_.max_message_size_mb);
-
-    server_ = builder.BuildAndStart();
-    LOG_INFO("Database gRPC server listening on port {}", config_.grpc_port);
-
-    running_ = true;
-
-    // Start background threads
-    snapshot_thread_ = std::thread(&DatabaseService::snapshotLoop, this);
-    ttl_thread_ = std::thread(&DatabaseService::ttlLoop, this);
-}
-
-void DatabaseService::stop() {
-    if (!running_) {
-        return;
-    }
-
-    LOG_INFO("Stopping database service...");
-
-    // Signal shutdown to background threads
-    {
-        std::lock_guard<std::mutex> lock(shutdown_mutex_);
-        running_ = false;
-    }
-    shutdown_cv_.notify_all();
-
-    // Stop background threads (they will wake up immediately now)
-    if (snapshot_thread_.joinable()) {
-        snapshot_thread_.join();
-    }
-    if (ttl_thread_.joinable()) {
-        ttl_thread_.join();
-    }
-
-    // Final snapshot before shutdown
-    LOG_INFO("Creating final snapshot before shutdown...");
-    createSnapshot();
-
-    // Stop WAL
-    wal_->stop();
-
-    // Stop gRPC server
-    if (server_) {
-        server_->Shutdown();
-    }
-
-    LOG_INFO("Database service stopped");
-}
-
-void DatabaseService::createPredefinedCollections() {
-    // Users collection
-    store_.createCollection(CollectionConfig("users"));
-
-    // Workflows collection
-    store_.createCollection(CollectionConfig("workflows"));
-
-    // Workflow groups collection
-    store_.createCollection(CollectionConfig("workflow_groups"));
-
-    // Executions collection with 7-day TTL
-    store_.createCollection(
-        CollectionConfig("executions").withTTL(7 * 24 * 60 * 60 * 1000));
-
-    // Sessions collection with 1-day TTL
-    store_.createCollection(
-        CollectionConfig("sessions").withTTL(24 * 60 * 60 * 1000));
-
-    // API keys collection
-    store_.createCollection(CollectionConfig("api_keys"));
-
-    // Credentials collection
-    store_.createCollection(CollectionConfig("credentials"));
-
-    // Runners collection
-    store_.createCollection(CollectionConfig("runners"));
-
-    // Nodes collection - stores node definitions
-    store_.createCollection(CollectionConfig("nodes"));
-
-    LOG_INFO("Created predefined collections");
-}
-
-void DatabaseService::createSnapshot() {
-    auto snapshot = store_.createSnapshot();
-    auto result = snapshot_manager_->save(snapshot);
-    if (result.ok()) {
-        // Truncate WAL after successful snapshot
-        wal_->truncate(snapshot.value("sequence", 0));
-    }
-}
-
-void DatabaseService::recoveryFromPersistence() {
-    LOG_INFO("Starting recovery from persistence...");
-
-    // Try to load latest snapshot
-    auto snapshot_result = snapshot_manager_->loadLatest();
-    if (snapshot_result.ok()) {
-        store_.loadSnapshot(snapshot_result.value());
-        LOG_INFO("Loaded snapshot");
-    }
-
-    // Replay WAL entries after snapshot
-    int64_t last_sequence = snapshot_manager_->getLatestSequence();
-    wal_->replay([this, last_sequence](const WalEntry& entry) {
-        if (entry.sequence <= last_sequence) {
-            return;  // Skip entries already in snapshot
-        }
-
-        // Apply WAL entry
-        switch (entry.type) {
-            case WalEntryType::Insert: {
-                auto doc = Document::fromJson(entry.data);
-                store_.getCollection(entry.collection)->loadFromSnapshot({doc});
-                break;
-            }
-            case WalEntryType::Update: {
-                auto doc = Document::fromJson(entry.data);
-                store_.getCollection(entry.collection)->loadFromSnapshot({doc});
-                break;
-            }
-            case WalEntryType::Delete: {
-                auto* col = store_.getCollection(entry.collection);
-                if (col) {
-                    col->remove(entry.document_id, 0);
-                }
-                break;
-            }
-            case WalEntryType::CreateCollection: {
-                CollectionConfig config(entry.collection);
-                if (entry.data.contains("default_ttl_ms")) {
-                    config.default_ttl_ms = entry.data["default_ttl_ms"];
-                }
-                if (entry.data.contains("versioning")) {
-                    config.versioning = VersioningConfig::fromJson(entry.data["versioning"]);
-                }
-                store_.createCollection(config);
-                break;
-            }
-            case WalEntryType::DropCollection: {
-                store_.dropCollection(entry.collection);
-                break;
-            }
-            case WalEntryType::StoreVersion: {
-                // Version entries are stored in-memory, no special recovery needed
-                // They are reconstructed from document updates during normal operation
-                break;
-            }
-            case WalEntryType::DeleteVersion: {
-                // Handled similarly to StoreVersion
-                break;
-            }
-        }
-    });
-
-    LOG_INFO("Recovery complete");
-}
-
-void DatabaseService::snapshotLoop() {
-    while (running_) {
-        // Wait for shutdown signal or timeout
-        {
-            std::unique_lock<std::mutex> lock(shutdown_mutex_);
-            if (shutdown_cv_.wait_for(lock,
-                    std::chrono::seconds(config_.snapshot.interval_sec),
-                    [this] { return !running_.load(); })) {
-                // Shutdown signaled, exit loop
-                break;
-            }
-        }
-
-        if (!running_) break;
-
-        // Check if WAL is large enough to trigger snapshot
-        if (wal_->getSize() >= config_.snapshot.wal_size_trigger) {
-            createSnapshot();
-        }
-    }
-}
-
-void DatabaseService::ttlLoop() {
-    constexpr int ttl_check_interval_sec = 60;  // Check every minute
-
-    while (running_) {
-        // Wait for shutdown signal or timeout
-        {
-            std::unique_lock<std::mutex> lock(shutdown_mutex_);
-            if (shutdown_cv_.wait_for(lock,
-                    std::chrono::seconds(ttl_check_interval_sec),
-                    [this] { return !running_.load(); })) {
-                // Shutdown signaled, exit loop
-                break;
-            }
-        }
-
-        if (!running_) break;
-
-        store_.expireAllDocuments();
-        store_.expireAllVersions();
-    }
-}
-
-} // namespace smartbotic::database

src/database/database_service.hpp

@@ -1,132 +0,0 @@
-#pragma once
-
-#include <memory>
-#include <thread>
-#include <atomic>
-#include <mutex>
-#include <condition_variable>
-#include <grpcpp/grpcpp.h>
-#include "proto/storage.grpc.pb.h"
-#include "memory_store.hpp"
-#include "persistence/wal.hpp"
-#include "persistence/snapshot.hpp"
-#include "config/config_loader.hpp"
-
-namespace smartbotic::database {
-
-// gRPC Storage Service implementation
-class DatabaseServiceImpl final : public proto::StorageService::Service {
-public:
-    explicit DatabaseServiceImpl(MemoryStore& store);
-
-    grpc::Status Get(grpc::ServerContext* context,
-                    const proto::GetRequest* request,
-                    proto::Document* response) override;
-
-    grpc::Status Query(grpc::ServerContext* context,
-                      const proto::QueryRequest* request,
-                      proto::QueryResponse* response) override;
-
-    grpc::Status Insert(grpc::ServerContext* context,
-                       const proto::InsertRequest* request,
-                       proto::MutationResponse* response) override;
-
-    grpc::Status Update(grpc::ServerContext* context,
-                       const proto::UpdateRequest* request,
-                       proto::MutationResponse* response) override;
-
-    grpc::Status Delete(grpc::ServerContext* context,
-                       const proto::DeleteRequest* request,
-                       proto::MutationResponse* response) override;
-
-    grpc::Status BatchInsert(grpc::ServerContext* context,
-                            const proto::BatchInsertRequest* request,
-                            proto::BatchMutationResponse* response) override;
-
-    grpc::Status BatchDelete(grpc::ServerContext* context,
-                            const proto::BatchDeleteRequest* request,
-                            proto::BatchMutationResponse* response) override;
-
-    grpc::Status CreateCollection(grpc::ServerContext* context,
-                                  const proto::CreateCollectionRequest* request,
-                                  proto::Empty* response) override;
-
-    grpc::Status DropCollection(grpc::ServerContext* context,
-                                const proto::DropCollectionRequest* request,
-                                proto::Empty* response) override;
-
-    grpc::Status ListCollections(grpc::ServerContext* context,
-                                 const proto::ListCollectionsRequest* request,
-                                 proto::ListCollectionsResponse* response) override;
-
-    // Version operations
-    grpc::Status GetVersion(grpc::ServerContext* context,
-                           const proto::GetVersionRequest* request,
-                           proto::Document* response) override;
-
-    grpc::Status ListVersions(grpc::ServerContext* context,
-                             const proto::ListVersionsRequest* request,
-                             proto::ListVersionsResponse* response) override;
-
-private:
-    void documentToProto(const Document& doc, proto::Document* proto);
-    void documentVersionToProto(const DocumentVersion& ver, proto::DocumentVersion* proto);
-    FilterOp protoToFilterOp(proto::FilterOp op);
-
-    MemoryStore& store_;
-};
-
-// Main database service
-class DatabaseService {
-public:
-    struct Config {
-        int grpc_port = 9001;
-        std::string data_directory = "./data/database";
-        int max_message_size_mb = 64;  // Max gRPC message size in MB
-        WAL::Config wal;
-        SnapshotManager::Config snapshot;
-    };
-
-    explicit DatabaseService(const Config& config);
-    ~DatabaseService();
-
-    // Load config from file
-    static Config loadConfig(const std::filesystem::path& path);
-
-    // Start service
-    void start();
-
-    // Stop service
-    void stop();
-
-    // Get memory store for direct access
-    MemoryStore& store() { return store_; }
-
-    // Create predefined collections
-    void createPredefinedCollections();
-
-    // Force snapshot
-    void createSnapshot();
-
-private:
-    void recoveryFromPersistence();
-    void snapshotLoop();
-    void ttlLoop();
-
-    Config config_;
-    MemoryStore store_;
-    std::unique_ptr<WAL> wal_;
-    std::unique_ptr<SnapshotManager> snapshot_manager_;
-    std::unique_ptr<DatabaseServiceImpl> service_impl_;
-    std::unique_ptr<grpc::Server> server_;
-
-    std::thread snapshot_thread_;
-    std::thread ttl_thread_;
-    std::atomic<bool> running_{false};
-
-    // Shutdown signaling
-    std::mutex shutdown_mutex_;
-    std::condition_variable shutdown_cv_;
-};
-
-} // namespace smartbotic::database

src/database/document.hpp

@@ -1,365 +0,0 @@
-#pragma once
-
-#include <string>
-#include <optional>
-#include <vector>
-#include <nlohmann/json.hpp>
-#include "common/uuid.hpp"
-#include "common/time_utils.hpp"
-
-namespace smartbotic::database {
-
-// Document structure
-struct Document {
-    std::string id;
-    std::string collection;
-    nlohmann::json data;
-    int64_t version = 1;
-    int64_t created_at = 0;
-    int64_t updated_at = 0;
-    int64_t expires_at = 0;  // 0 = no expiry
-
-    Document() = default;
-    Document(std::string id, std::string collection, nlohmann::json data)
-        : id(std::move(id))
-        , collection(std::move(collection))
-        , data(std::move(data)) {
-        auto now = common::TimeUtils::nowMs();
-        created_at = now;
-        updated_at = now;
-    }
-
-    bool isExpired() const {
-        return expires_at > 0 && common::TimeUtils::nowMs() > expires_at;
-    }
-
-    nlohmann::json toJson() const {
-        nlohmann::json j;
-        j["id"] = id;
-        j["collection"] = collection;
-        j["data"] = data;
-        j["version"] = version;
-        j["createdAt"] = created_at;
-        j["updatedAt"] = updated_at;
-        if (expires_at > 0) {
-            j["expiresAt"] = expires_at;
-        }
-        return j;
-    }
-
-    static Document fromJson(const nlohmann::json& j) {
-        Document doc;
-        doc.id = j.value("id", "");
-        doc.collection = j.value("collection", "");
-        doc.data = j.value("data", nlohmann::json::object());
-        doc.version = j.value("version", 1);
-        doc.created_at = j.value("createdAt", int64_t{0});
-        doc.updated_at = j.value("updatedAt", int64_t{0});
-        doc.expires_at = j.value("expiresAt", int64_t{0});
-        return doc;
-    }
-};
-
-// Filter operations
-enum class FilterOp {
-    Eq,       // Equal
-    Ne,       // Not equal
-    Gt,       // Greater than
-    Gte,      // Greater than or equal
-    Lt,       // Less than
-    Lte,      // Less than or equal
-    In,       // In array
-    Nin,      // Not in array
-    Contains, // Contains substring
-    Regex,    // Regular expression
-    Exists    // Field exists
-};
-
-// Single filter condition
-struct Filter {
-    std::string field;
-    FilterOp op = FilterOp::Eq;
-    nlohmann::json value;
-
-    Filter() = default;
-    Filter(std::string field, FilterOp op, nlohmann::json value)
-        : field(std::move(field)), op(op), value(std::move(value)) {}
-
-    // Convenience constructors
-    static Filter eq(const std::string& field, const nlohmann::json& value) {
-        return Filter(field, FilterOp::Eq, value);
-    }
-    static Filter ne(const std::string& field, const nlohmann::json& value) {
-        return Filter(field, FilterOp::Ne, value);
-    }
-    static Filter gt(const std::string& field, const nlohmann::json& value) {
-        return Filter(field, FilterOp::Gt, value);
-    }
-    static Filter gte(const std::string& field, const nlohmann::json& value) {
-        return Filter(field, FilterOp::Gte, value);
-    }
-    static Filter lt(const std::string& field, const nlohmann::json& value) {
-        return Filter(field, FilterOp::Lt, value);
-    }
-    static Filter lte(const std::string& field, const nlohmann::json& value) {
-        return Filter(field, FilterOp::Lte, value);
-    }
-    static Filter in(const std::string& field, const nlohmann::json& values) {
-        return Filter(field, FilterOp::In, values);
-    }
-    static Filter contains(const std::string& field, const std::string& substr) {
-        return Filter(field, FilterOp::Contains, substr);
-    }
-    static Filter regex(const std::string& field, const std::string& pattern) {
-        return Filter(field, FilterOp::Regex, pattern);
-    }
-    static Filter exists(const std::string& field, bool exists = true) {
-        return Filter(field, FilterOp::Exists, exists);
-    }
-};
-
-// Sort direction
-enum class SortDirection {
-    Asc,
-    Desc
-};
-
-// Sort specification
-struct Sort {
-    std::string field;
-    SortDirection direction = SortDirection::Asc;
-
-    Sort() = default;
-    Sort(std::string field, SortDirection direction = SortDirection::Asc)
-        : field(std::move(field)), direction(direction) {}
-
-    static Sort asc(const std::string& field) {
-        return Sort(field, SortDirection::Asc);
-    }
-    static Sort desc(const std::string& field) {
-        return Sort(field, SortDirection::Desc);
-    }
-};
-
-// Query request
-struct Query {
-    std::string collection;
-    std::vector<Filter> filters;
-    std::vector<Sort> sorts;
-    std::vector<std::string> fields;  // Projection
-    int32_t offset = 0;
-    int32_t limit = 100;
-
-    Query() = default;
-    explicit Query(std::string collection)
-        : collection(std::move(collection)) {}
-
-    Query& where(Filter filter) {
-        filters.push_back(std::move(filter));
-        return *this;
-    }
-
-    Query& orderBy(Sort sort) {
-        sorts.push_back(std::move(sort));
-        return *this;
-    }
-
-    Query& select(std::vector<std::string> fieldList) {
-        fields = std::move(fieldList);
-        return *this;
-    }
-
-    Query& skip(int32_t n) {
-        offset = n;
-        return *this;
-    }
-
-    Query& take(int32_t n) {
-        limit = n;
-        return *this;
-    }
-};
-
-// Query result
-struct QueryResult {
-    std::vector<Document> documents;
-    int64_t total_count = 0;
-    bool has_more = false;
-};
-
-// Versioning configuration for collections
-struct VersioningConfig {
-    bool enabled = false;
-    int32_t max_versions = 0;          // 0 = unlimited
-    int64_t version_ttl_ms = 0;        // 0 = no expiry
-    bool keep_on_delete = false;       // Archive deleted documents
-
-    VersioningConfig() = default;
-
-    VersioningConfig& withMaxVersions(int32_t max) {
-        max_versions = max;
-        return *this;
-    }
-
-    VersioningConfig& withTTL(int64_t ttl_ms) {
-        version_ttl_ms = ttl_ms;
-        return *this;
-    }
-
-    VersioningConfig& keepDeleted(bool keep = true) {
-        keep_on_delete = keep;
-        return *this;
-    }
-
-    nlohmann::json toJson() const {
-        return {
-            {"enabled", enabled},
-            {"maxVersions", max_versions},
-            {"versionTtlMs", version_ttl_ms},
-            {"keepOnDelete", keep_on_delete}
-        };
-    }
-
-    static VersioningConfig fromJson(const nlohmann::json& j) {
-        VersioningConfig config;
-        config.enabled = j.value("enabled", false);
-        config.max_versions = j.value("maxVersions", 0);
-        config.version_ttl_ms = j.value("versionTtlMs", 0);
-        config.keep_on_delete = j.value("keepOnDelete", false);
-        return config;
-    }
-
-    static VersioningConfig create(int32_t max_versions = 50) {
-        VersioningConfig config;
-        config.enabled = true;
-        config.max_versions = max_versions;
-        return config;
-    }
-};
-
-// Document version entry for version history
-struct DocumentVersion {
-    std::string id;              // "{documentId}__v{version}"
-    std::string document_id;
-    int64_t version = 0;
-    nlohmann::json data;
-    int64_t created_at = 0;
-    int64_t expires_at = 0;      // For TTL cleanup
-
-    DocumentVersion() = default;
-    DocumentVersion(const Document& doc)
-        : document_id(doc.id)
-        , version(doc.version)
-        , data(doc.data)
-        , created_at(common::TimeUtils::nowMs()) {
-        id = document_id + "__v" + std::to_string(version);
-    }
-
-    bool isExpired() const {
-        return expires_at > 0 && common::TimeUtils::nowMs() > expires_at;
-    }
-
-    nlohmann::json toJson() const {
-        return {
-            {"id", id},
-            {"documentId", document_id},
-            {"version", version},
-            {"data", data},
-            {"createdAt", created_at},
-            {"expiresAt", expires_at}
-        };
-    }
-
-    static DocumentVersion fromJson(const nlohmann::json& j) {
-        DocumentVersion ver;
-        ver.id = j.value("id", "");
-        ver.document_id = j.value("documentId", "");
-        ver.version = j.value("version", 0);
-        ver.data = j.value("data", nlohmann::json::object());
-        ver.created_at = j.value("createdAt", int64_t{0});
-        ver.expires_at = j.value("expiresAt", int64_t{0});
-        return ver;
-    }
-
-    Document toDocument() const {
-        Document doc;
-        doc.id = id;
-        doc.data = toJson();
-        doc.created_at = created_at;
-        doc.updated_at = created_at;
-        doc.expires_at = expires_at;
-        return doc;
-    }
-};
-
-// Collection configuration
-struct CollectionConfig {
-    std::string name;
-    std::optional<nlohmann::json> schema;  // JSON Schema for validation
-    int64_t default_ttl_ms = 0;  // Default TTL for documents
-    std::optional<VersioningConfig> versioning;  // Versioning configuration
-
-    CollectionConfig() = default;
-    explicit CollectionConfig(std::string name)
-        : name(std::move(name)) {}
-
-    CollectionConfig& withSchema(nlohmann::json s) {
-        schema = std::move(s);
-        return *this;
-    }
-
-    CollectionConfig& withTTL(int64_t ttl_ms) {
-        default_ttl_ms = ttl_ms;
-        return *this;
-    }
-
-    CollectionConfig& withVersioning(VersioningConfig config) {
-        versioning = std::move(config);
-        return *this;
-    }
-};
-
-// WAL entry types
-enum class WalEntryType {
-    Insert,
-    Update,
-    Delete,
-    CreateCollection,
-    DropCollection,
-    StoreVersion,      // Archive a document version
-    DeleteVersion      // Delete a specific version
-};
-
-// WAL entry for persistence
-struct WalEntry {
-    int64_t sequence = 0;
-    int64_t timestamp = 0;
-    WalEntryType type;
-    std::string collection;
-    std::string document_id;
-    nlohmann::json data;
-
-    nlohmann::json toJson() const {
-        nlohmann::json j;
-        j["seq"] = sequence;
-        j["ts"] = timestamp;
-        j["type"] = static_cast<int>(type);
-        j["col"] = collection;
-        j["id"] = document_id;
-        j["data"] = data;
-        return j;
-    }
-
-    static WalEntry fromJson(const nlohmann::json& j) {
-        WalEntry entry;
-        entry.sequence = j.value("seq", 0);
-        entry.timestamp = j.value("ts", 0);
-        entry.type = static_cast<WalEntryType>(j.value("type", 0));
-        entry.collection = j.value("col", "");
-        entry.document_id = j.value("id", "");
-        entry.data = j.value("data", nlohmann::json::object());
-        return entry;
-    }
-};
-
-} // namespace smartbotic::database

src/database/main.cpp

@@ -1,64 +0,0 @@
-#include <csignal>
-#include <iostream>
-#include "database_service.hpp"
-#include "logging/logger.hpp"
-#include "config/config_loader.hpp"
-
-using namespace smartbotic;
-
-std::atomic<bool> g_shutdown{false};
-
-void signalHandler(int signal) {
-    LOG_INFO("Received signal {}, shutting down...", signal);
-    g_shutdown = true;
-}
-
-int main(int argc, char* argv[]) {
-    // Initialize logging
-    logging::LogConfig log_config;
-    log_config.name = "database";
-    log_config.level = logging::LogLevel::Info;
-    logging::Logger::init(log_config);
-
-    LOG_INFO("SmartBotic Database Service starting...");
-
-    // Load configuration
-    std::filesystem::path config_path = "./config/database.json";
-    if (argc > 1) {
-        config_path = argv[1];
-    }
-
-    database::DatabaseService::Config config;
-    if (std::filesystem::exists(config_path)) {
-        config = database::DatabaseService::loadConfig(config_path);
-        LOG_INFO("Loaded configuration from {}", config_path.string());
-    } else {
-        LOG_WARN("Config file not found at {}, using defaults", config_path.string());
-    }
-
-    // Setup signal handlers
-    std::signal(SIGINT, signalHandler);
-    std::signal(SIGTERM, signalHandler);
-
-    // Create and start service
-    database::DatabaseService service(config);
-
-    try {
-        service.start();
-
-        // Wait for shutdown signal
-        while (!g_shutdown) {
-            std::this_thread::sleep_for(std::chrono::milliseconds(100));
-        }
-
-        service.stop();
-    } catch (const std::exception& e) {
-        LOG_ERROR("Fatal error: {}", e.what());
-        return 1;
-    }
-
-    LOG_INFO("SmartBotic Database Service stopped");
-    logging::Logger::shutdown();
-
-    return 0;
-}

src/database/memory_store.cpp

@@ -1,830 +0,0 @@
-#include "memory_store.hpp"
-#include "common/uuid.hpp"
-#include "common/time_utils.hpp"
-#include "logging/logger.hpp"
-#include <algorithm>
-
-namespace smartbotic::database {
-
-using namespace common;
-
-// Collection implementation
-Collection::Collection(const CollectionConfig& config)
-    : config_(config) {}
-
-Result<Document> Collection::get(const std::string& id) const {
-    std::shared_lock lock(mutex_);
-
-    auto it = documents_.find(id);
-    if (it == documents_.end()) {
-        return Error(ErrorCode::DocumentNotFound,
-                    "Document not found: " + id);
-    }
-
-    if (it->second.isExpired()) {
-        return Error(ErrorCode::DocumentNotFound,
-                    "Document expired: " + id);
-    }
-
-    return it->second;
-}
-
-Result<Document> Collection::insert(Document doc) {
-    if (doc.id.empty()) {
-        doc.id = UUID::generate();
-    }
-
-    if (!config_.schema || validateSchema(doc.data)) {
-        std::unique_lock lock(mutex_);
-
-        if (documents_.contains(doc.id)) {
-            return Error(ErrorCode::AlreadyExists,
-                        "Document already exists: " + doc.id);
-        }
-
-        auto now = TimeUtils::nowMs();
-        doc.created_at = now;
-        doc.updated_at = now;
-        doc.version = 1;
-        doc.collection = config_.name;
-
-        if (config_.default_ttl_ms > 0 && doc.expires_at == 0) {
-            doc.expires_at = now + config_.default_ttl_ms;
-        }
-
-        documents_[doc.id] = doc;
-        return doc;
-    }
-
-    return Error(ErrorCode::ValidationError, "Document failed schema validation");
-}
-
-Result<Document> Collection::update(const std::string& id, const nlohmann::json& data,
-                                    int64_t expected_version, bool partial) {
-    std::unique_lock lock(mutex_);
-
-    auto it = documents_.find(id);
-    if (it == documents_.end()) {
-        return Error(ErrorCode::DocumentNotFound,
-                    "Document not found: " + id);
-    }
-
-    if (it->second.isExpired()) {
-        documents_.erase(it);
-        return Error(ErrorCode::DocumentNotFound,
-                    "Document expired: " + id);
-    }
-
-    if (expected_version > 0 && it->second.version != expected_version) {
-        return Error(ErrorCode::VersionConflict,
-                    "Version conflict: expected " + std::to_string(expected_version) +
-                    ", got " + std::to_string(it->second.version));
-    }
-
-    // Store version before updating if versioning is enabled
-    if (config_.versioning && config_.versioning->enabled) {
-        storeVersionInternal(it->second);
-    }
-
-    nlohmann::json new_data;
-    if (partial) {
-        new_data = it->second.data;
-        new_data.merge_patch(data);
-    } else {
-        new_data = data;
-    }
-
-    if (config_.schema && !validateSchema(new_data)) {
-        return Error(ErrorCode::ValidationError, "Document failed schema validation");
-    }
-
-    it->second.data = std::move(new_data);
-    it->second.version++;
-    it->second.updated_at = TimeUtils::nowMs();
-
-    return it->second;
-}
-
-void Collection::storeVersionInternal(const Document& doc) {
-    // Note: caller must hold mutex_
-    DocumentVersion ver(doc);
-
-    // Set expiration if configured
-    if (config_.versioning && config_.versioning->version_ttl_ms > 0) {
-        ver.expires_at = TimeUtils::nowMs() + config_.versioning->version_ttl_ms;
-    }
-
-    versions_[doc.id].push_back(ver);
-    cleanupOldVersions(doc.id);
-}
-
-void Collection::storeVersion(const Document& doc) {
-    std::unique_lock lock(mutex_);
-    storeVersionInternal(doc);
-}
-
-void Collection::storeDeletedVersion(const Document& doc) {
-    if (!config_.versioning || !config_.versioning->enabled || !config_.versioning->keep_on_delete) {
-        return;
-    }
-    storeVersion(doc);
-}
-
-void Collection::cleanupOldVersions(const std::string& id) {
-    // Note: caller must hold mutex_
-    if (!config_.versioning || config_.versioning->max_versions <= 0) {
-        return;
-    }
-
-    auto& vers = versions_[id];
-    int32_t max_versions = config_.versioning->max_versions;
-
-    while (static_cast<int32_t>(vers.size()) > max_versions) {
-        vers.erase(vers.begin());  // Remove oldest
-    }
-}
-
-Result<DocumentVersion> Collection::getVersion(const std::string& id, int64_t version) {
-    std::shared_lock lock(mutex_);
-
-    auto it = versions_.find(id);
-    if (it == versions_.end()) {
-        return Error(ErrorCode::DocumentNotFound,
-                    "No versions found for document: " + id);
-    }
-
-    for (const auto& ver : it->second) {
-        if (ver.version == version) {
-            if (ver.isExpired()) {
-                return Error(ErrorCode::DocumentNotFound,
-                            "Version expired: " + id + " v" + std::to_string(version));
-            }
-            return ver;
-        }
-    }
-
-    return Error(ErrorCode::DocumentNotFound,
-                "Version not found: " + id + " v" + std::to_string(version));
-}
-
-QueryResult Collection::listVersions(const std::string& id, int32_t limit, int32_t offset) {
-    std::shared_lock lock(mutex_);
-
-    QueryResult result;
-
-    auto it = versions_.find(id);
-    if (it == versions_.end()) {
-        return result;
-    }
-
-    // Filter out expired versions and collect non-expired ones
-    std::vector<DocumentVersion> valid_versions;
-    for (const auto& ver : it->second) {
-        if (!ver.isExpired()) {
-            valid_versions.push_back(ver);
-        }
-    }
-
-    // Sort by version descending (newest first)
-    std::sort(valid_versions.begin(), valid_versions.end(),
-              [](const DocumentVersion& a, const DocumentVersion& b) {
-                  return a.version > b.version;
-              });
-
-    result.total_count = static_cast<int64_t>(valid_versions.size());
-
-    // Apply pagination
-    int32_t start = std::min(offset, static_cast<int32_t>(valid_versions.size()));
-    int32_t end = std::min(offset + limit, static_cast<int32_t>(valid_versions.size()));
-
-    result.has_more = end < static_cast<int32_t>(valid_versions.size());
-
-    for (int32_t i = start; i < end; ++i) {
-        result.documents.push_back(valid_versions[i].toDocument());
-    }
-
-    return result;
-}
-
-int64_t Collection::expireVersions() {
-    std::unique_lock lock(mutex_);
-
-    auto now = TimeUtils::nowMs();
-    int64_t expired = 0;
-
-    for (auto& [doc_id, vers] : versions_) {
-        for (auto it = vers.begin(); it != vers.end();) {
-            if (it->expires_at > 0 && it->expires_at < now) {
-                it = vers.erase(it);
-                ++expired;
-            } else {
-                ++it;
-            }
-        }
-    }
-
-    return expired;
-}
-
-Result<void> Collection::remove(const std::string& id, int64_t expected_version) {
-    std::unique_lock lock(mutex_);
-
-    auto it = documents_.find(id);
-    if (it == documents_.end()) {
-        return Error(ErrorCode::DocumentNotFound,
-                    "Document not found: " + id);
-    }
-
-    if (expected_version > 0 && it->second.version != expected_version) {
-        return Error(ErrorCode::VersionConflict,
-                    "Version conflict: expected " + std::to_string(expected_version) +
-                    ", got " + std::to_string(it->second.version));
-    }
-
-    // Store final version if keep_on_delete is enabled
-    if (config_.versioning && config_.versioning->enabled && config_.versioning->keep_on_delete) {
-        storeVersionInternal(it->second);
-    }
-
-    documents_.erase(it);
-    return Result<void>();
-}
-
-QueryResult Collection::query(const Query& q) const {
-    std::shared_lock lock(mutex_);
-
-    std::vector<Document> results;
-    results.reserve(documents_.size());
-
-    for (const auto& [id, doc] : documents_) {
-        if (!doc.isExpired() && matchesAllFilters(doc, q.filters)) {
-            results.push_back(doc);
-        }
-    }
-
-    QueryResult result;
-    result.total_count = static_cast<int64_t>(results.size());
-
-    // Sort
-    if (!q.sorts.empty()) {
-        sortDocuments(results, q.sorts);
-    }
-
-    // Pagination
-    int32_t start = std::min(q.offset, static_cast<int32_t>(results.size()));
-    int32_t end = std::min(q.offset + q.limit, static_cast<int32_t>(results.size()));
-
-    result.has_more = end < static_cast<int32_t>(results.size());
-
-    // Apply offset and limit
-    for (int32_t i = start; i < end; ++i) {
-        if (!q.fields.empty()) {
-            result.documents.push_back(applyProjection(results[i], q.fields));
-        } else {
-            result.documents.push_back(results[i]);
-        }
-    }
-
-    return result;
-}
-
-int64_t Collection::count(const std::vector<Filter>& filters) const {
-    std::shared_lock lock(mutex_);
-
-    if (filters.empty()) {
-        return static_cast<int64_t>(documents_.size());
-    }
-
-    int64_t cnt = 0;
-    for (const auto& [id, doc] : documents_) {
-        if (!doc.isExpired() && matchesAllFilters(doc, filters)) {
-            ++cnt;
-        }
-    }
-    return cnt;
-}
-
-bool Collection::exists(const std::string& id) const {
-    std::shared_lock lock(mutex_);
-    auto it = documents_.find(id);
-    return it != documents_.end() && !it->second.isExpired();
-}
-
-std::vector<Result<Document>> Collection::insertMany(std::vector<Document> docs) {
-    std::vector<Result<Document>> results;
-    results.reserve(docs.size());
-
-    for (auto& doc : docs) {
-        results.push_back(insert(std::move(doc)));
-    }
-
-    return results;
-}
-
-int64_t Collection::removeMany(const std::vector<std::string>& ids) {
-    std::unique_lock lock(mutex_);
-
-    int64_t count = 0;
-    for (const auto& id : ids) {
-        if (documents_.erase(id) > 0) {
-            ++count;
-        }
-    }
-    return count;
-}
-
-int64_t Collection::size() const {
-    std::shared_lock lock(mutex_);
-    return static_cast<int64_t>(documents_.size());
-}
-
-int64_t Collection::expireDocuments() {
-    std::unique_lock lock(mutex_);
-
-    auto now = TimeUtils::nowMs();
-    int64_t expired = 0;
-
-    for (auto it = documents_.begin(); it != documents_.end();) {
-        if (it->second.expires_at > 0 && it->second.expires_at < now) {
-            it = documents_.erase(it);
-            ++expired;
-        } else {
-            ++it;
-        }
-    }
-
-    return expired;
-}
-
-std::vector<Document> Collection::getAllDocuments() const {
-    std::shared_lock lock(mutex_);
-
-    std::vector<Document> docs;
-    docs.reserve(documents_.size());
-
-    for (const auto& [id, doc] : documents_) {
-        docs.push_back(doc);
-    }
-
-    return docs;
-}
-
-void Collection::loadFromSnapshot(const std::vector<Document>& docs) {
-    std::unique_lock lock(mutex_);
-
-    documents_.clear();
-    for (const auto& doc : docs) {
-        documents_[doc.id] = doc;
-    }
-}
-
-bool Collection::matchesFilter(const Document& doc, const Filter& filter) const {
-    auto value = getFieldValue(doc.data, filter.field);
-
-    switch (filter.op) {
-        case FilterOp::Eq:
-            return value == filter.value;
-
-        case FilterOp::Ne:
-            return value != filter.value;
-
-        case FilterOp::Gt:
-            return value > filter.value;
-
-        case FilterOp::Gte:
-            return value >= filter.value;
-
-        case FilterOp::Lt:
-            return value < filter.value;
-
-        case FilterOp::Lte:
-            return value <= filter.value;
-
-        case FilterOp::In:
-            if (filter.value.is_array()) {
-                for (const auto& v : filter.value) {
-                    if (value == v) return true;
-                }
-            }
-            return false;
-
-        case FilterOp::Nin:
-            if (filter.value.is_array()) {
-                for (const auto& v : filter.value) {
-                    if (value == v) return false;
-                }
-            }
-            return true;
-
-        case FilterOp::Contains:
-            if (value.is_string() && filter.value.is_string()) {
-                return value.get<std::string>().find(filter.value.get<std::string>()) != std::string::npos;
-            }
-            return false;
-
-        case FilterOp::Regex:
-            if (value.is_string() && filter.value.is_string()) {
-                try {
-                    std::regex re(filter.value.get<std::string>());
-                    return std::regex_search(value.get<std::string>(), re);
-                } catch (...) {
-                    return false;
-                }
-            }
-            return false;
-
-        case FilterOp::Exists:
-            return !value.is_null() == filter.value.get<bool>();
-    }
-
-    return false;
-}
-
-bool Collection::matchesAllFilters(const Document& doc, const std::vector<Filter>& filters) const {
-    for (const auto& filter : filters) {
-        if (!matchesFilter(doc, filter)) {
-            return false;
-        }
-    }
-    return true;
-}
-
-nlohmann::json Collection::getFieldValue(const nlohmann::json& data, const std::string& field) const {
-    // Support nested fields with dot notation
-    const nlohmann::json* current = &data;
-
-    size_t start = 0;
-    size_t pos;
-    while ((pos = field.find('.', start)) != std::string::npos) {
-        std::string part = field.substr(start, pos - start);
-        if (!current->is_object() || !current->contains(part)) {
-            return nlohmann::json(nullptr);
-        }
-        current = &(*current)[part];
-        start = pos + 1;
-    }
-
-    std::string last_part = field.substr(start);
-    if (!current->is_object() || !current->contains(last_part)) {
-        return nlohmann::json(nullptr);
-    }
-
-    return (*current)[last_part];
-}
-
-bool Collection::validateSchema(const nlohmann::json& data) const {
-    // Basic validation - in production, use a JSON Schema validator
-    if (!config_.schema) return true;
-
-    const auto& schema = *config_.schema;
-
-    // Check required fields
-    if (schema.contains("required") && schema["required"].is_array()) {
-        for (const auto& field : schema["required"]) {
-            if (!data.contains(field.get<std::string>())) {
-                return false;
-            }
-        }
-    }
-
-    return true;
-}
-
-void Collection::sortDocuments(std::vector<Document>& docs, const std::vector<Sort>& sorts) const {
-    std::sort(docs.begin(), docs.end(), [this, &sorts](const Document& a, const Document& b) {
-        for (const auto& sort : sorts) {
-            auto va = getFieldValue(a.data, sort.field);
-            auto vb = getFieldValue(b.data, sort.field);
-
-            if (va == vb) continue;
-
-            bool less = va < vb;
-            if (sort.direction == SortDirection::Desc) {
-                less = !less;
-            }
-            return less;
-        }
-        return false;
-    });
-}
-
-Document Collection::applyProjection(const Document& doc, const std::vector<std::string>& fields) const {
-    Document result = doc;
-    nlohmann::json projected = nlohmann::json::object();
-
-    for (const auto& field : fields) {
-        auto value = getFieldValue(doc.data, field);
-        if (!value.is_null()) {
-            projected[field] = value;
-        }
-    }
-
-    result.data = projected;
-    return result;
-}
-
-// MemoryStore implementation
-MemoryStore::MemoryStore() = default;
-MemoryStore::~MemoryStore() = default;
-
-void MemoryStore::setMutationCallback(MutationCallback callback) {
-    mutation_callback_ = std::move(callback);
-}
-
-Result<void> MemoryStore::createCollection(const CollectionConfig& config) {
-    std::unique_lock lock(mutex_);
-
-    if (collections_.contains(config.name)) {
-        return Error(ErrorCode::AlreadyExists,
-                    "Collection already exists: " + config.name);
-    }
-
-    collections_[config.name] = std::make_unique<Collection>(config);
-
-    WalEntry entry;
-    entry.type = WalEntryType::CreateCollection;
-    entry.collection = config.name;
-    entry.data = {
-        {"name", config.name},
-        {"default_ttl_ms", config.default_ttl_ms}
-    };
-    if (config.schema) {
-        entry.data["schema"] = *config.schema;
-    }
-    if (config.versioning) {
-        entry.data["versioning"] = config.versioning->toJson();
-    }
-    notifyMutation(std::move(entry));
-
-    return Result<void>();
-}
-
-Result<void> MemoryStore::dropCollection(const std::string& name) {
-    std::unique_lock lock(mutex_);
-
-    if (!collections_.contains(name)) {
-        return Error(ErrorCode::CollectionNotFound,
-                    "Collection not found: " + name);
-    }
-
-    collections_.erase(name);
-
-    WalEntry entry;
-    entry.type = WalEntryType::DropCollection;
-    entry.collection = name;
-    notifyMutation(std::move(entry));
-
-    return Result<void>();
-}
-
-bool MemoryStore::hasCollection(const std::string& name) const {
-    std::shared_lock lock(mutex_);
-    return collections_.contains(name);
-}
-
-std::vector<std::string> MemoryStore::listCollections() const {
-    std::shared_lock lock(mutex_);
-
-    std::vector<std::string> names;
-    names.reserve(collections_.size());
-
-    for (const auto& [name, _] : collections_) {
-        names.push_back(name);
-    }
-
-    return names;
-}
-
-Collection* MemoryStore::getCollection(const std::string& name) {
-    std::shared_lock lock(mutex_);
-    auto it = collections_.find(name);
-    return it != collections_.end() ? it->second.get() : nullptr;
-}
-
-const Collection* MemoryStore::getCollection(const std::string& name) const {
-    std::shared_lock lock(mutex_);
-    auto it = collections_.find(name);
-    return it != collections_.end() ? it->second.get() : nullptr;
-}
-
-Result<Document> MemoryStore::get(const std::string& collection, const std::string& id) {
-    auto* col = getCollection(collection);
-    if (!col) {
-        return Error(ErrorCode::CollectionNotFound,
-                    "Collection not found: " + collection);
-    }
-    return col->get(id);
-}
-
-Result<Document> MemoryStore::insert(const std::string& collection, Document doc) {
-    auto* col = getCollection(collection);
-    if (!col) {
-        return Error(ErrorCode::CollectionNotFound,
-                    "Collection not found: " + collection);
-    }
-
-    auto result = col->insert(std::move(doc));
-    if (result.ok()) {
-        WalEntry entry;
-        entry.type = WalEntryType::Insert;
-        entry.collection = collection;
-        entry.document_id = result.value().id;
-        entry.data = result.value().toJson();
-        notifyMutation(std::move(entry));
-    }
-    return result;
-}
-
-Result<Document> MemoryStore::update(const std::string& collection, const std::string& id,
-                                     const nlohmann::json& data, int64_t expected_version,
-                                     bool partial) {
-    auto* col = getCollection(collection);
-    if (!col) {
-        return Error(ErrorCode::CollectionNotFound,
-                    "Collection not found: " + collection);
-    }
-
-    auto result = col->update(id, data, expected_version, partial);
-    if (result.ok()) {
-        WalEntry entry;
-        entry.type = WalEntryType::Update;
-        entry.collection = collection;
-        entry.document_id = id;
-        entry.data = result.value().toJson();
-        notifyMutation(std::move(entry));
-    }
-    return result;
-}
-
-Result<void> MemoryStore::remove(const std::string& collection, const std::string& id,
-                                  int64_t expected_version) {
-    auto* col = getCollection(collection);
-    if (!col) {
-        return Error(ErrorCode::CollectionNotFound,
-                    "Collection not found: " + collection);
-    }
-
-    auto result = col->remove(id, expected_version);
-    if (result.ok()) {
-        WalEntry entry;
-        entry.type = WalEntryType::Delete;
-        entry.collection = collection;
-        entry.document_id = id;
-        notifyMutation(std::move(entry));
-    }
-    return result;
-}
-
-QueryResult MemoryStore::query(const Query& q) {
-    auto* col = getCollection(q.collection);
-    if (!col) {
-        return QueryResult{};
-    }
-    return col->query(q);
-}
-
-void MemoryStore::expireAllDocuments() {
-    std::shared_lock lock(mutex_);
-
-    for (auto& [name, col] : collections_) {
-        auto expired = col->expireDocuments();
-        if (expired > 0) {
-            LOG_DEBUG("Expired {} documents from collection {}", expired, name);
-        }
-    }
-}
-
-void MemoryStore::expireAllVersions() {
-    std::shared_lock lock(mutex_);
-
-    for (auto& [name, col] : collections_) {
-        auto expired = col->expireVersions();
-        if (expired > 0) {
-            LOG_DEBUG("Expired {} versions from collection {}", expired, name);
-        }
-    }
-}
-
-Result<DocumentVersion> MemoryStore::getVersion(const std::string& collection,
-                                                const std::string& id,
-                                                int64_t version) {
-    auto* col = getCollection(collection);
-    if (!col) {
-        return Error(ErrorCode::CollectionNotFound,
-                    "Collection not found: " + collection);
-    }
-    return col->getVersion(id, version);
-}
-
-QueryResult MemoryStore::listVersions(const std::string& collection,
-                                      const std::string& id,
-                                      int32_t limit,
-                                      int32_t offset) {
-    auto* col = getCollection(collection);
-    if (!col) {
-        return QueryResult{};
-    }
-    return col->listVersions(id, limit, offset);
-}
-
-nlohmann::json MemoryStore::createSnapshot() const {
-    std::shared_lock lock(mutex_);
-
-    nlohmann::json snapshot;
-    snapshot["version"] = 1;
-    snapshot["timestamp"] = TimeUtils::nowMs();
-    snapshot["sequence"] = sequence_.load();
-    snapshot["collections"] = nlohmann::json::object();
-
-    for (const auto& [name, col] : collections_) {
-        nlohmann::json col_data;
-        col_data["config"] = {
-            {"name", col->config().name},
-            {"default_ttl_ms", col->config().default_ttl_ms}
-        };
-        if (col->config().schema) {
-            col_data["config"]["schema"] = *col->config().schema;
-        }
-        if (col->config().versioning) {
-            col_data["config"]["versioning"] = col->config().versioning->toJson();
-        }
-
-        col_data["documents"] = nlohmann::json::array();
-        for (const auto& doc : col->getAllDocuments()) {
-            col_data["documents"].push_back(doc.toJson());
-        }
-
-        snapshot["collections"][name] = col_data;
-    }
-
-    return snapshot;
-}
-
-void MemoryStore::loadSnapshot(const nlohmann::json& snapshot) {
-    std::unique_lock lock(mutex_);
-
-    collections_.clear();
-
-    if (snapshot.contains("sequence")) {
-        sequence_ = snapshot["sequence"].get<int64_t>();
-    }
-
-    if (snapshot.contains("collections")) {
-        for (auto it = snapshot["collections"].begin(); it != snapshot["collections"].end(); ++it) {
-            const auto& col_data = it.value();
-
-            CollectionConfig config;
-            config.name = it.key();
-            if (col_data.contains("config")) {
-                config.default_ttl_ms = col_data["config"].value("default_ttl_ms", 0);
-                if (col_data["config"].contains("schema")) {
-                    config.schema = col_data["config"]["schema"];
-                }
-                if (col_data["config"].contains("versioning")) {
-                    config.versioning = VersioningConfig::fromJson(col_data["config"]["versioning"]);
-                }
-            }
-
-            collections_[config.name] = std::make_unique<Collection>(config);
-
-            if (col_data.contains("documents")) {
-                std::vector<Document> docs;
-                for (const auto& doc_json : col_data["documents"]) {
-                    docs.push_back(Document::fromJson(doc_json));
-                }
-                collections_[config.name]->loadFromSnapshot(docs);
-            }
-        }
-    }
-}
-
-MemoryStore::Stats MemoryStore::getStats() const {
-    std::shared_lock lock(mutex_);
-
-    Stats stats;
-    stats.collection_count = collections_.size();
-
-    for (const auto& [name, col] : collections_) {
-        stats.total_documents += col->size();
-    }
-
-    // Rough memory estimate
-    stats.memory_estimate_bytes = stats.total_documents * 1024;  // ~1KB per document estimate
-
-    return stats;
-}
-
-void MemoryStore::notifyMutation(WalEntry entry) {
-    entry.sequence = nextSequence();
-    entry.timestamp = TimeUtils::nowMs();
-
-    if (mutation_callback_) {
-        mutation_callback_(entry);
-    }
-}
-
-int64_t MemoryStore::nextSequence() {
-    return ++sequence_;
-}
-
-} // namespace smartbotic::database

src/database/memory_store.hpp

@@ -1,136 +0,0 @@
-#pragma once
-
-#include <string>
-#include <unordered_map>
-#include <unordered_set>
-#include <shared_mutex>
-#include <functional>
-#include <regex>
-#include <atomic>
-#include <nlohmann/json.hpp>
-#include "document.hpp"
-#include "common/error.hpp"
-
-namespace smartbotic::database {
-
-// Collection - stores documents with thread-safe access
-class Collection {
-public:
-    explicit Collection(const CollectionConfig& config);
-
-    // Document operations
-    common::Result<Document> get(const std::string& id) const;
-    common::Result<Document> insert(Document doc);
-    common::Result<Document> update(const std::string& id, const nlohmann::json& data,
-                                    int64_t expected_version = 0, bool partial = false);
-    common::Result<void> remove(const std::string& id, int64_t expected_version = 0);
-
-    // Query operations
-    QueryResult query(const Query& q) const;
-    int64_t count(const std::vector<Filter>& filters = {}) const;
-    bool exists(const std::string& id) const;
-
-    // Bulk operations
-    std::vector<common::Result<Document>> insertMany(std::vector<Document> docs);
-    int64_t removeMany(const std::vector<std::string>& ids);
-
-    // Collection info
-    const std::string& name() const { return config_.name; }
-    const CollectionConfig& config() const { return config_; }
-    int64_t size() const;
-
-    // TTL expiration
-    int64_t expireDocuments();
-
-    // Snapshot support
-    std::vector<Document> getAllDocuments() const;
-    void loadFromSnapshot(const std::vector<Document>& docs);
-
-    // Version operations
-    common::Result<DocumentVersion> getVersion(const std::string& id, int64_t version);
-    QueryResult listVersions(const std::string& id, int32_t limit = 100, int32_t offset = 0);
-    int64_t expireVersions();
-    void storeVersion(const Document& doc);
-    void storeDeletedVersion(const Document& doc);
-
-private:
-    void storeVersionInternal(const Document& doc);
-    void cleanupOldVersions(const std::string& id);
-    bool matchesFilter(const Document& doc, const Filter& filter) const;
-    bool matchesAllFilters(const Document& doc, const std::vector<Filter>& filters) const;
-    nlohmann::json getFieldValue(const nlohmann::json& data, const std::string& field) const;
-    bool validateSchema(const nlohmann::json& data) const;
-    void sortDocuments(std::vector<Document>& docs, const std::vector<Sort>& sorts) const;
-    Document applyProjection(const Document& doc, const std::vector<std::string>& fields) const;
-
-    CollectionConfig config_;
-    std::unordered_map<std::string, Document> documents_;
-    std::unordered_map<std::string, std::vector<DocumentVersion>> versions_;  // doc_id -> versions
-    mutable std::shared_mutex mutex_;
-};
-
-// MemoryStore - manages multiple collections
-class MemoryStore {
-public:
-    using MutationCallback = std::function<void(const WalEntry&)>;
-
-    MemoryStore();
-    ~MemoryStore();
-
-    // Set mutation callback for WAL
-    void setMutationCallback(MutationCallback callback);
-
-    // Collection management
-    common::Result<void> createCollection(const CollectionConfig& config);
-    common::Result<void> dropCollection(const std::string& name);
-    bool hasCollection(const std::string& name) const;
-    std::vector<std::string> listCollections() const;
-    Collection* getCollection(const std::string& name);
-    const Collection* getCollection(const std::string& name) const;
-
-    // Document operations (convenience wrappers)
-    common::Result<Document> get(const std::string& collection, const std::string& id);
-    common::Result<Document> insert(const std::string& collection, Document doc);
-    common::Result<Document> update(const std::string& collection, const std::string& id,
-                                    const nlohmann::json& data, int64_t expected_version = 0,
-                                    bool partial = false);
-    common::Result<void> remove(const std::string& collection, const std::string& id,
-                                int64_t expected_version = 0);
-    QueryResult query(const Query& q);
-
-    // TTL expiration (run periodically)
-    void expireAllDocuments();
-    void expireAllVersions();
-
-    // Version operations
-    common::Result<DocumentVersion> getVersion(const std::string& collection,
-                                               const std::string& id,
-                                               int64_t version);
-    QueryResult listVersions(const std::string& collection,
-                            const std::string& id,
-                            int32_t limit = 100,
-                            int32_t offset = 0);
-
-    // Snapshot support
-    nlohmann::json createSnapshot() const;
-    void loadSnapshot(const nlohmann::json& snapshot);
-
-    // Stats
-    struct Stats {
-        size_t collection_count = 0;
-        size_t total_documents = 0;
-        size_t memory_estimate_bytes = 0;
-    };
-    Stats getStats() const;
-
-private:
-    void notifyMutation(WalEntry entry);
-    int64_t nextSequence();
-
-    std::unordered_map<std::string, std::unique_ptr<Collection>> collections_;
-    mutable std::shared_mutex mutex_;
-    MutationCallback mutation_callback_;
-    std::atomic<int64_t> sequence_{0};
-};
-
-} // namespace smartbotic::database

src/database/persistence/snapshot.cpp

@@ -1,159 +0,0 @@
-#include "snapshot.hpp"
-#include "common/time_utils.hpp"
-#include "logging/logger.hpp"
-#include <fstream>
-#include <algorithm>
-
-namespace smartbotic::database {
-
-using namespace common;
-
-SnapshotManager::SnapshotManager(const Config& config)
-    : config_(config) {
-    if (config_.enabled && !config_.directory.empty()) {
-        std::filesystem::create_directories(config_.directory);
-    }
-}
-
-Result<std::filesystem::path> SnapshotManager::save(const nlohmann::json& data) {
-    if (!config_.enabled) {
-        return Error(ErrorCode::FailedPrecondition, "Snapshots disabled");
-    }
-
-    auto path = generatePath();
-
-    try {
-        std::ofstream file(path, std::ios::binary);
-        if (!file.is_open()) {
-            return Error(ErrorCode::Internal, "Failed to create snapshot file: " + path.string());
-        }
-
-        // Write with pretty formatting for debugging, or dump() for compact
-        file << data.dump();
-        file.close();
-
-        LOG_INFO("Created snapshot: {}", path.string());
-
-        // Update latest sequence
-        if (data.contains("sequence")) {
-            latest_sequence_ = data["sequence"].get<int64_t>();
-        }
-
-        // Cleanup old snapshots
-        cleanup();
-
-        return path;
-    } catch (const std::exception& e) {
-        return Error(ErrorCode::Internal, "Failed to save snapshot: " + std::string(e.what()));
-    }
-}
-
-Result<nlohmann::json> SnapshotManager::loadLatest() {
-    auto path = getLatestPath();
-    if (!path) {
-        return Error(ErrorCode::NotFound, "No snapshot found");
-    }
-    return load(*path);
-}
-
-Result<nlohmann::json> SnapshotManager::load(const std::filesystem::path& path) {
-    if (!std::filesystem::exists(path)) {
-        return Error(ErrorCode::NotFound, "Snapshot not found: " + path.string());
-    }
-
-    try {
-        std::ifstream file(path, std::ios::binary);
-        if (!file.is_open()) {
-            return Error(ErrorCode::Internal, "Failed to open snapshot: " + path.string());
-        }
-
-        nlohmann::json data = nlohmann::json::parse(file);
-
-        if (data.contains("sequence")) {
-            latest_sequence_ = data["sequence"].get<int64_t>();
-        }
-
-        LOG_INFO("Loaded snapshot: {}", path.string());
-        return data;
-    } catch (const std::exception& e) {
-        return Error(ErrorCode::Internal, "Failed to load snapshot: " + std::string(e.what()));
-    }
-}
-
-std::optional<std::filesystem::path> SnapshotManager::getLatestPath() const {
-    auto snapshots = getSnapshots();
-    if (snapshots.empty()) {
-        return std::nullopt;
-    }
-
-    // Sort by modification time (newest first)
-    std::sort(snapshots.begin(), snapshots.end(),
-        [](const auto& a, const auto& b) {
-            return std::filesystem::last_write_time(a) > std::filesystem::last_write_time(b);
-        });
-
-    return snapshots[0];
-}
-
-int64_t SnapshotManager::getLatestSequence() const {
-    if (latest_sequence_ > 0) {
-        return latest_sequence_;
-    }
-
-    auto path = getLatestPath();
-    if (!path) {
-        return 0;
-    }
-
-    try {
-        std::ifstream file(*path);
-        nlohmann::json data = nlohmann::json::parse(file);
-        latest_sequence_ = data.value("sequence", 0);
-        return latest_sequence_;
-    } catch (...) {
-        return 0;
-    }
-}
-
-void SnapshotManager::cleanup() {
-    auto snapshots = getSnapshots();
-    if (snapshots.size() <= static_cast<size_t>(config_.max_snapshots)) {
-        return;
-    }
-
-    // Sort by modification time (oldest first)
-    std::sort(snapshots.begin(), snapshots.end(),
-        [](const auto& a, const auto& b) {
-            return std::filesystem::last_write_time(a) < std::filesystem::last_write_time(b);
-        });
-
-    // Remove oldest snapshots
-    size_t to_remove = snapshots.size() - config_.max_snapshots;
-    for (size_t i = 0; i < to_remove; ++i) {
-        std::filesystem::remove(snapshots[i]);
-        LOG_INFO("Removed old snapshot: {}", snapshots[i].string());
-    }
-}
-
-std::vector<std::filesystem::path> SnapshotManager::getSnapshots() const {
-    std::vector<std::filesystem::path> snapshots;
-
-    if (!std::filesystem::exists(config_.directory)) {
-        return snapshots;
-    }
-
-    for (const auto& entry : std::filesystem::directory_iterator(config_.directory)) {
-        if (entry.is_regular_file() && entry.path().extension() == ".snapshot") {
-            snapshots.push_back(entry.path());
-        }
-    }
-
-    return snapshots;
-}
-
-std::filesystem::path SnapshotManager::generatePath() const {
-    auto timestamp = TimeUtils::nowMs();
-    return config_.directory / ("snapshot_" + std::to_string(timestamp) + ".snapshot");
-}
-
-} // namespace smartbotic::database

src/database/persistence/snapshot.hpp

@@ -1,53 +0,0 @@
-#pragma once
-
-#include <string>
-#include <filesystem>
-#include <nlohmann/json.hpp>
-#include "common/error.hpp"
-
-namespace smartbotic::database {
-
-class MemoryStore;
-
-// Snapshot manager for full state persistence
-class SnapshotManager {
-public:
-    struct Config {
-        std::filesystem::path directory;
-        int64_t interval_sec = 3600;  // Snapshot every hour
-        size_t wal_size_trigger = 100 * 1024 * 1024;  // Snapshot at 100MB WAL
-        int32_t max_snapshots = 5;  // Keep last N snapshots
-        bool enabled = true;
-    };
-
-    explicit SnapshotManager(const Config& config);
-
-    // Save snapshot
-    common::Result<std::filesystem::path> save(const nlohmann::json& data);
-
-    // Load latest snapshot
-    common::Result<nlohmann::json> loadLatest();
-
-    // Load specific snapshot
-    common::Result<nlohmann::json> load(const std::filesystem::path& path);
-
-    // Get latest snapshot path
-    std::optional<std::filesystem::path> getLatestPath() const;
-
-    // Get sequence number from latest snapshot
-    int64_t getLatestSequence() const;
-
-    // Cleanup old snapshots
-    void cleanup();
-
-    // Get all snapshot paths
-    std::vector<std::filesystem::path> getSnapshots() const;
-
-private:
-    std::filesystem::path generatePath() const;
-
-    Config config_;
-    mutable int64_t latest_sequence_ = 0;
-};
-
-} // namespace smartbotic::database

src/database/persistence/wal.cpp

@@ -1,243 +0,0 @@
-#include "wal.hpp"
-#include "logging/logger.hpp"
-#include <algorithm>
-
-namespace smartbotic::database {
-
-using namespace common;
-
-WAL::WAL(const Config& config)
-    : config_(config) {
-    if (config_.enabled && !config_.directory.empty()) {
-        std::filesystem::create_directories(config_.directory);
-    }
-}
-
-WAL::~WAL() {
-    stop();
-}
-
-void WAL::start() {
-    if (!config_.enabled || running_) {
-        return;
-    }
-
-    running_ = true;
-    sync_thread_ = std::thread(&WAL::syncLoop, this);
-    LOG_INFO("WAL started, directory: {}", config_.directory.string());
-}
-
-void WAL::stop() {
-    if (!running_) {
-        return;
-    }
-
-    running_ = false;
-    cv_.notify_all();
-
-    if (sync_thread_.joinable()) {
-        sync_thread_.join();
-    }
-
-    std::lock_guard lock(mutex_);
-    if (current_file_.is_open()) {
-        current_file_.flush();
-        current_file_.close();
-    }
-
-    LOG_INFO("WAL stopped");
-}
-
-Result<void> WAL::append(const WalEntry& entry) {
-    if (!config_.enabled) {
-        return Result<void>();
-    }
-
-    std::lock_guard lock(mutex_);
-
-    pending_entries_.push(entry);
-    current_sequence_ = std::max(current_sequence_, entry.sequence);
-
-    cv_.notify_one();
-
-    return Result<void>();
-}
-
-Result<void> WAL::replay(std::function<void(const WalEntry&)> callback) {
-    if (!config_.enabled) {
-        return Result<void>();
-    }
-
-    auto files = getWalFiles();
-    std::sort(files.begin(), files.end());
-
-    int64_t replayed = 0;
-
-    for (const auto& file_path : files) {
-        std::ifstream file(file_path);
-        if (!file.is_open()) {
-            LOG_WARN("Failed to open WAL file: {}", file_path.string());
-            continue;
-        }
-
-        std::string line;
-        while (std::getline(file, line)) {
-            if (line.empty()) continue;
-
-            try {
-                auto j = nlohmann::json::parse(line);
-                auto entry = WalEntry::fromJson(j);
-                callback(entry);
-                current_sequence_ = std::max(current_sequence_, entry.sequence);
-                ++replayed;
-            } catch (const std::exception& e) {
-                LOG_WARN("Failed to parse WAL entry: {}", e.what());
-            }
-        }
-    }
-
-    LOG_INFO("Replayed {} WAL entries, current sequence: {}", replayed, current_sequence_);
-    return Result<void>();
-}
-
-Result<void> WAL::truncate(int64_t up_to_sequence) {
-    if (!config_.enabled) {
-        return Result<void>();
-    }
-
-    std::lock_guard lock(mutex_);
-
-    // Close current file
-    if (current_file_.is_open()) {
-        current_file_.close();
-    }
-
-    // Remove old WAL files
-    auto files = getWalFiles();
-    for (const auto& file_path : files) {
-        // Parse sequence from filename
-        auto filename = file_path.filename().string();
-        if (filename.starts_with("wal_")) {
-            try {
-                auto seq_str = filename.substr(4, filename.find('.') - 4);
-                int64_t file_seq = std::stoll(seq_str);
-                if (file_seq <= up_to_sequence) {
-                    std::filesystem::remove(file_path);
-                    LOG_DEBUG("Removed WAL file: {}", file_path.string());
-                }
-            } catch (...) {
-                // Ignore parse errors
-            }
-        }
-    }
-
-    // Reset state
-    current_file_path_.clear();
-    current_file_size_ = 0;
-
-    return Result<void>();
-}
-
-size_t WAL::getSize() const {
-    if (!config_.enabled) {
-        return 0;
-    }
-
-    size_t total = 0;
-    auto files = getWalFiles();
-    for (const auto& file : files) {
-        total += std::filesystem::file_size(file);
-    }
-    return total;
-}
-
-void WAL::flush() {
-    std::lock_guard lock(mutex_);
-    if (current_file_.is_open()) {
-        current_file_.flush();
-    }
-}
-
-void WAL::syncLoop() {
-    while (running_) {
-        std::vector<WalEntry> entries_to_write;
-
-        {
-            std::unique_lock lock(mutex_);
-            cv_.wait_for(lock, std::chrono::milliseconds(config_.sync_interval_ms),
-                        [this] { return !pending_entries_.empty() || !running_; });
-
-            while (!pending_entries_.empty()) {
-                entries_to_write.push_back(std::move(pending_entries_.front()));
-                pending_entries_.pop();
-            }
-        }
-
-        if (entries_to_write.empty()) {
-            continue;
-        }
-
-        // Write entries outside the lock
-        {
-            std::lock_guard lock(mutex_);
-
-            // Rotate if needed
-            rotateIfNeeded();
-
-            // Ensure file is open
-            if (!current_file_.is_open()) {
-                current_file_path_ = getWalFilePath(current_sequence_);
-                current_file_.open(current_file_path_, std::ios::app | std::ios::binary);
-                if (!current_file_.is_open()) {
-                    LOG_ERROR("Failed to open WAL file: {}", current_file_path_.string());
-                    continue;
-                }
-            }
-
-            for (const auto& entry : entries_to_write) {
-                std::string line = entry.toJson().dump() + "\n";
-                current_file_ << line;
-                current_file_size_ += line.size();
-            }
-
-            current_file_.flush();
-        }
-    }
-}
-
-void WAL::rotateIfNeeded() {
-    if (current_file_size_ >= config_.max_file_size) {
-        if (current_file_.is_open()) {
-            current_file_.close();
-        }
-        current_file_path_ = getWalFilePath(current_sequence_);
-        current_file_.open(current_file_path_, std::ios::app | std::ios::binary);
-        current_file_size_ = 0;
-        LOG_INFO("Rotated WAL file: {}", current_file_path_.string());
-    }
-}
-
-std::filesystem::path WAL::getWalFilePath(int64_t sequence) const {
-    return config_.directory / ("wal_" + std::to_string(sequence) + ".log");
-}
-
-std::vector<std::filesystem::path> WAL::getWalFiles() const {
-    std::vector<std::filesystem::path> files;
-
-    if (!std::filesystem::exists(config_.directory)) {
-        return files;
-    }
-
-    for (const auto& entry : std::filesystem::directory_iterator(config_.directory)) {
-        if (entry.is_regular_file() && entry.path().extension() == ".log") {
-            auto filename = entry.path().filename().string();
-            if (filename.starts_with("wal_")) {
-                files.push_back(entry.path());
-            }
-        }
-    }
-
-    return files;
-}
-
-} // namespace smartbotic::database

src/database/persistence/wal.hpp

@@ -1,71 +0,0 @@
-#pragma once
-
-#include <string>
-#include <fstream>
-#include <mutex>
-#include <filesystem>
-#include <functional>
-#include <thread>
-#include <atomic>
-#include <condition_variable>
-#include <queue>
-#include "../document.hpp"
-#include "common/error.hpp"
-
-namespace smartbotic::database {
-
-// Write-Ahead Log for durability
-class WAL {
-public:
-    struct Config {
-        std::filesystem::path directory;
-        int64_t sync_interval_ms = 100;
-        size_t max_file_size = 100 * 1024 * 1024;  // 100MB per file
-        bool enabled = true;
-    };
-
-    explicit WAL(const Config& config);
-    ~WAL();
-
-    // Start/stop background sync
-    void start();
-    void stop();
-
-    // Write entry to log
-    common::Result<void> append(const WalEntry& entry);
-
-    // Replay log entries (used during recovery)
-    common::Result<void> replay(std::function<void(const WalEntry&)> callback);
-
-    // Get current sequence number
-    int64_t getCurrentSequence() const { return current_sequence_; }
-
-    // Truncate WAL after snapshot
-    common::Result<void> truncate(int64_t up_to_sequence);
-
-    // Get WAL size in bytes
-    size_t getSize() const;
-
-    // Force flush
-    void flush();
-
-private:
-    void syncLoop();
-    void rotateIfNeeded();
-    std::filesystem::path getWalFilePath(int64_t sequence) const;
-    std::vector<std::filesystem::path> getWalFiles() const;
-
-    Config config_;
-    std::ofstream current_file_;
-    std::filesystem::path current_file_path_;
-    int64_t current_sequence_ = 0;
-    size_t current_file_size_ = 0;
-
-    std::mutex mutex_;
-    std::queue<WalEntry> pending_entries_;
-    std::condition_variable cv_;
-    std::thread sync_thread_;
-    std::atomic<bool> running_{false};
-};
-
-} // namespace smartbotic::database