smartbotic-database/service/src/database_service.cpp

#include "database_service.hpp"
#include "json_parse.hpp"

#include <grpcpp/grpcpp.h>
#include <grpcpp/resource_quota.h>
#include <spdlog/spdlog.h>

#include <fstream>

#ifdef HAVE_SYSTEMD
#include <systemd/sd-daemon.h>
#endif

#if defined(__GLIBC__) && !defined(__APPLE__)
#include <malloc.h>
#endif

namespace smartbotic::database {

DatabaseService::DatabaseService(Config config)
    : config_(std::move(config))
{
}

DatabaseService::~DatabaseService() {
    stop();
}

std::string DatabaseService::readOnlyReason() const {
    std::lock_guard<std::mutex> lock(reason_mutex_);
    return read_only_reason_;
}

void DatabaseService::setReadOnly(bool value, const std::string& reason) {
    {
        std::lock_guard<std::mutex> lock(reason_mutex_);
        read_only_reason_ = value ? reason : "";
    }
    bool prev = read_only_.exchange(value, std::memory_order_acq_rel);
    if (prev != value) {
        if (value) {
            spdlog::error("Database entered READ-ONLY mode: {}", reason);
        } else {
            spdlog::info("Database unlocked -- writes accepted");
        }
    }
}

bool DatabaseService::initialize() {
    spdlog::info("Initializing database service (node: {})", config_.nodeId);

    try {
        // Create data directory if needed
        std::error_code ec;
        std::filesystem::create_directories(config_.dataDirectory, ec);
        if (ec) {
            spdlog::error("Failed to create data directory: {}", ec.message());
            return false;
        }

        setupComponents();

        // Initialize encryption
        if (config_.encryptionEnabled) {
            if (!encryption_->initialize()) {
                spdlog::error("Failed to initialize encryption");
                return false;
            }
        }

        // Recover from persistence
        recovery_outcome_ = persistence_->recover(*store_);

#if defined(__GLIBC__) && !defined(__APPLE__)
        // v1.9.3 — release freelist pages accumulated during recovery. v1.9.1
        // added trim at end of loadSnapshot, but the WAL replay phase that
        // runs afterward (`persistence_->recover`'s `replayWal` step) also
        // burns through GBs of small Document JSON allocations whose pages
        // sit on the per-thread freelist with no subsequent allocation to
        // shake them loose. On Zoe (BUG-memory-leak-zoe.md update 15:05)
        // this was 5.7 GB at 23 min uptime — fully reclaimable via
        // malloc_trim, just nothing called it. Paired with the periodic
        // every-5-minute trim in MemoryStore::logMemoryCheck, so any
        // post-boot bloat that escapes this trim gets cleaned up shortly.
        ::malloc_trim(0);
#endif

        if (recovery_outcome_.isFailure()) {
            const std::string modeStr =
                recoveryModeToString(config_.persistenceConfig.recoveryMode);
            spdlog::error("");
            spdlog::error("+------------------------------------------------------------------+");
            spdlog::error("| RECOVERY REFUSED                                                 |");
            spdlog::error("|                                                                  |");
            spdlog::error("| Mode:              {}", modeStr);
            spdlog::error("| Expected snapshot: {}", recovery_outcome_.expectedSnapshot.string());
            spdlog::error("| Reason:            {}", recovery_outcome_.failureReason);
            spdlog::error("|                                                                  |");
            spdlog::error("| Snapshots available:  {}", recovery_outcome_.snapshotsAvailable);
            spdlog::error("| Snapshots attempted:  {}", recovery_outcome_.snapshotsAttempted);
            spdlog::error("|                                                                  |");
            spdlog::error("| To escalate, restart with one of:                                |");
            spdlog::error("|   --recovery-mode=snapshot_fallback   Try older snapshots        |");
            spdlog::error("|   --recovery-mode=wal_only            Replay WAL only (slow)     |");
            spdlog::error("|   --recovery-mode=best_effort         Try all of the above       |");
            spdlog::error("|   --recovery-mode=force_empty         Start empty (LAST RESORT)  |");
            spdlog::error("|                                                                  |");
            spdlog::error("| Or set in config.json:  \"recovery\": {{ \"mode\": \"<mode>\" }}    |");
            spdlog::error("|                                                                  |");
            spdlog::error("| PRESERVE /var/lib/smartbotic-database/ BEFORE ESCALATING.        |");
            spdlog::error("+------------------------------------------------------------------+");
            throw std::runtime_error("recovery failed; refusing to start");
        }

        // Auto-readonly mode on non-trivial recovery
        if (recovery_outcome_.isNonTrivial() && !force_readwrite_) {
            std::string reason;
            switch (recovery_outcome_.kind) {
                case RecoveryOutcome::Kind::SnapshotFellBack:
                    reason = "fell back to snapshot " +
                             recovery_outcome_.snapshotUsed.filename().string() +
                             " because " +
                             recovery_outcome_.expectedSnapshot.filename().string() +
                             " failed: " + recovery_outcome_.failureReason;
                    break;
                case RecoveryOutcome::Kind::WalOnlyReplay:
                    reason = "WAL-only replay, no snapshot loaded (" +
                             std::to_string(recovery_outcome_.walEntriesReplayed) +
                             " entries)";
                    break;
                case RecoveryOutcome::Kind::ForcedEmpty:
                    reason = "forced empty by operator (--recovery-mode=force_empty)";
                    break;
                default:
                    reason = "non-trivial recovery";
                    break;
            }
            reason += ". Run `smartbotic-db-cli unlock` to accept this state, "
                      "or restart with --force-readwrite to bypass this check.";

            setReadOnly(true, reason);

            spdlog::error("");
            spdlog::error("+------------------------------------------------------------------+");
            spdlog::error("| [ERROR] Database booted in READ-ONLY mode after non-trivial      |");
            spdlog::error("|         recovery.                                                 |");
            spdlog::error("|                                                                  |");
            spdlog::error("| Reason: {}", reason);
            spdlog::error("|                                                                  |");
            spdlog::error("| Writes will be REJECTED until you acknowledge this state:        |");
            spdlog::error("|   smartbotic-db-cli unlock                 # live, no restart    |");
            spdlog::error("|   smartbotic-database --force-readwrite    # on next restart     |");
            spdlog::error("+------------------------------------------------------------------+");
        }

        // Set replication sequence after recovery (WAL sequence is now known)
        replication_->setSequence(persistence_->currentWalSequence());

        // v1.8.0 — start the persistence manager before loadFromStore /
        // migrations so any system-collection writes those phases produce
        // (view docs, _collection_meta entries, _migrations recordings)
        // reach the WAL like normal mutations. Pre-v1.8, persistence_ was
        // started later in start(), which silently dropped those writes
        // (running_=false → logInsert no-op). That meant migrated views
        // only "survived" because the runner re-created them every boot,
        // and a manually-created view that landed during the racy startup
        // window (before full readiness) could be lost.
        if (!persistence_->start()) {
            spdlog::error("Failed to start persistence manager before migrations");
            return false;
        }

        // Load view definitions from the _views system collection (which is now
        // populated by the persistence recovery above).
        view_manager_->loadFromStore();

        // Load per-collection configs from the _collection_meta system collection.
        // Must happen AFTER persistence recovery and BEFORE migrations so any
        // migration-created documents are stamped with the correct precision.
        config_manager_->loadFromStore();

        // Run migrations if enabled
        if (config_.migrations.enabled && !config_.migrations.directory.empty()) {
            if (!runMigrations()) {
                if (config_.migrations.failOnError) {
                    spdlog::error("Failed to run migrations");
                    return false;
                }
                spdlog::warn("Some migrations failed, continuing anyway");
            }
        }

        spdlog::info("Database service initialized successfully");
        return true;

    } catch (const std::exception& e) {
        spdlog::error("Failed to initialize database service: {}", e.what());
        return false;
    }
}

bool DatabaseService::runMigrations() {
    if (!config_.migrations.enabled) {
        return true;
    }

    MigrationRunner::Config migrationConfig;
    migrationConfig.directory = config_.migrations.directory;
    migrationConfig.autoApply = config_.migrations.autoApply;
    migrationConfig.failOnError = config_.migrations.failOnError;

    migrationRunner_ = std::make_unique<MigrationRunner>(*store_, *view_manager_, migrationConfig);
    return migrationRunner_->runMigrations();
}

void DatabaseService::start() {
    if (running_.exchange(true)) {
        return;
    }

    spdlog::info("Starting database service on {}:{}", config_.bindAddress, config_.rpcPort);

    // Start components
    store_->start();
    persistence_->start();
    events_->start();
    files_->start();

    if (config_.replicationEnabled) {
        replication_->start();
        // Set initial local collections for discovery
        replication_->setLocalCollections(store_->listCollections());
    }

    // Start gRPC server
    startGrpcServer();

    spdlog::info("Database service started");
}

void DatabaseService::stop() {
    if (!running_.exchange(false)) {
        return;
    }

    spdlog::info("Stopping database service...");

    // Stop gRPC server first
    stopGrpcServer();

    // Stop components in reverse order
    if (replication_) {
        replication_->stop();
    }
    if (files_) {
        files_->stop();
    }
    if (events_) {
        events_->stop();
    }
    if (persistence_ && store_) {
        // v1.8.0 — take a final snapshot on graceful shutdown so the next
        // boot's recovery is "trivial" (snapshot-loaded) instead of "WAL-only
        // replay" which trips auto-readonly mode. This makes ordinary
        // restarts pass through cleanly without needing
        // `smartbotic-db-cli unlock`.
        //
        // v1.8.2 — extend systemd's stop watchdog before doing it. On Zoe
        // (1.85M docs / ~5 GB tracked) the serialize takes ~70 s and the
        // default TimeoutStopSec=30s SIGKILLed the process mid-write, peak
        // RSS hit 11 GB (live state + uncompressed serialize buffer), and
        // the new snapshot never made it to disk anyway. EXTEND_TIMEOUT_USEC
        // tells systemd we're working — same protocol the recovery path
        // uses during startup. Paired with TimeoutStopSec=300 in the unit
        // file as the static cap.
#ifdef HAVE_SYSTEMD
        sd_notify(0, "EXTEND_TIMEOUT_USEC=600000000");
        sd_notify(0, "STATUS=Writing final snapshot");
#endif
        try {
            persistence_->forceSnapshot(*store_);
            spdlog::info("Final snapshot taken on shutdown");
        } catch (const std::exception& e) {
            spdlog::warn("Final snapshot on shutdown failed: {} — recovery on next "
                         "boot may be WAL-only and trip auto-readonly mode", e.what());
        }
    }
    if (persistence_) {
        persistence_->stop();
    }
    if (store_) {
        store_->stop();
    }

    spdlog::info("Database service stopped");
}

void DatabaseService::wait() {
    if (serverThread_.joinable()) {
        serverThread_.join();
    }
}

void DatabaseService::signalStop() {
    stopRequested_ = true;
    stop();
}

nlohmann::json DatabaseService::getStats() const {
    nlohmann::json stats;

    if (store_) {
        auto storeStats = store_->getStats();
        stats["documents"] = storeStats.totalDocuments;
        stats["collections"] = storeStats.totalCollections;
        stats["memory_bytes"] = storeStats.estimatedMemoryBytes;
        stats["inserts"] = storeStats.insertCount;
        stats["updates"] = storeStats.updateCount;
        stats["deletes"] = storeStats.deleteCount;
        stats["queries"] = storeStats.queryCount;
    }

    if (persistence_) {
        auto persistStats = persistence_->getStats();
        stats["wal_sequence"] = persistStats.walSequence;
        stats["wal_size_bytes"] = persistStats.walSizeBytes;
        stats["snapshot_count"] = persistStats.snapshotCount;
        stats["last_snapshot_sequence"] = persistStats.lastSnapshotSequence;
    }

    if (events_) {
        stats["subscriptions"] = events_->subscriptionCount();
    }

    return stats;
}

DatabaseService::Config DatabaseService::loadConfig(const std::filesystem::path& configPath) {
    std::ifstream file(configPath);
    if (!file) {
        throw std::runtime_error("Failed to open config file: " + configPath.string());
    }

    nlohmann::json json;
    file >> json;
    return parseConfig(json);
}

DatabaseService::Config DatabaseService::parseConfig(const nlohmann::json& json) {
    Config config;

    // Check for both "storage" and "database" keys for backward compatibility
    const nlohmann::json* dbConfig = nullptr;
    if (json.contains("database")) {
        dbConfig = &json["database"];
    } else if (json.contains("storage")) {
        dbConfig = &json["storage"];
    }

    if (dbConfig) {
        const auto& db = *dbConfig;

        config.nodeId = db.value("node_id", config.nodeId);
        config.bindAddress = db.value("bind_address", config.bindAddress);
        config.rpcPort = db.value("rpc_port", config.rpcPort);

        // Expand environment variables in data directory
        std::string dataDir = db.value("data_directory", "");
        if (dataDir.find("${HOME}") != std::string::npos) {
            const char* home = std::getenv("HOME");
            if (home) {
                size_t pos = dataDir.find("${HOME}");
                dataDir.replace(pos, 7, home);
            }
        }
        config.dataDirectory = dataDir;

        // Migrations settings
        if (db.contains("migrations")) {
            auto& migrations = db["migrations"];
            config.migrations.enabled = migrations.value("enabled", config.migrations.enabled);
            config.migrations.autoApply = migrations.value("auto_apply", config.migrations.autoApply);
            config.migrations.failOnError = migrations.value("fail_on_error", config.migrations.failOnError);

            std::string migDir = migrations.value("directory", "");
            if (migDir.find("${HOME}") != std::string::npos) {
                const char* home = std::getenv("HOME");
                if (home) {
                    size_t pos = migDir.find("${HOME}");
                    migDir.replace(pos, 7, home);
                }
            }
            config.migrations.directory = migDir;
        }

        // Memory eviction settings
        if (db.contains("memory")) {
            auto& memory = db["memory"];
            config.maxMemoryMb = memory.value("max_memory_mb", config.maxMemoryMb);
            config.evictionThresholdPercent = memory.value("eviction_threshold_percent", config.evictionThresholdPercent);
            config.evictionTargetPercent = memory.value("eviction_target_percent", config.evictionTargetPercent);
            config.evictionCheckIntervalMs = memory.value("eviction_check_interval_ms", config.evictionCheckIntervalMs);

            // NEW v1.7.0 eviction tuning (schema-only in T2; runtime use lands in T3/T4)
            config.evictionChunkSize = memory.value("eviction_chunk_size", config.evictionChunkSize);
            config.evictionChunkPauseMs = memory.value("eviction_chunk_pause_ms", config.evictionChunkPauseMs);
            config.maxEvictionPassesPerTrigger = memory.value("max_eviction_passes_per_trigger", config.maxEvictionPassesPerTrigger);
            config.hotWriteFloorMs = memory.value("hot_write_floor_ms", config.hotWriteFloorMs);
            config.memorySoftPercent = memory.value("memory_soft_percent", config.memorySoftPercent);
            config.memoryHardPercent = memory.value("memory_hard_percent", config.memoryHardPercent);
            config.memoryEmergencyPercent = memory.value("memory_emergency_percent", config.memoryEmergencyPercent);

            // v1.7.0 T10 — eviction burst event threshold
            config.evictionBurstThreshold = memory.value("eviction_burst_threshold", config.evictionBurstThreshold);
        }

        // Persistence settings
        if (db.contains("persistence")) {
            auto& persistence = db["persistence"];
            config.walSyncIntervalMs = persistence.value("wal_sync_interval_ms", config.walSyncIntervalMs);
            config.snapshotIntervalSec = persistence.value("snapshot_interval_sec", config.snapshotIntervalSec);
            config.compressionEnabled = persistence.value("compression", "lz4") != "none";

            // Snapshot durability (NEW in v1.6.1)
            if (persistence.contains("snapshots")) {
                const auto& snap = persistence["snapshots"];
                config.persistenceConfig.validateAfterWrite = snap.value("validate_after_write", config.persistenceConfig.validateAfterWrite);
                config.persistenceConfig.cleanupOnlyIfVerified = snap.value("cleanup_only_if_verified", config.persistenceConfig.cleanupOnlyIfVerified);
            }

            // Recovery (NEW in v1.6.1)
            if (persistence.contains("recovery")) {
                const auto& rec = persistence["recovery"];
                std::string modeStr = rec.value("mode", std::string("normal"));
                try {
                    config.persistenceConfig.recoveryMode = recoveryModeFromString(modeStr);
                } catch (const std::exception& e) {
                    spdlog::warn("Invalid recovery.mode '{}', defaulting to 'normal': {}",
                                 modeStr, e.what());
                    config.persistenceConfig.recoveryMode = RecoveryMode::Normal;
                }
                config.persistenceConfig.autoEscalate = rec.value("auto_escalate", config.persistenceConfig.autoEscalate);
                config.persistenceConfig.allowEmptyOnFreshInstall = rec.value("allow_empty_on_fresh_install", config.persistenceConfig.allowEmptyOnFreshInstall);
            }
        }

        // File settings
        if (db.contains("files")) {
            auto& files = db["files"];
            config.maxFileSizeMb = files.value("max_file_size_mb", config.maxFileSizeMb);
            config.fileCleanupIntervalSec = files.value("cleanup_orphans_interval_sec", config.fileCleanupIntervalSec);
            if (files.contains("allowed_types")) {
                for (const auto& type : files["allowed_types"]) {
                    config.allowedFileTypes.push_back(type.get<std::string>());
                }
            }
        }

        // Encryption settings
        if (db.contains("encryption")) {
            auto& encryption = db["encryption"];
            config.encryptionEnabled = encryption.value("enabled", config.encryptionEnabled);
            config.autoGenerateKey = encryption.value("auto_generate_key", config.autoGenerateKey);

            std::string keyFile = encryption.value("key_file", "");
            if (keyFile.find("${HOME}") != std::string::npos) {
                const char* home = std::getenv("HOME");
                if (home) {
                    size_t pos = keyFile.find("${HOME}");
                    keyFile.replace(pos, 7, home);
                }
            }
            config.keyFilePath = keyFile;
        }

        // Replication settings
        if (db.contains("replication")) {
            auto& replication = db["replication"];
            config.replicationEnabled = replication.value("enabled", config.replicationEnabled);
            config.conflictResolution = replication.value("conflict_resolution", config.conflictResolution);
            if (replication.contains("peers")) {
                for (const auto& peer : replication["peers"]) {
                    config.peerAddresses.push_back(peer.get<std::string>());
                }
            }
        }

        // gRPC settings (v1.6.2 — concurrency cap + configurable message sizes)
        if (db.contains("grpc")) {
            const auto& grpc = db["grpc"];
            config.grpc.maxReceiveMessageSizeMb =
                grpc.value("max_receive_message_size_mb", config.grpc.maxReceiveMessageSizeMb);
            config.grpc.maxSendMessageSizeMb =
                grpc.value("max_send_message_size_mb", config.grpc.maxSendMessageSizeMb);
            config.grpc.resourceQuotaMemoryMb =
                grpc.value("resource_quota_memory_mb", config.grpc.resourceQuotaMemoryMb);
            config.grpc.maxConcurrentSubscribeStreams =
                grpc.value("max_concurrent_subscribe_streams", config.grpc.maxConcurrentSubscribeStreams);
            config.grpc.maxConcurrentFileStreams =
                grpc.value("max_concurrent_file_streams", config.grpc.maxConcurrentFileStreams);
        }
    }

    return config;
}

void DatabaseService::setupComponents() {
    // Create memory store with eviction config
    MemoryStore::Config storeConfig;
    storeConfig.nodeId = config_.nodeId;
    storeConfig.maxMemoryBytes = config_.maxMemoryMb * 1024ULL * 1024ULL;
    storeConfig.evictionThresholdPercent = config_.evictionThresholdPercent;
    storeConfig.evictionTargetPercent = config_.evictionTargetPercent;
    storeConfig.evictionCheckIntervalMs = config_.evictionCheckIntervalMs;
    // NEW v1.7.0 eviction tuning (wired in T2; consumed in T3/T4).
    storeConfig.evictionChunkSize = config_.evictionChunkSize;
    storeConfig.evictionChunkPauseMs = config_.evictionChunkPauseMs;
    storeConfig.maxEvictionPassesPerTrigger = config_.maxEvictionPassesPerTrigger;
    storeConfig.hotWriteFloorMs = config_.hotWriteFloorMs;
    storeConfig.memorySoftPercent = config_.memorySoftPercent;
    storeConfig.memoryHardPercent = config_.memoryHardPercent;
    storeConfig.memoryEmergencyPercent = config_.memoryEmergencyPercent;
    storeConfig.evictionBurstThreshold = config_.evictionBurstThreshold;
    store_ = std::make_unique<MemoryStore>(storeConfig);

    // Create view manager (cache loaded in initialize() after persistence recovery)
    view_manager_ = std::make_unique<ViewManager>(*store_);

    // Create per-collection config manager and attach it to the store so the
    // write paths route document timestamp stamps through it.
    config_manager_ = std::make_unique<CollectionConfigManager>(*store_);
    store_->setConfigManager(config_manager_.get());

    // v1.9.0 — disk-resident version history. Replaces the in-heap
    // `CollectionData::versionHistory` deques that were the dominant
    // source of the Zoe untracked-RSS gap. One file per collection at
    // `<dataDir>/history/<collection>.hlog`. Lifecycle: constructed
    // here, wired into the store immediately so snapshot deserializer
    // can migrate pre-v1.9 in-memory history blocks straight to disk.
    HistoryStore::Config histConfig;
    histConfig.dataDir = config_.dataDirectory;
    history_store_ = std::make_unique<HistoryStore>(histConfig);
    store_->setHistoryStore(history_store_.get());

    // Create persistence manager. Start from any persistenceConfig values the
    // caller (config loader / CLI parser) has already populated — including
    // recoveryMode — and overlay the top-level convenience fields.
    PersistenceManager::Config persistConfig = config_.persistenceConfig;
    persistConfig.dataDir = config_.dataDirectory;
    persistConfig.walSyncIntervalMs = config_.walSyncIntervalMs;
    persistConfig.snapshotIntervalSec = config_.snapshotIntervalSec;
    persistConfig.compressionEnabled = config_.compressionEnabled;
    persistConfig.nodeId = config_.nodeId;
    // Mirror the resolved recoveryMode back into our Config so downstream code
    // (logging, RPC handlers) can inspect config_.persistenceConfig.recoveryMode
    // without having to reach into the PersistenceManager.
    config_.persistenceConfig = persistConfig;
    persistence_ = std::make_unique<PersistenceManager>(persistConfig);

    // Connect store callbacks to persistence - uses setPersistCallback for WAL logging
    store_->setPersistCallback([this](const std::string& collection, const std::string& id,
                                       const std::optional<Document>& doc, EventType eventType) {
        // Log to WAL based on event type. v1.7.4: capture the assigned WAL
        // sequence and record it on the store so eviction stubs can carry
        // a real seq (instead of doc.version, which was useless as a
        // `fromSequence` hint and forced a full WAL scan on every fault).
        uint64_t walSeq = 0;
        switch (eventType) {
            case EventType::INSERT:
                if (doc) walSeq = persistence_->logInsert(collection, *doc);
                break;
            case EventType::UPDATE:
                if (doc) walSeq = persistence_->logUpdate(collection, *doc);
                break;
            case EventType::DELETE:
                persistence_->logDelete(collection, id);
                break;
            default:
                break;
        }
        if (walSeq > 0 && (eventType == EventType::INSERT || eventType == EventType::UPDATE)) {
            store_->recordWalSequence(collection, id, walSeq);
        }

        // Queue for replication broadcast
        if (replication_ && config_.replicationEnabled) {
            databasepb::ReplicationEntry entry;
            entry.set_collection(collection);
            entry.set_document_id(id);
            // v1.8.0 — stamp our nodeId so peers can attribute the write to
            // its actual origin. Pre-v1.8 broadcasts left this empty, which
            // is now treated by receivers as "legacy / unknown origin".
            entry.set_node_id(config_.nodeId);
            entry.set_global_timestamp(
                std::chrono::duration_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now().time_since_epoch()
                ).count());

            switch (eventType) {
                case EventType::INSERT:
                    entry.set_op(databasepb::OP_INSERT);
                    // Send the full Document envelope (id, collection, data,
                    // version, timestamps, encryption state). The follower's
                    // applyReplicatedEntry uses Document::fromJson() which
                    // expects this envelope — sending only doc->data silently
                    // dropped every user field on the other side.
                    if (doc) entry.set_data(doc->toJson().dump());
                    break;
                case EventType::UPDATE:
                    entry.set_op(databasepb::OP_UPDATE);
                    if (doc) entry.set_data(doc->toJson().dump());
                    break;
                case EventType::DELETE:
                    entry.set_op(databasepb::OP_DELETE);
                    break;
                default:
                    break;
            }

            replication_->queueForReplication(entry);
        }

        // Publish event
        if (events_) {
            DatabaseEvent event;
            event.type = eventType;
            event.collection = collection;
            event.documentId = id;
            event.timestamp = std::chrono::duration_cast<std::chrono::milliseconds>(
                std::chrono::system_clock::now().time_since_epoch()
            ).count();
            event.nodeId = config_.nodeId;
            if (doc) {
                event.data = doc->data;
            }
            events_->publish(event);
        }
    });

    // Set up document load callback for LRU eviction recovery.
    //
    // v1.7.4: walSequence is now a real WAL sequence (set by the persist
    // callback below at append time, not the per-doc version counter).
    // Pass `walSequence - 1` as the exclusive `fromSequence` so replay
    // returns entries with seq >= walSequence — the doc's last write is
    // exactly at walSequence, so the very first hit is the one we want.
    // For docs whose seq is unknown (0 — e.g. loaded from a snapshot
    // that predates v1.7.4), fall back to the full-WAL scan from 0.
    store_->setDocumentLoadCallback([this](const std::string& collection,
                                           const std::string& id,
                                           uint64_t walSequence) -> std::optional<Document> {
        uint64_t fromSequence = walSequence > 0 ? walSequence - 1 : 0;
        return persistence_->loadDocument(collection, id, fromSequence);
    });

    // Create encryption manager
    EncryptionManager::Config encryptConfig;
    encryptConfig.enabled = config_.encryptionEnabled;
    encryptConfig.keyFilePath = config_.keyFilePath;
    encryptConfig.autoGenerateKey = config_.autoGenerateKey;
    encryption_ = std::make_unique<EncryptionManager>(encryptConfig);

    // Create event manager
    EventManager::Config eventConfig;
    eventConfig.nodeId = config_.nodeId;
    events_ = std::make_unique<EventManager>(eventConfig);

    // Create file manager
    FileManager::Config fileConfig;
    fileConfig.filesDir = config_.dataDirectory / "files";
    fileConfig.maxFileSizeMb = config_.maxFileSizeMb;
    fileConfig.allowedMimeTypes = config_.allowedFileTypes;
    fileConfig.cleanupIntervalSec = config_.fileCleanupIntervalSec;
    files_ = std::make_unique<FileManager>(fileConfig);

    // Create replication manager
    ReplicationManager::Config replConfig;
    replConfig.nodeId = config_.nodeId;
    replConfig.peerAddresses = config_.peerAddresses;
    replConfig.conflictResolution = config_.conflictResolution;
    replication_ = std::make_unique<ReplicationManager>(replConfig);

    // Wire replication entry apply callback
    replication_->setEntryApplyCallback([this](const databasepb::ReplicationEntry& entry) {
        applyReplicatedEntry(entry);
    });

    // Create gRPC implementations
    storageImpl_ = std::make_unique<DatabaseGrpcImpl>(
        *this, *store_, *persistence_, *events_, *files_, *encryption_, *view_manager_, *config_manager_
    );
    // v1.6.2 — wire the streaming-RPC concurrency limits from GrpcConfig.
    storageImpl_->setStreamLimits(
        config_.grpc.maxConcurrentSubscribeStreams,
        config_.grpc.maxConcurrentFileStreams);
    replicationImpl_ = std::make_unique<DatabaseReplicationGrpcImpl>(
        *store_, *replication_, *persistence_
    );
}

void DatabaseService::applyReplicatedEntry(const databasepb::ReplicationEntry& entry) {
    try {
        // v1.8.0 — also append the replicated entry to the local WAL, tagged
        // with the originating node's id. This is what makes follower-side
        // recovery (eviction → WAL fallback, restart → WAL replay) preserve
        // replicated documents. Echo amplification is prevented by:
        //   - GetEntriesSince emitting walEntry.nodeId (not local node id)
        //   - SyncProtocol skipping entries whose origin is the peer itself
        // both implemented in the same v1.8.0 cycle.
        const std::string origin = entry.node_id().empty()
            ? config_.nodeId   // legacy peer (pre-1.8) — best-guess local
            : entry.node_id();

        switch (entry.op()) {
            case databasepb::OP_INSERT:
            case databasepb::OP_UPDATE:
            case databasepb::OP_UPSERT: {
                if (entry.data().empty()) {
                    spdlog::warn("Replication entry has no data for op {}", static_cast<int>(entry.op()));
                    return;
                }

                auto json = smartbotic::db::parse_to_nlohmann(entry.data());
                Document doc = Document::fromJson(json);
                doc.id = entry.document_id();
                doc.nodeId = entry.node_id();

                // loadDocument bypasses the persist-and-broadcast callback to avoid
                // re-replicating; we explicitly write to the WAL right after with
                // the origin-aware overload so eviction/WAL recovery still works.
                store_->loadDocument(entry.collection(), doc);

                if (persistence_) {
                    uint64_t walSeq = (entry.op() == databasepb::OP_INSERT)
                        ? persistence_->logInsert(entry.collection(), doc, origin)
                        : persistence_->logUpdate(entry.collection(), doc, origin);
                    if (walSeq > 0) {
                        store_->recordWalSequence(entry.collection(), doc.id, walSeq);
                    }
                }

                spdlog::trace("Applied replicated {} to {}/{} from {}",
                             entry.op() == databasepb::OP_INSERT ? "insert" :
                             entry.op() == databasepb::OP_UPDATE ? "update" : "upsert",
                             entry.collection(), entry.document_id(), entry.node_id());
                break;
            }

            case databasepb::OP_DELETE: {
                store_->remove(entry.collection(), entry.document_id());
                if (persistence_) {
                    persistence_->logDelete(entry.collection(), entry.document_id(), origin);
                }
                spdlog::trace("Applied replicated delete to {}/{} from {}",
                             entry.collection(), entry.document_id(), entry.node_id());
                break;
            }

            case databasepb::OP_CREATE_COLLECTION: {
                CollectionOptions options;
                store_->createCollection(entry.collection(), options);
                if (persistence_) {
                    persistence_->logCreateCollection(entry.collection(), options, origin);
                }
                spdlog::trace("Applied replicated create collection {} from {}",
                             entry.collection(), entry.node_id());
                break;
            }

            case databasepb::OP_DROP_COLLECTION: {
                store_->dropCollection(entry.collection());
                if (persistence_) {
                    persistence_->logDropCollection(entry.collection(), origin);
                }
                spdlog::trace("Applied replicated drop collection {} from {}",
                             entry.collection(), entry.node_id());
                break;
            }

            default:
                spdlog::warn("Unknown replication operation type: {}", static_cast<int>(entry.op()));
                break;
        }
    } catch (const std::exception& e) {
        spdlog::error("Failed to apply replicated entry: {}", e.what());
    }
}

void DatabaseService::startGrpcServer() {
    std::string serverAddress = config_.bindAddress + ":" + std::to_string(config_.rpcPort);

    grpc::ServerBuilder builder;
    builder.AddListeningPort(serverAddress, grpc::InsecureServerCredentials());
    builder.RegisterService(storageImpl_.get());
    builder.RegisterService(replicationImpl_.get());

    // Configurable message sizes (default 100 MB for file uploads)
    const int64_t mb = 1024 * 1024;
    builder.SetMaxReceiveMessageSize(
        static_cast<int>(config_.grpc.maxReceiveMessageSizeMb * mb));
    builder.SetMaxSendMessageSize(
        static_cast<int>(config_.grpc.maxSendMessageSizeMb * mb));

    // ResourceQuota bounds total inbound buffer memory across all RPCs
    grpc::ResourceQuota quota("smartbotic-db");
    quota.Resize(config_.grpc.resourceQuotaMemoryMb * mb);
    builder.SetResourceQuota(quota);

    spdlog::info("gRPC config: recv={}MB send={}MB quota={}MB subs={} files={}",
                 config_.grpc.maxReceiveMessageSizeMb,
                 config_.grpc.maxSendMessageSizeMb,
                 config_.grpc.resourceQuotaMemoryMb,
                 config_.grpc.maxConcurrentSubscribeStreams,
                 config_.grpc.maxConcurrentFileStreams);

    grpcServer_ = builder.BuildAndStart();

    if (!grpcServer_) {
        throw std::runtime_error("Failed to start gRPC server on " + serverAddress);
    }

    spdlog::info("gRPC server listening on {}", serverAddress);
}

void DatabaseService::stopGrpcServer() {
    if (grpcServer_) {
        grpcServer_->Shutdown();
        grpcServer_.reset();
    }
}

} // namespace smartbotic::database