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

#pragma once

#include <zencore/stats.h>
#include <zenhttp/httpserver.h>
#include <zenhttp/httpstats.h>
#include <zenhttp/httpstatus.h>
#include <zenstore/cache/cacherpc.h>
#include <zenutil/cache/cache.h>
#include <zenutil/openprocesscache.h>

#include <memory>
#include <vector>

namespace zen {

struct CacheChunkRequest;
struct CacheKeyRequest;
struct PutRequestData;

class CidStore;
class CbObjectView;
class DiskWriteBlocker;
class HttpStructuredCacheService;
class ScrubContext;
class UpstreamCache;
class ZenCacheStore;

enum class CachePolicy : uint32_t;
enum class RpcAcceptOptions : uint16_t;

namespace cache {
	class IRpcRequestReplayer;
	class IRpcRequestRecorder;
	namespace detail {
		struct RecordBody;
		struct ChunkRequest;
	}  // namespace detail
}  // namespace cache

/**
 * 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,
							   const DiskWriteBlocker* InDiskWriteBlocker,
							   OpenProcessCache&	   InOpenProcessCache);
	~HttpStructuredCacheService();

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

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

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

	void HandleCacheRecordRequest(HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
	void HandleGetCacheRecord(HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
	void HandlePutCacheRecord(HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
	void HandleCacheChunkRequest(HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
	void HandleGetCacheChunk(HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
	void HandlePutCacheChunk(HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
	void HandleRpcRequest(HttpServerRequest& Request, std::string_view UriNamespace);
	void HandleDetailsRequest(HttpServerRequest& Request);

	void		 HandleCacheRequest(HttpServerRequest& Request);
	void		 HandleCacheNamespaceRequest(HttpServerRequest& Request, std::string_view Namespace);
	void		 HandleCacheBucketRequest(HttpServerRequest& Request, std::string_view Namespace, std::string_view Bucket);
	virtual void HandleStatsRequest(HttpServerRequest& Request) override;
	virtual void HandleStatusRequest(HttpServerRequest& Request) override;

	bool AreDiskWritesAllowed() const;

	LoggerRef				 Log() { return m_Log; }
	LoggerRef				 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;
	const DiskWriteBlocker*	 m_DiskWriteBlocker = nullptr;
	OpenProcessCache&		 m_OpenProcessCache;
	CacheRpcHandler			 m_RpcHandler;

	void ReplayRequestRecorder(const CacheRequestContext& Context, cache::IRpcRequestReplayer& Replayer, uint32_t ThreadCount);

	// This exists to avoid taking locks when recording is not enabled
	std::atomic_bool m_RequestRecordingEnabled{false};

	// This lock should be taken in SHARED mode when calling into the recorder,
	// and taken in EXCLUSIVE mode whenever the recorder is created or destroyed
	RwLock										m_RequestRecordingLock;
	std::unique_ptr<cache::IRpcRequestRecorder> m_RequestRecorder;
};

}  // namespace zen