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

#include "jupiter.h"

#include "diag/formatters.h"
#include "diag/logging.h"

#include <zencore/compactbinary.h>
#include <zencore/iobuffer.h>
#include <zencore/iohash.h>
#include <zencore/string.h>
#include <zencore/thread.h>
#include <zencore/trace.h>

ZEN_THIRD_PARTY_INCLUDES_START
#include <cpr/cpr.h>
#include <fmt/format.h>
ZEN_THIRD_PARTY_INCLUDES_END

#if ZEN_PLATFORM_WINDOWS
#	pragma comment(lib, "Crypt32.lib")
#	pragma comment(lib, "Wldap32.lib")
#endif

#include <json11.hpp>

using namespace std::literals;

namespace zen {

namespace detail {
	struct CloudCacheSessionState
	{
		CloudCacheSessionState(CloudCacheClient& Client) : m_Client(Client) {}

		const CloudCacheAccessToken& GetAccessToken(bool RefreshToken)
		{
			if (RefreshToken)
			{
				m_AccessToken = m_Client.AcquireAccessToken();
			}

			return m_AccessToken;
		}

		cpr::Session& GetSession() { return m_Session; }

		void Reset(std::chrono::milliseconds ConnectTimeout, std::chrono::milliseconds Timeout)
		{
			m_Session.SetBody({});
			m_Session.SetHeader({});
			m_Session.SetConnectTimeout(ConnectTimeout);
			m_Session.SetTimeout(Timeout);
		}

	private:
		friend class zen::CloudCacheClient;

		CloudCacheClient&	  m_Client;
		CloudCacheAccessToken m_AccessToken;
		cpr::Session		  m_Session;
	};

}  // namespace detail

CloudCacheSession::CloudCacheSession(CloudCacheClient* CacheClient) : m_Log(CacheClient->Logger()), m_CacheClient(CacheClient)
{
	m_SessionState = m_CacheClient->AllocSessionState();
}

CloudCacheSession::~CloudCacheSession()
{
	m_CacheClient->FreeSessionState(m_SessionState);
}

CloudCacheResult
CloudCacheSession::Authenticate()
{
	const bool					 RefreshToken = true;
	const CloudCacheAccessToken& AccessToken  = GetAccessToken(RefreshToken);

	return {.Success = AccessToken.IsValid()};
}

CloudCacheResult
CloudCacheSession::GetDerivedData(std::string_view BucketId, std::string_view Key)
{
	ZEN_TRACE_CPU("HordeClient::GetDerivedData");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/c/ddc/" << m_CacheClient->DdcNamespace() << "/" << BucketId << "/" << Key;

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});

	cpr::Response Response = Session.Get();
	ZEN_DEBUG("GET {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	const bool	   Success = Response.status_code == 200;
	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();

	return {.Response = Buffer, .Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
}

CloudCacheResult
CloudCacheSession::GetDerivedData(std::string_view BucketId, const IoHash& Key)
{
	return GetDerivedData(BucketId, Key.ToHexString());
}

CloudCacheResult
CloudCacheSession::GetRef(std::string_view BucketId, const IoHash& Key, ZenContentType RefType)
{
	const std::string ContentType = RefType == ZenContentType::kCbObject ? "application/x-ue-cb" : "application/octet-stream";

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
		<< Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", ContentType}});
	Session.SetOption(cpr::Body{});

	cpr::Response Response = Session.Get();
	ZEN_DEBUG("GET {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	const bool	   Success = Response.status_code == 200;
	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();

	return {.Response = Buffer, .Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
}

CloudCacheResult
CloudCacheSession::GetBlob(const IoHash& Key)
{
	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
	Session.SetOption(cpr::Body{});

	cpr::Response Response = Session.Get();
	ZEN_DEBUG("GET {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	const bool	   Success = Response.status_code == 200;
	const IoBuffer Buffer =
		Success && Response.text.size() > 0 ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();

	return {.Response = Buffer, .Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
}

CloudCacheResult
CloudCacheSession::GetCompressedBlob(const IoHash& Key)
{
	ZEN_TRACE_CPU("HordeClient::GetCompressedBlob");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-comp"}});
	Session.SetOption(cpr::Body{});

	cpr::Response Response = Session.Get();
	ZEN_DEBUG("GET {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	const bool	   Success = Response.status_code == 200;
	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();

	return {.Response = Buffer, .Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
}

CloudCacheResult
CloudCacheSession::GetObject(const IoHash& Key)
{
	ZEN_TRACE_CPU("HordeClient::GetObject");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
	Session.SetOption(cpr::Body{});

	cpr::Response Response = Session.Get();
	ZEN_DEBUG("GET {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	const bool	   Success = Response.status_code == 200;
	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();

	return {.Response = Buffer, .Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
}

CloudCacheResult
CloudCacheSession::PutDerivedData(std::string_view BucketId, std::string_view Key, IoBuffer DerivedData)
{
	ZEN_TRACE_CPU("HordeClient::PutDerivedData");

	IoHash Hash = IoHash::HashBuffer(DerivedData.Data(), DerivedData.Size());

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/c/ddc/" << m_CacheClient->DdcNamespace() << "/" << BucketId << "/" << Key;

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value},
								  {"X-Jupiter-IoHash", Hash.ToHexString()},
								  {"Content-Type", "application/octet-stream"}});
	Session.SetBody(cpr::Body{(const char*)DerivedData.Data(), DerivedData.Size()});

	cpr::Response Response = Session.Put();
	ZEN_DEBUG("PUT {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	return {.Bytes			= Response.uploaded_bytes,
			.ElapsedSeconds = Response.elapsed,
			.Success		= (Response.status_code == 200 || Response.status_code == 201)};
}

CloudCacheResult
CloudCacheSession::PutDerivedData(std::string_view BucketId, const IoHash& Key, IoBuffer DerivedData)
{
	return PutDerivedData(BucketId, Key.ToHexString(), DerivedData);
}

PutRefResult
CloudCacheSession::PutRef(std::string_view BucketId, const IoHash& Key, IoBuffer Ref, ZenContentType RefType)
{
	ZEN_TRACE_CPU("HordeClient::PutRef");

	IoHash Hash = IoHash::HashBuffer(Ref.Data(), Ref.Size());

	const std::string ContentType = RefType == ZenContentType::kCbObject ? "application/x-ue-cb" : "application/octet-stream";

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
		<< Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(
		cpr::Header{{"Authorization", AccessToken.Value}, {"X-Jupiter-IoHash", Hash.ToHexString()}, {"Content-Type", ContentType}});
	Session.SetBody(cpr::Body{(const char*)Ref.Data(), Ref.Size()});

	cpr::Response Response = Session.Put();
	ZEN_DEBUG("PUT {}", Response);

	if (Response.error)
	{
		PutRefResult Result;
		Result.ErrorCode = static_cast<int32_t>(Response.error.code);
		Result.Reason	 = std::move(Response.error.message);
		return Result;
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		PutRefResult Result;
		Result.ErrorCode = 401;
		Result.Reason	 = "Invalid access token"sv;
		return Result;
	}

	PutRefResult Result;
	Result.Success		  = (Response.status_code == 200 || Response.status_code == 201);
	Result.Bytes		  = Response.uploaded_bytes;
	Result.ElapsedSeconds = Response.elapsed;

	if (Result.Success)
	{
		std::string	 JsonError;
		json11::Json Json = json11::Json::parse(Response.text, JsonError);
		if (JsonError.empty())
		{
			json11::Json::array Needs = Json["needs"].array_items();
			for (const auto& Need : Needs)
			{
				Result.Needs.emplace_back(IoHash::FromHexString(Need.string_value()));
			}
		}
	}

	return Result;
}

FinalizeRefResult
CloudCacheSession::FinalizeRef(std::string_view BucketId, const IoHash& Key, const IoHash& RefHash)
{
	ZEN_TRACE_CPU("HordeClient::FinalizeRef");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
		<< Key.ToHexString() << "/finalize/" << RefHash.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value},
								  {"X-Jupiter-IoHash", RefHash.ToHexString()},
								  {"Content-Type", "application/x-ue-cb"}});
	Session.SetBody(cpr::Body{});

	cpr::Response Response = Session.Post();
	ZEN_DEBUG("POST {}", Response);

	if (Response.error)
	{
		FinalizeRefResult Result;
		Result.ErrorCode = static_cast<int32_t>(Response.error.code);
		Result.Reason	 = std::move(Response.error.message);
		return Result;
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		FinalizeRefResult Result;
		Result.ErrorCode = 401;
		Result.Reason	 = "Invalid access token"sv;
		return Result;
	}

	FinalizeRefResult Result;
	Result.Success		  = (Response.status_code == 200 || Response.status_code == 201);
	Result.Bytes		  = Response.uploaded_bytes;
	Result.ElapsedSeconds = Response.elapsed;

	if (Result.Success)
	{
		std::string	 JsonError;
		json11::Json Json = json11::Json::parse(Response.text, JsonError);
		if (JsonError.empty())
		{
			json11::Json::array Needs = Json["needs"].array_items();
			for (const auto& Need : Needs)
			{
				Result.Needs.emplace_back(IoHash::FromHexString(Need.string_value()));
			}
		}
	}

	return Result;
}

CloudCacheResult
CloudCacheSession::PutBlob(const IoHash& Key, IoBuffer Blob)
{
	ZEN_TRACE_CPU("HordeClient::PutBlob");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/octet-stream"}});
	Session.SetBody(cpr::Body{(const char*)Blob.Data(), Blob.Size()});

	cpr::Response Response = Session.Put();
	ZEN_DEBUG("PUT {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	return {.Bytes			= Response.uploaded_bytes,
			.ElapsedSeconds = Response.elapsed,
			.Success		= (Response.status_code == 200 || Response.status_code == 201)};
}

CloudCacheResult
CloudCacheSession::PutCompressedBlob(const IoHash& Key, IoBuffer Blob)
{
	ZEN_TRACE_CPU("HordeClient::PutCompressedBlob");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-comp"}});
	Session.SetBody(cpr::Body{(const char*)Blob.Data(), Blob.Size()});

	cpr::Response Response = Session.Put();
	ZEN_DEBUG("PUT {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	return {.Bytes			= Response.uploaded_bytes,
			.ElapsedSeconds = Response.elapsed,
			.Success		= (Response.status_code == 200 || Response.status_code == 201)};
}

CloudCacheResult
CloudCacheSession::PutObject(const IoHash& Key, IoBuffer Object)
{
	ZEN_TRACE_CPU("HordeClient::PutObject");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
	Session.SetBody(cpr::Body{(const char*)Object.Data(), Object.Size()});

	cpr::Response Response = Session.Put();
	ZEN_DEBUG("PUT {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	return {.Bytes			= Response.uploaded_bytes,
			.ElapsedSeconds = Response.elapsed,
			.Success		= (Response.status_code == 200 || Response.status_code == 201)};
}

CloudCacheResult
CloudCacheSession::RefExists(std::string_view BucketId, const IoHash& Key)
{
	ZEN_TRACE_CPU("HordeClient::RefExists");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
		<< Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}});
	Session.SetOption(cpr::Body{});

	cpr::Response Response = Session.Head();
	ZEN_DEBUG("HEAD {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	return {.ElapsedSeconds = Response.elapsed, .Success = Response.status_code == 200};
}

GetObjectReferencesResult
CloudCacheSession::GetObjectReferences(const IoHash& Key)
{
	ZEN_TRACE_CPU("HordeClient::GetObjectReferences");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString()
		<< "/references";

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
	Session.SetOption(cpr::Body{});

	cpr::Response Response = Session.Get();
	ZEN_DEBUG("GET {}", Response);

	if (Response.error)
	{
		return {CloudCacheResult{.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message}};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {CloudCacheResult{.ErrorCode = 401, .Reason = std::string("Invalid access token")}};
	}

	GetObjectReferencesResult Result{
		CloudCacheResult{.Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Response.status_code == 200}};

	if (Result.Success)
	{
		IoBuffer	   Buffer			  = IoBuffer(zen::IoBuffer::Wrap, Response.text.data(), Response.text.size());
		const CbObject ReferencesResponse = LoadCompactBinaryObject(Buffer);
		for (auto& Item : ReferencesResponse["references"sv])
		{
			Result.References.insert(Item.AsHash());
		}
	}

	return Result;
}

CloudCacheResult
CloudCacheSession::BlobExists(const IoHash& Key)
{
	return CacheTypeExists("blobs"sv, Key);
}

CloudCacheResult
CloudCacheSession::CompressedBlobExists(const IoHash& Key)
{
	return CacheTypeExists("compressed-blobs"sv, Key);
}

CloudCacheResult
CloudCacheSession::ObjectExists(const IoHash& Key)
{
	return CacheTypeExists("objects"sv, Key);
}

CloudCacheExistsResult
CloudCacheSession::BlobExists(const std::set<IoHash>& Keys)
{
	return CacheTypeExists("blobs"sv, Keys);
}

CloudCacheExistsResult
CloudCacheSession::CompressedBlobExists(const std::set<IoHash>& Keys)
{
	return CacheTypeExists("compressed-blobs"sv, Keys);
}

CloudCacheExistsResult
CloudCacheSession::ObjectExists(const std::set<IoHash>& Keys)
{
	return CacheTypeExists("objects"sv, Keys);
}

CloudCacheResult
CloudCacheSession::PostComputeTasks(IoBuffer TasksData)
{
	ZEN_TRACE_CPU("HordeClient::PostComputeTasks");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << m_CacheClient->ComputeCluster();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
	Session.SetBody(cpr::Body{(const char*)TasksData.Data(), TasksData.Size()});

	cpr::Response Response = Session.Post();
	ZEN_DEBUG("POST {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	return {.Bytes = Response.uploaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Response.status_code == 200};
}

CloudCacheResult
CloudCacheSession::GetComputeUpdates(std::string_view ChannelId, const uint32_t WaitSeconds)
{
	ZEN_TRACE_CPU("HordeClient::GetComputeUpdates");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << m_CacheClient->ComputeCluster() << "/updates/" << ChannelId
		<< "?wait=" << WaitSeconds;

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
	Session.SetOption(cpr::Body{});

	cpr::Response Response = Session.Post();
	ZEN_DEBUG("POST {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	const bool	   Success = Response.status_code == 200;
	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();

	return {.Response = Buffer, .Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
}

std::vector<IoHash>
CloudCacheSession::Filter(std::string_view BucketId, const std::vector<IoHash>& ChunkHashes)
{
	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl();
	Uri << "/api/v1/s/" << m_CacheClient->DdcNamespace();

	ZEN_UNUSED(BucketId, ChunkHashes);

	return {};
}

cpr::Session&
CloudCacheSession::GetSession()
{
	return m_SessionState->GetSession();
}

CloudCacheAccessToken
CloudCacheSession::GetAccessToken(bool RefreshToken)
{
	return m_SessionState->GetAccessToken(RefreshToken);
}

bool
CloudCacheSession::VerifyAccessToken(long StatusCode)
{
	return StatusCode != 401;
}

CloudCacheResult
CloudCacheSession::CacheTypeExists(std::string_view TypeId, const IoHash& Key)
{
	ZEN_TRACE_CPU("HordeClient::CacheTypeExists");

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/" << TypeId << "/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}});
	Session.SetOption(cpr::Body{});

	cpr::Response Response = Session.Head();
	ZEN_DEBUG("HEAD {}", Response);

	if (Response.error)
	{
		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
	}

	return {.ElapsedSeconds = Response.elapsed, .Success = Response.status_code == 200};
}

CloudCacheExistsResult
CloudCacheSession::CacheTypeExists(std::string_view TypeId, const std::set<IoHash>& Keys)
{
	ZEN_TRACE_CPU("HordeClient::CacheTypeExists");

	ExtendableStringBuilder<256> Body;
	Body << "[";
	for (const auto& Key : Keys)
	{
		Body << (Body.Size() != 1 ? ",\"" : "\"") << Key.ToHexString() << "\"";
	}
	Body << "]";

	ExtendableStringBuilder<256> Uri;
	Uri << m_CacheClient->ServiceUrl() << "/api/v1/" << TypeId << "/" << m_CacheClient->BlobStoreNamespace() << "/exist";

	cpr::Session&				 Session	 = GetSession();
	const CloudCacheAccessToken& AccessToken = GetAccessToken();

	Session.SetOption(cpr::Url{Uri.c_str()});
	Session.SetOption(
		cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}, {"Content-Type", "application/json"}});
	Session.SetOption(cpr::Body(Body.ToString()));

	cpr::Response Response = Session.Post();
	ZEN_DEBUG("POST {}", Response);

	if (Response.error)
	{
		return {CloudCacheResult{.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message}};
	}
	else if (!VerifyAccessToken(Response.status_code))
	{
		return {CloudCacheResult{.ErrorCode = 401, .Reason = std::string("Invalid access token")}};
	}

	CloudCacheExistsResult Result{
		CloudCacheResult{.Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Response.status_code == 200}};

	if (Result.Success)
	{
		IoBuffer	   Buffer		  = IoBuffer(zen::IoBuffer::Wrap, Response.text.data(), Response.text.size());
		const CbObject ExistsResponse = LoadCompactBinaryObject(Buffer);
		for (auto& Item : ExistsResponse["needs"sv])
		{
			Result.Needs.insert(Item.AsHash());
		}
	}

	return Result;
}

/**
 * An access token provider that holds a token that will never change.
 */
class StaticTokenProvider final : public CloudCacheTokenProvider
{
public:
	StaticTokenProvider(CloudCacheAccessToken Token) : m_Token(std::move(Token)) {}

	virtual ~StaticTokenProvider() = default;

	virtual CloudCacheAccessToken AcquireAccessToken() final override { return m_Token; }

private:
	CloudCacheAccessToken m_Token;
};

std::unique_ptr<CloudCacheTokenProvider>
CloudCacheTokenProvider::CreateFromStaticToken(CloudCacheAccessToken Token)
{
	return std::make_unique<StaticTokenProvider>(std::move(Token));
}

class OAuthClientCredentialsTokenProvider final : public CloudCacheTokenProvider
{
public:
	OAuthClientCredentialsTokenProvider(const CloudCacheTokenProvider::OAuthClientCredentialsParams& Params)
	{
		m_Url		   = std::string(Params.Url);
		m_ClientId	   = std::string(Params.ClientId);
		m_ClientSecret = std::string(Params.ClientSecret);
	}

	virtual ~OAuthClientCredentialsTokenProvider() = default;

	virtual CloudCacheAccessToken AcquireAccessToken() final override
	{
		using namespace std::chrono;

		std::string Body =
			fmt::format("client_id={}&scope=cache_access&grant_type=client_credentials&client_secret={}", m_ClientId, m_ClientSecret);

		cpr::Response Response =
			cpr::Post(cpr::Url{m_Url}, cpr::Header{{"Content-Type", "application/x-www-form-urlencoded"}}, cpr::Body{std::move(Body)});

		if (Response.error || Response.status_code != 200)
		{
			return {};
		}

		std::string	 JsonError;
		json11::Json Json = json11::Json::parse(Response.text, JsonError);

		if (JsonError.empty() == false)
		{
			return {};
		}

		std::string						 Token			  = Json["access_token"].string_value();
		int64_t							 ExpiresInSeconds = static_cast<int64_t>(Json["expires_in"].int_value());
		CloudCacheAccessToken::TimePoint ExpireTime		  = CloudCacheAccessToken::Clock::now() + seconds(ExpiresInSeconds);

		return {.Value = fmt::format("Bearer {}", Token), .ExpireTime = ExpireTime};
	}

private:
	std::string m_Url;
	std::string m_ClientId;
	std::string m_ClientSecret;
};

std::unique_ptr<CloudCacheTokenProvider>
CloudCacheTokenProvider::CreateFromOAuthClientCredentials(const OAuthClientCredentialsParams& Params)
{
	return std::make_unique<OAuthClientCredentialsTokenProvider>(Params);
}

class CallbackTokenProvider final : public CloudCacheTokenProvider
{
public:
	CallbackTokenProvider(std::function<CloudCacheAccessToken()>&& Callback) : m_Callback(std::move(Callback)) {}

	virtual ~CallbackTokenProvider() = default;

	virtual CloudCacheAccessToken AcquireAccessToken() final override { return m_Callback(); }

private:
	std::function<CloudCacheAccessToken()> m_Callback;
};

std::unique_ptr<CloudCacheTokenProvider>
CloudCacheTokenProvider::CreateFromCallback(std::function<CloudCacheAccessToken()>&& Callback)
{
	return std::make_unique<CallbackTokenProvider>(std::move(Callback));
}

CloudCacheClient::CloudCacheClient(const CloudCacheClientOptions& Options, std::unique_ptr<CloudCacheTokenProvider> TokenProvider)
: m_Log(zen::logging::Get("jupiter"))
, m_ServiceUrl(Options.ServiceUrl)
, m_DdcNamespace(Options.DdcNamespace)
, m_BlobStoreNamespace(Options.BlobStoreNamespace)
, m_ComputeCluster(Options.ComputeCluster)
, m_ConnectTimeout(Options.ConnectTimeout)
, m_Timeout(Options.Timeout)
, m_TokenProvider(std::move(TokenProvider))
{
	ZEN_ASSERT(m_TokenProvider.get() != nullptr);
}

CloudCacheClient::~CloudCacheClient()
{
	RwLock::ExclusiveLockScope _(m_SessionStateLock);

	for (auto State : m_SessionStateCache)
	{
		delete State;
	}
}

CloudCacheAccessToken
CloudCacheClient::AcquireAccessToken()
{
	ZEN_TRACE_CPU("HordeClient::AcquireAccessToken");

	return m_TokenProvider->AcquireAccessToken();
}

detail::CloudCacheSessionState*
CloudCacheClient::AllocSessionState()
{
	detail::CloudCacheSessionState* State = nullptr;

	bool IsTokenValid = false;

	{
		RwLock::ExclusiveLockScope _(m_SessionStateLock);

		if (m_SessionStateCache.empty() == false)
		{
			State		 = m_SessionStateCache.front();
			IsTokenValid = State->m_AccessToken.IsValid();

			m_SessionStateCache.pop_front();
		}
	}

	if (State == nullptr)
	{
		State = new detail::CloudCacheSessionState(*this);
	}

	State->Reset(m_ConnectTimeout, m_Timeout);

	if (IsTokenValid == false)
	{
		State->m_AccessToken = m_TokenProvider->AcquireAccessToken();
	}

	return State;
}

void
CloudCacheClient::FreeSessionState(detail::CloudCacheSessionState* State)
{
	RwLock::ExclusiveLockScope _(m_SessionStateLock);
	m_SessionStateCache.push_front(State);
}

}  // namespace zen