#include "w3cs_webp_cache.h"

#include <fcntl.h>
#include <sys/file.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <unistd.h>
#include <webp/decode.h>

#include <array>
#include <chrono>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <filesystem>
#include <string>
#include <thread>
#include <utility>
#include <vector>

namespace {

using w3cs::WebpBytes;
using w3cs::WebpCache;
using w3cs::WebpPolicy;
using w3cs::WebpPreparedSource;
using w3cs::WebpSource;

[[noreturn]] void fail(const char *message) {
  std::fprintf(stderr, "webp cache test failed: %s\n", message);
  std::exit(1);
}

void require(bool condition, const char *message) {
  if (!condition) fail(message);
}

bool write_exact(int fd, const void *input, size_t size) {
  const auto *cursor = static_cast<const uint8_t *>(input);
  while (size) {
    const ssize_t count = write(fd, cursor, size);
    if (count > 0) {
      cursor += count;
      size -= static_cast<size_t>(count);
      continue;
    }
    if (count < 0 && errno == EINTR) continue;
    return false;
  }
  return true;
}

bool read_exact(int fd, void *output, size_t size) {
  auto *cursor = static_cast<uint8_t *>(output);
  while (size) {
    const ssize_t count = read(fd, cursor, size);
    if (count > 0) {
      cursor += count;
      size -= static_cast<size_t>(count);
      continue;
    }
    if (count < 0 && errno == EINTR) continue;
    return false;
  }
  return true;
}

std::string temporary_directory(const char *suffix) {
  std::string pattern = std::string("/tmp/w3cs-webp-") + suffix + "-XXXXXX";
  std::vector<char> path(pattern.begin(), pattern.end());
  path.push_back(0);
  if (!mkdtemp(path.data())) fail("mkdtemp");
  return path.data();
}

WebpSource rgba_source(uint32_t width, uint32_t height, uint8_t seed = 0) {
  WebpSource source{width, height, width * 4, 21,
                    WebpBytes(static_cast<size_t>(width) * height * 4)};
  for (size_t index = 0; index < source.pixels.size(); index += 4) {
    source.pixels[index] = static_cast<uint8_t>(seed + index * 3);
    source.pixels[index + 1] = static_cast<uint8_t>(seed + index * 5);
    source.pixels[index + 2] = static_cast<uint8_t>(seed + index * 7);
    source.pixels[index + 3] = static_cast<uint8_t>((index / 4) & 1 ? 17 : 238);
  }
  return source;
}

void test_source_decode() {
  struct Vector {
    uint32_t format;
    uint16_t word;
    std::array<uint8_t, 4> rgba;
  };
  const std::array vectors{
    Vector{23, 0xf800, {255, 0, 0, 255}},
    Vector{23, 0x07e0, {0, 255, 0, 255}},
    Vector{24, 0x001f, {0, 0, 255, 255}},
    Vector{25, 0x8000, {0, 0, 0, 255}},
    Vector{25, 0x0000, {0, 0, 0, 0}},
    Vector{26, 0xf123, {17, 34, 51, 255}},
  };
  for (const auto &vector : vectors) {
    WebpSource source{1, 1, 2, vector.format,
                      {static_cast<uint8_t>(vector.word),
                       static_cast<uint8_t>(vector.word >> 8)}};
    auto rgba = WebpCache::decode_source_rgba(source);
    require(rgba && rgba->size() == 4, "decode 16-bit source");
    require(std::equal(rgba->begin(), rgba->end(), vector.rgba.begin()),
            "decode 16-bit value");
  }
  WebpSource bgra{1, 1, 4, 21, {3, 2, 1, 4}};
  auto rgba = WebpCache::decode_source_rgba(bgra);
  require(rgba && *rgba == WebpBytes({1, 2, 3, 4}), "decode BGRA");
  bgra.format = 22;
  rgba = WebpCache::decode_source_rgba(bgra);
  require(rgba && *rgba == WebpBytes({1, 2, 3, 255}), "decode BGRX");
  bgra.pitch = 3;
  require(!WebpCache::decode_source_rgba(bgra), "reject bad pitch");
}

void test_encoder_alpha() {
  WebpBytes rgba;
  for (unsigned index = 0; index < 64; ++index) {
    rgba.push_back(static_cast<uint8_t>(index * 3));
    rgba.push_back(static_cast<uint8_t>(index * 5));
    rgba.push_back(static_cast<uint8_t>(index * 7));
    rgba.push_back(static_cast<uint8_t>(index * 17));
  }
  for (uint8_t method : {uint8_t{0}, uint8_t{4}}) {
    auto encoded = WebpCache::encode_rgba(rgba, 8, 8, {40, method});
    require(encoded && !encoded->empty(), "encode supported method");
    int width = 0;
    int height = 0;
    uint8_t *decoded = WebPDecodeRGBA(encoded->data(), encoded->size(),
                                      &width, &height);
    require(decoded && width == 8 && height == 8, "decode encoded WebP");
    for (size_t pixel = 0; pixel < 64; ++pixel)
      require(decoded[pixel * 4 + 3] == rgba[pixel * 4 + 3],
              "WebP alpha is exact");
    WebPFree(decoded);
  }
}

void wait_encoded(WebpCache &cache, uint64_t count) {
  for (unsigned attempt = 0; attempt < 400; ++attempt) {
    if (cache.stats().encoded >= count) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail("background encode timeout");
}

void wait_encode_errors(WebpCache &cache, uint64_t count) {
  for (unsigned attempt = 0; attempt < 400; ++attempt) {
    if (cache.stats().encode_errors >= count) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail("background encode failure timeout");
}

void wait_disk_loads(WebpCache &cache, uint64_t count) {
  for (unsigned attempt = 0; attempt < 400; ++attempt) {
    if (cache.stats().disk_loads >= count) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail("background disk load timeout");
}

void wait_fallback_disk_loads(WebpCache &cache, uint64_t count) {
  for (unsigned attempt = 0; attempt < 400; ++attempt) {
    if (cache.stats().fallback_disk_loads >= count) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail("background fallback disk load timeout");
}

void wait_idle(WebpCache &cache) {
  for (unsigned attempt = 0; attempt < 400; ++attempt) {
    if (cache.stats().pending_jobs == 0) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail("background cache idle timeout");
}

void wait_disk_scans(WebpCache &cache, uint64_t count) {
  for (unsigned attempt = 0; attempt < 400; ++attempt) {
    if (cache.stats().disk_scans >= count) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail("background disk scan timeout");
}

void wait_stale_temp_removals(WebpCache &cache, uint64_t count) {
  for (unsigned attempt = 0; attempt < 400; ++attempt) {
    if (cache.stats().stale_temp_removals >= count) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail("background stale temporary cleanup timeout");
}

void wait_for_path(const std::string &path, bool exists) {
  for (unsigned attempt = 0; attempt < 400; ++attempt) {
    if (std::filesystem::exists(path) == exists) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail(exists ? "path creation timeout" : "path removal timeout");
}

void test_method_keys_and_legacy_disk_fallback() {
  const std::string directory = temporary_directory("method-migration");
  WebpPreparedSource fixed;
  fixed.encoder_version = 0x01020304u;
  fixed.width = 64;
  fixed.height = 32;
  fixed.pitch = 128;
  fixed.format = 23;
  fixed.decoded_size = 4096;
  for (size_t index = 0; index < fixed.source_digest.size(); ++index)
    fixed.source_digest[index] = static_cast<uint8_t>(index);
  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    const auto method0 = cache.prepare(fixed, {95, 0});
    const auto method4 = cache.prepare(fixed, {95, 4});
    require(method0.key ==
                "46325047634bedb89217a2196e9e9948415153954d1cab1c2c41ed8e8bef79d1",
            "deployed method-0 cache key domain stays stable");
    require(method4.key ==
                "605f78552f63c2b2a8727d711d16af466c853528075ea05d9f25ddd5a131999e",
            "method-4 cache key is stable and separate");
  }

  const WebpSource source = rgba_source(128, 128, 31);
  w3cs::WebpRequest legacy;
  w3cs::WebpRequest primary;
  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    const auto prepared = cache.prepare_source(source);
    legacy = cache.prepare(prepared, {80, 0});
    primary = cache.prepare(prepared, {80, 4});
    require(legacy.key != primary.key, "method separates live cache keys");
    require(cache.submit(legacy, source), "seed deployed method-0 entry");
    wait_encoded(cache, 1);
  }

  const std::string slot_path = directory + "/.slot-0.lock";
  const int slot = open(slot_path.c_str(), O_RDWR | O_CREAT, 0644);
  require(slot >= 0 && flock(slot, LOCK_EX) == 0,
          "hold encoder while legacy disk entry loads");
  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    require(!cache.lookup(primary) && !cache.lookup(legacy),
            "fresh process starts with both methods cold in memory");
    require(cache.submit(primary, source, legacy),
            "queue method 4 with read-only method-0 fallback");
    wait_fallback_disk_loads(cache, 1);
    require(cache.lookup(legacy).has_value(),
            "background worker publishes deployed method-0 disk entry");
    require(!cache.lookup(primary),
            "method 4 remains cold while encoder slot is occupied");
    const auto blocked = cache.stats();
    require(blocked.encoded == 0 && blocked.fallback_disk_loads == 1 &&
                blocked.disk_loads == 1,
            "legacy load does not run either encoder policy");

    flock(slot, LOCK_UN);
    wait_encoded(cache, 1);
    require(cache.lookup(primary).has_value(),
            "method 4 becomes primary after background encoding");
    const auto warmed = cache.stats();
    require(warmed.encoded == 1 && warmed.fallback_disk_loads == 1,
            "migration encodes method 4 exactly once");
  }
  close(slot);
  require(std::filesystem::exists(directory + "/" + legacy.key + ".webp") &&
              std::filesystem::exists(directory + "/" + primary.key +
                                      ".webp"),
          "migration retains legacy file and adds primary file");
  std::filesystem::remove_all(directory);
}

void test_legacy_deny_propagates_without_method0_encode() {
  const std::string directory = temporary_directory("method-deny");
  const WebpSource source = rgba_source(64, 64, 41);
  w3cs::WebpRequest legacy;
  w3cs::WebpRequest primary;
  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    const auto prepared = cache.prepare_source(source);
    legacy = cache.prepare(prepared, {95, 0});
    primary = cache.prepare(prepared, {95, 4});
    cache.deny(legacy);
    wait_for_path(directory + "/" + legacy.key + ".deny", true);
  }
  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    require(cache.submit(primary, source, legacy),
            "queue method 4 beside deployed method-0 deny");
    wait_for_path(directory + "/" + primary.key + ".deny", true);
    wait_idle(cache);
    const auto stats = cache.stats();
    require(stats.encoded == 0 && stats.fallback_disk_loads == 0,
            "legacy deny cannot encode or load a method-4 texture");
    require(!std::filesystem::exists(
                directory + "/" + primary.key + ".webp"),
            "legacy deny cannot resurrect a method-4 WebP file");
    require(!cache.submit(primary, source, legacy),
            "propagated deny rejects repeated local submissions");
  }
  std::filesystem::remove_all(directory);
}

void test_cache_hit_corruption_and_deny() {
  const std::string directory = temporary_directory("hit");
  const WebpSource source = rgba_source(64, 64, 9);
  w3cs::WebpRequest request;
  std::string path;
  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    require(cache.enabled(), "cache enabled");
    request = cache.prepare(source, {80, 0});
    const auto other_quality = cache.prepare(source, {95, 0});
    require(request.key != other_quality.key, "quality separates cache key");
    require(!cache.lookup(request), "cold miss");
    require(cache.submit(request, source), "queue miss");
    wait_encoded(cache, 1);
    const auto hit = cache.lookup(request);
    require(hit && !hit->empty(), "warm hit");
    const auto stats = cache.stats();
    require(stats.memory_entries == 1 && stats.memory_bytes == hit->size(),
            "encoded resource is published to bounded memory");
    require(stats.memory_bytes <= stats.memory_limit_bytes,
            "memory front stays within its byte limit");
    path = directory + "/" + request.key + ".webp";
  }

  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    require(!cache.lookup(request),
            "cold process does not perform disk I/O in lookup");
    require(cache.stats().disk_loads == 0,
            "cold lookup does not synchronously load disk cache");
    require(cache.submit(request, source), "queue existing disk resource");
    wait_disk_loads(cache, 1);
    require(cache.stats().encoded == 0,
            "valid disk entry does not run the encoder");
    require(cache.lookup(request).has_value(),
            "background disk load warms a later memory lookup");
  }

  {
    const int fd = open(path.c_str(), O_WRONLY | O_TRUNC);
    require(fd >= 0, "open cache entry for corruption test");
    require(write(fd, "bad", 3) == 3, "corrupt cache entry");
    close(fd);
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    require(!cache.lookup(request), "new process starts with memory miss");
    require(cache.submit(request, source), "queue corrupt disk resource");
    wait_encoded(cache, 1);
    require(cache.stats().corrupt_entries == 1, "corruption counted");
    require(cache.lookup(request).has_value(),
            "corrupt disk entry is replaced in the background");
    cache.deny(request);
    require(!cache.lookup(request), "deny selects raw");
    wait_for_path(directory + "/" + request.key + ".deny", true);
    wait_for_path(path, false);
    require(cache.stats().deny_queued == 1,
            "disk deny is queued in the background");
  }
  std::filesystem::remove_all(directory);
}

void test_queue_bound_and_duplicate_coalescing() {
  const std::string directory = temporary_directory("bound");
  const std::string slot_path = directory + "/.slot-0.lock";
  const int slot = open(slot_path.c_str(), O_RDWR | O_CREAT, 0644);
  require(slot >= 0 && flock(slot, LOCK_EX) == 0, "hold host encoder slot");
  {
    WebpCache cache(directory, 1, 1024 * 1024, 32 * 1024 * 1024);
    WebpSource first = rgba_source(512, 512, 1);
    WebpSource second = rgba_source(512, 512, 2);
    const auto first_request = cache.prepare(first, {40, 0});
    const auto second_request = cache.prepare(second, {40, 0});
    require(cache.submit(first_request, std::move(first)), "queue first job");
    require(cache.submit(first_request, rgba_source(512, 512, 1)),
            "coalesce duplicate in-flight job");
    require(cache.stats().queued == 1,
            "duplicate in-flight job is queued only once");
    require(!cache.submit(second_request, std::move(second)),
            "bounded pending bytes reject second job");
    require(cache.stats().queue_drops == 1, "queue drop counted");
    flock(slot, LOCK_UN);
    wait_encoded(cache, 1);
  }
  close(slot);
  std::filesystem::remove_all(directory);
}

std::string numbered_key(uint64_t number) {
  static constexpr char hex[] = "0123456789abcdef";
  std::string key(64, '0');
  for (size_t index = 0; index < 16; ++index) {
    key[63 - index] = hex[number & 15];
    number >>= 4;
  }
  return key;
}

void create_sparse_file(const std::string &path, off_t size,
                        time_t modified = 0) {
  const int fd = open(path.c_str(), O_WRONLY | O_CREAT | O_EXCL, 0644);
  require(fd >= 0, "create sparse cache fixture");
  require(ftruncate(fd, size) == 0, "size sparse cache fixture");
  if (modified) {
    const timespec times[2]{{modified, 0}, {modified, 0}};
    require(futimens(fd, times) == 0, "date sparse cache fixture");
  }
  close(fd);
}

uint64_t ledger_checksum(const std::array<uint8_t, 64> &bytes) {
  uint64_t hash = UINT64_C(14695981039346656037);
  for (size_t index = 0; index < 56; ++index) {
    hash ^= bytes[index];
    hash *= UINT64_C(1099511628211);
  }
  return hash;
}

void store_le32(uint8_t *output, uint32_t value) {
  for (unsigned index = 0; index < 4; ++index)
    output[index] = static_cast<uint8_t>(value >> (index * 8));
}

void store_le64(uint8_t *output, uint64_t value) {
  for (unsigned index = 0; index < 8; ++index)
    output[index] = static_cast<uint8_t>(value >> (index * 8));
}

void mark_usage_ledger_dirty(const std::string &directory) {
  const std::string path = directory + "/.usage-v1";
  const int fd = open(path.c_str(), O_RDWR);
  require(fd >= 0, "open usage ledger for dirty repair test");
  std::array<uint8_t, 64> bytes{};
  require(read_exact(fd, bytes.data(), bytes.size()), "read usage ledger");
  store_le32(bytes.data() + 12, 2);
  store_le64(bytes.data() + 56, ledger_checksum(bytes));
  require(lseek(fd, 0, SEEK_SET) == 0 &&
              write_exact(fd, bytes.data(), bytes.size()) && fsync(fd) == 0,
          "persist dirty usage ledger");
  close(fd);
}

void corrupt_usage_ledger(const std::string &directory) {
  const std::string path = directory + "/.usage-v1";
  const int fd = open(path.c_str(), O_RDWR);
  require(fd >= 0, "open usage ledger for corruption test");
  uint8_t byte = 0;
  require(read_exact(fd, &byte, sizeof(byte)), "read usage magic");
  byte ^= 0xff;
  require(lseek(fd, 0, SEEK_SET) == 0 &&
              write_exact(fd, &byte, sizeof(byte)) && fsync(fd) == 0,
          "corrupt usage ledger");
  close(fd);
}

void test_usage_ledger_repair_and_scan_scaling() {
  const std::string directory = temporary_directory("ledger");
  constexpr size_t mebibyte = 1024 * 1024;
  {
    WebpCache cache(directory, 1, 8 * mebibyte, 32 * mebibyte);
    wait_disk_scans(cache, 1);
    const WebpSource first = rgba_source(64, 64, 101);
    const uint64_t key_bytes_before = cache.stats().key_bytes;
    const auto prepared = cache.prepare_source(first);
    const auto q40 = cache.prepare(prepared, {40, 0});
    const auto q80 = cache.prepare(prepared, {80, 0});
    const auto q95 = cache.prepare(prepared, {95, 0});
    require(q40.key != q80.key && q80.key != q95.key,
            "prepared source separates quality keys");
    require(cache.stats().key_bytes - key_bytes_before == first.pixels.size(),
            "prepared source hashes texture bytes once");
    require(cache.prepare(first, {40, 0}).key == q40.key,
            "prepared and direct cache keys are identical");
    require(cache.submit(q40, first), "queue first ledger entry");
    WebpSource second = rgba_source(64, 64, 102);
    const auto second_request = cache.prepare(second, {40, 0});
    require(cache.submit(second_request, std::move(second)),
            "queue second ledger entry");
    wait_encoded(cache, 2);
    require(cache.stats().disk_scans == 1,
            "below-limit writes reuse the ledger without rescanning");
  }

  {
    WebpCache cache(directory, 1, 8 * mebibyte, 32 * mebibyte);
    WebpSource source = rgba_source(64, 64, 103);
    const auto request = cache.prepare(source, {40, 0});
    require(cache.submit(request, std::move(source)),
            "queue entry through a second cache instance");
    wait_encoded(cache, 1);
    require(cache.stats().disk_scans == 0,
            "second cache instance trusts a clean shared ledger");
    require(cache.stats().webp_entries == 3,
            "shared ledger carries exact entry counts");
  }

  mark_usage_ledger_dirty(directory);
  {
    WebpCache cache(directory, 1, 8 * mebibyte, 32 * mebibyte);
    wait_disk_scans(cache, 1);
    require(cache.stats().webp_entries == 3,
            "dirty ledger repair restores directory usage");
  }
  corrupt_usage_ledger(directory);
  {
    WebpCache cache(directory, 1, 8 * mebibyte, 32 * mebibyte);
    wait_disk_scans(cache, 1);
    require(cache.stats().webp_entries == 3,
            "corrupt ledger repair restores directory usage");
  }
  std::filesystem::remove_all(directory);
}

void test_disk_growth_bound_and_stale_cleanup() {
  const std::string directory = temporary_directory("prune");
  constexpr size_t mebibyte = 1024 * 1024;
  for (uint64_t index = 0; index < 20; ++index)
    create_sparse_file(directory + "/" + numbered_key(index) + ".webp",
                       mebibyte);
  create_sparse_file(directory + "/" + numbered_key(100) + ".deny",
                     mebibyte);
  const std::string stale = directory + "/.tmp-" + numbered_key(200) +
      "-999999-1";
  create_sparse_file(stale, 2 * mebibyte, std::time(nullptr) - 7200);
  const std::string stale_deny = directory + "/" + numbered_key(201) +
      ".deny.tmp-999999-1";
  create_sparse_file(stale_deny, mebibyte, std::time(nullptr) - 7200);

  {
    WebpCache cache(directory, 1, 8 * mebibyte, 16 * mebibyte);
    wait_disk_scans(cache, 1);
    wait_stale_temp_removals(cache, 2);
    const auto stats = cache.stats();
    require(stats.disk_limit_bytes == 16 * mebibyte,
            "report configured disk bound");
    require(stats.webp_bytes + stats.deny_bytes <=
                stats.disk_limit_bytes * 9 / 10,
            "startup pruning returns disk cache below target");
    require(stats.disk_evictions > 0 && stats.disk_evicted_bytes > 0,
            "disk evictions are reported");
    require(stats.stale_temp_removals == 2 &&
                stats.stale_temp_bytes == 3 * mebibyte,
            "stale temporary data is removed and reported");
    require(!std::filesystem::exists(stale) &&
                !std::filesystem::exists(stale_deny),
            "stale temporary file is deleted");
  }
  std::filesystem::remove_all(directory);

  const std::string runtime_directory = temporary_directory("runtime-prune");
  for (uint64_t index = 0; index < 16; ++index)
    create_sparse_file(runtime_directory + "/" + numbered_key(index) +
                           ".webp",
                       mebibyte);
  {
    WebpCache cache(runtime_directory, 1, 8 * mebibyte, 16 * mebibyte);
    wait_disk_scans(cache, 1);
    require(cache.stats().webp_bytes == 16 * mebibyte &&
                cache.stats().disk_evictions == 0,
            "cache at its hard limit does not prune at startup");
    WebpSource source = rgba_source(64, 64, 91);
    const auto request = cache.prepare(source, {80, 0});
    require(cache.submit(request, std::move(source)),
            "queue write that crosses disk limit");
    wait_encoded(cache, 1);
    const auto stats = cache.stats();
    require(stats.webp_bytes + stats.deny_bytes <=
                stats.disk_limit_bytes * 9 / 10,
            "successful write immediately restores disk bound");
    require(stats.disk_evictions > 0,
            "runtime write reports disk evictions");
  }
  std::filesystem::remove_all(runtime_directory);

  const std::string inode_directory = temporary_directory("entry-prune");
  for (uint64_t index = 0; index < 300; ++index)
    create_sparse_file(inode_directory + "/" + numbered_key(index) +
                           ".deny",
                       1);
  for (uint64_t index = 0; index < 20; ++index)
    create_sparse_file(inode_directory + "/.tmp-" +
                           numbered_key(1000 + index) + "-999999-" +
                           std::to_string(index),
                       1);
  {
    WebpCache cache(inode_directory, 1, 8 * mebibyte, 16 * mebibyte);
    wait_disk_scans(cache, 1);
    const auto stats = cache.stats();
    require(stats.disk_entry_limit == 256,
            "small cache reports derived disk entry bound");
    require(stats.webp_entries + stats.deny_entries <=
                stats.disk_entry_limit * 9 / 10,
            "startup pruning enforces disk entry bound");
    require(stats.temp_entry_limit == 16 &&
                stats.temp_entries <= stats.temp_entry_limit,
            "fresh temporary file count is bounded");
    require(stats.temp_evictions >= 4,
            "fresh temporary entry pressure is reported");
  }
  std::filesystem::remove_all(inode_directory);
}

[[noreturn]] void run_bound_writer_child(const std::string &directory,
                                         uint8_t seed) {
  bool success = false;
  {
    WebpCache cache(directory, 2, 8 * 1024 * 1024, 16 * 1024 * 1024);
    WebpSource source = rgba_source(128, 128, seed);
    const auto request = cache.prepare(source, {40, 0});
    if (cache.submit(request, std::move(source))) {
      for (unsigned attempt = 0; attempt < 1000; ++attempt) {
        const auto stats = cache.stats();
        if (stats.encoded == 1) {
          success = true;
          break;
        }
        if (stats.encode_errors != 0) break;
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
      }
    }
  }
  _exit(success ? 0 : 30);
}

void test_concurrent_writes_respect_hard_bound() {
  const std::string directory = temporary_directory("concurrent-bound");
  constexpr size_t mebibyte = 1024 * 1024;
  for (uint64_t index = 0; index < 16; ++index)
    create_sparse_file(directory + "/" + numbered_key(index) + ".webp",
                       mebibyte);
  std::array<pid_t, 2> children{};
  for (size_t index = 0; index < children.size(); ++index) {
    children[index] = fork();
    require(children[index] >= 0, "fork bounded cache writer");
    if (children[index] == 0)
      run_bound_writer_child(directory, static_cast<uint8_t>(120 + index));
  }
  for (pid_t child : children) {
    int status = 0;
    require(waitpid(child, &status, 0) == child && WIFEXITED(status) &&
                WEXITSTATUS(status) == 0,
            "concurrent bounded cache writer exits cleanly");
  }
  uint64_t bytes = 0;
  uint64_t entries = 0;
  for (const auto &item : std::filesystem::directory_iterator(directory)) {
    const std::string name = item.path().filename().string();
    const bool permanent = name.size() == 69 &&
        (name.ends_with(".webp") || name.ends_with(".deny"));
    if (!permanent || !item.is_regular_file()) continue;
    ++entries;
    bytes += item.file_size();
  }
  require(bytes <= 16 * mebibyte,
          "concurrent publications preserve the hard byte bound");
  require(entries <= 256,
          "concurrent publications preserve the hard entry bound");
  {
    WebpCache cache(directory, 1, 8 * mebibyte, 16 * mebibyte);
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    require(cache.stats().disk_scans == 0,
            "concurrent writers leave a clean exact ledger");
  }
  std::filesystem::remove_all(directory);
}

void test_async_failure_and_disabled_cache() {
  const std::string directory = temporary_directory("failure");
  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    const WebpSource source = rgba_source(64, 64, 33);
    const auto invalid_request = cache.prepare(source, {40, 7});
    require(cache.submit(invalid_request, source),
            "queue request with invalid encoder policy");
    wait_encode_errors(cache, 1);
    require(cache.stats().encoded == 0, "failed encode does not count success");
    require(!cache.lookup(invalid_request),
            "failed encode leaves an exact-stream cache miss");
    require(!std::filesystem::exists(
                directory + "/" + invalid_request.key + ".webp"),
            "failed encode leaves no cache entry");
  }

  const std::string blocker = directory + "/not-a-directory";
  const int fd = open(blocker.c_str(), O_WRONLY | O_CREAT | O_EXCL, 0644);
  require(fd >= 0, "create cache-directory blocker");
  close(fd);
  {
    WebpCache disabled(blocker + "/cache", 1, 8 * 1024 * 1024,
                       32 * 1024 * 1024);
    const WebpSource source = rgba_source(8, 8, 44);
    const auto request = disabled.prepare(source, {40, 0});
    require(!disabled.enabled(), "filesystem failure disables cache");
    require(!disabled.lookup(request), "disabled cache lookup is a miss");
    require(!disabled.submit(request, source),
            "disabled cache rejects background work");
  }

  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    const WebpSource source = rgba_source(8, 8, 45);
    auto request = cache.prepare(source, {40, 0});
    request.key.assign(64, '/');
    require(!cache.lookup(request), "reject malformed lookup key");
    require(!cache.submit(request, source), "reject malformed submit key");
    cache.deny(request);
  }
  std::filesystem::remove_all(directory);
}

void test_async_deny_is_immediate_and_cannot_resurrect_entry() {
  const std::string directory = temporary_directory("deny-race");
  std::string entry_path;
  std::string deny_path;
  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    const WebpSource source = rgba_source(256, 256, 55);
    const auto request = cache.prepare(source, {40, 0});
    entry_path = directory + "/" + request.key + ".webp";
    deny_path = directory + "/" + request.key + ".deny";
    const std::string key_lock_path = directory + "/.key-" +
        request.key.substr(0, 2) + ".lock";
    const int key_lock = open(key_lock_path.c_str(), O_RDWR | O_CREAT, 0644);
    require(key_lock >= 0 && flock(key_lock, LOCK_EX) == 0,
            "hold same-key worker lock");
    require(cache.submit(request, source), "queue job before deny");
    const auto started = std::chrono::steady_clock::now();
    cache.deny(request);
    const auto elapsed = std::chrono::steady_clock::now() - started;
    require(elapsed < std::chrono::milliseconds(250),
            "deny performs no synchronous disk work");
    cache.deny(request);
    require(!cache.submit(request, source), "deny rejects later submission");
    require(!cache.lookup(request), "deny forces exact raw fallback");
    require(!std::filesystem::exists(deny_path),
            "held worker lock delays persistent deny work");
    require(cache.stats().denied == 2 && cache.stats().deny_queued == 1,
            "duplicate deny disk work is coalesced");
    flock(key_lock, LOCK_UN);
    close(key_lock);
    wait_for_path(deny_path, true);
    wait_for_path(entry_path, false);
  }
  require(std::filesystem::exists(deny_path), "deny marker persists");
  require(!std::filesystem::exists(entry_path),
          "queued job cannot recreate denied entry");
  std::filesystem::remove_all(directory);
}

void test_shutdown_cancels_slot_wait() {
  const std::string directory = temporary_directory("shutdown");
  const std::string slot_path = directory + "/.slot-0.lock";
  const int slot = open(slot_path.c_str(), O_RDWR | O_CREAT, 0644);
  require(slot >= 0 && flock(slot, LOCK_EX) == 0, "hold shutdown slot");
  std::string entry_path;
  const auto started = std::chrono::steady_clock::now();
  {
    WebpCache cache(directory, 1, 8 * 1024 * 1024, 32 * 1024 * 1024);
    WebpSource source = rgba_source(512, 512, 66);
    const auto request = cache.prepare(source, {40, 0});
    entry_path = directory + "/" + request.key + ".webp";
    require(cache.submit(request, std::move(source)),
            "queue job while host slot is busy");
  }
  const auto elapsed = std::chrono::steady_clock::now() - started;
  require(elapsed < std::chrono::seconds(2),
          "shutdown does not wait for a busy host slot");
  require(!std::filesystem::exists(entry_path),
          "shutdown cancellation leaves no partial cache entry");
  flock(slot, LOCK_UN);
  close(slot);
  std::filesystem::remove_all(directory);
}

struct ChildReport {
  uint64_t encoded = 0;
  uint64_t encode_errors = 0;
  uint64_t disk_loads = 0;
  uint64_t queued = 0;
  uint8_t submitted = 0;
  uint8_t observed_hit = 0;
};

struct ChildProcess {
  pid_t pid = -1;
  int ready_fd = -1;
  int report_fd = -1;
};

[[noreturn]] void run_concurrent_child(const std::string &directory,
                                       int inherited_slot_zero,
                                       int inherited_slot_one, int ready_fd,
                                       int report_fd) {
  close(inherited_slot_zero);
  close(inherited_slot_one);
  ChildReport report;
  {
    WebpCache cache(directory, 2, 8 * 1024 * 1024, 32 * 1024 * 1024);
    const WebpSource source = rgba_source(512, 512, 77);
    const auto request = cache.prepare(source, {80, 0});
    report.submitted = cache.submit(request, source) ? 1 : 0;
    const uint8_t ready = 1;
    if (!write_exact(ready_fd, &ready, sizeof(ready))) _exit(20);
    close(ready_fd);

    const auto hit_deadline = std::chrono::steady_clock::now() +
        std::chrono::seconds(5);
    while (std::chrono::steady_clock::now() < hit_deadline) {
      if (cache.lookup(request)) {
        report.observed_hit = 1;
        break;
      }
      std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }

    // One worker encodes. The other worker loads the completed disk entry.
    // Wait for either terminal outcome before reporting per-process stats.
    const auto settle_deadline = std::chrono::steady_clock::now() +
        std::chrono::seconds(1);
    while (cache.stats().encoded == 0 && cache.stats().disk_loads == 0 &&
           std::chrono::steady_clock::now() < settle_deadline)
      std::this_thread::sleep_for(std::chrono::milliseconds(5));
    const auto stats = cache.stats();
    report.encoded = stats.encoded;
    report.encode_errors = stats.encode_errors;
    report.disk_loads = stats.disk_loads;
    report.queued = stats.queued;
  }
  if (!write_exact(report_fd, &report, sizeof(report))) _exit(21);
  close(report_fd);
  _exit(0);
}

ChildProcess spawn_concurrent_child(const std::string &directory,
                                    int inherited_slot_zero,
                                    int inherited_slot_one) {
  int ready_pipe[2]{};
  int report_pipe[2]{};
  require(pipe(ready_pipe) == 0, "create child ready pipe");
  require(pipe(report_pipe) == 0, "create child report pipe");
  const pid_t pid = fork();
  require(pid >= 0, "fork cache process");
  if (pid == 0) {
    close(ready_pipe[0]);
    close(report_pipe[0]);
    run_concurrent_child(directory, inherited_slot_zero, inherited_slot_one,
                         ready_pipe[1], report_pipe[1]);
  }
  close(ready_pipe[1]);
  close(report_pipe[1]);
  return {pid, ready_pipe[0], report_pipe[0]};
}

void test_multiprocess_same_key_encodes_once() {
  const std::string directory = temporary_directory("multiprocess");
  const std::string slot_zero_path = directory + "/.slot-0.lock";
  const std::string slot_one_path = directory + "/.slot-1.lock";
  const int slot_zero = open(slot_zero_path.c_str(), O_RDWR | O_CREAT, 0644);
  const int slot_one = open(slot_one_path.c_str(), O_RDWR | O_CREAT, 0644);
  require(slot_zero >= 0 && slot_one >= 0 &&
              flock(slot_zero, LOCK_EX) == 0 &&
              flock(slot_one, LOCK_EX) == 0,
          "hold both shared multi-process slots");

  std::array<ChildProcess, 2> children{
    spawn_concurrent_child(directory, slot_zero, slot_one),
    spawn_concurrent_child(directory, slot_zero, slot_one),
  };
  for (auto &child : children) {
    uint8_t ready = 0;
    require(read_exact(child.ready_fd, &ready, sizeof(ready)) && ready == 1,
            "child queued same-key request");
    close(child.ready_fd);
  }
  flock(slot_zero, LOCK_UN);
  flock(slot_one, LOCK_UN);
  close(slot_zero);
  close(slot_one);

  std::array<ChildReport, 2> reports{};
  for (size_t index = 0; index < children.size(); ++index) {
    require(read_exact(children[index].report_fd, &reports[index],
                       sizeof(reports[index])),
            "read child cache report");
    close(children[index].report_fd);
    int status = 0;
    require(waitpid(children[index].pid, &status, 0) == children[index].pid &&
                WIFEXITED(status) && WEXITSTATUS(status) == 0,
            "cache child exits cleanly");
  }

  require(reports[0].submitted && reports[1].submitted,
          "both relay processes accepted the same-key request");
  require(reports[0].queued == 1 && reports[1].queued == 1,
          "each relay queued one local request");
  require(reports[0].observed_hit && reports[1].observed_hit,
          "both relay processes observe the shared cache entry");
  require(reports[0].encode_errors + reports[1].encode_errors == 0,
          "same-key race has no encode errors");
  require(reports[0].encoded + reports[1].encoded == 1,
          "two host slots encode a same-key resource exactly once");
  require(reports[0].disk_loads + reports[1].disk_loads == 1,
          "second process loads the shared same-key entry in background");

  std::filesystem::remove_all(directory);
}

}  // namespace

int main() {
  test_source_decode();
  test_encoder_alpha();
  test_method_keys_and_legacy_disk_fallback();
  test_legacy_deny_propagates_without_method0_encode();
  test_cache_hit_corruption_and_deny();
  test_queue_bound_and_duplicate_coalescing();
  test_usage_ledger_repair_and_scan_scaling();
  test_disk_growth_bound_and_stale_cleanup();
  test_concurrent_writes_respect_hard_bound();
  test_async_failure_and_disabled_cache();
  test_async_deny_is_immediate_and_cannot_resurrect_entry();
  test_shutdown_cancels_slot_wait();
  test_multiprocess_same_key_encodes_once();
  std::puts("WebP cache tests passed");
  return 0;
}