Added upstream batch API.

1 files changed, 103 insertions, 122 deletions

diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index 172e122e4..59cc74d53 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -12,6 +12,7 @@
 #include <zenhttp/httpserver.h>
 #include <zenstore/CAS.h>
 
+#include "cachekey.h"
 #include "monitoring/httpstats.h"
 #include "structuredcache.h"
 #include "structuredcachestore.h"
@@ -36,111 +37,12 @@ using namespace std::literals;
 
 //////////////////////////////////////////////////////////////////////////
 
-namespace detail { namespace cacheopt {
-	constexpr std::string_view Local	   = "local"sv;
-	constexpr std::string_view Remote	   = "remote"sv;
-	constexpr std::string_view Data		   = "data"sv;
-	constexpr std::string_view Meta		   = "meta"sv;
-	constexpr std::string_view Value	   = "value"sv;
-	constexpr std::string_view Attachments = "attachments"sv;
-}}	// namespace detail::cacheopt
-
-//////////////////////////////////////////////////////////////////////////
-
-enum class CachePolicy : uint8_t
-{
-	None			= 0,
-	QueryLocal		= 1 << 0,
-	QueryRemote		= 1 << 1,
-	Query			= QueryLocal | QueryRemote,
-	StoreLocal		= 1 << 2,
-	StoreRemote		= 1 << 3,
-	Store			= StoreLocal | StoreRemote,
-	SkipMeta		= 1 << 4,
-	SkipValue		= 1 << 5,
-	SkipAttachments = 1 << 6,
-	SkipData		= SkipMeta | SkipValue | SkipAttachments,
-	SkipLocalCopy	= 1 << 7,
-	Local			= QueryLocal | StoreLocal,
-	Remote			= QueryRemote | StoreRemote,
-	Default			= Query | Store,
-	Disable			= None,
-};
-
-gsl_DEFINE_ENUM_BITMASK_OPERATORS(CachePolicy);
-
 CachePolicy
 ParseCachePolicy(const HttpServerRequest::QueryParams& QueryParams)
 {
-	CachePolicy QueryPolicy = CachePolicy::Query;
-
-	{
-		std::string_view Opts = QueryParams.GetValue("query"sv);
-		if (!Opts.empty())
-		{
-			QueryPolicy = CachePolicy::None;
-			ForEachStrTok(Opts, ',', [&QueryPolicy](const std::string_view& Opt) {
-				if (Opt == detail::cacheopt::Local)
-				{
-					QueryPolicy |= CachePolicy::QueryLocal;
-				}
-				if (Opt == detail::cacheopt::Remote)
-				{
-					QueryPolicy |= CachePolicy::QueryRemote;
-				}
-				return true;
-			});
-		}
-	}
-
-	CachePolicy StorePolicy = CachePolicy::Store;
-
-	{
-		std::string_view Opts = QueryParams.GetValue("store"sv);
-		if (!Opts.empty())
-		{
-			StorePolicy = CachePolicy::None;
-			ForEachStrTok(Opts, ',', [&StorePolicy](const std::string_view& Opt) {
-				if (Opt == detail::cacheopt::Local)
-				{
-					StorePolicy |= CachePolicy::StoreLocal;
-				}
-				if (Opt == detail::cacheopt::Remote)
-				{
-					StorePolicy |= CachePolicy::StoreRemote;
-				}
-				return true;
-			});
-		}
-	}
-
-	CachePolicy SkipPolicy = CachePolicy::None;
-
-	{
-		std::string_view Opts = QueryParams.GetValue("skip"sv);
-		if (!Opts.empty())
-		{
-			ForEachStrTok(Opts, ',', [&SkipPolicy](const std::string_view& Opt) {
-				if (Opt == detail::cacheopt::Meta)
-				{
-					SkipPolicy |= CachePolicy::SkipMeta;
-				}
-				if (Opt == detail::cacheopt::Value)
-				{
-					SkipPolicy |= CachePolicy::SkipValue;
-				}
-				if (Opt == detail::cacheopt::Attachments)
-				{
-					SkipPolicy |= CachePolicy::SkipAttachments;
-				}
-				if (Opt == detail::cacheopt::Data)
-				{
-					SkipPolicy |= CachePolicy::SkipData;
-				}
-				return true;
-			});
-		}
-	}
+	const CachePolicy QueryPolicy = zen::ParseQueryCachePolicy(QueryParams.GetValue("query"sv));
+	const CachePolicy StorePolicy = zen::ParseStoreCachePolicy(QueryParams.GetValue("store"sv));
+	const CachePolicy SkipPolicy  = zen::ParseSkipCachePolicy(QueryParams.GetValue("skip"sv));
 
 	return QueryPolicy | StorePolicy | SkipPolicy;
 }
@@ -882,7 +784,11 @@ HttpStructuredCacheService::ValidateKeyUri(HttpServerRequest& Request, CacheRef&
 void
 HttpStructuredCacheService::HandleBatchRequest(zen::HttpServerRequest& Request, CachePolicy Policy)
 {
-	ZEN_UNUSED(Policy);
+	using namespace fmt::literals;
+
+	const bool SkipData		   = (Policy & CachePolicy::SkipData) == CachePolicy::SkipData;
+	const bool SkipAttachments = (Policy & CachePolicy::SkipAttachments) == CachePolicy::SkipAttachments;
+	const bool QueryUpstream   = m_UpstreamCache && ((Policy & CachePolicy::QueryRemote) == CachePolicy::QueryRemote);
 
 	switch (auto Verb = Request.RequestVerb())
 	{
@@ -898,44 +804,119 @@ HttpStructuredCacheService::HandleBatchRequest(zen::HttpServerRequest& Request,
 					return Request.WriteResponse(HttpResponseCode::BadRequest);
 				}
 
-				CbObject BatchRequest = zen::LoadCompactBinaryObject(Request.ReadPayload());
+				CbObject			   BatchRequest = zen::LoadCompactBinaryObject(Request.ReadPayload());
+				const std::string_view Method		= BatchRequest["method"sv].AsString();
 
-				CbPackage	   Package;
-				CbObjectWriter BatchResponse;
+				if (Method != "getcacherecords"sv)
+				{
+					return Request.WriteResponse(HttpResponseCode::BadRequest,
+												 HttpContentType::kText,
+												 "Invalid method '{}'"_format(Method));
+				}
 
-				BatchResponse.BeginArray("records"sv);
+				CbObjectView Params = BatchRequest["params"sv].AsObjectView();
 
-				for (CbFieldView QueryView : BatchRequest["records"sv])
+				std::vector<CacheKey> CacheKeys;
+				std::vector<IoBuffer> CacheValues;
+				std::vector<IoBuffer> Payloads;
+				std::vector<size_t>	  Missing;
+
+				for (CbFieldView QueryView : Params["cachekeys"sv])
 				{
-					CbObjectView		   Query	= QueryView.AsObjectView();
-					const std::string_view Bucket	= Query["bucket"sv].AsString();
-					const IoHash		   CacheKey = Query["key"sv].AsHash();
+					CbObjectView Query = QueryView.AsObjectView();
+					CacheKeys.push_back(CacheKey::Create(Query["bucket"sv].AsString(), Query["key"sv].AsHash()));
+				}
 
-					ZenCacheValue CacheValue;
-					const bool	  Hit = m_CacheStore.Get(Bucket, CacheKey, CacheValue);
+				CacheValues.resize(CacheKeys.size());
 
-					if (Hit)
+				for (size_t Idx = 0; const CacheKey& Key : CacheKeys)
+				{
+					ZenCacheValue CacheValue;
+					if (m_CacheStore.Get(Key.Bucket, Key.Hash, CacheValue))
 					{
 						CbObjectView CacheRecord(CacheValue.Value.Data());
 
-						CacheRecord.IterateAttachments([this, &Package](CbFieldView AttachmentHash) {
-							if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
+						if (!SkipAttachments)
+						{
+							CacheRecord.IterateAttachments([this, &Payloads](CbFieldView AttachmentHash) {
+								if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
+								{
+									Payloads.push_back(Chunk);
+								}
+							});
+						}
+
+						CacheValues[Idx] = CacheValue.Value;
+					}
+					else
+					{
+						Missing.push_back(Idx);
+					}
+
+					++Idx;
+				}
+
+				if (QueryUpstream)
+				{
+					auto UpstreamResult = m_UpstreamCache->GetCacheRecords(
+						CacheKeys,
+						Missing,
+						[this, &CacheValues, &Payloads, SkipAttachments](size_t KeyIndex, IoBuffer UpstreamValue) {
+							CbPackage UpstreamPackage;
+							if (UpstreamValue && UpstreamPackage.TryLoad(UpstreamValue))
 							{
-								Package.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Chunk))));
+								CbObjectView CacheRecord = UpstreamPackage.GetObject();
+
+								CacheRecord.IterateAttachments([&](CbFieldView AttachmentHash) {
+									if (const CbAttachment* Attachment = UpstreamPackage.FindAttachment(AttachmentHash.AsHash()))
+									{
+										if (CompressedBuffer Chunk = Attachment->AsCompressedBinary())
+										{
+											m_CidStore.AddChunk(Chunk);
+
+											if (!SkipAttachments)
+											{
+												Payloads.push_back(Chunk.GetCompressed().Flatten().AsIoBuffer());
+											}
+										}
+									}
+								});
+
+								CacheValues[KeyIndex] = IoBufferBuilder::MakeCloneFromMemory(CacheRecord.GetView());
 							}
 						});
+				}
+
+				CbObjectWriter BatchResponse;
 
-						BatchResponse << CacheRecord;
+				BatchResponse.BeginArray("result"sv);
+				for (const IoBuffer& Value : CacheValues)
+				{
+					if (Value)
+					{
+						BatchResponse << CbObjectView(Value.Data());
+					}
+					else
+					{
+						BatchResponse.AddNull();
 					}
 				}
-
 				BatchResponse.EndArray();
+
+				CbPackage Package;
 				Package.SetObject(BatchResponse.Save());
-				
+
+				for (const IoBuffer& Payload : Payloads)
+				{
+					Package.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Payload))));
+				}
+
 				BinaryWriter MemStream;
 				Package.Save(MemStream);
-				
-				Request.WriteResponse(HttpResponseCode::OK, HttpContentType::kCbPackage, IoBuffer(IoBuffer::Wrap, MemStream.GetData(), MemStream.GetSize()));
+
+				Request.WriteResponse(HttpResponseCode::OK,
+									  HttpContentType::kCbPackage,
+									  IoBuffer(IoBuffer::Wrap, MemStream.GetData(), MemStream.GetSize()));
 			}
 			break;
 		default: