Implemented new URI addressing scheme for the Zen cache endpoints, and prepared for additional indexing capabilities

4 files changed, 147 insertions, 33 deletions

diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index 2704bedc3..728cfaded 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -2,6 +2,8 @@
 
 #pragma once
 
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/compactbinaryvalidation.h>
 #include <zencore/fmtutils.h>
 #include <zencore/httpserver.h>
 #include <zencore/timer.h>
@@ -53,6 +55,21 @@ HttpStructuredCacheService::HandleRequest(zen::HttpServerRequest& Request)
 		return Request.WriteResponse(zen::HttpResponse::BadRequest);  // invalid URL
 	}
 
+	if (Ref.PayloadId == IoHash::Zero)
+	{
+		return HandleCacheRecordRequest(Request, Ref);
+	}
+	else
+	{
+		return HandleCachePayloadRequest(Request, Ref);
+	}
+
+	return;
+}
+
+void
+HttpStructuredCacheService::HandleCacheRecordRequest(zen::HttpServerRequest& Request, CacheRef& Ref)
+{
 	switch (auto Verb = Request.RequestVerb())
 	{
 		using enum zen::HttpVerb;
@@ -65,20 +82,15 @@ HttpStructuredCacheService::HandleRequest(zen::HttpServerRequest& Request)
 
 				if (!Success)
 				{
-					Request.WriteResponse(zen::HttpResponse::NotFound);
+					return Request.WriteResponse(zen::HttpResponse::NotFound);
 				}
-				else
+
+				if (Verb == kHead)
 				{
-					if (Verb == kHead)
-					{
-						Request.SetSuppressResponseBody();
-						Request.WriteResponse(zen::HttpResponse::OK, zen::HttpContentType::kBinary, Value.Value);
-					}
-					else
-					{
-						Request.WriteResponse(zen::HttpResponse::OK, zen::HttpContentType::kBinary, Value.Value);
-					}
+					Request.SetSuppressResponseBody();
 				}
+
+				return Request.WriteResponse(zen::HttpResponse::OK, zen::HttpContentType::kBinary, Value.Value);
 			}
 			break;
 
@@ -91,21 +103,55 @@ HttpStructuredCacheService::HandleRequest(zen::HttpServerRequest& Request)
 
 					HttpContentType ContentType = Request.RequestContentType();
 
+					bool IsCompactBinary;
+
 					switch (ContentType)
 					{
 						case HttpContentType::kUnknownContentType:
 						case HttpContentType::kBinary:
-							Value.IsCompactBinary = false;
+							IsCompactBinary = false;
 							break;
 
 						case HttpContentType::kCbObject:
-							Value.IsCompactBinary = true;
+							IsCompactBinary = true;
 							break;
 
 						default:
 							return Request.WriteResponse(zen::HttpResponse::BadRequest);
 					}
 
+					// Compute index data
+
+					if (IsCompactBinary)
+					{
+						// Validate payload before accessing it
+						zen::CbValidateError ValidationResult =
+							zen::ValidateCompactBinary(MemoryView(Body.Data(), Body.Size()), zen::CbValidateMode::All);
+
+						if (ValidationResult != CbValidateError::None)
+						{
+							// TODO: add details in response
+							return Request.WriteResponse(HttpResponse::BadRequest);
+						}
+
+						// Extract data for index
+						zen::CbObjectView Cbo(Body.Data());
+
+						int ReferenceCount = 0;
+
+						zen::CbObjectWriter Idx;
+						Idx.BeginArray();
+
+						Cbo.IterateAttachments([&](CbFieldView AttachmentView) {
+							Idx.AddHash(AttachmentView.AsHash());
+							++ReferenceCount;
+						});
+
+						Idx.EndArray();
+
+						// TODO: store references in index
+					}
+
 					m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, Value);
 
 					// This is currently synchronous for simplicity and debuggability but should be
@@ -152,7 +198,76 @@ HttpStructuredCacheService::HandleRequest(zen::HttpServerRequest& Request)
 	}
 }
 
-[[nodiscard]] bool
+void
+HttpStructuredCacheService::HandleCachePayloadRequest(zen::HttpServerRequest& Request, CacheRef& Ref)
+{
+	// Note: the URL references the uncompressed payload hash - so this maintains the mapping
+	// from uncompressed CAS identity to the stored payload hash
+	//
+	// this is a PITA but a consequence of the fact that the client side code is not able to
+	// address data by compressed hash
+
+	switch (auto Verb = Request.RequestVerb())
+	{
+		using enum zen::HttpVerb;
+
+		case kHead:
+		case kGet:
+			{
+				// TODO: need to map from uncompressed content address into the storage 
+				// (compressed) content address
+
+				zen::IoBuffer Payload = m_CasStore.FindChunk(Ref.PayloadId);
+
+				if (!Payload)
+				{
+					return Request.WriteResponse(zen::HttpResponse::NotFound);
+				}
+
+				if (Verb == kHead)
+				{
+					Request.SetSuppressResponseBody();
+				}
+
+				return Request.WriteResponse(zen::HttpResponse::OK, zen::HttpContentType::kBinary, Payload);
+			}
+			break;
+
+		case kPut:
+			{
+				if (zen::IoBuffer Body = Request.ReadPayload())
+				{
+					zen::IoHash ChunkHash = zen::IoHash::HashMemory(Body);
+
+					if (ChunkHash != Ref.PayloadId)
+					{
+						// the URL and data hashes don't match!
+						return Request.WriteResponse(HttpResponse::BadRequest);
+					}
+
+					zen::CasStore::InsertResult Result = m_CasStore.InsertChunk(Body, ChunkHash);
+
+					if (Result.New)
+					{
+						return Request.WriteResponse(zen::HttpResponse::Created);
+					}
+					else
+					{
+						return Request.WriteResponse(zen::HttpResponse::OK);
+					}
+				}
+			}
+			break;
+
+		case kPost:
+			break;
+
+		default:
+			break;
+	}
+}
+
+bool
 HttpStructuredCacheService::ValidateUri(zen::HttpServerRequest& Request, CacheRef& OutRef)
 {
 	std::string_view			Key				  = Request.RelativeUri();
@@ -184,20 +299,24 @@ HttpStructuredCacheService::ValidateUri(zen::HttpServerRequest& Request, CacheRe
 		PayloadSegment = Key.substr(PayloadSplitOffset + 1);
 	}
 
-	if (HashSegment.size() != (2 * sizeof OutRef.HashKey.Hash))
+	if (HashSegment.size() != zen::IoHash::StringLength)
 	{
 		return false;
 	}
 
-	if (!PayloadSegment.empty() && PayloadSegment.size() != 24)
+	if (!PayloadSegment.empty() && PayloadSegment.size() == zen::IoHash::StringLength)
 	{
-		OutRef.PayloadId = zen::Oid::FromHexString(PayloadSegment);
+		const bool IsOk = zen::ParseHexBytes(PayloadSegment.data(), PayloadSegment.size(), OutRef.PayloadId.Hash);
 
-		if (!OutRef.PayloadId)
+		if (!IsOk)
 		{
 			return false;
 		}
 	}
+	else
+	{
+		OutRef.PayloadId = zen::IoHash::Zero;
+	}
 
 	const bool IsOk = zen::ParseHexBytes(HashSegment.data(), HashSegment.size(), OutRef.HashKey.Hash);
 
diff --git a/zenserver/cache/structuredcache.h b/zenserver/cache/structuredcache.h
index 111a4af50..761bf3399 100644
--- a/zenserver/cache/structuredcache.h
+++ b/zenserver/cache/structuredcache.h
@@ -39,10 +39,12 @@ private:
 	{
 		std::string BucketSegment;
 		IoHash		HashKey;
-		Oid			PayloadId;
+		IoHash		PayloadId;
 	};
 
 	[[nodiscard]] bool ValidateUri(zen::HttpServerRequest& Request, CacheRef& OutRef);
+	void			   HandleCacheRecordRequest(zen::HttpServerRequest& Request, CacheRef& Ref);
+	void			   HandleCachePayloadRequest(zen::HttpServerRequest& Request, CacheRef& Ref);
 
 	zen::CasStore&			 m_CasStore;
 	ZenCacheStore			 m_CacheStore;
diff --git a/zenserver/cache/structuredcachestore.cpp b/zenserver/cache/structuredcachestore.cpp
index 3fd9f5ec4..59588af48 100644
--- a/zenserver/cache/structuredcachestore.cpp
+++ b/zenserver/cache/structuredcachestore.cpp
@@ -59,23 +59,14 @@ ZenCacheStore::Get(std::string_view InBucket, const zen::IoHash& HashKey, ZenCac
 void
 ZenCacheStore::Put(std::string_view InBucket, const zen::IoHash& HashKey, const ZenCacheValue& Value)
 {
+	// Store value and index
+
 	if (Value.Value.Size() <= m_DiskLayerSizeThreshold)
 	{
 		m_MemLayer.Put(InBucket, HashKey, Value);
 	}
 
 	m_DiskLayer.Put(InBucket, HashKey, Value);
-
-	if (Value.IsCompactBinary)
-	{
-		zen::CbObject Cbo(SharedBuffer(Value.Value));
-
-		std::vector<IoHash> ReferencedChunks;
-
-		Cbo.IterateAttachments([&](CbFieldView AttachmentView) { ReferencedChunks.push_back(AttachmentView.AsHash()); });
-
-		// TODO: store references in index
-	}
 }
 
 //////////////////////////////////////////////////////////////////////////
@@ -176,6 +167,7 @@ struct DiskLocation
 {
 	uint64_t Offset;
 	uint32_t Size;
+	uint32_t IndexDataSize;
 };
 
 struct DiskIndexEntry
@@ -186,7 +178,7 @@ struct DiskIndexEntry
 
 #pragma pack(pop)
 
-static_assert(sizeof(DiskIndexEntry) == 32);
+static_assert(sizeof(DiskIndexEntry) == 36);
 
 struct ZenCacheDiskLayer::CacheBucket
 {
diff --git a/zenserver/cache/structuredcachestore.h b/zenserver/cache/structuredcachestore.h
index 364644cf7..47302dc10 100644
--- a/zenserver/cache/structuredcachestore.h
+++ b/zenserver/cache/structuredcachestore.h
@@ -2,7 +2,8 @@
 
 #pragma once
 
-#include <zencore/IoBuffer.h>
+#include <zencore/compactbinary.h>
+#include <zencore/iobuffer.h>
 #include <zencore/iohash.h>
 #include <zencore/thread.h>
 #include <zencore/uid.h>
@@ -44,7 +45,7 @@ class CasStore;
 struct ZenCacheValue
 {
 	zen::IoBuffer Value;
-	bool		  IsCompactBinary = false;
+	zen::CbObject IndexData;
 };
 
 class ZenCacheMemoryLayer