oplog upload/download (#214)

- Feature: Zen server endpoint `prj/{project}/oplog/{log}/chunks` to post multiple attachments in one request.
- Feature: Zen server endpoint `prj/{project}/oplog/{log}/save` to save an oplog container. Accepts `CbObject` containing a compressed oplog and attachment references organized in blocks. 
- Feature: Zen server endpoint `prj/{project}/oplog/{log}/load` to request an oplog container. Responds with an `CbObject` containing a compressed oplog and attachment references organized in blocks.
- Feature: Zen server endpoint `{project}/oplog/{log}/rpc` to initiate an import to or export from an external location and other operations. Use either JSon or CbPackage as payload.
  - CbObject/JSon RPC format for `import` and `export` methods:
  - CbObject RPC format for `getchunks` method, returns CbPackage with the found chunks, if all chunks are found the number of attachments matches number of chunks requested.
- Feature: Zen server `{project}/oplog/{log}/{hash}` now accepts `HttpVerb::kPost` as well as `HttpVerb::kGet`.
- Feature: Zen command line tool `oplog-export` to export an oplog to an external target using the zenserver oplog export endpoint.
- Feature: Zen command line tool `oplog-import` to import an oplog from an external source using the zenserver oplog import endpoint.

1 files changed, 42 insertions, 0 deletions

diff --git a/zenserver/upstream/jupiter.cpp b/zenserver/upstream/jupiter.cpp
index 260b83355..dbb185bec 100644
--- a/zenserver/upstream/jupiter.cpp
+++ b/zenserver/upstream/jupiter.cpp
@@ -6,6 +6,7 @@
 #include "diag/logging.h"
 
 #include <zencore/compactbinary.h>
+#include <zencore/compositebuffer.h>
 #include <zencore/iobuffer.h>
 #include <zencore/iohash.h>
 #include <zencore/string.h>
@@ -437,6 +438,47 @@ CloudCacheSession::PutCompressedBlob(std::string_view Namespace, const IoHash& K
 }
 
 CloudCacheResult
+CloudCacheSession::PutCompressedBlob(std::string_view Namespace, const IoHash& Key, const CompositeBuffer& Payload)
+{
+	ZEN_TRACE_CPU("HordeClient::PutCompressedBlob");
+
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << Namespace << "/" << Key.ToHexString();
+
+	cpr::Session&				 Session	 = GetSession();
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
+
+	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-comp"}});
+	uint64_t				  SizeLeft	   = Payload.GetSize();
+	CompositeBuffer::Iterator BufferIt	   = Payload.GetIterator(0);
+	auto					  ReadCallback = [&Payload, &BufferIt, &SizeLeft](char* buffer, size_t& size, intptr_t) {
+		 size = Min<size_t>(size, SizeLeft);
+		 MutableMemoryView Data(buffer, size);
+		 Payload.CopyTo(Data, BufferIt);
+		 SizeLeft -= size;
+		 return true;
+	};
+	Session.SetReadCallback(cpr::ReadCallback(gsl::narrow<cpr::cpr_off_t>(SizeLeft), ReadCallback));
+
+	cpr::Response Response = Session.Put();
+	ZEN_DEBUG("PUT {}", Response);
+
+	if (Response.error)
+	{
+		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
+	}
+	else if (!VerifyAccessToken(Response.status_code))
+	{
+		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
+	}
+
+	return {.Bytes			= Response.uploaded_bytes,
+			.ElapsedSeconds = Response.elapsed,
+			.Success		= (Response.status_code == 200 || Response.status_code == 201)};
+}
+
+CloudCacheResult
 CloudCacheSession::PutObject(std::string_view Namespace, const IoHash& Key, IoBuffer Object)
 {
 	ZEN_TRACE_CPU("HordeClient::PutObject");