Try upload missing payloads and finalize Jupiter cache ref.

1 files changed, 96 insertions, 7 deletions

diff --git a/zenserver/upstream/jupiter.cpp b/zenserver/upstream/jupiter.cpp
index b93635e76..0397ddaa0 100644
--- a/zenserver/upstream/jupiter.cpp
+++ b/zenserver/upstream/jupiter.cpp
@@ -299,13 +299,16 @@ CloudCacheSession::PutDerivedData(std::string_view BucketId, const IoHash& Key,
 	return PutDerivedData(BucketId, Key.ToHexString(), DerivedData);
 }
 
-CloudCacheResult
+PutRefResult
 CloudCacheSession::PutRef(std::string_view BucketId, const IoHash& Key, IoBuffer Ref, ZenContentType RefType)
 {
 	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 	if (!AccessToken.IsValid())
 	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
+		PutRefResult Result;
+		Result.ErrorCode = 401;
+		Result.Reason	 = "Invalid access token"sv;
+		return Result;
 	}
 
 	IoHash Hash = IoHash::HashBuffer(Ref.Data(), Ref.Size());
@@ -328,16 +331,102 @@ CloudCacheSession::PutRef(std::string_view BucketId, const IoHash& Key, IoBuffer
 
 	if (Response.error)
 	{
-		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
+		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))
 	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
+		PutRefResult Result;
+		Result.ErrorCode = 401;
+		Result.Reason	 = "Invalid access token"sv;
+		return Result;
 	}
 
-	return {.Bytes			= Response.uploaded_bytes,
-			.ElapsedSeconds = Response.elapsed,
-			.Success		= (Response.status_code == 200 || Response.status_code == 201)};
+	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)
+{
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
+	if (!AccessToken.IsValid())
+	{
+		FinalizeRefResult Result;
+		Result.ErrorCode = 401;
+		Result.Reason	 = "Invalid access token"sv;
+		return Result;
+	}
+
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
+		<< Key.ToHexString() << "/finalize/" << RefHash.ToHexString();
+
+	cpr::Session& Session = m_SessionState->Session;
+
+	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"}});
+
+	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