#include "w3cs_webp_cache.h"

#include <dirent.h>
#include <fcntl.h>
#include <glib.h>
#include <sys/file.h>
#include <sys/stat.h>
#include <unistd.h>
#include <webp/decode.h>
#include <webp/encode.h>

#include <algorithm>
#include <atomic>
#include <cerrno>
#include <chrono>
#include <condition_variable>
#include <cstdio>
#include <cstring>
#include <ctime>
#include <deque>
#include <filesystem>
#include <iterator>
#include <list>
#include <mutex>
#include <optional>
#include <string>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

namespace w3cs {
namespace {

constexpr uint32_t kFormatA8R8G8B8 = 21;
constexpr uint32_t kFormatX8R8G8B8 = 22;
constexpr uint32_t kFormatR5G6B5 = 23;
constexpr uint32_t kFormatX1R5G5B5 = 24;
constexpr uint32_t kFormatA1R5G5B5 = 25;
constexpr uint32_t kFormatA4R4G4B4 = 26;
constexpr uint16_t kCacheVersion = 1;
constexpr char kCacheMagic[4] = {'W', '3', 'W', 'P'};
constexpr uint16_t kUsageVersion = 1;
constexpr uint32_t kUsageClean = 1;
constexpr uint32_t kUsageDirty = 2;
constexpr char kUsageMagic[8] = {'W', '3', 'U', 'S', 'A', 'G', 'E', '1'};
constexpr size_t kMaximumImageBytes = 256u * 1024u * 1024u;
constexpr size_t kMaximumMemoryFrontBytes = 64u * 1024u * 1024u;
constexpr size_t kMinimumDiskEntries = 256u;
constexpr size_t kMaximumDiskEntries = 65536u;
constexpr size_t kDiskEntrySizingBytes = 64u * 1024u;
constexpr size_t kMinimumTempBytes = 4u * 1024u * 1024u;
constexpr size_t kMaximumTempBytes = 512u * 1024u * 1024u;
constexpr time_t kStaleTempAgeSeconds = 60 * 60;

#pragma pack(push, 1)
struct CacheHeader {
  char magic[4];
  uint16_t version;
  uint16_t header_size;
  uint8_t key[32];
  uint8_t encoded_digest[32];
  uint32_t encoder_version;
  uint32_t width;
  uint32_t height;
  uint32_t pitch;
  uint32_t format;
  uint32_t decoded_size;
  uint32_t encoded_size;
  uint8_t quality;
  uint8_t method;
  uint8_t reserved[2];
};

struct UsageLedger {
  char magic[8];
  uint16_t version;
  uint16_t header_size;
  uint32_t state;
  uint64_t generation;
  uint64_t webp_entries;
  uint64_t webp_bytes;
  uint64_t deny_entries;
  uint64_t deny_bytes;
  uint64_t checksum;
};
#pragma pack(pop)

static_assert(sizeof(CacheHeader) == 104);
static_assert(sizeof(UsageLedger) == 64);

struct DiskUsage {
  uint64_t generation = 0;
  uint64_t webp_entries = 0;
  uint64_t webp_bytes = 0;
  uint64_t deny_entries = 0;
  uint64_t deny_bytes = 0;
};

uint64_t usage_checksum(const UsageLedger &ledger) {
  const auto *bytes = reinterpret_cast<const uint8_t *>(&ledger);
  uint64_t hash = UINT64_C(14695981039346656037);
  for (size_t index = 0; index < offsetof(UsageLedger, checksum); ++index) {
    hash ^= bytes[index];
    hash *= UINT64_C(1099511628211);
  }
  return hash;
}

uint64_t usage_bytes(const DiskUsage &usage) {
  return usage.webp_bytes + usage.deny_bytes;
}

uint64_t usage_entries(const DiskUsage &usage) {
  return usage.webp_entries + usage.deny_entries;
}

bool exceeds_limit(uint64_t current, uint64_t added, uint64_t limit) {
  return current > limit || added > limit - current;
}

std::array<uint8_t, 32> sha256(const uint8_t *data, size_t size) {
  std::array<uint8_t, 32> digest{};
  GChecksum *checksum = g_checksum_new(G_CHECKSUM_SHA256);
  if (!checksum) return digest;
  if (size) g_checksum_update(checksum, data, size);
  gsize digest_size = digest.size();
  g_checksum_get_digest(checksum, digest.data(), &digest_size);
  g_checksum_free(checksum);
  if (digest_size != digest.size()) digest.fill(0);
  return digest;
}

std::string hex_digest(const std::array<uint8_t, 32> &digest) {
  static constexpr char hex[] = "0123456789abcdef";
  std::string result(digest.size() * 2, '0');
  for (size_t index = 0; index < digest.size(); ++index) {
    result[index * 2] = hex[digest[index] >> 4];
    result[index * 2 + 1] = hex[digest[index] & 15];
  }
  return result;
}

void append_u32(WebpBytes &bytes, uint32_t value) {
  value = GUINT32_TO_LE(value);
  const auto *raw = reinterpret_cast<const uint8_t *>(&value);
  bytes.insert(bytes.end(), raw, raw + sizeof(value));
}

bool read_all(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;
}

bool write_all(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 hex_key(const char *name) {
  if (!name) return false;
  for (size_t index = 0; index < 64; ++index) {
    const char byte = name[index];
    if (!((byte >= '0' && byte <= '9') || (byte >= 'a' && byte <= 'f')))
      return false;
  }
  return true;
}

bool valid_key(const std::string &key) {
  return key.size() == 64 && hex_key(key.c_str());
}

bool key_name(const char *name, const char *suffix) {
  return name && std::strlen(name) == 64 + std::strlen(suffix) &&
      hex_key(name) && !std::strcmp(name + 64, suffix);
}

bool temp_name(const char *name) {
  if (!name) return false;
  if (!std::strncmp(name, ".tmp-", 5)) {
    const char *key = name + 5;
    return std::strlen(key) > 65 && hex_key(key) && key[64] == '-';
  }
  return std::strlen(name) > 74 && hex_key(name) &&
      !std::strncmp(name + 64, ".deny.tmp-", 10);
}

uint8_t expand5(uint16_t value) {
  return static_cast<uint8_t>((value << 3) | (value >> 2));
}

uint8_t expand6(uint16_t value) {
  return static_cast<uint8_t>((value << 2) | (value >> 4));
}

}  // namespace

class WebpCache::Impl {
 public:
  struct Job {
    WebpRequest request;
    WebpSource source;
    std::optional<WebpRequest> read_fallback;
  };

  struct MemoryEntry {
    WebpBytes encoded;
    std::list<std::string>::iterator lru;
  };

  Impl(std::string directory, unsigned slots, size_t queue_limit,
       size_t cache_limit)
      : cache_dir(std::move(directory)), host_slots(std::clamp(slots, 1u, 16u)),
        queue_byte_limit(std::max<size_t>(queue_limit, 1024 * 1024)),
        cache_byte_limit(std::max<size_t>(cache_limit, 16 * 1024 * 1024)),
        cache_entry_limit(std::clamp<size_t>(
            cache_byte_limit / kDiskEntrySizingBytes,
            kMinimumDiskEntries, kMaximumDiskEntries)),
        temp_entry_limit(std::clamp<size_t>(cache_entry_limit / 16u,
                                            16u, 256u)),
        temp_byte_limit(std::min(cache_byte_limit,
            std::clamp<size_t>(cache_byte_limit / 8u,
                               kMinimumTempBytes, kMaximumTempBytes))),
        memory_byte_limit(std::min(cache_byte_limit,
                                   kMaximumMemoryFrontBytes)),
        deny_queue_limit(std::clamp<size_t>(queue_byte_limit / 64, 1024,
                                            65536)) {
    if (cache_dir.empty()) return;
    std::error_code error;
    std::filesystem::create_directories(cache_dir, error);
    if (error) {
      std::fprintf(stderr, "native WebP cache disabled: mkdir %s: %s\n",
                   cache_dir.c_str(), error.message().c_str());
      return;
    }
    enabled_flag = true;
    worker = std::thread([this] { work(); });
  }

  ~Impl() {
    {
      std::lock_guard lock(mutex);
      stopping = true;
      queue.clear();
      inflight.clear();
      queue_bytes = 0;
    }
    changed.notify_all();
    if (worker.joinable()) worker.join();
  }

  std::string entry_path(const std::string &key) const {
    return cache_dir + "/" + key + ".webp";
  }

  std::string deny_path(const std::string &key) const {
    return cache_dir + "/" + key + ".deny";
  }

  std::string key_lock_path(const std::string &key) const {
    // The first digest byte selects one of 256 fixed lock buckets. This keeps
    // lock-file growth bounded while still serializing every identical key.
    return cache_dir + "/.key-" + key.substr(0, 2) + ".lock";
  }

  std::string usage_path() const { return cache_dir + "/.usage-v1"; }

  std::string prune_lock_path() const {
    return cache_dir + "/.prune.lock";
  }

  bool locally_denied(const std::string &key) {
    std::lock_guard lock(memory_mutex);
    return memory_denied.count(key) != 0;
  }

  bool try_locally_denied(const std::string &key) {
    std::unique_lock lock(memory_mutex, std::try_to_lock);
    return lock.owns_lock() && memory_denied.count(key) != 0;
  }

  void update_memory_occupancy_locked() {
    stat_memory_entries.store(memory.size());
    stat_memory_bytes.store(memory_bytes);
  }

  void remove_memory_locked(const std::string &key) {
    const auto found = memory.find(key);
    if (found == memory.end()) return;
    memory_bytes -= std::min(memory_bytes, found->second.encoded.size());
    memory_lru.erase(found->second.lru);
    memory.erase(found);
    update_memory_occupancy_locked();
  }

  bool publish_memory(const std::string &key, WebpBytes encoded) {
    if (encoded.empty() || encoded.size() > memory_byte_limit) return false;
    std::lock_guard lock(memory_mutex);
    if (memory_denied.count(key)) return false;
    remove_memory_locked(key);
    while (!memory_lru.empty() &&
           encoded.size() > memory_byte_limit -
               std::min(memory_bytes, memory_byte_limit)) {
      const std::string oldest = memory_lru.back();
      const auto found = memory.find(oldest);
      if (found == memory.end()) {
        memory_lru.pop_back();
        continue;
      }
      const size_t removed = found->second.encoded.size();
      memory_bytes -= std::min(memory_bytes, removed);
      memory_lru.erase(found->second.lru);
      memory.erase(found);
      ++stat_memory_evictions;
      stat_memory_evicted_bytes += removed;
    }
    memory_lru.push_front(key);
    auto lru = memory_lru.begin();
    memory_bytes += encoded.size();
    memory.emplace(key, MemoryEntry{std::move(encoded), lru});
    update_memory_occupancy_locked();
    return true;
  }

  std::optional<WebpBytes> lookup_memory(const std::string &key) {
    std::unique_lock lock(memory_mutex, std::try_to_lock);
    if (!lock.owns_lock()) {
      ++stat_misses;
      ++stat_memory_busy_misses;
      return std::nullopt;
    }
    const auto found = memory.find(key);
    if (found == memory.end() || memory_denied.count(key)) {
      ++stat_misses;
      return std::nullopt;
    }
    memory_lru.splice(memory_lru.begin(), memory_lru, found->second.lru);
    ++stat_hits;
    return found->second.encoded;
  }

  bool denied(const std::string &key) const {
    struct stat state{};
    return lstat(deny_path(key).c_str(), &state) == 0 &&
        S_ISREG(state.st_mode);
  }

  std::optional<WebpBytes> read_entry(const WebpRequest &request) {
    if (denied(request.key)) return std::nullopt;
    const std::string path = entry_path(request.key);
    const int fd = open(path.c_str(), O_RDONLY | O_CLOEXEC | O_NOFOLLOW);
    if (fd < 0) return std::nullopt;
    struct stat state{};
    CacheHeader header{};
    bool valid = fstat(fd, &state) == 0 && S_ISREG(state.st_mode) &&
        read_all(fd, &header, sizeof(header));
    const uint16_t version = GUINT16_FROM_LE(header.version);
    const uint16_t header_size = GUINT16_FROM_LE(header.header_size);
    const uint32_t encoded_size = GUINT32_FROM_LE(header.encoded_size);
    valid = valid && !std::memcmp(header.magic, kCacheMagic, sizeof(kCacheMagic)) &&
        version == kCacheVersion && header_size == sizeof(header) &&
        !std::memcmp(header.key, request.digest.data(), request.digest.size()) &&
        GUINT32_FROM_LE(header.encoder_version) == request.encoder_version &&
        GUINT32_FROM_LE(header.width) == request.width &&
        GUINT32_FROM_LE(header.height) == request.height &&
        GUINT32_FROM_LE(header.pitch) == request.pitch &&
        GUINT32_FROM_LE(header.format) == request.format &&
        GUINT32_FROM_LE(header.decoded_size) == request.decoded_size &&
        header.quality == request.policy.quality &&
        header.method == request.policy.method &&
        encoded_size > 0 && encoded_size <= kMaximumImageBytes &&
        state.st_size == static_cast<off_t>(sizeof(header) + encoded_size);
    WebpBytes encoded(valid ? encoded_size : 0);
    valid = valid && read_all(fd, encoded.data(), encoded.size());
    close(fd);
    int webp_width = 0;
    int webp_height = 0;
    const auto encoded_digest = sha256(encoded.data(), encoded.size());
    valid = valid && !std::memcmp(header.encoded_digest,
                                  encoded_digest.data(),
                                  encoded_digest.size()) &&
        WebPGetInfo(encoded.data(), encoded.size(), &webp_width,
                    &webp_height) &&
        webp_width == static_cast<int>(request.width) &&
        webp_height == static_cast<int>(request.height);
    if (!valid) {
      ++stat_corrupt_entries;
      remove_corrupt_entry(path);
      return std::nullopt;
    }
    return encoded;
  }

  bool load_disk_entry(const WebpRequest &request, bool fallback) {
    auto cached = read_entry(request);
    if (!cached) return false;
    const size_t bytes = cached->size();
    if (publish_memory(request.key, std::move(*cached))) {
      ++stat_disk_loads;
      stat_disk_loaded_bytes += bytes;
      if (fallback) {
        ++stat_fallback_disk_loads;
        stat_fallback_disk_loaded_bytes += bytes;
      }
    }
    return true;
  }

  bool propagate_deny(const WebpRequest &request) {
    {
      std::lock_guard lock(memory_mutex);
      memory_denied.insert(request.key);
      remove_memory_locked(request.key);
    }
    const bool written = write_deny_marker(request.key);
    if (!written) {
      std::lock_guard lock(memory_mutex);
      memory_denied.erase(request.key);
    }
    ++stat_denied;
    if (!written) ++stat_deny_write_errors;
    return written;
  }

  bool write_entry(const WebpRequest &request, const WebpBytes &encoded) {
    if (denied(request.key) || encoded.empty() ||
        encoded.size() > kMaximumImageBytes ||
        sizeof(CacheHeader) > cache_byte_limit ||
        encoded.size() > cache_byte_limit - sizeof(CacheHeader) ||
        encoded.size() > temp_byte_limit -
            std::min(temp_byte_limit, sizeof(CacheHeader)))
      return false;
    int width = 0;
    int height = 0;
    if (!WebPGetInfo(encoded.data(), encoded.size(), &width, &height) ||
        width != static_cast<int>(request.width) ||
        height != static_cast<int>(request.height))
      return false;

    CacheHeader header{};
    std::memcpy(header.magic, kCacheMagic, sizeof(kCacheMagic));
    header.version = GUINT16_TO_LE(kCacheVersion);
    header.header_size = GUINT16_TO_LE(sizeof(header));
    std::memcpy(header.key, request.digest.data(), request.digest.size());
    const auto encoded_digest = sha256(encoded.data(), encoded.size());
    std::memcpy(header.encoded_digest, encoded_digest.data(),
                encoded_digest.size());
    header.encoder_version = GUINT32_TO_LE(request.encoder_version);
    header.width = GUINT32_TO_LE(request.width);
    header.height = GUINT32_TO_LE(request.height);
    header.pitch = GUINT32_TO_LE(request.pitch);
    header.format = GUINT32_TO_LE(request.format);
    header.decoded_size = GUINT32_TO_LE(request.decoded_size);
    header.encoded_size = GUINT32_TO_LE(static_cast<uint32_t>(encoded.size()));
    header.quality = request.policy.quality;
    header.method = request.policy.method;

    const uint64_t record_size = sizeof(header) + encoded.size();
    const int prune_lock = acquire_lock(prune_lock_path());
    if (prune_lock < 0) return false;
    DiskUsage usage;
    bool ready = ensure_usage_locked(&usage);
    if (ready &&
        (exceeds_limit(usage_bytes(usage), record_size, cache_byte_limit) ||
         exceeds_limit(usage_entries(usage), 1, cache_entry_limit))) {
      ready = scan_and_prune_locked(record_size, 1, usage.generation,
                                    &usage);
    }
    if (!ready || denied(request.key) ||
        exceeds_limit(usage_bytes(usage), record_size, cache_byte_limit) ||
        exceeds_limit(usage_entries(usage), 1, cache_entry_limit)) {
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }

    const uint64_t serial = temp_serial.fetch_add(1);
    const std::string final_path = entry_path(request.key);
    const std::string temp_path = cache_dir + "/.tmp-" + request.key + "-" +
        std::to_string(getpid()) + "-" + std::to_string(serial);
    struct stat old_state{};
    const bool replaced = lstat(final_path.c_str(), &old_state) == 0 &&
        S_ISREG(old_state.st_mode);
    const uint64_t old_size = replaced
        ? static_cast<uint64_t>(old_state.st_size) : 0;
    const uint64_t generation = usage.generation + 1;
    if (!write_usage_locked(usage, kUsageDirty, generation)) {
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }
    const int fd = open(temp_path.c_str(), O_WRONLY | O_CREAT | O_EXCL |
                        O_CLOEXEC | O_NOFOLLOW, 0644);
    if (fd < 0) {
      write_usage_locked(usage, kUsageClean, generation);
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }
    const bool written = write_all(fd, &header, sizeof(header)) &&
        write_all(fd, encoded.data(), encoded.size()) && fsync(fd) == 0;
    close(fd);
    if (!written || denied(request.key)) {
      const bool removed = unlink(temp_path.c_str()) == 0 || errno == ENOENT;
      if (removed && fsync_cache_directory())
        write_usage_locked(usage, kUsageClean, generation);
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }
    if (rename(temp_path.c_str(), final_path.c_str()) != 0) {
      const bool removed = unlink(temp_path.c_str()) == 0 || errno == ENOENT;
      if (removed && fsync_cache_directory())
        write_usage_locked(usage, kUsageClean, generation);
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }
    if (!fsync_cache_directory()) {
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }
    if (replaced) {
      usage.webp_bytes -= std::min(usage.webp_bytes, old_size);
    } else {
      ++usage.webp_entries;
    }
    usage.webp_bytes += record_size;
    usage.generation = generation;
    const bool committed = write_usage_locked(usage, kUsageClean,
                                               generation);
    if (committed) publish_usage_stats(usage);
    flock(prune_lock, LOCK_UN);
    close(prune_lock);
    return committed;
  }

  bool write_deny_marker(const std::string &key) {
    const std::string final_path = deny_path(key);
    const std::string webp_path = entry_path(key);
    const std::string line = key + "\n";
    const int prune_lock = acquire_lock(prune_lock_path());
    if (prune_lock < 0) return false;
    DiskUsage usage;
    bool ready = ensure_usage_locked(&usage);
    struct stat deny_state{};
    bool had_deny = lstat(final_path.c_str(), &deny_state) == 0 &&
        S_ISREG(deny_state.st_mode);
    if (ready && !had_deny &&
        (exceeds_limit(usage_bytes(usage), line.size(), cache_byte_limit) ||
         exceeds_limit(usage_entries(usage), 1, cache_entry_limit))) {
      ready = scan_and_prune_locked(line.size(), 1, usage.generation,
                                    &usage);
      had_deny = lstat(final_path.c_str(), &deny_state) == 0 &&
          S_ISREG(deny_state.st_mode);
    }
    if (!ready || (!had_deny &&
        (exceeds_limit(usage_bytes(usage), line.size(), cache_byte_limit) ||
         exceeds_limit(usage_entries(usage), 1, cache_entry_limit)))) {
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }
    struct stat webp_state{};
    const bool had_webp = lstat(webp_path.c_str(), &webp_state) == 0 &&
        S_ISREG(webp_state.st_mode);
    const uint64_t generation = usage.generation + 1;
    if (!write_usage_locked(usage, kUsageDirty, generation)) {
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }
    if (!had_deny) {
      const uint64_t serial = temp_serial.fetch_add(1);
      const std::string temp_path = final_path + ".tmp-" +
          std::to_string(getpid()) + "-" + std::to_string(serial);
      const int fd = open(temp_path.c_str(), O_WRONLY | O_CREAT | O_EXCL |
                          O_CLOEXEC | O_NOFOLLOW, 0644);
      if (fd < 0) {
        write_usage_locked(usage, kUsageClean, generation);
        flock(prune_lock, LOCK_UN);
        close(prune_lock);
        return false;
      }
      const bool written = write_all(fd, line.data(), line.size()) &&
          fsync(fd) == 0;
      close(fd);
      if (!written || rename(temp_path.c_str(), final_path.c_str()) != 0) {
        const bool removed = unlink(temp_path.c_str()) == 0 || errno == ENOENT;
        if (removed && fsync_cache_directory())
          write_usage_locked(usage, kUsageClean, generation);
        flock(prune_lock, LOCK_UN);
        close(prune_lock);
        return false;
      }
    }
    if (unlink(webp_path.c_str()) != 0 && errno != ENOENT) {
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }
    if (!fsync_cache_directory()) {
      flock(prune_lock, LOCK_UN);
      close(prune_lock);
      return false;
    }
    if (!had_deny) {
      ++usage.deny_entries;
      usage.deny_bytes += line.size();
    }
    if (had_webp) {
      --usage.webp_entries;
      usage.webp_bytes -= std::min(
          usage.webp_bytes, static_cast<uint64_t>(webp_state.st_size));
    }
    usage.generation = generation;
    const bool committed = write_usage_locked(usage, kUsageClean,
                                               generation);
    if (committed) publish_usage_stats(usage);
    flock(prune_lock, LOCK_UN);
    close(prune_lock);
    return committed;
  }

  int acquire_lock(const std::string &path) {
    while (true) {
      {
        std::lock_guard lock(mutex);
        if (stopping) return -1;
      }
      const int fd = open(path.c_str(), O_RDWR | O_CREAT | O_CLOEXEC |
                          O_NOFOLLOW, 0644);
      if (fd >= 0) {
        if (flock(fd, LOCK_EX | LOCK_NB) == 0) return fd;
        close(fd);
      }
      std::unique_lock lock(mutex);
      changed.wait_for(lock, std::chrono::milliseconds(5),
                       [this] { return stopping; });
      if (stopping) return -1;
    }
  }

  int acquire_slot() {
    while (true) {
      {
        std::lock_guard lock(mutex);
        if (stopping) return -1;
      }
      for (unsigned slot = 0; slot < host_slots; ++slot) {
        const std::string path = cache_dir + "/.slot-" +
            std::to_string(slot) + ".lock";
        const int fd = open(path.c_str(), O_RDWR | O_CREAT | O_CLOEXEC |
                            O_NOFOLLOW, 0644);
        if (fd < 0) continue;
        if (flock(fd, LOCK_EX | LOCK_NB) == 0) return fd;
        close(fd);
      }
      std::unique_lock lock(mutex);
      changed.wait_for(lock, std::chrono::milliseconds(5),
                       [this] { return stopping; });
      if (stopping) return -1;
    }
  }

  int acquire_key_lock(const std::string &key) {
    return acquire_lock(key_lock_path(key));
  }

  void finish_job(const std::string &key, size_t bytes) {
    std::lock_guard lock(mutex);
    inflight.erase(key);
    queue_bytes -= std::min(queue_bytes, bytes);
  }

  void finish_deny(const std::string &key) {
    std::unique_lock memory_lock(memory_mutex);
    std::lock_guard queue_lock(mutex);
    memory_denied.erase(key);
    deny_pending.erase(key);
  }

  void process_denies(const std::vector<std::string> &keys) {
    for (const std::string &key : keys) {
      const int key_lock = acquire_key_lock(key);
      bool written = false;
      if (key_lock >= 0) {
        written = write_deny_marker(key);
        flock(key_lock, LOCK_UN);
        close(key_lock);
      }
      if (!written) ++stat_deny_write_errors;
      finish_deny(key);
    }
  }

  void work() {
    initialize_usage();
    while (true) {
      Job job;
      bool have_job = false;
      std::vector<std::string> denies;
      {
        std::unique_lock lock(mutex);
        changed.wait(lock, [this] {
          return stopping || !deny_queue.empty() || !queue.empty();
        });
        if (!deny_queue.empty()) {
          denies.assign(std::make_move_iterator(deny_queue.begin()),
                        std::make_move_iterator(deny_queue.end()));
          deny_queue.clear();
        } else if (stopping) {
          return;
        } else {
          job = std::move(queue.front());
          queue.pop_front();
          have_job = true;
        }
      }
      if (!denies.empty()) {
        process_denies(denies);
        continue;
      }
      if (!have_job) continue;
      bool inherited_deny = false;
      if (job.read_fallback) {
        const int fallback_lock = acquire_key_lock(job.read_fallback->key);
        if (fallback_lock < 0) {
          finish_job(job.request.key, job.source.pixels.size());
          continue;
        }
        inherited_deny = locally_denied(job.read_fallback->key) ||
            denied(job.read_fallback->key);
        flock(fallback_lock, LOCK_UN);
        close(fallback_lock);
      }
      int key_lock = acquire_key_lock(job.request.key);
      if (key_lock < 0) {
        finish_job(job.request.key, job.source.pixels.size());
        continue;
      }

      bool complete = locally_denied(job.request.key) ||
          denied(job.request.key);
      if (!complete && inherited_deny) {
        propagate_deny(job.request);
        complete = true;
      }
      if (!complete) complete = load_disk_entry(job.request, false);

      /* A method-4 miss can reuse the deployed method-0 object while the new
       * variant warms. Do this read under its own key lock and never pass the
       * fallback policy to the encoder. Release the primary lock first so
       * two cache workers cannot deadlock on different lock buckets. */
      if (!complete && job.read_fallback) {
        flock(key_lock, LOCK_UN);
        close(key_lock);
        key_lock = -1;
        bool fallback_denied = false;
        const int fallback_lock = acquire_key_lock(job.read_fallback->key);
        if (fallback_lock >= 0) {
          fallback_denied = locally_denied(job.read_fallback->key) ||
              denied(job.read_fallback->key);
          if (!fallback_denied)
            load_disk_entry(*job.read_fallback, true);
          flock(fallback_lock, LOCK_UN);
          close(fallback_lock);
        }
        key_lock = acquire_key_lock(job.request.key);
        if (key_lock < 0) {
          finish_job(job.request.key, job.source.pixels.size());
          continue;
        }
        if (fallback_denied) {
          /* A deny classifies the source as dynamic, independent of encoder
           * effort. Carry a deployed method-0 deny forward so method 4 cannot
           * resurrect that texture after an upgrade. */
          if (!locally_denied(job.request.key) && !denied(job.request.key))
            propagate_deny(job.request);
          complete = true;
        } else {
          complete = locally_denied(job.request.key) ||
              denied(job.request.key) || load_disk_entry(job.request, false);
        }
      }

      int slot = -1;
      if (!complete) {
        slot = acquire_slot();
        if (slot < 0) {
          flock(key_lock, LOCK_UN);
          close(key_lock);
          finish_job(job.request.key, job.source.pixels.size());
          continue;
        }
        // Recheck after waiting for a host slot. Another process can publish
        // or deny the key while this worker is waiting.
        if (locally_denied(job.request.key) || denied(job.request.key)) {
          complete = true;
        } else {
          complete = load_disk_entry(job.request, false);
        }
      }

      if (!complete) {
        const auto started = std::chrono::steady_clock::now();
        auto rgba = WebpCache::decode_source_rgba(job.source);
        auto encoded = rgba ? WebpCache::encode_rgba(
            *rgba, job.source.width, job.source.height,
            job.request.policy) : std::nullopt;
        const auto elapsed = std::chrono::duration_cast<
            std::chrono::microseconds>(std::chrono::steady_clock::now() -
                                      started).count();
        stat_encode_microseconds += static_cast<uint64_t>(
            std::max<int64_t>(0, elapsed));
        if (encoded && !locally_denied(job.request.key) &&
            !denied(job.request.key) && write_entry(job.request, *encoded)) {
          const size_t bytes = encoded->size();
          publish_memory(job.request.key, std::move(*encoded));
          ++stat_encoded;
          stat_source_bytes += job.source.pixels.size();
          stat_encoded_bytes += bytes;
        } else if (!locally_denied(job.request.key) &&
                   !denied(job.request.key)) {
          ++stat_encode_errors;
        }
      }
      if (slot >= 0) {
        flock(slot, LOCK_UN);
        close(slot);
      }
      flock(key_lock, LOCK_UN);
      close(key_lock);
      changed.notify_all();
      finish_job(job.request.key, job.source.pixels.size());
    }
  }

  bool fsync_cache_directory() const {
    const int fd = open(cache_dir.c_str(), O_RDONLY | O_CLOEXEC);
    if (fd < 0) return false;
    const bool synced = fsync(fd) == 0;
    close(fd);
    return synced;
  }

  bool read_usage_locked(DiskUsage *usage, uint32_t *state) const {
    if (!usage || !state) return false;
    const int fd = open(usage_path().c_str(), O_RDONLY | O_CLOEXEC |
                        O_NOFOLLOW);
    if (fd < 0) return false;
    struct stat file_state{};
    UsageLedger ledger{};
    const bool read_ok = fstat(fd, &file_state) == 0 &&
        S_ISREG(file_state.st_mode) &&
        file_state.st_size == static_cast<off_t>(sizeof(ledger)) &&
        read_all(fd, &ledger, sizeof(ledger));
    close(fd);
    if (!read_ok ||
        std::memcmp(ledger.magic, kUsageMagic, sizeof(kUsageMagic)) != 0 ||
        GUINT16_FROM_LE(ledger.version) != kUsageVersion ||
        GUINT16_FROM_LE(ledger.header_size) != sizeof(ledger) ||
        GUINT64_FROM_LE(ledger.checksum) != usage_checksum(ledger))
      return false;
    const uint32_t decoded_state = GUINT32_FROM_LE(ledger.state);
    if (decoded_state != kUsageClean && decoded_state != kUsageDirty)
      return false;
    DiskUsage decoded;
    decoded.generation = GUINT64_FROM_LE(ledger.generation);
    decoded.webp_entries = GUINT64_FROM_LE(ledger.webp_entries);
    decoded.webp_bytes = GUINT64_FROM_LE(ledger.webp_bytes);
    decoded.deny_entries = GUINT64_FROM_LE(ledger.deny_entries);
    decoded.deny_bytes = GUINT64_FROM_LE(ledger.deny_bytes);
    if (decoded.generation == UINT64_MAX ||
        decoded.webp_entries > UINT64_MAX - decoded.deny_entries ||
        decoded.webp_bytes > UINT64_MAX - decoded.deny_bytes)
      return false;
    *usage = decoded;
    *state = decoded_state;
    return true;
  }

  bool write_usage_locked(const DiskUsage &usage, uint32_t state,
                          uint64_t generation) const {
    UsageLedger ledger{};
    std::memcpy(ledger.magic, kUsageMagic, sizeof(kUsageMagic));
    ledger.version = GUINT16_TO_LE(kUsageVersion);
    ledger.header_size = GUINT16_TO_LE(sizeof(ledger));
    ledger.state = GUINT32_TO_LE(state);
    ledger.generation = GUINT64_TO_LE(generation);
    ledger.webp_entries = GUINT64_TO_LE(usage.webp_entries);
    ledger.webp_bytes = GUINT64_TO_LE(usage.webp_bytes);
    ledger.deny_entries = GUINT64_TO_LE(usage.deny_entries);
    ledger.deny_bytes = GUINT64_TO_LE(usage.deny_bytes);
    ledger.checksum = GUINT64_TO_LE(usage_checksum(ledger));
    const int fd = open(usage_path().c_str(), O_WRONLY | O_CREAT | O_CLOEXEC |
                        O_NOFOLLOW, 0644);
    if (fd < 0) return false;
    const bool written = ftruncate(fd, sizeof(ledger)) == 0 &&
        lseek(fd, 0, SEEK_SET) == 0 &&
        write_all(fd, &ledger, sizeof(ledger)) && fsync(fd) == 0;
    close(fd);
    return written;
  }

  void publish_usage_stats(const DiskUsage &usage) {
    stat_webp_entries.store(usage.webp_entries);
    stat_webp_bytes.store(usage.webp_bytes);
    stat_deny_entries.store(usage.deny_entries);
    stat_deny_bytes.store(usage.deny_bytes);
  }

  bool scan_and_prune_locked(uint64_t reserve_bytes,
                             uint64_t reserve_entries,
                             uint64_t previous_generation,
                             DiskUsage *result) {
    if (!result || reserve_bytes > cache_byte_limit ||
        reserve_entries > cache_entry_limit)
      return false;
    const uint64_t generation = previous_generation + 1;
    DiskUsage dirty;
    dirty.generation = previous_generation;
    if (!write_usage_locked(dirty, kUsageDirty, generation)) return false;
    ++stat_disk_scans;
    enum class Kind { kWebp, kDeny };
    struct Entry {
      std::string path;
      off_t size;
      timespec modified;
      Kind kind;
    };
    struct TempEntry {
      std::string path;
      off_t size;
      timespec modified;
    };
    std::vector<Entry> entries;
    std::vector<TempEntry> temporary;
    uint64_t total = 0;
    uint64_t webp_entries = 0;
    uint64_t webp_bytes = 0;
    uint64_t deny_entries = 0;
    uint64_t deny_bytes = 0;
    uint64_t stale_removals = 0;
    uint64_t stale_bytes = 0;
    uint64_t temp_entries = 0;
    uint64_t temp_bytes = 0;
    uint64_t temp_evictions = 0;
    uint64_t temp_evicted_bytes = 0;
    bool directory_changed = false;
    const time_t now = std::time(nullptr);
    DIR *directory = opendir(cache_dir.c_str());
    if (!directory) return false;
    while (dirent *item = readdir(directory)) {
      const std::string path = cache_dir + "/" + item->d_name;
      struct stat state{};
      if (lstat(path.c_str(), &state) != 0 || !S_ISREG(state.st_mode))
        continue;
#if defined(__APPLE__)
      const timespec modified = state.st_mtimespec;
#else
      const timespec modified = state.st_mtim;
#endif
      if (temp_name(item->d_name)) {
        const bool stale = now >= modified.tv_sec &&
            now - modified.tv_sec >= kStaleTempAgeSeconds;
        if (unlink(path.c_str()) == 0) {
          if (stale) {
            ++stale_removals;
            stale_bytes += static_cast<uint64_t>(state.st_size);
          } else {
            ++temp_evictions;
            temp_evicted_bytes += static_cast<uint64_t>(state.st_size);
          }
          directory_changed = true;
        } else {
          ++temp_entries;
          temp_bytes += static_cast<uint64_t>(state.st_size);
          temporary.push_back({path, state.st_size, modified});
        }
        continue;
      }
      Kind kind;
      if (key_name(item->d_name, ".webp")) {
        kind = Kind::kWebp;
        ++webp_entries;
        webp_bytes += static_cast<uint64_t>(state.st_size);
      } else if (key_name(item->d_name, ".deny")) {
        kind = Kind::kDeny;
        ++deny_entries;
        deny_bytes += static_cast<uint64_t>(state.st_size);
      } else {
        continue;
      }
      total += static_cast<uint64_t>(state.st_size);
      entries.push_back({path, state.st_size, modified, kind});
    }
    closedir(directory);
    if (temp_entries > temp_entry_limit || temp_bytes > temp_byte_limit) {
      std::sort(temporary.begin(), temporary.end(),
                [](const TempEntry &left, const TempEntry &right) {
        if (left.modified.tv_sec != right.modified.tv_sec)
          return left.modified.tv_sec < right.modified.tv_sec;
        return left.modified.tv_nsec < right.modified.tv_nsec;
      });
      for (const TempEntry &entry : temporary) {
        if (temp_entries <= temp_entry_limit &&
            temp_bytes <= temp_byte_limit)
          break;
        if (unlink(entry.path.c_str()) != 0) continue;
        const uint64_t removed = static_cast<uint64_t>(entry.size);
        --temp_entries;
        temp_bytes -= std::min(temp_bytes, removed);
        ++temp_evictions;
        temp_evicted_bytes += removed;
        directory_changed = true;
      }
    }
    // Every cooperative publication holds .prune.lock from before creating a
    // temporary file until after its rename. Any temporary file visible while
    // this lock is held is abandoned. Never mark the ledger clean while such
    // an unaccounted file remains.
    if (temp_entries != 0) return false;
    uint64_t permanent_entries = entries.size();
    uint64_t disk_total = total + temp_bytes;
    if (exceeds_limit(disk_total, reserve_bytes, cache_byte_limit) ||
        exceeds_limit(permanent_entries, reserve_entries,
                      cache_entry_limit)) {
      std::sort(entries.begin(), entries.end(),
                [](const Entry &left, const Entry &right) {
        if (left.modified.tv_sec != right.modified.tv_sec)
          return left.modified.tv_sec < right.modified.tv_sec;
        return left.modified.tv_nsec < right.modified.tv_nsec;
      });
      const uint64_t target = std::max<uint64_t>(
          cache_byte_limit * 9 / 10, reserve_bytes);
      const uint64_t entry_target = std::max<uint64_t>(
          cache_entry_limit * 9 / 10, reserve_entries);
      for (const Entry &entry : entries) {
        if (!exceeds_limit(disk_total, reserve_bytes, target) &&
            !exceeds_limit(permanent_entries, reserve_entries,
                           entry_target))
          break;
        if (unlink(entry.path.c_str()) != 0) continue;
        const uint64_t removed = static_cast<uint64_t>(entry.size);
        total -= std::min(total, removed);
        disk_total -= std::min(disk_total, removed);
        --permanent_entries;
        if (entry.kind == Kind::kWebp) {
          --webp_entries;
          webp_bytes -= std::min(webp_bytes, removed);
        } else {
          --deny_entries;
          deny_bytes -= std::min(deny_bytes, removed);
        }
        ++stat_disk_evictions;
        stat_disk_evicted_bytes += removed;
        directory_changed = true;
      }
    }
    if (directory_changed && !fsync_cache_directory()) return false;
    DiskUsage usage;
    usage.generation = generation;
    usage.webp_entries = webp_entries;
    usage.webp_bytes = webp_bytes;
    usage.deny_entries = deny_entries;
    usage.deny_bytes = deny_bytes;
    if (!write_usage_locked(usage, kUsageClean, generation)) return false;
    publish_usage_stats(usage);
    stat_temp_entries.store(temp_entries);
    stat_temp_bytes.store(temp_bytes);
    stat_temp_evictions += temp_evictions;
    stat_temp_evicted_bytes += temp_evicted_bytes;
    stat_stale_temp_removals += stale_removals;
    stat_stale_temp_bytes += stale_bytes;
    *result = usage;
    return true;
  }

  bool ensure_usage_locked(DiskUsage *usage) {
    uint32_t state = 0;
    DiskUsage decoded;
    if (read_usage_locked(&decoded, &state) && state == kUsageClean &&
        usage_bytes(decoded) <= cache_byte_limit &&
        usage_entries(decoded) <= cache_entry_limit) {
      publish_usage_stats(decoded);
      *usage = decoded;
      return true;
    }
    const uint64_t generation = decoded.generation;
    return scan_and_prune_locked(0, 0, generation, usage);
  }

  void initialize_usage() {
    const int lock_fd = acquire_lock(prune_lock_path());
    if (lock_fd < 0) return;
    DiskUsage usage;
    ensure_usage_locked(&usage);
    flock(lock_fd, LOCK_UN);
    close(lock_fd);
  }

  void remove_corrupt_entry(const std::string &path) {
    const int lock_fd = acquire_lock(prune_lock_path());
    if (lock_fd < 0) return;
    DiskUsage usage;
    if (!ensure_usage_locked(&usage)) {
      flock(lock_fd, LOCK_UN);
      close(lock_fd);
      return;
    }
    struct stat state{};
    if (lstat(path.c_str(), &state) != 0 || !S_ISREG(state.st_mode)) {
      flock(lock_fd, LOCK_UN);
      close(lock_fd);
      return;
    }
    // The file can have been truncated, so its current size cannot safely be
    // subtracted from the previous ledger total. Leave a durable dirty ledger;
    // the encode path (or the next process) repairs it with one source-of-truth
    // directory scan before any new publication.
    const uint64_t generation = usage.generation + 1;
    if (write_usage_locked(usage, kUsageDirty, generation) &&
        unlink(path.c_str()) == 0)
      fsync_cache_directory();
    flock(lock_fd, LOCK_UN);
    close(lock_fd);
  }

  std::string cache_dir;
  unsigned host_slots;
  size_t queue_byte_limit;
  size_t cache_byte_limit;
  size_t cache_entry_limit;
  size_t temp_entry_limit;
  size_t temp_byte_limit;
  size_t memory_byte_limit;
  size_t deny_queue_limit;
  bool enabled_flag = false;
  bool stopping = false;
  std::thread worker;
  std::mutex mutex;
  std::condition_variable changed;
  std::deque<Job> queue;
  std::unordered_set<std::string> inflight;
  std::deque<std::string> deny_queue;
  std::unordered_set<std::string> deny_pending;
  size_t queue_bytes = 0;
  std::mutex memory_mutex;
  std::unordered_map<std::string, MemoryEntry> memory;
  std::list<std::string> memory_lru;
  std::unordered_set<std::string> memory_denied;
  size_t memory_bytes = 0;
  std::atomic<uint64_t> temp_serial{1};
  std::atomic<uint64_t> stat_hits{0};
  std::atomic<uint64_t> stat_misses{0};
  std::atomic<uint64_t> stat_denied{0};
  std::atomic<uint64_t> stat_deny_queued{0};
  std::atomic<uint64_t> stat_deny_queue_drops{0};
  std::atomic<uint64_t> stat_deny_write_errors{0};
  std::atomic<uint64_t> stat_queued{0};
  std::atomic<uint64_t> stat_queue_drops{0};
  std::atomic<uint64_t> stat_encoded{0};
  std::atomic<uint64_t> stat_encode_errors{0};
  std::atomic<uint64_t> stat_corrupt_entries{0};
  std::atomic<uint64_t> stat_source_bytes{0};
  std::atomic<uint64_t> stat_encoded_bytes{0};
  std::atomic<uint64_t> stat_encode_microseconds{0};
  std::atomic<uint64_t> stat_key_bytes{0};
  std::atomic<uint64_t> stat_key_microseconds{0};
  std::atomic<uint64_t> stat_disk_loads{0};
  std::atomic<uint64_t> stat_disk_loaded_bytes{0};
  std::atomic<uint64_t> stat_fallback_disk_loads{0};
  std::atomic<uint64_t> stat_fallback_disk_loaded_bytes{0};
  std::atomic<uint64_t> stat_disk_scans{0};
  std::atomic<uint64_t> stat_memory_entries{0};
  std::atomic<uint64_t> stat_memory_bytes{0};
  std::atomic<uint64_t> stat_memory_busy_misses{0};
  std::atomic<uint64_t> stat_memory_evictions{0};
  std::atomic<uint64_t> stat_memory_evicted_bytes{0};
  std::atomic<uint64_t> stat_webp_entries{0};
  std::atomic<uint64_t> stat_webp_bytes{0};
  std::atomic<uint64_t> stat_deny_entries{0};
  std::atomic<uint64_t> stat_deny_bytes{0};
  std::atomic<uint64_t> stat_disk_evictions{0};
  std::atomic<uint64_t> stat_disk_evicted_bytes{0};
  std::atomic<uint64_t> stat_temp_entries{0};
  std::atomic<uint64_t> stat_temp_bytes{0};
  std::atomic<uint64_t> stat_temp_evictions{0};
  std::atomic<uint64_t> stat_temp_evicted_bytes{0};
  std::atomic<uint64_t> stat_stale_temp_removals{0};
  std::atomic<uint64_t> stat_stale_temp_bytes{0};
};

WebpCache::WebpCache(std::string cache_dir, unsigned host_slots,
                     size_t queue_byte_limit, size_t cache_byte_limit)
    : impl_(std::make_unique<Impl>(std::move(cache_dir), host_slots,
                                  queue_byte_limit, cache_byte_limit)) {}

WebpCache::~WebpCache() = default;

bool WebpCache::enabled() const { return impl_->enabled_flag; }

WebpPreparedSource WebpCache::prepare_source(const WebpSource &source) const {
  const auto started = std::chrono::steady_clock::now();
  WebpPreparedSource prepared;
  prepared.source_digest = sha256(source.pixels.data(), source.pixels.size());
  prepared.encoder_version = static_cast<uint32_t>(WebPGetEncoderVersion());
  prepared.width = source.width;
  prepared.height = source.height;
  prepared.pitch = source.pitch;
  prepared.format = source.format;
  prepared.decoded_size = static_cast<uint32_t>(source.pixels.size());
  const auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(
      std::chrono::steady_clock::now() - started).count();
  impl_->stat_key_bytes += source.pixels.size();
  impl_->stat_key_microseconds += static_cast<uint64_t>(
      std::max<int64_t>(0, elapsed));
  return prepared;
}

WebpRequest WebpCache::prepare(const WebpPreparedSource &source,
                               const WebpPolicy &policy) const {
  const auto started = std::chrono::steady_clock::now();
  WebpRequest request;
  request.encoder_version = source.encoder_version;
  request.policy = policy;
  request.width = source.width;
  request.height = source.height;
  request.pitch = source.pitch;
  request.format = source.format;
  request.decoded_size = source.decoded_size;
  WebpBytes canonical;
  /* This deployed domain is immutable. The method byte below already gives
   * method 0 and method 4 separate keys. Renaming the domain would make the
   * existing method-0 disk cache undiscoverable during migration. */
  static constexpr char domain[] = "W3CS-WEBP-M0-CACHE-V2";
  canonical.insert(canonical.end(), domain, domain + sizeof(domain));
  append_u32(canonical, request.encoder_version);
  append_u32(canonical, source.width);
  append_u32(canonical, source.height);
  append_u32(canonical, source.pitch);
  append_u32(canonical, source.format);
  append_u32(canonical, request.decoded_size);
  canonical.push_back(policy.quality);
  canonical.push_back(policy.method);
  canonical.push_back(1);  // exact transparent RGB
  canonical.push_back(100);  // lossless alpha quality
  canonical.insert(canonical.end(), source.source_digest.begin(),
                   source.source_digest.end());
  request.digest = sha256(canonical.data(), canonical.size());
  request.key = hex_digest(request.digest);
  const auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(
      std::chrono::steady_clock::now() - started).count();
  impl_->stat_key_microseconds += static_cast<uint64_t>(
      std::max<int64_t>(0, elapsed));
  return request;
}

WebpRequest WebpCache::prepare(const WebpSource &source,
                               const WebpPolicy &policy) const {
  return prepare(prepare_source(source), policy);
}

std::optional<WebpBytes> WebpCache::lookup(const WebpRequest &request) {
  if (!enabled() || !valid_key(request.key)) return std::nullopt;
  return impl_->lookup_memory(request.key);
}

bool WebpCache::submit(const WebpRequest &request, WebpSource source,
                       std::optional<WebpRequest> read_fallback) {
  if (!enabled() || !valid_key(request.key) || !supported_source(source) ||
      impl_->try_locally_denied(request.key))
    return false;
  if (read_fallback &&
      (!valid_key(read_fallback->key) ||
       read_fallback->key == request.key ||
       read_fallback->encoder_version != request.encoder_version ||
       read_fallback->width != request.width ||
       read_fallback->height != request.height ||
       read_fallback->pitch != request.pitch ||
       read_fallback->format != request.format ||
       read_fallback->decoded_size != request.decoded_size ||
       read_fallback->policy.quality != request.policy.quality))
    return false;
  std::lock_guard lock(impl_->mutex);
  if (impl_->stopping) return false;
  if (impl_->inflight.count(request.key)) return true;
  if (source.pixels.size() > impl_->queue_byte_limit -
      std::min(impl_->queue_bytes, impl_->queue_byte_limit)) {
    ++impl_->stat_queue_drops;
    return false;
  }
  impl_->queue_bytes += source.pixels.size();
  impl_->inflight.insert(request.key);
  impl_->queue.push_back(
      {request, std::move(source), std::move(read_fallback)});
  ++impl_->stat_queued;
  impl_->changed.notify_one();
  return true;
}

void WebpCache::deny(const WebpRequest &request) {
  if (!enabled() || !valid_key(request.key)) return;
  bool queued = false;
  {
    std::unique_lock memory_lock(impl_->memory_mutex);
    impl_->remove_memory_locked(request.key);
    std::lock_guard queue_lock(impl_->mutex);
    if (impl_->deny_pending.count(request.key)) {
      impl_->memory_denied.insert(request.key);
    } else if (!impl_->stopping &&
               impl_->deny_pending.size() < impl_->deny_queue_limit) {
      impl_->memory_denied.insert(request.key);
      impl_->deny_pending.insert(request.key);
      impl_->deny_queue.push_back(request.key);
      ++impl_->stat_deny_queued;
      queued = true;
    } else {
      ++impl_->stat_deny_queue_drops;
    }
  }
  ++impl_->stat_denied;
  if (queued) impl_->changed.notify_one();
}

WebpCacheStats WebpCache::stats() const {
  WebpCacheStats result;
  result.hits = impl_->stat_hits.load();
  result.misses = impl_->stat_misses.load();
  result.denied = impl_->stat_denied.load();
  result.deny_queued = impl_->stat_deny_queued.load();
  result.deny_queue_drops = impl_->stat_deny_queue_drops.load();
  result.deny_write_errors = impl_->stat_deny_write_errors.load();
  {
    std::lock_guard lock(impl_->mutex);
    result.deny_pending = impl_->deny_pending.size();
    result.deny_queue_limit = impl_->deny_queue_limit;
    result.pending_jobs = impl_->inflight.size();
    result.pending_bytes = impl_->queue_bytes;
    result.queue_limit_bytes = impl_->queue_byte_limit;
  }
  result.queued = impl_->stat_queued.load();
  result.queue_drops = impl_->stat_queue_drops.load();
  result.encoded = impl_->stat_encoded.load();
  result.encode_errors = impl_->stat_encode_errors.load();
  result.corrupt_entries = impl_->stat_corrupt_entries.load();
  result.source_bytes = impl_->stat_source_bytes.load();
  result.encoded_bytes = impl_->stat_encoded_bytes.load();
  result.encode_microseconds = impl_->stat_encode_microseconds.load();
  result.key_bytes = impl_->stat_key_bytes.load();
  result.key_microseconds = impl_->stat_key_microseconds.load();
  result.disk_loads = impl_->stat_disk_loads.load();
  result.disk_loaded_bytes = impl_->stat_disk_loaded_bytes.load();
  result.fallback_disk_loads = impl_->stat_fallback_disk_loads.load();
  result.fallback_disk_loaded_bytes =
      impl_->stat_fallback_disk_loaded_bytes.load();
  result.disk_scans = impl_->stat_disk_scans.load();
  result.memory_entries = impl_->stat_memory_entries.load();
  result.memory_bytes = impl_->stat_memory_bytes.load();
  result.memory_limit_bytes = impl_->memory_byte_limit;
  result.memory_busy_misses = impl_->stat_memory_busy_misses.load();
  result.memory_evictions = impl_->stat_memory_evictions.load();
  result.memory_evicted_bytes = impl_->stat_memory_evicted_bytes.load();
  result.webp_entries = impl_->stat_webp_entries.load();
  result.webp_bytes = impl_->stat_webp_bytes.load();
  result.deny_entries = impl_->stat_deny_entries.load();
  result.deny_bytes = impl_->stat_deny_bytes.load();
  result.disk_limit_bytes = impl_->cache_byte_limit;
  result.disk_entry_limit = impl_->cache_entry_limit;
  result.disk_evictions = impl_->stat_disk_evictions.load();
  result.disk_evicted_bytes = impl_->stat_disk_evicted_bytes.load();
  result.temp_entries = impl_->stat_temp_entries.load();
  result.temp_bytes = impl_->stat_temp_bytes.load();
  result.temp_entry_limit = impl_->temp_entry_limit;
  result.temp_byte_limit = impl_->temp_byte_limit;
  result.temp_evictions = impl_->stat_temp_evictions.load();
  result.temp_evicted_bytes = impl_->stat_temp_evicted_bytes.load();
  result.stale_temp_removals = impl_->stat_stale_temp_removals.load();
  result.stale_temp_bytes = impl_->stat_stale_temp_bytes.load();
  return result;
}

bool WebpCache::supported_source(const WebpSource &source) {
  if (!source.width || !source.height || source.width > 16384 ||
      source.height > 16384)
    return false;
  const bool four = source.format == kFormatA8R8G8B8 ||
      source.format == kFormatX8R8G8B8;
  const bool two = source.format == kFormatR5G6B5 ||
      source.format == kFormatX1R5G5B5 ||
      source.format == kFormatA1R5G5B5 ||
      source.format == kFormatA4R4G4B4;
  if (!four && !two) return false;
  const uint64_t pitch = static_cast<uint64_t>(source.width) * (four ? 4 : 2);
  const uint64_t size = pitch * source.height;
  return pitch == source.pitch && size == source.pixels.size() &&
      size > 0 && size <= kMaximumImageBytes;
}

std::optional<WebpBytes> WebpCache::decode_source_rgba(
    const WebpSource &source) {
  if (!supported_source(source)) return std::nullopt;
  WebpBytes rgba(static_cast<size_t>(source.width) * source.height * 4);
  const size_t pixels = static_cast<size_t>(source.width) * source.height;
  if (source.format == kFormatA8R8G8B8 ||
      source.format == kFormatX8R8G8B8) {
    const bool opaque = source.format == kFormatX8R8G8B8;
    for (size_t pixel = 0; pixel < pixels; ++pixel) {
      const size_t offset = pixel * 4;
      rgba[offset] = source.pixels[offset + 2];
      rgba[offset + 1] = source.pixels[offset + 1];
      rgba[offset + 2] = source.pixels[offset];
      rgba[offset + 3] = opaque ? 255 : source.pixels[offset + 3];
    }
    return rgba;
  }
  for (size_t pixel = 0; pixel < pixels; ++pixel) {
    uint16_t value = 0;
    std::memcpy(&value, source.pixels.data() + pixel * 2, sizeof(value));
    value = GUINT16_FROM_LE(value);
    uint8_t red = 0;
    uint8_t green = 0;
    uint8_t blue = 0;
    uint8_t alpha = 255;
    if (source.format == kFormatR5G6B5) {
      red = expand5((value >> 11) & 31);
      green = expand6((value >> 5) & 63);
      blue = expand5(value & 31);
    } else if (source.format == kFormatA4R4G4B4) {
      red = static_cast<uint8_t>(((value >> 8) & 15) * 17);
      green = static_cast<uint8_t>(((value >> 4) & 15) * 17);
      blue = static_cast<uint8_t>((value & 15) * 17);
      alpha = static_cast<uint8_t>(((value >> 12) & 15) * 17);
    } else {
      red = expand5((value >> 10) & 31);
      green = expand5((value >> 5) & 31);
      blue = expand5(value & 31);
      alpha = source.format == kFormatX1R5G5B5 || (value & 0x8000)
          ? 255 : 0;
    }
    const size_t offset = pixel * 4;
    rgba[offset] = red;
    rgba[offset + 1] = green;
    rgba[offset + 2] = blue;
    rgba[offset + 3] = alpha;
  }
  return rgba;
}

std::optional<WebpBytes> WebpCache::encode_rgba(
    const WebpBytes &rgba, uint32_t width, uint32_t height,
    const WebpPolicy &policy) {
  if (!width || !height || policy.quality > 100 || policy.method > 6 ||
      rgba.size() != static_cast<size_t>(width) * height * 4)
    return std::nullopt;
  WebPConfig config;
  if (!WebPConfigInit(&config)) return std::nullopt;
  config.lossless = 0;
  config.quality = policy.quality;
  config.method = policy.method;
  config.alpha_quality = 100;
  config.exact = 1;
  config.thread_level = 0;
  if (!WebPValidateConfig(&config)) return std::nullopt;
  WebPPicture picture;
  if (!WebPPictureInit(&picture)) return std::nullopt;
  picture.width = static_cast<int>(width);
  picture.height = static_cast<int>(height);
  picture.use_argb = 1;
  WebPMemoryWriter writer;
  WebPMemoryWriterInit(&writer);
  picture.writer = WebPMemoryWrite;
  picture.custom_ptr = &writer;
  const bool imported = WebPPictureImportRGBA(
      &picture, rgba.data(), static_cast<int>(width * 4));
  const bool encoded_ok = imported && WebPEncode(&config, &picture);
  WebpBytes output;
  if (encoded_ok && writer.mem && writer.size)
    output.assign(writer.mem, writer.mem + writer.size);
  WebPPictureFree(&picture);
  WebPMemoryWriterClear(&writer);
  if (output.empty()) return std::nullopt;
  return output;
}

}  // namespace w3cs