Implemented generic CbPackage attachments filtering

Package transmission will also need to be updated (up next) for the new scheme to be effective

6 files changed, 539 insertions, 284 deletions

diff --git a/zenhttp/httpclient.cpp b/zenhttp/httpclient.cpp
index 9d692271d..fb31e0a8b 100644
--- a/zenhttp/httpclient.cpp
+++ b/zenhttp/httpclient.cpp
@@ -2,12 +2,102 @@
 
 #include <zenhttp/httpclient.h>
 
-#include <spdlog/spdlog.h>
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/compactbinarypackage.h>
+#include <zencore/logging.h>
+#include <zencore/stream.h>
 
 #include <doctest/doctest.h>
 
 namespace zen {
 
+HttpClient::HttpClient(std::string_view BaseUri) : m_BaseUri(BaseUri)
+{
+}
+
+HttpClient::~HttpClient()
+{
+}
+
+void
+HttpClient::TransactPackage(std::string_view Url, CbPackage Package)
+{
+	cpr::Session Sess;
+	Sess.SetUrl(m_BaseUri + std::string(Url));
+
+	// First, list of offered chunks for filtering on the server end
+
+	std::vector<IoHash>			  AttachmentsToSend;
+	std::span<const CbAttachment> Attachments = Package.GetAttachments();
+
+	if (Attachments.empty() == false)
+	{
+		CbObjectWriter Writer;
+		Writer.BeginArray("offer");
+
+		for (const CbAttachment& Attachment : Attachments)
+		{
+			IoHash Hash = Attachment.GetHash();
+
+			Writer.AddHash(Hash);
+		}
+
+		Writer.EndArray();
+
+		MemoryOutStream MemOut;
+		BinaryWriter	MemWriter(MemOut);
+		Writer.Save(MemWriter);
+
+		Sess.SetHeader({{"Content-Type", "application/x-ue-offer"}, {"UE-Session", "123456789012345678901234"}, {"UE-Request", "1"}});
+		Sess.SetBody(cpr::Body{(const char*)MemOut.Data(), MemOut.Size()});
+
+		cpr::Response FilterResponse = Sess.Post();
+
+		if (FilterResponse.status_code == 200)
+		{
+			IoBuffer ResponseBuffer(IoBuffer::Wrap, FilterResponse.text.data(), FilterResponse.text.size());
+			CbObject ResponseObject = LoadCompactBinaryObject(ResponseBuffer);
+
+			for (auto& Entry : ResponseObject["need"])
+			{
+				ZEN_ASSERT(Entry.IsHash());
+				AttachmentsToSend.push_back(Entry.AsHash());
+			}
+		}
+	}
+
+	CbPackage SendPackage;
+	SendPackage.SetObject(Package.GetObject(), Package.GetObjectHash());
+	;
+
+	for (const IoHash& AttachmentCid : AttachmentsToSend)
+	{
+		const CbAttachment* Attachment = Package.FindAttachment(AttachmentCid);
+
+		if (Attachment)
+		{
+			SendPackage.AddAttachment(*Attachment);
+		}
+		else
+		{
+			// This should be an error -- server asked to have something we can't find
+		}
+	}
+
+	{
+		MemoryOutStream MemOut;
+		BinaryWriter	MemWriter(MemOut);
+		SendPackage.Save(MemWriter);
+
+		Sess.SetHeader({{"Content-Type", "application/x-ue-cbpkg"}, {"UE-Session", "123456789012345678901234"}, {"UE-Request", "1"}});
+		Sess.SetBody(cpr::Body{(const char*)MemOut.Data(), MemOut.Size()});
+
+		cpr::Response FilterResponse = Sess.Post();
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////
+
 TEST_CASE("httpclient")
 {
 	using namespace std::literals;
diff --git a/zenhttp/httpserver.cpp b/zenhttp/httpserver.cpp
index 75678d433..39bec435d 100644
--- a/zenhttp/httpserver.cpp
+++ b/zenhttp/httpserver.cpp
@@ -49,11 +49,211 @@ MapContentTypeToString(HttpContentType ContentType)
 		case HttpContentType::kCbPackage:
 			return "application/x-ue-cbpkg"sv;
 
+		case HttpContentType::kCbPackageOffer:
+			return "application/x-ue-offer"sv;
+
 		case HttpContentType::kYAML:
 			return "text/yaml"sv;
 	}
 }
 
+static const uint32_t HashBinary					= HashStringDjb2("application/octet-stream"sv);
+static const uint32_t HashJson						= HashStringDjb2("application/json"sv);
+static const uint32_t HashYaml						= HashStringDjb2("text/yaml"sv);
+static const uint32_t HashText						= HashStringDjb2("text/plain"sv);
+static const uint32_t HashCompactBinary				= HashStringDjb2("application/x-ue-cb"sv);
+static const uint32_t HashCompactBinaryPackage		= HashStringDjb2("application/x-ue-cbpkg"sv);
+static const uint32_t HashCompactBinaryPackageOffer = HashStringDjb2("application/x-ue-offer"sv);
+
+HttpContentType
+ParseContentType(const std::string_view& ContentTypeString)
+{
+	if (!ContentTypeString.empty())
+	{
+		const uint32_t CtHash = HashStringDjb2(ContentTypeString);
+
+		if (CtHash == HashBinary)
+		{
+			return HttpContentType::kBinary;
+		}
+		else if (CtHash == HashCompactBinary)
+		{
+			return HttpContentType::kCbObject;
+		}
+		else if (CtHash == HashCompactBinaryPackage)
+		{
+			return HttpContentType::kCbPackage;
+		}
+		else if (CtHash == HashCompactBinaryPackageOffer)
+		{
+			return HttpContentType::kCbPackageOffer;
+		}
+		else if (CtHash == HashJson)
+		{
+			return HttpContentType::kJSON;
+		}
+		else if (CtHash == HashYaml)
+		{
+			return HttpContentType::kYAML;
+		}
+		else if (CtHash == HashText)
+		{
+			return HttpContentType::kText;
+		}
+	}
+
+	return HttpContentType::kUnknownContentType;
+}
+
+const char*
+ReasonStringForHttpResultCode(int HttpCode)
+{
+	switch (HttpCode)
+	{
+			// 1xx Informational
+
+		case 100:
+			return "Continue";
+		case 101:
+			return "Switching Protocols";
+
+			// 2xx Success
+
+		case 200:
+			return "OK";
+		case 201:
+			return "Created";
+		case 202:
+			return "Accepted";
+		case 204:
+			return "No Content";
+		case 205:
+			return "Reset Content";
+		case 206:
+			return "Partial Content";
+
+			// 3xx Redirection
+
+		case 300:
+			return "Multiple Choices";
+		case 301:
+			return "Moved Permanently";
+		case 302:
+			return "Found";
+		case 303:
+			return "See Other";
+		case 304:
+			return "Not Modified";
+		case 305:
+			return "Use Proxy";
+		case 306:
+			return "Switch Proxy";
+		case 307:
+			return "Temporary Redirect";
+		case 308:
+			return "Permanent Redirect";
+
+			// 4xx Client errors
+
+		case 400:
+			return "Bad Request";
+		case 401:
+			return "Unauthorized";
+		case 402:
+			return "Payment Required";
+		case 403:
+			return "Forbidden";
+		case 404:
+			return "Not Found";
+		case 405:
+			return "Method Not Allowed";
+		case 406:
+			return "Not Acceptable";
+		case 407:
+			return "Proxy Authentication Required";
+		case 408:
+			return "Request Timeout";
+		case 409:
+			return "Conflict";
+		case 410:
+			return "Gone";
+		case 411:
+			return "Length Required";
+		case 412:
+			return "Precondition Failed";
+		case 413:
+			return "Payload Too Large";
+		case 414:
+			return "URI Too Long";
+		case 415:
+			return "Unsupported Media Type";
+		case 416:
+			return "Range Not Satisifiable";
+		case 417:
+			return "Expectation Failed";
+		case 418:
+			return "I'm a teapot";
+		case 421:
+			return "Misdirected Request";
+		case 422:
+			return "Unprocessable Entity";
+		case 423:
+			return "Locked";
+		case 424:
+			return "Failed Dependency";
+		case 425:
+			return "Too Early";
+		case 426:
+			return "Upgrade Required";
+		case 428:
+			return "Precondition Required";
+		case 429:
+			return "Too Many Requests";
+		case 431:
+			return "Request Header Fields Too Large";
+
+			// 5xx Server errors
+
+		case 500:
+			return "Internal Server Error";
+		case 501:
+			return "Not Implemented";
+		case 502:
+			return "Bad Gateway";
+		case 503:
+			return "Service Unavailable";
+		case 504:
+			return "Gateway Timeout";
+		case 505:
+			return "HTTP Version Not Supported";
+		case 506:
+			return "Variant Also Negotiates";
+		case 507:
+			return "Insufficient Storage";
+		case 508:
+			return "Loop Detected";
+		case 510:
+			return "Not Extended";
+		case 511:
+			return "Network Authentication Required";
+
+		default:
+			return "Unknown Result";
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////
+
+Ref<IHttpPackageHandler>
+HttpService::HandlePackageRequest(HttpServerRequest& HttpServiceRequest)
+{
+	ZEN_UNUSED(HttpServiceRequest);
+
+	return nullptr;
+}
+
+//////////////////////////////////////////////////////////////////////////
+
 HttpServerRequest::HttpServerRequest()
 {
 }
@@ -197,6 +397,18 @@ HttpServerRequest::GetQueryParams()
 	return Params;
 }
 
+Oid
+HttpServerRequest::SessionId() const
+{
+	return {};
+}
+
+uint32_t
+HttpServerRequest::RequestId() const
+{
+	return {};
+}
+
 CbObject
 HttpServerRequest::ReadPayloadObject()
 {
diff --git a/zenhttp/httpsys.cpp b/zenhttp/httpsys.cpp
index 8aa30344b..f5f9d4249 100644
--- a/zenhttp/httpsys.cpp
+++ b/zenhttp/httpsys.cpp
@@ -2,6 +2,8 @@
 
 #include "httpsys.h"
 
+#include <zencore/compactbinary.h>
+#include <zencore/compactbinarybuilder.h>
 #include <zencore/except.h>
 #include <zencore/logging.h>
 #include <zencore/scopeguard.h>
@@ -59,197 +61,14 @@ UTF8_to_wstring(const char* in)
 
 namespace zen {
 
+using namespace std::literals;
+
 class HttpSysServer;
 class HttpSysTransaction;
 class HttpMessageResponseRequest;
 
-
-//////////////////////////////////////////////////////////////////////////
-
-using namespace std::literals;
-
-static const uint32_t HashBinary			   = HashStringDjb2("application/octet-stream"sv);
-static const uint32_t HashJson				   = HashStringDjb2("application/json"sv);
-static const uint32_t HashYaml				   = HashStringDjb2("text/yaml"sv);
-static const uint32_t HashText				   = HashStringDjb2("text/plain"sv);
-static const uint32_t HashCompactBinary		   = HashStringDjb2("application/x-ue-cb"sv);
-static const uint32_t HashCompactBinaryPackage = HashStringDjb2("application/x-ue-cbpkg"sv);
-
-HttpContentType
-MapContentType(const std::string_view& ContentTypeString)
-{
-	if (!ContentTypeString.empty())
-	{
-		const uint32_t CtHash = HashStringDjb2(ContentTypeString);
-
-		if (CtHash == HashBinary)
-		{
-			return HttpContentType::kBinary;
-		}
-		else if (CtHash == HashCompactBinary)
-		{
-			return HttpContentType::kCbObject;
-		}
-		else if (CtHash == HashCompactBinaryPackage)
-		{
-			return HttpContentType::kCbPackage;
-		}
-		else if (CtHash == HashJson)
-		{
-			return HttpContentType::kJSON;
-		}
-		else if (CtHash == HashYaml)
-		{
-			return HttpContentType::kYAML;
-		}
-		else if (CtHash == HashText)
-		{
-			return HttpContentType::kText;
-		}
-	}
-
-	return HttpContentType::kUnknownContentType;
-}
-
 //////////////////////////////////////////////////////////////////////////
 
-const char*
-ReasonStringForHttpResultCode(int HttpCode)
-{
-	switch (HttpCode)
-	{
-			// 1xx Informational
-
-		case 100:
-			return "Continue";
-		case 101:
-			return "Switching Protocols";
-
-			// 2xx Success
-
-		case 200:
-			return "OK";
-		case 201:
-			return "Created";
-		case 202:
-			return "Accepted";
-		case 204:
-			return "No Content";
-		case 205:
-			return "Reset Content";
-		case 206:
-			return "Partial Content";
-
-			// 3xx Redirection
-
-		case 300:
-			return "Multiple Choices";
-		case 301:
-			return "Moved Permanently";
-		case 302:
-			return "Found";
-		case 303:
-			return "See Other";
-		case 304:
-			return "Not Modified";
-		case 305:
-			return "Use Proxy";
-		case 306:
-			return "Switch Proxy";
-		case 307:
-			return "Temporary Redirect";
-		case 308:
-			return "Permanent Redirect";
-
-			// 4xx Client errors
-
-		case 400:
-			return "Bad Request";
-		case 401:
-			return "Unauthorized";
-		case 402:
-			return "Payment Required";
-		case 403:
-			return "Forbidden";
-		case 404:
-			return "Not Found";
-		case 405:
-			return "Method Not Allowed";
-		case 406:
-			return "Not Acceptable";
-		case 407:
-			return "Proxy Authentication Required";
-		case 408:
-			return "Request Timeout";
-		case 409:
-			return "Conflict";
-		case 410:
-			return "Gone";
-		case 411:
-			return "Length Required";
-		case 412:
-			return "Precondition Failed";
-		case 413:
-			return "Payload Too Large";
-		case 414:
-			return "URI Too Long";
-		case 415:
-			return "Unsupported Media Type";
-		case 416:
-			return "Range Not Satisifiable";
-		case 417:
-			return "Expectation Failed";
-		case 418:
-			return "I'm a teapot";
-		case 421:
-			return "Misdirected Request";
-		case 422:
-			return "Unprocessable Entity";
-		case 423:
-			return "Locked";
-		case 424:
-			return "Failed Dependency";
-		case 425:
-			return "Too Early";
-		case 426:
-			return "Upgrade Required";
-		case 428:
-			return "Precondition Required";
-		case 429:
-			return "Too Many Requests";
-		case 431:
-			return "Request Header Fields Too Large";
-
-			// 5xx Server errors
-
-		case 500:
-			return "Internal Server Error";
-		case 501:
-			return "Not Implemented";
-		case 502:
-			return "Bad Gateway";
-		case 503:
-			return "Service Unavailable";
-		case 504:
-			return "Gateway Timeout";
-		case 505:
-			return "HTTP Version Not Supported";
-		case 506:
-			return "Variant Also Negotiates";
-		case 507:
-			return "Insufficient Storage";
-		case 508:
-			return "Loop Detected";
-		case 510:
-			return "Not Extended";
-		case 511:
-			return "Network Authentication Required";
-
-		default:
-			return "Unknown Result";
-	}
-}
-
 HttpVerb
 TranslateHttpVerb(HTTP_VERB ReqVerb)
 {
@@ -296,14 +115,14 @@ HttpContentType
 GetContentType(const HTTP_REQUEST* HttpRequest)
 {
 	const HTTP_KNOWN_HEADER& CtHdr = HttpRequest->Headers.KnownHeaders[HttpHeaderContentType];
-	return MapContentType({CtHdr.pRawValue, CtHdr.RawValueLength});
+	return ParseContentType({CtHdr.pRawValue, CtHdr.RawValueLength});
 };
 
 HttpContentType
 GetAcceptType(const HTTP_REQUEST* HttpRequest)
 {
 	const HTTP_KNOWN_HEADER& CtHdr = HttpRequest->Headers.KnownHeaders[HttpHeaderAccept];
-	return MapContentType({CtHdr.pRawValue, CtHdr.RawValueLength});
+	return ParseContentType({CtHdr.pRawValue, CtHdr.RawValueLength});
 };
 
 /**
@@ -357,11 +176,16 @@ public:
 	HttpSysServerRequest(HttpSysTransaction& Tx, HttpService& Service, IoBuffer PayloadBuffer);
 	~HttpSysServerRequest() = default;
 
+	virtual Oid		 SessionId() const override;
+	virtual uint32_t RequestId() const override;
+
 	virtual IoBuffer ReadPayload() override;
 	virtual void	 WriteResponse(HttpResponseCode ResponseCode) override;
 	virtual void	 WriteResponse(HttpResponseCode ResponseCode, HttpContentType ContentType, std::span<IoBuffer> Blobs) override;
 	virtual void	 WriteResponse(HttpResponseCode ResponseCode, HttpContentType ContentType, std::u8string_view ResponseString) override;
 
+	using HttpServerRequest::WriteResponse;
+
 	HttpSysTransaction&			m_HttpTx;
 	HttpMessageResponseRequest* m_Response = nullptr;  // TODO: make this more general
 	IoBuffer					m_PayloadBuffer;
@@ -400,12 +224,18 @@ public:
 	inline HttpSysServer& Server() { return m_HttpServer; }
 	inline HTTP_REQUEST*  HttpRequest() { return m_InitialHttpHandler.HttpRequest(); }
 
+	HttpSysServerRequest& InvokeRequestHandler(HttpService& Service, IoBuffer Payload);
+
 private:
-	OVERLAPPED			   m_HttpOverlapped{};
-	HttpSysServer&		   m_HttpServer;
-	HttpSysRequestHandler* m_CompletionHandler{nullptr};  // Tracks which handler is due to handle the next I/O completion event
-	RwLock				   m_CompletionMutex;
-	InitialRequestHandler  m_InitialHttpHandler{*this};
+	OVERLAPPED	   m_HttpOverlapped{};
+	HttpSysServer& m_HttpServer;
+
+	// Tracks which handler is due to handle the next I/O completion event
+	HttpSysRequestHandler*				m_CompletionHandler = nullptr;
+	RwLock								m_CompletionMutex;
+	InitialRequestHandler				m_InitialHttpHandler{*this};
+	std::optional<HttpSysServerRequest> m_HandlerRequest;
+	Ref<IHttpPackageHandler>			m_PackageHandler;
 };
 
 //////////////////////////////////////////////////////////////////////////
@@ -435,22 +265,19 @@ public:
 
 	virtual void				   IssueRequest(std::error_code& ErrorCode) override final;
 	virtual HttpSysRequestHandler* HandleCompletion(ULONG IoResult, ULONG_PTR NumberOfBytesTransferred) override;
-
-	void SuppressResponseBody();
+	void						   SuppressResponseBody();	// typically used for HEAD requests
 
 private:
 	std::vector<HTTP_DATA_CHUNK> m_HttpDataChunks;
-	uint64_t					 m_TotalDataSize = 0;  // Sum of all chunk sizes
-
-	uint16_t m_ResponseCode		   = 0;
-	uint32_t m_NextDataChunkOffset = 0;	 // This is used for responses where the number of chunks exceed the maximum number for one API call
-	uint32_t m_RemainingChunkCount = 0;
-	bool	 m_IsInitialResponse   = true;
-	HttpContentType m_ContentType  = HttpContentType::kBinary;
-
-	void Initialize(uint16_t ResponseCode, std::span<IoBuffer> Blobs);
-
-	std::vector<IoBuffer> m_DataBuffers;
+	uint64_t					 m_TotalDataSize	   = 0;	 // Sum of all chunk sizes
+	uint16_t					 m_ResponseCode		   = 0;
+	uint32_t					 m_NextDataChunkOffset = 0;	 // Cursor used for very large chunk lists
+	uint32_t					 m_RemainingChunkCount = 0;	 // Backlog for multi-call sends
+	bool						 m_IsInitialResponse   = true;
+	HttpContentType				 m_ContentType		   = HttpContentType::kBinary;
+	std::vector<IoBuffer>		 m_DataBuffers;
+
+	void InitializeForPayload(uint16_t ResponseCode, std::span<IoBuffer> Blobs);
 };
 
 HttpMessageResponseRequest::HttpMessageResponseRequest(HttpSysTransaction& InRequest, uint16_t ResponseCode)
@@ -458,7 +285,7 @@ HttpMessageResponseRequest::HttpMessageResponseRequest(HttpSysTransaction& InReq
 {
 	std::array<IoBuffer, 0> EmptyBufferList;
 
-	Initialize(ResponseCode, EmptyBufferList);
+	InitializeForPayload(ResponseCode, EmptyBufferList);
 }
 
 HttpMessageResponseRequest::HttpMessageResponseRequest(HttpSysTransaction& InRequest, uint16_t ResponseCode, std::string_view Message)
@@ -468,7 +295,7 @@ HttpMessageResponseRequest::HttpMessageResponseRequest(HttpSysTransaction& InReq
 	IoBuffer				MessageBuffer(IoBuffer::Wrap, Message.data(), Message.size());
 	std::array<IoBuffer, 1> SingleBufferList({MessageBuffer});
 
-	Initialize(ResponseCode, SingleBufferList);
+	InitializeForPayload(ResponseCode, SingleBufferList);
 }
 
 HttpMessageResponseRequest::HttpMessageResponseRequest(HttpSysTransaction& InRequest,
@@ -482,7 +309,7 @@ HttpMessageResponseRequest::HttpMessageResponseRequest(HttpSysTransaction& InReq
 	IoBuffer				MessageBuffer(IoBuffer::Wrap, Payload, PayloadSize);
 	std::array<IoBuffer, 1> SingleBufferList({MessageBuffer});
 
-	Initialize(ResponseCode, SingleBufferList);
+	InitializeForPayload(ResponseCode, SingleBufferList);
 }
 
 HttpMessageResponseRequest::HttpMessageResponseRequest(HttpSysTransaction& InRequest,
@@ -490,10 +317,9 @@ HttpMessageResponseRequest::HttpMessageResponseRequest(HttpSysTransaction& InReq
 													   HttpContentType	   ContentType,
 													   std::span<IoBuffer> BlobList)
 : HttpSysRequestHandler(InRequest)
+, m_ContentType(ContentType)
 {
-	Initialize(ResponseCode, BlobList);
-
-	ZEN_UNUSED(ContentType);
+	InitializeForPayload(ResponseCode, BlobList);
 }
 
 HttpMessageResponseRequest::~HttpMessageResponseRequest()
@@ -501,7 +327,7 @@ HttpMessageResponseRequest::~HttpMessageResponseRequest()
 }
 
 void
-HttpMessageResponseRequest::Initialize(uint16_t ResponseCode, std::span<IoBuffer> BlobList)
+HttpMessageResponseRequest::InitializeForPayload(uint16_t ResponseCode, std::span<IoBuffer> BlobList)
 {
 	m_ResponseCode = ResponseCode;
 
@@ -908,10 +734,10 @@ HttpSysServer::IssueNewRequestMaybe()
 
 	std::unique_ptr<HttpSysTransaction> Request = std::make_unique<HttpSysTransaction>(*this);
 
-	std::error_code ec;
-	Request->IssueInitialRequest(ec);
+	std::error_code ErrorCode;
+	Request->IssueInitialRequest(ErrorCode);
 
-	if (ec)
+	if (ErrorCode)
 	{
 		// No request was actually issued. What is the appropriate response?
 
@@ -1054,12 +880,12 @@ HttpSysTransaction::HandleCompletion(ULONG IoResult, ULONG_PTR NumberOfBytesTran
 		{
 			try
 			{
-				std::error_code ec;
-				m_CompletionHandler->IssueRequest(ec);
+				std::error_code ErrorCode;
+				m_CompletionHandler->IssueRequest(ErrorCode);
 
-				if (ec)
+				if (ErrorCode)
 				{
-					spdlog::error("IssueRequest() failed {}"sv, ec.message());
+					spdlog::error("IssueRequest() failed {}"sv, ErrorCode.message());
 				}
 				else
 				{
@@ -1095,6 +921,66 @@ HttpSysTransaction::HandleCompletion(ULONG IoResult, ULONG_PTR NumberOfBytesTran
 	return Status::kDone;
 }
 
+HttpSysServerRequest&
+HttpSysTransaction::InvokeRequestHandler(HttpService& Service, IoBuffer Payload)
+{
+	HttpSysServerRequest& ThisRequest = m_HandlerRequest.emplace(*this, Service, Payload);
+
+	if ((ThisRequest.RequestContentType() == HttpContentType::kCbPackageOffer) && (ThisRequest.RequestVerb() == HttpVerb::kPost))
+	{
+		// The client is presenting us with a package attachments offer, we need
+		// to filter it down to the list of attachments we need them to send in
+		// the follow-up request
+
+		m_PackageHandler = Service.HandlePackageRequest(ThisRequest);
+
+		if (m_PackageHandler)
+		{
+			CbObject OfferMessage = LoadCompactBinaryObject(Payload);
+
+			std::vector<IoHash> OfferCids;
+
+			for (auto& CidEntry : OfferMessage["offer"])
+			{
+				if (!CidEntry.IsHash())
+				{
+					// Should yield bad request response?
+
+					continue;
+				}
+
+				OfferCids.push_back(CidEntry.AsHash(IoHash::Zero));
+			}
+
+			m_PackageHandler->FilterOffer(OfferCids);
+
+			CbObjectWriter ResponseWriter;
+			ResponseWriter.BeginArray("need");
+
+			for (const IoHash& Cid : OfferCids)
+			{
+				ResponseWriter.AddHash(Cid);
+			}
+
+			ResponseWriter.EndArray();
+
+			// Emit filter response
+			ThisRequest.WriteResponse(HttpResponseCode::OK, ResponseWriter.Save());
+
+			return ThisRequest;
+		}
+	}
+	else if ((ThisRequest.RequestContentType() == HttpContentType::kCbPackage) && (ThisRequest.RequestVerb() == HttpVerb::kPost))
+	{
+	}
+
+	// Default request handling
+
+	Service.HandleRequest(ThisRequest);
+
+	return ThisRequest;
+}
+
 //////////////////////////////////////////////////////////////////////////
 
 HttpSysServerRequest::HttpSysServerRequest(HttpSysTransaction& Tx, HttpService& Service, IoBuffer PayloadBuffer)
@@ -1137,6 +1023,62 @@ HttpSysServerRequest::HttpSysServerRequest(HttpSysTransaction& Tx, HttpService&
 	m_AcceptType	= GetAcceptType(HttpRequestPtr);
 }
 
+Oid
+HttpSysServerRequest::SessionId() const
+{
+	if (m_Flags & kHaveSessionId)
+	{
+		return m_SessionId;
+	}
+
+	const HTTP_REQUEST* HttpRequestPtr = m_HttpTx.HttpRequest();
+
+	for (int i = 0; i < HttpRequestPtr->Headers.UnknownHeaderCount; ++i)
+	{
+		HTTP_UNKNOWN_HEADER& Header = HttpRequestPtr->Headers.pUnknownHeaders[i];
+		std::string_view	 HeaderName{Header.pName, Header.NameLength};
+
+		if (HeaderName == "UE-Session"sv)
+		{
+			if (Header.RawValueLength == Oid::StringLength)
+			{
+				m_SessionId = Oid::FromHexString({Header.pRawValue, Header.RawValueLength});
+			}
+		}
+	}
+
+	m_Flags |= kHaveSessionId;
+
+	return m_SessionId;
+}
+
+uint32_t
+HttpSysServerRequest::RequestId() const
+{
+	if (m_Flags & kHaveRequestId)
+	{
+		return m_RequestId;
+	}
+
+	const HTTP_REQUEST* HttpRequestPtr = m_HttpTx.HttpRequest();
+
+	for (int i = 0; i < HttpRequestPtr->Headers.UnknownHeaderCount; ++i)
+	{
+		HTTP_UNKNOWN_HEADER& Header = HttpRequestPtr->Headers.pUnknownHeaders[i];
+		std::string_view	 HeaderName{Header.pName, Header.NameLength};
+
+		if (HeaderName == "UE-Request"sv)
+		{
+			std::string_view RequestValue{Header.pRawValue, Header.RawValueLength};
+			std::from_chars(RequestValue.data(), RequestValue.data() + RequestValue.size(), m_RequestId);
+		}
+	}
+
+	m_Flags |= kHaveRequestId;
+
+	return m_RequestId;
+}
+
 IoBuffer
 HttpSysServerRequest::ReadPayload()
 {
@@ -1146,47 +1088,47 @@ HttpSysServerRequest::ReadPayload()
 void
 HttpSysServerRequest::WriteResponse(HttpResponseCode ResponseCode)
 {
-	ZEN_ASSERT(m_IsHandled == false);
+	ZEN_ASSERT(IsHandled() == false);
 
 	m_Response = new HttpMessageResponseRequest(m_HttpTx, (uint16_t)ResponseCode);
 
-	if (m_SuppressBody)
+	if (SuppressBody())
 	{
 		m_Response->SuppressResponseBody();
 	}
 
-	m_IsHandled = true;
+	SetIsHandled();
 }
 
 void
 HttpSysServerRequest::WriteResponse(HttpResponseCode ResponseCode, HttpContentType ContentType, std::span<IoBuffer> Blobs)
 {
-	ZEN_ASSERT(m_IsHandled == false);
+	ZEN_ASSERT(IsHandled() == false);
 
 	m_Response = new HttpMessageResponseRequest(m_HttpTx, (uint16_t)ResponseCode, ContentType, Blobs);
 
-	if (m_SuppressBody)
+	if (SuppressBody())
 	{
 		m_Response->SuppressResponseBody();
 	}
 
-	m_IsHandled = true;
+	SetIsHandled();
 }
 
 void
 HttpSysServerRequest::WriteResponse(HttpResponseCode ResponseCode, HttpContentType ContentType, std::u8string_view ResponseString)
 {
-	ZEN_ASSERT(m_IsHandled == false);
+	ZEN_ASSERT(IsHandled() == false);
 
 	m_Response =
 		new HttpMessageResponseRequest(m_HttpTx, (uint16_t)ResponseCode, ContentType, ResponseString.data(), ResponseString.size());
 
-	if (m_SuppressBody)
+	if (SuppressBody())
 	{
 		m_Response->SuppressResponseBody();
 	}
 
-	m_IsHandled = true;
+	SetIsHandled();
 }
 
 //////////////////////////////////////////////////////////////////////////
@@ -1388,18 +1330,16 @@ InitialRequestHandler::HandleCompletion(ULONG IoResult, ULONG_PTR NumberOfBytesT
 
 				// Body received completely - call request handler
 
-				HttpSysServerRequest ThisRequest(Transaction(), *Service, m_PayloadBuffer);
-
-				Service->HandleRequest(ThisRequest);
+				HttpSysServerRequest& ThisRequest = Transaction().InvokeRequestHandler(*Service, m_PayloadBuffer);
 
 				if (!ThisRequest.IsHandled())
 				{
 					return new HttpMessageResponseRequest(Transaction(), 404, "Not found"sv);
 				}
 
-				if (ThisRequest.m_Response)
+				if (HttpMessageResponseRequest* Response = ThisRequest.m_Response)
 				{
-					return ThisRequest.m_Response;
+					return Response;
 				}
 			}
 			else
diff --git a/zenhttp/include/zenhttp/httpclient.h b/zenhttp/include/zenhttp/httpclient.h
index 4a266e59e..10829a58c 100644
--- a/zenhttp/include/zenhttp/httpclient.h
+++ b/zenhttp/include/zenhttp/httpclient.h
@@ -4,17 +4,33 @@
 
 #include "zenhttp.h"
 
-#include <zencore/string.h>
-#include <gsl/gsl-lite.hpp>
+#include <zencore/windows.h>
+
+// For some reason, these don't seem to stick, so we disable the warnings
+//#	define _SILENCE_CXX17_C_HEADER_DEPRECATION_WARNING 1
+//#	define _SILENCE_ALL_CXX17_DEPRECATION_WARNINGS		1
+#pragma warning(push)
+#pragma warning(disable : 4004)
+#pragma warning(disable : 4996)
+#include <cpr/cpr.h>
+#pragma warning(pop)
 
 namespace zen {
 
+class CbPackage;
+
 /** Asynchronous HTTP client implementation for Zen use cases
  */
 class HttpClient
 {
 public:
+	HttpClient(std::string_view BaseUri);
+	~HttpClient();
+
+	void TransactPackage(std::string_view Url, CbPackage Package);
+
 private:
+	std::string m_BaseUri;
 };
 
 }  // namespace zen
diff --git a/zenhttp/include/zenhttp/httpserver.h b/zenhttp/include/zenhttp/httpserver.h
index b8b63ac4d..f859eb038 100644
--- a/zenhttp/include/zenhttp/httpserver.h
+++ b/zenhttp/include/zenhttp/httpserver.h
@@ -6,8 +6,10 @@
 
 #include <zencore/enumflags.h>
 #include <zencore/iobuffer.h>
+#include <zencore/iohash.h>
 #include <zencore/refcount.h>
 #include <zencore/string.h>
+#include <zencore/uid.h>
 
 #include <functional>
 #include <gsl/gsl-lite.hpp>
@@ -26,8 +28,10 @@ class CbPackage;
 class StringBuilderBase;
 
 std::string_view MapContentTypeToString(HttpContentType ContentType);
+HttpContentType	 ParseContentType(const std::string_view& ContentTypeString);
+const char*		 ReasonStringForHttpResultCode(int HttpCode);
 
-enum class HttpVerb
+enum class HttpVerb : uint8_t
 {
 	kGet	 = 1 << 0,
 	kPut	 = 1 << 1,
@@ -179,7 +183,6 @@ public:
 
 	[[nodiscard]] inline std::string_view RelativeUri() const { return m_UriUtf8; }	 // Returns URI without service prefix
 	[[nodiscard]] inline std::string_view QueryString() const { return m_QueryStringUtf8; }
-	inline bool							  IsHandled() const { return m_IsHandled; }
 
 	struct QueryParams
 	{
@@ -210,25 +213,16 @@ public:
 	inline HttpContentType RequestContentType() { return m_ContentType; }
 	inline HttpContentType AcceptContentType() { return m_AcceptType; }
 
-	const char*		HeaderAccept() const;
-	const char*		HeaderAcceptEncoding() const;
-	const char*		HeaderContentType() const;
-	const char*		HeaderContentEncoding() const;
-	inline uint64_t HeaderContentLength() const { return m_ContentLength; }
+	inline uint64_t	 ContentLength() const { return m_ContentLength; }
+	virtual Oid		 SessionId() const;
+	virtual uint32_t RequestId() const;
 
-	void SetSuppressResponseBody() { m_SuppressBody = true; }
+	inline bool IsHandled() const { return !!(m_Flags & kIsHandled); }
+	inline bool SuppressBody() const { return !!(m_Flags & kSuppressBody); }
+	inline void SetSuppressResponseBody() { m_Flags |= kSuppressBody; }
 
-	// Asynchronous operations
-
-	/** Read POST/PUT payload
-
-		This will return a null buffer if the contents are not fully available yet, and the handler should
-		at that point return - another completion request will be issued once the contents have been received
-		fully.
-
-		NOTE: in practice, via the http.sys implementation this always operates synchronously. This should
-		be updated to provide fully asynchronous operation for better scalability on shared instances
-	  */
+	/** Read POST/PUT payload for request body, which is always available without delay
+	 */
 	virtual IoBuffer ReadPayload() = 0;
 
 	ZENCORE_API CbObject  ReadPayloadObject();
@@ -236,6 +230,9 @@ public:
 
 	/** Respond with payload
 
+		No data will have been sent when any of these functions return. Instead, the response will be transmitted
+		asynchronously, after returning from a request handler function.
+
 		Note that this is destructive in the sense that the IoBuffer instances referred to by Blobs will be
 		moved into our response handler array where they are kept alive, in order to reduce ref-counting storms
 	  */
@@ -249,14 +246,34 @@ public:
 	void WriteResponse(HttpResponseCode ResponseCode, HttpContentType ContentType, IoBuffer Blob);
 
 protected:
-	bool						 m_IsHandled	 = false;
-	bool						 m_SuppressBody	 = false;
+	enum
+	{
+		kIsHandled	   = 1 << 0,
+		kSuppressBody  = 1 << 1,
+		kHaveRequestId = 1 << 2,
+		kHaveSessionId = 1 << 3,
+	};
+
+	mutable uint32_t			 m_Flags		 = 0;
 	HttpVerb					 m_Verb			 = HttpVerb::kGet;
-	uint64_t					 m_ContentLength = ~0ull;
 	HttpContentType				 m_ContentType	 = HttpContentType::kBinary;
 	HttpContentType				 m_AcceptType	 = HttpContentType::kUnknownContentType;
-	ExtendableStringBuilder<256> m_UriUtf8;
-	ExtendableStringBuilder<256> m_QueryStringUtf8;
+	uint64_t					 m_ContentLength = ~0ull;
+	ExtendableStringBuilder<128> m_UriUtf8;
+	ExtendableStringBuilder<128> m_QueryStringUtf8;
+	mutable uint32_t			 m_RequestId = ~uint32_t(0);
+	mutable Oid					 m_SessionId = Oid::Zero;
+
+	inline void SetIsHandled() { m_Flags |= kIsHandled; }
+};
+
+class IHttpPackageHandler : public RefCounted
+{
+public:
+	virtual void	 FilterOffer(std::vector<IoHash>& OfferCids)		   = 0;
+	virtual void	 OnBeginChunks()									   = 0;
+	virtual IoBuffer CreateTarget(const IoHash& Cid, uint64_t StorageSize) = 0;
+	virtual void	 OnEndChunks()										   = 0;
 };
 
 /**
@@ -272,8 +289,9 @@ public:
 	HttpService()		   = default;
 	virtual ~HttpService() = default;
 
-	virtual const char* BaseUri() const										 = 0;
-	virtual void		HandleRequest(HttpServerRequest& HttpServiceRequest) = 0;
+	virtual const char*				 BaseUri() const									  = 0;
+	virtual void					 HandleRequest(HttpServerRequest& HttpServiceRequest) = 0;
+	virtual Ref<IHttpPackageHandler> HandlePackageRequest(HttpServerRequest& HttpServiceRequest);
 
 	// Internals
 
diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index 77cae2d99..b2f8d191c 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -25,27 +25,6 @@ namespace zen {
 
 using namespace std::literals;
 
-zen::HttpContentType
-MapToHttpContentType(zen::ZenContentType Type)
-{
-	switch (Type)
-	{
-		default:
-		case zen::ZenContentType::kBinary:
-			return zen::HttpContentType::kBinary;
-		case zen::ZenContentType::kCbObject:
-			return zen::HttpContentType::kCbObject;
-		case zen::ZenContentType::kCbPackage:
-			return zen::HttpContentType::kCbPackage;
-		case zen::ZenContentType::kText:
-			return zen::HttpContentType::kText;
-		case zen::ZenContentType::kJSON:
-			return zen::HttpContentType::kJSON;
-		case zen::ZenContentType::kYAML:
-			return zen::HttpContentType::kYAML;
-	}
-};
-
 //////////////////////////////////////////////////////////////////////////
 
 HttpStructuredCacheService::HttpStructuredCacheService(::ZenCacheStore&				  InCacheStore,
@@ -224,7 +203,7 @@ HttpStructuredCacheService::HandleCacheRecordRequest(zen::HttpServerRequest& Req
 							Value.Value.GetContentType(),
 							InUpstreamCache ? "upstream" : "local");
 
-				return Request.WriteResponse(zen::HttpResponseCode::OK, MapToHttpContentType(Value.Value.GetContentType()), Value.Value);
+				return Request.WriteResponse(zen::HttpResponseCode::OK, Value.Value.GetContentType(), Value.Value);
 			}
 			break;