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

#include <zenutil/cloud/s3client.h>

#include <zenutil/cloud/imdscredentials.h>
#include <zenutil/cloud/minioprocess.h>
#include <zenutil/cloud/s3response.h>

#include <zencore/except_fmt.h>
#include <zencore/iobuffer.h>
#include <zencore/memoryview.h>
#include <zencore/string.h>
#include <zencore/testing.h>

ZEN_THIRD_PARTY_INCLUDES_START
#include <fmt/format.h>
ZEN_THIRD_PARTY_INCLUDES_END

#include <algorithm>

namespace zen {

std::string_view S3GetObjectResult::NotFoundErrorText = "Not found";

S3Client::S3Client(const S3ClientOptions& Options)
: m_Log(logging::Get("s3"))
, m_Builder(Options.Region, Options.BucketName, Options.Endpoint, Options.PathStyle)
, m_Credentials(Options.Credentials)
, m_CredentialProvider(Options.CredentialProvider)
, m_HttpClient(std::string(m_Builder.Endpoint()), Options.HttpSettings)
, m_Verbose(Options.HttpSettings.Verbose)
{
	ZEN_INFO("S3 client configured for bucket '{}' in region '{}' (endpoint: {}, {})",
			 m_Builder.BucketName(),
			 m_Builder.Region(),
			 m_HttpClient.GetBaseUri(),
			 m_Builder.PathStyle() ? "path-style" : "virtual-hosted");
}

S3Client::~S3Client() = default;

SigV4Credentials
S3Client::GetCurrentCredentials()
{
	if (m_CredentialProvider)
	{
		SigV4Credentials Creds = m_CredentialProvider->GetCredentials();
		if (!Creds.AccessKeyId.empty())
		{
			// Update stored credentials atomically so callers see a consistent snapshot.
			// The builder's signing-key cache is keyed on (DateStamp, AccessKeyId), so it
			// self-invalidates on the next Sign() call when either rotates.
			RwLock::ExclusiveLockScope _(m_CredentialsLock);
			m_Credentials = Creds;
			return Creds;
		}
		// IMDS returned empty credentials; fall back to the last known-good credentials.
		RwLock::SharedLockScope _(m_CredentialsLock);
		return m_Credentials;
	}
	return m_Credentials;
}

std::string
S3Client::BuildNoCredentialsError(std::string Context)
{
	std::string Err = fmt::format("{}: no credentials available", Context);
	ZEN_WARN("{}", Err);
	return Err;
}

S3Result
S3Client::PutObject(std::string_view Key, IoBuffer Content)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 PUT '{}' failed", Key); !Err.empty())
	{
		return S3Result{std::move(Err)};
	}

	std::string Path = m_Builder.KeyToPath(Key);

	// Hash the payload
	std::string PayloadHash = Sha256ToHex(ComputeSha256(Content.GetData(), Content.GetSize()));

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "PUT", Path, "", PayloadHash);

	HttpClient::Response Response = m_HttpClient.Put(Path, Content, Headers);
	if (!Response.IsSuccess())
	{
		std::string Err = S3ErrorMessage("S3 PUT failed", Response);
		ZEN_WARN("S3 PUT '{}' failed: {}", Key, Err);
		return S3Result{std::move(Err)};
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 PUT '{}' succeeded ({} bytes)", Key, Content.GetSize());
	}
	return {};
}

S3GetObjectResult
S3Client::GetObject(std::string_view Key, const std::filesystem::path& TempFilePath)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 GET '{}' failed", Key); !Err.empty())
	{
		return S3GetObjectResult{S3Result{std::move(Err)}, {}};
	}

	std::string Path = m_Builder.KeyToPath(Key);

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "GET", Path, "", S3EmptyPayloadHash);

	HttpClient::Response Response = m_HttpClient.Download(Path, TempFilePath, Headers);
	if (!Response.IsSuccess())
	{
		if (Response.StatusCode == HttpResponseCode::NotFound)
		{
			return S3GetObjectResult{S3Result{.Error = std::string(S3GetObjectResult::NotFoundErrorText)}, {}};
		}

		std::string Err = S3ErrorMessage("S3 GET failed", Response);
		ZEN_WARN("S3 GET '{}' failed: {}", Key, Err);
		return S3GetObjectResult{S3Result{std::move(Err)}, {}};
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 GET '{}' succeeded ({} bytes)", Key, Response.ResponsePayload.GetSize());
	}
	return S3GetObjectResult{{}, std::move(Response.ResponsePayload)};
}

S3GetObjectResult
S3Client::GetObjectRange(std::string_view Key, uint64_t RangeStart, uint64_t RangeSize)
{
	ZEN_ASSERT(RangeSize > 0);

	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 GET range '{}' [{}-{}] failed", Key, RangeStart, RangeStart + RangeSize - 1);
		!Err.empty())
	{
		return S3GetObjectResult{S3Result{std::move(Err)}, {}};
	}

	std::string Path = m_Builder.KeyToPath(Key);

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "GET", Path, "", S3EmptyPayloadHash);
	Headers->emplace("Range", fmt::format("bytes={}-{}", RangeStart, RangeStart + RangeSize - 1));

	HttpClient::Response Response = m_HttpClient.Get(Path, Headers);
	if (!Response.IsSuccess())
	{
		if (Response.StatusCode == HttpResponseCode::NotFound)
		{
			return S3GetObjectResult{S3Result{.Error = std::string(S3GetObjectResult::NotFoundErrorText)}, {}};
		}

		std::string Err = S3ErrorMessage("S3 GET range failed", Response);
		ZEN_WARN("S3 GET range '{}' [{}-{}] failed: {}", Key, RangeStart, RangeStart + RangeSize - 1, Err);
		return S3GetObjectResult{S3Result{std::move(Err)}, {}};
	}

	// Callers are expected to request only ranges that lie within the known object size (e.g.
	// by calling HeadObject first). Treat a short read as an error rather than silently
	// returning a truncated buffer - a partial write is more dangerous than a hard failure.
	if (Response.ResponsePayload.GetSize() != RangeSize)
	{
		std::string Err = fmt::format("S3 GET range '{}' [{}-{}] returned {} bytes, expected {}",
									  Key,
									  RangeStart,
									  RangeStart + RangeSize - 1,
									  Response.ResponsePayload.GetSize(),
									  RangeSize);
		ZEN_WARN("{}", Err);
		return S3GetObjectResult{S3Result{std::move(Err)}, {}};
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 GET range '{}' [{}-{}] succeeded ({} bytes)",
				 Key,
				 RangeStart,
				 RangeStart + RangeSize - 1,
				 Response.ResponsePayload.GetSize());
	}
	return S3GetObjectResult{{}, std::move(Response.ResponsePayload)};
}

S3Result
S3Client::DeleteObject(std::string_view Key)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 DELETE '{}' failed", Key); !Err.empty())
	{
		return S3Result{std::move(Err)};
	}

	std::string Path = m_Builder.KeyToPath(Key);

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "DELETE", Path, "", S3EmptyPayloadHash);

	HttpClient::Response Response = m_HttpClient.Delete(Path, Headers);
	if (!Response.IsSuccess())
	{
		std::string Err = S3ErrorMessage("S3 DELETE failed", Response);
		ZEN_WARN("S3 DELETE '{}' failed: {}", Key, Err);
		return S3Result{std::move(Err)};
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 DELETE '{}' succeeded", Key);
	}
	return {};
}

S3Result
S3Client::Touch(std::string_view Key)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 Touch '{}' failed", Key); !Err.empty())
	{
		return S3Result{std::move(Err)};
	}

	std::string Path = m_Builder.KeyToPath(Key);

	// x-amz-copy-source is always "/bucket/key" regardless of addressing style.
	// Key must be URI-encoded except for '/' separators. When source and destination
	// are identical, REPLACE is required; COPY is rejected with InvalidRequest.
	const std::array<std::pair<std::string, std::string>, 2> ExtraSigned{{
		{"x-amz-copy-source", fmt::format("/{}/{}", m_Builder.BucketName(), AwsUriEncode(Key, false))},
		{"x-amz-metadata-directive", "REPLACE"},
	}};

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "PUT", Path, "", S3EmptyPayloadHash, ExtraSigned);

	HttpClient::Response Response = m_HttpClient.Put(Path, IoBuffer{}, Headers);
	if (!Response.IsSuccess())
	{
		std::string Err = S3ErrorMessage("S3 Touch failed", Response);
		ZEN_WARN("S3 Touch '{}' failed: {}", Key, Err);
		return S3Result{std::move(Err)};
	}

	// Copy operations can return HTTP 200 with an error in the XML body.
	std::string_view ResponseBody = Response.AsText();
	std::string_view ErrorCode;
	std::string_view ErrorMessage;
	if (S3ExtractError(ResponseBody, ErrorCode, ErrorMessage))
	{
		std::string Err = fmt::format("S3 Touch '{}' returned error: {} - {}", Key, ErrorCode, ErrorMessage);
		ZEN_WARN("{}", Err);
		return S3Result{std::move(Err)};
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 Touch '{}' succeeded", Key);
	}
	return {};
}

S3HeadObjectResult
S3Client::HeadObject(std::string_view Key)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 HEAD '{}' failed", Key); !Err.empty())
	{
		return S3HeadObjectResult{S3Result{std::move(Err)}, {}, HeadObjectResult::Error};
	}

	std::string Path = m_Builder.KeyToPath(Key);

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "HEAD", Path, "", S3EmptyPayloadHash);

	HttpClient::Response Response = m_HttpClient.Head(Path, Headers);
	if (!Response.IsSuccess())
	{
		if (Response.StatusCode == HttpResponseCode::NotFound)
		{
			return S3HeadObjectResult{{}, {}, HeadObjectResult::NotFound};
		}

		std::string Err = S3ErrorMessage("S3 HEAD failed", Response);
		ZEN_WARN("S3 HEAD '{}' failed: {}", Key, Err);
		return S3HeadObjectResult{S3Result{std::move(Err)}, {}, HeadObjectResult::Error};
	}

	S3ObjectInfo Info;
	Info.Key = std::string(Key);

	if (const std::string* V = S3FindResponseHeader(Response.Header, "content-length"))
	{
		Info.Size = ParseInt<uint64_t>(*V).value_or(0);
	}

	if (const std::string* V = S3FindResponseHeader(Response.Header, "etag"))
	{
		Info.ETag = *V;
	}

	if (const std::string* V = S3FindResponseHeader(Response.Header, "last-modified"))
	{
		Info.LastModified = *V;
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 HEAD '{}' succeeded (size={})", Key, Info.Size);
	}
	return S3HeadObjectResult{{}, std::move(Info), HeadObjectResult::Found};
}

S3ListObjectsResult
S3Client::ListObjects(std::string_view Prefix, uint32_t MaxKeys)
{
	S3ListObjectsResult Result;

	std::string ContinuationToken;

	for (;;)
	{
		SigV4Credentials Credentials;
		if (std::string Err = RequireCredentials(Credentials, "S3 ListObjectsV2 prefix='{}' failed", Prefix); !Err.empty())
		{
			Result.Error = std::move(Err);
			return Result;
		}

		// Build query parameters for ListObjectsV2
		std::vector<std::pair<std::string, std::string>> QueryParams;
		QueryParams.emplace_back("list-type", "2");
		if (!Prefix.empty())
		{
			QueryParams.emplace_back("prefix", std::string(Prefix));
		}
		if (MaxKeys > 0)
		{
			QueryParams.emplace_back("max-keys", fmt::format("{}", MaxKeys));
		}
		if (!ContinuationToken.empty())
		{
			QueryParams.emplace_back("continuation-token", ContinuationToken);
		}

		std::string				CanonicalQS = BuildCanonicalQueryString(std::move(QueryParams));
		std::string				RootPath	= m_Builder.BucketRootPath();
		HttpClient::KeyValueMap Headers		= m_Builder.SignRequest(Credentials, "GET", RootPath, CanonicalQS, S3EmptyPayloadHash);

		std::string			 FullPath = S3BuildRequestPath(RootPath, CanonicalQS);
		HttpClient::Response Response = m_HttpClient.Get(FullPath, Headers);
		if (!Response.IsSuccess())
		{
			std::string Err = S3ErrorMessage("S3 ListObjectsV2 failed", Response);
			ZEN_WARN("S3 ListObjectsV2 prefix='{}' failed: {}", Prefix, Err);
			Result.Error = std::move(Err);
			return Result;
		}

		// Parse the XML response to extract object keys
		std::string_view ResponseBody = Response.AsText();

		// Find all <Contents> elements
		std::string_view Remaining = ResponseBody;
		while (true)
		{
			size_t ContentsStart = Remaining.find("<Contents>");
			if (ContentsStart == std::string_view::npos)
			{
				break;
			}

			size_t ContentsEnd = Remaining.find("</Contents>", ContentsStart);
			if (ContentsEnd == std::string_view::npos)
			{
				break;
			}

			std::string_view ContentsXml = Remaining.substr(ContentsStart, ContentsEnd - ContentsStart + 11);

			S3ObjectInfo Info;
			Info.Key		  = S3DecodeXmlEntities(S3ExtractXmlValue(ContentsXml, "Key"));
			Info.ETag		  = S3DecodeXmlEntities(S3ExtractXmlValue(ContentsXml, "ETag"));
			Info.LastModified = std::string(S3ExtractXmlValue(ContentsXml, "LastModified"));

			std::string_view SizeStr = S3ExtractXmlValue(ContentsXml, "Size");
			if (!SizeStr.empty())
			{
				Info.Size = ParseInt<uint64_t>(SizeStr).value_or(0);
			}

			if (!Info.Key.empty())
			{
				Result.Objects.push_back(std::move(Info));
			}

			Remaining = Remaining.substr(ContentsEnd + 11);
		}

		// Check if there are more pages
		std::string_view IsTruncated = S3ExtractXmlValue(ResponseBody, "IsTruncated");
		if (IsTruncated != "true")
		{
			break;
		}

		std::string_view NextToken = S3ExtractXmlValue(ResponseBody, "NextContinuationToken");
		if (NextToken.empty())
		{
			break;
		}

		ContinuationToken = std::string(NextToken);
		if (m_Verbose)
		{
			ZEN_INFO("S3 ListObjectsV2 prefix='{}' fetching next page ({} objects so far)", Prefix, Result.Objects.size());
		}
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 ListObjectsV2 prefix='{}' returned {} objects", Prefix, Result.Objects.size());
	}
	return Result;
}

//////////////////////////////////////////////////////////////////////////
// Multipart Upload

S3CreateMultipartUploadResult
S3Client::CreateMultipartUpload(std::string_view Key)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 CreateMultipartUpload '{}' failed", Key); !Err.empty())
	{
		return S3CreateMultipartUploadResult{S3Result{std::move(Err)}, {}};
	}

	std::string Path		= m_Builder.KeyToPath(Key);
	std::string CanonicalQS = BuildCanonicalQueryString({{"uploads", ""}});

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "POST", Path, CanonicalQS, S3EmptyPayloadHash);

	std::string			 FullPath = S3BuildRequestPath(Path, CanonicalQS);
	HttpClient::Response Response = m_HttpClient.Post(FullPath, Headers);
	if (!Response.IsSuccess())
	{
		std::string Err = S3ErrorMessage("S3 CreateMultipartUpload failed", Response);
		ZEN_WARN("S3 CreateMultipartUpload '{}' failed: {}", Key, Err);
		return S3CreateMultipartUploadResult{S3Result{std::move(Err)}, {}};
	}

	// Parse UploadId from XML response:
	// <InitiateMultipartUploadResult>
	//   <Bucket>...</Bucket>
	//   <Key>...</Key>
	//   <UploadId>...</UploadId>
	// </InitiateMultipartUploadResult>
	std::string_view ResponseBody = Response.AsText();
	std::string_view UploadId	  = S3ExtractXmlValue(ResponseBody, "UploadId");
	if (UploadId.empty())
	{
		std::string Err = "failed to parse UploadId from CreateMultipartUpload response";
		ZEN_WARN("S3 CreateMultipartUpload '{}': {}", Key, Err);
		return S3CreateMultipartUploadResult{S3Result{std::move(Err)}, {}};
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 CreateMultipartUpload '{}' succeeded (uploadId={})", Key, UploadId);
	}
	return S3CreateMultipartUploadResult{{}, std::string(UploadId)};
}

S3UploadPartResult
S3Client::UploadPart(std::string_view Key, std::string_view UploadId, uint32_t PartNumber, IoBuffer Content)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 UploadPart '{}' part {} failed", Key, PartNumber); !Err.empty())
	{
		return S3UploadPartResult{S3Result{std::move(Err)}, {}};
	}

	std::string Path		= m_Builder.KeyToPath(Key);
	std::string CanonicalQS = BuildCanonicalQueryString({
		{"partNumber", fmt::format("{}", PartNumber)},
		{"uploadId", std::string(UploadId)},
	});

	std::string PayloadHash = Sha256ToHex(ComputeSha256(Content.GetData(), Content.GetSize()));

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "PUT", Path, CanonicalQS, PayloadHash);

	std::string			 FullPath = S3BuildRequestPath(Path, CanonicalQS);
	HttpClient::Response Response = m_HttpClient.Put(FullPath, Content, Headers);
	if (!Response.IsSuccess())
	{
		std::string Err = S3ErrorMessage(fmt::format("S3 UploadPart {} failed", PartNumber), Response);
		ZEN_WARN("S3 UploadPart '{}' part {} failed: {}", Key, PartNumber, Err);
		return S3UploadPartResult{S3Result{std::move(Err)}, {}};
	}

	// Extract ETag from response headers
	const std::string* ETag = S3FindResponseHeader(Response.Header, "etag");
	if (!ETag)
	{
		std::string Err = "S3 UploadPart response missing ETag header";
		ZEN_WARN("S3 UploadPart '{}' part {}: {}", Key, PartNumber, Err);
		return S3UploadPartResult{S3Result{std::move(Err)}, {}};
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 UploadPart '{}' part {} succeeded ({} bytes, etag={})", Key, PartNumber, Content.GetSize(), *ETag);
	}
	return S3UploadPartResult{{}, *ETag};
}

S3Result
S3Client::CompleteMultipartUpload(std::string_view									   Key,
								  std::string_view									   UploadId,
								  const std::vector<std::pair<uint32_t, std::string>>& PartETags)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 CompleteMultipartUpload '{}' failed", Key); !Err.empty())
	{
		return S3Result{std::move(Err)};
	}

	std::string Path		= m_Builder.KeyToPath(Key);
	std::string CanonicalQS = BuildCanonicalQueryString({{"uploadId", std::string(UploadId)}});

	// Build the CompleteMultipartUpload XML payload
	ExtendableStringBuilder<1024> XmlBody;
	XmlBody.Append("<CompleteMultipartUpload>");
	for (const auto& [PartNumber, ETag] : PartETags)
	{
		XmlBody.Append(fmt::format("<Part><PartNumber>{}</PartNumber><ETag>{}</ETag></Part>", PartNumber, ETag));
	}
	XmlBody.Append("</CompleteMultipartUpload>");

	std::string_view XmlView	 = XmlBody.ToView();
	std::string		 PayloadHash = Sha256ToHex(ComputeSha256(XmlView));

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "POST", Path, CanonicalQS, PayloadHash);
	Headers->emplace("Content-Type", "application/xml");

	IoBuffer Payload(IoBuffer::Clone, XmlView.data(), XmlView.size());

	std::string			 FullPath = S3BuildRequestPath(Path, CanonicalQS);
	HttpClient::Response Response = m_HttpClient.Post(FullPath, Payload, Headers);
	if (!Response.IsSuccess())
	{
		std::string Err = S3ErrorMessage("S3 CompleteMultipartUpload failed", Response);
		ZEN_WARN("S3 CompleteMultipartUpload '{}' failed: {}", Key, Err);
		return S3Result{std::move(Err)};
	}

	// Check for error in response body - S3 can return 200 with an error in the XML body
	std::string_view ResponseBody = Response.AsText();
	std::string_view ErrorCode;
	std::string_view ErrorMessage;
	if (S3ExtractError(ResponseBody, ErrorCode, ErrorMessage))
	{
		std::string Err = fmt::format("S3 CompleteMultipartUpload '{}' returned error: {} - {}", Key, ErrorCode, ErrorMessage);
		ZEN_WARN("{}", Err);
		return S3Result{std::move(Err)};
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 CompleteMultipartUpload '{}' succeeded ({} parts)", Key, PartETags.size());
	}
	return {};
}

S3Result
S3Client::AbortMultipartUpload(std::string_view Key, std::string_view UploadId)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 AbortMultipartUpload '{}' failed", Key); !Err.empty())
	{
		return S3Result{std::move(Err)};
	}

	std::string Path		= m_Builder.KeyToPath(Key);
	std::string CanonicalQS = BuildCanonicalQueryString({{"uploadId", std::string(UploadId)}});

	HttpClient::KeyValueMap Headers = m_Builder.SignRequest(Credentials, "DELETE", Path, CanonicalQS, S3EmptyPayloadHash);

	std::string			 FullPath = S3BuildRequestPath(Path, CanonicalQS);
	HttpClient::Response Response = m_HttpClient.Delete(FullPath, Headers);
	if (!Response.IsSuccess())
	{
		std::string Err = S3ErrorMessage("S3 AbortMultipartUpload failed", Response);
		ZEN_WARN("S3 AbortMultipartUpload '{}' failed: {}", Key, Err);
		return S3Result{std::move(Err)};
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 AbortMultipartUpload '{}' succeeded (uploadId={})", Key, UploadId);
	}
	return {};
}

std::string
S3Client::GeneratePresignedGetUrl(std::string_view Key, std::chrono::seconds ExpiresIn)
{
	return GeneratePresignedUrlForMethod(Key, "GET", ExpiresIn);
}

std::string
S3Client::GeneratePresignedPutUrl(std::string_view Key, std::chrono::seconds ExpiresIn)
{
	return GeneratePresignedUrlForMethod(Key, "PUT", ExpiresIn);
}

std::string
S3Client::GeneratePresignedUrlForMethod(std::string_view Key, std::string_view Method, std::chrono::seconds ExpiresIn)
{
	SigV4Credentials Credentials;
	if (std::string Err = RequireCredentials(Credentials, "S3 GeneratePresignedUrl '{}' {} failed", Key, Method); !Err.empty())
	{
		return {};
	}

	std::string Path   = m_Builder.KeyToPath(Key);
	std::string Scheme = "https";

	if (!m_Builder.Endpoint().empty() && m_Builder.Endpoint().starts_with("http://"))
	{
		Scheme = "http";
	}

	return GeneratePresignedUrl(Credentials, Method, Scheme, m_Builder.Host(), Path, m_Builder.Region(), "s3", ExpiresIn);
}

S3Result
S3Client::PutObjectMultipart(std::string_view										 Key,
							 uint64_t												 TotalSize,
							 std::function<IoBuffer(uint64_t Offset, uint64_t Size)> FetchRange,
							 uint64_t												 PartSize)
{
	// If the content fits in a single part, just use PutObject
	if (TotalSize <= PartSize)
	{
		return PutObject(Key, TotalSize > 0 ? FetchRange(0, TotalSize) : IoBuffer{});
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 multipart upload '{}': {} bytes in ~{} parts", Key, TotalSize, (TotalSize + PartSize - 1) / PartSize);
	}

	S3CreateMultipartUploadResult InitResult = CreateMultipartUpload(Key);
	if (!InitResult)
	{
		return S3Result{std::move(InitResult.Error)};
	}

	const std::string& UploadId = InitResult.UploadId;

	// Cleanup helper: AbortMultipartUpload itself can throw on transport failure;
	// inside the catch (...) below that would replace the original exception with
	// a less actionable transport one. Swallow + log.
	auto SafeAbort = [this, &Key, &UploadId]() noexcept {
		try
		{
			AbortMultipartUpload(Key, UploadId);
		}
		catch (const std::exception& Ex)
		{
			ZEN_WARN("S3 AbortMultipartUpload '{}' threw during cleanup: {}", Key, Ex.what());
		}
		catch (...)
		{
			ZEN_WARN("S3 AbortMultipartUpload '{}' threw during cleanup", Key);
		}
	};

	// Sequential upload by design; for parallel multipart use S3AsyncStorage::PutMultipart
	// in the hub hydration path.

	std::vector<std::pair<uint32_t, std::string>> PartETags;
	uint64_t									  Offset	 = 0;
	uint32_t									  PartNumber = 1;

	try
	{
		while (Offset < TotalSize)
		{
			uint64_t		   ThisPartSize = std::min(PartSize, TotalSize - Offset);
			IoBuffer		   PartContent	= FetchRange(Offset, ThisPartSize);
			S3UploadPartResult PartResult	= UploadPart(Key, UploadId, PartNumber, std::move(PartContent));
			if (!PartResult)
			{
				SafeAbort();
				return S3Result{std::move(PartResult.Error)};
			}

			PartETags.emplace_back(PartNumber, std::move(PartResult.ETag));
			Offset += ThisPartSize;
			PartNumber++;
		}

		S3Result CompleteResult = CompleteMultipartUpload(Key, UploadId, PartETags);
		if (!CompleteResult)
		{
			SafeAbort();
			return CompleteResult;
		}
	}
	catch (...)
	{
		SafeAbort();
		throw;
	}

	if (m_Verbose)
	{
		ZEN_INFO("S3 multipart upload '{}' completed ({} parts, {} bytes)", Key, PartETags.size(), TotalSize);
	}
	return {};
}

S3Result
S3Client::PutObjectMultipart(std::string_view Key, IoBuffer Content, uint64_t PartSize)
{
	return PutObjectMultipart(
		Key,
		Content.GetSize(),
		[&Content](uint64_t Offset, uint64_t Size) { return IoBuffer(Content, Offset, Size); },
		PartSize);
}

//////////////////////////////////////////////////////////////////////////
// Tests

#if ZEN_WITH_TESTS

void
s3client_forcelink()
{
}

TEST_SUITE_BEGIN("util.cloud.s3client");

TEST_CASE("s3client.xml_extract")
{
	std::string_view Xml =
		"<Contents><Key>test/file.txt</Key><Size>1234</Size>"
		"<ETag>\"abc123\"</ETag><LastModified>2024-01-01T00:00:00Z</LastModified></Contents>";

	CHECK(S3ExtractXmlValue(Xml, "Key") == "test/file.txt");
	CHECK(S3ExtractXmlValue(Xml, "Size") == "1234");
	CHECK(S3ExtractXmlValue(Xml, "ETag") == "\"abc123\"");
	CHECK(S3ExtractXmlValue(Xml, "LastModified") == "2024-01-01T00:00:00Z");
	CHECK(S3ExtractXmlValue(Xml, "NonExistent") == "");
}

TEST_CASE("s3client.xml_entity_decode")
{
	CHECK(S3DecodeXmlEntities("no entities") == "no entities");
	CHECK(S3DecodeXmlEntities("a&amp;b") == "a&b");
	CHECK(S3DecodeXmlEntities("&lt;tag&gt;") == "<tag>");
	CHECK(S3DecodeXmlEntities("&quot;hello&apos;") == "\"hello'");
	CHECK(S3DecodeXmlEntities("&amp;&amp;") == "&&");
	CHECK(S3DecodeXmlEntities("") == "");

	// Key with entities as S3 would return it
	std::string_view Xml = "<Key>path/file&amp;name&lt;1&gt;.txt</Key>";
	CHECK(S3DecodeXmlEntities(S3ExtractXmlValue(Xml, "Key")) == "path/file&name<1>.txt");
}

TEST_CASE("s3client.path_style_addressing")
{
	// Verify path-style builds /{bucket}/{key} paths
	S3ClientOptions Opts;
	Opts.BucketName					 = "test-bucket";
	Opts.Region						 = "us-east-1";
	Opts.Endpoint					 = "http://localhost:9000";
	Opts.PathStyle					 = true;
	Opts.Credentials.AccessKeyId	 = "minioadmin";
	Opts.Credentials.SecretAccessKey = "minioadmin";

	S3Client Client(Opts);
	CHECK(Client.BucketName() == "test-bucket");
	CHECK(Client.Region() == "us-east-1");
}

TEST_CASE("s3client.virtual_hosted_addressing")
{
	// Verify virtual-hosted style derives endpoint from region + bucket
	S3ClientOptions Opts;
	Opts.BucketName					 = "my-bucket";
	Opts.Region						 = "eu-west-1";
	Opts.PathStyle					 = false;
	Opts.Credentials.AccessKeyId	 = "key";
	Opts.Credentials.SecretAccessKey = "secret";

	S3Client Client(Opts);
	CHECK(Client.BucketName() == "my-bucket");
	CHECK(Client.Region() == "eu-west-1");
}

TEST_CASE("s3requestbuilder.path_style_paths")
{
	S3RequestBuilder Builder("us-east-1", "test-bucket", "http://localhost:9000", /*PathStyle*/ true);

	CHECK(Builder.Region() == "us-east-1");
	CHECK(Builder.BucketName() == "test-bucket");
	CHECK(Builder.PathStyle());
	CHECK(Builder.Endpoint() == "http://localhost:9000");
	CHECK(Builder.Host() == "localhost:9000");
	CHECK(Builder.KeyToPath("foo/bar") == "/test-bucket/foo/bar");
	CHECK(Builder.BucketRootPath() == "/test-bucket/");
}

TEST_CASE("s3requestbuilder.virtual_hosted_paths")
{
	S3RequestBuilder Builder("eu-west-1", "my-bucket", /*Endpoint*/ "", /*PathStyle*/ false);

	CHECK(Builder.Endpoint() == "https://my-bucket.s3.eu-west-1.amazonaws.com");
	CHECK(Builder.Host() == "my-bucket.s3.eu-west-1.amazonaws.com");
	CHECK(Builder.KeyToPath("foo/bar") == "/foo/bar");
	CHECK(Builder.BucketRootPath() == "/");
}

TEST_CASE("s3requestbuilder.derived_endpoint_path_style")
{
	S3RequestBuilder Builder("us-west-2", "another-bucket", /*Endpoint*/ "", /*PathStyle*/ true);
	CHECK(Builder.Endpoint() == "https://s3.us-west-2.amazonaws.com");
	CHECK(Builder.Host() == "s3.us-west-2.amazonaws.com");
}

TEST_CASE("s3requestbuilder.sign_request_headers")
{
	S3RequestBuilder Builder("us-east-1", "bucket", "http://localhost:9000", /*PathStyle*/ true);

	SigV4Credentials Creds;
	Creds.AccessKeyId	  = "AKIDEXAMPLE";
	Creds.SecretAccessKey = "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY";

	HttpClient::KeyValueMap Headers = Builder.SignRequest(Creds, "GET", "/bucket/foo", "", S3EmptyPayloadHash);

	// All four required SigV4 headers present.
	CHECK(Headers->find("Authorization") != Headers->end());
	CHECK(Headers->find("x-amz-date") != Headers->end());
	CHECK(Headers->find("x-amz-content-sha256") != Headers->end());

	// SessionToken absent -> no x-amz-security-token header.
	CHECK(Headers->find("x-amz-security-token") == Headers->end());

	const std::string& Auth = Headers->find("Authorization")->second;
	CHECK(Auth.starts_with("AWS4-HMAC-SHA256 "));
	CHECK(Auth.find("Credential=AKIDEXAMPLE/") != std::string::npos);
	CHECK(Auth.find("/us-east-1/s3/aws4_request") != std::string::npos);
	CHECK(Auth.find("Signature=") != std::string::npos);
}

TEST_CASE("s3requestbuilder.session_token_emits_security_header")
{
	S3RequestBuilder Builder("us-east-1", "bucket", "http://localhost:9000", true);

	SigV4Credentials Creds;
	Creds.AccessKeyId	  = "ASIA-tmp";
	Creds.SecretAccessKey = "secret";
	Creds.SessionToken	  = "sts-session-token-value";

	HttpClient::KeyValueMap Headers = Builder.SignRequest(Creds, "PUT", "/bucket/key", "", S3EmptyPayloadHash);

	auto It = Headers->find("x-amz-security-token");
	REQUIRE(It != Headers->end());
	CHECK(It->second == "sts-session-token-value");
}

TEST_CASE("s3requestbuilder.signing_key_cache_invalidates_on_key_rotate")
{
	// Two consecutive Sign() calls with the same date but different AccessKeyId
	// must produce different Authorization signatures, proving the cache is keyed
	// on (DateStamp, AccessKeyId) and not date alone.
	S3RequestBuilder Builder("us-east-1", "bucket", "http://localhost:9000", true);

	SigV4Credentials A;
	A.AccessKeyId			   = "AKIDEXAMPLE";
	A.SecretAccessKey		   = "secretA";
	HttpClient::KeyValueMap HA = Builder.SignRequest(A, "GET", "/bucket/foo", "", S3EmptyPayloadHash);

	SigV4Credentials B;
	B.AccessKeyId			   = "AKIDEXAMPLE2";
	B.SecretAccessKey		   = "secretB";
	HttpClient::KeyValueMap HB = Builder.SignRequest(B, "GET", "/bucket/foo", "", S3EmptyPayloadHash);

	CHECK(HA->find("Authorization")->second != HB->find("Authorization")->second);
}

TEST_CASE("s3client.minio_integration")
{
	using namespace std::literals;

	// Spawn a single MinIO server for the entire test case. Previously each SUBCASE re-entered
	// the TEST_CASE from the top, spawning and killing MinIO per subcase - slow and flaky on
	// macOS CI. Sequential sections avoid the re-entry while still sharing one MinIO instance
	// that is torn down via RAII at scope exit.
	MinioProcessOptions MinioOpts;
	MinioOpts.Port		   = 19000;
	MinioOpts.RootUser	   = "testuser";
	MinioOpts.RootPassword = "testpassword";

	MinioProcess Minio(MinioOpts);
	Minio.SpawnMinioServer();
	Minio.CreateBucket("integration-test");

	S3ClientOptions Opts;
	Opts.BucketName					 = "integration-test";
	Opts.Region						 = "us-east-1";
	Opts.Endpoint					 = Minio.Endpoint();
	Opts.PathStyle					 = true;
	Opts.Credentials.AccessKeyId	 = std::string(Minio.RootUser());
	Opts.Credentials.SecretAccessKey = std::string(Minio.RootPassword());

	S3Client Client(Opts);

	// -- put_get_delete -------------------------------------------------------
	{
		// PUT
		std::string_view TestData = "hello, minio integration test!"sv;
		IoBuffer		 Content  = IoBufferBuilder::MakeFromMemory(MakeMemoryView(TestData));
		S3Result		 PutRes	  = Client.PutObject("test/hello.txt", std::move(Content));
		REQUIRE(PutRes.IsSuccess());

		// GET
		S3GetObjectResult GetRes = Client.GetObject("test/hello.txt");
		REQUIRE(GetRes.IsSuccess());
		CHECK(GetRes.AsText() == TestData);

		// HEAD
		S3HeadObjectResult HeadRes = Client.HeadObject("test/hello.txt");
		REQUIRE(HeadRes.IsSuccess());
		CHECK(HeadRes.Status == HeadObjectResult::Found);
		CHECK(HeadRes.Info.Size == TestData.size());

		// DELETE
		S3Result DelRes = Client.DeleteObject("test/hello.txt");
		REQUIRE(DelRes.IsSuccess());

		// HEAD after delete
		S3HeadObjectResult HeadRes2 = Client.HeadObject("test/hello.txt");
		REQUIRE(HeadRes2.IsSuccess());
		CHECK(HeadRes2.Status == HeadObjectResult::NotFound);
	}

	// -- touch ----------------------------------------------------------------
	{
		std::string_view TestData = "touch-me"sv;
		IoBuffer		 Content  = IoBufferBuilder::MakeFromMemory(MakeMemoryView(TestData));
		S3Result		 PutRes	  = Client.PutObject("touch/obj.txt", std::move(Content));
		REQUIRE(PutRes.IsSuccess());

		S3HeadObjectResult Before = Client.HeadObject("touch/obj.txt");
		REQUIRE(Before.IsSuccess());
		REQUIRE(Before.Status == HeadObjectResult::Found);

		// S3 LastModified has second precision; sleep past the second boundary so
		// the touched timestamp is strictly greater.
		Sleep(1100);

		S3Result TouchRes = Client.Touch("touch/obj.txt");
		REQUIRE(TouchRes.IsSuccess());

		S3HeadObjectResult After = Client.HeadObject("touch/obj.txt");
		REQUIRE(After.IsSuccess());
		REQUIRE(After.Status == HeadObjectResult::Found);
		CHECK(After.Info.Size == Before.Info.Size);
		CHECK(After.Info.LastModified != Before.Info.LastModified);

		// Content must be unchanged by a self-copy.
		S3GetObjectResult GetRes = Client.GetObject("touch/obj.txt");
		REQUIRE(GetRes.IsSuccess());
		CHECK(GetRes.AsText() == TestData);

		// Touching a missing key must fail.
		S3Result MissRes = Client.Touch("touch/does-not-exist.txt");
		CHECK_FALSE(MissRes.IsSuccess());

		Client.DeleteObject("touch/obj.txt");
	}

	// -- head_not_found -------------------------------------------------------
	{
		S3HeadObjectResult Res = Client.HeadObject("nonexistent/key.dat");
		CHECK(Res.IsSuccess());
		CHECK(Res.Status == HeadObjectResult::NotFound);
	}

	// -- list_objects ---------------------------------------------------------
	{
		// Upload several objects with a common prefix
		for (int i = 0; i < 3; ++i)
		{
			std::string Key		= fmt::format("list-test/item-{}.txt", i);
			std::string Payload = fmt::format("payload-{}", i);
			IoBuffer	Buf		= IoBufferBuilder::MakeFromMemory(MakeMemoryView(Payload));
			S3Result	Res		= Client.PutObject(Key, std::move(Buf));
			REQUIRE(Res.IsSuccess());
		}

		// List with prefix
		S3ListObjectsResult ListRes = Client.ListObjects("list-test/");
		REQUIRE(ListRes.IsSuccess());
		CHECK(ListRes.Objects.size() == 3);

		// Verify keys are present
		std::vector<std::string> Keys;
		for (const S3ObjectInfo& Obj : ListRes.Objects)
		{
			Keys.push_back(Obj.Key);
		}
		std::sort(Keys.begin(), Keys.end());
		CHECK(Keys[0] == "list-test/item-0.txt");
		CHECK(Keys[1] == "list-test/item-1.txt");
		CHECK(Keys[2] == "list-test/item-2.txt");

		// Cleanup
		for (int i = 0; i < 3; ++i)
		{
			Client.DeleteObject(fmt::format("list-test/item-{}.txt", i));
		}
	}

	// -- multipart_upload -----------------------------------------------------
	{
		// Create a payload large enough to exercise multipart (use minimum part size)
		constexpr uint64_t PartSize	   = 5 * 1024 * 1024;  // 5 MB minimum
		constexpr uint64_t PayloadSize = PartSize + 1024;  // slightly over one part

		std::string LargePayload(PayloadSize, 'X');
		// Add some variation
		for (uint64_t i = 0; i < PayloadSize; i += 1024)
		{
			LargePayload[i] = char('A' + (i / 1024) % 26);
		}

		IoBuffer Content = IoBufferBuilder::MakeFromMemory(MakeMemoryView(LargePayload));
		S3Result Res	 = Client.PutObjectMultipart("multipart/large.bin", std::move(Content), PartSize);
		REQUIRE(Res.IsSuccess());

		// Verify via GET
		S3GetObjectResult GetRes = Client.GetObject("multipart/large.bin");
		REQUIRE(GetRes.IsSuccess());
		CHECK(GetRes.Content.GetSize() == PayloadSize);
		CHECK(GetRes.AsText() == std::string_view(LargePayload));

		// Cleanup
		Client.DeleteObject("multipart/large.bin");
	}

	// -- presigned_urls -------------------------------------------------------
	{
		// Upload an object
		std::string_view TestData = "presigned-url-test-data"sv;
		IoBuffer		 Content  = IoBufferBuilder::MakeFromMemory(MakeMemoryView(TestData));
		S3Result		 PutRes	  = Client.PutObject("presigned/test.txt", std::move(Content));
		REQUIRE(PutRes.IsSuccess());

		// Generate a pre-signed GET URL
		std::string Url = Client.GeneratePresignedGetUrl("presigned/test.txt", std::chrono::seconds(60));
		CHECK(!Url.empty());
		CHECK(Url.find("X-Amz-Signature") != std::string::npos);

		// Fetch via the pre-signed URL (no auth headers needed)
		HttpClient Hc(Minio.Endpoint());
		// Extract the path+query from the full URL
		std::string_view	 UrlView   = Url;
		size_t				 PathStart = UrlView.find('/', UrlView.find("://") + 3);
		std::string			 PathAndQuery(UrlView.substr(PathStart));
		HttpClient::Response Resp = Hc.Get(PathAndQuery);
		REQUIRE(Resp.IsSuccess());
		CHECK(Resp.AsText() == TestData);

		// Cleanup
		Client.DeleteObject("presigned/test.txt");
	}
}

TEST_SUITE_END();

#endif

}  // namespace zen