#include "smartbotic/webserver/database_client.hpp" #include #include #include #include #include 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 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 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(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(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(config_.initial_backoff.count())); args.SetInt(GRPC_ARG_MAX_RECONNECT_BACKOFF_MS, static_cast(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(config_.initial_backoff.count()); auto multiplier = std::pow(config_.backoff_multiplier, static_cast(attempt)); auto backoff_ms = static_cast(base_backoff * multiplier); // Cap at max backoff backoff_ms = std::min(backoff_ms, static_cast(config_.max_backoff.count())); // Add jitter (±10%) auto jitter = static_cast(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