#include "smartbotic/database/client.hpp" #include #include #include #include #include #include namespace smartbotic::database { // ===== PIMPL Implementation ===== class Client::Impl { public: explicit Impl(Config config) : config_(std::move(config)) {} ~Impl() { disconnect(); } bool connect() { try { auto channelArgs = grpc::ChannelArguments(); // Use longer keepalive intervals to avoid "too_many_pings" errors from server channelArgs.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, 60000); // 60 seconds channelArgs.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 20000); // 20 seconds channelArgs.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 0); // Only ping when there are active calls // Set max message size to match server (100MB for file uploads) channelArgs.SetMaxReceiveMessageSize(100 * 1024 * 1024); channelArgs.SetMaxSendMessageSize(100 * 1024 * 1024); channel_ = grpc::CreateCustomChannel( config_.address, grpc::InsecureChannelCredentials(), channelArgs ); stub_ = smartbotic::databasepb::DatabaseService::NewStub(channel_); connected_ = true; spdlog::info("Database client connected to {}", config_.address); return true; } catch (const std::exception& e) { spdlog::error("Database client connection failed: {}", e.what()); return false; } } void disconnect() { connected_ = false; stub_.reset(); channel_.reset(); } bool isConnected() const { if (!connected_ || !channel_) { return false; } // Check if channel is in a usable state (not failed or shutdown) auto state = channel_->GetState(false); return state == GRPC_CHANNEL_READY || state == GRPC_CHANNEL_IDLE || state == GRPC_CHANNEL_CONNECTING; } // ===== Document Operations ===== std::string insert(const std::string& collection, const nlohmann::json& data, const std::string& id, uint32_t ttlSeconds, const std::string& actor) { smartbotic::databasepb::InsertRequest request; request.set_collection(collection); request.set_data(data.dump()); if (!id.empty()) { request.set_id(id); } if (ttlSeconds > 0) { request.set_ttl_seconds(ttlSeconds); } if (!actor.empty()) { request.set_actor(actor); } smartbotic::databasepb::InsertResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->Insert(&context, request, &response); if (!status.ok()) { spdlog::error("Client::insert failed: {}", status.error_message()); throw std::runtime_error(status.error_message()); } return response.id(); } std::optional get(const std::string& collection, const std::string& id) { smartbotic::databasepb::GetRequest request; request.set_collection(collection); request.set_id(id); smartbotic::databasepb::GetResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->Get(&context, request, &response); if (!status.ok()) { spdlog::error("Client::get failed: {}", status.error_message()); return std::nullopt; } if (!response.found()) { return std::nullopt; } auto json = nlohmann::json::parse(response.document().data()); json["_id"] = response.document().id(); // Include document ID in returned data return json; } bool update(const std::string& collection, const std::string& id, const nlohmann::json& data, const std::string& actor) { smartbotic::databasepb::UpdateRequest request; request.set_collection(collection); request.set_id(id); request.set_data(data.dump()); if (!actor.empty()) { request.set_actor(actor); } smartbotic::databasepb::UpdateResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->Update(&context, request, &response); if (!status.ok()) { spdlog::error("Client::update failed: {}", status.error_message()); return false; } return response.success(); } bool updateIfVersion(const std::string& collection, const std::string& id, const nlohmann::json& data, uint64_t expectedVersion, const std::string& actor) { smartbotic::databasepb::UpdateRequest request; request.set_collection(collection); request.set_id(id); request.set_data(data.dump()); request.set_expected_version(expectedVersion); if (!actor.empty()) { request.set_actor(actor); } smartbotic::databasepb::UpdateResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->Update(&context, request, &response); if (!status.ok()) { spdlog::error("Client::updateIfVersion failed: {}", status.error_message()); return false; } return response.success(); } std::pair upsert(const std::string& collection, const nlohmann::json& data, const std::string& id, uint32_t ttlSeconds, const std::string& actor) { smartbotic::databasepb::UpsertRequest request; request.set_collection(collection); request.set_data(data.dump()); if (!id.empty()) { request.set_id(id); } if (ttlSeconds > 0) { request.set_ttl_seconds(ttlSeconds); } if (!actor.empty()) { request.set_actor(actor); } smartbotic::databasepb::UpsertResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->Upsert(&context, request, &response); if (!status.ok()) { spdlog::error("Client::upsert failed: {}", status.error_message()); throw std::runtime_error(status.error_message()); } return {response.id(), response.inserted()}; } bool remove(const std::string& collection, const std::string& id) { smartbotic::databasepb::DeleteRequest request; request.set_collection(collection); request.set_id(id); smartbotic::databasepb::DeleteResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->Delete(&context, request, &response); if (!status.ok()) { spdlog::error("Client::remove failed: {}", status.error_message()); return false; } return response.deleted(); } bool exists(const std::string& collection, const std::string& id) { smartbotic::databasepb::ExistsRequest request; request.set_collection(collection); request.set_id(id); smartbotic::databasepb::ExistsResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->Exists(&context, request, &response); if (!status.ok()) { spdlog::error("Client::exists failed: {}", status.error_message()); return false; } return response.exists(); } // ===== Query Operations ===== std::vector find(const std::string& collection, const Client::QueryOptions& options) { smartbotic::databasepb::FindRequest request; request.set_collection(collection); // Set filters for (const auto& [field, value] : options.filters) { auto* filter = request.add_filters(); filter->set_field(field); filter->set_value(value.dump()); // Use SEARCH op for _search field, EQ for others if (field == "_search") { filter->set_op(smartbotic::databasepb::FILTER_OP_SEARCH); } else { filter->set_op(smartbotic::databasepb::FILTER_OP_EQ); } } // Set sorting if (!options.sortField.empty()) { auto* sort = request.mutable_sort(); sort->set_field(options.sortField); sort->set_descending(options.sortDescending); } // Set pagination request.set_limit(options.limit); request.set_offset(options.offset); smartbotic::databasepb::FindResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->Find(&context, request, &response); if (!status.ok()) { spdlog::error("Client::find failed: {}", status.error_message()); return {}; } std::vector results; results.reserve(response.documents_size()); for (const auto& doc : response.documents()) { auto json = nlohmann::json::parse(doc.data()); json["_id"] = doc.id(); // Include document ID in returned data json["_created_at"] = doc.created_at(); // Creation timestamp (ms) json["_updated_at"] = doc.updated_at(); // Last update timestamp (ms) json["_created_by"] = doc.created_by(); // User ID who created json["_updated_by"] = doc.updated_by(); // User ID who last updated results.push_back(json); } return results; } uint64_t count(const std::string& collection, const std::vector>& filters) { smartbotic::databasepb::CountRequest request; request.set_collection(collection); for (const auto& [field, value] : filters) { auto* filter = request.add_filters(); filter->set_field(field); filter->set_value(value.dump()); // Use SEARCH op for _search field, EQ for others if (field == "_search") { filter->set_op(smartbotic::databasepb::FILTER_OP_SEARCH); } else { filter->set_op(smartbotic::databasepb::FILTER_OP_EQ); } } smartbotic::databasepb::CountResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->Count(&context, request, &response); if (!status.ok()) { spdlog::error("Client::count failed: {}", status.error_message()); return 0; } return response.count(); } // ===== Set Operations ===== bool setAdd(const std::string& collection, const std::string& setId, const std::string& member) { smartbotic::databasepb::SetAddRequest request; request.set_collection(collection); request.set_set_id(setId); request.set_member(member); smartbotic::databasepb::SetAddResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->SetAdd(&context, request, &response); if (!status.ok()) { spdlog::error("Client::setAdd failed: {}", status.error_message()); return false; } return response.added(); } bool setRemove(const std::string& collection, const std::string& setId, const std::string& member) { smartbotic::databasepb::SetRemoveRequest request; request.set_collection(collection); request.set_set_id(setId); request.set_member(member); smartbotic::databasepb::SetRemoveResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->SetRemove(&context, request, &response); if (!status.ok()) { spdlog::error("Client::setRemove failed: {}", status.error_message()); return false; } return response.removed(); } std::vector setMembers(const std::string& collection, const std::string& setId) { smartbotic::databasepb::SetMembersRequest request; request.set_collection(collection); request.set_set_id(setId); smartbotic::databasepb::SetMembersResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->SetMembers(&context, request, &response); if (!status.ok()) { spdlog::error("Client::setMembers failed: {}", status.error_message()); return {}; } return {response.members().begin(), response.members().end()}; } bool setIsMember(const std::string& collection, const std::string& setId, const std::string& member) { smartbotic::databasepb::SetIsMemberRequest request; request.set_collection(collection); request.set_set_id(setId); request.set_member(member); smartbotic::databasepb::SetIsMemberResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->SetIsMember(&context, request, &response); if (!status.ok()) { spdlog::error("Client::setIsMember failed: {}", status.error_message()); return false; } return response.is_member(); } // ===== Collection Management ===== bool createCollection(const std::string& name, uint32_t defaultTtlSeconds, bool encrypted, uint32_t maxVersions) { smartbotic::databasepb::CreateCollectionRequest request; request.set_name(name); auto* options = request.mutable_options(); if (defaultTtlSeconds > 0) { options->set_default_ttl_seconds(defaultTtlSeconds); } options->set_encrypted(encrypted); if (maxVersions > 0) { options->set_max_versions(maxVersions); } smartbotic::databasepb::CreateCollectionResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->CreateCollection(&context, request, &response); if (!status.ok()) { spdlog::error("Client::createCollection failed: {}", status.error_message()); return false; } return response.created(); } bool dropCollection(const std::string& name) { smartbotic::databasepb::DropCollectionRequest request; request.set_name(name); smartbotic::databasepb::DropCollectionResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->DropCollection(&context, request, &response); if (!status.ok()) { spdlog::error("Client::dropCollection failed: {}", status.error_message()); return false; } return response.dropped(); } std::vector listCollections() { smartbotic::databasepb::ListCollectionsRequest request; smartbotic::databasepb::ListCollectionsResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->ListCollections(&context, request, &response); if (!status.ok()) { spdlog::error("Client::listCollections failed: {}", status.error_message()); return {}; } return {response.names().begin(), response.names().end()}; } std::optional getCollectionInfo(const std::string& name) { smartbotic::databasepb::GetCollectionInfoRequest request; request.set_name(name); smartbotic::databasepb::GetCollectionInfoResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->GetCollectionInfo(&context, request, &response); if (!status.ok()) { spdlog::error("Client::getCollectionInfo failed: {}", status.error_message()); return std::nullopt; } if (!response.found()) { return std::nullopt; } const auto& info = response.info(); Client::CollectionInfo result; result.name = info.name(); result.documentCount = info.document_count(); result.sizeBytes = info.size_bytes(); result.defaultTtlSeconds = info.options().default_ttl_seconds(); result.encrypted = info.options().encrypted(); result.maxVersions = info.options().max_versions(); result.createdAt = info.created_at(); result.updatedAt = info.updated_at(); return result; } // ===== Event Subscription ===== class SubscriptionHandle { public: SubscriptionHandle(std::shared_ptr ctx, std::unique_ptr> reader, std::thread readerThread) : context_(std::move(ctx)) , reader_(std::move(reader)) , readerThread_(std::move(readerThread)) , active_(true) {} ~SubscriptionHandle() { cancel(); } void cancel() { if (active_.exchange(false)) { context_->TryCancel(); if (readerThread_.joinable()) { readerThread_.join(); } } } private: std::shared_ptr context_; std::unique_ptr> reader_; std::thread readerThread_; std::atomic active_; }; std::shared_ptr subscribe(const std::vector& collections, Client::EventCallback callback) { auto context = std::make_shared(); smartbotic::databasepb::SubscribeRequest request; for (const auto& coll : collections) { request.add_collections(coll); } request.set_include_data(true); auto reader = stub_->Subscribe(context.get(), request); // Create reader thread auto readerThread = std::thread([reader = reader.get(), callback = std::move(callback)]() { smartbotic::databasepb::DatabaseEvent event; while (reader->Read(&event)) { std::optional data; if (!event.data().empty()) { data = nlohmann::json::parse(event.data()); } // Convert proto event type to string std::string eventType; switch (event.type()) { case smartbotic::databasepb::EVENT_INSERT: eventType = "insert"; break; case smartbotic::databasepb::EVENT_UPDATE: eventType = "update"; break; case smartbotic::databasepb::EVENT_DELETE: eventType = "delete"; break; case smartbotic::databasepb::EVENT_EXPIRE: eventType = "expire"; break; case smartbotic::databasepb::EVENT_INVALIDATE: eventType = "invalidate"; break; default: eventType = "unknown"; break; } callback( event.collection(), event.document_id(), eventType, data ); } }); return std::make_shared( std::move(context), std::move(reader), std::move(readerThread) ); } // ===== Version History ===== Client::VersionHistoryResult getVersionHistory(const std::string& collection, const std::string& id, uint32_t limit, uint32_t offset) { smartbotic::databasepb::GetVersionHistoryRequest request; request.set_collection(collection); request.set_id(id); if (limit > 0) request.set_limit(limit); if (offset > 0) request.set_offset(offset); smartbotic::databasepb::GetVersionHistoryResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->GetVersionHistory(&context, request, &response); if (!status.ok()) { spdlog::error("Client::getVersionHistory failed: {}", status.error_message()); return {}; } Client::VersionHistoryResult result; result.currentVersion = response.current_version(); result.totalCount = response.total_count(); result.documentDeleted = response.document_deleted(); result.versions.reserve(response.versions_size()); for (const auto& ver : response.versions()) { Client::VersionEntry entry; entry.version = ver.version(); entry.timestamp = ver.timestamp(); entry.updatedBy = ver.updated_by(); if (!ver.data().empty()) { entry.data = nlohmann::json::parse(ver.data()); } result.versions.push_back(std::move(entry)); } return result; } std::optional getDocumentVersion(const std::string& collection, const std::string& id, uint64_t version) { smartbotic::databasepb::GetDocumentVersionRequest request; request.set_collection(collection); request.set_id(id); request.set_version(version); smartbotic::databasepb::GetDocumentVersionResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->GetDocumentVersion(&context, request, &response); if (!status.ok()) { spdlog::error("Client::getDocumentVersion failed: {}", status.error_message()); return std::nullopt; } if (!response.found()) { return std::nullopt; } Client::VersionEntry entry; entry.version = response.version_entry().version(); entry.timestamp = response.version_entry().timestamp(); entry.updatedBy = response.version_entry().updated_by(); if (!response.version_entry().data().empty()) { entry.data = nlohmann::json::parse(response.version_entry().data()); } return entry; } uint64_t restoreVersion(const std::string& collection, const std::string& id, uint64_t version, const std::string& actor) { smartbotic::databasepb::RestoreVersionRequest request; request.set_collection(collection); request.set_id(id); request.set_version(version); if (!actor.empty()) request.set_actor(actor); smartbotic::databasepb::RestoreVersionResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->RestoreVersion(&context, request, &response); if (!status.ok()) { spdlog::error("Client::restoreVersion failed: {}", status.error_message()); return 0; } if (!response.success()) { spdlog::error("Client::restoreVersion: {}", response.error()); return 0; } return response.new_version(); } std::pair restoreToDate(const std::string& collection, const std::string& id, uint64_t timestamp, const std::string& actor) { smartbotic::databasepb::RestoreToDateRequest request; request.set_collection(collection); request.set_id(id); request.set_timestamp(timestamp); if (!actor.empty()) request.set_actor(actor); smartbotic::databasepb::RestoreToDateResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->RestoreToDate(&context, request, &response); if (!status.ok()) { spdlog::error("Client::restoreToDate failed: {}", status.error_message()); return {0, 0}; } if (!response.success()) { spdlog::error("Client::restoreToDate: {}", response.error()); return {0, 0}; } return {response.restored_version(), response.new_version()}; } // ===== Health ===== bool healthCheck() { smartbotic::databasepb::HealthCheckRequest request; smartbotic::databasepb::HealthCheckResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->HealthCheck(&context, request, &response); if (!status.ok()) { return false; } return response.healthy(); } std::optional getHealthInfo() { smartbotic::databasepb::HealthCheckRequest request; smartbotic::databasepb::HealthCheckResponse response; grpc::ClientContext context; setDeadline(context); auto status = stub_->HealthCheck(&context, request, &response); if (!status.ok()) { return std::nullopt; } Client::HealthInfo info; info.healthy = response.healthy(); info.uptimeMs = response.uptime_seconds() * 1000; info.documentCount = response.document_count(); info.memoryUsedBytes = response.memory_used_bytes(); info.walSizeBytes = response.wal_size_bytes(); return info; } private: void setDeadline(grpc::ClientContext& context) { context.set_deadline( std::chrono::system_clock::now() + std::chrono::milliseconds(config_.timeoutMs) ); } Config config_; std::shared_ptr channel_; std::unique_ptr stub_; std::atomic connected_{false}; }; // ===== Client Public Interface Implementation ===== Client::Client(Config config) : impl_(std::make_unique(std::move(config))) {} Client::~Client() = default; Client::Client(Client&&) noexcept = default; Client& Client::operator=(Client&&) noexcept = default; bool Client::connect() { return impl_->connect(); } bool Client::isConnected() const { return impl_->isConnected(); } std::string Client::insert(const std::string& collection, const nlohmann::json& data, const std::string& id, uint32_t ttlSeconds, const std::string& actor) { return impl_->insert(collection, data, id, ttlSeconds, actor); } std::optional Client::get(const std::string& collection, const std::string& id) { return impl_->get(collection, id); } bool Client::update(const std::string& collection, const std::string& id, const nlohmann::json& data, const std::string& actor) { return impl_->update(collection, id, data, actor); } bool Client::updateIfVersion(const std::string& collection, const std::string& id, const nlohmann::json& data, uint64_t expectedVersion, const std::string& actor) { return impl_->updateIfVersion(collection, id, data, expectedVersion, actor); } std::pair Client::upsert(const std::string& collection, const nlohmann::json& data, const std::string& id, uint32_t ttlSeconds, const std::string& actor) { return impl_->upsert(collection, data, id, ttlSeconds, actor); } bool Client::remove(const std::string& collection, const std::string& id) { return impl_->remove(collection, id); } bool Client::exists(const std::string& collection, const std::string& id) { return impl_->exists(collection, id); } Client::VersionHistoryResult Client::getVersionHistory(const std::string& collection, const std::string& id, uint32_t limit, uint32_t offset) { return impl_->getVersionHistory(collection, id, limit, offset); } std::optional Client::getDocumentVersion(const std::string& collection, const std::string& id, uint64_t version) { return impl_->getDocumentVersion(collection, id, version); } uint64_t Client::restoreVersion(const std::string& collection, const std::string& id, uint64_t version, const std::string& actor) { return impl_->restoreVersion(collection, id, version, actor); } std::pair Client::restoreToDate(const std::string& collection, const std::string& id, uint64_t timestamp, const std::string& actor) { return impl_->restoreToDate(collection, id, timestamp, actor); } std::vector Client::find(const std::string& collection, const QueryOptions& options) { return impl_->find(collection, options); } std::vector Client::find(const std::string& collection) { return impl_->find(collection, QueryOptions{}); } uint64_t Client::count(const std::string& collection, const std::vector>& filters) { return impl_->count(collection, filters); } uint64_t Client::count(const std::string& collection) { return impl_->count(collection, {}); } bool Client::setAdd(const std::string& collection, const std::string& setId, const std::string& member) { return impl_->setAdd(collection, setId, member); } bool Client::setRemove(const std::string& collection, const std::string& setId, const std::string& member) { return impl_->setRemove(collection, setId, member); } std::vector Client::setMembers(const std::string& collection, const std::string& setId) { return impl_->setMembers(collection, setId); } bool Client::setIsMember(const std::string& collection, const std::string& setId, const std::string& member) { return impl_->setIsMember(collection, setId, member); } bool Client::createCollection(const std::string& name, uint32_t defaultTtlSeconds, bool encrypted, uint32_t maxVersions) { return impl_->createCollection(name, defaultTtlSeconds, encrypted, maxVersions); } bool Client::dropCollection(const std::string& name) { return impl_->dropCollection(name); } std::vector Client::listCollections() { return impl_->listCollections(); } std::optional Client::getCollectionInfo(const std::string& name) { return impl_->getCollectionInfo(name); } std::shared_ptr Client::subscribe(const std::vector& collections, EventCallback callback) { return impl_->subscribe(collections, std::move(callback)); } bool Client::healthCheck() { return impl_->healthCheck(); } std::optional Client::getHealthInfo() { return impl_->getHealthInfo(); } } // namespace smartbotic::database