feat: server-side atomic PatchDocument RPC (v1.4.0)

Adds PatchDocument RPC that atomically merges fields into an existing
document on the server. The read-merge-write happens within the
collection lock, eliminating the lost-update race condition for
concurrent partial updates.

Two clients can now safely modify different fields of the same document
simultaneously — neither write is lost.

Client API: db.patch(collection, id, {field: value})
Server: MemoryStore::patchDocument() uses json::merge_patch() atomically
Proto: PatchDocument(PatchDocumentRequest) returns (PatchDocumentResponse)

VERSION

@@ -1 +1 @@
-1.3.0
+1.4.0

client/include/smartbotic/database/client.hpp

@@ -74,6 +74,16 @@ public:
                         const nlohmann::json& data, uint64_t expectedVersion,
                         const std::string& actor = "");
 
+    /**
+     * Atomically merge fields into an existing document (server-side).
+     * Only the specified fields are modified — other fields remain unchanged.
+     * Safe for concurrent use: two clients can patch different fields simultaneously.
+     * @param fields JSON object with fields to merge
+     * @return new version number, or 0 on failure
+     */
+    uint64_t patch(const std::string& collection, const std::string& id,
+                   const nlohmann::json& fields, const std::string& actor = "");
+
     /**
      * Upsert a document.
      * @param actor User ID performing the operation (for audit trail)

client/src/client.cpp

@@ -192,6 +192,34 @@ public:
         return response.success();
     }
 
+    uint64_t patch(const std::string& collection, const std::string& id,
+                   const nlohmann::json& fields, const std::string& actor) {
+        smartbotic::databasepb::PatchDocumentRequest request;
+        request.set_collection(collection);
+        request.set_id(id);
+        request.set_patch_json(fields.dump());
+        if (!actor.empty()) {
+            request.set_actor(actor);
+        }
+
+        smartbotic::databasepb::PatchDocumentResponse response;
+        grpc::ClientContext context;
+        setDeadline(context);
+
+        auto status = stub_->PatchDocument(&context, request, &response);
+        if (!status.ok()) {
+            spdlog::error("Client::patch failed: {}", status.error_message());
+            return 0;
+        }
+
+        if (!response.success()) {
+            spdlog::error("Client::patch failed: {}", response.error());
+            return 0;
+        }
+
+        return response.new_version();
+    }
+
     std::pair<std::string, bool> upsert(const std::string& collection, const nlohmann::json& data,
                                         const std::string& id, uint32_t ttlSeconds,
                                         const std::string& actor) {
@@ -1119,6 +1147,11 @@ bool Client::updateIfVersion(const std::string& collection, const std::string& i
     return impl_->updateIfVersion(collection, id, data, expectedVersion, actor);
 }
 
+uint64_t Client::patch(const std::string& collection, const std::string& id,
+                       const nlohmann::json& fields, const std::string& actor) {
+    return impl_->patch(collection, id, fields, actor);
+}
+
 std::pair<std::string, bool> Client::upsert(const std::string& collection, const nlohmann::json& data,
                                                     const std::string& id, uint32_t ttlSeconds,
                                                     const std::string& actor) {

proto/database.proto

@@ -27,6 +27,7 @@ service DatabaseService {
     rpc Update(UpdateRequest) returns (UpdateResponse);
     rpc Upsert(UpsertRequest) returns (UpsertResponse);
     rpc Delete(DeleteRequest) returns (DeleteResponse);
+    rpc PatchDocument(PatchDocumentRequest) returns (PatchDocumentResponse);
     rpc Exists(ExistsRequest) returns (ExistsResponse);
 
     // Version history operations
@@ -164,6 +165,19 @@ message DeleteResponse {
     bool deleted = 1;
 }
 
+message PatchDocumentRequest {
+    string collection = 1;
+    string id = 2;
+    bytes patch_json = 3;             // JSON fields to merge into existing document
+    string actor = 4;                 // User ID performing the operation (for audit)
+}
+
+message PatchDocumentResponse {
+    uint64 new_version = 1;
+    bool success = 2;
+    string error = 3;
+}
+
 message ExistsRequest {
     string collection = 1;
     string id = 2;

service/src/database_grpc_impl.cpp

@@ -166,6 +166,51 @@ grpc::Status DatabaseGrpcImpl::Update(
     }
 }
 
+grpc::Status DatabaseGrpcImpl::PatchDocument(
+    grpc::ServerContext* /*context*/,
+    const pb::PatchDocumentRequest* request,
+    pb::PatchDocumentResponse* response
+) {
+    try {
+        nlohmann::json patch = nlohmann::json::parse(
+            request->patch_json().begin(), request->patch_json().end()
+        );
+
+        uint64_t newVersion = store_.patchDocument(
+            request->collection(),
+            request->id(),
+            patch,
+            request->actor()
+        );
+
+        if (newVersion == 0) {
+            response->set_success(false);
+            response->set_error("Document not found");
+            return grpc::Status::OK;
+        }
+
+        // Persist vector changes (store_.patchDocument handles WAL for the document itself
+        // via emitPersist, but vector WAL entries need explicit logging)
+        auto* vectors = store_.getCollectionVectors(request->collection());
+        if (vectors) {
+            auto vit = vectors->find(request->id());
+            if (vit != vectors->end()) {
+                persistence_.logVecPut(request->collection(), request->id(), vit->second);
+            }
+        }
+
+        response->set_new_version(newVersion);
+        response->set_success(true);
+        return grpc::Status::OK;
+
+    } catch (const nlohmann::json::exception& e) {
+        return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
+                           "Invalid JSON: " + std::string(e.what()));
+    } catch (const std::exception& e) {
+        return grpc::Status(grpc::StatusCode::INTERNAL, e.what());
+    }
+}
+
 grpc::Status DatabaseGrpcImpl::Upsert(
     grpc::ServerContext* /*context*/,
     const pb::UpsertRequest* request,

service/src/database_grpc_impl.hpp

@@ -66,6 +66,12 @@ public:
         pb::DeleteResponse* response
     ) override;
 
+    grpc::Status PatchDocument(
+        grpc::ServerContext* context,
+        const pb::PatchDocumentRequest* request,
+        pb::PatchDocumentResponse* response
+    ) override;
+
     grpc::Status Exists(
         grpc::ServerContext* context,
         const pb::ExistsRequest* request,

service/src/memory_store.cpp

@@ -657,6 +657,81 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
     return true;
 }
 
+uint64_t MemoryStore::patchDocument(const std::string& collection, const std::string& id,
+                                     const nlohmann::json& patch, const std::string& actor) {
+    CollectionData* coll = getOrCreateCollection(collection);
+
+    std::unique_lock<std::shared_mutex> lock(coll->mutex);
+
+    auto it = coll->documents.find(id);
+    if (it == coll->documents.end()) {
+        return 0;
+    }
+
+    // Track memory change (old size)
+    uint64_t oldSize = estimateDocumentSize(it->second);
+
+    // Remove from old expiration index
+    if (it->second.expiresAt > 0) {
+        removeFromExpirationIndex(*coll, id, it->second.expiresAt);
+    }
+
+    // Save current state to history
+    saveToHistory(*coll, it->second);
+
+    // Merge patch into existing data (RFC 7396 merge patch)
+    it->second.data.merge_patch(patch);
+
+    // Handle vector field if present in patch
+    auto vec = extractVector(*coll, it->second.data);
+    if (!vec.empty()) {
+        storeVector(*coll, id, vec);
+    }
+
+    // Update metadata
+    it->second.version++;
+    it->second.updatedAt = currentTimeMs();
+    it->second.nodeId = config_.nodeId;
+    if (!actor.empty()) {
+        it->second.updatedBy = actor;
+    }
+
+    // Add to new expiration index
+    if (it->second.expiresAt > 0) {
+        addToExpirationIndex(*coll, id, it->second.expiresAt);
+    }
+
+    coll->updatedAt = it->second.updatedAt;
+    uint64_t newVersion = it->second.version;
+
+    // Track memory change (new size)
+    uint64_t newSize = estimateDocumentSize(it->second);
+
+    // Capture the updated document for callbacks after releasing the lock
+    Document updatedDoc = it->second;
+
+    lock.unlock();
+
+    // Update stats
+    {
+        std::lock_guard<std::mutex> statsLock(statsMutex_);
+        stats_.updateCount++;
+    }
+
+    // Update memory tracking atomically (subtract old, add new)
+    if (newSize > oldSize) {
+        estimatedMemoryBytes_.fetch_add(newSize - oldSize, std::memory_order_relaxed);
+    } else if (oldSize > newSize) {
+        estimatedMemoryBytes_.fetch_sub(oldSize - newSize, std::memory_order_relaxed);
+    }
+
+    // Emit callbacks (WAL persistence + event notification)
+    emitPersist(collection, id, updatedDoc, EventType::UPDATE);
+    emitEvent(EventType::UPDATE, collection, id, updatedDoc.data);
+
+    return newVersion;
+}
+
 // ===== Query Operations =====
 
 QueryResult MemoryStore::find(const std::string& collection, const Query& query) const {

service/src/memory_store.hpp

@@ -178,6 +178,16 @@ public:
     bool updateIfVersion(const std::string& collection, const std::string& id,
                          const Document& doc, uint64_t expectedVersion);
 
+    /**
+     * Atomically merge fields into an existing document.
+     * Reads the document, merges patch fields, increments version, writes back.
+     * All within the collection lock — no race condition possible.
+     * @param patch JSON object whose fields are merged into the existing document
+     * @return new version number, or 0 if document not found
+     */
+    uint64_t patchDocument(const std::string& collection, const std::string& id,
+                           const nlohmann::json& patch, const std::string& actor = "");
+
     // ===== Query Operations =====
 
     /**