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

#define _SILENCE_CXX17_C_HEADER_DEPRECATION_WARNING

#include <zencore/compactbinary.h>
#include <zencore/compactbinarybuilder.h>
#include <zencore/compactbinarypackage.h>
#include <zencore/compress.h>
#include <zencore/except.h>
#include <zencore/filesystem.h>
#include <zencore/fmtutils.h>
#include <zencore/iohash.h>
#include <zencore/logging.h>
#include <zencore/memory.h>
#include <zencore/refcount.h>
#include <zencore/stream.h>
#include <zencore/string.h>
#include <zencore/thread.h>
#include <zencore/timer.h>
#include <zenhttp/httpclient.h>
#include <zenhttp/httpshared.h>
#include <zenhttp/websocket.h>
#include <zenhttp/zenhttp.h>
#include <zenutil/cache/cache.h>
#include <zenutil/zenserverprocess.h>

#if ZEN_USE_MIMALLOC
ZEN_THIRD_PARTY_INCLUDES_START
#	include <mimalloc.h>
ZEN_THIRD_PARTY_INCLUDES_END
#endif

#include <http_parser.h>

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

ZEN_THIRD_PARTY_INCLUDES_START
#include <cpr/cpr.h>
#undef GetObject
ZEN_THIRD_PARTY_INCLUDES_END

#include <atomic>
#include <filesystem>
#include <map>
#include <random>
#include <span>
#include <thread>
#include <unordered_map>

#if ZEN_PLATFORM_WINDOWS
#	include <ppl.h>
#	include <atlbase.h>
#	include <process.h>
#endif

#include <asio.hpp>

//////////////////////////////////////////////////////////////////////////

#include "projectclient.h"

//////////////////////////////////////////////////////////////////////////

#if ZEN_WITH_TESTS
#	define DOCTEST_CONFIG_IMPLEMENT
#	include <zencore/testing.h>
#	undef DOCTEST_CONFIG_IMPLEMENT
#endif

using namespace std::literals;

#if ZEN_PLATFORM_LINUX || ZEN_PLATFORM_MAC
struct Concurrency
{
	template<typename... T>
	static void parallel_invoke(T&&... t)
	{
		constexpr size_t NumTs			= sizeof...(t);
		std::thread		 Threads[NumTs] = {
			 std::thread(std::forward<T>(t))...,
		 };

		for (std::thread& Thread : Threads)
		{
			Thread.join();
		}
	}
};
#endif

//////////////////////////////////////////////////////////////////////////
//
// Custom logging -- test code, this should be tweaked
//

namespace logging {
using namespace spdlog;
using namespace spdlog::details;
using namespace std::literals;

class full_test_formatter final : public spdlog::formatter
{
public:
	full_test_formatter(std::string_view LogId, std::chrono::time_point<std::chrono::system_clock> Epoch) : m_Epoch(Epoch), m_LogId(LogId)
	{
	}

	virtual std::unique_ptr<formatter> clone() const override { return std::make_unique<full_test_formatter>(m_LogId, m_Epoch); }

	static constexpr bool UseDate = false;

	virtual void format(const details::log_msg& msg, memory_buf_t& dest) override
	{
		using std::chrono::duration_cast;
		using std::chrono::milliseconds;
		using std::chrono::seconds;

		if constexpr (UseDate)
		{
			auto secs = std::chrono::duration_cast<seconds>(msg.time.time_since_epoch());
			if (secs != m_LastLogSecs)
			{
				m_CachedTm	  = os::localtime(log_clock::to_time_t(msg.time));
				m_LastLogSecs = secs;
			}
		}

		const auto& tm_time = m_CachedTm;

		// cache the date/time part for the next second.
		auto duration = msg.time - m_Epoch;
		auto secs	  = duration_cast<seconds>(duration);

		if (m_CacheTimestamp != secs || m_CachedDatetime.size() == 0)
		{
			m_CachedDatetime.clear();
			m_CachedDatetime.push_back('[');

			if constexpr (UseDate)
			{
				fmt_helper::append_int(tm_time.tm_year + 1900, m_CachedDatetime);
				m_CachedDatetime.push_back('-');

				fmt_helper::pad2(tm_time.tm_mon + 1, m_CachedDatetime);
				m_CachedDatetime.push_back('-');

				fmt_helper::pad2(tm_time.tm_mday, m_CachedDatetime);
				m_CachedDatetime.push_back(' ');

				fmt_helper::pad2(tm_time.tm_hour, m_CachedDatetime);
				m_CachedDatetime.push_back(':');

				fmt_helper::pad2(tm_time.tm_min, m_CachedDatetime);
				m_CachedDatetime.push_back(':');

				fmt_helper::pad2(tm_time.tm_sec, m_CachedDatetime);
			}
			else
			{
				int Count = int(secs.count());

				const int LogSecs = Count % 60;
				Count /= 60;

				const int LogMins = Count % 60;
				Count /= 60;

				const int LogHours = Count;

				fmt_helper::pad2(LogHours, m_CachedDatetime);
				m_CachedDatetime.push_back(':');
				fmt_helper::pad2(LogMins, m_CachedDatetime);
				m_CachedDatetime.push_back(':');
				fmt_helper::pad2(LogSecs, m_CachedDatetime);
			}

			m_CachedDatetime.push_back('.');

			m_CacheTimestamp = secs;
		}

		dest.append(m_CachedDatetime.begin(), m_CachedDatetime.end());

		auto millis = fmt_helper::time_fraction<milliseconds>(msg.time);
		fmt_helper::pad3(static_cast<uint32_t>(millis.count()), dest);
		dest.push_back(']');
		dest.push_back(' ');

		if (!m_LogId.empty())
		{
			dest.push_back('[');
			fmt_helper::append_string_view(m_LogId, dest);
			dest.push_back(']');
			dest.push_back(' ');
		}

		// append logger name if exists
		if (msg.logger_name.size() > 0)
		{
			dest.push_back('[');
			fmt_helper::append_string_view(msg.logger_name, dest);
			dest.push_back(']');
			dest.push_back(' ');
		}

		dest.push_back('[');
		// wrap the level name with color
		msg.color_range_start = dest.size();
		fmt_helper::append_string_view(level::to_string_view(msg.level), dest);
		msg.color_range_end = dest.size();
		dest.push_back(']');
		dest.push_back(' ');

		// add source location if present
		if (!msg.source.empty())
		{
			dest.push_back('[');
			const char* filename = details::short_filename_formatter<details::null_scoped_padder>::basename(msg.source.filename);
			fmt_helper::append_string_view(filename, dest);
			dest.push_back(':');
			fmt_helper::append_int(msg.source.line, dest);
			dest.push_back(']');
			dest.push_back(' ');
		}

		fmt_helper::append_string_view(msg.payload, dest);
		fmt_helper::append_string_view("\n"sv, dest);
	}

private:
	std::chrono::time_point<std::chrono::system_clock> m_Epoch;
	std::tm											   m_CachedTm;
	std::chrono::seconds							   m_LastLogSecs;
	std::chrono::seconds							   m_CacheTimestamp{0};
	memory_buf_t									   m_CachedDatetime;
	std::string										   m_LogId;
};
}  // namespace logging

//////////////////////////////////////////////////////////////////////////

#if 0

int
main()
{
	mi_version();

	zen::Sleep(1000);

	zen::Stopwatch timer;

	const int RequestCount = 100000;

	cpr::Session Sessions[10];

	for (auto& Session : Sessions)
	{
		Session.SetUrl(cpr::Url{"http://localhost:1337/test/hello"});
		//Session.SetUrl(cpr::Url{ "http://arn-wd-l0182:1337/test/hello" });
	}

	auto Run = [](cpr::Session& Session) {
		for (int i = 0; i < 10000; ++i)
		{
			cpr::Response Result = Session.Get();
			
			if (Result.status_code != 200)
			{
				ZEN_WARN("request response: {}", Result.status_code);
			}
		}
	};

	Concurrency::parallel_invoke([&] { Run(Sessions[0]); },
								 [&] { Run(Sessions[1]); },
								 [&] { Run(Sessions[2]); },
								 [&] { Run(Sessions[3]); },
								 [&] { Run(Sessions[4]); },
								 [&] { Run(Sessions[5]); },
								 [&] { Run(Sessions[6]); },
								 [&] { Run(Sessions[7]); },
								 [&] { Run(Sessions[8]); },
								 [&] { Run(Sessions[9]); });

	// cpr::Response r = cpr::Get(cpr::Url{ "http://localhost:1337/test/hello" });

	ZEN_INFO("{} requests in {} ({})",
				 RequestCount,
				 zen::NiceTimeSpanMs(timer.GetElapsedTimeMs()),
				 zen::NiceRate(RequestCount, (uint32_t)timer.GetElapsedTimeMs(), "req"));

	return 0;
}
#elif 0
//#include <restinio/all.hpp>

int
main()
{
	mi_version();
	restinio::run(restinio::on_thread_pool(32).port(8080).request_handler(
		[](auto req) { return req->create_response().set_body("Hello, World!").done(); }));
	return 0;
}
#elif ZEN_WITH_TESTS

zen::ZenServerEnvironment TestEnv;

int
main(int argc, char** argv)
{
#	if ZEN_USE_MIMALLOC
	mi_version();
#	endif

	zen::zencore_forcelinktests();
	zen::zenhttp_forcelinktests();

	zen::logging::InitializeLogging();

	spdlog::set_level(spdlog::level::debug);
	spdlog::set_formatter(std::make_unique< ::logging::full_test_formatter>("test", std::chrono::system_clock::now()));

	std::filesystem::path ProgramBaseDir = std::filesystem::path(argv[0]).parent_path();
	std::filesystem::path TestBaseDir	 = ProgramBaseDir.parent_path().parent_path() / ".test";

	TestEnv.InitializeForTest(ProgramBaseDir, TestBaseDir);

	ZEN_INFO("Running tests...(base dir: '{}')", TestBaseDir);
	return doctest::Context(argc, argv).run();
}

namespace zen::tests {

TEST_CASE("default.single")
{
	std::filesystem::path TestDir = TestEnv.CreateNewTestDir();

	ZenServerInstance Instance(TestEnv);
	Instance.SetTestDir(TestDir);
	Instance.SpawnServer(13337);

	ZEN_INFO("Waiting...");

	Instance.WaitUntilReady();

	std::atomic<uint64_t> RequestCount{0};
	std::atomic<uint64_t> BatchCounter{0};

	ZEN_INFO("Running single server test...");

	auto IssueTestRequests = [&] {
		const uint64_t BatchNo	= BatchCounter.fetch_add(1);
		const int	   ThreadId = zen::GetCurrentThreadId();

		ZEN_INFO("query batch {} started (thread {})", BatchNo, ThreadId);
		cpr::Session cli;
		cli.SetUrl(cpr::Url{"http://localhost:13337/test/hello"});

		for (int i = 0; i < 10000; ++i)
		{
			auto res = cli.Get();
			++RequestCount;
		}
		ZEN_INFO("query batch {} ended (thread {})", BatchNo, ThreadId);
	};

	zen::Stopwatch timer;

	Concurrency::parallel_invoke(IssueTestRequests,
								 IssueTestRequests,
								 IssueTestRequests,
								 IssueTestRequests,
								 IssueTestRequests,
								 IssueTestRequests,
								 IssueTestRequests,
								 IssueTestRequests,
								 IssueTestRequests,
								 IssueTestRequests);

	uint64_t Elapsed = timer.GetElapsedTimeMs();

	ZEN_INFO("{} requests in {} ({})", RequestCount, zen::NiceTimeSpanMs(Elapsed), zen::NiceRate(RequestCount, (uint32_t)Elapsed, "req"));
}

TEST_CASE("multi.basic")
{
	ZenServerInstance	  Instance1(TestEnv);
	std::filesystem::path TestDir1 = TestEnv.CreateNewTestDir();
	Instance1.SetTestDir(TestDir1);
	Instance1.SpawnServer(13337);

	ZenServerInstance	  Instance2(TestEnv);
	std::filesystem::path TestDir2 = TestEnv.CreateNewTestDir();
	Instance2.SetTestDir(TestDir2);
	Instance2.SpawnServer(13338);

	ZEN_INFO("Waiting...");

	Instance1.WaitUntilReady();
	Instance2.WaitUntilReady();

	std::atomic<uint64_t> RequestCount{0};
	std::atomic<uint64_t> BatchCounter{0};

	auto IssueTestRequests = [&](int PortNumber) {
		const uint64_t BatchNo	= BatchCounter.fetch_add(1);
		const int	   ThreadId = zen::GetCurrentThreadId();

		ZEN_INFO("query batch {} started (thread {}) for port {}", BatchNo, ThreadId, PortNumber);

		cpr::Session cli;
		cli.SetUrl(cpr::Url{fmt::format("http://localhost:{}/test/hello", PortNumber)});

		for (int i = 0; i < 10000; ++i)
		{
			auto res = cli.Get();
			++RequestCount;
		}
		ZEN_INFO("query batch {} ended (thread {})", BatchNo, ThreadId);
	};

	zen::Stopwatch timer;

	ZEN_INFO("Running multi-server test...");

	Concurrency::parallel_invoke([&] { IssueTestRequests(13337); },
								 [&] { IssueTestRequests(13338); },
								 [&] { IssueTestRequests(13337); },
								 [&] { IssueTestRequests(13338); });

	uint64_t Elapsed = timer.GetElapsedTimeMs();

	ZEN_INFO("{} requests in {} ({})", RequestCount, zen::NiceTimeSpanMs(Elapsed), zen::NiceRate(RequestCount, (uint32_t)Elapsed, "req"));
}

TEST_CASE("cas.basic")
{
	std::filesystem::path TestDir = TestEnv.CreateNewTestDir();

	const uint16_t PortNumber = 13337;

	const int IterationCount = 1000;

	std::vector<int>		 ChunkSizes(IterationCount);
	std::vector<zen::IoHash> ChunkHashes(IterationCount);

	{
		ZenServerInstance Instance1(TestEnv);
		Instance1.SetTestDir(TestDir);
		Instance1.SpawnServer(PortNumber);
		Instance1.WaitUntilReady();

		std::atomic<uint64_t> RequestCount{0};

		zen::Stopwatch timer;

		std::mt19937_64 mt;

		auto BaseUri = fmt::format("http://localhost:{}/cas", PortNumber);

		cpr::Session cli;
		cli.SetUrl(cpr::Url{BaseUri});

		// Populate CAS with some generated data

		for (int i = 0; i < IterationCount; ++i)
		{
			const int	ChunkSize = mt() % 10000 + 5;
			std::string body	  = fmt::format("{}", i);
			body.resize(ChunkSize, ' ');

			ChunkSizes[i]  = ChunkSize;
			ChunkHashes[i] = zen::IoHash::HashBuffer(body.data(), body.size());

			cli.SetBody(body);

			auto res = cli.Post();
			CHECK(!res.error);

			++RequestCount;
		}

		// Verify that the chunks persisted

		for (int i = 0; i < IterationCount; ++i)
		{
			zen::ExtendableStringBuilder<128> Uri;
			Uri << BaseUri << "/";
			ChunkHashes[i].ToHexString(Uri);

			auto res = cpr::Get(cpr::Url{Uri.c_str()});
			CHECK(!res.error);
			CHECK(res.status_code == 200);
			CHECK(res.text.size() == ChunkSizes[i]);

			zen::IoHash Hash = zen::IoHash::HashBuffer(res.text.data(), res.text.size());

			CHECK(ChunkHashes[i] == Hash);

			++RequestCount;
		}

		uint64_t Elapsed = timer.GetElapsedTimeMs();

		ZEN_INFO("{} requests in {} ({})",
				 RequestCount,
				 zen::NiceTimeSpanMs(Elapsed),
				 zen::NiceRate(RequestCount, (uint32_t)Elapsed, "req"));
	}

	// Verify that the data persists between process runs (the previous server has exited at this point)

	{
		ZenServerInstance Instance2(TestEnv);
		Instance2.SetTestDir(TestDir);
		Instance2.SpawnServer(PortNumber);
		Instance2.WaitUntilReady();

		for (int i = 0; i < IterationCount; ++i)
		{
			zen::ExtendableStringBuilder<128> Uri;
			Uri << fmt::format("http://localhost:{}/cas/", PortNumber);
			ChunkHashes[i].ToHexString(Uri);

			auto res = cpr::Get(cpr::Url{Uri.c_str()});
			CHECK(res.status_code == 200);
			CHECK(res.text.size() == ChunkSizes[i]);

			zen::IoHash Hash = zen::IoHash::HashBuffer(res.text.data(), res.text.size());

			CHECK(ChunkHashes[i] == Hash);
		}
	}
}

TEST_CASE("project.basic")
{
	using namespace std::literals;

	std::filesystem::path TestDir = TestEnv.CreateNewTestDir();

	const uint16_t PortNumber = 13337;

	ZenServerInstance Instance1(TestEnv);
	Instance1.SetTestDir(TestDir);
	Instance1.SpawnServer(PortNumber);
	Instance1.WaitUntilReady();

	std::atomic<uint64_t> RequestCount{0};

	zen::Stopwatch timer;

	std::mt19937_64 mt;

	zen::StringBuilder<64> BaseUri;
	BaseUri << fmt::format("http://localhost:{}/prj/test", PortNumber);

	std::filesystem::path BinPath  = zen::GetRunningExecutablePath();
	std::filesystem::path RootPath = BinPath.parent_path().parent_path();
	BinPath						   = BinPath.lexically_relative(RootPath);

	SUBCASE("build store init")
	{
		{
			{
				zen::CbObjectWriter Body;
				Body << "id"
					 << "test";
				Body << "root" << RootPath.c_str();
				Body << "project"
					 << "/zooom";
				Body << "engine"
					 << "/zooom";

				zen::BinaryWriter MemOut;
				Body.Save(MemOut);

				auto Response = cpr::Post(cpr::Url{BaseUri.c_str()}, cpr::Body{(const char*)MemOut.Data(), MemOut.Size()});
				CHECK(Response.status_code == 201);
			}

			{
				auto Response = cpr::Get(cpr::Url{BaseUri.c_str()});
				CHECK(Response.status_code == 200);

				zen::CbObjectView ResponseObject = zen::CbFieldView(Response.text.data()).AsObjectView();

				CHECK(ResponseObject["id"].AsString() == "test"sv);
				CHECK(ResponseObject["root"].AsString() == PathToUtf8(RootPath.c_str()));
			}
		}

		BaseUri << "/oplog/foobar";

		{
			{
				zen::StringBuilder<64> PostUri;
				PostUri << BaseUri;
				auto Response = cpr::Post(cpr::Url{PostUri.c_str()});
				CHECK(Response.status_code == 201);
			}

			{
				auto Response = cpr::Get(cpr::Url{BaseUri.c_str()});
				CHECK(Response.status_code == 200);

				zen::CbObjectView ResponseObject = zen::CbFieldView(Response.text.data()).AsObjectView();

				CHECK(ResponseObject["id"].AsString() == "foobar"sv);
				CHECK(ResponseObject["project"].AsString() == "test"sv);
			}
		}

		SUBCASE("build store persistence")
		{
			uint8_t AttachData[] = {1, 2, 3};

			zen::CbAttachment Attach{zen::CompressedBuffer::Compress(zen::SharedBuffer::Clone(zen::MemoryView{AttachData, 3}))};

			zen::CbObjectWriter OpWriter;
			OpWriter << "key"
					 << "foo"
					 << "attachment" << Attach;

			const std::string_view ChunkId{
				"00000000"
				"00000000"
				"00010000"};
			auto FileOid = zen::Oid::FromHexString(ChunkId);

			OpWriter.BeginArray("files");
			OpWriter.BeginObject();
			OpWriter << "id" << FileOid;
			OpWriter << "clientpath"
					 << "/{engine}/client/side/path";
			OpWriter << "serverpath" << BinPath.c_str();
			OpWriter.EndObject();
			OpWriter.EndArray();

			zen::CbObject Op = OpWriter.Save();

			zen::CbPackage OpPackage(Op);
			OpPackage.AddAttachment(Attach);

			zen::BinaryWriter MemOut;
			legacy::SaveCbPackage(OpPackage, MemOut);

			{
				zen::StringBuilder<64> PostUri;
				PostUri << BaseUri << "/new";
				auto Response = cpr::Post(cpr::Url{PostUri.c_str()}, cpr::Body{(const char*)MemOut.Data(), MemOut.Size()});

				REQUIRE(!Response.error);
				CHECK(Response.status_code == 201);
			}

			// Read file data

			{
				zen::StringBuilder<128> ChunkGetUri;
				ChunkGetUri << BaseUri << "/" << ChunkId;
				auto Response = cpr::Get(cpr::Url{ChunkGetUri.c_str()});

				REQUIRE(!Response.error);
				CHECK(Response.status_code == 200);
			}

			{
				zen::StringBuilder<128> ChunkGetUri;
				ChunkGetUri << BaseUri << "/" << ChunkId << "?offset=1&size=10";
				auto Response = cpr::Get(cpr::Url{ChunkGetUri.c_str()});

				REQUIRE(!Response.error);
				CHECK(Response.status_code == 200);
				CHECK(Response.text.size() == 10);
			}

			ZEN_INFO("+++++++");
		}
		SUBCASE("build store op commit") { ZEN_INFO("-------"); }
	}

	const uint64_t Elapsed = timer.GetElapsedTimeMs();

	ZEN_INFO("{} requests in {} ({})", RequestCount, zen::NiceTimeSpanMs(Elapsed), zen::NiceRate(RequestCount, (uint32_t)Elapsed, "req"));
}

#	if 0  // this is extremely WIP
TEST_CASE("project.pipe")
{
	using namespace std::literals;

	std::filesystem::path TestDir = TestEnv.CreateNewTestDir();

	const uint16_t PortNumber = 13337;

	ZenServerInstance Instance1(TestEnv);
	Instance1.SetTestDir(TestDir);
	Instance1.SpawnServer(PortNumber);
	Instance1.WaitUntilReady();

	zen::LocalProjectClient LocalClient(PortNumber);

	zen::CbObjectWriter Cbow;
	Cbow << "hey" << 42;

	zen::CbObject Response = LocalClient.MessageTransaction(Cbow.Save());
}
#	endif

namespace utils {

	struct ZenConfig
	{
		std::filesystem::path DataDir;
		uint16_t			  Port;
		std::string			  BaseUri;
		std::string			  Args;

		static ZenConfig New(uint16_t Port = 13337, std::string Args = "")
		{
			return ZenConfig{.DataDir = TestEnv.CreateNewTestDir(),
							 .Port	  = Port,
							 .BaseUri = fmt::format("http://localhost:{}/z$", Port),
							 .Args	  = std::move(Args)};
		}

		static ZenConfig NewWithUpstream(uint16_t UpstreamPort)
		{
			return New(13337, fmt::format("--debug --upstream-thread-count=0 --upstream-zen-url=http://localhost:{}", UpstreamPort));
		}

		void Spawn(ZenServerInstance& Inst)
		{
			Inst.SetTestDir(DataDir);
			Inst.SpawnServer(Port, Args);
			Inst.WaitUntilReady();
		}
	};

	void SpawnServer(ZenServerInstance& Server, ZenConfig& Cfg)
	{
		Server.SetTestDir(Cfg.DataDir);
		Server.SpawnServer(Cfg.Port, Cfg.Args);
		Server.WaitUntilReady();
	}

}  // namespace utils

TEST_CASE("zcache.basic")
{
	using namespace std::literals;

	std::filesystem::path TestDir = TestEnv.CreateNewTestDir();

	const uint16_t PortNumber = 13337;

	const int  kIterationCount = 100;
	const auto BaseUri		   = fmt::format("http://localhost:{}/z$", PortNumber);

	auto HashKey = [](int i) -> zen::IoHash { return zen::IoHash::HashBuffer(&i, sizeof i); };

	{
		ZenServerInstance Instance1(TestEnv);
		Instance1.SetTestDir(TestDir);
		Instance1.SpawnServer(PortNumber);
		Instance1.WaitUntilReady();

		// Populate with some simple data

		for (int i = 0; i < kIterationCount; ++i)
		{
			zen::CbObjectWriter Cbo;
			Cbo << "index" << i;

			zen::BinaryWriter MemOut;
			Cbo.Save(MemOut);

			zen::IoHash Key = HashKey(i);

			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}", BaseUri, "test", Key)},
											cpr::Body{(const char*)MemOut.Data(), MemOut.Size()},
											cpr::Header{{"Content-Type", "application/x-ue-cb"}});

			CHECK(Result.status_code == 201);
		}

		// Retrieve data

		for (int i = 0; i < kIterationCount; ++i)
		{
			zen::IoHash Key = zen::IoHash::HashBuffer(&i, sizeof i);

			cpr::Response Result =
				cpr::Get(cpr::Url{fmt::format("{}/{}/{}", BaseUri, "test", Key)}, cpr::Header{{"Accept", "application/x-ue-cbpkg"}});

			CHECK(Result.status_code == 200);
		}

		// Ensure bad bucket identifiers are rejected

		{
			zen::CbObjectWriter Cbo;
			Cbo << "index" << 42;

			zen::BinaryWriter MemOut;
			Cbo.Save(MemOut);

			zen::IoHash Key = HashKey(442);

			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}", BaseUri, "te!st", Key)},
											cpr::Body{(const char*)MemOut.Data(), MemOut.Size()},
											cpr::Header{{"Content-Type", "application/x-ue-cb"}});

			CHECK(Result.status_code == 400);
		}
	}

	// Verify that the data persists between process runs (the previous server has exited at this point)

	{
		ZenServerInstance Instance1(TestEnv);
		Instance1.SetTestDir(TestDir);
		Instance1.SpawnServer(PortNumber);
		Instance1.WaitUntilReady();

		// Retrieve data again

		for (int i = 0; i < kIterationCount; ++i)
		{
			zen::IoHash Key = HashKey(i);

			cpr::Response Result =
				cpr::Get(cpr::Url{fmt::format("{}/{}/{}", BaseUri, "test", Key)}, cpr::Header{{"Accept", "application/x-ue-cbpkg"}});

			CHECK(Result.status_code == 200);
		}
	}
}

TEST_CASE("zcache.cbpackage")
{
	using namespace std::literals;

	auto CreateTestPackage = [](zen::IoHash& OutAttachmentKey) -> zen::CbPackage {
		auto Data			= zen::SharedBuffer::Clone(zen::MakeMemoryView<uint8_t>({1, 2, 3, 4, 5, 6, 7, 8, 9}));
		auto CompressedData = zen::CompressedBuffer::Compress(Data);

		OutAttachmentKey = zen::IoHash::FromBLAKE3(CompressedData.GetRawHash());

		zen::CbWriter Obj;
		Obj.BeginObject("obj"sv);
		Obj.AddBinaryAttachment("data", OutAttachmentKey);
		Obj.EndObject();

		zen::CbPackage Package;
		Package.SetObject(Obj.Save().AsObject());
		Package.AddAttachment(zen::CbAttachment(CompressedData));

		return Package;
	};

	auto SerializeToBuffer = [](zen::CbPackage Package) -> zen::IoBuffer {
		zen::BinaryWriter MemStream;

		Package.Save(MemStream);

		return zen::IoBuffer(zen::IoBuffer::Clone, MemStream.Data(), MemStream.Size());
	};

	auto IsEqual = [](zen::CbPackage Lhs, zen::CbPackage Rhs) -> bool {
		std::span<const zen::CbAttachment> LhsAttachments = Lhs.GetAttachments();
		std::span<const zen::CbAttachment> RhsAttachments = Rhs.GetAttachments();

		if (LhsAttachments.size() != RhsAttachments.size())
		{
			return false;
		}

		for (const zen::CbAttachment& LhsAttachment : LhsAttachments)
		{
			const zen::CbAttachment* RhsAttachment = Rhs.FindAttachment(LhsAttachment.GetHash());
			CHECK(RhsAttachment);

			zen::SharedBuffer LhsBuffer = LhsAttachment.AsCompressedBinary().Decompress();
			CHECK(!LhsBuffer.IsNull());

			zen::SharedBuffer RhsBuffer = RhsAttachment->AsCompressedBinary().Decompress();
			CHECK(!RhsBuffer.IsNull());

			if (!LhsBuffer.GetView().EqualBytes(RhsBuffer.GetView()))
			{
				return false;
			}
		}

		return true;
	};

	SUBCASE("PUT/GET returns correct package")
	{
		std::filesystem::path TestDir	 = TestEnv.CreateNewTestDir();
		const uint16_t		  PortNumber = 13337;
		const auto			  BaseUri	 = fmt::format("http://localhost:{}/z$", PortNumber);

		ZenServerInstance Instance1(TestEnv);
		Instance1.SetTestDir(TestDir);
		Instance1.SpawnServer(PortNumber);
		Instance1.WaitUntilReady();

		const std::string_view Bucket = "mosdef"sv;
		zen::IoHash			   Key;
		zen::CbPackage		   ExpectedPackage = CreateTestPackage(Key);

		// PUT
		{
			zen::IoBuffer Body	 = SerializeToBuffer(ExpectedPackage);
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}", BaseUri, Bucket, Key)},
											cpr::Body{(const char*)Body.Data(), Body.Size()},
											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 201);
		}

		// GET
		{
			cpr::Response Result =
				cpr::Get(cpr::Url{fmt::format("{}/{}/{}", BaseUri, Bucket, Key)}, cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 200);

			zen::IoBuffer Response(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size());

			zen::CbPackage Package;
			const bool	   Ok = Package.TryLoad(Response);
			CHECK(Ok);
			CHECK(IsEqual(Package, ExpectedPackage));
		}
	}

	SUBCASE("PUT propagates upstream")
	{
		// Setup local and remote server
		std::filesystem::path LocalDataDir	   = TestEnv.CreateNewTestDir();
		std::filesystem::path RemoteDataDir	   = TestEnv.CreateNewTestDir();
		const uint16_t		  LocalPortNumber  = 13337;
		const uint16_t		  RemotePortNumber = 13338;

		const auto LocalBaseUri	 = fmt::format("http://localhost:{}/z$", LocalPortNumber);
		const auto RemoteBaseUri = fmt::format("http://localhost:{}/z$", RemotePortNumber);

		ZenServerInstance RemoteInstance(TestEnv);
		RemoteInstance.SetTestDir(RemoteDataDir);
		RemoteInstance.SpawnServer(RemotePortNumber);
		RemoteInstance.WaitUntilReady();

		ZenServerInstance LocalInstance(TestEnv);
		LocalInstance.SetTestDir(LocalDataDir);
		LocalInstance.SpawnServer(LocalPortNumber,
								  fmt::format("--upstream-thread-count=0 --upstream-zen-url=http://localhost:{}", RemotePortNumber));
		LocalInstance.WaitUntilReady();

		const std::string_view Bucket = "mosdef"sv;
		zen::IoHash			   Key;
		zen::CbPackage		   ExpectedPackage = CreateTestPackage(Key);

		// Store the cache record package in the local instance
		{
			zen::IoBuffer Body	 = SerializeToBuffer(ExpectedPackage);
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}", LocalBaseUri, Bucket, Key)},
											cpr::Body{(const char*)Body.Data(), Body.Size()},
											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});

			CHECK(Result.status_code == 201);
		}

		// The cache record can be retrieved as a package from the local instance
		{
			cpr::Response Result =
				cpr::Get(cpr::Url{fmt::format("{}/{}/{}", LocalBaseUri, Bucket, Key)}, cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 200);

			zen::IoBuffer  Body(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size());
			zen::CbPackage Package;
			const bool	   Ok = Package.TryLoad(Body);
			CHECK(Ok);
			CHECK(IsEqual(Package, ExpectedPackage));
		}

		// The cache record can be retrieved as a package from the remote instance
		{
			cpr::Response Result =
				cpr::Get(cpr::Url{fmt::format("{}/{}/{}", RemoteBaseUri, Bucket, Key)}, cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 200);

			zen::IoBuffer  Body(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size());
			zen::CbPackage Package;
			const bool	   Ok = Package.TryLoad(Body);
			CHECK(Ok);
			CHECK(IsEqual(Package, ExpectedPackage));
		}
	}

	SUBCASE("GET finds upstream when missing in local")
	{
		// Setup local and remote server
		std::filesystem::path LocalDataDir	   = TestEnv.CreateNewTestDir();
		std::filesystem::path RemoteDataDir	   = TestEnv.CreateNewTestDir();
		const uint16_t		  LocalPortNumber  = 13337;
		const uint16_t		  RemotePortNumber = 13338;

		const auto LocalBaseUri	 = fmt::format("http://localhost:{}/z$", LocalPortNumber);
		const auto RemoteBaseUri = fmt::format("http://localhost:{}/z$", RemotePortNumber);

		ZenServerInstance RemoteInstance(TestEnv);
		RemoteInstance.SetTestDir(RemoteDataDir);
		RemoteInstance.SpawnServer(RemotePortNumber);
		RemoteInstance.WaitUntilReady();

		ZenServerInstance LocalInstance(TestEnv);
		LocalInstance.SetTestDir(LocalDataDir);
		LocalInstance.SpawnServer(LocalPortNumber,
								  fmt::format("--upstream-thread-count=0 --upstream-zen-url=http://localhost:{}", RemotePortNumber));
		LocalInstance.WaitUntilReady();

		const std::string_view Bucket = "mosdef"sv;
		zen::IoHash			   Key;
		zen::CbPackage		   ExpectedPackage = CreateTestPackage(Key);

		// Store the cache record package in upstream cache
		{
			zen::IoBuffer Body	 = SerializeToBuffer(ExpectedPackage);
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}", RemoteBaseUri, Bucket, Key)},
											cpr::Body{(const char*)Body.Data(), Body.Size()},
											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});

			CHECK(Result.status_code == 201);
		}

		// The cache record can be retrieved as a package from the local cache
		{
			cpr::Response Result =
				cpr::Get(cpr::Url{fmt::format("{}/{}/{}", LocalBaseUri, Bucket, Key)}, cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 200);

			zen::IoBuffer  Body(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size());
			zen::CbPackage Package;
			const bool	   Ok = Package.TryLoad(Body);
			CHECK(Ok);
			CHECK(IsEqual(Package, ExpectedPackage));
		}
	}
}

TEST_CASE("zcache.policy")
{
	using namespace std::literals;
	using namespace utils;

	auto GenerateData = [](uint64_t Size, zen::IoHash& OutHash) -> zen::UniqueBuffer {
		auto	 Buf  = zen::UniqueBuffer::Alloc(Size);
		uint8_t* Data = reinterpret_cast<uint8_t*>(Buf.GetData());
		for (uint64_t Idx = 0; Idx < Size; Idx++)
		{
			Data[Idx] = Idx % 256;
		}
		OutHash = zen::IoHash::HashBuffer(Data, Size);
		return Buf;
	};

	auto GeneratePackage = [](zen::IoHash& OutRecordKey, zen::IoHash& OutAttachmentKey) -> zen::CbPackage {
		auto Data			= zen::SharedBuffer::Clone(zen::MakeMemoryView<uint8_t>({1, 2, 3, 4, 5, 6, 7, 8, 9}));
		auto CompressedData = zen::CompressedBuffer::Compress(Data);
		OutAttachmentKey	= zen::IoHash::FromBLAKE3(CompressedData.GetRawHash());

		zen::CbWriter Writer;
		Writer.BeginObject("obj"sv);
		Writer.AddBinaryAttachment("data", OutAttachmentKey);
		Writer.EndObject();
		CbObject CacheRecord = Writer.Save().AsObject();

		OutRecordKey = IoHash::HashBuffer(CacheRecord.GetBuffer().GetView());

		zen::CbPackage Package;
		Package.SetObject(CacheRecord);
		Package.AddAttachment(zen::CbAttachment(CompressedData));

		return Package;
	};

	auto ToBuffer = [](zen::CbPackage Package) -> zen::IoBuffer {
		zen::BinaryWriter MemStream;
		Package.Save(MemStream);

		return zen::IoBuffer(zen::IoBuffer::Clone, MemStream.Data(), MemStream.Size());
	};

	SUBCASE("query - 'local' does not query upstream (binary)")
	{
		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
		ZenServerInstance UpstreamInst(TestEnv);
		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
		ZenServerInstance LocalInst(TestEnv);
		const auto		  Bucket = "legacy"sv;

		UpstreamCfg.Spawn(UpstreamInst);
		LocalCfg.Spawn(LocalInst);

		zen::IoHash Key;
		auto		BinaryValue = GenerateData(1024, Key);

		// Store binary cache value upstream
		{
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}", UpstreamCfg.BaseUri, Bucket, Key)},
											cpr::Body{(const char*)BinaryValue.GetData(), BinaryValue.GetSize()},
											cpr::Header{{"Content-Type", "application/octet-stream"}});
			CHECK(Result.status_code == 201);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}?Policy=QueryLocal,Store", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/octet-stream"}});
			CHECK(Result.status_code == 404);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}?Policy=Query,Store", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/octet-stream"}});
			CHECK(Result.status_code == 200);
		}
	}

	SUBCASE("store - 'local' does not store upstream (binary)")
	{
		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
		ZenServerInstance UpstreamInst(TestEnv);
		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
		ZenServerInstance LocalInst(TestEnv);
		const auto		  Bucket = "legacy"sv;

		UpstreamCfg.Spawn(UpstreamInst);
		LocalCfg.Spawn(LocalInst);

		zen::IoHash Key;
		auto		BinaryValue = GenerateData(1024, Key);

		// Store binary cache value locally
		{
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}?Policy=Query,StoreLocal", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Body{(const char*)BinaryValue.GetData(), BinaryValue.GetSize()},
											cpr::Header{{"Content-Type", "application/octet-stream"}});
			CHECK(Result.status_code == 201);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", UpstreamCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/octet-stream"}});
			CHECK(Result.status_code == 404);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/octet-stream"}});
			CHECK(Result.status_code == 200);
		}
	}

	SUBCASE("store - 'local/remote' stores local and upstream (binary)")
	{
		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
		ZenServerInstance UpstreamInst(TestEnv);
		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
		ZenServerInstance LocalInst(TestEnv);
		const auto		  Bucket = "legacy"sv;

		UpstreamCfg.Spawn(UpstreamInst);
		LocalCfg.Spawn(LocalInst);

		zen::IoHash Key;
		auto		BinaryValue = GenerateData(1024, Key);

		// Store binary cache value locally and upstream
		{
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}?Policy=Query,Store", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Body{(const char*)BinaryValue.GetData(), BinaryValue.GetSize()},
											cpr::Header{{"Content-Type", "application/octet-stream"}});
			CHECK(Result.status_code == 201);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", UpstreamCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/octet-stream"}});
			CHECK(Result.status_code == 200);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/octet-stream"}});
			CHECK(Result.status_code == 200);
		}
	}

	SUBCASE("query - 'local' does not query upstream (cppackage)")
	{
		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
		ZenServerInstance UpstreamInst(TestEnv);
		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
		ZenServerInstance LocalInst(TestEnv);
		const auto		  Bucket = "legacy"sv;

		UpstreamCfg.Spawn(UpstreamInst);
		LocalCfg.Spawn(LocalInst);

		zen::IoHash	   Key;
		zen::IoHash	   PayloadId;
		zen::CbPackage Package = GeneratePackage(Key, PayloadId);
		auto		   Buf	   = ToBuffer(Package);

		// Store package upstream
		{
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}", UpstreamCfg.BaseUri, Bucket, Key)},
											cpr::Body{(const char*)Buf.GetData(), Buf.GetSize()},
											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 201);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}?Policy=QueryLocal,Store", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 404);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}?Policy=Query,Store", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 200);
		}
	}

	SUBCASE("store - 'local' does not store upstream (cbpackge)")
	{
		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
		ZenServerInstance UpstreamInst(TestEnv);
		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
		ZenServerInstance LocalInst(TestEnv);
		const auto		  Bucket = "legacy"sv;

		UpstreamCfg.Spawn(UpstreamInst);
		LocalCfg.Spawn(LocalInst);

		zen::IoHash	   Key;
		zen::IoHash	   PayloadId;
		zen::CbPackage Package = GeneratePackage(Key, PayloadId);
		auto		   Buf	   = ToBuffer(Package);

		// Store packge locally
		{
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}?Policy=Query,StoreLocal", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Body{(const char*)Buf.GetData(), Buf.GetSize()},
											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 201);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", UpstreamCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 404);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 200);
		}
	}

	SUBCASE("store - 'local/remote' stores local and upstream (cbpackage)")
	{
		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
		ZenServerInstance UpstreamInst(TestEnv);
		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
		ZenServerInstance LocalInst(TestEnv);
		const auto		  Bucket = "legacy"sv;

		UpstreamCfg.Spawn(UpstreamInst);
		LocalCfg.Spawn(LocalInst);

		zen::IoHash	   Key;
		zen::IoHash	   PayloadId;
		zen::CbPackage Package = GeneratePackage(Key, PayloadId);
		auto		   Buf	   = ToBuffer(Package);

		// Store package locally and upstream
		{
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}?Policy=Query,Store", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Body{(const char*)Buf.GetData(), Buf.GetSize()},
											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 201);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", UpstreamCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 200);
		}

		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", LocalCfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 200);
		}
	}

	SUBCASE("skip - 'data' returns cache record without attachments/empty payload")
	{
		ZenConfig		  Cfg = ZenConfig::New();
		ZenServerInstance Instance(TestEnv);
		const auto		  Bucket = "test"sv;

		Cfg.Spawn(Instance);

		zen::IoHash	   Key;
		zen::IoHash	   PayloadId;
		zen::CbPackage Package = GeneratePackage(Key, PayloadId);
		auto		   Buf	   = ToBuffer(Package);

		// Store package
		{
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}", Cfg.BaseUri, Bucket, Key)},
											cpr::Body{(const char*)Buf.GetData(), Buf.GetSize()},
											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});
			CHECK(Result.status_code == 201);
		}

		// Get package
		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}?Policy=Default,SkipData", Cfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
			CHECK(IsHttpSuccessCode(Result.status_code));
			IoBuffer  Buffer(IoBuffer::Wrap, Result.text.c_str(), Result.text.size());
			CbPackage ResponsePackage;
			CHECK(ResponsePackage.TryLoad(Buffer));
			CHECK(ResponsePackage.GetAttachments().size() == 0);
		}

		// Get record
		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}?Policy=Default,SkipData", Cfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/x-ue-cb"}});
			CHECK(IsHttpSuccessCode(Result.status_code));
			IoBuffer Buffer(IoBuffer::Wrap, Result.text.c_str(), Result.text.size());
			CbObject ResponseObject = zen::LoadCompactBinaryObject(Buffer);
			CHECK((bool)ResponseObject);
		}

		// Get payload
		{
			cpr::Response Result =
				cpr::Get(cpr::Url{fmt::format("{}/{}/{}/{}?Policy=Default,SkipData", Cfg.BaseUri, Bucket, Key, PayloadId)},
						 cpr::Header{{"Accept", "application/x-ue-comp"}});
			CHECK(IsHttpSuccessCode(Result.status_code));
			CHECK(Result.text.size() == 0);
		}
	}

	SUBCASE("skip - 'data' returns empty binary value")
	{
		ZenConfig		  Cfg = ZenConfig::New();
		ZenServerInstance Instance(TestEnv);
		const auto		  Bucket = "test"sv;

		Cfg.Spawn(Instance);

		zen::IoHash Key;
		auto		BinaryValue = GenerateData(1024, Key);

		// Store binary cache value
		{
			cpr::Response Result = cpr::Put(cpr::Url{fmt::format("{}/{}/{}", Cfg.BaseUri, Bucket, Key)},
											cpr::Body{(const char*)BinaryValue.GetData(), BinaryValue.GetSize()},
											cpr::Header{{"Content-Type", "application/octet-stream"}});
			CHECK(Result.status_code == 201);
		}

		// Get package
		{
			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}?Policy=Default,SkipData", Cfg.BaseUri, Bucket, Key)},
											cpr::Header{{"Accept", "application/octet-stream"}});
			CHECK(IsHttpSuccessCode(Result.status_code));
			CHECK(Result.text.size() == 0);
		}
	}
}

TEST_CASE("zcache.rpc")
{
	using namespace std::literals;

	auto AppendCacheRecord = [](CbPackage&			 Package,
								CbWriter&			 Writer,
								const zen::CacheKey& CacheKey,
								size_t				 PayloadSize,
								CachePolicy /*	 BatchDefaultPolicy */,
								CachePolicy RecordPolicy) {
		std::vector<uint8_t> Data;
		Data.resize(PayloadSize);
		for (size_t Idx = 0; Idx < PayloadSize; ++Idx)
		{
			Data[Idx] = Idx % 255;
		}

		CompressedBuffer  Value = zen::CompressedBuffer::Compress(SharedBuffer::MakeView(Data.data(), Data.size()));
		zen::CbAttachment Attachment(Value);

		Writer.BeginObject();
		{
			Writer.BeginObject("Record"sv);
			{
				Writer.BeginObject("Key"sv);
				{
					Writer << "Bucket"sv << CacheKey.Bucket << "Hash"sv << CacheKey.Hash;
				}
				Writer.EndObject();
				Writer.BeginArray("Values"sv);
				{
					Writer.BeginObject();
					{
						Writer.AddObjectId("Id"sv, Oid::NewOid());
						Writer.AddBinaryAttachment("RawHash"sv, IoHash::FromBLAKE3(Value.GetRawHash()));
						Writer.AddInteger("RawSize"sv, Value.GetRawSize());
					}
					Writer.EndObject();
				}
				Writer.EndArray();
			}
			Writer.EndObject();
			Writer.SetName("Policy"sv);
			CacheRecordPolicy(RecordPolicy).Save(Writer);
		}
		Writer.EndObject();

		Package.AddAttachment(Attachment);
	};

	auto ToIoBuffer = [](zen::CbPackage Package) -> zen::IoBuffer {
		zen::BinaryWriter MemStream;
		Package.Save(MemStream);
		return zen::IoBuffer(zen::IoBuffer::Clone, MemStream.Data(), MemStream.Size());
	};

	auto PutCacheRecords =
		[&AppendCacheRecord,
		 &ToIoBuffer](std::string_view BaseUri, std::string_view Bucket, size_t Num, size_t PayloadSize = 1024) -> std::vector<CacheKey> {
		std::vector<zen::CacheKey> OutKeys;

		for (uint32_t Key = 1; Key <= Num; ++Key)
		{
			zen::IoHash KeyHash;
			((uint32_t*)(KeyHash.Hash))[0] = Key;
			const zen::CacheKey CacheKey   = zen::CacheKey::Create(Bucket, KeyHash);
			CbPackage			Package;
			CbWriter			Writer;

			Writer.BeginObject();
			{
				Writer << "Method"sv
					   << "PutCacheRecords"sv;
				Writer.BeginObject("Params"sv);
				{
					CachePolicy BatchDefaultPolicy = CachePolicy::Default;
					Writer << "DefaultPolicy"sv << WriteToString<128>(BatchDefaultPolicy);
					Writer.BeginArray("Requests"sv);
					{
						AppendCacheRecord(Package, Writer, CacheKey, PayloadSize, BatchDefaultPolicy, CachePolicy::Default);
					}
					Writer.EndArray();
				}
				Writer.EndObject();
			}
			Writer.EndObject();
			Package.SetObject(Writer.Save().AsObject());

			OutKeys.push_back(CacheKey);

			IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", BaseUri)},
											 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});

			CHECK(Result.status_code == 200);
		}

		return OutKeys;
	};

	struct GetCacheRecordResult
	{
		zen::CbPackage				  Response;
		std::vector<zen::CbFieldView> Records;
		bool						  Success;
	};

	auto GetCacheRecords = [](std::string_view BaseUri, std::span<zen::CacheKey> Keys, zen::CachePolicy Policy) -> GetCacheRecordResult {
		using namespace zen;

		CbObjectWriter Request;
		Request << "Method"sv
				<< "GetCacheRecords"sv;
		Request.BeginObject("Params"sv);
		{
			Request << "DefaultPolicy"sv << WriteToString<128>(Policy);
			Request.BeginArray("Requests"sv);
			for (const CacheKey& Key : Keys)
			{
				Request.BeginObject();
				{
					Request.BeginObject("Key"sv);
					{
						Request << "Bucket"sv << Key.Bucket << "Hash"sv << Key.Hash;
					}
					Request.EndObject();
				}
				Request.EndObject();
			}
			Request.EndArray();
		}
		Request.EndObject();

		BinaryWriter Body;
		Request.Save(Body);

		cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", BaseUri)},
										 cpr::Header{{"Content-Type", "application/x-ue-cb"}, {"Accept", "application/x-ue-cbpkg"}},
										 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});

		GetCacheRecordResult OutResult;

		if (Result.status_code == 200)
		{
			CbPackage Response;
			if (Response.TryLoad(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size())))
			{
				OutResult.Response			= std::move(Response);
				CbObjectView ResponseObject = OutResult.Response.GetObject();

				for (CbFieldView RecordView : ResponseObject["Result"])
				{
					OutResult.Records.push_back(RecordView);
				}

				OutResult.Success = true;
			}
		}

		return OutResult;
	};

	auto LoadKey = [](zen::CbFieldView KeyView) -> zen::CacheKey {
		if (zen::CbObjectView KeyObj = KeyView.AsObjectView())
		{
			return CacheKey::Create(KeyObj["Bucket"sv].AsString(), KeyObj["Hash"].AsHash());
		}
		return CacheKey::Empty;
	};

	SUBCASE("get cache records")
	{
		std::filesystem::path TestDir	 = TestEnv.CreateNewTestDir();
		const uint16_t		  PortNumber = 13337;
		const auto			  BaseUri	 = fmt::format("http://localhost:{}/z$", PortNumber);

		ZenServerInstance Inst(TestEnv);
		Inst.SetTestDir(TestDir);
		Inst.SpawnServer(PortNumber);
		Inst.WaitUntilReady();

		CachePolicy				   Policy = CachePolicy::Default;
		std::vector<zen::CacheKey> Keys	  = PutCacheRecords(BaseUri, "mastodon"sv, 128);
		GetCacheRecordResult	   Result = GetCacheRecords(BaseUri, Keys, Policy);

		CHECK(Result.Records.size() == Keys.size());

		for (size_t Index = 0; CbFieldView RecordView : Result.Records)
		{
			const CacheKey& ExpectedKey = Keys[Index++];

			CbObjectView   RecordObj = RecordView.AsObjectView();
			CbObjectView   KeyObj	 = RecordObj["Key"sv].AsObjectView();
			const CacheKey Key		 = CacheKey::Create(KeyObj["Bucket"sv].AsString(), KeyObj["Hash"].AsHash());
			IoHash		   AttachmentHash;
			size_t		   NumValues = 0;
			for (CbFieldView Value : RecordObj["Values"sv])
			{
				AttachmentHash = Value.AsObjectView()["RawHash"sv].AsHash();
				++NumValues;
			}
			CHECK(NumValues == 1);
			const CbAttachment* Attachment = Result.Response.FindAttachment(AttachmentHash);

			CHECK(Key == ExpectedKey);
			CHECK(Attachment != nullptr);
		}
	}

	SUBCASE("get missing cache records")
	{
		std::filesystem::path TestDir	 = TestEnv.CreateNewTestDir();
		const uint16_t		  PortNumber = 13337;
		const auto			  BaseUri	 = fmt::format("http://localhost:{}/z$", PortNumber);

		ZenServerInstance Inst(TestEnv);
		Inst.SetTestDir(TestDir);
		Inst.SpawnServer(PortNumber);
		Inst.WaitUntilReady();

		CachePolicy				   Policy		= CachePolicy::Default;
		std::vector<zen::CacheKey> ExistingKeys = PutCacheRecords(BaseUri, "mastodon"sv, 128);
		std::vector<zen::CacheKey> Keys;

		for (const zen::CacheKey& Key : ExistingKeys)
		{
			Keys.push_back(Key);
			Keys.push_back(CacheKey::Create("missing"sv, IoHash::Zero));
		}

		GetCacheRecordResult Result = GetCacheRecords(BaseUri, Keys, Policy);

		CHECK(Result.Records.size() == Keys.size());

		size_t KeyIndex = 0;
		for (size_t Index = 0; CbFieldView RecordView : Result.Records)
		{
			const bool Missing = Index++ % 2 != 0;

			if (Missing)
			{
				CHECK(RecordView.IsNull());
			}
			else
			{
				const CacheKey& ExpectedKey = ExistingKeys[KeyIndex++];
				CbObjectView	RecordObj	= RecordView.AsObjectView();
				zen::CacheKey	Key			= LoadKey(RecordObj["Key"sv]);
				IoHash			AttachmentHash;
				size_t			NumValues = 0;
				for (CbFieldView Value : RecordObj["Values"sv])
				{
					AttachmentHash = Value.AsObjectView()["RawHash"sv].AsHash();
					++NumValues;
				}
				CHECK(NumValues == 1);
				const CbAttachment* Attachment = Result.Response.FindAttachment(AttachmentHash);
				CHECK(Key == ExpectedKey);
				CHECK(Attachment != nullptr);
			}
		}
	}

	SUBCASE("policy - 'QueryLocal' does not query upstream")
	{
		using namespace utils;

		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
		ZenServerInstance UpstreamServer(TestEnv);
		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
		ZenServerInstance LocalServer(TestEnv);

		SpawnServer(UpstreamServer, UpstreamCfg);
		SpawnServer(LocalServer, LocalCfg);

		std::vector<zen::CacheKey> Keys = PutCacheRecords(UpstreamCfg.BaseUri, "mastodon"sv, 4);

		CachePolicy			 Policy = CachePolicy::QueryLocal;
		GetCacheRecordResult Result = GetCacheRecords(LocalCfg.BaseUri, Keys, Policy);

		CHECK(Result.Records.size() == Keys.size());

		for (CbFieldView RecordView : Result.Records)
		{
			CHECK(RecordView.IsNull());
		}
	}

	SUBCASE("policy - 'QueryRemote' does query upstream")
	{
		using namespace utils;

		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
		ZenServerInstance UpstreamServer(TestEnv);
		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
		ZenServerInstance LocalServer(TestEnv);

		SpawnServer(UpstreamServer, UpstreamCfg);
		SpawnServer(LocalServer, LocalCfg);

		std::vector<zen::CacheKey> Keys = PutCacheRecords(UpstreamCfg.BaseUri, "mastodon"sv, 4);

		CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
		GetCacheRecordResult Result = GetCacheRecords(LocalCfg.BaseUri, Keys, Policy);

		CHECK(Result.Records.size() == Keys.size());

		for (size_t Index = 0; CbFieldView RecordView : Result.Response.GetObject()["Result"sv])
		{
			const zen::CacheKey& ExpectedKey = Keys[Index++];
			CbObjectView		 RecordObj	 = RecordView.AsObjectView();
			zen::CacheKey		 Key		 = LoadKey(RecordObj["Key"sv]);
			CHECK(Key == ExpectedKey);
		}
	}
}

TEST_CASE("zcache.rpc.allpolicies")
{
	using namespace std::literals;
	using namespace utils;

	ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
	ZenServerInstance UpstreamServer(TestEnv);
	ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
	ZenServerInstance LocalServer(TestEnv);
	const uint16_t	  LocalPortNumber = 13337;
	const auto		  BaseUri		  = fmt::format("http://localhost:{}/z$", LocalPortNumber);

	SpawnServer(UpstreamServer, UpstreamCfg);
	SpawnServer(LocalServer, LocalCfg);

	std::string_view TestVersion = "F72150A02AE34B57A9EC91D36BA1CE08"sv;
	std::string_view TestBucket	 = "allpoliciestest"sv;

	// NumKeys = (2 Value vs Record)*(2 SkipData vs Default)*(2 ForceMiss vs Not)*(2 use local)
	// *(2 use remote)*(2 UseValue Policy vs not)*(4 cases per type)
	constexpr int NumKeys	= 256;
	constexpr int NumValues = 4;
	Oid			  ValueIds[NumValues];
	IoHash		  Hash;
	for (int ValueIndex = 0; ValueIndex < NumValues; ++ValueIndex)
	{
		ExtendableStringBuilder<16> ValueName;
		ValueName << "ValueId_"sv << ValueIndex;
		static_assert(sizeof(IoHash) >= sizeof(Oid));
		ValueIds[ValueIndex] = Oid::FromMemory(IoHash::HashBuffer(ValueName.Data(), ValueName.Size() * sizeof(ValueName.Data()[0])).Hash);
	}

	struct KeyData;
	struct UserData
	{
		UserData& Set(KeyData* InKeyData, int InValueIndex)
		{
			Data	   = InKeyData;
			ValueIndex = InValueIndex;
			return *this;
		}
		KeyData* Data		= nullptr;
		int		 ValueIndex = 0;
	};
	struct KeyData
	{
		CompressedBuffer BufferValues[NumValues];
		uint64_t		 IntValues[NumValues];
		UserData		 ValueUserData[NumValues];
		bool			 ReceivedChunk[NumValues];
		CacheKey		 Key;
		UserData		 KeyUserData;
		uint32_t		 KeyIndex		  = 0;
		bool			 GetRequestsData  = true;
		bool			 UseValueAPI	  = false;
		bool			 UseValuePolicy	  = false;
		bool			 ForceMiss		  = false;
		bool			 UseLocal		  = true;
		bool			 UseRemote		  = true;
		bool			 ShouldBeHit	  = true;
		bool			 ReceivedPut	  = false;
		bool			 ReceivedGet	  = false;
		bool			 ReceivedPutValue = false;
		bool			 ReceivedGetValue = false;
	};
	struct CachePutRequest
	{
		CacheKey		  Key;
		CbObject		  Record;
		CacheRecordPolicy Policy;
		KeyData*		  Values;
		UserData*		  Data;
	};
	struct CachePutValueRequest
	{
		CacheKey		 Key;
		CompressedBuffer Value;
		CachePolicy		 Policy;
		UserData*		 Data;
	};
	struct CacheGetRequest
	{
		CacheKey		  Key;
		CacheRecordPolicy Policy;
		UserData*		  Data;
	};
	struct CacheGetValueRequest
	{
		CacheKey	Key;
		CachePolicy Policy;
		UserData*	Data;
	};
	struct CacheGetChunkRequest
	{
		CacheKey	Key;
		Oid			ValueId;
		uint64_t	RawOffset;
		uint64_t	RawSize;
		IoHash		RawHash;
		CachePolicy Policy;
		UserData*	Data;
	};

	KeyData							  KeyDatas[NumKeys];
	std::vector<CachePutRequest>	  PutRequests;
	std::vector<CachePutValueRequest> PutValueRequests;
	std::vector<CacheGetRequest>	  GetRequests;
	std::vector<CacheGetValueRequest> GetValueRequests;
	std::vector<CacheGetChunkRequest> ChunkRequests;

	for (uint32_t KeyIndex = 0; KeyIndex < NumKeys; ++KeyIndex)
	{
		IoHashStream KeyWriter;
		KeyWriter.Append(TestVersion.data(), TestVersion.length() * sizeof(TestVersion.data()[0]));
		KeyWriter.Append(&KeyIndex, sizeof(KeyIndex));
		IoHash	 KeyHash = KeyWriter.GetHash();
		KeyData& KeyData = KeyDatas[KeyIndex];

		KeyData.Key				 = CacheKey::Create(TestBucket, KeyHash);
		KeyData.KeyIndex		 = KeyIndex;
		KeyData.GetRequestsData	 = (KeyIndex & (1 << 1)) == 0;
		KeyData.UseValueAPI		 = (KeyIndex & (1 << 2)) != 0;
		KeyData.UseValuePolicy	 = (KeyIndex & (1 << 3)) != 0;
		KeyData.ForceMiss		 = (KeyIndex & (1 << 4)) == 0;
		KeyData.UseLocal		 = (KeyIndex & (1 << 5)) == 0;
		KeyData.UseRemote		 = (KeyIndex & (1 << 6)) == 0;
		KeyData.ShouldBeHit		 = !KeyData.ForceMiss && (KeyData.UseLocal || KeyData.UseRemote);
		CachePolicy SharedPolicy = KeyData.UseLocal ? CachePolicy::Local : CachePolicy::None;
		SharedPolicy |= KeyData.UseRemote ? CachePolicy::Remote : CachePolicy::None;
		CachePolicy PutPolicy = SharedPolicy;
		CachePolicy GetPolicy = SharedPolicy;
		GetPolicy |= !KeyData.GetRequestsData ? CachePolicy::SkipData : CachePolicy::None;
		CacheKey& Key = KeyData.Key;

		for (int ValueIndex = 0; ValueIndex < NumValues; ++ValueIndex)
		{
			KeyData.IntValues[ValueIndex] = static_cast<uint64_t>(KeyIndex) | (static_cast<uint64_t>(ValueIndex) << 32);
			KeyData.BufferValues[ValueIndex] =
				CompressedBuffer::Compress(SharedBuffer::MakeView(&KeyData.IntValues[ValueIndex], sizeof(KeyData.IntValues[ValueIndex])));
			KeyData.ReceivedChunk[ValueIndex] = false;
		}

		UserData& KeyUserData = KeyData.KeyUserData.Set(&KeyData, -1);
		for (int ValueIndex = 0; ValueIndex < NumValues; ++ValueIndex)
		{
			KeyData.ValueUserData[ValueIndex].Set(&KeyData, ValueIndex);
		}
		if (!KeyData.UseValueAPI)
		{
			CbObjectWriter Builder;
			Builder.BeginObject("key"sv);
			Builder << "Bucket"sv << Key.Bucket << "Hash"sv << Key.Hash;
			Builder.EndObject();
			Builder.BeginArray("Values"sv);
			for (int ValueIndex = 0; ValueIndex < NumValues; ++ValueIndex)
			{
				Builder.BeginObject();
				Builder.AddObjectId("Id"sv, ValueIds[ValueIndex]);
				Builder.AddBinaryAttachment("RawHash"sv, IoHash::FromBLAKE3(KeyData.BufferValues[ValueIndex].GetRawHash()));
				Builder.AddInteger("RawSize"sv, KeyData.BufferValues[ValueIndex].GetRawSize());
				Builder.EndObject();
			}
			Builder.EndArray();

			CacheRecordPolicy PutRecordPolicy;
			CacheRecordPolicy GetRecordPolicy;
			if (!KeyData.UseValuePolicy)
			{
				PutRecordPolicy = CacheRecordPolicy(PutPolicy);
				GetRecordPolicy = CacheRecordPolicy(GetPolicy);
			}
			else
			{
				// Switch the SkipData field in the Record policy so that if the CacheStore ignores the ValuePolicies
				// it will use the wrong value for SkipData and fail our tests.
				CacheRecordPolicyBuilder PutBuilder(PutPolicy ^ CachePolicy::SkipData);
				CacheRecordPolicyBuilder GetBuilder(GetPolicy ^ CachePolicy::SkipData);
				for (int ValueIndex = 0; ValueIndex < NumValues; ++ValueIndex)
				{
					PutBuilder.AddValuePolicy(ValueIds[ValueIndex], PutPolicy);
					GetBuilder.AddValuePolicy(ValueIds[ValueIndex], GetPolicy);
				}
				PutRecordPolicy = PutBuilder.Build();
				GetRecordPolicy = GetBuilder.Build();
			}
			if (!KeyData.ForceMiss)
			{
				PutRequests.push_back({Key, Builder.Save(), PutRecordPolicy, &KeyData, &KeyUserData});
			}
			GetRequests.push_back({Key, GetRecordPolicy, &KeyUserData});
			for (int ValueIndex = 0; ValueIndex < NumValues; ++ValueIndex)
			{
				UserData& ValueUserData = KeyData.ValueUserData[ValueIndex];
				ChunkRequests.push_back({Key, ValueIds[ValueIndex], 0, UINT64_MAX, IoHash(), GetPolicy, &ValueUserData});
			}
		}
		else
		{
			if (!KeyData.ForceMiss)
			{
				PutValueRequests.push_back({Key, KeyData.BufferValues[0], PutPolicy, &KeyUserData});
			}
			GetValueRequests.push_back({Key, GetPolicy, &KeyUserData});
			ChunkRequests.push_back({Key, Oid::Zero, 0, UINT64_MAX, IoHash(), GetPolicy, &KeyUserData});
		}
	}

	// PutCacheRecords
	{
		CbPackage	   Package;
		CbObjectWriter Writer;
		Writer << "Method"sv
			   << "PutCacheRecords"sv;
		Writer.BeginObject("Params"sv);
		{
			CachePolicy BatchDefaultPolicy = CachePolicy::Default;
			Writer << "DefaultPolicy"sv << WriteToString<128>(BatchDefaultPolicy);
			Writer.BeginArray("Requests"sv);
			for (CachePutRequest& Request : PutRequests)
			{
				Writer.BeginObject();
				{
					Writer.BeginObject("Record"sv);
					{
						Writer.BeginObject("Key"sv);
						{
							Writer << "Bucket"sv << Request.Key.Bucket << "Hash"sv << Request.Key.Hash;
						}
						Writer.EndObject();
						Writer.BeginArray("Values"sv);
						for (int ValueIndex = 0; ValueIndex < NumValues; ++ValueIndex)
						{
							Writer.BeginObject();
							{
								CompressedBuffer Buffer = Request.Values->BufferValues[ValueIndex];
								Writer.AddObjectId("Id"sv, ValueIds[ValueIndex]);
								Writer.AddBinaryAttachment("RawHash"sv, IoHash::FromBLAKE3(Buffer.GetRawHash()));
								Package.AddAttachment(CbAttachment(Buffer));
								Writer.AddInteger("RawSize"sv, Buffer.GetRawSize());
							}
							Writer.EndObject();
						}
						Writer.EndArray();
					}
					Writer.EndObject();
					Writer.SetName("Policy"sv);
					Request.Policy.Save(Writer);
				}
				Writer.EndObject();
				Request.Data->Data->ReceivedPut = true;
			}
			Writer.EndArray();
		}
		Writer.EndObject();
		Package.SetObject(Writer.Save());
		IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
		cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", BaseUri)},
										 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
										 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
		CHECK_MESSAGE(Result.status_code == 200, WriteToString<32>("PutCacheRecords unexpectedly failed.").c_str());
	}

	// PutCacheValues
	{
		CbPackage	   Package;
		CbObjectWriter Writer;
		Writer << "Method"sv
			   << "PutCacheValues"sv;
		Writer.BeginObject("Params"sv);
		{
			CachePolicy BatchDefaultPolicy = CachePolicy::Default;
			Writer << "DefaultPolicy"sv << WriteToString<128>(BatchDefaultPolicy);
			Writer.BeginArray("Requests"sv);
			for (CachePutValueRequest& Request : PutValueRequests)
			{
				Writer.BeginObject();
				{
					Writer.BeginObject("Key"sv);
					{
						Writer << "Bucket"sv << Request.Key.Bucket << "Hash"sv << Request.Key.Hash;
					}
					Writer.EndObject();
					CompressedBuffer Buffer = Request.Value;
					Writer.AddBinaryAttachment("RawHash"sv, IoHash::FromBLAKE3(Buffer.GetRawHash()));
					Package.AddAttachment(CbAttachment(Buffer));
					Writer.AddString("Policy"sv, WriteToString<128>(Request.Policy));
				}
				Writer.EndObject();
				Request.Data->Data->ReceivedPutValue = true;
			}
			Writer.EndArray();
		}
		Writer.EndObject();
		Package.SetObject(Writer.Save());
		IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
		cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", BaseUri)},
										 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
										 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
		CHECK_MESSAGE(Result.status_code == 200, WriteToString<32>("PutCacheValues unexpectedly failed.").c_str());
	}

	for (KeyData& KeyData : KeyDatas)
	{
		if (!KeyData.ForceMiss)
		{
			if (!KeyData.UseValueAPI)
			{
				CHECK_MESSAGE(KeyData.ReceivedPut, WriteToString<32>("Key ", KeyData.KeyIndex, " was unexpectedly not put.").c_str());
			}
			else
			{
				CHECK_MESSAGE(KeyData.ReceivedPutValue,
							  WriteToString<32>("Key ", KeyData.KeyIndex, " was unexpectedly not put to ValueAPI.").c_str());
			}
		}
	}

	// GetCacheRecords
	{
		CbPackage	   Package;
		CbObjectWriter Writer;
		Writer << "Method"sv
			   << "GetCacheRecords"sv;
		Writer.BeginObject("Params"sv);
		{
			CachePolicy BatchDefaultPolicy = CachePolicy::Default;
			Writer << "DefaultPolicy"sv << WriteToString<128>(BatchDefaultPolicy);
			Writer.BeginArray("Requests"sv);
			for (CacheGetRequest& Request : GetRequests)
			{
				Writer.BeginObject();
				{
					Writer.BeginObject("Key"sv);
					{
						Writer << "Bucket"sv << Request.Key.Bucket << "Hash"sv << Request.Key.Hash;
					}
					Writer.EndObject();
					Writer.SetName("Policy"sv);
					Request.Policy.Save(Writer);
				}
				Writer.EndObject();
			}
			Writer.EndArray();
		}
		Writer.EndObject();
		Package.SetObject(Writer.Save());
		IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
		cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", BaseUri)},
										 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
										 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
		CHECK_MESSAGE(Result.status_code == 200, WriteToString<32>("GetCacheRecords unexpectedly failed.").c_str());
		CbPackage Response;
		bool	  Loaded = Response.TryLoad(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size()));
		CHECK_MESSAGE(Loaded, WriteToString<32>("GetCacheRecords response failed to load.").c_str());
		CbObjectView ResponseObject = Response.GetObject();
		CbArrayView	 Responses		= ResponseObject["Result"sv].AsArrayView();
		CHECK_MESSAGE(Responses.Num() == GetRequests.size(),
					  WriteToString<32>("GetCacheRecords response count did not match request count.").c_str());
		int Index = 0;
		for (CbFieldView ResponseField : Responses)
		{
			CbObjectView RecordView = ResponseField.AsObjectView();
			bool		 Succeeded	= !ResponseField.HasError();

			CacheGetRequest& Request = GetRequests[Index++];
			KeyData*		 KeyData = Request.Data->Data;
			KeyData->ReceivedGet	 = true;
			WriteToString<32> Name("Get(", KeyData->KeyIndex, ")");

			if (KeyData->ShouldBeHit)
			{
				CHECK_MESSAGE(Succeeded, WriteToString<32>(Name, " unexpectedly failed.").c_str());
			}
			else if (KeyData->ForceMiss)
			{
				CHECK_MESSAGE(!Succeeded, WriteToString<32>(Name, " unexpectedly succeeded.").c_str());
			}
			if (!KeyData->ForceMiss && Succeeded)
			{
				CbArrayView ValuesArray = RecordView["Values"sv].AsArrayView();
				CHECK_MESSAGE(ValuesArray.Num() == NumValues, WriteToString<32>(Name, " number of values did not match.").c_str());

				for (CbFieldView ValueField : ValuesArray)
				{
					CbObjectView ValueObject		= ValueField.AsObjectView();
					Oid			 ActualValueId		= ValueObject["Id"sv].AsObjectId();
					int			 ExpectedValueIndex = 0;
					for (; ExpectedValueIndex < NumValues; ++ExpectedValueIndex)
					{
						if (ValueIds[ExpectedValueIndex] == ActualValueId)
						{
							break;
						}
					}
					CHECK_MESSAGE(ExpectedValueIndex < NumValues, WriteToString<32>(Name, " could not find matching ValueId.").c_str());
					WriteToString<32>	ValueName("Get(", KeyData->KeyIndex, ",", ExpectedValueIndex, ")");
					IoHash				ActualRawHash = ValueObject["RawHash"sv].AsHash();
					const CbAttachment* Attachment	  = Response.FindAttachment(ActualRawHash);
					CompressedBuffer	ActualBuffer  = Attachment ? Attachment->AsCompressedBinary() : CompressedBuffer();
					uint64_t			ActualRawSize = UINT64_MAX;
					if (ActualBuffer)
					{
						ActualRawSize = ActualBuffer.GetRawSize();
					}
					else
					{
						ActualRawSize = ValueObject["RawSize"sv].AsUInt64(UINT64_MAX);
					}
					CompressedBuffer ExpectedValue = KeyData->BufferValues[ExpectedValueIndex];
					CHECK_MESSAGE(ActualRawHash == IoHash::FromBLAKE3(ExpectedValue.GetRawHash()),
								  WriteToString<32>(ValueName, " RawHash did not match.").c_str());
					CHECK_MESSAGE(ActualRawSize == ExpectedValue.GetRawSize(),
								  WriteToString<32>(ValueName, " RawSize did not match.").c_str());

					if (KeyData->GetRequestsData)
					{
						SharedBuffer Buffer = ActualBuffer.Decompress();
						CHECK_MESSAGE(Buffer.GetSize() == ActualRawSize,
									  WriteToString<32>(ValueName, " BufferSize did not match RawSize.").c_str());
						uint64_t ActualIntValue	  = ((const uint64_t*)Buffer.GetData())[0];
						uint64_t ExpectedIntValue = KeyData->IntValues[ExpectedValueIndex];
						CHECK_MESSAGE(ActualIntValue == ExpectedIntValue, WriteToString<32>(ValueName, " had unexpected data.").c_str());
					}
				}
			}
		}
	}

	// GetCacheValues
	{
		CbPackage	   Package;
		CbObjectWriter Writer;
		Writer << "Method"sv
			   << "GetCacheValues"sv;
		Writer.BeginObject("Params"sv);
		{
			CachePolicy BatchDefaultPolicy = CachePolicy::Default;
			Writer << "DefaultPolicy"sv << WriteToString<128>(BatchDefaultPolicy);
			Writer.BeginArray("Requests"sv);
			for (CacheGetValueRequest& Request : GetValueRequests)
			{
				Writer.BeginObject();
				{
					Writer.BeginObject("Key"sv);
					{
						Writer << "Bucket"sv << Request.Key.Bucket << "Hash"sv << Request.Key.Hash;
					}
					Writer.EndObject();
					Writer.AddString("Policy"sv, WriteToString<128>(Request.Policy));
				}
				Writer.EndObject();
			}
			Writer.EndArray();
		}
		Writer.EndObject();
		Package.SetObject(Writer.Save());
		IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
		cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", BaseUri)},
										 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
										 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
		CHECK_MESSAGE(Result.status_code == 200, WriteToString<32>("GetCacheValues unexpectedly failed.").c_str());
		CbPackage Response;
		bool	  Loaded = Response.TryLoad(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size()));
		CHECK_MESSAGE(Loaded, WriteToString<32>("GetCacheValues response failed to load.").c_str());
		CbObjectView ResponseObject = Response.GetObject();
		CbArrayView	 Responses		= ResponseObject["Result"sv].AsArrayView();
		CHECK_MESSAGE(Responses.Num() == GetValueRequests.size(),
					  WriteToString<32>("GetCacheValues response count did not match request count.").c_str());
		int Index = 0;
		for (CbFieldView RequestResultView : Responses)
		{
			CbObjectView RequestResultObject = RequestResultView.AsObjectView();
			CbFieldView	 RawHashField		 = RequestResultObject["RawHash"sv];
			IoHash		 ActualRawHash		 = RawHashField.AsHash();
			bool		 Succeeded			 = !RawHashField.HasError();

			CacheGetValueRequest& Request = GetValueRequests[Index++];
			KeyData*			  KeyData = Request.Data->Data;
			KeyData->ReceivedGetValue	  = true;
			WriteToString<32> Name("GetValue("sv, KeyData->KeyIndex, ")"sv);

			if (KeyData->ShouldBeHit)
			{
				CHECK_MESSAGE(Succeeded, WriteToString<32>(Name, " unexpectedly failed.").c_str());
			}
			else if (KeyData->ForceMiss)
			{
				CHECK_MESSAGE(!Succeeded, WriteToString<32>(Name, "unexpectedly succeeded.").c_str());
			}
			if (!KeyData->ForceMiss && Succeeded)
			{
				const CbAttachment* Attachment	  = Response.FindAttachment(ActualRawHash);
				CompressedBuffer	ActualBuffer  = Attachment ? Attachment->AsCompressedBinary() : CompressedBuffer();
				uint64_t			ActualRawSize = UINT64_MAX;
				if (ActualBuffer)
				{
					ActualRawSize = ActualBuffer.GetRawSize();
				}
				else
				{
					ActualRawSize = RequestResultObject["RawSize"sv].AsUInt64(UINT64_MAX);
				}
				CompressedBuffer ExpectedValue = KeyData->BufferValues[0];
				CHECK_MESSAGE(ActualRawHash == IoHash::FromBLAKE3(ExpectedValue.GetRawHash()),
							  WriteToString<32>(Name, " RawHash did not match.").c_str());
				CHECK_MESSAGE(ActualRawSize == ExpectedValue.GetRawSize(), WriteToString<32>(Name, " RawSize did not match.").c_str());

				if (KeyData->GetRequestsData)
				{
					SharedBuffer Buffer = ActualBuffer.Decompress();
					CHECK_MESSAGE(Buffer.GetSize() == ActualRawSize, WriteToString<32>(Name, " BufferSize did not match RawSize.").c_str());
					uint64_t ActualIntValue	  = ((const uint64_t*)Buffer.GetData())[0];
					uint64_t ExpectedIntValue = KeyData->IntValues[0];
					CHECK_MESSAGE(ActualIntValue == ExpectedIntValue, WriteToString<32>(Name, " had unexpected data.").c_str());
				}
			}
		}
	}

	// GetCacheChunks
	{
		CbPackage	   Package;
		CbObjectWriter Writer;
		std::sort(ChunkRequests.begin(), ChunkRequests.end(), [](CacheGetChunkRequest& A, CacheGetChunkRequest& B) {
			return A.Key.Hash < B.Key.Hash;
		});
		Writer << "Method"sv
			   << "GetCacheChunks"sv;
		Writer.BeginObject("Params"sv);
		{
			CachePolicy BatchDefaultPolicy = CachePolicy::Default;
			Writer << "DefaultPolicy"sv << WriteToString<128>(BatchDefaultPolicy);
			Writer.BeginArray("ChunkRequests"sv);
			for (CacheGetChunkRequest& Request : ChunkRequests)
			{
				Writer.BeginObject();
				{
					Writer.BeginObject("Key"sv);
					{
						Writer << "Bucket"sv << Request.Key.Bucket << "Hash"sv << Request.Key.Hash;
					}
					Writer.EndObject();
					Writer.AddObjectId("ValueId"sv, Request.ValueId);
					Writer.AddInteger("RawOffset"sv, Request.RawOffset);
					Writer.AddInteger("RawSize"sv, Request.RawSize);
					Writer.AddHash("ChunkId"sv, IoHash());
					Writer.AddString("Policy"sv, WriteToString<128>(Request.Policy));
				}
				Writer.EndObject();
			}
			Writer.EndArray();
		}
		Writer.EndObject();
		Package.SetObject(Writer.Save());
		IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
		cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", BaseUri)},
										 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
										 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
		CHECK_MESSAGE(Result.status_code == 200, WriteToString<32>("GetCacheChunks unexpectedly failed.").c_str());
		CbPackage Response;
		bool	  Loaded = Response.TryLoad(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size()));
		CHECK_MESSAGE(Loaded, WriteToString<32>("GetCacheChunks response failed to load.").c_str());
		CbObjectView ResponseObject = Response.GetObject();
		CbArrayView	 Responses		= ResponseObject["Result"sv].AsArrayView();
		CHECK_MESSAGE(Responses.Num() == ChunkRequests.size(),
					  WriteToString<32>("GetCacheChunks response count did not match request count.").c_str());
		int Index = 0;
		for (CbFieldView RequestResultView : Responses)
		{
			CbObjectView RequestResultObject = RequestResultView.AsObjectView();
			CbFieldView	 RawHashField		 = RequestResultObject["RawHash"sv];
			IoHash		 ActualRawHash		 = RawHashField.AsHash();
			bool		 Succeeded			 = !RawHashField.HasError();

			CacheGetChunkRequest& Request	   = ChunkRequests[Index++];
			KeyData*			  KeyData	   = Request.Data->Data;
			int					  ValueIndex   = Request.Data->ValueIndex >= 0 ? Request.Data->ValueIndex : 0;
			KeyData->ReceivedChunk[ValueIndex] = true;
			WriteToString<32> Name("GetChunks("sv, KeyData->KeyIndex, ","sv, ValueIndex, ")"sv);

			if (KeyData->ShouldBeHit)
			{
				CHECK_MESSAGE(Succeeded, WriteToString<256>(Name, " unexpectedly failed."sv).c_str());
			}
			else if (KeyData->ForceMiss)
			{
				CHECK_MESSAGE(!Succeeded, WriteToString<256>(Name, " unexpectedly succeeded."sv).c_str());
			}
			if (KeyData->ShouldBeHit && Succeeded)
			{
				const CbAttachment* Attachment	  = Response.FindAttachment(ActualRawHash);
				CompressedBuffer	ActualBuffer  = Attachment ? Attachment->AsCompressedBinary() : CompressedBuffer();
				uint64_t			ActualRawSize = UINT64_MAX;
				if (ActualBuffer)
				{
					ActualRawSize = ActualBuffer.GetRawSize();
				}
				else
				{
					ActualRawSize = RequestResultObject["RawSize"sv].AsUInt64(UINT64_MAX);
				}
				CompressedBuffer ExpectedValue = KeyData->BufferValues[ValueIndex];
				CHECK_MESSAGE(ActualRawHash == IoHash::FromBLAKE3(ExpectedValue.GetRawHash()),
							  WriteToString<32>(Name, " had unexpected RawHash.").c_str());
				CHECK_MESSAGE(ActualRawSize == ExpectedValue.GetRawSize(), WriteToString<32>(Name, " had unexpected RawSize.").c_str());

				if (KeyData->GetRequestsData)
				{
					SharedBuffer Buffer = ActualBuffer.Decompress();
					CHECK_MESSAGE(Buffer.GetSize() == ActualRawSize, WriteToString<32>(Name, " BufferSize did not match RawSize.").c_str());
					uint64_t ActualIntValue	  = ((const uint64_t*)Buffer.GetData())[0];
					uint64_t ExpectedIntValue = KeyData->IntValues[ValueIndex];
					CHECK_MESSAGE(ActualIntValue == ExpectedIntValue, WriteToString<32>(Name, " had unexpected data.").c_str());
				}
			}
		}
	}

	for (KeyData& KeyData : KeyDatas)
	{
		if (!KeyData.UseValueAPI)
		{
			CHECK_MESSAGE(KeyData.ReceivedGet, WriteToString<32>("Get(", KeyData.KeyIndex, ") was unexpectedly not received.").c_str());
			for (int ValueIndex = 0; ValueIndex < NumValues; ++ValueIndex)
			{
				CHECK_MESSAGE(
					KeyData.ReceivedChunk[ValueIndex],
					WriteToString<32>("GetChunks(", KeyData.KeyIndex, ",", ValueIndex, ") was unexpectedly not received.").c_str());
			}
		}
		else
		{
			CHECK_MESSAGE(KeyData.ReceivedGetValue,
						  WriteToString<32>("GetValue(", KeyData.KeyIndex, ") was unexpectedly not received.").c_str());
			CHECK_MESSAGE(KeyData.ReceivedChunk[0],
						  WriteToString<32>("GetChunks(", KeyData.KeyIndex, ") was unexpectedly not received.").c_str());
		}
	}
}

#	if ZEN_WITH_EXEC_SERVICES

struct RemoteExecutionRequest
{
	RemoteExecutionRequest(std::string_view Host, int Port, std::filesystem::path& TreePath)
	: m_HostName(Host)
	, m_PortNumber(Port)
	, m_TreePath(TreePath)
	{
	}

	void Build(std::string_view Command, std::string_view Arguments)
	{
		zen::FileSystemTraversal Traversal;
		Traversal.TraverseFileSystem(m_TreePath, m_Visit);

		zen::CbObjectWriter PrepReq;

		PrepReq << "cmd" << Command;
		PrepReq << "args" << Arguments;

		PrepReq.BeginArray("files");

		for (const auto& Kv : m_Visit.m_Files)
		{
			PrepReq.BeginObject();
			PrepReq << "file" << zen::PathToUtf8(Kv.first) << "size" << Kv.second.Size << "hash" << Kv.second.Hash;
			PrepReq.EndObject();
		}
		PrepReq.EndArray();

		PrepReq.Save(m_MemOut);
	}

	void Prep()
	{
		cpr::Response Response =
			cpr::Post(cpr::Url(fmt::format("{}/prep", m_BaseUri)), cpr::Body((const char*)m_MemOut.Data(), m_MemOut.Size()));

		if (Response.status_code < 300)
		{
			zen::IoBuffer Payload(zen::IoBuffer::Clone, Response.text.data(), Response.text.size());
			zen::CbObject Result = zen::LoadCompactBinaryObject(Payload);

			for (auto& Need : Result["need"])
			{
				zen::IoHash NeedHash = Need.AsHash();

				if (auto It = m_Visit.m_HashToFile.find(NeedHash); It != m_Visit.m_HashToFile.end())
				{
					zen::IoBuffer FileData = zen::IoBufferBuilder::MakeFromFile(It->second);

					cpr::Response CasResponse = cpr::Post(cpr::Url(m_CasUri), cpr::Body((const char*)FileData.Data(), FileData.Size()));

					if (CasResponse.status_code >= 300)
					{
						ZEN_ERROR("CAS put failed with {}", CasResponse.status_code);
					}
				}
				else
				{
					ZEN_ERROR("unknown hash in 'need' list: {}", NeedHash);
				}
			}
		}
	}

	zen::CbObject Exec()
	{
		cpr::Response JobResponse = cpr::Post(cpr::Url(m_BaseUri), cpr::Body((const char*)m_MemOut.Data(), m_MemOut.Size()));

		if (JobResponse.status_code < 300)
		{
			zen::IoBuffer Payload(zen::IoBuffer::Clone, JobResponse.text.data(), JobResponse.text.size());
			return zen::LoadCompactBinaryObject(std::move(Payload));
		}

		ZEN_INFO("job exec: {}", JobResponse.status_code);
		return {};
	}

private:
	struct Visitor : public zen::FileSystemTraversal::TreeVisitor
	{
		const std::filesystem::path& m_RootPath;

		Visitor(const std::filesystem::path& RootPath) : m_RootPath(RootPath) {}

		virtual void VisitFile(const std::filesystem::path& Parent, const path_view& FileName, uint64_t FileSize) override
		{
			std::filesystem::path FullPath = Parent / FileName;

			zen::IoHashStream Ios;
			zen::ScanFile(FullPath, 64 * 1024, [&](const void* Data, size_t Size) { Ios.Append(Data, Size); });
			zen::IoHash Hash = Ios.GetHash();

			auto RelativePath = FullPath.lexically_relative(m_RootPath).native();
			//			ZEN_INFO("File: {:32} => {} ({})", zen::PathToUtf8(RelativePath), Hash, FileSize);

			FileEntry& Entry = m_Files[RelativePath];
			Entry.Hash		 = Hash;
			Entry.Size		 = FileSize;

			m_HashToFile[Hash] = FullPath;
		}

		virtual bool VisitDirectory(const std::filesystem::path& Parent, const path_view& DirectoryName) override
		{
			std::filesystem::path FullPath = Parent / DirectoryName;

			if (DirectoryName.starts_with('.'))
			{
				return false;
			}

			return true;
		}

		struct FileEntry
		{
			uint64_t	Size;
			zen::IoHash Hash;
		};

		std::map<std::filesystem::path::string_type, FileEntry>						m_Files;
		std::unordered_map<zen::IoHash, std::filesystem::path, zen::IoHash::Hasher> m_HashToFile;
	};

	std::string			  m_HostName;
	int					  m_PortNumber;
	std::filesystem::path m_TreePath;
	const std::string	  m_BaseUri = fmt::format("http://{}:{}/exec/jobs", m_HostName, m_PortNumber);
	const std::string	  m_CasUri	= fmt::format("http://{}:{}/cas", m_HostName, m_PortNumber);
	Visitor				  m_Visit{m_TreePath};
	zen::BinaryWriter	  m_MemOut;
};

TEST_CASE("exec.basic" * doctest::skip(true))
{
#		if ZEN_WITH_EXEC_SERVICES
	using namespace std::literals;

	std::filesystem::path TestDir = TestEnv.CreateNewTestDir();

	const uint16_t PortNumber = 13337;

	ZenServerInstance Zen1(TestEnv);
	Zen1.SetTestDir(TestDir);
	Zen1.SpawnServer(PortNumber);
	Zen1.WaitUntilReady();

	std::filesystem::path TreePath = TestEnv.GetTestRootDir("test/remote1");

	{
		RemoteExecutionRequest RemoteRequest("localhost", PortNumber, TreePath);
		RemoteRequest.Build("zentest-appstub.exe", "");
		RemoteRequest.Prep();
		zen::CbObject Result = RemoteRequest.Exec();

		CHECK(Result["exitcode"sv].AsInt32(-1) == 0);
	}

	{
		RemoteExecutionRequest RemoteRequest("localhost", PortNumber, TreePath);
		RemoteRequest.Build("zentest-appstub.exe", "-f=1");
		RemoteRequest.Prep();
		zen::CbObject Result = RemoteRequest.Exec();

		CHECK(Result["exitcode"sv].AsInt32(-1) == 1);
	}
#		endif	// ZEN_WITH_EXEC_SERVICES
}

TEST_CASE("mesh.basic")
{
// --mesh option only available with ZEN_ENABLE_MESH
#		if ZEN_ENABLE_MESH
	using namespace std::literals;

	const int kInstanceCount = 4;

	ZEN_INFO("spawning {} instances", kInstanceCount);

	std::unique_ptr<ZenServerInstance> Instances[kInstanceCount];

	for (int i = 0; i < kInstanceCount; ++i)
	{
		auto& Instance = Instances[i];

		Instance = std::make_unique<ZenServerInstance>(TestEnv);
		Instance->SetTestDir(TestEnv.CreateNewTestDir());
		Instance->EnableMesh();
		Instance->SpawnServer(13337 + i);
	}

	for (int i = 0; i < kInstanceCount; ++i)
	{
		auto& Instance = Instances[i];

		Instance->WaitUntilReady();
	}
#		endif
}

#	endif	// ZEN_WITH_EXEC_SERVICES

class ZenServerTestHelper
{
public:
	ZenServerTestHelper(std::string_view HelperId, int ServerCount) : m_HelperId{HelperId}, m_ServerCount{ServerCount} {}
	~ZenServerTestHelper() {}

	void SpawnServers()
	{
		SpawnServers([](ZenServerInstance&) {});
	}

	void SpawnServers(auto&& Callback)
	{
		ZEN_INFO("{}: spawning {} server instances", m_HelperId, m_ServerCount);

		m_Instances.resize(m_ServerCount);

		for (int i = 0; i < m_ServerCount; ++i)
		{
			auto& Instance = m_Instances[i];

			Instance = std::make_unique<ZenServerInstance>(TestEnv);
			Instance->SetTestDir(TestEnv.CreateNewTestDir());

			Callback(*Instance);

			Instance->SpawnServer(13337 + i);
		}

		for (int i = 0; i < m_ServerCount; ++i)
		{
			auto& Instance = m_Instances[i];

			Instance->WaitUntilReady();
		}
	}

	ZenServerInstance& GetInstance(int Index) { return *m_Instances[Index]; }

private:
	std::string										 m_HelperId;
	int												 m_ServerCount = 0;
	std::vector<std::unique_ptr<ZenServerInstance> > m_Instances;
};

class IoDispatcher
{
public:
	IoDispatcher(asio::io_context& IoCtx) : m_IoCtx(IoCtx) {}
	~IoDispatcher() { Stop(); }

	void Run()
	{
		Stop();

		m_Running = true;

		m_IoThread = std::thread([this]() {
			try
			{
				m_IoCtx.run();
			}
			catch (std::exception& Error)
			{
				m_Error = Error;
			}
		});
	}

	void Stop()
	{
		if (m_Running)
		{
			m_Running = false;

			if (m_IoThread.joinable())
			{
				m_IoThread.join();
			}
		}
	}

	bool IsRunning() const { return m_Running; }

	const std::exception& Error() { return m_Error; }

private:
	asio::io_context& m_IoCtx;
	std::thread		  m_IoThread;
	std::exception	  m_Error;
	std::atomic_bool  m_Running{false};
};

TEST_CASE("http.basics")
{
	using namespace std::literals;

	ZenServerTestHelper Servers{"http.basics"sv, 1};
	Servers.SpawnServers();

	ZenServerInstance& Instance = Servers.GetInstance(0);
	const std::string  BaseUri	= Instance.GetBaseUri();

	{
		cpr::Response r = cpr::Get(cpr::Url{fmt::format("{}/testing/hello", BaseUri)});
		CHECK(IsHttpSuccessCode(r.status_code));
	}

	{
		cpr::Response r = cpr::Post(cpr::Url{fmt::format("{}/testing/hello", BaseUri)});
		CHECK_EQ(r.status_code, 404);
	}

	{
		cpr::Response r = cpr::Post(cpr::Url{fmt::format("{}/testing/echo", BaseUri)}, cpr::Body{"yoyoyoyo"});
		CHECK_EQ(r.status_code, 200);
		CHECK_EQ(r.text, "yoyoyoyo");
	}
}

TEST_CASE("http.package")
{
	using namespace std::literals;

	ZenServerTestHelper Servers{"http.package"sv, 1};
	Servers.SpawnServers();

	ZenServerInstance& Instance = Servers.GetInstance(0);
	const std::string  BaseUri	= Instance.GetBaseUri();

	static const uint8_t Data1[] = {0, 1, 2, 3};
	static const uint8_t Data2[] = {0, 1, 2, 3, 4, 5, 6, 7, 8};

	zen::CbAttachment Attach1{zen::CompressedBuffer::Compress(zen::SharedBuffer::Clone({Data1, 4}),
															  zen::OodleCompressor::NotSet,
															  zen::OodleCompressionLevel::None)};
	zen::CbAttachment Attach2{zen::CompressedBuffer::Compress(zen::SharedBuffer::Clone({Data2, 8}),
															  zen::OodleCompressor::NotSet,
															  zen::OodleCompressionLevel::None)};

	zen::CbObjectWriter Writer;

	Writer.AddAttachment("attach1", Attach1);
	Writer.AddAttachment("attach2", Attach2);

	zen::CbObject CoreObject = Writer.Save();

	zen::CbPackage TestPackage;
	TestPackage.SetObject(CoreObject);
	TestPackage.AddAttachment(Attach1);
	TestPackage.AddAttachment(Attach2);

	zen::HttpClient			  TestClient(BaseUri);
	zen::HttpClient::Response Response = TestClient.TransactPackage("/testing/package"sv, TestPackage);

	zen::CbPackage ResponsePackage = ParsePackageMessage(Response.ResponsePayload);

	CHECK_EQ(ResponsePackage, TestPackage);
}

TEST_CASE("websocket.basic" * doctest::skip(true))
{
	std::filesystem::path TestDir	  = TestEnv.CreateNewTestDir();
	const uint16_t		  PortNumber  = 13337;
	const auto			  MaxWaitTime = std::chrono::seconds(5);

	ZenServerInstance Inst(TestEnv);
	Inst.SetTestDir(TestDir);
	Inst.SpawnServer(PortNumber, "--websocket-port=8848"sv);
	Inst.WaitUntilReady();

	asio::io_context IoCtx;
	IoDispatcher	 IoDispatcher(IoCtx);
	auto			 WebSocket = WebSocketClient::Create(IoCtx);

	auto ConnectFuture = WebSocket->Connect({.Host = "127.0.0.1", .Port = 8848, .Endpoint = "/zen"});
	IoDispatcher.Run();

	ConnectFuture.wait_for(MaxWaitTime);
	CHECK(ConnectFuture.get());

	for (size_t Idx = 0; Idx < 10; Idx++)
	{
		CbObjectWriter Request;
		Request << "Method"sv
				<< "SayHello"sv;

		WebSocketMessage RequestMsg;
		RequestMsg.SetMessageType(WebSocketMessageType::kRequest);
		RequestMsg.SetBody(Request.Save());

		auto ResponseFuture = WebSocket->SendRequest(std::move(RequestMsg));
		ResponseFuture.wait_for(MaxWaitTime);

		CbObject		 Response = ResponseFuture.get().Body().GetObject();
		std::string_view Message  = Response["Result"].AsString();

		CHECK(Message == "Hello Friend!!"sv);
	}

	WebSocket->Disconnect();

	IoCtx.stop();
	IoDispatcher.Stop();
}

#	if 0
TEST_CASE("lifetime.owner")
{
	// This test is designed to verify that the hand-over of sponsor processes is handled
	// correctly for the case when a second or third process is launched on the same port
	//
	// Due to the nature of it, it cannot be

	const uint16_t PortNumber = 23456;

	ZenServerInstance	  Zen1(TestEnv);
	std::filesystem::path TestDir1 = TestEnv.CreateNewTestDir();
	Zen1.SetTestDir(TestDir1);
	Zen1.SpawnServer(PortNumber);
	Zen1.WaitUntilReady();
	Zen1.Detach();

	ZenServerInstance	  Zen2(TestEnv);
	std::filesystem::path TestDir2 = TestEnv.CreateNewTestDir();
	Zen2.SetTestDir(TestDir2);
	Zen2.SpawnServer(PortNumber);
	Zen2.WaitUntilReady();
	Zen2.Detach();
}

TEST_CASE("lifetime.owner.2")
{
	// This test is designed to verify that the hand-over of sponsor processes is handled
	// correctly for the case when a second or third process is launched on the same port
	//
	// Due to the nature of it, it cannot be

	const uint16_t PortNumber = 13456;

	std::filesystem::path TestDir1 = TestEnv.CreateNewTestDir();
	std::filesystem::path TestDir2 = TestEnv.CreateNewTestDir();

	ZenServerInstance Zen1(TestEnv);
	Zen1.SetTestDir(TestDir1);
	Zen1.SpawnServer(PortNumber);
	Zen1.WaitUntilReady();

	ZenServerInstance Zen2(TestEnv);
	Zen2.SetTestDir(TestDir2);
	Zen2.SetOwnerPid(Zen1.GetPid());
	Zen2.SpawnServer(PortNumber + 1);
	Zen2.Detach();

	ZenServerInstance Zen3(TestEnv);
	Zen3.SetTestDir(TestDir2);
	Zen3.SetOwnerPid(Zen1.GetPid());
	Zen3.SpawnServer(PortNumber + 1);
	Zen3.Detach();

	ZenServerInstance Zen4(TestEnv);
	Zen4.SetTestDir(TestDir2);
	Zen4.SetOwnerPid(Zen1.GetPid());
	Zen4.SpawnServer(PortNumber + 1);
	Zen4.Detach();
}
#	endif

}  // namespace zen::tests
#else
int
main()
{
}
#endif