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

#include "compactcas.h"

#include <zenstore/cas.h>

#include <zencore/except.h>
#include <zencore/filesystem.h>
#include <zencore/fmtutils.h>
#include <zencore/logging.h>
#include <zencore/workthreadpool.h>
#include <gsl/gsl-lite.hpp>

#if ZEN_WITH_TESTS
#	include <zencore/compactbinarybuilder.h>
#	include <zencore/testing.h>
#	include <zencore/testutils.h>
#	include <zenstore/cidstore.h>
#	include <algorithm>
#	include <random>
#endif

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

namespace zen {

struct CasDiskIndexHeader
{
	static constexpr uint32_t ExpectedMagic	 = 0x75696478;	// 'uidx';
	static constexpr uint32_t CurrentVersion = 1;
	uint32_t				  Magic			 = ExpectedMagic;
	uint32_t				  Version		 = CurrentVersion;
	uint32_t				  PayloadAlignement;
	uint32_t				  Reserved0 = 0;
	uint64_t				  EntryCount;
	uint32_t				  Reserved1 = 0;
	uint32_t				  Reserved2 = 0;
};

static_assert(sizeof(CasDiskIndexHeader) == 32);

namespace {
	std::filesystem::path BuildUcasPath(const std::filesystem::path& BlocksBasePath, const uint32_t BlockIndex)
	{
		ExtendablePathBuilder<256> Path;

		char BlockHexString[9];
		FormatHex(BlockIndex, BlockHexString);

		Path.Append(BlocksBasePath);
		Path.AppendSeparator();
		Path.AppendAsciiRange(BlockHexString, BlockHexString + 4);
		Path.AppendSeparator();
		Path.Append(BlockHexString);
		Path.Append(".ucas");
		return Path.ToPath();
	}

}  // namespace

struct CasContainerStrategy::ChunkBlock
{
	explicit ChunkBlock(const std::filesystem::path& BlockPath);
	~ChunkBlock();
	const std::filesystem::path& GetPath() const;
	void						 Open();
	void						 Create(uint64_t InitialSize);
	void						 MarkAsDeleteOnClose(std::error_code& Ec);
	uint64_t					 FileSize();
	IoBuffer					 GetChunk(uint64_t Offset, uint64_t Size);
	void						 Read(void* Data, uint64_t Size, uint64_t FileOffset);
	void						 Write(const void* Data, uint64_t Size, uint64_t FileOffset);
	void						 Flush();
	void StreamByteRange(uint64_t FileOffset, uint64_t Size, std::function<void(const void* Data, uint64_t Size)>&& ChunkFun);

private:
	void						InternalOpen();
	const std::filesystem::path m_Path;
	RwLock						m_OpenLock;
	BasicFile					m_File;
	IoBuffer					m_IoBuffer;
};

CasContainerStrategy::ChunkBlock::ChunkBlock(const std::filesystem::path& BlockPath) : m_Path(BlockPath)
{
}

CasContainerStrategy::ChunkBlock::~ChunkBlock()
{
	RwLock::ExclusiveLockScope _(m_OpenLock);
	m_File.Detach();
}

const std::filesystem::path&
CasContainerStrategy::ChunkBlock::GetPath() const
{
	return m_Path;
}

void
CasContainerStrategy::ChunkBlock::InternalOpen()
{
	if (m_File.Handle())
	{
		return;
	}
	m_File.Open(m_Path, false);
	void* FileHandle = m_File.Handle();
	m_IoBuffer		 = IoBuffer(IoBuffer::File, FileHandle, 0, m_File.FileSize());
}

void
CasContainerStrategy::ChunkBlock::Open()
{
	RwLock::ExclusiveLockScope _(m_OpenLock);
	InternalOpen();
}

void
CasContainerStrategy::ChunkBlock::Create(uint64_t InitialSize)
{
	RwLock::ExclusiveLockScope _(m_OpenLock);

	auto ParentPath = m_Path.parent_path();
	if (!std::filesystem::is_directory(ParentPath))
	{
		CreateDirectories(ParentPath);
	}

	m_File.Open(m_Path, true);
	if (InitialSize > 0)
	{
		m_File.SetFileSize(InitialSize);
	}
	void* FileHandle = m_File.Handle();
	m_IoBuffer		 = IoBuffer(IoBuffer::File, FileHandle, 0, InitialSize);
}

uint64_t
CasContainerStrategy::ChunkBlock::FileSize()
{
	RwLock::SharedLockScope _(m_OpenLock);
	return m_File.FileSize();
}

void
CasContainerStrategy::ChunkBlock::MarkAsDeleteOnClose(std::error_code& Ec)
{
	RwLock::ExclusiveLockScope _(m_OpenLock);
	if (m_File.Handle())
	{
		m_File.MarkAsDeleteOnClose(Ec);
		return;
	}
	if (std::filesystem::is_regular_file(m_Path))
	{
		Ec.clear();
		std::filesystem::remove(m_Path, Ec);
	}
}

IoBuffer
CasContainerStrategy::ChunkBlock::GetChunk(uint64_t Offset, uint64_t Size)
{
	InternalOpen();
	return IoBuffer(m_IoBuffer, Offset, Size);
}

void
CasContainerStrategy::ChunkBlock::Read(void* Data, uint64_t Size, uint64_t FileOffset)
{
	RwLock::SharedLockScope _(m_OpenLock);
	m_File.Read(Data, Size, FileOffset);
}

void
CasContainerStrategy::ChunkBlock::Write(const void* Data, uint64_t Size, uint64_t FileOffset)
{
	RwLock::SharedLockScope _(m_OpenLock);
	m_File.Write(Data, Size, FileOffset);
}

void
CasContainerStrategy::ChunkBlock::Flush()
{
	RwLock::ExclusiveLockScope _(m_OpenLock);
	if (!m_File.Handle())
	{
		return;
	}
	m_File.Flush();
}

void
CasContainerStrategy::ChunkBlock::StreamByteRange(uint64_t												 FileOffset,
												  uint64_t												 Size,
												  std::function<void(const void* Data, uint64_t Size)>&& ChunkFun)
{
	RwLock::SharedLockScope _(m_OpenLock);
	m_File.StreamByteRange(FileOffset, Size, std::move(ChunkFun));
}

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

CasContainerStrategy::CasContainerStrategy(const CasStoreConfiguration& Config, CasGc& Gc)
: GcStorage(Gc)
, m_Config(Config)
, m_Log(logging::Get("containercas"))
{
}

CasContainerStrategy::~CasContainerStrategy()
{
}

void
CasContainerStrategy::Initialize(const std::string_view ContainerBaseName, uint32_t MaxBlockSize, uint64_t Alignment, bool IsNewStore)
{
	ZEN_ASSERT(IsPow2(Alignment));
	ZEN_ASSERT(!m_IsInitialized);
	ZEN_ASSERT(MaxBlockSize > 0);

	m_ContainerBaseName = ContainerBaseName;
	m_PayloadAlignment	= Alignment;
	m_MaxBlockSize		= MaxBlockSize;
	m_BlocksBasePath	= m_Config.RootDirectory / m_ContainerBaseName / "blocks";

	OpenContainer(IsNewStore);

	m_IsInitialized = true;
}

CasStore::InsertResult
CasContainerStrategy::InsertChunk(const void* ChunkData, size_t ChunkSize, const IoHash& ChunkHash)
{
	uint32_t					WriteBlockIndex;
	std::shared_ptr<ChunkBlock> WriteBlock;
	uint64_t					InsertOffset;
	{
		RwLock::ExclusiveLockScope _i(m_InsertLock);

		{
			RwLock::SharedLockScope _l(m_LocationMapLock);
			auto					KeyIt = m_LocationMap.find(ChunkHash);

			if (KeyIt != m_LocationMap.end())
			{
				return CasStore::InsertResult{.New = false};
			}
		}

		// New entry

		WriteBlockIndex = m_WriteBlockIndex.load(std::memory_order_acquire);
		WriteBlock		= m_WriteBlock.lock();
		if (!WriteBlock || (m_CurrentInsertOffset + ChunkSize) > m_MaxBlockSize)
		{
			if (WriteBlock)
			{
				WriteBlock->Flush();
			}
			{
				RwLock::ExclusiveLockScope __(m_LocationMapLock);
				if (m_ChunkBlocks.size() == CasDiskLocation::MaxBlockIndex)
				{
					throw std::runtime_error(fmt::format("unable to allocate a new block in {}", m_ContainerBaseName));
				}
				WriteBlockIndex += WriteBlock ? 1 : 0;
				while (m_ChunkBlocks.contains(WriteBlockIndex))
				{
					WriteBlockIndex = (WriteBlockIndex + 1) & CasDiskLocation::MaxBlockIndex;
				}
				auto BlockPath				   = BuildUcasPath(m_BlocksBasePath, WriteBlockIndex);
				WriteBlock					   = std::make_shared<ChunkBlock>(BlockPath);
				m_ChunkBlocks[WriteBlockIndex] = WriteBlock;
				m_WriteBlockIndex.store(WriteBlockIndex, std::memory_order_release);
			}
			m_WriteBlock		  = WriteBlock;
			m_CurrentInsertOffset = 0;
			WriteBlock->Create(m_MaxBlockSize);
		}
		else
		{
			WriteBlock->Open();
		}
		InsertOffset		  = m_CurrentInsertOffset;
		m_CurrentInsertOffset = RoundUp(InsertOffset + ChunkSize, m_PayloadAlignment);
	}

	WriteBlock->Write(ChunkData, ChunkSize, InsertOffset);

	const CasLocation Location(WriteBlockIndex, InsertOffset, ChunkSize);
	CasDiskIndexEntry IndexEntry{.Key = ChunkHash, .Location = CasDiskLocation(Location, m_PayloadAlignment)};

	m_TotalSize.fetch_add(static_cast<uint64_t>(ChunkSize));
	{
		RwLock::ExclusiveLockScope __(m_LocationMapLock);
		m_LocationMap.emplace(ChunkHash, CasDiskLocation(Location, m_PayloadAlignment));
		m_CasLog.Append(IndexEntry);
	}

	return CasStore::InsertResult{.New = true};
}

CasStore::InsertResult
CasContainerStrategy::InsertChunk(IoBuffer Chunk, const IoHash& ChunkHash)
{
	return InsertChunk(Chunk.Data(), Chunk.Size(), ChunkHash);
}

IoBuffer
CasContainerStrategy::FindChunk(const IoHash& ChunkHash)
{
	std::shared_ptr<ChunkBlock> ChunkBlock;
	CasLocation					Location;
	{
		RwLock::SharedLockScope _(m_LocationMapLock);
		auto					KeyIt = m_LocationMap.find(ChunkHash);
		if (KeyIt == m_LocationMap.end())
		{
			return IoBuffer();
		}
		Location	 = KeyIt->second.Get(m_PayloadAlignment);
		auto BlockIt = m_ChunkBlocks.find(Location.BlockIndex);
		if (BlockIt == m_ChunkBlocks.end())
		{
			return IoBuffer();
		}
		ChunkBlock = BlockIt->second;
	}
	return ChunkBlock->GetChunk(Location.Offset, Location.Size);
}

bool
CasContainerStrategy::HaveChunk(const IoHash& ChunkHash)
{
	RwLock::SharedLockScope _(m_LocationMapLock);
	return m_LocationMap.contains(ChunkHash);
}

void
CasContainerStrategy::FilterChunks(CasChunkSet& InOutChunks)
{
	// This implementation is good enough for relatively small
	// chunk sets (in terms of chunk identifiers), but would
	// benefit from a better implementation which removes
	// items incrementally for large sets, especially when
	// we're likely to already have a large proportion of the
	// chunks in the set

	InOutChunks.RemoveChunksIf([&](const IoHash& Hash) { return HaveChunk(Hash); });
}

void
CasContainerStrategy::Flush()
{
	{
		RwLock::ExclusiveLockScope _(m_InsertLock);
		if (m_CurrentInsertOffset > 0)
		{
			uint32_t WriteBlockIndex = m_WriteBlockIndex.load(std::memory_order_acquire);
			auto	 WriteBlock		 = m_WriteBlock.lock();
			WriteBlockIndex++;
			while (m_ChunkBlocks.contains(WriteBlockIndex))
			{
				WriteBlockIndex = (WriteBlockIndex + 1) & CasDiskLocation::MaxBlockIndex;
			}
			WriteBlock->Flush();
			m_WriteBlock.reset();
			m_WriteBlockIndex.store(WriteBlockIndex, std::memory_order_release);
			m_CurrentInsertOffset = 0;
		}
	}
	MakeIndexSnapshot();
}

void
CasContainerStrategy::Scrub(ScrubContext& Ctx)
{
	const uint64_t WindowSize = 4 * 1024 * 1024;

	std::vector<CasDiskIndexEntry> BigChunks;
	std::vector<CasDiskIndexEntry> BadChunks;

	// We do a read sweep through the payloads file and validate
	// any entries that are contained within each segment, with
	// the assumption that most entries will be checked in this
	// pass. An alternative strategy would be to use memory mapping.

	{
		IoBuffer ReadBuffer{WindowSize};
		void*	 BufferBase = ReadBuffer.MutableData();

		RwLock::SharedLockScope _(m_InsertLock);  // TODO: Refactor so we don't have to keep m_InsertLock all the time?
		RwLock::SharedLockScope __(m_LocationMapLock);

		for (const auto& Block : m_ChunkBlocks)
		{
			uint64_t WindowStart = 0;
			uint64_t WindowEnd	 = WindowSize;
			auto&	 BlockFile	 = *Block.second;
			BlockFile.Open();
			const uint64_t FileSize = BlockFile.FileSize();

			do
			{
				const uint64_t ChunkSize = Min(WindowSize, FileSize - WindowStart);
				BlockFile.Read(BufferBase, ChunkSize, WindowStart);

				for (auto& Entry : m_LocationMap)
				{
					const CasLocation Location	  = Entry.second.Get(m_PayloadAlignment);
					const uint64_t	  EntryOffset = Location.Offset;

					if ((EntryOffset >= WindowStart) && (EntryOffset < WindowEnd))
					{
						const uint64_t EntryEnd = EntryOffset + Location.Size;

						if (EntryEnd >= WindowEnd)
						{
							BigChunks.push_back({.Key = Entry.first, .Location = Entry.second});

							continue;
						}

						const IoHash ComputedHash =
							IoHash::HashBuffer(reinterpret_cast<uint8_t*>(BufferBase) + Location.Offset - WindowStart, Location.Size);

						if (Entry.first != ComputedHash)
						{
							// Hash mismatch
							BadChunks.push_back({.Key = Entry.first, .Location = Entry.second});
						}
					}
				}

				WindowStart += WindowSize;
				WindowEnd += WindowSize;
			} while (WindowStart < FileSize);
		}

		// Deal with large chunks

		for (const CasDiskIndexEntry& Entry : BigChunks)
		{
			IoHashStream	  Hasher;
			const CasLocation Location	= Entry.Location.Get(m_PayloadAlignment);
			auto&			  BlockFile = *m_ChunkBlocks[Location.BlockIndex];
			BlockFile.StreamByteRange(Location.Offset, Location.Size, [&](const void* Data, uint64_t Size) { Hasher.Append(Data, Size); });
			IoHash ComputedHash = Hasher.GetHash();

			if (Entry.Key != ComputedHash)
			{
				BadChunks.push_back(Entry);
			}
		}
	}

	if (BadChunks.empty())
	{
		return;
	}

	ZEN_ERROR("Scrubbing found {} bad chunks in '{}'", BadChunks.size(), m_ContainerBaseName);

	// Deal with bad chunks by removing them from our lookup map

	std::vector<IoHash> BadChunkHashes;

	RwLock::ExclusiveLockScope _(m_LocationMapLock);
	for (const CasDiskIndexEntry& Entry : BadChunks)
	{
		BadChunkHashes.push_back(Entry.Key);
		m_CasLog.Append({.Key = Entry.Key, .Location = Entry.Location, .Flags = CasDiskIndexEntry::kTombstone});
		m_LocationMap.erase(Entry.Key);
	}

	// Let whomever it concerns know about the bad chunks. This could
	// be used to invalidate higher level data structures more efficiently
	// than a full validation pass might be able to do

	Ctx.ReportBadCasChunks(BadChunkHashes);
}

void
CasContainerStrategy::CollectGarbage(GcContext& GcCtx)
{
	namespace fs = std::filesystem;

	// It collects all the blocks that we want to delete chunks from. For each such
	// block we keep a list of chunks to retain.
	//
	// It will first remove any chunks that are flushed from the m_LocationMap.
	//
	// It then checks to see if we want to purge any chunks that are in the currently
	// active block. If so, we break off the current block and start on a new block,
	// otherwise we just let the active block be.
	//
	// Next it will iterate over all blocks that we want to remove chunks from.
	// If the block is empty after removal of chunks we mark the block as pending
	// delete - we want to delete it as soon as there are no IoBuffers using the
	// block file.
	//
	// If the block is non-empty we write out the chunks we want to keep to a new
	// block file (creating new block files as needed).
	//
	// We update the index as we complete each new block file. This makes it possible
	// to break the GC if we want to limit time for execution.
	//
	// GC can fairly parallell to regular operation - it will block while figuring
	// out which chunks to remove and what blocks to rewrite but the actual
	// reading and writing of data to new block files does not block regular operation.
	//
	// While moving blocks it will do a blocking operation and update the m_LocationMap
	// after each new block is written and it will also block when figuring out the
	// path to the next new block.

	ZEN_INFO("collecting garbage from '{}'", m_Config.RootDirectory / m_ContainerBaseName);
	std::unordered_map<IoHash, CasDiskLocation, IoHash::Hasher> LocationMap;
	size_t														BlockCount;
	{
		RwLock::SharedLockScope _i(m_InsertLock);
		RwLock::SharedLockScope _l(m_LocationMapLock);
		LocationMap.reserve(m_LocationMap.size());
		bool	 IsWriting	  = !m_WriteBlock.expired();
		uint32_t WritingBlock = m_WriteBlockIndex.load(std::memory_order_acquire);
		for (const auto& Entry : m_LocationMap)
		{
			if (IsWriting && Entry.second.GetBlockIndex() == WritingBlock)
			{
				continue;
			}
			LocationMap.emplace(Entry.first, Entry.second);
		}
		BlockCount = m_ChunkBlocks.size();
	}

	if (LocationMap.empty())
	{
		ZEN_INFO("garbage collect SKIPPED, for '{}', container is empty", m_Config.RootDirectory / m_ContainerBaseName);
		return;
	}

	const uint64_t TotalChunkCount = LocationMap.size();

	std::unordered_set<uint32_t>							BlocksToReWrite;
	std::unordered_map<uint32_t, size_t>					BlockIndexToChunkMapIndex;
	std::vector<std::unordered_set<IoHash, IoHash::Hasher>> KeepChunks;
	std::vector<std::unordered_set<IoHash, IoHash::Hasher>> DeleteChunks;

	BlocksToReWrite.reserve(BlockCount);
	BlockIndexToChunkMapIndex.reserve(BlockCount);
	KeepChunks.reserve(BlockCount);
	DeleteChunks.reserve(BlockCount);
	size_t GuesstimateCountPerBlock = TotalChunkCount / BlockCount / 2;

	std::vector<IoHash> TotalChunkHashes;
	TotalChunkHashes.reserve(TotalChunkCount);
	for (const auto& Entry : LocationMap)
	{
		TotalChunkHashes.push_back(Entry.first);
		uint32_t BlockIndex = Entry.second.GetBlockIndex();
		if (BlockIndexToChunkMapIndex.contains(BlockIndex))
		{
			continue;
		}
		BlockIndexToChunkMapIndex[BlockIndex] = KeepChunks.size();
		KeepChunks.resize(KeepChunks.size() + 1);
		KeepChunks.back().reserve(GuesstimateCountPerBlock);
		DeleteChunks.resize(DeleteChunks.size() + 1);
		DeleteChunks.back().reserve(GuesstimateCountPerBlock);
	}

	const bool PerformDelete = GcCtx.IsDeletionMode() && GcCtx.CollectSmallObjects();
	uint64_t   DeleteCount	 = {};
	uint64_t   MoveCount	 = {};

	uint64_t NewTotalSize = 0;
	GcCtx.FilterCas(TotalChunkHashes, [&](const IoHash& ChunkHash, bool Keep) {
		auto	 KeyIt		   = LocationMap.find(ChunkHash);
		uint32_t BlockIndex	   = KeyIt->second.GetBlockIndex();
		size_t	 ChunkMapIndex = BlockIndexToChunkMapIndex[BlockIndex];
		if (Keep)
		{
			auto& ChunkMap = KeepChunks[ChunkMapIndex];
			ChunkMap.insert(ChunkHash);
			NewTotalSize += KeyIt->second.GetSize();
		}
		else
		{
			auto& ChunkMap = DeleteChunks[ChunkMapIndex];
			ChunkMap.insert(ChunkHash);
			DeleteCount++;
			BlocksToReWrite.insert(BlockIndex);
		}
	});

	if (!PerformDelete)
	{
		uint64_t TotalSize = m_TotalSize.load(std::memory_order_relaxed);
		ZEN_INFO("garbage collect from '{}' DISABLED, found #{} {} chunks of total #{} {}",
				 m_Config.RootDirectory / m_ContainerBaseName,
				 DeleteCount,
				 NiceBytes(TotalSize - NewTotalSize),
				 TotalChunkCount,
				 NiceBytes(TotalSize));
		return;
	}

	// Move all chunks in blocks that have chunks removed to new blocks

	std::shared_ptr<ChunkBlock> NewBlockFile;
	uint64_t					WriteOffset	  = {};
	uint32_t					NewBlockIndex = {};
	std::vector<IoHash>			DeletedChunks;
	DeletedChunks.reserve(DeleteCount);
	std::vector<IoHash> MovedChunks;
	DeletedChunks.reserve(MoveCount);

	std::unordered_map<IoHash, CasDiskLocation> MovedBlockChunks;
	for (auto BlockIndex : BlocksToReWrite)
	{
		const size_t ChunkMapIndex = BlockIndexToChunkMapIndex[BlockIndex];
		const auto&	 KeepMap	   = KeepChunks[ChunkMapIndex];
		if (KeepMap.empty())
		{
			std::shared_ptr<ChunkBlock> BlockFile;
			{
				RwLock::ExclusiveLockScope _i(m_LocationMapLock);
				const auto&				   DeleteMap = DeleteChunks[ChunkMapIndex];
				for (const auto& ChunkHash : DeleteMap)
				{
					auto	 KeyIt	   = m_LocationMap.find(ChunkHash);
					uint64_t ChunkSize = KeyIt->second.GetSize();
					m_CasLog.Append({.Key = ChunkHash, .Location = KeyIt->second, .Flags = CasDiskIndexEntry::kTombstone});
					m_TotalSize.fetch_sub(ChunkSize);
					m_LocationMap.erase(KeyIt);
				}
				DeletedChunks.insert(DeletedChunks.end(), DeleteMap.begin(), DeleteMap.end());
				m_ChunkBlocks[BlockIndex].swap(BlockFile);
			}
			ZEN_DEBUG("marking cas store file for delete {}, block {}", m_ContainerBaseName, std::to_string(BlockIndex));
			std::error_code Ec;
			BlockFile->MarkAsDeleteOnClose(Ec);
			if (Ec)
			{
				ZEN_WARN("Failed to flag file '{}' for deletion: '{}'", BlockFile->GetPath(), Ec.message());
			}
			continue;
		}

		std::shared_ptr<ChunkBlock> OldBlockFile;
		{
			RwLock::SharedLockScope _i(m_LocationMapLock);
			OldBlockFile = m_ChunkBlocks[BlockIndex];
			OldBlockFile->Open();
		}

		MovedChunks.reserve(MovedChunks.size() + KeepMap.size());
		std::vector<uint8_t> Chunk;
		for (const auto& ChunkHash : KeepMap)
		{
			auto			  KeyIt			= LocationMap.find(ChunkHash);
			const CasLocation ChunkLocation = KeyIt->second.Get(m_PayloadAlignment);
			Chunk.resize(ChunkLocation.Size);
			OldBlockFile->Read(Chunk.data(), Chunk.size(), ChunkLocation.Offset);

			if (!NewBlockFile || (WriteOffset + Chunk.size() > m_MaxBlockSize))
			{
				uint32_t NextBlockIndex = m_WriteBlockIndex.load(std::memory_order::memory_order_relaxed);
				{
					RwLock::ExclusiveLockScope _l(m_LocationMapLock);
					if (NewBlockFile)
					{
						for (const auto& MovedEntry : MovedBlockChunks)
						{
							m_LocationMap[MovedEntry.first] = MovedEntry.second;
							m_CasLog.Append({.Key = MovedEntry.first, .Location = MovedEntry.second});
							MovedChunks.push_back(MovedEntry.first);
						}
						if (m_ChunkBlocks.size() == CasDiskLocation::MaxBlockIndex)
						{
							ZEN_ERROR("unable to allocate a new block in {}, count limit {} exeeded",
									  m_ContainerBaseName,
									  static_cast<uint64_t>(std::numeric_limits<uint32_t>::max()) + 1);
							return;
						}
					}
					if (m_ChunkBlocks.size() == CasDiskLocation::MaxBlockIndex)
					{
						throw std::runtime_error(fmt::format("unable to allocate a new block in {}", m_ContainerBaseName));
					}
					while (m_ChunkBlocks.contains(NextBlockIndex))
					{
						NextBlockIndex = (NextBlockIndex + 1) & CasDiskLocation::MaxBlockIndex;
					}
					auto NewBlockPath			  = BuildUcasPath(m_BlocksBasePath, NextBlockIndex);
					NewBlockFile				  = std::make_shared<ChunkBlock>(NewBlockPath);
					m_ChunkBlocks[NextBlockIndex] = NewBlockFile;
				}

				std::error_code Error;
				DiskSpace		Space = DiskSpaceInfo(m_Config.RootDirectory, Error);
				if (Error)
				{
					ZEN_ERROR("get disk space in {} FAILED, reason '{}'", m_ContainerBaseName, Error.message());
					return;
				}

				if (Space.Free < m_MaxBlockSize)
				{
					std::filesystem::path GCReservePath = m_Config.RootDirectory / (m_ContainerBaseName + ".gc.reserve.ucas");
					if (!std::filesystem::is_regular_file(GCReservePath))
					{
						ZEN_INFO("garbage collect from '{}' FAILED, required disk space {}, free {}",
								 m_Config.RootDirectory / m_ContainerBaseName,
								 m_MaxBlockSize,
								 NiceBytes(Space.Free));
						RwLock::ExclusiveLockScope _l(m_LocationMapLock);
						m_ChunkBlocks.erase(NextBlockIndex);
						return;
					}

					ZEN_INFO("using gc reserve for '{}', disk free {}",
							 m_Config.RootDirectory / m_ContainerBaseName,
							 NiceBytes(Space.Free));
					auto NewBlockPath = BuildUcasPath(m_BlocksBasePath, NextBlockIndex);
					std::filesystem::rename(GCReservePath, NewBlockPath);
					NewBlockFile->Open();
				}
				else
				{
					NewBlockFile->Create(m_MaxBlockSize);
				}
				NewBlockIndex = NextBlockIndex;
				MovedBlockChunks.clear();
				WriteOffset = 0;
			}

			NewBlockFile->Write(Chunk.data(), Chunk.size(), WriteOffset);
			CasLocation NewChunkLocation(NewBlockIndex, WriteOffset, Chunk.size());
			MovedBlockChunks.emplace(ChunkHash, CasDiskLocation(NewChunkLocation, m_PayloadAlignment));
			WriteOffset = RoundUp(WriteOffset + Chunk.size(), m_PayloadAlignment);
		}
		Chunk.clear();

		if (NewBlockFile)
		{
			const auto& DeleteMap = DeleteChunks[ChunkMapIndex];
			// Remap moved chunks to the new block file
			{
				RwLock::ExclusiveLockScope _l(m_LocationMapLock);
				for (const auto& MovedEntry : MovedBlockChunks)
				{
					m_LocationMap[MovedEntry.first] = MovedEntry.second;
					m_CasLog.Append({.Key = MovedEntry.first, .Location = MovedEntry.second});
					MovedChunks.push_back(MovedEntry.first);
				}

				for (const auto& ChunkHash : DeleteMap)
				{
					auto	 KeyIt	   = m_LocationMap.find(ChunkHash);
					uint64_t ChunkSize = KeyIt->second.GetSize();
					m_CasLog.Append({.Key = ChunkHash, .Location = KeyIt->second, .Flags = CasDiskIndexEntry::kTombstone});
					m_TotalSize.fetch_sub(ChunkSize);
					m_LocationMap.erase(KeyIt);
				}
				m_ChunkBlocks[BlockIndex].reset();
			}
			MovedBlockChunks.clear();
			DeletedChunks.insert(DeletedChunks.end(), DeleteMap.begin(), DeleteMap.end());
		}
		ZEN_DEBUG("marking cas store file for delete {}, block index {}", m_ContainerBaseName, BlockIndex);
		std::error_code Ec;
		OldBlockFile->MarkAsDeleteOnClose(Ec);
		if (Ec)
		{
			ZEN_WARN("Failed to flag file '{}' for deletion: '{}'", OldBlockFile->GetPath(), Ec.message());
		}
		OldBlockFile.reset();
	}

	GcCtx.DeletedCas(DeletedChunks);

	ZEN_INFO("garbage collection complete '{}', deleted {} and moved {} of {} chunks",
			 m_Config.RootDirectory / m_ContainerBaseName,
			 DeletedChunks.size(),
			 MovedChunks.size(),
			 TotalChunkCount);

	std::filesystem::path GCReservePath = m_Config.RootDirectory / (m_ContainerBaseName + ".gc.reserve.ucas");
	if (std::filesystem::is_regular_file(GCReservePath))
	{
		return;
	}

	std::error_code Error;
	DiskSpace		Space = DiskSpaceInfo(m_Config.RootDirectory, Error);
	if (Error)
	{
		ZEN_ERROR("get disk space in {} FAILED, reason '{}'", m_ContainerBaseName, Error.message());
		return;
	}
	if (Space.Free < m_MaxBlockSize)
	{
		ZEN_INFO("not enough space for garbage collect reserve '{}' FAILED, required disk space {}, free {}",
				 m_Config.RootDirectory / m_ContainerBaseName,
				 m_MaxBlockSize,
				 NiceBytes(Space.Free));
		return;
	}
	BasicFile GCReserveFile;
	GCReserveFile.Open(GCReservePath, true);
	GCReserveFile.SetFileSize(m_MaxBlockSize);

	ZEN_DEBUG("recreated garbage collect reserve '{}' FAILED, {} bytes",
			  m_Config.RootDirectory / m_ContainerBaseName,
			  NiceBytes(Space.Free));
}

void
CasContainerStrategy::MakeIndexSnapshot()
{
	ZEN_INFO("writing index snapshot for '{}'", m_Config.RootDirectory / m_ContainerBaseName);

	namespace fs = std::filesystem;

	fs::path SlogPath		   = m_Config.RootDirectory / (m_ContainerBaseName + ".ulog");
	fs::path SidxPath		   = m_Config.RootDirectory / (m_ContainerBaseName + ".uidx");
	fs::path STmplogPath	   = m_Config.RootDirectory / (m_ContainerBaseName + ".tmp.ulog");
	fs::path STmpSidxPath	   = m_Config.RootDirectory / (m_ContainerBaseName + ".tmp.uidx");
	fs::path SRecoveredlogPath = m_Config.RootDirectory / (m_ContainerBaseName + ".recover.ulog");

	// Index away, we keep it if something goes wrong
	if (fs::is_regular_file(STmpSidxPath))
	{
		fs::remove(STmpSidxPath);
	}
	if (fs::is_regular_file(SidxPath))
	{
		fs::rename(SidxPath, STmpSidxPath);
	}

	// Move cas away, we keep it if something goes wrong, any new chunks will be added to the new log
	{
		RwLock::ExclusiveLockScope _(m_InsertLock);
		RwLock::ExclusiveLockScope __(m_LocationMapLock);
		m_CasLog.Close();

		if (fs::is_regular_file(STmplogPath))
		{
			fs::remove(STmplogPath);
		}

		fs::rename(SlogPath, STmplogPath);

		// Open an new log
		m_CasLog.Open(SlogPath, true);
	}

	try
	{
		// Write the current state of the location map to a new index state
		std::vector<CasDiskIndexEntry> Entries;

		{
			RwLock::SharedLockScope _l(m_LocationMapLock);
			Entries.resize(m_LocationMap.size());

			uint64_t EntryIndex = 0;
			for (auto& Entry : m_LocationMap)
			{
				CasDiskIndexEntry& IndexEntry = Entries[EntryIndex++];
				IndexEntry.Key				  = Entry.first;
				IndexEntry.Location			  = Entry.second;
			}
		}

		BasicFile ObjectIndexFile;
		ObjectIndexFile.Open(SidxPath, true);
		CasDiskIndexHeader Header = {.PayloadAlignement = gsl::narrow<uint32_t>(m_PayloadAlignment), .EntryCount = Entries.size()};
		ObjectIndexFile.Write(&Header, sizeof(CasDiskIndexEntry), 0);
		ObjectIndexFile.Write(Entries.data(), Entries.size() * sizeof(CasDiskIndexEntry), sizeof(CasDiskIndexEntry));
		ObjectIndexFile.Close();
	}
	catch (std::exception& Err)
	{
		ZEN_ERROR("snapshot FAILED, reason '{}'", Err.what());

		// Reconstruct the log from old log and any added log entries
		RwLock::ExclusiveLockScope _(m_LocationMapLock);
		if (fs::is_regular_file(STmplogPath))
		{
			std::vector<CasDiskIndexEntry> Records;
			Records.reserve(m_LocationMap.size());
			{
				TCasLogFile<CasDiskIndexEntry> OldCasLog;
				OldCasLog.Open(STmplogPath, false);
				OldCasLog.Replay([&](const CasDiskIndexEntry& Record) { Records.push_back(Record); });
			}
			{
				m_CasLog.Replay([&](const CasDiskIndexEntry& Record) { Records.push_back(Record); });
			}

			TCasLogFile<CasDiskIndexEntry> RecoveredCasLog;
			RecoveredCasLog.Open(SRecoveredlogPath, true);
			for (const auto& Record : Records)
			{
				RecoveredCasLog.Append(Record);
			}
			RecoveredCasLog.Close();

			fs::remove(SlogPath);
			fs::rename(SRecoveredlogPath, SlogPath);
			fs::remove(STmplogPath);
		}

		if (fs::is_regular_file(SidxPath))
		{
			fs::remove(SidxPath);
		}

		// Restore any previous snapshot
		if (fs::is_regular_file(STmpSidxPath))
		{
			fs::remove(SidxPath);
			fs::rename(STmpSidxPath, SidxPath);
		}
	}
	if (fs::is_regular_file(STmpSidxPath))
	{
		fs::remove(STmpSidxPath);
	}
	if (fs::is_regular_file(STmplogPath))
	{
		fs::remove(STmplogPath);
	}
}

namespace {
	struct LegacyCasDiskLocation
	{
		LegacyCasDiskLocation(uint64_t InOffset, uint64_t InSize)
		{
			ZEN_ASSERT(InOffset <= 0xff'ffff'ffff);
			ZEN_ASSERT(InSize <= 0xff'ffff'ffff);

			memcpy(&m_Offset[0], &InOffset, sizeof m_Offset);
			memcpy(&m_Size[0], &InSize, sizeof m_Size);
		}

		LegacyCasDiskLocation() = default;

		inline uint64_t GetOffset() const
		{
			uint64_t Offset = 0;
			memcpy(&Offset, &m_Offset, sizeof m_Offset);
			return Offset;
		}

		inline uint64_t GetSize() const
		{
			uint64_t Size = 0;
			memcpy(&Size, &m_Size, sizeof m_Size);
			return Size;
		}

	private:
		uint8_t m_Offset[5];
		uint8_t m_Size[5];
	};

	struct LegacyCasDiskIndexEntry
	{
		static const uint8_t kTombstone = 0x01;

		IoHash				  Key;
		LegacyCasDiskLocation Location;
		ZenContentType		  ContentType = ZenContentType::kUnknownContentType;
		uint8_t				  Flags		  = 0;
	};

}  // namespace

void
CasContainerStrategy::OpenContainer(bool IsNewStore)
{
	m_TotalSize = 0;

	m_LocationMap.clear();

	std::filesystem::path SidxPath		 = m_Config.RootDirectory / (m_ContainerBaseName + ".uidx");
	std::filesystem::path SlogPath		 = m_Config.RootDirectory / (m_ContainerBaseName + ".ulog");
	std::filesystem::path LegacySobsPath = m_Config.RootDirectory / (m_ContainerBaseName + ".ucas");

	bool CasLogEmpty = true;

	if (IsNewStore)
	{
		if (std::filesystem::is_regular_file(LegacySobsPath))
		{
			std::filesystem::remove(LegacySobsPath);
		}
		if (std::filesystem::is_regular_file(SlogPath))
		{
			std::filesystem::remove(SlogPath);
		}
		if (std::filesystem::is_regular_file(SidxPath))
		{
			std::filesystem::remove(SidxPath);
		}
		m_CasLog.Open(SlogPath, true);
	}
	else
	{
		if (std::filesystem::is_regular_file(LegacySobsPath))
		{
			ZEN_INFO("migrating store {} from {} to chunks", m_Config.RootDirectory / m_ContainerBaseName, LegacySobsPath);
			std::error_code Error;
			DiskSpace		Space = DiskSpaceInfo(m_Config.RootDirectory, Error);
			if (Error)
			{
				ZEN_ERROR("get disk space in {} FAILED, reason '{}'", m_ContainerBaseName, Error.message());
				return;
			}

			if (Space.Free < m_MaxBlockSize)  // Never let GC steal the last block space
			{
				ZEN_ERROR("legacy store migration from '{}' FAILED, required disk space {}, free {}",
						  m_Config.RootDirectory / m_ContainerBaseName,
						  m_MaxBlockSize,
						  NiceBytes(Space.Free));
				return;
			}

			BasicFile BlockFile;
			BlockFile.Open(LegacySobsPath, false);

			std::unordered_map<IoHash, LegacyCasDiskIndexEntry, IoHash::Hasher> LegacyDiskIndex;

			TCasLogFile<LegacyCasDiskIndexEntry> LegacyCasLog;
			LegacyCasLog.Open(SlogPath, false);
			LegacyCasLog.Replay([&](const LegacyCasDiskIndexEntry& Record) {
				if (Record.Flags & LegacyCasDiskIndexEntry::kTombstone)
				{
					LegacyDiskIndex.erase(Record.Key);
				}
				else
				{
					LegacyDiskIndex[Record.Key] = Record;
				}
			});

			uint64_t MaxUsedSize = 0;
			for (const auto& Entry : LegacyDiskIndex)
			{
				const LegacyCasDiskIndexEntry& Record(Entry.second);
				uint64_t					   EntryEnd = Record.Location.GetOffset() + Record.Location.GetSize();
				if (EntryEnd > MaxUsedSize)
				{
					MaxUsedSize = EntryEnd;
				}
			}

			BlockFile.SetFileSize(MaxUsedSize);

			uint64_t MaxRequiredChunkCount = MaxUsedSize / m_MaxBlockSize;
			if (MaxRequiredChunkCount > CasDiskLocation::MaxBlockIndex)
			{
				ZEN_ERROR("legacy store migration from '{}' FAILED, required block count {}, possible {}",
						  m_Config.RootDirectory / m_ContainerBaseName,
						  MaxRequiredChunkCount,
						  CasDiskLocation::MaxBlockIndex);
				return;
			}

			std::vector<IoHash> ChunkHashes;
			ChunkHashes.reserve(LegacyDiskIndex.size());
			for (const auto& Entry : LegacyDiskIndex)
			{
				ChunkHashes.push_back(Entry.first);
			}
			LegacyCasLog.Close();

			// Sort from biggest position to smallest
			std::sort(begin(ChunkHashes), end(ChunkHashes), [&](IoHash Lhs, IoHash Rhs) {
				auto LhsKeyIt = LegacyDiskIndex.find(Lhs);
				auto RhsKeyIt = LegacyDiskIndex.find(Rhs);
				return RhsKeyIt->second.Location.GetOffset() < LhsKeyIt->second.Location.GetOffset();
			});

			m_CasLog.Open(SlogPath, true);

			std::unique_ptr<ChunkBlock> NewBlockFile;
			uint64_t					WriteOffset	  = {};
			uint32_t					NewBlockIndex = {};

			std::vector<uint8_t> Chunk;
			for (const auto& ChunkHash : ChunkHashes)
			{
				const auto&					 Entry		   = LegacyDiskIndex[ChunkHash];
				const LegacyCasDiskLocation& ChunkLocation = Entry.Location;
				Chunk.resize(ChunkLocation.GetSize());
				BlockFile.Read(Chunk.data(), Chunk.size(), ChunkLocation.GetOffset());
				if (!NewBlockFile)
				{
					auto BlockPath = BuildUcasPath(m_BlocksBasePath, NewBlockIndex);
					NewBlockFile   = std::make_unique<ChunkBlock>(BlockPath);
					NewBlockFile->Create(m_MaxBlockSize);
				}
				else if (WriteOffset + Chunk.size() > m_MaxBlockSize)
				{
					uint64_t ChunkEnd = ChunkLocation.GetOffset() + Chunk.size();
					BlockFile.SetFileSize(ChunkEnd);
					NewBlockIndex  = NewBlockIndex + 1;
					auto BlockPath = BuildUcasPath(m_BlocksBasePath, NewBlockIndex);
					NewBlockFile   = std::make_unique<ChunkBlock>(BlockPath);
					NewBlockFile->Create(m_MaxBlockSize);
					WriteOffset = 0;
				}
				NewBlockFile->Write(Chunk.data(), Chunk.size(), WriteOffset);
				CasLocation NewChunkLocation(NewBlockIndex, WriteOffset, Chunk.size());
				m_CasLog.Append({.Key = ChunkHash, .Location = CasDiskLocation(NewChunkLocation, m_PayloadAlignment)});
				WriteOffset = RoundUp(WriteOffset + Chunk.size(), m_PayloadAlignment);
			}
			m_CasLog.Close();

			BlockFile.Close();
			std::filesystem::remove(LegacySobsPath);

			ZEN_INFO("migrated store {} to {} to chunks", m_Config.RootDirectory / m_ContainerBaseName, NewBlockIndex + 1);
			CasLogEmpty = false;
		}

		if (std::filesystem::is_regular_file(SidxPath))
		{
			BasicFile ObjectIndexFile;
			ObjectIndexFile.Open(SidxPath, false);
			uint64_t Size = ObjectIndexFile.FileSize();
			if (Size >= sizeof(CasDiskIndexHeader))
			{
				uint64_t		   ExpectedEntryCount = (Size - sizeof(sizeof(CasDiskIndexHeader))) / sizeof(CasDiskIndexEntry);
				CasDiskIndexHeader Header;
				ObjectIndexFile.Read(&Header, sizeof(Header), 0);
				if (Header.Magic == CasDiskIndexHeader::ExpectedMagic && Header.Version == CasDiskIndexHeader::CurrentVersion &&
					Header.PayloadAlignement > 0 && Header.EntryCount == ExpectedEntryCount)
				{
					std::vector<CasDiskIndexEntry> Entries{Header.EntryCount};
					ObjectIndexFile.Read(Entries.data(), Header.EntryCount * sizeof(CasDiskIndexEntry), sizeof(CasDiskIndexHeader));
					ObjectIndexFile.Close();
					for (const auto& Entry : Entries)
					{
						m_LocationMap[Entry.Key] = Entry.Location;
					}
					m_PayloadAlignment = Header.PayloadAlignement;
				}
			}
		}

		m_CasLog.Open(SlogPath, false);
		m_CasLog.Replay([&](const CasDiskIndexEntry& Record) {
			if (Record.Flags & CasDiskIndexEntry::kTombstone)
			{
				m_LocationMap.erase(Record.Key);
			}
			else
			{
				m_LocationMap[Record.Key] = Record.Location;
			}
			CasLogEmpty = false;
		});
	}

	std::unordered_set<uint32_t> BlockUsage;
	for (const auto& Entry : m_LocationMap)
	{
		const auto& Location = Entry.second;
		m_TotalSize.fetch_add(Location.GetSize());
		BlockUsage.insert(Location.GetBlockIndex());
	}

	if (std::filesystem::is_directory(m_BlocksBasePath))
	{
		std::vector<std::filesystem::path> FoldersToScan;
		FoldersToScan.push_back(m_BlocksBasePath);
		size_t FolderOffset = 0;
		while (FolderOffset < FoldersToScan.size())
		{
			for (const std::filesystem::directory_entry& Entry : std::filesystem::directory_iterator(FoldersToScan[FolderOffset]))
			{
				if (Entry.is_directory())
				{
					FoldersToScan.push_back(Entry.path());
					continue;
				}
				if (Entry.is_regular_file())
				{
					const std::filesystem::path Path = Entry.path();
					if (Path.extension() != ".ucas")
					{
						continue;
					}
					if (IsNewStore)
					{
						std::filesystem::remove(Path);
						continue;
					}
					std::string FileName = Path.stem().string();
					uint32_t	BlockIndex;
					bool		OK = ParseHex(FileName, BlockIndex);
					if (!OK)
					{
						continue;
					}
					if (!BlockUsage.contains(BlockIndex))
					{
						// Clear out unused blocks
						std::filesystem::remove(Path);
						continue;
					}
					auto BlockPath			  = BuildUcasPath(m_BlocksBasePath, BlockIndex);
					auto BlockFile			  = std::make_shared<ChunkBlock>(BlockPath);
					m_ChunkBlocks[BlockIndex] = BlockFile;
				}
			}
			++FolderOffset;
		}
	}

	// Create GC reserve file if possible
	std::filesystem::path GCReservePath = m_Config.RootDirectory / (m_ContainerBaseName + ".gc.reserve.ucas");

	std::error_code Error;
	DiskSpace		Space = DiskSpaceInfo(m_Config.RootDirectory, Error);
	if (Error)
	{
		ZEN_ERROR("get disk space in {} FAILED, reason '{}'", m_ContainerBaseName, Error.message());
		return;
	}

	BasicFile GCReserveFile;
	if (std::filesystem::is_regular_file(GCReservePath))
	{
		GCReserveFile.Open(GCReservePath, false);
		std::uint64_t CurrentSize = GCReserveFile.FileSize();
		if (CurrentSize != m_MaxBlockSize)
		{
			if (CurrentSize > m_MaxBlockSize)
			{
				GCReserveFile.SetFileSize(m_MaxBlockSize);
			}
			else
			{
				std::uint64_t ExtraSpace = m_MaxBlockSize - CurrentSize;
				if (Space.Free >= ExtraSpace)
				{
					GCReserveFile.SetFileSize(m_MaxBlockSize);
				}
				else
				{
					// We need it to be the proper size if we are to use it
					std::filesystem::remove(GCReservePath);
				}
			}
		}
	}
	else
	{
		if (Space.Free > m_MaxBlockSize)
		{
			GCReserveFile.Open(GCReservePath, true);
			GCReserveFile.SetFileSize(m_MaxBlockSize);
		}
	}

	if (!CasLogEmpty)
	{
		MakeIndexSnapshot();
	}

	// TODO: should validate integrity of container files here
}

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

#if ZEN_WITH_TESTS

namespace {
	static IoBuffer CreateChunk(uint64_t Size)
	{
		static std::random_device rd;
		static std::mt19937		  g(rd());

		std::vector<uint8_t> Values;
		Values.resize(Size);
		for (size_t Idx = 0; Idx < Size; ++Idx)
		{
			Values[Idx] = static_cast<uint8_t>(Idx);
		}
		std::shuffle(Values.begin(), Values.end(), g);

		return IoBufferBuilder::MakeCloneFromMemory(Values.data(), Values.size());
	}
}  // namespace

bool
operator==(const CasLocation& Lhs, const CasLocation& Rhs)
{
	return Lhs.BlockIndex == Rhs.BlockIndex && Lhs.Offset == Rhs.Offset && Lhs.Size == Rhs.Size;
}

TEST_CASE("compactcas.casdisklocation")
{
	CasLocation Zero = CasLocation{.BlockIndex = 0, .Offset = 0, .Size = 0};
	CHECK(Zero == CasDiskLocation(Zero, 4).Get(4));

	CasLocation MaxBlockIndex = CasLocation{.BlockIndex = CasDiskLocation::MaxBlockIndex, .Offset = 0, .Size = 0};
	CHECK(MaxBlockIndex == CasDiskLocation(MaxBlockIndex, 4).Get(4));

	CasLocation MaxOffset = CasLocation{.BlockIndex = 0, .Offset = CasDiskLocation::MaxOffset * 4, .Size = 0};
	CHECK(MaxOffset == CasDiskLocation(MaxOffset, 4).Get(4));

	CasLocation MaxSize = CasLocation{.BlockIndex = 0, .Offset = 0, .Size = std::numeric_limits<uint32_t>::max()};
	CHECK(MaxSize == CasDiskLocation(MaxSize, 4).Get(4));

	CasLocation MaxBlockIndexAndOffset =
		CasLocation{.BlockIndex = CasDiskLocation::MaxBlockIndex, .Offset = CasDiskLocation::MaxOffset * 4, .Size = 0};
	CHECK(MaxBlockIndexAndOffset == CasDiskLocation(MaxBlockIndexAndOffset, 4).Get(4));

	CasLocation MaxAll = CasLocation{.BlockIndex = CasDiskLocation::MaxBlockIndex,
									 .Offset	 = CasDiskLocation::MaxOffset * 4,
									 .Size		 = std::numeric_limits<uint32_t>::max()};
	CHECK(MaxAll == CasDiskLocation(MaxAll, 4).Get(4));

	CasLocation MaxAll4096 = CasLocation{.BlockIndex = CasDiskLocation::MaxBlockIndex,
										 .Offset	 = CasDiskLocation::MaxOffset * 4096,
										 .Size		 = std::numeric_limits<uint32_t>::max()};
	CHECK(MaxAll4096 == CasDiskLocation(MaxAll4096, 4096).Get(4096));

	CasLocation Middle = CasLocation{.BlockIndex = (CasDiskLocation::MaxBlockIndex) / 2,
									 .Offset	 = ((CasDiskLocation::MaxOffset) / 2) * 4,
									 .Size		 = std::numeric_limits<uint32_t>::max() / 2};
	CHECK(Middle == CasDiskLocation(Middle, 4).Get(4));
}

TEST_CASE("compactcas.hex")
{
	uint32_t	Value;
	std::string HexString;
	CHECK(!ParseHex("", Value));
	char Hex[9];

	FormatHex(0, Hex);
	HexString = std::string(Hex);
	CHECK(ParseHex(HexString, Value));
	CHECK(Value == 0);

	FormatHex(std::numeric_limits<std::uint32_t>::max(), Hex);
	HexString = std::string(Hex);
	CHECK(HexString == "ffffffff");
	CHECK(ParseHex(HexString, Value));
	CHECK(Value == std::numeric_limits<std::uint32_t>::max());

	FormatHex(0xadf14711, Hex);
	HexString = std::string(Hex);
	CHECK(HexString == "adf14711");
	CHECK(ParseHex(HexString, Value));
	CHECK(Value == 0xadf14711);

	FormatHex(0x80000000, Hex);
	HexString = std::string(Hex);
	CHECK(HexString == "80000000");
	CHECK(ParseHex(HexString, Value));
	CHECK(Value == 0x80000000);

	FormatHex(0x718293a4, Hex);
	HexString = std::string(Hex);
	CHECK(HexString == "718293a4");
	CHECK(ParseHex(HexString, Value));
	CHECK(Value == 0x718293a4);
}

TEST_CASE("compactcas.compact.gc")
{
	ScopedTemporaryDirectory TempDir;

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	const int kIterationCount = 1000;

	std::vector<IoHash> Keys(kIterationCount);

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 65536, 16, true);

		for (int i = 0; i < kIterationCount; ++i)
		{
			CbObjectWriter Cbo;
			Cbo << "id" << i;
			CbObject Obj = Cbo.Save();

			IoBuffer	 ObjBuffer = Obj.GetBuffer().AsIoBuffer();
			const IoHash Hash	   = HashBuffer(ObjBuffer);

			Cas.InsertChunk(ObjBuffer, Hash);

			Keys[i] = Hash;
		}

		for (int i = 0; i < kIterationCount; ++i)
		{
			IoBuffer Chunk = Cas.FindChunk(Keys[i]);

			CHECK(!!Chunk);

			CbObject Value = LoadCompactBinaryObject(Chunk);

			CHECK_EQ(Value["id"].AsInt32(), i);
		}
	}

	// Validate that we can still read the inserted data after closing
	// the original cas store

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 65536, 16, false);

		for (int i = 0; i < kIterationCount; ++i)
		{
			IoBuffer Chunk = Cas.FindChunk(Keys[i]);

			CHECK(!!Chunk);

			CbObject Value = LoadCompactBinaryObject(Chunk);

			CHECK_EQ(Value["id"].AsInt32(), i);
		}

		GcContext Ctx;
		Cas.CollectGarbage(Ctx);
	}
}

TEST_CASE("compactcas.compact.totalsize")
{
	std::random_device rd;
	std::mt19937	   g(rd());

	ScopedTemporaryDirectory TempDir;

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();

	CreateDirectories(CasConfig.RootDirectory);

	const uint64_t kChunkSize  = 1024;
	const int32_t  kChunkCount = 16;

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 65536, 16, true);

		for (int32_t Idx = 0; Idx < kChunkCount; ++Idx)
		{
			IoBuffer	 Chunk		  = CreateChunk(kChunkSize);
			const IoHash Hash		  = HashBuffer(Chunk);
			auto		 InsertResult = Cas.InsertChunk(Chunk, Hash);
			ZEN_ASSERT(InsertResult.New);
		}

		const uint64_t TotalSize = Cas.StorageSize().DiskSize;
		CHECK_EQ(kChunkSize * kChunkCount, TotalSize);
	}

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 65536, 16, false);

		const uint64_t TotalSize = Cas.StorageSize().DiskSize;
		CHECK_EQ(kChunkSize * kChunkCount, TotalSize);
	}
}

TEST_CASE("compactcas.gc.basic")
{
	ScopedTemporaryDirectory TempDir;

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	CasGc				 Gc;
	CasContainerStrategy Cas(CasConfig, Gc);
	Cas.Initialize("cb", 65536, 1 << 4, true);

	IoBuffer Chunk	   = CreateChunk(128);
	IoHash	 ChunkHash = IoHash::HashBuffer(Chunk);

	const auto InsertResult = Cas.InsertChunk(Chunk, ChunkHash);
	CHECK(InsertResult.New);
	Cas.Flush();

	GcContext GcCtx;
	GcCtx.CollectSmallObjects(true);

	Cas.CollectGarbage(GcCtx);

	CHECK(!Cas.HaveChunk(ChunkHash));
}

TEST_CASE("compactcas.gc.removefile")
{
	ScopedTemporaryDirectory TempDir;

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	IoBuffer Chunk	   = CreateChunk(128);
	IoHash	 ChunkHash = IoHash::HashBuffer(Chunk);
	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("cb", 65536, 1 << 4, true);

		const auto InsertResult = Cas.InsertChunk(Chunk, ChunkHash);
		CHECK(InsertResult.New);
		const auto InsertResultDup = Cas.InsertChunk(Chunk, ChunkHash);
		CHECK(!InsertResultDup.New);
		Cas.Flush();
	}

	CasGc				 Gc;
	CasContainerStrategy Cas(CasConfig, Gc);
	Cas.Initialize("cb", 65536, 1 << 4, false);

	GcContext GcCtx;
	GcCtx.CollectSmallObjects(true);

	Cas.CollectGarbage(GcCtx);

	CHECK(!Cas.HaveChunk(ChunkHash));
}

TEST_CASE("compactcas.gc.compact")
{
	ScopedTemporaryDirectory TempDir;

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	CasGc				 Gc;
	CasContainerStrategy Cas(CasConfig, Gc);
	Cas.Initialize("cb", 2048, 1 << 4, true);

	uint64_t			  ChunkSizes[9] = {128, 541, 1023, 781, 218, 37, 4, 997, 5};
	std::vector<IoBuffer> Chunks;
	Chunks.reserve(9);
	for (const auto& Size : ChunkSizes)
	{
		Chunks.push_back(CreateChunk(Size));
	}

	std::vector<IoHash> ChunkHashes;
	ChunkHashes.reserve(9);
	for (const auto& Chunk : Chunks)
	{
		ChunkHashes.push_back(IoHash::HashBuffer(Chunk.Data(), Chunk.Size()));
	}

	CHECK(Cas.InsertChunk(Chunks[0], ChunkHashes[0]).New);
	CHECK(Cas.InsertChunk(Chunks[1], ChunkHashes[1]).New);
	CHECK(Cas.InsertChunk(Chunks[2], ChunkHashes[2]).New);
	CHECK(Cas.InsertChunk(Chunks[3], ChunkHashes[3]).New);
	CHECK(Cas.InsertChunk(Chunks[4], ChunkHashes[4]).New);
	CHECK(Cas.InsertChunk(Chunks[5], ChunkHashes[5]).New);
	CHECK(Cas.InsertChunk(Chunks[6], ChunkHashes[6]).New);
	CHECK(Cas.InsertChunk(Chunks[7], ChunkHashes[7]).New);
	CHECK(Cas.InsertChunk(Chunks[8], ChunkHashes[8]).New);

	CHECK(Cas.HaveChunk(ChunkHashes[0]));
	CHECK(Cas.HaveChunk(ChunkHashes[1]));
	CHECK(Cas.HaveChunk(ChunkHashes[2]));
	CHECK(Cas.HaveChunk(ChunkHashes[3]));
	CHECK(Cas.HaveChunk(ChunkHashes[4]));
	CHECK(Cas.HaveChunk(ChunkHashes[5]));
	CHECK(Cas.HaveChunk(ChunkHashes[6]));
	CHECK(Cas.HaveChunk(ChunkHashes[7]));
	CHECK(Cas.HaveChunk(ChunkHashes[8]));

	auto InitialSize = Cas.StorageSize().DiskSize;

	// Keep first and last
	{
		GcContext GcCtx;
		GcCtx.CollectSmallObjects(true);

		std::vector<IoHash> KeepChunks;
		KeepChunks.push_back(ChunkHashes[0]);
		KeepChunks.push_back(ChunkHashes[8]);
		GcCtx.ContributeCas(KeepChunks);

		Cas.Flush();
		Cas.CollectGarbage(GcCtx);

		CHECK(Cas.HaveChunk(ChunkHashes[0]));
		CHECK(!Cas.HaveChunk(ChunkHashes[1]));
		CHECK(!Cas.HaveChunk(ChunkHashes[2]));
		CHECK(!Cas.HaveChunk(ChunkHashes[3]));
		CHECK(!Cas.HaveChunk(ChunkHashes[4]));
		CHECK(!Cas.HaveChunk(ChunkHashes[5]));
		CHECK(!Cas.HaveChunk(ChunkHashes[6]));
		CHECK(!Cas.HaveChunk(ChunkHashes[7]));
		CHECK(Cas.HaveChunk(ChunkHashes[8]));

		CHECK(ChunkHashes[0] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[0])));
		CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));
	}

	Cas.InsertChunk(Chunks[1], ChunkHashes[1]);
	Cas.InsertChunk(Chunks[2], ChunkHashes[2]);
	Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
	Cas.InsertChunk(Chunks[4], ChunkHashes[4]);
	Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
	Cas.InsertChunk(Chunks[6], ChunkHashes[6]);
	Cas.InsertChunk(Chunks[7], ChunkHashes[7]);

	// Keep last
	{
		GcContext GcCtx;
		GcCtx.CollectSmallObjects(true);
		std::vector<IoHash> KeepChunks;
		KeepChunks.push_back(ChunkHashes[8]);
		GcCtx.ContributeCas(KeepChunks);

		Cas.Flush();
		Cas.CollectGarbage(GcCtx);

		CHECK(!Cas.HaveChunk(ChunkHashes[0]));
		CHECK(!Cas.HaveChunk(ChunkHashes[1]));
		CHECK(!Cas.HaveChunk(ChunkHashes[2]));
		CHECK(!Cas.HaveChunk(ChunkHashes[3]));
		CHECK(!Cas.HaveChunk(ChunkHashes[4]));
		CHECK(!Cas.HaveChunk(ChunkHashes[5]));
		CHECK(!Cas.HaveChunk(ChunkHashes[6]));
		CHECK(!Cas.HaveChunk(ChunkHashes[7]));
		CHECK(Cas.HaveChunk(ChunkHashes[8]));

		CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));

		Cas.InsertChunk(Chunks[1], ChunkHashes[1]);
		Cas.InsertChunk(Chunks[2], ChunkHashes[2]);
		Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
		Cas.InsertChunk(Chunks[4], ChunkHashes[4]);
		Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
		Cas.InsertChunk(Chunks[6], ChunkHashes[6]);
		Cas.InsertChunk(Chunks[7], ChunkHashes[7]);
	}

	// Keep mixed
	{
		GcContext GcCtx;
		GcCtx.CollectSmallObjects(true);
		std::vector<IoHash> KeepChunks;
		KeepChunks.push_back(ChunkHashes[1]);
		KeepChunks.push_back(ChunkHashes[4]);
		KeepChunks.push_back(ChunkHashes[7]);
		GcCtx.ContributeCas(KeepChunks);

		Cas.Flush();
		Cas.CollectGarbage(GcCtx);

		CHECK(!Cas.HaveChunk(ChunkHashes[0]));
		CHECK(Cas.HaveChunk(ChunkHashes[1]));
		CHECK(!Cas.HaveChunk(ChunkHashes[2]));
		CHECK(!Cas.HaveChunk(ChunkHashes[3]));
		CHECK(Cas.HaveChunk(ChunkHashes[4]));
		CHECK(!Cas.HaveChunk(ChunkHashes[5]));
		CHECK(!Cas.HaveChunk(ChunkHashes[6]));
		CHECK(Cas.HaveChunk(ChunkHashes[7]));
		CHECK(!Cas.HaveChunk(ChunkHashes[8]));

		CHECK(ChunkHashes[1] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[1])));
		CHECK(ChunkHashes[4] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[4])));
		CHECK(ChunkHashes[7] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[7])));

		Cas.InsertChunk(Chunks[0], ChunkHashes[0]);
		Cas.InsertChunk(Chunks[2], ChunkHashes[2]);
		Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
		Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
		Cas.InsertChunk(Chunks[6], ChunkHashes[6]);
		Cas.InsertChunk(Chunks[8], ChunkHashes[8]);
	}

	// Keep multiple at end
	{
		GcContext GcCtx;
		GcCtx.CollectSmallObjects(true);
		std::vector<IoHash> KeepChunks;
		KeepChunks.push_back(ChunkHashes[6]);
		KeepChunks.push_back(ChunkHashes[7]);
		KeepChunks.push_back(ChunkHashes[8]);
		GcCtx.ContributeCas(KeepChunks);

		Cas.Flush();
		Cas.CollectGarbage(GcCtx);

		CHECK(!Cas.HaveChunk(ChunkHashes[0]));
		CHECK(!Cas.HaveChunk(ChunkHashes[1]));
		CHECK(!Cas.HaveChunk(ChunkHashes[2]));
		CHECK(!Cas.HaveChunk(ChunkHashes[3]));
		CHECK(!Cas.HaveChunk(ChunkHashes[4]));
		CHECK(!Cas.HaveChunk(ChunkHashes[5]));
		CHECK(Cas.HaveChunk(ChunkHashes[6]));
		CHECK(Cas.HaveChunk(ChunkHashes[7]));
		CHECK(Cas.HaveChunk(ChunkHashes[8]));

		CHECK(ChunkHashes[6] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[6])));
		CHECK(ChunkHashes[7] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[7])));
		CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));

		Cas.InsertChunk(Chunks[0], ChunkHashes[0]);
		Cas.InsertChunk(Chunks[1], ChunkHashes[1]);
		Cas.InsertChunk(Chunks[2], ChunkHashes[2]);
		Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
		Cas.InsertChunk(Chunks[4], ChunkHashes[4]);
		Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
	}

	// Keep every other
	{
		GcContext GcCtx;
		GcCtx.CollectSmallObjects(true);
		std::vector<IoHash> KeepChunks;
		KeepChunks.push_back(ChunkHashes[0]);
		KeepChunks.push_back(ChunkHashes[2]);
		KeepChunks.push_back(ChunkHashes[4]);
		KeepChunks.push_back(ChunkHashes[6]);
		KeepChunks.push_back(ChunkHashes[8]);
		GcCtx.ContributeCas(KeepChunks);

		Cas.Flush();
		Cas.CollectGarbage(GcCtx);

		CHECK(Cas.HaveChunk(ChunkHashes[0]));
		CHECK(!Cas.HaveChunk(ChunkHashes[1]));
		CHECK(Cas.HaveChunk(ChunkHashes[2]));
		CHECK(!Cas.HaveChunk(ChunkHashes[3]));
		CHECK(Cas.HaveChunk(ChunkHashes[4]));
		CHECK(!Cas.HaveChunk(ChunkHashes[5]));
		CHECK(Cas.HaveChunk(ChunkHashes[6]));
		CHECK(!Cas.HaveChunk(ChunkHashes[7]));
		CHECK(Cas.HaveChunk(ChunkHashes[8]));

		CHECK(ChunkHashes[0] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[0])));
		CHECK(ChunkHashes[2] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[2])));
		CHECK(ChunkHashes[4] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[4])));
		CHECK(ChunkHashes[6] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[6])));
		CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));

		Cas.InsertChunk(Chunks[1], ChunkHashes[1]);
		Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
		Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
		Cas.InsertChunk(Chunks[7], ChunkHashes[7]);
	}

	// Verify that we nicely appended blocks even after all GC operations
	CHECK(ChunkHashes[0] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[0])));
	CHECK(ChunkHashes[1] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[1])));
	CHECK(ChunkHashes[2] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[2])));
	CHECK(ChunkHashes[3] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[3])));
	CHECK(ChunkHashes[4] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[4])));
	CHECK(ChunkHashes[5] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[5])));
	CHECK(ChunkHashes[6] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[6])));
	CHECK(ChunkHashes[7] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[7])));
	CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));

	auto FinalSize = Cas.StorageSize().DiskSize;
	CHECK(InitialSize == FinalSize);
}

TEST_CASE("compactcas.gc.deleteblockonopen")
{
	ScopedTemporaryDirectory TempDir;

	uint64_t			  ChunkSizes[20] = {128, 541, 311, 181, 218, 37, 4, 397, 5, 92, 551, 721, 31, 92, 16, 99, 131, 41, 541, 84};
	std::vector<IoBuffer> Chunks;
	Chunks.reserve(20);
	for (const auto& Size : ChunkSizes)
	{
		Chunks.push_back(CreateChunk(Size));
	}

	std::vector<IoHash> ChunkHashes;
	ChunkHashes.reserve(20);
	for (const auto& Chunk : Chunks)
	{
		ChunkHashes.push_back(IoHash::HashBuffer(Chunk.Data(), Chunk.Size()));
	}

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);
	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 1024, 16, true);

		for (size_t i = 0; i < 20; i++)
		{
			CHECK(Cas.InsertChunk(Chunks[i], ChunkHashes[i]).New);
		}

		// GC every other block
		{
			GcContext GcCtx;
			GcCtx.CollectSmallObjects(true);
			std::vector<IoHash> KeepChunks;
			for (size_t i = 0; i < 20; i += 2)
			{
				KeepChunks.push_back(ChunkHashes[i]);
			}
			GcCtx.ContributeCas(KeepChunks);

			Cas.Flush();
			Cas.CollectGarbage(GcCtx);

			for (size_t i = 0; i < 20; i += 2)
			{
				CHECK(Cas.HaveChunk(ChunkHashes[i]));
				CHECK(!Cas.HaveChunk(ChunkHashes[i + 1]));
				CHECK(ChunkHashes[i] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[i])));
			}
		}
	}
	{
		// Re-open
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 1024, 16, false);

		for (size_t i = 0; i < 20; i += 2)
		{
			CHECK(Cas.HaveChunk(ChunkHashes[i]));
			CHECK(!Cas.HaveChunk(ChunkHashes[i + 1]));
			CHECK(ChunkHashes[i] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[i])));
		}
	}
}

TEST_CASE("compactcas.gc.handleopeniobuffer")
{
	ScopedTemporaryDirectory TempDir;

	uint64_t			  ChunkSizes[20] = {128, 541, 311, 181, 218, 37, 4, 397, 5, 92, 551, 721, 31, 92, 16, 99, 131, 41, 541, 84};
	std::vector<IoBuffer> Chunks;
	Chunks.reserve(20);
	for (const auto& Size : ChunkSizes)
	{
		Chunks.push_back(CreateChunk(Size));
	}

	std::vector<IoHash> ChunkHashes;
	ChunkHashes.reserve(20);
	for (const auto& Chunk : Chunks)
	{
		ChunkHashes.push_back(IoHash::HashBuffer(Chunk.Data(), Chunk.Size()));
	}

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	CasGc				 Gc;
	CasContainerStrategy Cas(CasConfig, Gc);
	Cas.Initialize("test", 1024, 16, true);

	for (size_t i = 0; i < 20; i++)
	{
		CHECK(Cas.InsertChunk(Chunks[i], ChunkHashes[i]).New);
	}

	auto RetainChunk = Cas.FindChunk(ChunkHashes[5]);
	Cas.Flush();

	// GC everything
	GcContext GcCtx;
	GcCtx.CollectSmallObjects(true);
	Cas.CollectGarbage(GcCtx);

	for (size_t i = 0; i < 20; i++)
	{
		CHECK(!Cas.HaveChunk(ChunkHashes[i]));
	}

	CHECK(ChunkHashes[5] == IoHash::HashBuffer(RetainChunk));
}

TEST_CASE("compactcas.legacyconversion")
{
	ScopedTemporaryDirectory TempDir;

	uint64_t ChunkSizes[]	 = {2041, 1123, 1223, 1239, 341, 1412, 912, 774, 341, 431, 554, 1098, 2048, 339, 561, 16, 16, 2048, 2048};
	size_t	 ChunkCount		 = sizeof(ChunkSizes) / sizeof(uint64_t);
	size_t	 SingleBlockSize = 0;
	std::vector<IoBuffer> Chunks;
	Chunks.reserve(ChunkCount);
	for (const auto& Size : ChunkSizes)
	{
		Chunks.push_back(CreateChunk(Size));
		SingleBlockSize += Size;
	}

	std::vector<IoHash> ChunkHashes;
	ChunkHashes.reserve(ChunkCount);
	for (const auto& Chunk : Chunks)
	{
		ChunkHashes.push_back(IoHash::HashBuffer(Chunk.Data(), Chunk.Size()));
	}

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", gsl::narrow<uint32_t>(SingleBlockSize * 2), 16, true);

		for (size_t i = 0; i < ChunkCount; i++)
		{
			CHECK(Cas.InsertChunk(Chunks[i], ChunkHashes[i]).New);
		}

		std::vector<IoHash> KeepChunks;
		for (size_t i = 0; i < ChunkCount; i += 2)
		{
			KeepChunks.push_back(ChunkHashes[i]);
		}
		GcContext GcCtx;
		GcCtx.CollectSmallObjects(true);
		GcCtx.ContributeCas(KeepChunks);
		Cas.Flush();
		Gc.CollectGarbage(GcCtx);
	}

	auto CasPath	   = BuildUcasPath(CasConfig.RootDirectory / "test" / "blocks", 1);
	auto LegacyCasPath = CasConfig.RootDirectory / "test.ucas";
	std::filesystem::rename(CasPath, LegacyCasPath);

	std::vector<CasDiskIndexEntry> LogEntries;
	std::filesystem::path		   SidxPath = CasConfig.RootDirectory / ("test.uidx");
	if (std::filesystem::is_regular_file(SidxPath))
	{
		BasicFile ObjectIndexFile;
		ObjectIndexFile.Open(SidxPath, false);
		uint64_t Size = ObjectIndexFile.FileSize();
		if (Size >= sizeof(CasDiskIndexHeader))
		{
			uint64_t		   ExpectedEntryCount = (Size - sizeof(sizeof(CasDiskIndexHeader))) / sizeof(CasDiskIndexEntry);
			CasDiskIndexHeader Header;
			ObjectIndexFile.Read(&Header, sizeof(Header), 0);
			if (Header.Magic == CasDiskIndexHeader::ExpectedMagic && Header.Version == CasDiskIndexHeader::CurrentVersion &&
				Header.PayloadAlignement > 0 && Header.EntryCount == ExpectedEntryCount)
			{
				LogEntries.resize(Header.EntryCount);
				ObjectIndexFile.Read(LogEntries.data(), Header.EntryCount * sizeof(CasDiskIndexEntry), sizeof(CasDiskIndexHeader));
			}
		}
		ObjectIndexFile.Close();
		std::filesystem::remove(SidxPath);
	}

	std::filesystem::path SlogPath = CasConfig.RootDirectory / "test.ulog";
	{
		TCasLogFile<CasDiskIndexEntry> CasLog;
		CasLog.Open(SlogPath, false);
		CasLog.Replay([&](const CasDiskIndexEntry& Record) { LogEntries.push_back(Record); });
	}
	TCasLogFile<LegacyCasDiskIndexEntry> LegacyCasLog;
	LegacyCasLog.Open(SlogPath, true);
	for (const auto& Entry : LogEntries)
	{
		CasLocation				Location = Entry.Location.Get(16);
		LegacyCasDiskLocation	LegacyLocation(Location.Offset, Location.Size);
		LegacyCasDiskIndexEntry LegacyEntry = {.Key			= Entry.Key,
											   .Location	= LegacyLocation,
											   .ContentType = Entry.ContentType,
											   .Flags		= Entry.Flags};
		LegacyCasLog.Append(LegacyEntry);
	}
	LegacyCasLog.Close();

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 2048, 16, false);

		for (size_t i = 0; i < ChunkCount; i += 2)
		{
			CHECK(Cas.HaveChunk(ChunkHashes[i]));
			CHECK(!Cas.HaveChunk(ChunkHashes[i + 1]));
			CHECK(ChunkHashes[i] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[i])));
		}
	}
}

TEST_CASE("compactcas.threadedinsert")	// * doctest::skip(true))
{
	//	for (uint32_t i = 0; i < 100; ++i)
	{
		ScopedTemporaryDirectory TempDir;

		CasStoreConfiguration CasConfig;
		CasConfig.RootDirectory = TempDir.Path();

		CreateDirectories(CasConfig.RootDirectory);

		const uint64_t kChunkSize  = 1048;
		const int32_t  kChunkCount = 8192;

		std::vector<IoHash> ChunkHashes;
		ChunkHashes.reserve(kChunkCount);
		std::vector<IoBuffer> Chunks;
		Chunks.reserve(kChunkCount);

		for (int32_t Idx = 0; Idx < kChunkCount; ++Idx)
		{
			IoBuffer Chunk = CreateChunk(kChunkSize);
			IoHash	 Hash  = HashBuffer(Chunk);
			ChunkHashes.emplace_back(Hash);
			Chunks.emplace_back(Chunk);
		}

		WorkerThreadPool	 ThreadPool(4);
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 32768, 16, true);
		{
			for (int32_t Idx = 0; Idx < kChunkCount; ++Idx)
			{
				const IoBuffer& Chunk = Chunks[Idx];
				const IoHash&	Hash  = ChunkHashes[Idx];
				ThreadPool.ScheduleWork([&Cas, Chunk, Hash]() {
					auto InsertResult = Cas.InsertChunk(Chunk, Hash);
					ZEN_ASSERT(InsertResult.New);
				});
			}
			ThreadPool.Flush();
		}

		const uint64_t TotalSize = Cas.StorageSize().DiskSize;
		CHECK_EQ(kChunkSize * kChunkCount, TotalSize);

		{
			std::vector<IoHash> OldChunkHashes(ChunkHashes.begin(), ChunkHashes.end());
			for (int32_t Idx = 0; Idx < kChunkCount; ++Idx)
			{
				ThreadPool.ScheduleWork([&Cas, &OldChunkHashes, Idx]() {
					auto ChunkHash = OldChunkHashes[Idx];
					auto Chunk	   = Cas.FindChunk(ChunkHash);
					auto Hash	   = IoHash::HashBuffer(Chunk);
					CHECK(ChunkHash == Hash);
				});
			}
			ThreadPool.Flush();
		}

		std::unordered_set<IoHash, IoHash::Hasher> GcChunkHashes(ChunkHashes.begin(), ChunkHashes.end());
		{
			std::vector<IoHash> OldChunkHashes(ChunkHashes.begin(), ChunkHashes.end());
			std::vector<IoHash> NewChunkHashes;
			NewChunkHashes.reserve(kChunkCount);
			std::vector<IoBuffer> NewChunks;
			NewChunks.reserve(kChunkCount);

			for (int32_t Idx = 0; Idx < kChunkCount; ++Idx)
			{
				IoBuffer Chunk = CreateChunk(kChunkSize);
				IoHash	 Hash  = HashBuffer(Chunk);
				NewChunkHashes.emplace_back(Hash);
				NewChunks.emplace_back(Chunk);
			}

			RwLock				 ChunkHashesLock;
			std::atomic_uint32_t AddedChunkCount;

			for (int32_t Idx = 0; Idx < kChunkCount; ++Idx)
			{
				const IoBuffer& Chunk = NewChunks[Idx];
				const IoHash&	Hash  = NewChunkHashes[Idx];
				ThreadPool.ScheduleWork([&Cas, Chunk, Hash, &AddedChunkCount]() {
					auto InsertResult = Cas.InsertChunk(Chunk, Hash);
					ZEN_ASSERT(InsertResult.New);
					AddedChunkCount.fetch_add(1);
				});
				ThreadPool.ScheduleWork([&Cas, &ChunkHashesLock, &OldChunkHashes, Idx]() {
					IoHash ChunkHash = OldChunkHashes[Idx];
					auto   Chunk	 = Cas.FindChunk(OldChunkHashes[Idx]);
					if (Chunk)
					{
						CHECK(ChunkHash == IoHash::HashBuffer(Chunk));
					}
				});
			}

			while (AddedChunkCount.load() < kChunkCount)
			{
				std::vector<IoHash> AddedHashes;
				{
					RwLock::ExclusiveLockScope _(ChunkHashesLock);
					AddedHashes.swap(NewChunkHashes);
				}
				// Need to be careful since we might GC blocks we don't know outside of RwLock::ExclusiveLockScope
				for (auto ChunkHash : AddedHashes)
				{
					if (Cas.HaveChunk(ChunkHash))
					{
						GcChunkHashes.emplace(ChunkHash);
					}
				}
				std::vector<IoHash> KeepHashes(GcChunkHashes.begin(), GcChunkHashes.end());
				int32_t				C = 0;
				while (C < KeepHashes.size())
				{
					if (C % 155 == 0)
					{
						if (C < KeepHashes.size() - 1)
						{
							KeepHashes[C] = KeepHashes[KeepHashes.size() - 1];
							KeepHashes.pop_back();
						}
						if (C + 3 < KeepHashes.size() - 1)
						{
							KeepHashes[C + 3] = KeepHashes[KeepHashes.size() - 1];
							KeepHashes.pop_back();
						}
					}
					C++;
				}

				GcContext GcCtx;
				GcCtx.CollectSmallObjects(true);
				GcCtx.ContributeCas(KeepHashes);
				Cas.CollectGarbage(GcCtx);
				CasChunkSet& Deleted = GcCtx.DeletedCas();
				Deleted.IterateChunks([&GcChunkHashes](const IoHash& ChunkHash) { GcChunkHashes.erase(ChunkHash); });
			}

			ThreadPool.Flush();

			{
				std::vector<IoHash> AddedHashes;
				{
					RwLock::ExclusiveLockScope _(ChunkHashesLock);
					AddedHashes.swap(NewChunkHashes);
				}
				// Need to be careful since we might GC blocks we don't know outside of RwLock::ExclusiveLockScope
				for (auto ChunkHash : AddedHashes)
				{
					if (Cas.HaveChunk(ChunkHash))
					{
						GcChunkHashes.emplace(ChunkHash);
					}
				}
				std::vector<IoHash> KeepHashes(GcChunkHashes.begin(), GcChunkHashes.end());
				int32_t				C = 0;
				while (C < KeepHashes.size())
				{
					if (C % 77 == 0 && C < KeepHashes.size() - 1)
					{
						KeepHashes[C] = KeepHashes[KeepHashes.size() - 1];
						KeepHashes.pop_back();
					}
					C++;
				}

				GcContext GcCtx;
				GcCtx.CollectSmallObjects(true);
				GcCtx.ContributeCas(KeepHashes);
				Cas.CollectGarbage(GcCtx);
				CasChunkSet& Deleted = GcCtx.DeletedCas();
				Deleted.IterateChunks([&GcChunkHashes](const IoHash& ChunkHash) { GcChunkHashes.erase(ChunkHash); });
			}
		}
		{
			for (const auto& ChunkHash : GcChunkHashes)
			{
				ThreadPool.ScheduleWork([&Cas, ChunkHash]() {
					CHECK(Cas.HaveChunk(ChunkHash));
					CHECK(ChunkHash == IoHash::HashBuffer(Cas.FindChunk(ChunkHash)));
				});
			}
			ThreadPool.Flush();
		}
	}
}

#endif

void
compactcas_forcelink()
{
}

}  // namespace zen