path: root/src/zenserver/projectstore/jupiterremoteprojectstore.cpp

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

#include "jupiterremoteprojectstore.h"

#include <zencore/compress.h>
#include <zencore/fmtutils.h>

#include <upstream/jupiter.h>
#include <zenhttp/auth/authmgr.h>

namespace zen {

using namespace std::literals;

class JupiterRemoteStore : public RemoteProjectStore
{
public:
	JupiterRemoteStore(Ref<CloudCacheClient>&& CloudClient,
					   std::string_view		   Namespace,
					   std::string_view		   Bucket,
					   const IoHash&		   Key,
					   bool					   ForceDisableBlocks,
					   bool					   ForceDisableTempBlocks)
	: m_CloudClient(std::move(CloudClient))
	, m_Namespace(Namespace)
	, m_Bucket(Bucket)
	, m_Key(Key)
	{
		if (ForceDisableBlocks)
		{
			m_EnableBlocks = false;
		}
		if (ForceDisableTempBlocks)
		{
			m_UseTempBlocks = false;
		}
	}

	virtual RemoteStoreInfo GetInfo() const override
	{
		return {.CreateBlocks	   = m_EnableBlocks,
				.UseTempBlockFiles = m_UseTempBlocks,
				.Description	   = fmt::format("[cloud] {} as {}/{}/{}"sv, m_CloudClient->ServiceUrl(), m_Namespace, m_Bucket, m_Key)};
	}

	virtual SaveResult SaveContainer(const IoBuffer& Payload) override
	{
		const int32_t MaxAttempts = 3;
		PutRefResult  PutResult;
		{
			CloudCacheSession Session(m_CloudClient.Get());
			for (int32_t Attempt = 0; Attempt < MaxAttempts && !PutResult.Success; Attempt++)
			{
				PutResult = Session.PutRef(m_Namespace, m_Bucket, m_Key, Payload, ZenContentType::kCbObject);
			}
		}

		SaveResult Result{ConvertResult(PutResult), {PutResult.Needs.begin(), PutResult.Needs.end()}, IoHash::HashBuffer(Payload)};
		if (Result.ErrorCode)
		{
			Result.Reason = fmt::format("Failed saving oplog container to {}/{}/{}/{}. Reason: '{}'",
										m_CloudClient->ServiceUrl(),
										m_Namespace,
										m_Bucket,
										m_Key,
										Result.Reason);
		}
		return Result;
	}

	virtual SaveAttachmentResult SaveAttachment(const CompositeBuffer& Payload, const IoHash& RawHash) override
	{
		const int32_t	 MaxAttempts = 3;
		CloudCacheResult PutResult;
		{
			CloudCacheSession Session(m_CloudClient.Get());
			for (int32_t Attempt = 0; Attempt < MaxAttempts && !PutResult.Success; Attempt++)
			{
				PutResult = Session.PutCompressedBlob(m_Namespace, RawHash, Payload);
			}
		}

		SaveAttachmentResult Result{ConvertResult(PutResult)};
		if (Result.ErrorCode)
		{
			Result.Reason = fmt::format("Failed saving oplog attachment to {}/{}/{}. Reason: '{}'",
										m_CloudClient->ServiceUrl(),
										m_Namespace,
										RawHash,
										Result.Reason);
		}
		return Result;
	}

	virtual SaveAttachmentsResult SaveAttachments(const std::vector<SharedBuffer>& Chunks) override
	{
		SaveAttachmentsResult Result;
		for (const SharedBuffer& Chunk : Chunks)
		{
			CompressedBuffer	 Compressed	 = CompressedBuffer::FromCompressedNoValidate(Chunk.AsIoBuffer());
			SaveAttachmentResult ChunkResult = SaveAttachment(Compressed.GetCompressed(), Compressed.DecodeRawHash());
			if (ChunkResult.ErrorCode)
			{
				return SaveAttachmentsResult{ChunkResult};
			}
		}
		return Result;
	}

	virtual Result FinalizeContainer(const IoHash& RawHash) override
	{
		const int32_t	 MaxAttempts = 3;
		CloudCacheResult FinalizeResult;
		{
			CloudCacheSession Session(m_CloudClient.Get());
			for (int32_t Attempt = 0; Attempt < MaxAttempts && !FinalizeResult.Success; Attempt++)
			{
				FinalizeResult = Session.FinalizeRef(m_Namespace, m_Bucket, m_Key, RawHash);
			}
		}
		Result Result{ConvertResult(FinalizeResult)};
		if (Result.ErrorCode)
		{
			Result.Reason = fmt::format("Failed finalizing oplog container to {}/{}/{}/{}. Reason: '{}'",
										m_CloudClient->ServiceUrl(),
										m_Namespace,
										m_Bucket,
										m_Key,
										Result.Reason);
		}
		return Result;
	}

	virtual LoadContainerResult LoadContainer() override
	{
		const int32_t	 MaxAttempts = 3;
		CloudCacheResult GetResult;
		{
			CloudCacheSession Session(m_CloudClient.Get());
			for (int32_t Attempt = 0; Attempt < MaxAttempts && !GetResult.Success; Attempt++)
			{
				GetResult = Session.GetRef(m_Namespace, m_Bucket, m_Key, ZenContentType::kCbObject);
			}
		}

		if (GetResult.ErrorCode || !GetResult.Success)
		{
			LoadContainerResult Result{ConvertResult(GetResult)};
			Result.Reason = fmt::format("Failed fetching oplog container from {}/{}/{}/{}. Reason: '{}'",
										m_CloudClient->ServiceUrl(),
										m_Namespace,
										m_Bucket,
										m_Key,
										Result.Reason);
			return Result;
		}

		CbObject ContainerObject = LoadCompactBinaryObject(GetResult.Response);
		if (!ContainerObject)
		{
			return LoadContainerResult{
				RemoteProjectStore::Result{.ErrorCode	   = gsl::narrow<int32_t>(HttpResponseCode::InternalServerError),
										   .ElapsedSeconds = GetResult.ElapsedSeconds,
										   .Reason = fmt::format("The ref {}/{}/{}/{} is not formatted as a compact binary object"sv,
																 m_CloudClient->ServiceUrl(),
																 m_Namespace,
																 m_Bucket,
																 m_Key)},
				{}};
		}
		return LoadContainerResult{ConvertResult(GetResult), std::move(ContainerObject)};
	}

	virtual LoadAttachmentResult LoadAttachment(const IoHash& RawHash) override
	{
		const int32_t	 MaxAttempts = 3;
		CloudCacheResult GetResult;
		{
			CloudCacheSession Session(m_CloudClient.Get());
			for (int32_t Attempt = 0; Attempt < MaxAttempts && !GetResult.Success; Attempt++)
			{
				GetResult = Session.GetCompressedBlob(m_Namespace, RawHash);
			}
		}
		LoadAttachmentResult Result{ConvertResult(GetResult), std::move(GetResult.Response)};
		if (GetResult.ErrorCode)
		{
			Result.Reason = fmt::format("Failed fetching oplog attachment from {}/{}/{}. Reason: '{}'",
										m_CloudClient->ServiceUrl(),
										m_Namespace,
										RawHash,
										Result.Reason);
		}
		return Result;
	}

	virtual LoadAttachmentsResult LoadAttachments(const std::vector<IoHash>& RawHashes) override
	{
		LoadAttachmentsResult Result;
		for (const IoHash& Hash : RawHashes)
		{
			LoadAttachmentResult ChunkResult = LoadAttachment(Hash);
			if (ChunkResult.ErrorCode)
			{
				return LoadAttachmentsResult{ChunkResult};
			}
			ZEN_DEBUG("Loaded attachment in {}", NiceTimeSpanMs(static_cast<uint64_t>(ChunkResult.ElapsedSeconds * 1000)));
			Result.Chunks.emplace_back(
				std::pair<IoHash, CompressedBuffer>{Hash, CompressedBuffer::FromCompressedNoValidate(std::move(ChunkResult.Bytes))});
		}
		return Result;
	}

private:
	static Result ConvertResult(const CloudCacheResult& Response)
	{
		std::string Text;
		int32_t		ErrorCode = 0;
		if (Response.ErrorCode != 0)
		{
			ErrorCode = Response.ErrorCode;
		}
		else if (!Response.Success)
		{
			ErrorCode = gsl::narrow<int32_t>(HttpResponseCode::InternalServerError);
			if (Response.Response.GetContentType() == ZenContentType::kText)
			{
				Text =
					std::string(reinterpret_cast<const std::string::value_type*>(Response.Response.GetData()), Response.Response.GetSize());
			}
		}
		return {.ErrorCode = ErrorCode, .ElapsedSeconds = Response.ElapsedSeconds, .Reason = Response.Reason, .Text = Text};
	}

	Ref<CloudCacheClient> m_CloudClient;
	const std::string	  m_Namespace;
	const std::string	  m_Bucket;
	const IoHash		  m_Key;
	bool				  m_EnableBlocks  = true;
	bool				  m_UseTempBlocks = true;
};

std::unique_ptr<RemoteProjectStore>
CreateJupiterRemoteStore(const JupiterRemoteStoreOptions& Options)
{
	std::string Url = Options.Url;
	if (Url.find("://"sv) == std::string::npos)
	{
		// Assume https URL
		Url = fmt::format("https://{}"sv, Url);
	}
	CloudCacheClientOptions ClientOptions{.Name			  = "Remote store"sv,
										  .ServiceUrl	  = Url,
										  .ConnectTimeout = std::chrono::milliseconds(2000),
										  .Timeout		  = std::chrono::milliseconds(60000)};
	// 1) Access token as parameter in request
	// 2) Environment variable (different win vs linux/mac)
	// 3) openid-provider (assumes oidctoken.exe -Zen true has been run with matching Options.OpenIdProvider

	std::unique_ptr<CloudCacheTokenProvider> TokenProvider;
	if (!Options.AccessToken.empty())
	{
		TokenProvider = CloudCacheTokenProvider::CreateFromCallback([AccessToken = "Bearer " + Options.AccessToken]() {
			return CloudCacheAccessToken{.Value = AccessToken, .ExpireTime = GcClock::TimePoint::max()};
		});
	}
	else
	{
		TokenProvider =
			CloudCacheTokenProvider::CreateFromCallback([&AuthManager = Options.AuthManager, OpenIdProvider = Options.OpenIdProvider]() {
				AuthMgr::OpenIdAccessToken Token = AuthManager.GetOpenIdAccessToken(OpenIdProvider.empty() ? "Default" : OpenIdProvider);
				return CloudCacheAccessToken{.Value = Token.AccessToken, .ExpireTime = Token.ExpireTime};
			});
	}

	Ref<CloudCacheClient> CloudClient(new CloudCacheClient(ClientOptions, std::move(TokenProvider)));

	std::unique_ptr<RemoteProjectStore> RemoteStore = std::make_unique<JupiterRemoteStore>(std::move(CloudClient),
																						   Options.Namespace,
																						   Options.Bucket,
																						   Options.Key,
																						   Options.ForceDisableBlocks,
																						   Options.ForceDisableTempBlocks);
	return RemoteStore;
}

}  // namespace zen