// Copyright Epic Games, Inc. All Rights Reserved.

#include "structuredcache.h"

#include <zencore/compactbinary.h>
#include <zencore/compactbinarybuilder.h>
#include <zencore/compactbinarypackage.h>
#include <zencore/compactbinaryvalidation.h>
#include <zencore/compress.h>
#include <zencore/trace.h>
#include <zencore/fmtutils.h>
#include <zencore/logging.h>
#include <zencore/scopeguard.h>
#include <zencore/stream.h>
#include <zencore/timer.h>
#include <zenhttp/httpserver.h>
#include <zenstore/cas.h>
#include <zenutil/cache/cache.h>

//#include "cachekey.h"
#include "monitoring/httpstats.h"
#include "structuredcachestore.h"
#include "upstream/jupiter.h"
#include "upstream/upstreamcache.h"
#include "upstream/zen.h"
#include "zenstore/cidstore.h"

#include <algorithm>
#include <atomic>
#include <filesystem>
#include <queue>
#include <thread>

#include <gsl/gsl-lite.hpp>

namespace zen {

using namespace std::literals;

//////////////////////////////////////////////////////////////////////////

CachePolicy
ParseCachePolicy(const HttpServerRequest::QueryParams& QueryParams)
{
	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;
}

struct AttachmentCount
{
	uint32_t New	 = 0;
	uint32_t Valid	 = 0;
	uint32_t Invalid = 0;
	uint32_t Total	 = 0;
};

//////////////////////////////////////////////////////////////////////////

HttpStructuredCacheService::HttpStructuredCacheService(ZenCacheStore&				  InCacheStore,
													   CidStore&					  InCidStore,
													   HttpStatsService&			  StatsService,
													   HttpStatusService&			  StatusService,
													   std::unique_ptr<UpstreamCache> UpstreamCache)
: m_Log(logging::Get("cache"))
, m_CacheStore(InCacheStore)
, m_StatsService(StatsService)
, m_StatusService(StatusService)
, m_CidStore(InCidStore)
, m_UpstreamCache(std::move(UpstreamCache))
{
	m_StatsService.RegisterHandler("z$", *this);
	m_StatusService.RegisterHandler("z$", *this);
}

HttpStructuredCacheService::~HttpStructuredCacheService()
{
	ZEN_INFO("closing structured cache");

	m_StatsService.UnregisterHandler("z$", *this);
	m_StatusService.UnregisterHandler("z$", *this);
}

const char*
HttpStructuredCacheService::BaseUri() const
{
	return "/z$/";
}

void
HttpStructuredCacheService::Flush()
{
}

void
HttpStructuredCacheService::Scrub(ScrubContext& Ctx)
{
	if (m_LastScrubTime == Ctx.ScrubTimestamp())
	{
		return;
	}

	m_LastScrubTime = Ctx.ScrubTimestamp();

	m_CidStore.Scrub(Ctx);
	m_CacheStore.Scrub(Ctx);
}

void
HttpStructuredCacheService::HandleRequest(HttpServerRequest& Request)
{
	CacheRef Ref;

	metrics::OperationTiming::Scope $(m_HttpRequests);

	if (!ValidateKeyUri(Request, /* out */ Ref))
	{
		std::string_view Key = Request.RelativeUri();

		if (Key == "$rpc")
		{
			return HandleRpcRequest(Request);
		}

		if (std::all_of(begin(Key), end(Key), [](const char c) { return std::isalnum(c); }))
		{
			// Bucket reference

			return HandleCacheBucketRequest(Request, Key);
		}

		return Request.WriteResponse(HttpResponseCode::BadRequest);	 // invalid URL
	}

	const auto	QueryParams = Request.GetQueryParams();
	CachePolicy Policy		= ParseCachePolicy(QueryParams);

	if (Ref.PayloadId == IoHash::Zero)
	{
		return HandleCacheRecordRequest(Request, Ref, Policy);
	}
	else
	{
		return HandleCachePayloadRequest(Request, Ref, Policy);
	}

	return;
}

void
HttpStructuredCacheService::HandleCacheBucketRequest(HttpServerRequest& Request, std::string_view Bucket)
{
	switch (Request.RequestVerb())
	{
		case HttpVerb::kHead:
		case HttpVerb::kGet:
			{
				// Query stats
			}
			break;

		case HttpVerb::kDelete:
			// Drop bucket

			if (m_CacheStore.DropBucket(Bucket))
			{
				return Request.WriteResponse(HttpResponseCode::OK);
			}
			else
			{
				return Request.WriteResponse(HttpResponseCode::NotFound);
			}
			break;

		default:
			break;
	}
}

void
HttpStructuredCacheService::HandleCacheRecordRequest(HttpServerRequest& Request, const CacheRef& Ref, CachePolicy Policy)
{
	switch (Request.RequestVerb())
	{
		case HttpVerb::kHead:
		case HttpVerb::kGet:
			{
				HandleGetCacheRecord(Request, Ref, Policy);
			}
			break;

		case HttpVerb::kPut:
			HandlePutCacheRecord(Request, Ref, Policy);
			break;
		default:
			break;
	}
}

void
HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy Policy)
{
	const ZenContentType AcceptType		 = Request.AcceptContentType();
	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);

	bool		  Success = false;
	ZenCacheValue LocalCacheValue;

	if (m_CacheStore.Get(Ref.BucketSegment, Ref.HashKey, LocalCacheValue))
	{
		Success = true;

		if (!SkipData && AcceptType == ZenContentType::kCbPackage)
		{
			CbPackage	 Package;
			CbObjectView CacheRecord(LocalCacheValue.Value.Data());
			uint32_t	 AttachmentCount = 0;
			uint32_t	 ValidCount		 = 0;

			if (!SkipAttachments)
			{
				CacheRecord.IterateAttachments([this, &Ref, &Package, &AttachmentCount, &ValidCount](CbFieldView AttachmentHash) {
					if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
					{
						Package.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Chunk))));
						ValidCount++;
					}
					AttachmentCount++;
				});

				if (ValidCount != AttachmentCount)
				{
					// Success = false;
					ZEN_WARN("GET - '{}/{}' '{}' is partial, found '{}' of '{}' attachments",
							 Ref.BucketSegment,
							 Ref.HashKey,
							 ToString(AcceptType),
							 ValidCount,
							 AttachmentCount);
				}
			}

			Package.SetObject(LoadCompactBinaryObject(LocalCacheValue.Value));

			BinaryWriter MemStream;
			Package.Save(MemStream);

			LocalCacheValue.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
			LocalCacheValue.Value.SetContentType(HttpContentType::kCbPackage);
		}
	}

	if (Success)
	{
		ZEN_DEBUG("HIT - '{}/{}' {} '{}' (LOCAL)",
				  Ref.BucketSegment,
				  Ref.HashKey,
				  NiceBytes(LocalCacheValue.Value.Size()),
				  ToString(LocalCacheValue.Value.GetContentType()));

		m_CacheStats.HitCount++;

		if (SkipData)
		{
			return Request.WriteResponse(HttpResponseCode::OK);
		}
		else
		{
			return Request.WriteResponse(HttpResponseCode::OK, LocalCacheValue.Value.GetContentType(), LocalCacheValue.Value);
		}
	}
	else if (!QueryUpstream)
	{
		ZEN_DEBUG("MISS - '{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
		m_CacheStats.MissCount++;
		return Request.WriteResponse(HttpResponseCode::NotFound);
	}

	// Issue upstream query asynchronously in order to keep requests flowing without
	// hogging I/O servicing threads with blocking work

	Request.WriteResponseAsync([this, AcceptType, SkipData, SkipAttachments, Ref](HttpServerRequest& AsyncRequest) {
		bool		  Success = false;
		ZenCacheValue UpstreamCacheValue;

		metrics::OperationTiming::Scope $(m_UpstreamGetRequestTiming);

		if (GetUpstreamCacheResult UpstreamResult = m_UpstreamCache->GetCacheRecord({Ref.BucketSegment, Ref.HashKey}, AcceptType);
			UpstreamResult.Success)
		{
			Success					 = true;
			UpstreamCacheValue.Value = UpstreamResult.Value;

			UpstreamCacheValue.Value.SetContentType(AcceptType);

			if (AcceptType == ZenContentType::kBinary || AcceptType == ZenContentType::kCbObject)
			{
				if (AcceptType == ZenContentType::kCbObject)
				{
					const CbValidateError ValidationResult = ValidateCompactBinary(UpstreamResult.Value, CbValidateMode::All);

					if (ValidationResult == CbValidateError::None)
					{
						CbObjectView CacheRecord(UpstreamResult.Value.Data());

						CbObjectWriter IndexData;
						IndexData.BeginArray("references");
						CacheRecord.IterateAttachments([&](CbFieldView Attachment) { IndexData.AddHash(Attachment.AsHash()); });
						IndexData.EndArray();

						UpstreamCacheValue.IndexData = IndexData.Save();
					}
					else
					{
						Success = false;
						ZEN_WARN("Get - '{}/{}' '{}' FAILED, invalid compact binary object from upstream",
								 Ref.BucketSegment,
								 Ref.HashKey,
								 ToString(AcceptType));
					}
				}

				if (Success)
				{
					m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, UpstreamCacheValue);
				}
			}
			else if (AcceptType == ZenContentType::kCbPackage)
			{
				CbPackage Package;
				if (Package.TryLoad(UpstreamCacheValue.Value))
				{
					CbObject CacheRecord	 = Package.GetObject();
					uint32_t AttachmentCount = 0;
					uint32_t ValidCount		 = 0;

					CacheRecord.IterateAttachments([this, &Package, &Ref, &AttachmentCount, &ValidCount](CbFieldView AttachmentHash) {
						if (const CbAttachment* Attachment = Package.FindAttachment(AttachmentHash.AsHash()))
						{
							if (CompressedBuffer Chunk = Attachment->AsCompressedBinary())
							{
								m_CidStore.AddChunk(Chunk);
								ValidCount++;
							}
							else
							{
								ZEN_WARN("Get - '{}/{}' '{}' FAILED, upstream attachment not compressed",
										 Ref.BucketSegment,
										 Ref.HashKey,
										 ToString(ZenContentType::kCbPackage));
							}
						}
						AttachmentCount++;
					});

					if (ValidCount == AttachmentCount)
					{
						m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, {.Value = CacheRecord.GetBuffer().AsIoBuffer()});

						if (SkipAttachments)
						{
							CbPackage PackageWithoutAttachments;
							PackageWithoutAttachments.SetObject(CacheRecord);

							BinaryWriter MemStream;
							PackageWithoutAttachments.Save(MemStream);

							UpstreamCacheValue.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
						}
					}
					else
					{
						Success = false;
						ZEN_WARN("Get - '{}/{}' '{}' FAILED, attachments missing in upstream package",
								 Ref.BucketSegment,
								 Ref.HashKey,
								 ToString(AcceptType));
					}
				}
				else
				{
					Success = false;
					ZEN_WARN("Get - '{}/{}' '{}' FAILED, invalid upstream package", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
				}
			}
		}

		if (Success)
		{
			ZEN_DEBUG("HIT - '{}/{}' {} '{}' (UPSTREAM)",
					  Ref.BucketSegment,
					  Ref.HashKey,
					  NiceBytes(UpstreamCacheValue.Value.Size()),
					  ToString(UpstreamCacheValue.Value.GetContentType()));

			m_CacheStats.HitCount++;
			m_CacheStats.UpstreamHitCount++;

			if (SkipData)
			{
				AsyncRequest.WriteResponse(HttpResponseCode::OK);
			}
			else
			{
				AsyncRequest.WriteResponse(HttpResponseCode::OK, UpstreamCacheValue.Value.GetContentType(), UpstreamCacheValue.Value);
			}
		}
		else
		{
			ZEN_DEBUG("MISS - '{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
			m_CacheStats.MissCount++;
			AsyncRequest.WriteResponse(HttpResponseCode::NotFound);
		}
	});
}

void
HttpStructuredCacheService::HandlePutCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy Policy)
{
	IoBuffer Body = Request.ReadPayload();

	if (!Body || Body.Size() == 0)
	{
		return Request.WriteResponse(HttpResponseCode::BadRequest);
	}

	const HttpContentType ContentType	= Request.RequestContentType();
	const bool			  StoreUpstream = m_UpstreamCache && (CachePolicy::StoreRemote == (Policy & CachePolicy::StoreRemote));

	Body.SetContentType(ContentType);

	if (ContentType == HttpContentType::kBinary)
	{
		ZEN_DEBUG("PUT - '{}/{}' {} '{}'", Ref.BucketSegment, Ref.HashKey, NiceBytes(Body.Size()), ToString(ContentType));
		m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, {.Value = Body});

		if (StoreUpstream)
		{
			ZEN_ASSERT(m_UpstreamCache);
			m_UpstreamCache->EnqueueUpstream({.Type = ZenContentType::kBinary, .Key = {Ref.BucketSegment, Ref.HashKey}});
		}

		Request.WriteResponse(HttpResponseCode::Created);
	}
	else if (ContentType == HttpContentType::kCbObject)
	{
		const CbValidateError ValidationResult = ValidateCompactBinary(MemoryView(Body.GetData(), Body.GetSize()), CbValidateMode::All);

		if (ValidationResult != CbValidateError::None)
		{
			ZEN_WARN("PUT - '{}/{}' '{}' FAILED, invalid compact binary", Ref.BucketSegment, Ref.HashKey, ToString(ContentType));
			return Request.WriteResponse(HttpResponseCode::BadRequest, HttpContentType::kText, "Compact binary validation failed"sv);
		}

		CbObjectView		CacheRecord(Body.Data());
		std::vector<IoHash> ValidAttachments;
		int32_t				TotalCount = 0;

		CacheRecord.IterateAttachments([this, &TotalCount, &ValidAttachments](CbFieldView AttachmentHash) {
			const IoHash Hash = AttachmentHash.AsHash();
			if (m_CidStore.ContainsChunk(Hash))
			{
				ValidAttachments.emplace_back(Hash);
			}
			TotalCount++;
		});

		ZEN_DEBUG("PUT - '{}/{}' {} '{}' attachments '{}/{}' (valid/total)",
				  Ref.BucketSegment,
				  Ref.HashKey,
				  NiceBytes(Body.Size()),
				  ToString(ContentType),
				  TotalCount,
				  ValidAttachments.size());

		Body.SetContentType(ZenContentType::kCbObject);
		m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, {.Value = Body});

		const bool IsPartialRecord = TotalCount != static_cast<int32_t>(ValidAttachments.size());

		if (StoreUpstream && !IsPartialRecord)
		{
			ZEN_ASSERT(m_UpstreamCache);
			m_UpstreamCache->EnqueueUpstream(
				{.Type = ZenContentType::kCbObject, .Key = {Ref.BucketSegment, Ref.HashKey}, .PayloadIds = std::move(ValidAttachments)});
		}

		Request.WriteResponse(HttpResponseCode::Created);
	}
	else if (ContentType == HttpContentType::kCbPackage)
	{
		CbPackage Package;

		if (!Package.TryLoad(Body))
		{
			ZEN_WARN("PUT - '{}/{}' '{}' FAILED, invalid package", Ref.BucketSegment, Ref.HashKey, ToString(ContentType));
			return Request.WriteResponse(HttpResponseCode::BadRequest, HttpContentType::kText, "Invalid package"sv);
		}

		CbObject			CacheRecord = Package.GetObject();
		AttachmentCount		Count;
		std::vector<IoHash> ValidAttachments;

		ValidAttachments.reserve(Package.GetAttachments().size());

		CacheRecord.IterateAttachments([this, &Ref, &Package, &ValidAttachments, &Count](CbFieldView HashView) {
			const IoHash Hash = HashView.AsHash();
			if (const CbAttachment* Attachment = Package.FindAttachment(Hash))
			{
				if (Attachment->IsCompressedBinary())
				{
					CompressedBuffer	   Chunk		= Attachment->AsCompressedBinary();
					CidStore::InsertResult InsertResult = m_CidStore.AddChunk(Chunk);

					ValidAttachments.emplace_back(InsertResult.DecompressedId);

					if (InsertResult.New)
					{
						Count.New++;
					}
					Count.Valid++;
				}
				else
				{
					ZEN_WARN("PUT - '{}/{}' '{}' FAILED, attachment '{}' is not compressed",
							 Ref.BucketSegment,
							 Ref.HashKey,
							 ToString(HttpContentType::kCbPackage),
							 Hash);
					Count.Invalid++;
				}
			}
			else if (m_CidStore.ContainsChunk(Hash))
			{
				ValidAttachments.emplace_back(Hash);
				Count.Valid++;
			}
			Count.Total++;
		});

		if (Count.Invalid > 0)
		{
			return Request.WriteResponse(HttpResponseCode::BadRequest, HttpContentType::kText, "Invalid attachment(s)"sv);
		}

		ZEN_DEBUG("PUT - '{}/{}' {} '{}', attachments '{}/{}/{}' (new/valid/total)",
				  Ref.BucketSegment,
				  Ref.HashKey,
				  NiceBytes(Body.GetSize()),
				  ToString(ContentType),
				  Count.New,
				  Count.Valid,
				  Count.Total);

		IoBuffer CacheRecordValue = CacheRecord.GetBuffer().AsIoBuffer();
		CacheRecordValue.SetContentType(ZenContentType::kCbObject);

		m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, {.Value = CacheRecord.GetBuffer().AsIoBuffer()});

		const bool IsPartialRecord = Count.Valid != Count.Total;

		if (StoreUpstream && !IsPartialRecord)
		{
			ZEN_ASSERT(m_UpstreamCache);
			m_UpstreamCache->EnqueueUpstream(
				{.Type = ZenContentType::kCbPackage, .Key = {Ref.BucketSegment, Ref.HashKey}, .PayloadIds = std::move(ValidAttachments)});
		}

		Request.WriteResponse(HttpResponseCode::Created);
	}
	else
	{
		return Request.WriteResponse(HttpResponseCode::BadRequest, HttpContentType::kText, "Content-Type invalid"sv);
	}
}

void
HttpStructuredCacheService::HandleCachePayloadRequest(HttpServerRequest& Request, const CacheRef& Ref, CachePolicy Policy)
{
	switch (Request.RequestVerb())
	{
		case HttpVerb::kHead:
		case HttpVerb::kGet:
			{
				HandleGetCachePayload(Request, Ref, Policy);
			}
			break;
		case HttpVerb::kPut:
			HandlePutCachePayload(Request, Ref, Policy);
			break;
		default:
			break;
	}
}

void
HttpStructuredCacheService::HandleGetCachePayload(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy Policy)
{
	IoBuffer   Payload		   = m_CidStore.FindChunkByCid(Ref.PayloadId);
	bool	   InUpstreamCache = false;
	const bool QueryUpstream   = !Payload && m_UpstreamCache && (CachePolicy::QueryRemote == (Policy & CachePolicy::QueryRemote));

	if (QueryUpstream)
	{
		if (auto UpstreamResult = m_UpstreamCache->GetCachePayload({Ref.BucketSegment, Ref.HashKey}, Ref.PayloadId); UpstreamResult.Success)
		{
			if (CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(UpstreamResult.Value)))
			{
				/*CidStore::InsertResult Result =*/ m_CidStore.AddChunk(Compressed);
				InUpstreamCache				  = true;
			}
			else
			{
				ZEN_WARN("got uncompressed upstream cache payload");
			}
		}
	}

	if (!Payload)
	{
		ZEN_DEBUG("MISS - '{}/{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, Ref.PayloadId, ToString(Request.AcceptContentType()));
		m_CacheStats.MissCount++;
		return Request.WriteResponse(HttpResponseCode::NotFound);
	}

	ZEN_DEBUG("HIT - '{}/{}/{}' {} '{}' ({})",
			  Ref.BucketSegment,
			  Ref.HashKey,
			  Ref.PayloadId,
			  NiceBytes(Payload.Size()),
			  ToString(Payload.GetContentType()),
			  InUpstreamCache ? "UPSTREAM" : "LOCAL");

	m_CacheStats.HitCount++;
	if (InUpstreamCache)
	{
		m_CacheStats.UpstreamHitCount++;
	}

	if ((Policy & CachePolicy::SkipData) == CachePolicy::SkipData)
	{
		Request.WriteResponse(HttpResponseCode::OK);
	}
	else
	{
		Request.WriteResponse(HttpResponseCode::OK, HttpContentType::kBinary, Payload);
	}
}

void
HttpStructuredCacheService::HandlePutCachePayload(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy Policy)
{
	// Note: Individual cache payloads are not propagated upstream until a valid cache record has been stored
	ZEN_UNUSED(Policy);

	IoBuffer Body = Request.ReadPayload();

	if (!Body || Body.Size() == 0)
	{
		return Request.WriteResponse(HttpResponseCode::BadRequest);
	}

	Body.SetContentType(Request.RequestContentType());

	/*IoHash			 ChunkHash	=*/ IoHash::HashBuffer(Body);
	CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(Body));

	if (!Compressed)
	{
		return Request.WriteResponse(HttpResponseCode::BadRequest, HttpContentType::kText, "Attachments must be compressed"sv);
	}

	if (IoHash::FromBLAKE3(Compressed.GetRawHash()) != Ref.PayloadId)
	{
		return Request.WriteResponse(HttpResponseCode::BadRequest, HttpContentType::kText, "Payload ID does not match attachment hash"sv);
	}

	CidStore::InsertResult Result = m_CidStore.AddChunk(Compressed);

	ZEN_DEBUG("PUT - '{}/{}/{}' {} '{}' ({})",
			  Ref.BucketSegment,
			  Ref.HashKey,
			  Ref.PayloadId,
			  NiceBytes(Body.Size()),
			  ToString(Body.GetContentType()),
			  Result.New ? "NEW" : "OLD");

	const HttpResponseCode ResponseCode = Result.New ? HttpResponseCode::Created : HttpResponseCode::OK;

	Request.WriteResponse(ResponseCode);
}

bool
HttpStructuredCacheService::ValidateKeyUri(HttpServerRequest& Request, CacheRef& OutRef)
{
	std::string_view			Key				  = Request.RelativeUri();
	std::string_view::size_type BucketSplitOffset = Key.find_first_of('/');

	if (BucketSplitOffset == std::string_view::npos)
	{
		return false;
	}

	OutRef.BucketSegment = ToLower(Key.substr(0, BucketSplitOffset));

	if (!std::all_of(begin(OutRef.BucketSegment), end(OutRef.BucketSegment), [](const char c) { return std::isalnum(c); }))
	{
		return false;
	}

	std::string_view HashSegment;
	std::string_view PayloadSegment;

	std::string_view::size_type PayloadSplitOffset = Key.find_last_of('/');

	// We know there is a slash so no need to check for npos return

	if (PayloadSplitOffset == BucketSplitOffset)
	{
		// Basic cache record lookup
		HashSegment = Key.substr(BucketSplitOffset + 1);
	}
	else
	{
		// Cache record + payload lookup
		HashSegment	   = Key.substr(BucketSplitOffset + 1, PayloadSplitOffset - BucketSplitOffset - 1);
		PayloadSegment = Key.substr(PayloadSplitOffset + 1);
	}

	if (HashSegment.size() != IoHash::StringLength)
	{
		return false;
	}

	if (!PayloadSegment.empty() && PayloadSegment.size() == IoHash::StringLength)
	{
		const bool IsOk = ParseHexBytes(PayloadSegment.data(), PayloadSegment.size(), OutRef.PayloadId.Hash);

		if (!IsOk)
		{
			return false;
		}
	}
	else
	{
		OutRef.PayloadId = IoHash::Zero;
	}

	const bool IsOk = ParseHexBytes(HashSegment.data(), HashSegment.size(), OutRef.HashKey.Hash);

	if (!IsOk)
	{
		return false;
	}

	return true;
}

void
HttpStructuredCacheService::HandleRpcRequest(zen::HttpServerRequest& Request)
{
	switch (Request.RequestVerb())
	{
		case HttpVerb::kPost:
			{
				const HttpContentType ContentType = Request.RequestContentType();
				const HttpContentType AcceptType  = Request.AcceptContentType();

				if (ContentType != HttpContentType::kCbObject || AcceptType != HttpContentType::kCbPackage)
				{
					return Request.WriteResponse(HttpResponseCode::BadRequest);
				}

				Request.WriteResponseAsync(
					[this, RpcRequest = zen::LoadCompactBinaryObject(Request.ReadPayload())](HttpServerRequest& AsyncRequest) {
						const std::string_view Method = RpcRequest["Method"sv].AsString();
						if (Method == "GetCacheRecords"sv)
						{
							HandleRpcGetCacheRecords(AsyncRequest, RpcRequest);
						}
						else if (Method == "GetCachePayloads"sv)
						{
							HandleRpcGetCachePayloads(AsyncRequest, RpcRequest);
						}
						else
						{
							AsyncRequest.WriteResponse(HttpResponseCode::BadRequest);
						}
					});
			}
			break;
		default:
			Request.WriteResponse(HttpResponseCode::BadRequest);
			break;
	}
}

void
HttpStructuredCacheService::HandleRpcGetCacheRecords(zen::HttpServerRequest& Request, CbObjectView RpcRequest)
{
	ZEN_TRACE_CPU("Z$::RpcGetCacheRecords");

	CbPackage			  RpcResponse;
	CacheRecordPolicy	  Policy;
	CbObjectView		  Params = RpcRequest["Params"sv].AsObjectView();
	std::vector<CacheKey> CacheKeys;
	std::vector<IoBuffer> CacheValues;
	std::vector<size_t>	  UpstreamRequests;

	ZEN_ASSERT(RpcRequest["Method"sv].AsString() == "GetCacheRecords"sv);

	CacheRecordPolicy::Load(Params["Policy"sv].AsObjectView(), Policy);

	const bool PartialOnError  = Policy.HasRecordPolicy(CachePolicy::PartialOnError);
	const bool SkipAttachments = Policy.HasRecordPolicy(CachePolicy::SkipAttachments);
	const bool QueryRemote	   = Policy.HasRecordPolicy(CachePolicy::QueryRemote) && m_UpstreamCache;

	for (CbFieldView KeyView : Params["CacheKeys"sv])
	{
		CbObjectView KeyObject = KeyView.AsObjectView();
		CacheKeys.push_back(CacheKey::Create(KeyObject["Bucket"sv].AsString(), KeyObject["Hash"sv].AsHash()));
	}

	if (CacheKeys.empty())
	{
		return Request.WriteResponse(HttpResponseCode::BadRequest);
	}

	CacheValues.resize(CacheKeys.size());

	for (size_t KeyIndex = 0; const CacheKey& Key : CacheKeys)
	{
		ZenCacheValue CacheValue;
		uint32_t	  MissingCount = 0;

		if (m_CacheStore.Get(Key.Bucket, Key.Hash, CacheValue))
		{
			CbObjectView CacheRecord(CacheValue.Value.Data());
			CacheRecord.IterateAttachments([this, SkipAttachments, &MissingCount, &RpcResponse](CbFieldView AttachmentHash) {
				if (SkipAttachments && MissingCount == 0)
				{
					if (!m_CidStore.ContainsChunk(AttachmentHash.AsHash()))
					{
						MissingCount++;
					}
				}
				else
				{
					if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
					{
						ZEN_ASSERT(Chunk.GetSize() > 0);
						RpcResponse.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Chunk))));
					}
					else
					{
						MissingCount++;
					}
				}
			});
		}

		if (CacheValue.Value && (MissingCount == 0 || PartialOnError))
		{
			ZEN_DEBUG("HIT - '{}/{}' {} '{}' (LOCAL) {}",
					  Key.Bucket,
					  Key.Hash,
					  NiceBytes(CacheValue.Value.Size()),
					  ToString(CacheValue.Value.GetContentType()),
					  MissingCount ? "(PARTIAl)" : ""sv);

			CacheValues[KeyIndex] = std::move(CacheValue.Value);
			m_CacheStats.HitCount++;
		}
		else if (QueryRemote)
		{
			UpstreamRequests.push_back(KeyIndex);
		}
		else
		{
			ZEN_DEBUG("MISS - '{}/{}' {}", Key.Bucket, Key.Hash, MissingCount ? "(PARTIAl)"sv : ""sv);
			m_CacheStats.MissCount++;
		}

		++KeyIndex;
	}

	if (!UpstreamRequests.empty() && m_UpstreamCache)
	{
		const auto OnCacheRecordGetComplete =
			[this, &CacheKeys, &CacheValues, &RpcResponse, PartialOnError, SkipAttachments](CacheRecordGetCompleteParams&& Params) {
				ZEN_ASSERT(Params.KeyIndex < CacheValues.size());

				IoBuffer		CacheValue;
				AttachmentCount Count;

				if (Params.Record)
				{
					Params.Record.IterateAttachments([this, &RpcResponse, SkipAttachments, &Params, &Count](CbFieldView HashView) {
						if (const CbAttachment* Attachment = Params.Package.FindAttachment(HashView.AsHash()))
						{
							if (CompressedBuffer Compressed = Attachment->AsCompressedBinary())
							{
								auto InsertResult = m_CidStore.AddChunk(Compressed);
								if (InsertResult.New)
								{
									Count.New++;
								}
								Count.Valid++;

								if (!SkipAttachments)
								{
									RpcResponse.AddAttachment(CbAttachment(Compressed));
								}
							}
							else
							{
								ZEN_DEBUG("Uncompressed payload '{}' from upstream cache record '{}/{}'",
										  HashView.AsHash(),
										  Params.Key.Bucket,
										  Params.Key.Hash);
								Count.Invalid++;
							}
						}
						else if (m_CidStore.ContainsChunk(HashView.AsHash()))
						{
							Count.Valid++;
						}
						Count.Total++;
					});

					if ((Count.Valid == Count.Total) || PartialOnError)
					{
						CacheValue = CbObject::Clone(Params.Record).GetBuffer().AsIoBuffer();
					}
				}

				if (CacheValue)
				{
					ZEN_DEBUG("HIT - '{}/{}' {} '{}' attachments '{}/{}/{}' (new/valid/total) (UPSTREAM)",
							  Params.Key.Bucket,
							  Params.Key.Hash,
							  NiceBytes(CacheValue.GetSize()),
							  ToString(HttpContentType::kCbPackage),
							  Count.New,
							  Count.Valid,
							  Count.Total);

					CacheValue.SetContentType(ZenContentType::kCbObject);

					CacheValues[Params.KeyIndex] = CacheValue;
					m_CacheStore.Put(Params.Key.Bucket, Params.Key.Hash, {.Value = CacheValue});

					m_CacheStats.HitCount++;
					m_CacheStats.UpstreamHitCount++;
				}
				else
				{
					const bool IsPartial = Count.Valid != Count.Total;
					ZEN_DEBUG("MISS - '{}/{}' {}", Params.Key.Bucket, Params.Key.Hash, IsPartial ? "(partial)"sv : ""sv);
					m_CacheStats.MissCount++;
				}
			};

		m_UpstreamCache->GetCacheRecords(CacheKeys, UpstreamRequests, Policy, std::move(OnCacheRecordGetComplete));
	}

	CbObjectWriter ResponseObject;

	ResponseObject.BeginArray("Result"sv);
	for (const IoBuffer& Value : CacheValues)
	{
		if (Value)
		{
			CbObjectView Record(Value.Data());
			ResponseObject << Record;
		}
		else
		{
			ResponseObject.AddNull();
		}
	}
	ResponseObject.EndArray();

	RpcResponse.SetObject(ResponseObject.Save());

	BinaryWriter MemStream;
	RpcResponse.Save(MemStream);

	Request.WriteResponse(HttpResponseCode::OK,
						  HttpContentType::kCbPackage,
						  IoBuffer(IoBuffer::Wrap, MemStream.GetData(), MemStream.GetSize()));
}

void
HttpStructuredCacheService::HandleRpcGetCachePayloads(zen::HttpServerRequest& Request, CbObjectView RpcRequest)
{
	ZEN_TRACE_CPU("Z$::RpcGetCachePayloads");

	ZEN_ASSERT(RpcRequest["Method"sv].AsString() == "GetCachePayloads"sv);

	std::vector<CacheChunkRequest> ChunkRequests;
	std::vector<size_t>			   UpstreamRequests;
	std::vector<IoBuffer>		   Chunks;
	CbObjectView				   Params = RpcRequest["Params"sv].AsObjectView();

	for (CbFieldView RequestView : Params["ChunkRequests"sv])
	{
		CbObjectView   RequestObject = RequestView.AsObjectView();
		CbObjectView   KeyObject	 = RequestObject["Key"sv].AsObjectView();
		const CacheKey Key			 = CacheKey::Create(KeyObject["Bucket"sv].AsString(), KeyObject["Hash"sv].AsHash());
		const IoHash   ChunkId		 = RequestObject["ChunkId"sv].AsHash();
		const Oid	   PayloadId	 = RequestObject["PayloadId"sv].AsObjectId();
		const uint64_t RawOffset	 = RequestObject["RawOffset"sv].AsUInt64();
		const uint64_t RawSize		 = RequestObject["RawSize"sv].AsUInt64();
		const uint32_t ChunkPolicy	 = RequestObject["Policy"sv].AsUInt32();

		ChunkRequests.emplace_back(Key, ChunkId, PayloadId, RawOffset, RawSize, static_cast<CachePolicy>(ChunkPolicy));
	}

	if (ChunkRequests.empty())
	{
		return Request.WriteResponse(HttpResponseCode::BadRequest);
	}

	Chunks.resize(ChunkRequests.size());

	// Unreal uses a 12 byte ID to address cache record payloads. When the uncompressed hash (ChunkId)
	// is missing, load the cache record and try to find the raw hash from the payload ID.
	{
		const auto GetChunkIdFromPayloadId = [](CbObjectView Record, const Oid& PayloadId) -> IoHash {
			if (CbObjectView ValueObject = Record["Value"sv].AsObjectView())
			{
				const Oid Id = ValueObject["Id"sv].AsObjectId();
				if (Id == PayloadId)
				{
					return ValueObject["RawHash"sv].AsHash();
				}
			}

			for (CbFieldView ValueView : Record["Values"sv])
			{
				CbObjectView ValueObject = ValueView.AsObjectView();
				const Oid	 Id			 = ValueObject["Id"sv].AsObjectId();

				if (Id == PayloadId)
				{
					return ValueObject["RawHash"sv].AsHash();
				}
			}

			for (CbFieldView AttachmentView : Record["Attachments"sv])
			{
				CbObjectView AttachmentObject = AttachmentView.AsObjectView();
				const Oid	 Id				  = AttachmentObject["Id"sv].AsObjectId();

				if (Id == PayloadId)
				{
					return AttachmentObject["RawHash"sv].AsHash();
				}
			}

			return IoHash::Zero;
		};

		CacheKey CurrentKey = CacheKey::Empty;
		IoBuffer CurrentRecordBuffer;

		for (CacheChunkRequest& ChunkRequest : ChunkRequests)
		{
			if (ChunkRequest.ChunkId != IoHash::Zero)
			{
				continue;
			}

			if (ChunkRequest.Key != CurrentKey)
			{
				CurrentKey = ChunkRequest.Key;

				ZenCacheValue CacheValue;
				if (m_CacheStore.Get(CurrentKey.Bucket, CurrentKey.Hash, CacheValue))
				{
					CurrentRecordBuffer = CacheValue.Value;
				}
			}

			if (CurrentRecordBuffer)
			{
				ChunkRequest.ChunkId = GetChunkIdFromPayloadId(CbObjectView(CurrentRecordBuffer.GetData()), ChunkRequest.PayloadId);
			}
		}
	}

	for (size_t RequestIndex = 0; const CacheChunkRequest& ChunkRequest : ChunkRequests)
	{
		const bool QueryLocal  = (ChunkRequest.Policy & CachePolicy::QueryLocal) == CachePolicy::QueryLocal;
		const bool QueryRemote = (ChunkRequest.Policy & CachePolicy::QueryRemote) == CachePolicy::QueryRemote;

		if (QueryLocal)
		{
			if (IoBuffer Chunk = m_CidStore.FindChunkByCid(ChunkRequest.ChunkId))
			{
				ZEN_ASSERT(Chunk.GetSize() > 0);

				ZEN_DEBUG("HIT - '{}/{}/{}' {} '{}' ({})",
						  ChunkRequest.Key.Bucket,
						  ChunkRequest.Key.Hash,
						  ChunkRequest.ChunkId,
						  NiceBytes(Chunk.Size()),
						  ToString(Chunk.GetContentType()),
						  "LOCAL");

				Chunks[RequestIndex] = Chunk;
				m_CacheStats.HitCount++;
			}
			else if (QueryRemote)
			{
				UpstreamRequests.push_back(RequestIndex);
			}
			else
			{
				ZEN_DEBUG("MISS - '{}/{}/{}'", ChunkRequest.Key.Bucket, ChunkRequest.Key.Hash, ChunkRequest.ChunkId);
				m_CacheStats.MissCount++;
			}
		}
		else
		{
			ZEN_DEBUG("SKIP - '{}/{}/{}'", ChunkRequest.Key.Bucket, ChunkRequest.Key.Hash, ChunkRequest.ChunkId);
		}

		++RequestIndex;
	}

	if (!UpstreamRequests.empty() && m_UpstreamCache)
	{
		const auto OnCachePayloadGetComplete = [this, &ChunkRequests, &Chunks](CachePayloadGetCompleteParams&& Params) {
			if (CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(Params.Payload)))
			{
				m_CidStore.AddChunk(Compressed);

				ZEN_DEBUG("HIT - '{}/{}/{}' {} ({})",
						  Params.Request.Key.Bucket,
						  Params.Request.Key.Hash,
						  Params.Request.ChunkId,
						  NiceBytes(Params.Payload.GetSize()),
						  "UPSTREAM");

				ZEN_ASSERT(Params.RequestIndex < Chunks.size());
				Chunks[Params.RequestIndex] = std::move(Params.Payload);

				m_CacheStats.HitCount++;
				m_CacheStats.UpstreamHitCount++;
			}
			else
			{
				ZEN_DEBUG("MISS - '{}/{}/{}'", Params.Request.Key.Bucket, Params.Request.Key.Hash, Params.Request.ChunkId);
				m_CacheStats.MissCount++;
			}
		};

		m_UpstreamCache->GetCachePayloads(ChunkRequests, UpstreamRequests, std::move(OnCachePayloadGetComplete));
	}

	CbPackage	   RpcResponse;
	CbObjectWriter ResponseObject;

	ResponseObject.BeginArray("Result"sv);

	for (size_t ChunkIndex = 0; ChunkIndex < Chunks.size(); ++ChunkIndex)
	{
		if (Chunks[ChunkIndex])
		{
			ResponseObject << ChunkRequests[ChunkIndex].ChunkId;
			RpcResponse.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(std::move(Chunks[ChunkIndex])))));
		}
		else
		{
			ResponseObject << IoHash::Zero;
		}
	}
	ResponseObject.EndArray();

	RpcResponse.SetObject(ResponseObject.Save());

	BinaryWriter MemStream;
	RpcResponse.Save(MemStream);

	Request.WriteResponse(HttpResponseCode::OK,
						  HttpContentType::kCbPackage,
						  IoBuffer(IoBuffer::Wrap, MemStream.GetData(), MemStream.GetSize()));
}

void
HttpStructuredCacheService::HandleStatsRequest(zen::HttpServerRequest& Request)
{
	CbObjectWriter Cbo;

	EmitSnapshot("requests", m_HttpRequests, Cbo);
	EmitSnapshot("upstream_gets", m_UpstreamGetRequestTiming, Cbo);

	const uint64_t HitCount			= m_CacheStats.HitCount;
	const uint64_t UpstreamHitCount = m_CacheStats.UpstreamHitCount;
	const uint64_t MissCount		= m_CacheStats.MissCount;
	const uint64_t TotalCount		= HitCount + MissCount;

	const CasStoreSize	CasSize	  = m_CidStore.CasSize();
	const GcStorageSize CacheSize = m_CacheStore.StorageSize();

	Cbo.BeginObject("cache");
	Cbo.BeginObject("size");
	Cbo << "disk" << CacheSize.DiskSize;
	Cbo << "memory" << CacheSize.MemorySize;
	Cbo.EndObject();
	Cbo << "upstream_ratio" << (HitCount > 0 ? (double(UpstreamHitCount) / double(HitCount)) : 0.0);
	Cbo << "hits" << HitCount << "misses" << MissCount;
	Cbo << "hit_ratio" << (TotalCount > 0 ? (double(HitCount) / double(TotalCount)) : 0.0);
	Cbo << "upstream_hits" << m_CacheStats.UpstreamHitCount;
	Cbo << "upstream_ratio" << (HitCount > 0 ? (double(UpstreamHitCount) / double(HitCount)) : 0.0);
	Cbo.EndObject();

	if (m_UpstreamCache)
	{
		Cbo.BeginObject("upstream");
		m_UpstreamCache->GetStatus(Cbo);
		Cbo.EndObject();
	}

	Cbo.BeginObject("cas");
	Cbo.BeginObject("size");
	Cbo << "tiny" << CasSize.TinySize;
	Cbo << "small" << CasSize.SmallSize;
	Cbo << "large" << CasSize.LargeSize;
	Cbo << "total" << CasSize.TotalSize;
	Cbo.EndObject();
	Cbo.EndObject();

	Request.WriteResponse(HttpResponseCode::OK, Cbo.Save());
}

void
HttpStructuredCacheService::HandleStatusRequest(zen::HttpServerRequest& Request)
{
	CbObjectWriter Cbo;
	Cbo << "ok" << true;
	Request.WriteResponse(HttpResponseCode::OK, Cbo.Save());
}

}  // namespace zen