smartbotic-crm/webserver/src/database_client.cpp

#include "smartbotic/webserver/database_client.hpp"

#include <spdlog/spdlog.h>

#include <algorithm>
#include <cmath>
#include <cstdlib>
#include <thread>

namespace smartbotic::webserver {

DatabaseClient::DatabaseClient(DatabaseClientConfig config) : config_(std::move(config)) {}

DatabaseClient::DatabaseClient(const HttpServerConfig& server_config) {
    config_.address = server_config.database_address;
}

DatabaseClient::~DatabaseClient() {
    Disconnect();
}

auto DatabaseClient::Connect() -> bool {
    std::lock_guard<std::mutex> lock(mutex_);

    if (connected_.load()) {
        spdlog::debug("Database client already connected");
        return true;
    }

    spdlog::info("Connecting to database at {}", config_.address);

    try {
        CreateChannel();
        CreateStubs();

        // Wait for the channel to become ready
        if (!WaitForReady(config_.connect_timeout)) {
            spdlog::error("Failed to connect to database at {}: connection timeout", config_.address);
            return false;
        }

        connected_.store(true);
        spdlog::info("Successfully connected to database at {}", config_.address);
        return true;
    } catch (const std::exception& e) {
        spdlog::error("Failed to connect to database: {}", e.what());
        return false;
    }
}

void DatabaseClient::Disconnect() {
    std::lock_guard<std::mutex> lock(mutex_);

    if (!connected_.load()) {
        return;
    }

    spdlog::info("Disconnecting from database");

    // Clear stubs
    document_stub_.reset();
    collection_stub_.reset();
    query_stub_.reset();
    admin_stub_.reset();
    subscription_stub_.reset();

    // Release channel
    channel_.reset();

    connected_.store(false);
    spdlog::info("Disconnected from database");
}

auto DatabaseClient::IsConnected() const -> bool {
    if (!connected_.load()) {
        return false;
    }

    // Also verify channel state
    if (channel_) {
        auto state = channel_->GetState(false);
        return state == GRPC_CHANNEL_READY || state == GRPC_CHANNEL_IDLE;
    }

    return false;
}

auto DatabaseClient::HealthCheck() -> HealthCheckResult {
    HealthCheckResult result;

    if (!connected_.load() || !admin_stub_) {
        result.healthy = false;
        result.message = "Not connected to database";
        return result;
    }

    auto start = std::chrono::steady_clock::now();

    // Use GetDatabaseStats as a health check - it's lightweight
    grpc::ClientContext context;
    context.set_deadline(std::chrono::system_clock::now() + std::chrono::seconds(5));

    ::smartbotic::database::GetDatabaseStatsRequest request;
    ::smartbotic::database::DatabaseStats response;

    auto status = admin_stub_->GetDatabaseStats(&context, request, &response);

    auto end = std::chrono::steady_clock::now();
    result.latency = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);

    if (status.ok()) {
        result.healthy = true;
        result.message = "Database is healthy";
    } else {
        result.healthy = false;
        result.message = "Database health check failed: " + status.error_message();
        spdlog::warn("Database health check failed: {} (code: {})", status.error_message(),
                     static_cast<int>(status.error_code()));
    }

    return result;
}

auto DatabaseClient::WaitForReady(std::chrono::milliseconds timeout) -> bool {
    if (!channel_) {
        return false;
    }

    auto deadline = std::chrono::system_clock::now() + timeout;

    // Try to establish connection
    auto state = channel_->GetState(true);  // true = try to connect

    while (state != GRPC_CHANNEL_READY) {
        if (state == GRPC_CHANNEL_SHUTDOWN) {
            spdlog::error("Channel is shutdown");
            return false;
        }

        if (!channel_->WaitForStateChange(state, deadline)) {
            // Timeout
            spdlog::error("Timeout waiting for channel to become ready (current state: {})", GetChannelStateString());
            return false;
        }

        state = channel_->GetState(false);
    }

    return true;
}

auto DatabaseClient::GetChannelState() const -> grpc_connectivity_state {
    if (!channel_) {
        return GRPC_CHANNEL_SHUTDOWN;
    }
    return channel_->GetState(false);
}

auto DatabaseClient::GetChannelStateString() const -> std::string {
    switch (GetChannelState()) {
        case GRPC_CHANNEL_IDLE:
            return "IDLE";
        case GRPC_CHANNEL_CONNECTING:
            return "CONNECTING";
        case GRPC_CHANNEL_READY:
            return "READY";
        case GRPC_CHANNEL_TRANSIENT_FAILURE:
            return "TRANSIENT_FAILURE";
        case GRPC_CHANNEL_SHUTDOWN:
            return "SHUTDOWN";
        default:
            return "UNKNOWN";
    }
}

auto DatabaseClient::GetDocumentService() -> ::smartbotic::database::DocumentService::Stub* {
    return document_stub_.get();
}

auto DatabaseClient::GetCollectionService() -> ::smartbotic::database::CollectionService::Stub* {
    return collection_stub_.get();
}

auto DatabaseClient::GetQueryService() -> ::smartbotic::database::QueryService::Stub* {
    return query_stub_.get();
}

auto DatabaseClient::GetAdminService() -> ::smartbotic::database::AdminService::Stub* {
    return admin_stub_.get();
}

auto DatabaseClient::GetSubscriptionService() -> ::smartbotic::database::SubscriptionService::Stub* {
    return subscription_stub_.get();
}

void DatabaseClient::CreateChannel() {
    grpc::ChannelArguments args;

    // Set keepalive parameters for long-lived connections
    args.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, 30000);           // 30 seconds
    args.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 10000);        // 10 seconds
    args.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);  // Allow keepalive without active calls
    args.SetInt(GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS, 5000);

    // Set maximum message sizes
    args.SetMaxReceiveMessageSize(64 * 1024 * 1024);  // 64 MB
    args.SetMaxSendMessageSize(64 * 1024 * 1024);     // 64 MB

    // Initial reconnect backoff
    args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, static_cast<int>(config_.initial_backoff.count()));
    args.SetInt(GRPC_ARG_MAX_RECONNECT_BACKOFF_MS, static_cast<int>(config_.max_backoff.count()));

    // Create channel with insecure credentials (for local development)
    // In production, this should use TLS credentials
    channel_ = grpc::CreateCustomChannel(config_.address, grpc::InsecureChannelCredentials(), args);

    spdlog::debug("Created gRPC channel to {}", config_.address);
}

void DatabaseClient::CreateStubs() {
    document_stub_ = ::smartbotic::database::DocumentService::NewStub(channel_);
    collection_stub_ = ::smartbotic::database::CollectionService::NewStub(channel_);
    query_stub_ = ::smartbotic::database::QueryService::NewStub(channel_);
    admin_stub_ = ::smartbotic::database::AdminService::NewStub(channel_);
    subscription_stub_ = ::smartbotic::database::SubscriptionService::NewStub(channel_);

    spdlog::debug("Created all gRPC service stubs");
}

auto DatabaseClient::CalculateBackoff(uint32_t attempt) const -> std::chrono::milliseconds {
    // Exponential backoff with jitter
    auto base_backoff = static_cast<double>(config_.initial_backoff.count());
    auto multiplier = std::pow(config_.backoff_multiplier, static_cast<double>(attempt));
    auto backoff_ms = static_cast<int64_t>(base_backoff * multiplier);

    // Cap at max backoff
    backoff_ms = std::min(backoff_ms, static_cast<int64_t>(config_.max_backoff.count()));

    // Add jitter (±10%)
    auto jitter = static_cast<int64_t>(backoff_ms * 0.1);
    backoff_ms += (rand() % (2 * jitter + 1)) - jitter;  // NOLINT(concurrency-mt-unsafe)

    return std::chrono::milliseconds(backoff_ms);
}

auto DatabaseClient::IsRetryable(grpc::StatusCode code) -> bool {
    switch (code) {
        case grpc::StatusCode::UNAVAILABLE:
        case grpc::StatusCode::DEADLINE_EXCEEDED:
        case grpc::StatusCode::ABORTED:
        case grpc::StatusCode::RESOURCE_EXHAUSTED:
            return true;
        default:
            return false;
    }
}

}  // namespace smartbotic::webserver