/* vim:set ts=2 sw=2 sts=2 et: */
/**
 * \author     Marcus Holland-Moritz (github@mhxnet.de)
 * \copyright  Copyright (c) Marcus Holland-Moritz
 *
 * This file is part of dwarfs.
 *
 * dwarfs is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * dwarfs is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with dwarfs.  If not, see <https://www.gnu.org/licenses/>.
 */

#include <algorithm>
#include <cassert>
#include <cerrno>
#include <climits>
#include <cstring>
#include <ctime>
#include <filesystem>
#include <numeric>
#include <ostream>

#include <boost/algorithm/string.hpp>

#include <thrift/lib/cpp2/frozen/FrozenUtil.h>
#include <thrift/lib/cpp2/protocol/DebugProtocol.h>
#include <thrift/lib/cpp2/protocol/Serializer.h>

#include <fmt/chrono.h>
#include <fmt/format.h>

#include <folly/container/Enumerate.h>
#include <folly/container/F14Set.h>
#include <folly/portability/Stdlib.h>
#include <folly/portability/Unistd.h>
#include <folly/stats/Histogram.h>

#include <fsst.h>

#include "dwarfs/error.h"
#include "dwarfs/features.h"
#include "dwarfs/file_stat.h"
#include "dwarfs/fstypes.h"
#include "dwarfs/logger.h"
#include "dwarfs/metadata_v2.h"
#include "dwarfs/options.h"
#include "dwarfs/string_table.h"
#include "dwarfs/util.h"
#include "dwarfs/vfs_stat.h"

#include "dwarfs/gen-cpp2/metadata_layouts.h"
#include "dwarfs/gen-cpp2/metadata_types_custom_protocol.h"

#include "thrift/lib/thrift/gen-cpp2/frozen_types_custom_protocol.h"

namespace dwarfs {

namespace fs = std::filesystem;

namespace {

using ::apache::thrift::frozen::MappedFrozen;

template <class T>
std::pair<std::vector<uint8_t>, std::vector<uint8_t>>
freeze_to_buffer(const T& x) {
  using namespace ::apache::thrift::frozen;

  Layout<T> layout;
  size_t content_size = LayoutRoot::layout(x, layout);

  std::string schema;
  serializeRootLayout(layout, schema);

  size_t schema_size = schema.size();
  auto schema_begin = reinterpret_cast<uint8_t const*>(schema.data());
  std::vector<uint8_t> schema_buffer(schema_begin, schema_begin + schema_size);

  std::vector<uint8_t> data_buffer;
  data_buffer.resize(content_size, 0);

  folly::MutableByteRange content_range(data_buffer.data(), data_buffer.size());
  ByteRangeFreezer::freeze(layout, x, content_range);

  data_buffer.resize(data_buffer.size() - content_range.size());

  return {schema_buffer, data_buffer};
}

void check_schema(std::span<uint8_t const> data) {
  using namespace ::apache::thrift;
  frozen::schema::Schema schema;
  size_t schemaSize = CompactSerializer::deserialize(data, schema);
  // std::cerr << debugString(schema) << '\n';
  if (schemaSize != data.size()) {
    DWARFS_THROW(runtime_error, "invalid schema size");
  }
  if (schema.layouts()->count(*schema.rootLayout()) == 0) {
    DWARFS_THROW(runtime_error, "invalid rootLayout in schema");
  }
  for (auto const& kvl : *schema.layouts()) {
    auto const& layout = kvl.second;
    if (kvl.first >= static_cast<int64_t>(schema.layouts()->size())) {
      DWARFS_THROW(runtime_error, "invalid layout key in schema");
    }
    if (*layout.size() < 0) {
      DWARFS_THROW(runtime_error, "negative size in schema");
    }
    if (*layout.bits() < 0) {
      DWARFS_THROW(runtime_error, "negative bits in schema");
    }
    for (auto const& kvf : *layout.fields()) {
      auto const& field = kvf.second;
      if (schema.layouts()->count(*field.layoutId()) == 0) {
        DWARFS_THROW(runtime_error, "invalid layoutId in field");
      }
    }
  }
}

template <typename T>
MappedFrozen<T>
map_frozen(std::span<uint8_t const> schema, std::span<uint8_t const> data) {
  using namespace ::apache::thrift::frozen;
  check_schema(schema);
  auto layout = std::make_unique<Layout<T>>();
  folly::ByteRange tmp(schema.data(), schema.size());
  deserializeRootLayout(tmp, *layout);
  MappedFrozen<T> ret(layout->view({data.data(), 0}));
  ret.hold(std::move(layout));
  return ret;
}

MappedFrozen<thrift::metadata::metadata>
check_frozen(MappedFrozen<thrift::metadata::metadata> meta) {
  if (meta.features()) {
    auto unsupported = feature_set::get_unsupported(meta.features()->thaw());
    if (!unsupported.empty()) {
      DWARFS_THROW(runtime_error,
                   fmt::format("file system uses the following features "
                               "unsupported by this build: {}",
                               boost::join(unsupported, ", ")));
    }
  }
  return meta;
}

global_metadata::Meta const&
check_metadata_consistency(logger& lgr, global_metadata::Meta const& meta,
                           bool force_consistency_check) {
  if (force_consistency_check) {
    global_metadata::check_consistency(lgr, meta);
  }
  return meta;
}

void analyze_frozen(std::ostream& os,
                    MappedFrozen<thrift::metadata::metadata> const& meta,
                    size_t total_size, int detail) {
  using namespace ::apache::thrift::frozen;
  std::ostringstream oss;
  stream_logger lgr(oss);

  auto layout = meta.findFirstOfType<
      std::unique_ptr<Layout<thrift::metadata::metadata>>>();

  auto& l = *layout;
  std::vector<std::pair<size_t, std::string>> usage;

#if FMT_VERSION >= 70000
#define DWARFS_FMT_L "L"
#else
#define DWARFS_FMT_L "n"
#endif

  auto fmt_size = [&](auto const& name, size_t count, size_t size) {
    return fmt::format("{0:>14" DWARFS_FMT_L "} {1:.<20}{2:.>16" DWARFS_FMT_L
                       "} bytes {3:5.1f}% {4:5.1f} bytes/item\n",
                       count, name, size, 100.0 * size / total_size,
                       count > 0 ? static_cast<double>(size) / count : 0.0);
  };

  auto fmt_detail = [&](auto const& name, size_t count, size_t size,
                        std::string num) {
    return fmt::format(
        "               {0:<20}{1:>16" DWARFS_FMT_L "} bytes {2:>6} "
        "{3:5.1f} bytes/item\n",
        name, size, num, count > 0 ? static_cast<double>(size) / count : 0.0);
  };

  auto fmt_detail_pct = [&](auto const& name, size_t count, size_t size) {
    return fmt_detail(name, count, size,
                      fmt::format("{0:5.1f}%", 100.0 * size / total_size));
  };

  auto add_size = [&](auto const& name, size_t count, size_t size) {
    usage.emplace_back(size, fmt_size(name, count, size));
  };

  auto list_size = [&](auto const& list, auto const& field) {
    return (list.size() * field.layout.itemField.layout.bits + 7) / 8;
  };

  auto add_list_size = [&](auto const& name, auto const& list,
                           auto const& field) {
    add_size(name, list.size(), list_size(list, field));
  };

  auto add_string_list_size = [&](auto const& name, auto const& list,
                                  auto const& field) {
    auto count = list.size();
    if (count > 0) {
      auto index_size = list_size(list, field);
      auto data_size = list.back().end() - list.front().begin();
      auto size = index_size + data_size;
      auto fmt = fmt_size(name, count, size) +
                 fmt_detail_pct("|- data", count, data_size) +
                 fmt_detail_pct("'- index", count, index_size);
      usage.emplace_back(size, fmt);
    }
  };

  auto add_string_table_size = [&](auto const& name, auto const& table,
                                   auto const& field) {
    if (auto data_size = table.buffer().size(); data_size > 0) {
      auto dict_size =
          table.symtab() ? table.symtab()->size() : static_cast<size_t>(0);
      auto index_size = list_size(table.index(), field.layout.indexField);
      auto size = index_size + data_size + dict_size;
      auto count = table.index().size() - (table.packed_index() ? 0 : 1);
      auto fmt = fmt_size(name, count, size) +
                 fmt_detail_pct("|- data", count, data_size);
      if (table.symtab()) {
        string_table st(lgr, "tmp", table);
        auto unpacked_size = st.unpacked_size();
        fmt += fmt_detail(
            "|- unpacked", count, unpacked_size,
            fmt::format("{0:5.2f}x",
                        static_cast<double>(unpacked_size) / data_size));
        fmt += fmt_detail_pct("|- dict", count, dict_size);
      }
      fmt += fmt_detail_pct("'- index", count, index_size);
      usage.emplace_back(size, fmt);
    }
  };

#define META_LIST_SIZE(x) add_list_size(#x, meta.x(), l->x##Field)

#define META_STRING_LIST_SIZE(x) add_string_list_size(#x, meta.x(), l->x##Field)

#define META_OPT_LIST_SIZE(x)                                                  \
  do {                                                                         \
    if (auto list = meta.x()) {                                                \
      add_list_size(#x, *list, l->x##Field.layout.valueField);                 \
    }                                                                          \
  } while (0)

#define META_OPT_STRING_LIST_SIZE(x)                                           \
  do {                                                                         \
    if (auto list = meta.x()) {                                                \
      add_string_list_size(#x, *list, l->x##Field.layout.valueField);          \
    }                                                                          \
  } while (0)

#define META_OPT_STRING_SET_SIZE(x) META_OPT_STRING_LIST_SIZE(x)

#define META_OPT_STRING_TABLE_SIZE(x)                                          \
  do {                                                                         \
    if (auto table = meta.x()) {                                               \
      add_string_table_size(#x, *table, l->x##Field.layout.valueField);        \
    }                                                                          \
  } while (0)

  META_LIST_SIZE(chunks);
  META_LIST_SIZE(directories);
  META_LIST_SIZE(inodes);
  META_LIST_SIZE(chunk_table);
  META_LIST_SIZE(symlink_table);
  META_LIST_SIZE(uids);
  META_LIST_SIZE(gids);
  META_LIST_SIZE(modes);

  META_OPT_LIST_SIZE(devices);
  META_OPT_LIST_SIZE(dir_entries);
  META_OPT_LIST_SIZE(shared_files_table);

  META_OPT_STRING_TABLE_SIZE(compact_names);
  META_OPT_STRING_TABLE_SIZE(compact_symlinks);

  META_STRING_LIST_SIZE(names);
  META_STRING_LIST_SIZE(symlinks);

  META_OPT_STRING_SET_SIZE(features);

  META_OPT_STRING_LIST_SIZE(category_names);
  META_OPT_LIST_SIZE(block_categories);

#undef META_LIST_SIZE
#undef META_OPT_STRING_SET_SIZE
#undef META_OPT_STRING_LIST_SIZE
#undef META_STRING_LIST_SIZE
#undef META_OPT_LIST_SIZE
#undef META_OPT_STRING_TABLE_SIZE

  std::sort(usage.begin(), usage.end(), [](auto const& a, auto const& b) {
    return a.first > b.first || (a.first == b.first && a.second < b.second);
  });

  os << "metadata memory usage:\n";
  os << fmt::format("               {0:.<20}{1:.>16" DWARFS_FMT_L
                    "} bytes       {2:6.1f} bytes/inode\n",
                    "total metadata", total_size,
                    static_cast<double>(total_size) / meta.inodes().size());

#undef DWARFS_FMT_L

  for (auto const& u : usage) {
    os << u.second;
  }

  if (detail > 3) {
    l->print(os, 0);
    os << '\n';
  }
}

template <typename Function>
void parse_metadata_options(
    MappedFrozen<thrift::metadata::metadata> const& meta, Function&& func) {
  if (auto opt = meta.options()) {
    func("mtime_only", opt->mtime_only());
    func("packed_chunk_table", opt->packed_chunk_table());
    func("packed_directories", opt->packed_directories());
    func("packed_shared_files_table", opt->packed_shared_files_table());
  }
  if (auto names = meta.compact_names()) {
    func("packed_names", static_cast<bool>(names->symtab()));
    func("packed_names_index", names->packed_index());
  }
  if (auto symlinks = meta.compact_symlinks()) {
    func("packed_symlinks", static_cast<bool>(symlinks->symtab()));
    func("packed_symlinks_index", symlinks->packed_index());
  }
}

struct category_info {
  size_t count{0};
  size_t compressed_size{0};
  size_t uncompressed_size{0};
  bool uncompressed_size_is_estimate{false};
};

std::map<size_t, category_info>
get_category_info(MappedFrozen<thrift::metadata::metadata> const& meta,
                  filesystem_info const& fsinfo) {
  std::map<size_t, category_info> catinfo;

  if (auto blockcat = meta.block_categories()) {
    for (auto [block, category] : folly::enumerate(blockcat.value())) {
      auto& ci = catinfo[category];
      ++ci.count;
      ci.compressed_size += fsinfo.compressed_block_sizes.at(block);
      if (auto size = fsinfo.uncompressed_block_sizes.at(block)) {
        ci.uncompressed_size += *size;
      } else {
        ci.uncompressed_size_is_estimate = true;
      }
    }
  }

  return catinfo;
}

const uint16_t READ_ONLY_MASK = ~uint16_t(
    fs::perms::owner_write | fs::perms::group_write | fs::perms::others_write);

} // namespace

template <typename LoggerPolicy>
class metadata_ final : public metadata_v2::impl {
 public:
  metadata_(logger& lgr, std::span<uint8_t const> schema,
            std::span<uint8_t const> data, metadata_options const& options,
            int inode_offset, bool force_consistency_check)
      : data_(data)
      , meta_(
            check_frozen(map_frozen<thrift::metadata::metadata>(schema, data_)))
      , global_(lgr, check_metadata_consistency(lgr, meta_,
                                                options.check_consistency ||
                                                    force_consistency_check))
      , root_(dir_entry_view::from_dir_entry_index(0, global_))
      , LOG_PROXY_INIT(lgr)
      , inode_offset_(inode_offset)
      , symlink_inode_offset_(find_inode_offset(inode_rank::INO_LNK))
      , file_inode_offset_(find_inode_offset(inode_rank::INO_REG))
      , dev_inode_offset_(find_inode_offset(inode_rank::INO_DEV))
      , inode_count_(meta_.dir_entries() ? meta_.inodes().size()
                                         : meta_.entry_table_v2_2().size())
      , nlinks_(build_nlinks(options))
      , chunk_table_(unpack_chunk_table())
      , shared_files_(decompress_shared_files())
      , unique_files_(dev_inode_offset_ - file_inode_offset_ -
                      (shared_files_.empty()
                           ? meta_.shared_files_table()
                                 ? meta_.shared_files_table()->size()
                                 : 0
                           : shared_files_.size()))
      , options_(options)
      , symlinks_(meta_.compact_symlinks()
                      ? string_table(lgr, "symlinks", *meta_.compact_symlinks())
                      : string_table(meta_.symlinks())) {
    if (static_cast<int>(meta_.directories().size() - 1) !=
        symlink_inode_offset_) {
      DWARFS_THROW(
          runtime_error,
          fmt::format("metadata inconsistency: number of directories ({}) does "
                      "not match link index ({})",
                      meta_.directories().size() - 1, symlink_inode_offset_));
    }

    if (static_cast<int>(meta_.symlink_table().size()) !=
        (file_inode_offset_ - symlink_inode_offset_)) {
      DWARFS_THROW(
          runtime_error,
          fmt::format(
              "metadata inconsistency: number of symlinks ({}) does not match "
              "chunk/symlink table delta ({} - {} = {})",
              meta_.symlink_table().size(), file_inode_offset_,
              symlink_inode_offset_,
              file_inode_offset_ - symlink_inode_offset_));
    }

    if (!meta_.shared_files_table()) {
      if (static_cast<int>(meta_.chunk_table().size() - 1) !=
          (dev_inode_offset_ - file_inode_offset_)) {
        DWARFS_THROW(
            runtime_error,
            fmt::format(
                "metadata inconsistency: number of files ({}) does not match "
                "device/chunk index delta ({} - {} = {})",
                meta_.chunk_table().size() - 1, dev_inode_offset_,
                file_inode_offset_, dev_inode_offset_ - file_inode_offset_));
      }
    }

    if (auto devs = meta_.devices()) {
      int other_offset = find_inode_offset(inode_rank::INO_OTH);

      if (static_cast<int>(devs->size()) !=
          (other_offset - dev_inode_offset_)) {
        DWARFS_THROW(
            runtime_error,
            fmt::format("metadata inconsistency: number of devices ({}) does "
                        "not match other/device index delta ({} - {} = {})",
                        devs->size(), other_offset, dev_inode_offset_,
                        other_offset - dev_inode_offset_));
      }
    }
  }

  void check_consistency() const override;

  void dump(std::ostream& os, int detail_level, filesystem_info const& fsinfo,
            std::function<void(const std::string&, uint32_t)> const& icb)
      const override;

  folly::dynamic info_as_dynamic(int detail_level,
                                 filesystem_info const& fsinfo) const override;

  folly::dynamic as_dynamic() const override;
  std::string serialize_as_json(bool simple) const override;

  size_t size() const override { return data_.size(); }

  bool empty() const override { return data_.empty(); }

  void walk(std::function<void(dir_entry_view)> const& func) const override {
    walk_tree([&](uint32_t self_index, uint32_t parent_index) {
      walk_call(func, self_index, parent_index);
    });
  }

  void walk_data_order(
      std::function<void(dir_entry_view)> const& func) const override {
    walk_data_order_impl(func);
  }

  std::optional<inode_view> find(const char* path) const override;
  std::optional<inode_view> find(int inode) const override;
  std::optional<inode_view> find(int inode, const char* name) const override;

  int getattr(inode_view iv, file_stat* stbuf) const override;

  std::optional<directory_view> opendir(inode_view iv) const override;

  std::optional<std::pair<inode_view, std::string>>
  readdir(directory_view dir, size_t offset) const override;

  size_t dirsize(directory_view dir) const override {
    return 2 + dir.entry_count(); // adds '.' and '..', which we fake in ;-)
  }

  int access(inode_view iv, int mode, uid_t uid, gid_t gid) const override;

  int open(inode_view iv) const override;

  int readlink(inode_view iv, std::string* buf,
               readlink_mode mode) const override;

  folly::Expected<std::string, int>
  readlink(inode_view iv, readlink_mode mode) const override;

  int statvfs(vfs_stat* stbuf) const override;

  std::optional<chunk_range> get_chunks(int inode) const override;

  size_t block_size() const override { return meta_.block_size(); }

  bool has_symlinks() const override { return !meta_.symlink_table().empty(); }

  folly::dynamic get_inode_info(inode_view iv) const override;

  std::optional<std::string>
  get_block_category(size_t block_number) const override;

  std::vector<std::string> get_all_block_categories() const override;

 private:
  template <typename K>
  using set_type = folly::F14ValueSet<K>;

  thrift::metadata::metadata unpack_metadata() const;

  inode_view make_inode_view(uint32_t inode) const {
    // TODO: move compatibility details to metadata_types
    uint32_t index =
        meta_.dir_entries() ? inode : meta_.entry_table_v2_2()[inode];
    return inode_view(meta_.inodes()[index], inode, meta_);
  }

  dir_entry_view
  make_dir_entry_view(uint32_t self_index, uint32_t parent_index) const {
    return dir_entry_view::from_dir_entry_index(self_index, parent_index,
                                                global_);
  }

  // This represents the order in which inodes are stored in inodes
  // (or entry_table_v2_2 for older file systems)
  enum class inode_rank {
    INO_DIR,
    INO_LNK,
    INO_REG,
    INO_DEV,
    INO_OTH,
  };

  // TODO: merge with mode_rank in metadata_types
  static inode_rank get_inode_rank(uint16_t mode) {
    switch (posix_file_type::from_mode(mode)) {
    case posix_file_type::directory:
      return inode_rank::INO_DIR;
    case posix_file_type::symlink:
      return inode_rank::INO_LNK;
    case posix_file_type::regular:
      return inode_rank::INO_REG;
    case posix_file_type::block:
    case posix_file_type::character:
      return inode_rank::INO_DEV;
    case posix_file_type::socket:
    case posix_file_type::fifo:
      return inode_rank::INO_OTH;
    default:
      DWARFS_THROW(runtime_error,
                   fmt::format("unknown file type: {:#06x}", mode));
    }
  }

  static char get_filetype_label(uint16_t mode) {
    switch (posix_file_type::from_mode(mode)) {
    case posix_file_type::regular:
      return '-';
    case posix_file_type::directory:
      return 'd';
    case posix_file_type::symlink:
      return 'l';
    case posix_file_type::block:
      return 'b';
    case posix_file_type::character:
      return 'c';
    case posix_file_type::fifo:
      return 'p';
    case posix_file_type::socket:
      return 's';
    default:
      DWARFS_THROW(runtime_error,
                   fmt::format("unknown file type: {:#06x}", mode));
    }
  }

  size_t find_inode_offset(inode_rank rank) const {
    if (meta_.dir_entries()) {
      auto range = boost::irange(size_t(0), meta_.inodes().size());

      auto it = std::lower_bound(
          range.begin(), range.end(), rank, [&](auto inode, inode_rank r) {
            auto mode = meta_.modes()[meta_.inodes()[inode].mode_index()];
            return get_inode_rank(mode) < r;
          });

      return *it;
    } else {
      auto range = boost::irange(size_t(0), meta_.entry_table_v2_2().size());

      auto it = std::lower_bound(range.begin(), range.end(), rank,
                                 [&](auto inode, inode_rank r) {
                                   auto iv = make_inode_view(inode);
                                   return get_inode_rank(iv.mode()) < r;
                                 });

      return *it;
    }
  }

  directory_view make_directory_view(inode_view iv) const {
    // TODO: revisit: is this the way to do it?
    return directory_view(iv.inode_num(), global_);
  }

  void analyze_chunks(std::ostream& os) const;

  // TODO: see if we really need to pass the extra dir_entry_view in
  //       addition to directory_view
  void dump(std::ostream& os, const std::string& indent, dir_entry_view entry,
            int detail_level,
            std::function<void(const std::string&, uint32_t)> const& icb) const;
  void dump(std::ostream& os, const std::string& indent, directory_view dir,
            dir_entry_view entry, int detail_level,
            std::function<void(const std::string&, uint32_t)> const& icb) const;

  folly::dynamic as_dynamic(dir_entry_view entry) const;
  folly::dynamic as_dynamic(directory_view dir, dir_entry_view entry) const;

  std::optional<inode_view>
  find(directory_view dir, std::string_view name) const;

  std::string modestring(uint16_t mode) const;

  uint32_t chunk_table_lookup(uint32_t ino) const {
    return chunk_table_.empty() ? meta_.chunk_table()[ino] : chunk_table_[ino];
  }

  int file_inode_to_chunk_index(int inode) const {
    inode -= file_inode_offset_;

    if (inode >= unique_files_) {
      inode -= unique_files_;

      if (!shared_files_.empty()) {
        if (inode < static_cast<int>(shared_files_.size())) {
          inode = shared_files_[inode] + unique_files_;
        }
      } else if (auto sfp = meta_.shared_files_table()) {
        if (inode < static_cast<int>(sfp->size())) {
          inode = (*sfp)[inode] + unique_files_;
        }
      }
    }

    return inode;
  }

  std::optional<chunk_range> get_chunk_range(int inode) const {
    std::optional<chunk_range> rv;

    inode = file_inode_to_chunk_index(inode);

    if (inode >= 0 &&
        inode < (static_cast<int>(meta_.chunk_table().size()) - 1)) {
      uint32_t begin = chunk_table_lookup(inode);
      uint32_t end = chunk_table_lookup(inode + 1);
      rv = chunk_range(meta_, begin, end);
    }

    return rv;
  }

  size_t reg_file_size(inode_view iv) const {
    auto cr = get_chunk_range(iv.inode_num());
    DWARFS_CHECK(cr, "invalid chunk range");
    return std::accumulate(
        cr->begin(), cr->end(), static_cast<size_t>(0),
        [](size_t s, chunk_view cv) { return s + cv.size(); });
  }

  size_t file_size(inode_view iv, uint16_t mode) const {
    switch (posix_file_type::from_mode(mode)) {
    case posix_file_type::regular:
      return reg_file_size(iv);
    case posix_file_type::symlink:
      return link_value(iv).size();
    default:
      return 0;
    }
  }

  // TODO: cleanup the walk logic
  void walk_call(std::function<void(dir_entry_view)> const& func,
                 uint32_t self_index, uint32_t parent_index) const {
    func(make_dir_entry_view(self_index, parent_index));
  }

  template <typename T>
  void walk(uint32_t self_index, uint32_t parent_index, set_type<int>& seen,
            T&& func) const;

  template <typename T>
  void walk_tree(T&& func) const {
    set_type<int> seen;
    walk(0, 0, seen, std::forward<T>(func));
  }

  void
  walk_data_order_impl(std::function<void(dir_entry_view)> const& func) const;

  std::optional<inode_view> get_entry(int inode) const {
    inode -= inode_offset_;
    std::optional<inode_view> rv;
    if (inode >= 0 && inode < inode_count_) {
      rv = make_inode_view(inode);
    }
    return rv;
  }

  std::string
  link_value(inode_view iv, readlink_mode mode = readlink_mode::raw) const {
    std::string rv =
        symlinks_[meta_
                      .symlink_table()[iv.inode_num() - symlink_inode_offset_]];

    if (mode != readlink_mode::raw) {
      char meta_preferred = '/';
      if (auto ps = meta_.preferred_path_separator()) {
        meta_preferred = static_cast<char>(*ps);
      }
      char host_preferred =
          static_cast<char>(std::filesystem::path::preferred_separator);
      if (mode == readlink_mode::unix) {
        host_preferred = '/';
      }
      if (meta_preferred != host_preferred) {
        std::replace(rv.begin(), rv.end(), meta_preferred, host_preferred);
      }
    }

    return rv;
  }

  uint64_t get_device_id(int inode) const {
    if (auto devs = meta_.devices()) {
      return (*devs)[inode - dev_inode_offset_];
    }
    LOG_ERROR << "get_device_id() called, but no devices in file system";
    return 0;
  }

  std::vector<uint32_t> unpack_chunk_table() const {
    std::vector<uint32_t> chunk_table;

    if (auto opts = meta_.options(); opts and opts->packed_chunk_table()) {
      auto ti = LOG_TIMED_DEBUG;

      chunk_table.resize(meta_.chunk_table().size());
      std::partial_sum(meta_.chunk_table().begin(), meta_.chunk_table().end(),
                       chunk_table.begin());

      ti << "unpacked chunk table ("
         << size_with_unit(sizeof(chunk_table.front()) * chunk_table.capacity())
         << ")";
    }

    return chunk_table;
  }

  std::vector<uint32_t> decompress_shared_files() const {
    std::vector<uint32_t> decompressed;

    if (auto opts = meta_.options();
        opts and opts->packed_shared_files_table()) {
      if (auto sfp = meta_.shared_files_table(); sfp and !sfp->empty()) {
        auto ti = LOG_TIMED_DEBUG;

        auto size = std::accumulate(sfp->begin(), sfp->end(), 2 * sfp->size());
        decompressed.reserve(size);

        uint32_t index = 0;
        for (auto c : *sfp) {
          decompressed.insert(decompressed.end(), c + 2, index++);
        }

        DWARFS_CHECK(decompressed.size() == size,
                     "unexpected decompressed shared files count");

        ti << "decompressed shared files table ("
           << size_with_unit(sizeof(decompressed.front()) *
                             decompressed.capacity())
           << ")";
      }
    }

    return decompressed;
  }

  std::vector<uint32_t> build_nlinks(metadata_options const& options) const {
    std::vector<uint32_t> nlinks;

    if (options.enable_nlink) {
      auto ti = LOG_TIMED_DEBUG;

      nlinks.resize(dev_inode_offset_ - file_inode_offset_);

      if (auto de = meta_.dir_entries()) {
        for (auto e : *de) {
          int index = int(e.inode_num()) - file_inode_offset_;
          if (index >= 0 && index < int(nlinks.size())) {
            ++nlinks[index];
          }
        }
      } else {
        for (auto e : meta_.inodes()) {
          int index = int(e.inode_v2_2()) - file_inode_offset_;
          if (index >= 0 && index < int(nlinks.size())) {
            ++nlinks[index];
          }
        }
      }

      ti << "built hardlink table ("
         << size_with_unit(sizeof(nlinks.front()) * nlinks.capacity()) << ")";
    }

    return nlinks;
  }

  std::span<uint8_t const> data_;
  MappedFrozen<thrift::metadata::metadata> meta_;
  const global_metadata global_;
  dir_entry_view root_;
  LOG_PROXY_DECL(LoggerPolicy);
  const int inode_offset_;
  const int symlink_inode_offset_;
  const int file_inode_offset_;
  const int dev_inode_offset_;
  const int inode_count_;
  const std::vector<uint32_t> nlinks_;
  const std::vector<uint32_t> chunk_table_;
  const std::vector<uint32_t> shared_files_;
  const int unique_files_;
  const metadata_options options_;
  const string_table symlinks_;
};

template <typename LoggerPolicy>
void metadata_<LoggerPolicy>::analyze_chunks(std::ostream& os) const {
  folly::Histogram<size_t> block_refs{1, 0, 1024};
  folly::Histogram<size_t> chunk_count{1, 0, 65536};
  size_t mergeable_chunks{0};

  for (size_t i = 1; i < meta_.chunk_table().size(); ++i) {
    uint32_t beg = chunk_table_lookup(i - 1);
    uint32_t end = chunk_table_lookup(i);
    uint32_t num = end - beg;

    assert(beg <= end);

    if (num > 1) {
      std::unordered_set<size_t> blocks;

      for (uint32_t k = beg; k < end; ++k) {
        auto chk = meta_.chunks()[k];
        blocks.emplace(chk.block());

        if (k > beg) {
          auto prev = meta_.chunks()[k - 1];
          if (prev.block() == chk.block()) {
            if (prev.offset() + prev.size() == chk.offset()) {
              ++mergeable_chunks;
            }
          }
        }
      }

      block_refs.addValue(blocks.size());
    } else {
      block_refs.addValue(num);
    }

    chunk_count.addValue(num);
  }

  {
    auto pct = [&](double p) { return block_refs.getPercentileEstimate(p); };

    os << "single file block refs p50: " << pct(0.5) << ", p75: " << pct(0.75)
       << ", p90: " << pct(0.9) << ", p95: " << pct(0.95)
       << ", p99: " << pct(0.99) << ", p99.9: " << pct(0.999) << "\n";
  }

  {
    auto pct = [&](double p) { return chunk_count.getPercentileEstimate(p); };

    os << "single file chunk count p50: " << pct(0.5) << ", p75: " << pct(0.75)
       << ", p90: " << pct(0.9) << ", p95: " << pct(0.95)
       << ", p99: " << pct(0.99) << ", p99.9: " << pct(0.999) << "\n";
  }

  // TODO: we can remove this once we have no more mergeable chunks :-)
  os << "mergeable chunks: " << mergeable_chunks << "/" << meta_.chunks().size()
     << "\n";
}

template <typename LoggerPolicy>
void metadata_<LoggerPolicy>::check_consistency() const {
  global_metadata::check_consistency(LOG_GET_LOGGER, meta_);
}

template <typename LoggerPolicy>
void metadata_<LoggerPolicy>::dump(
    std::ostream& os, const std::string& indent, dir_entry_view entry,
    int detail_level,
    std::function<void(const std::string&, uint32_t)> const& icb) const {
  auto iv = entry.inode();
  auto mode = iv.mode();
  auto inode = iv.inode_num();

  os << indent << "<inode:" << inode << "> " << modestring(mode);

  if (inode > 0) {
    os << " " << entry.name();
  }

  switch (posix_file_type::from_mode(mode)) {
  case posix_file_type::regular: {
    auto cr = get_chunk_range(inode);
    DWARFS_CHECK(cr, "invalid chunk range");
    os << " [" << cr->begin_ << ", " << cr->end_ << "]";
    os << " " << file_size(iv, mode) << "\n";
    if (detail_level > 4) {
      icb(indent + "  ", inode);
    }
  } break;

  case posix_file_type::directory:
    dump(os, indent + "  ", make_directory_view(iv), entry, detail_level,
         std::move(icb));
    break;

  case posix_file_type::symlink:
    os << " -> " << link_value(iv) << "\n";
    break;

  case posix_file_type::block:
    os << " (block device: " << get_device_id(inode) << ")\n";
    break;

  case posix_file_type::character:
    os << " (char device: " << get_device_id(inode) << ")\n";
    break;

  case posix_file_type::fifo:
    os << " (named pipe)\n";
    break;

  case posix_file_type::socket:
    os << " (socket)\n";
    break;
  }
}

template <typename LoggerPolicy>
folly::dynamic
metadata_<LoggerPolicy>::info_as_dynamic(int detail_level,
                                         filesystem_info const& fsinfo) const {
  folly::dynamic info = folly::dynamic::object;
  vfs_stat stbuf;
  statvfs(&stbuf);

  if (auto version = meta_.dwarfs_version()) {
    info["created_by"] = version.value();
  }

  if (auto ts = meta_.create_timestamp()) {
    info["created_on"] =
        fmt::format("{:%Y-%m-%dT%H:%M:%S}", fmt::localtime(ts.value()));
  }

  if (detail_level > 0) {
    info["block_size"] = stbuf.bsize;
    info["block_count"] = fsinfo.block_count;
    info["inode_count"] = stbuf.files;
    if (auto ps = meta_.preferred_path_separator()) {
      info["preferred_path_separator"] = std::string(1, static_cast<char>(*ps));
    }
    info["original_filesystem_size"] = stbuf.blocks;
    info["compressed_block_size"] = fsinfo.compressed_block_size;
    if (!fsinfo.uncompressed_block_size_is_estimate) {
      info["uncompressed_block_size"] = fsinfo.uncompressed_block_size;
    }
    info["compressed_metadata_size"] = fsinfo.compressed_metadata_size;
    if (!fsinfo.uncompressed_metadata_size_is_estimate) {
      info["uncompressed_metadata_size"] = fsinfo.uncompressed_metadata_size;
    }

    if (auto opt = meta_.options()) {
      folly::dynamic options = folly::dynamic::array;
      parse_metadata_options(meta_, [&](auto const& name, bool value) {
        if (value) {
          options.push_back(name);
        }
      });
      info["options"] = std::move(options);
      if (auto res = opt->time_resolution_sec()) {
        info["time_resolution"] = *res;
      }
    }

    if (meta_.block_categories()) {
      auto catnames = *meta_.category_names();
      auto catinfo = get_category_info(meta_, fsinfo);
      folly::dynamic categories = folly::dynamic::object;
      for (auto const& [category, ci] : catinfo) {
        categories[catnames[category]] = folly::dynamic::object(
            "block_count", ci.count)("compressed_size", ci.compressed_size);
        if (!ci.uncompressed_size_is_estimate) {
          categories[catnames[category]]["uncompressed_size"] =
              ci.uncompressed_size;
        }
      }
      info["categories"] = std::move(categories);
    }
  }

  if (detail_level > 2) {
    folly::dynamic meta = folly::dynamic::object;
    meta["symlink_inode_offset"] = symlink_inode_offset_;
    meta["file_inode_offset"] = file_inode_offset_;
    meta["dev_inode_offset"] = dev_inode_offset_;
    meta["chunks"] = meta_.chunks().size();
    meta["directories"] = meta_.directories().size();
    meta["inodes"] = meta_.inodes().size();
    meta["chunk_table"] = meta_.chunk_table().size();
    meta["entry_table_v2_2"] = meta_.entry_table_v2_2().size();
    meta["symlink_table"] = meta_.symlink_table().size();
    meta["uids"] = meta_.uids().size();
    meta["gids"] = meta_.gids().size();
    meta["modes"] = meta_.modes().size();
    meta["names"] = meta_.names().size();
    meta["symlinks"] = meta_.symlinks().size();

    if (auto dev = meta_.devices()) {
      meta["devices"] = dev->size();
    }

    if (auto de = meta_.dir_entries()) {
      meta["dir_entries"] = de->size();
    }

    if (auto sfp = meta_.shared_files_table()) {
      if (meta_.options()->packed_shared_files_table()) {
        meta["packed_shared_files_table"] = sfp->size();
        meta["unpacked_shared_files_table"] = shared_files_.size();
      } else {
        meta["shared_files_table"] = sfp->size();
      }
      meta["unique_files"] = unique_files_;
    }

    info["meta"] = std::move(meta);
  }

  if (detail_level > 3) {
    info["root"] = as_dynamic(root_);
  }

  return info;
}

// TODO: can we move this to dir_entry_view?
template <typename LoggerPolicy>
void metadata_<LoggerPolicy>::dump(
    std::ostream& os, const std::string& indent, directory_view dir,
    dir_entry_view entry, int detail_level,
    std::function<void(const std::string&, uint32_t)> const& icb) const {
  auto count = dir.entry_count();
  auto first = dir.first_entry();

  os << " (" << count << " entries, parent=" << dir.parent_entry() << ")\n";

  for (size_t i = 0; i < count; ++i) {
    dump(os, indent, make_dir_entry_view(first + i, entry.self_index()),
         detail_level, icb);
  }
}

template <typename LoggerPolicy>
void metadata_<LoggerPolicy>::dump(
    std::ostream& os, int detail_level, filesystem_info const& fsinfo,
    std::function<void(const std::string&, uint32_t)> const& icb) const {
  vfs_stat stbuf;
  statvfs(&stbuf);

  if (auto version = meta_.dwarfs_version()) {
    os << "created by: " << *version << "\n";
  }

  if (auto ts = meta_.create_timestamp()) {
    time_t tp = *ts;
    std::string str(32, '\0');
    str.resize(
        std::strftime(str.data(), str.size(), "%F %T", std::localtime(&tp)));
    os << "created on: " << str << "\n";
  }

  if (detail_level > 0) {
    os << "block size: " << size_with_unit(stbuf.bsize) << "\n";
    os << "block count: " << fsinfo.block_count << "\n";
    os << "inode count: " << stbuf.files << "\n";
    if (auto ps = meta_.preferred_path_separator()) {
      os << "preferred path separator: " << static_cast<char>(*ps) << "\n";
    }
    os << "original filesystem size: " << size_with_unit(stbuf.blocks) << "\n";
    os << "compressed block size: "
       << size_with_unit(fsinfo.compressed_block_size);
    if (!fsinfo.uncompressed_block_size_is_estimate) {
      os << fmt::format(" ({0:.2f}%)", (100.0 * fsinfo.compressed_block_size) /
                                           fsinfo.uncompressed_block_size);
    }
    os << "\n";
    os << "uncompressed block size: ";
    if (fsinfo.uncompressed_block_size_is_estimate) {
      os << "(at least) ";
    }
    os << size_with_unit(fsinfo.uncompressed_block_size) << "\n";
    os << "compressed metadata size: "
       << size_with_unit(fsinfo.compressed_metadata_size);
    if (!fsinfo.uncompressed_metadata_size_is_estimate) {
      os << fmt::format(" ({0:.2f}%)",
                        (100.0 * fsinfo.compressed_metadata_size) /
                            fsinfo.uncompressed_metadata_size);
    }
    os << "\n";
    os << "uncompressed metadata size: ";
    if (fsinfo.uncompressed_metadata_size_is_estimate) {
      os << "(at least) ";
    }
    os << size_with_unit(fsinfo.uncompressed_metadata_size) << "\n";
    if (auto opt = meta_.options()) {
      std::vector<std::string> options;
      parse_metadata_options(meta_, [&](auto const& name, bool value) {
        if (value) {
          options.push_back(name);
        }
      });
      os << "options: " << boost::join(options, "\n         ") << "\n";
      if (auto res = opt->time_resolution_sec()) {
        os << "time resolution: " << *res << " seconds\n";
      }
    }

    if (meta_.block_categories()) {
      auto catnames = *meta_.category_names();
      auto catinfo = get_category_info(meta_, fsinfo);
      os << "categories:\n";
      for (auto const& [category, ci] : catinfo) {
        os << "  " << catnames[category] << ": " << ci.count << " blocks";
        if (ci.uncompressed_size_is_estimate ||
            ci.uncompressed_size != ci.compressed_size) {
          os << ", " << size_with_unit(ci.compressed_size) << " compressed";
        }
        if (!ci.uncompressed_size_is_estimate) {
          os << ", " << size_with_unit(ci.uncompressed_size) << " uncompressed";
          if (ci.uncompressed_size != ci.compressed_size) {
            os << fmt::format(" ({0:.2f}%)", (100.0 * ci.compressed_size) /
                                                 ci.uncompressed_size);
          }
        }
        os << "\n";
      }
    }
  }

  if (detail_level > 1) {
    analyze_frozen(os, meta_, data_.size(), detail_level);
  }

  if (detail_level > 2) {
    os << "symlink_inode_offset: " << symlink_inode_offset_ << "\n";
    os << "file_inode_offset: " << file_inode_offset_ << "\n";
    os << "dev_inode_offset: " << dev_inode_offset_ << "\n";
    os << "chunks: " << meta_.chunks().size() << "\n";
    os << "directories: " << meta_.directories().size() << "\n";
    os << "inodes: " << meta_.inodes().size() << "\n";
    os << "chunk_table: " << meta_.chunk_table().size() << "\n";
    os << "entry_table_v2_2: " << meta_.entry_table_v2_2().size() << "\n";
    os << "symlink_table: " << meta_.symlink_table().size() << "\n";
    os << "uids: " << meta_.uids().size() << "\n";
    os << "gids: " << meta_.gids().size() << "\n";
    os << "modes: " << meta_.modes().size() << "\n";
    os << "names: " << meta_.names().size() << "\n";
    os << "symlinks: " << meta_.symlinks().size() << "\n";
    if (auto dev = meta_.devices()) {
      os << "devices: " << dev->size() << "\n";
    }
    if (auto de = meta_.dir_entries()) {
      os << "dir_entries: " << de->size() << "\n";
    }
    if (auto sfp = meta_.shared_files_table()) {
      if (meta_.options()->packed_shared_files_table()) {
        os << "packed shared_files_table: " << sfp->size() << "\n";
        os << "unpacked shared_files_table: " << shared_files_.size() << "\n";
      } else {
        os << "shared_files_table: " << sfp->size() << "\n";
      }
      os << "unique files: " << unique_files_ << "\n";
    }
    analyze_chunks(os);
  }

  if (detail_level > 5) {
    os << ::apache::thrift::debugString(meta_.thaw()) << '\n';
  }

  if (detail_level > 3) {
    dump(os, "", root_, detail_level, icb);
  }
}

template <typename LoggerPolicy>
folly::dynamic metadata_<LoggerPolicy>::as_dynamic(directory_view dir,
                                                   dir_entry_view entry) const {
  folly::dynamic obj = folly::dynamic::array;

  auto count = dir.entry_count();
  auto first = dir.first_entry();

  for (size_t i = 0; i < count; ++i) {
    obj.push_back(
        as_dynamic(make_dir_entry_view(first + i, entry.self_index())));
  }

  return obj;
}

template <typename LoggerPolicy>
folly::dynamic metadata_<LoggerPolicy>::as_dynamic(dir_entry_view entry) const {
  folly::dynamic obj = folly::dynamic::object;

  auto iv = entry.inode();
  auto mode = iv.mode();
  auto inode = iv.inode_num();

  obj["mode"] = mode;
  obj["modestring"] = modestring(mode);
  obj["inode"] = inode;

  if (inode > 0) {
    obj["name"] = std::string(entry.name());
  }

  switch (posix_file_type::from_mode(mode)) {
  case posix_file_type::regular:
    obj["type"] = "file";
    obj["size"] = file_size(iv, mode);
    break;

  case posix_file_type::directory:
    obj["type"] = "directory";
    obj["inodes"] = as_dynamic(make_directory_view(iv), entry);
    break;

  case posix_file_type::symlink:
    obj["type"] = "link";
    obj["target"] = link_value(iv);
    break;

  case posix_file_type::block:
    obj["type"] = "blockdev";
    obj["device_id"] = get_device_id(inode);
    break;

  case posix_file_type::character:
    obj["type"] = "chardev";
    obj["device_id"] = get_device_id(inode);
    break;

  case posix_file_type::fifo:
    obj["type"] = "fifo";
    break;

  case posix_file_type::socket:
    obj["type"] = "socket";
    break;
  }

  return obj;
}

template <typename LoggerPolicy>
folly::dynamic metadata_<LoggerPolicy>::as_dynamic() const {
  folly::dynamic obj = folly::dynamic::object;

  vfs_stat stbuf;
  statvfs(&stbuf);

  obj["statvfs"] = folly::dynamic::object("f_bsize", stbuf.bsize)(
      "f_files", stbuf.files)("f_blocks", stbuf.blocks);

  obj["root"] = as_dynamic(root_);

  return obj;
}

template <typename LoggerPolicy>
thrift::metadata::metadata metadata_<LoggerPolicy>::unpack_metadata() const {
  auto meta = meta_.thaw();

  if (auto opts = meta.options()) {
    if (opts->packed_chunk_table().value()) {
      meta.chunk_table() = chunk_table_;
    }
    if (opts->packed_directories().value()) {
      meta.directories() = global_.directories();
    }
    if (opts->packed_shared_files_table().value()) {
      meta.shared_files_table() = shared_files_;
    }
    if (auto const& names = global_.names(); names.is_packed()) {
      meta.names() = names.unpack();
      meta.compact_names().reset();
    }
    if (symlinks_.is_packed()) {
      meta.symlinks() = symlinks_.unpack();
      meta.compact_symlinks().reset();
    }
    opts->packed_chunk_table() = false;
    opts->packed_directories() = false;
    opts->packed_shared_files_table() = false;
  }

  return meta;
}

template <typename LoggerPolicy>
std::string metadata_<LoggerPolicy>::serialize_as_json(bool simple) const {
  std::string json;
  if (simple) {
    apache::thrift::SimpleJSONSerializer serializer;
    serializer.serialize(unpack_metadata(), &json);
  } else {
    apache::thrift::JSONSerializer serializer;
    serializer.serialize(unpack_metadata(), &json);
  }
  return json;
}

template <typename LoggerPolicy>
std::string metadata_<LoggerPolicy>::modestring(uint16_t mode) const {
  std::ostringstream oss;

  oss << (mode & uint16_t(fs::perms::set_uid) ? 'U' : '-');
  oss << (mode & uint16_t(fs::perms::set_gid) ? 'G' : '-');
  oss << (mode & uint16_t(fs::perms::sticky_bit) ? 'S' : '-');
  oss << get_filetype_label(mode);
  oss << (mode & uint16_t(fs::perms::owner_read) ? 'r' : '-');
  oss << (mode & uint16_t(fs::perms::owner_write) ? 'w' : '-');
  oss << (mode & uint16_t(fs::perms::owner_exec) ? 'x' : '-');
  oss << (mode & uint16_t(fs::perms::group_read) ? 'r' : '-');
  oss << (mode & uint16_t(fs::perms::group_write) ? 'w' : '-');
  oss << (mode & uint16_t(fs::perms::group_exec) ? 'x' : '-');
  oss << (mode & uint16_t(fs::perms::others_read) ? 'r' : '-');
  oss << (mode & uint16_t(fs::perms::others_write) ? 'w' : '-');
  oss << (mode & uint16_t(fs::perms::others_exec) ? 'x' : '-');

  return oss.str();
}

template <typename LoggerPolicy>
template <typename T>
void metadata_<LoggerPolicy>::walk(uint32_t self_index, uint32_t parent_index,
                                   set_type<int>& seen, T&& func) const {
  func(self_index, parent_index);

  auto entry = make_dir_entry_view(self_index, parent_index);
  auto iv = entry.inode();

  if (iv.is_directory()) {
    auto inode = iv.inode_num();

    if (!seen.emplace(inode).second) {
      DWARFS_THROW(runtime_error, "cycle detected during directory walk");
    }

    auto dir = make_directory_view(iv);

    for (auto cur_index : dir.entry_range()) {
      walk(cur_index, self_index, seen, func);
    }

    seen.erase(inode);
  }
}

template <typename LoggerPolicy>
void metadata_<LoggerPolicy>::walk_data_order_impl(
    std::function<void(dir_entry_view)> const& func) const {
  std::vector<std::pair<uint32_t, uint32_t>> entries;

  if (auto dep = meta_.dir_entries()) {
    entries.reserve(dep->size());
  } else {
    entries.reserve(meta_.inodes().size());
  }

  {
    auto td = LOG_TIMED_DEBUG;

    walk_tree([&](uint32_t self_index, uint32_t parent_index) {
      entries.emplace_back(self_index, parent_index);
    });

    if (auto dep = meta_.dir_entries()) {
      // 1. partition non-files / files
      auto mid =
          std::stable_partition(entries.begin(), entries.end(),
                                [de = *dep, beg = file_inode_offset_,
                                 end = dev_inode_offset_](auto const& e) {
                                  int ino = de[e.first].inode_num();
                                  return ino < beg or ino >= end;
                                });

      // 2. order files by chunk block number
      // 2a. build mapping inode -> first chunk block
      std::vector<uint32_t> first_chunk_block;

      {
        auto td2 = LOG_TIMED_DEBUG;

        first_chunk_block.resize(dep->size());

        for (size_t ix = 0; ix < first_chunk_block.size(); ++ix) {
          int ino = (*dep)[ix].inode_num();
          if (ino >= file_inode_offset_ and ino < dev_inode_offset_) {
            ino = file_inode_to_chunk_index(ino);
            if (auto beg = chunk_table_lookup(ino);
                beg != chunk_table_lookup(ino + 1)) {
              first_chunk_block[ix] = meta_.chunks()[beg].block();
            }
          }
        }

        td2 << "prepare first chunk block vector";
      }

      // 2b. sort second partition accordingly
      {
        auto td2 = LOG_TIMED_DEBUG;

        std::stable_sort(mid, entries.end(),
                         [&first_chunk_block](auto const& a, auto const& b) {
                           return first_chunk_block[a.first] <
                                  first_chunk_block[b.first];
                         });

        td2 << "final sort of " << std::distance(mid, entries.end())
            << " file entries";
      }
    } else {
      std::sort(entries.begin(), entries.end(),
                [this](auto const& a, auto const& b) {
                  return meta_.inodes()[a.first].inode_v2_2() <
                         meta_.inodes()[b.first].inode_v2_2();
                });
    }

    td << "ordered " << entries.size() << " entries by file data order";
  }

  for (auto [self_index, parent_index] : entries) {
    walk_call(func, self_index, parent_index);
  }
}

template <typename LoggerPolicy>
std::optional<inode_view>
metadata_<LoggerPolicy>::find(directory_view dir, std::string_view name) const {
  auto range = dir.entry_range();

  auto it = std::lower_bound(range.begin(), range.end(), name,
                             [&](auto ix, std::string_view name) {
                               return dir_entry_view::name(ix, global_) < name;
                             });

  std::optional<inode_view> rv;

  if (it != range.end()) {
    if (dir_entry_view::name(*it, global_) == name) {
      rv = dir_entry_view::inode(*it, global_);
    }
  }

  return rv;
}

template <typename LoggerPolicy>
std::optional<inode_view>
metadata_<LoggerPolicy>::find(const char* path) const {
  while (*path == '/') {
    ++path;
  }

  std::optional<inode_view> iv = root_.inode();

  while (*path) {
    const char* next = ::strchr(path, '/');
    size_t clen = next ? next - path : ::strlen(path); // Flawfinder: ignore

    iv = find(make_directory_view(*iv), std::string_view(path, clen));

    if (!iv) {
      break;
    }

    path = next ? next + 1 : path + clen;
  }

  return iv;
}

template <typename LoggerPolicy>
std::optional<inode_view> metadata_<LoggerPolicy>::find(int inode) const {
  return get_entry(inode);
}

template <typename LoggerPolicy>
std::optional<inode_view>
metadata_<LoggerPolicy>::find(int inode, const char* name) const {
  auto iv = get_entry(inode);

  if (iv) {
    iv = find(make_directory_view(*iv), std::string_view(name));
  }

  return iv;
}

template <typename LoggerPolicy>
int metadata_<LoggerPolicy>::getattr(inode_view iv, file_stat* stbuf) const {
  ::memset(stbuf, 0, sizeof(*stbuf));

  auto mode = iv.mode();
  auto timebase = meta_.timestamp_base();
  auto inode = iv.inode_num();
  bool mtime_only = meta_.options() && meta_.options()->mtime_only();
  uint32_t resolution = 1;
  if (meta_.options()) {
    if (auto res = meta_.options()->time_resolution_sec()) {
      resolution = *res;
      assert(resolution > 0);
    }
  }

  stbuf->mode = mode;

  if (options_.readonly) {
    stbuf->mode &= READ_ONLY_MASK;
  }

  stbuf->size = stbuf->is_directory() ? make_directory_view(iv).entry_count()
                                      : file_size(iv, mode);
  stbuf->ino = inode + inode_offset_;
  stbuf->blksize = 512;
  stbuf->blocks = (stbuf->size + 511) / 512;
  stbuf->uid = iv.getuid();
  stbuf->gid = iv.getgid();
  stbuf->mtime = resolution * (timebase + iv.mtime_offset());
  if (mtime_only) {
    stbuf->atime = stbuf->ctime = stbuf->mtime;
  } else {
    stbuf->atime = resolution * (timebase + iv.atime_offset());
    stbuf->ctime = resolution * (timebase + iv.ctime_offset());
  }
  stbuf->nlink = options_.enable_nlink && stbuf->is_regular_file()
                     ? DWARFS_NOTHROW(nlinks_.at(inode - file_inode_offset_))
                     : 1;

  if (stbuf->is_device()) {
    stbuf->rdev = get_device_id(inode);
  }

  return 0;
}

template <typename LoggerPolicy>
std::optional<directory_view>
metadata_<LoggerPolicy>::opendir(inode_view iv) const {
  std::optional<directory_view> rv;

  if (iv.is_directory()) {
    rv = make_directory_view(iv);
  }

  return rv;
}

template <typename LoggerPolicy>
std::optional<std::pair<inode_view, std::string>>
metadata_<LoggerPolicy>::readdir(directory_view dir, size_t offset) const {
  switch (offset) {
  case 0:
    return std::pair(make_inode_view(dir.inode()), ".");

  case 1:
    return std::pair(make_inode_view(dir.parent_inode()), "..");

  default:
    offset -= 2;

    if (offset >= dir.entry_count()) {
      break;
    }

    auto index = dir.first_entry() + offset;
    auto inode = dir_entry_view::inode(index, global_);
    return std::pair(inode, dir_entry_view::name(index, global_));
  }

  return std::nullopt;
}

template <typename LoggerPolicy>
int metadata_<LoggerPolicy>::access(inode_view iv, int mode, uid_t uid,
                                    gid_t gid) const {
  if (mode == F_OK) {
    // easy; we're only interested in the file's existance
    return 0;
  }

  int access_mode = 0;

  auto test = [e_mode = iv.mode(), &access_mode](fs::perms r_bit,
                                                 fs::perms x_bit) {
    if (e_mode & uint16_t(r_bit)) {
      access_mode |= R_OK;
    }
    if (e_mode & uint16_t(x_bit)) {
      access_mode |= X_OK;
    }
  };

  // Let's build the inode's access mask
  test(fs::perms::others_read, fs::perms::others_exec);

  if (iv.getgid() == gid) {
    test(fs::perms::group_read, fs::perms::group_exec);
  }

  if (iv.getuid() == uid) {
    test(fs::perms::owner_read, fs::perms::owner_exec);
  }

  return (access_mode & mode) == mode ? 0 : EACCES;
}

template <typename LoggerPolicy>
int metadata_<LoggerPolicy>::open(inode_view iv) const {
  if (iv.is_regular_file()) {
    return iv.inode_num();
  }

  return -1;
}

template <typename LoggerPolicy>
int metadata_<LoggerPolicy>::readlink(inode_view iv, std::string* buf,
                                      readlink_mode mode) const {
  if (iv.is_symlink()) {
    buf->assign(link_value(iv, mode));
    return 0;
  }

  return -EINVAL;
}

template <typename LoggerPolicy>
folly::Expected<std::string, int>
metadata_<LoggerPolicy>::readlink(inode_view iv, readlink_mode mode) const {
  if (iv.is_symlink()) {
    return link_value(iv, mode);
  }

  return folly::makeUnexpected(-EINVAL);
}

template <typename LoggerPolicy>
int metadata_<LoggerPolicy>::statvfs(vfs_stat* stbuf) const {
  ::memset(stbuf, 0, sizeof(*stbuf));

  stbuf->bsize = meta_.block_size();
  stbuf->frsize = 1UL;
  stbuf->blocks = meta_.total_fs_size();
  if (!options_.enable_nlink) {
    if (auto ths = meta_.total_hardlink_size()) {
      stbuf->blocks += *ths;
    }
  }
  stbuf->files = inode_count_;
  stbuf->readonly = true;
  stbuf->namemax = PATH_MAX;

  return 0;
}

template <typename LoggerPolicy>
std::optional<chunk_range>
metadata_<LoggerPolicy>::get_chunks(int inode) const {
  return get_chunk_range(inode - inode_offset_);
}

template <typename LoggerPolicy>
folly::dynamic metadata_<LoggerPolicy>::get_inode_info(inode_view iv) const {
  folly::dynamic obj = folly::dynamic::object;

  auto chunk_range = get_chunk_range(iv.inode_num());

  if (chunk_range) {
    obj["chunks"] = folly::dynamic::array;

    for (auto const& chunk : *chunk_range) {
      folly::dynamic chk = folly::dynamic::object;

      chk["block"] = chunk.block();
      chk["offset"] = chunk.offset();
      chk["size"] = chunk.size();

      if (auto catname = get_block_category(chunk.block())) {
        chk["category"] = catname.value();
      }

      obj["chunks"].push_back(chk);
    }
  }

  obj["mode"] = iv.mode();
  obj["modestring"] = modestring(iv.mode());
  obj["uid"] = iv.getuid();
  obj["gid"] = iv.getgid();

  return obj;
}

template <typename LoggerPolicy>
std::optional<std::string>
metadata_<LoggerPolicy>::get_block_category(size_t block_number) const {
  if (auto catnames = meta_.category_names()) {
    if (auto categories = meta_.block_categories()) {
      return std::string(catnames.value()[categories.value()[block_number]]);
    }
  }
  return std::nullopt;
}

template <typename LoggerPolicy>
std::vector<std::string>
metadata_<LoggerPolicy>::get_all_block_categories() const {
  std::vector<std::string> rv;

  if (auto catnames = meta_.category_names()) {
    rv.reserve(catnames.value().size());
    for (auto const& name : catnames.value()) {
      rv.emplace_back(name);
    }
  }

  return rv;
}

std::pair<std::vector<uint8_t>, std::vector<uint8_t>>
metadata_v2::freeze(const thrift::metadata::metadata& data) {
  return freeze_to_buffer(data);
}

metadata_v2::metadata_v2(logger& lgr, std::span<uint8_t const> schema,
                         std::span<uint8_t const> data,
                         metadata_options const& options, int inode_offset,
                         bool force_consistency_check)
    : impl_(make_unique_logging_object<metadata_v2::impl, metadata_,
                                       logger_policies>(
          lgr, schema, data, options, inode_offset, force_consistency_check)) {}

} // namespace dwarfs
