Bulk Write

Bulk insert/delete documents. Corresponding version history documents for inserted and/or deleted documents are created unless skipVersion is specified.

Note, the individual actions in the request are executed sequentially by the service. Individual requests in the batch may fail. This does not lead to service breaking out of the processing loop.

Data models that represents the payload to be submitted to the service for creating/deleting documents.

//
// Created by Rakesh on 15/12/2024.
//

#pragma once

#include "action.hpp"

#if defined __has_include
  #if __has_include("../../../common/util/serialise.hpp")
    #include "../../../common/util/serialise.hpp"
    #include "../../../common/util/json.hpp"
    #include "../../../common/visit_struct/visit_struct_intrusive.hpp"
  #else
    #include <mongo-service/common/util/serialise.hpp>
    #include <mongo-service/common/util/json.hpp>
    #include <mongo-service/common/visit_struct/visit_struct_intrusive.hpp>
  #endif
#endif

namespace spt::mongoservice::api::model::request
{
  template <util::Visitable Document, util::Visitable Metadata, util::Visitable Delete>
  requires std::is_same_v<decltype(Document::id), bsoncxx::oid> && std::constructible_from<Delete, bsoncxx::document::view>
  struct Bulk
  {
    struct Payload
    {
      Payload() = default;
      ~Payload() = default;
      Payload(Payload&&) = default;
      Payload& operator=(Payload&&) = default;

      Payload(const Payload&) = delete;
      Payload& operator=(const Payload&) = delete;

      BEGIN_VISITABLES(Payload);
      VISITABLE(std::vector<Document>, insert);
      VISITABLE(std::vector<Delete>, remove);
      END_VISITABLES;
    };

    Bulk() = default;
    ~Bulk() = default;
    Bulk(Bulk&&) = default;
    Bulk& operator=(Bulk&&) = default;

    Bulk(const Bulk&) = delete;
    Bulk& operator=(const Bulk&) = delete;

    BEGIN_VISITABLES(Bulk);
    VISITABLE(Payload, document);
    VISITABLE(std::optional<Metadata>, metadata);
    std::string database;
    std::string collection;
    std::string application;
    std::string correlationId;
    Action action{Action::bulk};
    bool skipVersion{false};
    bool skipMetric{false};
    END_VISITABLES;
  };
}

Data models that represents the payloads the service responds with when creating/deleting document(s).

//
// Created by Rakesh on 18/12/2024.
//

#pragma once

#if defined __has_include
  #if __has_include("../../../common/visit_struct/visit_struct_intrusive.hpp")
    #include "../../../common/visit_struct/visit_struct_intrusive.hpp"
    #include "../../../common/util/serialise.hpp"
  #else
    #include <mongo-service/common/visit_struct/visit_struct_intrusive.hpp>
    #include <mongo-service/common/util/serialise.hpp>
  #endif
#endif

#include <cstdint>

namespace spt::mongoservice::api::model::response
{
  struct Bulk
  {
    explicit Bulk( bsoncxx::document::view document ) { util::unmarshall( *this, document ); }
    Bulk() = default;
    ~Bulk() = default;
    Bulk(Bulk&&) = default;
    Bulk& operator=(Bulk&&) = default;

    Bulk(const Bulk&) = delete;
    Bulk& operator=(const Bulk&) = delete;

    BEGIN_VISITABLES(Bulk);
    VISITABLE(int32_t, create);
    VISITABLE(int32_t, remove);
    VISITABLE(int32_t, history);
    END_VISITABLES;
  };
}

Sample code illustrating the bulk action.

#include <mongo-service/api/repository/repository.hpp>
#include <log/NanoLog.hpp>

namespace example
{
  struct Document
  {
    explicit Document( bsoncxx::document::view bson ) { spt::util::unmarshall( *this, bson ); }
    Document() = default;
    ~Document() = default;
    Document(Document&&) = default;
    Document& operator=(Document&&) = default;
    bool operator==(const Document&) const = default;

    Document(const Document&) = delete;
    Document& operator=(const Document&) = delete;

    BEGIN_VISITABLES(Document);
    VISITABLE(bsoncxx::oid, id);
    VISITABLE(std::string, str);
    VISITABLE_DIRECT_INIT(spt::util::DateTimeMs, created, {std::chrono::duration_cast<std::chrono::milliseconds>( std::chrono::system_clock::now().time_since_epoch() )});
    VISITABLE_DIRECT_INIT(int64_t, integer, {5});
    VISITABLE_DIRECT_INIT(double, floating, {10.345});
    VISITABLE_DIRECT_INIT(bool, boolean, {true});
    END_VISITABLES;
  };

  struct Metadata
  {
    explicit Metadata( bsoncxx::document::view bson ) { spt::util::unmarshall( *this, bson ); }
    Metadata() = default;
    ~Metadata() = default;
    Metadata(Metadata&&) = default;
    Metadata& operator=(Metadata&&) = default;
    bool operator==(const Metadata&) const = default;

    Metadata(const Metadata&) = delete;
    Metadata& operator=(const Metadata&) = delete;

    BEGIN_VISITABLES(Metadata);
    VISITABLE(std::string, project);
    VISITABLE(std::string, product);
    END_VISITABLES;
  };
}

int main()
{
  using namespace spt::mongoservice::api;
  auto bulk = model::request::Bulk<example::Document, example::Metadata, model::request::IdFilter>{};
  for ( auto i = 0; i < 10; ++i )
  {
    bulk.document.insert.emplace_back();
    bulk.document.remove.emplace_back();
    bulk.document.remove.back().id = bulk.document.insert.back().id;
  }
  bulk.database = "test";
  bulk.collection = "test";

  auto result = repository::bulk( bulk );
  if ( !result.has_value() )
  {
    LOG_WARN << "Error executing bulk statements. " << magic_enum::enum_name( result.error().cause ) << ". " << result.error().message;
  }
}

Last modified: 18 February 2025