feat(storage): adapt lib/storage to upstream smartbotic-db-client

Replace gRPC stubs over our internal proto::StorageService with calls into
smartbotic::database::Client (libsmartbotic-db-client). Public API of
StorageClient is unchanged; PIMPL hides the upstream client. Field projection
and multi-sort warnings emitted; TTL converted ms->s with round-up.

CMakeLists.txt

@@ -59,12 +59,12 @@ add_library(smartbotic_storage STATIC
 )
 target_include_directories(smartbotic_storage PUBLIC
     ${CMAKE_CURRENT_SOURCE_DIR}/lib
-    ${CMAKE_CURRENT_BINARY_DIR}
 )
 target_link_libraries(smartbotic_storage PUBLIC
     smartbotic_common
     smartbotic_logging
-    smartbotic_proto
+    nlohmann_json::nlohmann_json
+    smartbotic::db-client
 )
 
 # Crypto library

lib/storage/storage_client.cpp

@@ -1,76 +1,98 @@
 #include "storage/storage_client.hpp"
+
+#include <chrono>
+#include <smartbotic/database/client.hpp>
+
 #include "logging/logger.hpp"
 
 namespace smartbotic::storage {
 
-using namespace common;
+namespace dbc = smartbotic::database;
+using common::Error;
+using common::ErrorCode;
+using common::Result;
 
-StorageClient::StorageClient(const StorageClientConfig& config)
-    : config_(config) {
-    grpc::ChannelArguments args;
-    int max_size = config_.max_message_size_mb * 1024 * 1024;
-    args.SetMaxReceiveMessageSize(max_size);
-    args.SetMaxSendMessageSize(max_size);
-    channel_ = grpc::CreateCustomChannel(config_.address, grpc::InsecureChannelCredentials(), args);
-    stub_ = proto::StorageService::NewStub(channel_);
-    LOG_DEBUG("Storage client created for {} (max message size: {} MB)", config_.address, config_.max_message_size_mb);
-}
+namespace {
 
-Result<nlohmann::json> StorageClient::get(const std::string& collection, const std::string& id) {
-    proto::GetRequest request;
-    request.set_collection(collection);
-    request.set_id(id);
+uint32_t msToSec(int64_t ms) {
+    if (ms <= 0) return 0;
+    return static_cast<uint32_t>((ms + 999) / 1000);
+}
 
-    proto::Document response;
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
+dbc::Client::FilterOp translateOp(const std::string& op) {
+    if (op == "eq" || op.empty()) return dbc::Client::FilterOp::EQ;
+    if (op == "ne") return dbc::Client::FilterOp::NE;
+    if (op == "gt") return dbc::Client::FilterOp::GT;
+    if (op == "gte") return dbc::Client::FilterOp::GTE;
+    if (op == "lt") return dbc::Client::FilterOp::LT;
+    if (op == "lte") return dbc::Client::FilterOp::LTE;
+    if (op == "in") return dbc::Client::FilterOp::IN;
+    if (op == "contains") return dbc::Client::FilterOp::CONTAINS;
+    if (op == "exists") return dbc::Client::FilterOp::EXISTS;
+    if (op == "regex") return dbc::Client::FilterOp::REGEX;
+    if (op == "search") return dbc::Client::FilterOp::SEARCH;
+    LOG_WARN("Storage: unknown filter op '{}', defaulting to EQ", op);
+    return dbc::Client::FilterOp::EQ;
+}
 
-    auto status = stub_->Get(&context, request, &response);
+dbc::Client::Filter buildFilter(const std::string& field, const nlohmann::json& v) {
+    dbc::Client::Filter f;
+    f.field = field;
+    if (v.is_object() && v.contains("op")) {
+        f.op = translateOp(v.value("op", "eq"));
+        f.value = v.value("value", nlohmann::json{});
+    } else {
+        f.op = dbc::Client::FilterOp::EQ;
+        f.value = v;
+    }
+    return f;
+}
 
-    if (!status.ok()) {
-        if (status.error_code() == grpc::StatusCode::NOT_FOUND) {
-            return Error(ErrorCode::DocumentNotFound, status.error_message());
+} // namespace
+
+class StorageClient::Impl {
+public:
+    explicit Impl(const StorageClientConfig& cfg) {
+        dbc::Client::Config config;
+        config.address = cfg.address;
+        config.timeoutMs = static_cast<uint32_t>(cfg.timeout_ms);
+        client_ = std::make_unique<dbc::Client>(std::move(config));
+        if (!client_->connect()) {
+            LOG_WARN("StorageClient: connect() to {} returned false", cfg.address);
         }
-        return Error(ErrorCode::DatabaseError, status.error_message());
     }
 
-    nlohmann::json result = nlohmann::json::parse(response.data());
-    result["_id"] = response.id();
-    result["_version"] = response.version();
-    result["_createdAt"] = response.created_at();
-    result["_updatedAt"] = response.updated_at();
+    std::unique_ptr<dbc::Client> client_;
+};
+
+StorageClient::StorageClient(const StorageClientConfig& config)
+    : impl_(std::make_unique<Impl>(config)) {}
+
+StorageClient::~StorageClient() = default;
+StorageClient::StorageClient(StorageClient&&) noexcept = default;
+StorageClient& StorageClient::operator=(StorageClient&&) noexcept = default;
 
-    return result;
+bool StorageClient::isConnected() const {
+    return impl_ && impl_->client_ && impl_->client_->isConnected();
+}
+
+Result<nlohmann::json> StorageClient::get(const std::string& collection, const std::string& id) {
+    auto opt = impl_->client_->get(collection, id);
+    if (!opt) {
+        return Error(ErrorCode::DocumentNotFound, "Document not found: " + collection + "/" + id);
+    }
+    return *opt;
 }
 
 Result<std::string> StorageClient::insert(const std::string& collection,
                                           const nlohmann::json& data,
                                           const std::string& id,
                                           int64_t ttl_ms) {
-    proto::InsertRequest request;
-    request.set_collection(collection);
-    request.set_id(id);
-    request.set_data(data.dump());
-    request.set_ttl_ms(ttl_ms);
-
-    proto::MutationResponse response;
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
-
-    auto status = stub_->Insert(&context, request, &response);
-
-    if (!status.ok()) {
-        return Error(ErrorCode::DatabaseError, status.error_message());
+    auto new_id = impl_->client_->insert(collection, data, id, msToSec(ttl_ms));
+    if (new_id.empty()) {
+        return Error(ErrorCode::DatabaseError, "Insert failed: " + collection);
     }
-
-    if (!response.success()) {
-        return Error(static_cast<ErrorCode>(response.error().code()),
-                    response.error().message());
-    }
-
-    return response.id();
+    return new_id;
 }
 
 Result<int64_t> StorageClient::update(const std::string& collection,
@@ -78,270 +100,125 @@ Result<int64_t> StorageClient::update(const std::string& collection,
                                       const nlohmann::json& data,
                                       int64_t expected_version,
                                       bool partial) {
-    proto::UpdateRequest request;
-    request.set_collection(collection);
-    request.set_id(id);
-    request.set_data(data.dump());
-    request.set_expected_version(expected_version);
-    request.set_partial(partial);
-
-    proto::MutationResponse response;
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
-
-    auto status = stub_->Update(&context, request, &response);
-
-    if (!status.ok()) {
-        return Error(ErrorCode::DatabaseError, status.error_message());
-    }
-
-    if (!response.success()) {
-        return Error(static_cast<ErrorCode>(response.error().code()),
-                    response.error().message());
-    }
-
-    return response.version();
+    if (partial) {
+        uint64_t v = impl_->client_->patch(collection, id, data);
+        if (v == 0) return Error(ErrorCode::DatabaseError, "Patch failed: " + collection + "/" + id);
+        return static_cast<int64_t>(v);
+    }
+    if (expected_version != 0) {
+        bool ok = impl_->client_->updateIfVersion(
+            collection, id, data, static_cast<uint64_t>(expected_version));
+        if (!ok) return Error(ErrorCode::VersionConflict, "Version conflict: " + collection + "/" + id);
+        return expected_version + 1;
+    }
+    bool ok = impl_->client_->update(collection, id, data);
+    if (!ok) return Error(ErrorCode::DatabaseError, "Update failed: " + collection + "/" + id);
+    return 0;
 }
 
 Result<void> StorageClient::remove(const std::string& collection,
                                    const std::string& id,
                                    int64_t expected_version) {
-    proto::DeleteRequest request;
-    request.set_collection(collection);
-    request.set_id(id);
-    request.set_expected_version(expected_version);
-
-    proto::MutationResponse response;
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
-
-    auto status = stub_->Delete(&context, request, &response);
-
-    if (!status.ok()) {
-        return Error(ErrorCode::DatabaseError, status.error_message());
-    }
-
-    if (!response.success()) {
-        return Error(static_cast<ErrorCode>(response.error().code()),
-                    response.error().message());
-    }
-
-    return Result<void>();
+    (void)expected_version;  // upstream remove() is unconditional in 1.7.x
+    bool ok = impl_->client_->remove(collection, id);
+    if (!ok) return Error(ErrorCode::DocumentNotFound, "Remove failed: " + collection + "/" + id);
+    return {};
 }
 
 Result<QueryResult> StorageClient::query(const std::string& collection) {
-    return query(collection, QueryOptions{});
+    QueryOptions opts;
+    return query(collection, opts);
 }
 
-Result<QueryResult> StorageClient::query(const std::string& collection,
-                                         const QueryOptions& options) {
-    proto::QueryRequest request;
-    request.set_collection(collection);
-
-    // Add filters
-    for (const auto& [field, value] : options.filters) {
-        auto* filter = request.add_filters();
-        filter->set_field(field);
-        filter->set_op(proto::FILTER_OP_EQ);
-        filter->set_value(value.dump());
-    }
-
-    // Add sorts
-    for (const auto& [field, ascending] : options.sorts) {
-        auto* sort = request.add_sorts();
-        sort->set_field(field);
-        sort->set_direction(ascending ? proto::SORT_DIRECTION_ASC : proto::SORT_DIRECTION_DESC);
-    }
-
-    // Add projection
-    for (const auto& field : options.fields) {
-        request.add_fields(field);
+Result<QueryResult> StorageClient::query(const std::string& collection, const QueryOptions& opts) {
+    dbc::Client::QueryOptions up;
+    for (const auto& [field, val] : opts.filters) {
+        up.filters.push_back(buildFilter(field, val));
     }
-
-    // Pagination
-    auto* pagination = request.mutable_pagination();
-    pagination->set_page(options.page);
-    pagination->set_page_size(options.page_size);
-
-    proto::QueryResponse response;
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
-
-    auto status = stub_->Query(&context, request, &response);
-
-    if (!status.ok()) {
-        return Error(ErrorCode::DatabaseError, status.error_message());
+    if (!opts.sorts.empty()) {
+        if (opts.sorts.size() > 1) {
+            LOG_WARN("Storage: upstream client supports only one sort field; using first");
+        }
+        up.sortField = opts.sorts.front().first;
+        up.sortDescending = !opts.sorts.front().second;
     }
-
-    QueryResult result;
-    result.total_count = response.pagination().total_count();
-    result.has_more = response.pagination().has_more();
-
-    for (const auto& doc : response.documents()) {
-        nlohmann::json j = nlohmann::json::parse(doc.data());
-        j["_id"] = doc.id();
-        j["_version"] = doc.version();
-        j["_createdAt"] = doc.created_at();
-        j["_updatedAt"] = doc.updated_at();
-        result.documents.push_back(std::move(j));
+    if (!opts.fields.empty()) {
+        LOG_WARN("Storage: field projection not supported by upstream client; returning all fields");
     }
+    up.limit = static_cast<uint32_t>(opts.page_size);
+    up.offset = static_cast<uint32_t>(std::max(0, (opts.page - 1) * opts.page_size));
 
-    return result;
+    auto fr = impl_->client_->findWithMetrics(collection, up);
+    QueryResult out;
+    out.documents = std::move(fr.documents);
+    out.total_count = static_cast<int64_t>(fr.totalCount);
+    out.has_more = fr.hasMore;
+    return out;
 }
 
 Result<void> StorageClient::createCollection(const std::string& name,
                                              const nlohmann::json& schema,
                                              int64_t default_ttl_ms,
                                              const VersioningOptions* versioning) {
-    proto::CreateCollectionRequest request;
-    request.set_name(name);
-    if (!schema.is_null()) {
-        request.set_schema(schema.dump());
-    }
-    request.set_default_ttl_ms(default_ttl_ms);
-
-    // Set versioning configuration if provided
-    if (versioning) {
-        auto* ver_config = request.mutable_versioning();
-        ver_config->set_enabled(versioning->enabled);
-        ver_config->set_max_versions(versioning->max_versions);
-        ver_config->set_version_ttl_ms(versioning->version_ttl_ms);
-        ver_config->set_keep_on_delete(versioning->keep_on_delete);
-    }
-
-    proto::Empty response;
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
-
-    auto status = stub_->CreateCollection(&context, request, &response);
-
-    if (!status.ok()) {
-        if (status.error_code() == grpc::StatusCode::ALREADY_EXISTS) {
-            return Error(ErrorCode::AlreadyExists, status.error_message());
+    if (!schema.is_null() && !schema.empty()) {
+        LOG_WARN("Storage: createCollection 'schema' parameter is ignored by upstream client (no server-side schema validation in 1.7.x)");
+    }
+    uint32_t maxVersions = 0;
+    if (versioning && versioning->enabled) {
+        maxVersions = static_cast<uint32_t>(versioning->max_versions);
+        if (versioning->version_ttl_ms != 0 || versioning->keep_on_delete) {
+            LOG_WARN("Storage: createCollection version_ttl_ms / keep_on_delete are ignored — upstream 1.7.x createCollection only takes maxVersions");
         }
-        return Error(ErrorCode::DatabaseError, status.error_message());
     }
-
-    return Result<void>();
+    bool ok = impl_->client_->createCollection(
+        name,
+        msToSec(default_ttl_ms),
+        /*encrypted=*/false,
+        maxVersions,
+        /*vectorDimension=*/0);
+    if (!ok) return Error(ErrorCode::CollectionNotFound, "createCollection failed: " + name);
+    return {};
 }
 
 Result<void> StorageClient::dropCollection(const std::string& name) {
-    proto::DropCollectionRequest request;
-    request.set_name(name);
-
-    proto::Empty response;
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
-
-    auto status = stub_->DropCollection(&context, request, &response);
-
-    if (!status.ok()) {
-        if (status.error_code() == grpc::StatusCode::NOT_FOUND) {
-            return Error(ErrorCode::CollectionNotFound, status.error_message());
-        }
-        return Error(ErrorCode::DatabaseError, status.error_message());
-    }
-
-    return Result<void>();
+    bool ok = impl_->client_->dropCollection(name);
+    if (!ok) return Error(ErrorCode::CollectionNotFound, "dropCollection failed: " + name);
+    return {};
 }
 
 std::vector<std::string> StorageClient::listCollections() {
-    proto::ListCollectionsRequest request;
-    proto::ListCollectionsResponse response;
-
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
-
-    auto status = stub_->ListCollections(&context, request, &response);
-
-    if (!status.ok()) {
-        return {};
-    }
-
-    std::vector<std::string> result;
-    for (const auto& name : response.collections()) {
-        result.push_back(name);
-    }
-    return result;
-}
-
-bool StorageClient::isConnected() const {
-    return channel_->GetState(false) == GRPC_CHANNEL_READY;
+    return impl_->client_->listCollections();
 }
 
 Result<nlohmann::json> StorageClient::getVersion(const std::string& collection,
                                                  const std::string& id,
                                                  int64_t version) {
-    proto::GetVersionRequest request;
-    request.set_collection(collection);
-    request.set_id(id);
-    request.set_version(version);
-
-    proto::Document response;
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
-
-    auto status = stub_->GetVersion(&context, request, &response);
-
-    if (!status.ok()) {
-        if (status.error_code() == grpc::StatusCode::NOT_FOUND) {
-            return Error(ErrorCode::DocumentNotFound, status.error_message());
-        }
-        return Error(ErrorCode::DatabaseError, status.error_message());
-    }
-
-    nlohmann::json result = nlohmann::json::parse(response.data());
-    result["_id"] = response.id();
-    result["_version"] = response.version();
-    result["_createdAt"] = response.created_at();
-    result["_updatedAt"] = response.updated_at();
-
-    return result;
+    auto entry = impl_->client_->getDocumentVersion(collection, id, static_cast<uint64_t>(version));
+    if (!entry) return Error(ErrorCode::DocumentNotFound, "Version not found");
+    return entry->data;
 }
 
 Result<VersionListResult> StorageClient::listVersions(const std::string& collection,
                                                       const std::string& id,
                                                       int32_t limit,
                                                       int32_t offset) {
-    proto::ListVersionsRequest request;
-    request.set_collection(collection);
-    request.set_id(id);
-    request.set_limit(limit);
-    request.set_offset(offset);
-
-    proto::ListVersionsResponse response;
-    grpc::ClientContext context;
-    context.set_deadline(std::chrono::system_clock::now() +
-                        std::chrono::milliseconds(config_.timeout_ms));
-
-    auto status = stub_->ListVersions(&context, request, &response);
-
-    if (!status.ok()) {
-        return Error(ErrorCode::DatabaseError, status.error_message());
-    }
-
-    VersionListResult result;
-    result.total_count = response.total_count();
-    result.has_more = response.has_more();
-
-    for (const auto& ver : response.versions()) {
+    auto h = impl_->client_->getVersionHistory(collection, id,
+                                               static_cast<uint32_t>(limit),
+                                               static_cast<uint32_t>(offset));
+    VersionListResult out;
+    out.total_count = static_cast<int64_t>(h.totalCount);
+    out.has_more = (offset + static_cast<int32_t>(h.versions.size())) <
+                   static_cast<int32_t>(h.totalCount);
+    for (auto& v : h.versions) {
         DocumentVersionInfo info;
-        info.id = ver.id();
-        info.document_id = ver.document_id();
-        info.version = ver.version();
-        info.data = nlohmann::json::parse(ver.data());
-        info.created_at = ver.created_at();
-        result.versions.push_back(std::move(info));
-    }
-
-    return result;
+        info.id = id;
+        info.document_id = id;
+        info.version = static_cast<int64_t>(v.version);
+        info.data = std::move(v.data);
+        info.created_at = static_cast<int64_t>(v.timestamp);
+        out.versions.push_back(std::move(info));
+    }
+    return out;
 }
 
 } // namespace smartbotic::storage

lib/storage/storage_client.hpp

@@ -1,22 +1,26 @@
 #pragma once
 
+#include <memory>
 #include <string>
 #include <vector>
-#include <memory>
-#include <grpcpp/grpcpp.h>
-#include "proto/storage.grpc.pb.h"
+
+#include <nlohmann/json.hpp>
+
 #include "common/error.hpp"
 
 namespace smartbotic::storage {
 
 // Storage client configuration
 struct StorageClientConfig {
-    std::string address = "localhost:9001";
+    std::string address = "localhost:9010";  // upstream smartbotic-database default in our deployments
     int timeout_ms = 5000;
-    int max_message_size_mb = 64;  // Max gRPC message size in MB
+    int max_message_size_mb = 64;            // kept for back-compat; ignored by adapter
 };
 
-// Query options - defined outside class to avoid default initializer issues
+// Query options - field projection and multi-sort kept for source compatibility,
+// but the upstream client supports only a single sort field. The adapter will
+// log a warning and use only the first entry of `sorts`. Field projection is
+// silently ignored by upstream; callers must filter in-process if needed.
 struct QueryOptions {
     std::vector<std::pair<std::string, nlohmann::json>> filters;
     std::vector<std::pair<std::string, bool>> sorts;  // field, ascending
@@ -25,14 +29,12 @@ struct QueryOptions {
     int32_t page_size = 100;
 };
 
-// Query result
 struct QueryResult {
     std::vector<nlohmann::json> documents;
     int64_t total_count = 0;
     bool has_more = false;
 };
 
-// Versioning configuration for storage client
 struct VersioningOptions {
     bool enabled = false;
     int32_t max_versions = 0;
@@ -40,7 +42,6 @@ struct VersioningOptions {
     bool keep_on_delete = false;
 };
 
-// Document version info
 struct DocumentVersionInfo {
     std::string id;
     std::string document_id;
@@ -49,19 +50,24 @@ struct DocumentVersionInfo {
     int64_t created_at = 0;
 };
 
-// Version list result
 struct VersionListResult {
     std::vector<DocumentVersionInfo> versions;
     int64_t total_count = 0;
     bool has_more = false;
 };
 
-// RAII storage client for gRPC communication with database service
+// Adapter over smartbotic::database::Client. Public surface unchanged from the
+// pre-1.0 internal-DB version; PIMPL hides the upstream client.
 class StorageClient {
 public:
     explicit StorageClient(const StorageClientConfig& config);
+    ~StorageClient();
+
+    StorageClient(const StorageClient&) = delete;
+    StorageClient& operator=(const StorageClient&) = delete;
+    StorageClient(StorageClient&&) noexcept;
+    StorageClient& operator=(StorageClient&&) noexcept;
 
-    // Document operations
     common::Result<nlohmann::json> get(const std::string& collection, const std::string& id);
 
     common::Result<std::string> insert(const std::string& collection,
@@ -79,11 +85,9 @@ public:
                                 const std::string& id,
                                 int64_t expected_version = 0);
 
-    // Query operations - overloaded versions
     common::Result<QueryResult> query(const std::string& collection);
     common::Result<QueryResult> query(const std::string& collection, const QueryOptions& options);
 
-    // Collection operations
     common::Result<void> createCollection(const std::string& name,
                                           const nlohmann::json& schema = nlohmann::json{},
                                           int64_t default_ttl_ms = 0,
@@ -93,7 +97,6 @@ public:
 
     std::vector<std::string> listCollections();
 
-    // Version operations
     common::Result<nlohmann::json> getVersion(const std::string& collection,
                                               const std::string& id,
                                               int64_t version);
@@ -103,13 +106,11 @@ public:
                                                    int32_t limit = 100,
                                                    int32_t offset = 0);
 
-    // Connection status
     bool isConnected() const;
 
 private:
-    StorageClientConfig config_;
-    std::shared_ptr<grpc::Channel> channel_;
-    std::unique_ptr<proto::StorageService::Stub> stub_;
+    class Impl;
+    std::unique_ptr<Impl> impl_;
 };
 
 } // namespace smartbotic::storage