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

#pragma once

#include <zencore/stats.h>
#include <zenhttp/httpserver.h>

#include "monitoring/httpstats.h"
#include "monitoring/httpstatus.h"

#include <memory>
#include <vector>

namespace spdlog {
class logger;
}

namespace zen {

struct CacheChunkRequest;
struct CacheKeyRequest;
class CasStore;
class CidStore;
class CbObjectView;
struct PutRequestData;
class ScrubContext;
class UpstreamCache;
class ZenCacheStore;
enum class CachePolicy : uint32_t;

namespace cache::detail {
	struct RecordBody;
	struct ChunkRequest;
}  // namespace cache::detail

/**
 * Structured cache service. Imposes constraints on keys, supports blobs and
 * structured values
 *
 * Keys are structured as:
 *
 *   {BucketId}/{KeyHash}
 *
 * Where BucketId is a lower-case alphanumeric string, and KeyHash is a 40-character
 * hexadecimal sequence. The hash value may be derived in any number of ways, it's
 * up to the application to pick an approach.
 *
 * Values may be structured or unstructured. Structured values are encoded using Unreal
 * Engine's compact binary encoding (see CbObject)
 *
 * Additionally, attachments may be addressed as:
 *
 *   {BucketId}/{KeyHash}/{ValueHash}
 *
 * Where the two initial components are the same as for the main endpoint
 *
 * The storage strategy is as follows:
 *
 *  - Structured values are stored in a dedicated backing store per bucket
 *  - Unstructured values and attachments are stored in the CAS pool
 *
 */

class HttpStructuredCacheService : public HttpService, public IHttpStatsProvider, public IHttpStatusProvider
{
public:
	HttpStructuredCacheService(ZenCacheStore&	  InCacheStore,
							   CidStore&		  InCidStore,
							   HttpStatsService&  StatsService,
							   HttpStatusService& StatusService,
							   UpstreamCache&	  UpstreamCache);
	~HttpStructuredCacheService();

	virtual const char* BaseUri() const override;
	virtual void		HandleRequest(zen::HttpServerRequest& Request) override;

	void Flush();
	void Scrub(ScrubContext& Ctx);

private:
	struct CacheRef
	{
		std::string Namespace;
		std::string BucketSegment;
		IoHash		HashKey;
		IoHash		ValueContentId;
	};

	struct CacheStats
	{
		std::atomic_uint64_t HitCount{};
		std::atomic_uint64_t UpstreamHitCount{};
		std::atomic_uint64_t MissCount{};
	};
	enum class PutResult
	{
		Success,
		Fail,
		Invalid,
	};

	[[nodiscard]] bool ValidateKeyUri(zen::HttpServerRequest& Request, CacheRef& OutRef);
	void			   HandleCacheRecordRequest(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
	void			   HandleGetCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
	void			   HandlePutCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
	void			   HandleCacheValueRequest(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
	void			   HandleGetCacheValue(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
	void			   HandlePutCacheValue(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
	void			   HandleRpcRequest(zen::HttpServerRequest& Request);
	void			   HandleRpcPutCacheRecords(zen::HttpServerRequest& Request, const CbPackage& BatchRequest);
	void			   HandleRpcGetCacheRecords(zen::HttpServerRequest& Request, CbObjectView BatchRequest);
	void			   HandleRpcPutCacheValues(zen::HttpServerRequest& Request, const CbPackage& BatchRequest);
	void			   HandleRpcGetCacheValues(zen::HttpServerRequest& Request, CbObjectView BatchRequest);
	void			   HandleRpcGetCacheChunks(zen::HttpServerRequest& Request, CbObjectView BatchRequest);
	void			   HandleCacheBucketRequest(zen::HttpServerRequest& Request, std::string_view Bucket);
	virtual void	   HandleStatsRequest(zen::HttpServerRequest& Request) override;
	virtual void	   HandleStatusRequest(zen::HttpServerRequest& Request) override;
	PutResult		   PutCacheRecord(PutRequestData& Request, const CbPackage* Package);

	/** HandleRpcGetCacheChunks Helper: Parse the Body object into RecordValue Requests and Value Requests. */
	bool ParseGetCacheChunksRequest(std::vector<CacheKeyRequest>&			   RecordKeys,
									std::vector<cache::detail::RecordBody>&	   Records,
									std::vector<CacheChunkRequest>&			   RequestKeys,
									std::vector<cache::detail::ChunkRequest>&  Requests,
									std::vector<cache::detail::ChunkRequest*>& RecordRequests,
									std::vector<cache::detail::ChunkRequest*>& ValueRequests,
									CbObjectView							   RpcRequest);
	/** HandleRpcGetCacheChunks Helper: Load records to get ContentId for RecordRequests, and load their payloads if they exist locally. */
	void GetLocalCacheRecords(std::vector<CacheKeyRequest>&				 RecordKeys,
							  std::vector<cache::detail::RecordBody>&	 Records,
							  std::vector<cache::detail::ChunkRequest*>& RecordRequests,
							  std::vector<CacheChunkRequest*>&			 OutUpstreamChunks);
	/** HandleRpcGetCacheChunks Helper: For ValueRequests, load their payloads if they exist locally. */
	void GetLocalCacheValues(std::vector<cache::detail::ChunkRequest*>& ValueRequests, std::vector<CacheChunkRequest*>& OutUpstreamChunks);
	/** HandleRpcGetCacheChunks Helper: Load payloads from upstream that did not exist locally. */
	void GetUpstreamCacheChunks(std::vector<CacheChunkRequest*>&		  UpstreamChunks,
								std::vector<CacheChunkRequest>&			  RequestKeys,
								std::vector<cache::detail::ChunkRequest>& Requests);
	/** HandleRpcGetCacheChunks Helper: Send response message containing all chunk results. */
	void WriteGetCacheChunksResponse(std::vector<cache::detail::ChunkRequest>& Requests, zen::HttpServerRequest& HttpRequest);

	spdlog::logger&			 Log() { return m_Log; }
	spdlog::logger&			 m_Log;
	ZenCacheStore&			 m_CacheStore;
	HttpStatsService&		 m_StatsService;
	HttpStatusService&		 m_StatusService;
	CidStore&				 m_CidStore;
	UpstreamCache&			 m_UpstreamCache;
	uint64_t				 m_LastScrubTime = 0;
	metrics::OperationTiming m_HttpRequests;
	metrics::OperationTiming m_UpstreamGetRequestTiming;
	CacheStats				 m_CacheStats;
};

/** Recognize both kBinary and kCompressedBinary as kCompressedBinary for structured cache value keys.
 * We need this until the content type is preserved for kCompressedBinary when passing to and from upstream servers. */
inline bool
IsCompressedBinary(ZenContentType Type)
{
	return Type == ZenContentType::kBinary || Type == ZenContentType::kCompressedBinary;
}

}  // namespace zen