//===----------------------------------------------------------------------===//
//
// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
// See https://llvm.org/LICENSE.txt for license information.
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
//
//===----------------------------------------------------------------------===//
///
/// \file
/// Implementation of the LLVM CAS plugin API, for testing purposes.
///
/// It is backed by \c UnifiedOnDiskCache and can optionally be given a second
/// on-disk path via the \c upstream-path option, which it uses to simulate
/// "uploading"/"downloading" objects to/from a distributed CAS.
///
//===----------------------------------------------------------------------===//

#include "llvm-c/CAS/PluginAPI_functions.h"
#include "llvm/CAS/BuiltinObjectHasher.h"
#include "llvm/CAS/CASID.h"
#include "llvm/CAS/OnDiskKeyValueDB.h"
#include "llvm/CAS/UnifiedOnDiskCache.h"
#include "llvm/Support/CBindingWrapping.h"
#include "llvm/Support/Errc.h"
#include "llvm/Support/Error.h"
#include "llvm/Support/MemoryBuffer.h"
#include "llvm/Support/SHA1.h"
#include "llvm/Support/ThreadPool.h"
#include <mutex>

using namespace llvm;
using namespace llvm::cas;
using namespace llvm::cas::ondisk;

namespace llvm::cas::ondisk {
/// Declared in the private "OnDiskCommon.h"; see \c setSmallMaxMappingSize.
void setMaxMappingSize(uint64_t Size);
} // namespace llvm::cas::ondisk

/// This plugin exists only for testing, and a test process can create many
/// instances of it. Keep the on-disk mappings small so that they stay cheap;
/// the default sizes are measured in gigabytes per instance.
///
/// This has to happen inside the plugin: it links its own copy of LLVMCAS, so
/// the setting the test binary applies to itself does not reach us.
static void setSmallMaxMappingSize() {
  static std::once_flag Flag;
  std::call_once(Flag, [] { setMaxMappingSize(100 * 1024 * 1024); });
}

static char *copyNewMallocString(StringRef Str) {
  char *c_str = (char *)malloc(Str.size() + 1);
  std::uninitialized_copy(Str.begin(), Str.end(), c_str);
  c_str[Str.size()] = '\0';
  return c_str;
}

template <typename ResT>
static ResT reportError(Error &&E, char **error, ResT Result = ResT()) {
  if (error)
    *error = copyNewMallocString(toString(std::move(E)));
  return Result;
}

void llcas_get_plugin_version(unsigned *major, unsigned *minor) {
  *major = LLCAS_VERSION_MAJOR;
  *minor = LLCAS_VERSION_MINOR;
}

void llcas_string_dispose(char *str) { free(str); }

namespace {

struct CancellableState {
  std::atomic<bool> Cancelled{false};
};

struct CancellableWrap {
  std::shared_ptr<CancellableState> State;
};

DEFINE_SIMPLE_CONVERSION_FUNCTIONS(CancellableWrap, llcas_cancellable_t)

} // namespace

void llcas_cancellable_cancel(llcas_cancellable_t c_cancellable) {
  unwrap(c_cancellable)->State->Cancelled = true;
}

void llcas_cancellable_dispose(llcas_cancellable_t c_cancellable) {
  delete unwrap(c_cancellable);
}

namespace {

struct CASPluginOptions {
  std::string OnDiskPath;
  std::string UpstreamPath;
  std::string FirstPrefix;
  std::string SecondPrefix;
  bool SimulateMissingObjects = false;
  bool Logging = true;
  /// If true, \c llcas_cas_validate_if_needed crashes while checking hashes,
  /// for testing recovery from a crashed validation.
  bool CrashOnValidate = false;

  Error setOption(StringRef Name, StringRef Value);
};

DEFINE_SIMPLE_CONVERSION_FUNCTIONS(CASPluginOptions, llcas_cas_options_t)

} // namespace

Error CASPluginOptions::setOption(StringRef Name, StringRef Value) {
  if (Name == "first-prefix")
    FirstPrefix = Value;
  else if (Name == "second-prefix")
    SecondPrefix = Value;
  else if (Name == "upstream-path")
    UpstreamPath = Value;
  else if (Name == "simulate-missing-objects")
    SimulateMissingObjects = true;
  else if (Name == "no-logging")
    Logging = false;
  else if (Name == "crash-on-validate")
    CrashOnValidate = true;
  else
    return createStringError(errc::invalid_argument,
                             Twine("unknown option: ") + Name);
  return Error::success();
}

llcas_cas_options_t llcas_cas_options_create(void) {
  return wrap(new CASPluginOptions());
}

void llcas_cas_options_dispose(llcas_cas_options_t c_opts) {
  delete unwrap(c_opts);
}

void llcas_cas_options_set_ondisk_path(llcas_cas_options_t c_opts,
                                       const char *path) {
  auto &Opts = *unwrap(c_opts);
  Opts.OnDiskPath = path;
}

bool llcas_cas_options_set_option(llcas_cas_options_t c_opts, const char *name,
                                  const char *value, char **error) {
  auto &Opts = *unwrap(c_opts);
  if (Error E = Opts.setOption(name, value))
    return reportError(std::move(E), error, true);
  return false;
}

namespace {

using HasherT = SHA1;
using HashType = decltype(HasherT::hash(std::declval<ArrayRef<uint8_t> &>()));

class PluginCASContext : public CASContext {
  void printIDImpl(raw_ostream &OS, const CASID &ID) const final {
    PluginCASContext::printID(ID.getHash(), OS);
  }

public:
  static StringRef getHashName() { return "SHA1"; }
  StringRef getHashSchemaIdentifier() const final {
    static const std::string ID =
        ("llvm.cas.builtin.v2[" + getHashName() + "]").str();
    return ID;
  }

  PluginCASContext() = default;

  static Expected<HashType> parseID(StringRef Reference) {
    if (!Reference.consume_front("llvmcas://"))
      return createStringError(
          std::make_error_code(std::errc::invalid_argument),
          "invalid cas-id '" + Reference + "'");

    if (Reference.size() != 2 * sizeof(HashType))
      return createStringError(
          std::make_error_code(std::errc::invalid_argument),
          "wrong size for cas-id hash '" + Reference + "'");

    std::string Binary;
    if (!tryGetFromHex(Reference, Binary))
      return createStringError(
          std::make_error_code(std::errc::invalid_argument),
          "invalid hash in cas-id '" + Reference + "'");

    assert(Binary.size() == sizeof(HashType));
    HashType Digest;
    llvm::copy(Binary, Digest.data());
    return Digest;
  }

  static void printID(ArrayRef<uint8_t> Digest, raw_ostream &OS) {
    SmallString<64> Hash;
    toHex(Digest, /*LowerCase=*/true, Hash);
    OS << "llvmcas://" << Hash;
  }
};

struct CASWrapper {
  std::string FirstPrefix;
  std::string SecondPrefix;
  /// If true, asynchronous "download" of an object will treat it as missing.
  bool SimulateMissingObjects = false;
  bool Logging = true;
  std::unique_ptr<UnifiedOnDiskCache> DB;
  /// Used for testing the \c globally parameter of action cache APIs. Simulates
  /// "uploading"/"downloading" objects from/to the primary on-disk path.
  std::unique_ptr<UnifiedOnDiskCache> UpstreamDB;
  DefaultThreadPool Pool{llvm::hardware_concurrency()};

  std::mutex Lock{};

  /// Check if the object is contained, in the "local" CAS only or "globally".
  bool containsObject(ObjectID ID, bool Globally);

  /// Load the object, potentially "downloading" it from upstream.
  Expected<std::optional<ondisk::ObjectHandle>> loadObject(ObjectID ID);

  /// "Uploads" a key and the associated full node graph.
  Error upstreamKey(ArrayRef<uint8_t> Key, ObjectID Value);

  /// "Downloads" the ID associated with the key but not the node data. The node
  /// itself and the rest of the nodes in the graph will be "downloaded" lazily
  /// as they are visited.
  Expected<std::optional<ObjectID>> downstreamKey(ArrayRef<uint8_t> Key);

  /// Synchronized access to \c llvm::errs().
  void syncErrs(llvm::function_ref<void(raw_ostream &OS)> Fn) {
    if (!Logging) {
      // Ignore log output.
      SmallString<32> Buf;
      raw_svector_ostream OS(Buf);
      Fn(OS);
      return;
    }
    std::unique_lock<std::mutex> LockGuard(Lock);
    Fn(errs());
    errs().flush();
  }

private:
  /// "Uploads" the full object node graph.
  Expected<ObjectID> upstreamNode(ObjectID Node);
  /// "Downloads" only a single object node. The rest of the nodes in the graph
  /// will be "downloaded" lazily as they are visited.
  Expected<ObjectID> downstreamNode(ObjectID Node);
};

DEFINE_SIMPLE_CONVERSION_FUNCTIONS(CASWrapper, llcas_cas_t)

} // namespace

bool CASWrapper::containsObject(ObjectID ID, bool Globally) {
  if (DB->getGraphDB().containsObject(ID))
    return true;
  if (!Globally || !UpstreamDB)
    return false;

  auto UpstreamID = expectedToOptional(
      UpstreamDB->getGraphDB().getReference(DB->getGraphDB().getDigest(ID)));

  if (!UpstreamID)
    return false;

  return UpstreamDB->getGraphDB().containsObject(*UpstreamID);
}

Expected<std::optional<ondisk::ObjectHandle>>
CASWrapper::loadObject(ObjectID ID) {
  std::optional<ondisk::ObjectHandle> Obj;
  if (Error E = DB->getGraphDB().load(ID).moveInto(Obj))
    return std::move(E);
  if (Obj)
    return Obj;
  if (!UpstreamDB)
    return std::nullopt;

  // Try "downloading" the node from upstream.
  auto UpstreamID =
      UpstreamDB->getGraphDB().getReference(DB->getGraphDB().getDigest(ID));
  if (!UpstreamID)
    return UpstreamID.takeError();
  std::optional<ObjectID> Ret;
  if (Error E = downstreamNode(*UpstreamID).moveInto(Ret))
    return std::move(E);
  return DB->getGraphDB().load(ID);
}

/// Imports a single object node.
static Expected<ObjectID> importNode(ObjectID FromID, OnDiskGraphDB &FromDB,
                                     OnDiskGraphDB &ToDB) {
  auto ToID = ToDB.getReference(FromDB.getDigest(FromID));
  if (!ToID)
    return ToID.takeError();
  if (ToDB.containsObject(*ToID))
    return ToID;

  std::optional<ondisk::ObjectHandle> FromH;
  if (Error E = FromDB.load(FromID).moveInto(FromH))
    return std::move(E);
  if (!FromH)
    return ToID;

  auto Data = FromDB.getObjectData(*FromH);
  auto FromRefs = FromDB.getObjectRefs(*FromH);
  SmallVector<ObjectID> Refs;
  for (ObjectID FromRef : FromRefs) {
    auto Ref = ToDB.getReference(FromDB.getDigest(FromRef));
    if (!Ref)
      return Ref.takeError();
    Refs.push_back(*Ref);
  }

  if (Error E = ToDB.store(*ToID, Refs, Data))
    return std::move(E);
  return ToID;
}

Expected<ObjectID> CASWrapper::upstreamNode(ObjectID Node) {
  OnDiskGraphDB &FromDB = DB->getGraphDB();
  OnDiskGraphDB &ToDB = UpstreamDB->getGraphDB();

  std::optional<ondisk::ObjectHandle> FromH;
  if (Error E = FromDB.load(Node).moveInto(FromH))
    return std::move(E);
  if (!FromH)
    return createStringError(errc::invalid_argument, "node doesn't exist");

  for (ObjectID Ref : FromDB.getObjectRefs(*FromH)) {
    std::optional<ObjectID> ID;
    if (Error E = upstreamNode(Ref).moveInto(ID))
      return std::move(E);
  }

  return importNode(Node, FromDB, ToDB);
}

Expected<ObjectID> CASWrapper::downstreamNode(ObjectID Node) {
  OnDiskGraphDB &FromDB = UpstreamDB->getGraphDB();
  OnDiskGraphDB &ToDB = DB->getGraphDB();
  return importNode(Node, FromDB, ToDB);
}

static Expected<ObjectID> cachePut(OnDiskKeyValueDB &DB, ArrayRef<uint8_t> Key,
                                   ObjectID ID) {
  auto Value = UnifiedOnDiskCache::getValueFromObjectID(ID);
  auto Result = DB.put(Key, Value);
  if (!Result)
    return Result.takeError();
  return UnifiedOnDiskCache::getObjectIDFromValue(*Result);
}

static Expected<std::optional<ObjectID>> cacheGet(OnDiskKeyValueDB &DB,
                                                  ArrayRef<uint8_t> Key) {
  auto Result = DB.get(Key);
  if (!Result)
    return Result.takeError();
  if (!*Result)
    return std::nullopt;
  return UnifiedOnDiskCache::getObjectIDFromValue(**Result);
}

Error CASWrapper::upstreamKey(ArrayRef<uint8_t> Key, ObjectID Value) {
  if (!UpstreamDB)
    return Error::success();
  Expected<ObjectID> UpstreamVal = upstreamNode(Value);
  if (!UpstreamVal)
    return UpstreamVal.takeError();
  Expected<ObjectID> PutValue =
      cachePut(UpstreamDB->getKeyValueDB(), Key, *UpstreamVal);
  if (!PutValue)
    return PutValue.takeError();
  assert(*PutValue == *UpstreamVal);
  return Error::success();
}

Expected<std::optional<ObjectID>>
CASWrapper::downstreamKey(ArrayRef<uint8_t> Key) {
  if (!UpstreamDB)
    return std::nullopt;
  std::optional<ObjectID> UpstreamValue;
  if (Error E =
          cacheGet(UpstreamDB->getKeyValueDB(), Key).moveInto(UpstreamValue))
    return std::move(E);
  if (!UpstreamValue)
    return std::nullopt;

  auto Value = DB->getGraphDB().getReference(
      UpstreamDB->getGraphDB().getDigest(*UpstreamValue));
  if (!Value)
    return Value.takeError();
  Expected<ObjectID> PutValue = cachePut(DB->getKeyValueDB(), Key, *Value);
  if (!PutValue)
    return PutValue.takeError();
  assert(*PutValue == *Value);
  return PutValue;
}

llcas_cas_t llcas_cas_create(llcas_cas_options_t c_opts, char **error) {
  auto &Opts = *unwrap(c_opts);
  setSmallMaxMappingSize();
  Expected<std::unique_ptr<UnifiedOnDiskCache>> DB = UnifiedOnDiskCache::open(
      Opts.OnDiskPath, /*SizeLimit=*/std::nullopt,
      PluginCASContext::getHashName(), sizeof(HashType));
  if (!DB)
    return reportError<llcas_cas_t>(DB.takeError(), error);

  std::unique_ptr<UnifiedOnDiskCache> UpstreamDB;
  if (!Opts.UpstreamPath.empty()) {
    if (Error E = UnifiedOnDiskCache::open(
                      Opts.UpstreamPath, /*SizeLimit=*/std::nullopt,
                      PluginCASContext::getHashName(), sizeof(HashType))
                      .moveInto(UpstreamDB))
      return reportError<llcas_cas_t>(std::move(E), error);
  }

  return wrap(new CASWrapper{Opts.FirstPrefix, Opts.SecondPrefix,
                             Opts.SimulateMissingObjects, Opts.Logging,
                             std::move(*DB), std::move(UpstreamDB)});
}

void llcas_cas_dispose(llcas_cas_t c_cas) { delete unwrap(c_cas); }

int64_t llcas_cas_get_ondisk_size(llcas_cas_t c_cas, char **error) {
  return unwrap(c_cas)->DB->getStorageSize();
}

bool llcas_cas_set_ondisk_size_limit(llcas_cas_t c_cas, int64_t size_limit,
                                     char **error) {
  std::optional<uint64_t> SizeLimit;
  if (size_limit < 0) {
    return reportError(
        llvm::createStringError(
            llvm::inconvertibleErrorCode(),
            "invalid size limit passed to llcas_cas_set_ondisk_size_limit"),
        error, true);
  }
  if (size_limit > 0) {
    SizeLimit = size_limit;
  }
  unwrap(c_cas)->DB->setSizeLimit(SizeLimit);
  return false;
}

bool llcas_cas_prune_ondisk_data(llcas_cas_t c_cas, char **error) {
  if (Error E = unwrap(c_cas)->DB->collectGarbage())
    return reportError(std::move(E), error, true);
  return false;
}

static void hashObject(ArrayRef<ArrayRef<uint8_t>> Refs, ArrayRef<char> Data,
                       SmallVectorImpl<uint8_t> &Result) {
  HashType Digest = BuiltinObjectHasher<HasherT>::hashObject(Refs, Data);
  Result.assign(Digest.begin(), Digest.end());
}

bool llcas_cas_validate(llcas_cas_t c_cas, bool check_hash, char **error) {
  if (Error E =
          unwrap(c_cas)->DB->getGraphDB().validate(check_hash, hashObject))
    return reportError(std::move(E), error, true);
  return false;
}

static llcas_validation_result_t
toValidationResult(Expected<ValidationResult> Result, char **error) {
  if (!Result)
    return reportError(Result.takeError(), error,
                       LLCAS_VALIDATION_RESULT_ERROR);
  switch (*Result) {
  case ValidationResult::Valid:
    return LLCAS_VALIDATION_RESULT_VALID;
  case ValidationResult::Recovered:
    return LLCAS_VALIDATION_RESULT_RECOVERED;
  case ValidationResult::Skipped:
    return LLCAS_VALIDATION_RESULT_SKIPPED;
  }
  llvm_unreachable("unknown ValidationResult value");
}

llcas_validation_result_t
llcas_cas_validate_if_needed(llcas_cas_options_t c_opts, bool check_hash,
                             bool force, char **error) {
  auto &Opts = *unwrap(c_opts);
  setSmallMaxMappingSize();
  auto HashFn = [&](ArrayRef<ArrayRef<uint8_t>> Refs, ArrayRef<char> Data,
                    SmallVectorImpl<uint8_t> &Result) {
    if (Opts.CrashOnValidate)
      abort();
    hashObject(Refs, Data, Result);
  };
  return toValidationResult(UnifiedOnDiskCache::validateIfNeeded(
                                Opts.OnDiskPath,
                                PluginCASContext::getHashName(),
                                sizeof(HashType), check_hash, HashFn, force),
                            error);
}

llcas_validation_result_t
llcas_cas_recover_ondisk_data(llcas_cas_options_t c_opts, char **error) {
  auto &Opts = *unwrap(c_opts);
  return toValidationResult(UnifiedOnDiskCache::recover(Opts.OnDiskPath),
                            error);
}

void llcas_cas_options_set_client_version(llcas_cas_options_t, unsigned major,
                                          unsigned minor) {
  // Ignore for now.
}

char *llcas_cas_get_hash_schema_name(llcas_cas_t) {
  // Using same name as builtin CAS so that it's interchangeable for testing
  // purposes.
  return copyNewMallocString("llvm.cas.builtin.v2[BLAKE3]");
}

unsigned llcas_digest_parse(llcas_cas_t c_cas, const char *printed_digest,
                            uint8_t *bytes, size_t bytes_size, char **error) {
  auto &Wrapper = *unwrap(c_cas);
  if (bytes_size < sizeof(HashType))
    return sizeof(HashType);

  StringRef PrintedDigest = printed_digest;
  bool Consumed = PrintedDigest.consume_front(Wrapper.FirstPrefix);
  assert(Consumed);
  (void)Consumed;
  Consumed = PrintedDigest.consume_front(Wrapper.SecondPrefix);
  assert(Consumed);
  (void)Consumed;

  Expected<HashType> Digest = PluginCASContext::parseID(PrintedDigest);
  if (!Digest)
    return reportError(Digest.takeError(), error, 0);
  std::uninitialized_copy(Digest->begin(), Digest->end(), bytes);
  return Digest->size();
}

bool llcas_digest_print(llcas_cas_t c_cas, llcas_digest_t c_digest,
                        char **printed_id, char **error) {
  auto &Wrapper = *unwrap(c_cas);
  SmallString<74> PrintDigest;
  raw_svector_ostream OS(PrintDigest);
  // Include these for testing purposes.
  OS << Wrapper.FirstPrefix << Wrapper.SecondPrefix;
  PluginCASContext::printID(ArrayRef(c_digest.data, c_digest.size), OS);
  *printed_id = copyNewMallocString(PrintDigest);
  return false;
}

bool llcas_cas_get_objectid(llcas_cas_t c_cas, llcas_digest_t c_digest,
                            llcas_objectid_t *c_id_p, char **error) {
  auto &CAS = unwrap(c_cas)->DB->getGraphDB();
  auto ID = CAS.getReference(ArrayRef(c_digest.data, c_digest.size));
  if (!ID)
    return reportError(ID.takeError(), error, true);

  *c_id_p = llcas_objectid_t{ID->getOpaqueData()};
  return false;
}

llcas_digest_t llcas_objectid_get_digest(llcas_cas_t c_cas,
                                         llcas_objectid_t c_id) {
  auto &CAS = unwrap(c_cas)->DB->getGraphDB();
  ObjectID ID = ObjectID::fromOpaqueData(c_id.opaque);
  ArrayRef<uint8_t> Digest = CAS.getDigest(ID);
  return llcas_digest_t{Digest.data(), Digest.size()};
}

llcas_lookup_result_t llcas_cas_contains_object(llcas_cas_t c_cas,
                                                llcas_objectid_t c_id,
                                                bool globally, char **error) {
  ObjectID ID = ObjectID::fromOpaqueData(c_id.opaque);
  return unwrap(c_cas)->containsObject(ID, globally)
             ? LLCAS_LOOKUP_RESULT_SUCCESS
             : LLCAS_LOOKUP_RESULT_NOTFOUND;
}

llcas_lookup_result_t llcas_cas_load_object(llcas_cas_t c_cas,
                                            llcas_objectid_t c_id,
                                            llcas_loaded_object_t *c_obj_p,
                                            char **error) {
  ObjectID ID = ObjectID::fromOpaqueData(c_id.opaque);
  Expected<std::optional<ondisk::ObjectHandle>> ObjOpt =
      unwrap(c_cas)->loadObject(ID);
  if (!ObjOpt)
    return reportError(ObjOpt.takeError(), error, LLCAS_LOOKUP_RESULT_ERROR);
  if (!*ObjOpt)
    return LLCAS_LOOKUP_RESULT_NOTFOUND;

  ondisk::ObjectHandle Obj = **ObjOpt;
  *c_obj_p = llcas_loaded_object_t{Obj.getOpaqueData()};
  return LLCAS_LOOKUP_RESULT_SUCCESS;
}

void llcas_cas_load_object_async(llcas_cas_t c_cas, llcas_objectid_t c_id,
                                 void *ctx_cb, llcas_cas_load_object_cb cb,
                                 llcas_cancellable_t *c_cancellable) {
  auto CancelState = std::make_shared<CancellableState>();
  if (c_cancellable) {
    *c_cancellable = wrap(new CancellableWrap{CancelState});
  }

  std::string PrintedDigest;
  {
    llcas_digest_t c_digest = llcas_objectid_get_digest(c_cas, c_id);
    char *printed_id;
    char *c_err;
    bool failed = llcas_digest_print(c_cas, c_digest, &printed_id, &c_err);
    if (failed)
      report_fatal_error(Twine("digest printing failed: ") + c_err);
    PrintedDigest = printed_id;
    llcas_string_dispose(printed_id);
  }

  auto passObject = [ctx_cb,
                     cb](Expected<std::optional<ondisk::ObjectHandle>> Obj) {
    if (!Obj) {
      cb(ctx_cb, LLCAS_LOOKUP_RESULT_ERROR, llcas_loaded_object_t(),
         copyNewMallocString(toString(Obj.takeError())));
    } else if (!*Obj) {
      cb(ctx_cb, LLCAS_LOOKUP_RESULT_NOTFOUND, llcas_loaded_object_t(),
         nullptr);
    } else {
      cb(ctx_cb, LLCAS_LOOKUP_RESULT_SUCCESS,
         llcas_loaded_object_t{(*Obj)->getOpaqueData()}, nullptr);
    }
  };

  auto &CAS = unwrap(c_cas)->DB->getGraphDB();
  ObjectID ID = ObjectID::fromOpaqueData(c_id.opaque);
  if (CAS.containsObject(ID)) {
    unwrap(c_cas)->syncErrs([&](raw_ostream &OS) {
      OS << "load_object_async existing: " << PrintedDigest << '\n';
    });
    return passObject(unwrap(c_cas)->loadObject(ID));
  }

  if (!unwrap(c_cas)->UpstreamDB)
    return passObject(std::nullopt);

  // Try "downloading" the node from upstream.

  unwrap(c_cas)->syncErrs([&](raw_ostream &OS) {
    OS << "load_object_async downstream begin: " << PrintedDigest << '\n';
  });
  unwrap(c_cas)->Pool.async([=] {
#if LLVM_ENABLE_THREADS
    // Wait a bit for the caller to proceed.
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
#endif
    auto &Wrap = *unwrap(c_cas);
    if (CancelState->Cancelled) {
      Wrap.syncErrs([&](raw_ostream &OS) {
        OS << "load_object_async cancelled: " << PrintedDigest << '\n';
      });
      return passObject(std::nullopt);
    }
    Wrap.syncErrs([&](raw_ostream &OS) {
      OS << "load_object_async downstream end: " << PrintedDigest << '\n';
    });
    if (Wrap.SimulateMissingObjects)
      return passObject(std::nullopt);
    passObject(Wrap.loadObject(ID));
  });
}

bool llcas_cas_store_object(llcas_cas_t c_cas, llcas_data_t c_data,
                            const llcas_objectid_t *c_refs, size_t c_refs_count,
                            llcas_objectid_t *c_id_p, char **error) {
  auto &CAS = unwrap(c_cas)->DB->getGraphDB();
  SmallVector<ObjectID, 64> Refs;
  Refs.reserve(c_refs_count);
  for (unsigned I = 0; I != c_refs_count; ++I) {
    Refs.push_back(ObjectID::fromOpaqueData(c_refs[I].opaque));
  }
  ArrayRef Data((const char *)c_data.data, c_data.size);

  SmallVector<ArrayRef<uint8_t>, 8> RefHashes;
  RefHashes.reserve(c_refs_count);
  for (ObjectID Ref : Refs)
    RefHashes.push_back(CAS.getDigest(Ref));
  HashType Digest = BuiltinObjectHasher<HasherT>::hashObject(RefHashes, Data);
  auto StoredID = CAS.getReference(Digest);
  if (!StoredID)
    return reportError(StoredID.takeError(), error, true);

  if (Error E = CAS.store(*StoredID, Refs, Data))
    return reportError(std::move(E), error, true);
  *c_id_p = llcas_objectid_t{StoredID->getOpaqueData()};
  return false;
}

llcas_data_t llcas_loaded_object_get_data(llcas_cas_t c_cas,
                                          llcas_loaded_object_t c_obj) {
  auto &CAS = unwrap(c_cas)->DB->getGraphDB();
  ondisk::ObjectHandle Obj = ondisk::ObjectHandle(c_obj.opaque);
  auto Data = CAS.getObjectData(Obj);
  return llcas_data_t{Data.data(), Data.size()};
}

/// The \c MemoryBuffer objects handed out by
/// \c llcas_loaded_object_get_standalone_data, keyed by the bytes the C API
/// reports, so \c llcas_standalone_data_dispose can find the owner again. The
/// C API passes back only the buffer, and these outlive the \c llcas_cas_t, so
/// they cannot be tracked on it.
/// Intentionally leaked, since a buffer may be disposed of during static
/// destruction, after a non-leaked map would already be gone.
static std::mutex StandaloneBuffersLock;
static auto *StandaloneBuffers =
    new DenseMap<const void *, std::unique_ptr<MemoryBuffer>>();

llcas_data_t
llcas_loaded_object_get_standalone_data(llcas_cas_t c_cas,
                                        llcas_loaded_object_t c_obj) {
  auto &CAS = unwrap(c_cas)->DB->getGraphDB();
  ondisk::ObjectHandle Obj = ondisk::ObjectHandle(c_obj.opaque);
  // The underlying database already knows how to produce a buffer that does
  // not reference it, so use that rather than copying the data again. The
  // plugin API requires a nul terminator, which costs a copy for the objects
  // whose file has no byte to spare for one.
  std::unique_ptr<MemoryBuffer> Buffer = CAS.getStandaloneMemoryBuffer(
      Obj, /*Name=*/"", /*RequiresNullTerminator=*/true);
  const char *Data = Buffer->getBufferStart();
  size_t Size = Buffer->getBufferSize();
  {
    std::lock_guard<std::mutex> Lock(StandaloneBuffersLock);
    (*StandaloneBuffers)[Data] = std::move(Buffer);
  }
  return llcas_data_t{Data, Size};
}

void llcas_standalone_data_dispose(llcas_data_t c_data) {
  std::lock_guard<std::mutex> Lock(StandaloneBuffersLock);
  StandaloneBuffers->erase(c_data.data);
}

llcas_object_refs_t llcas_loaded_object_get_refs(llcas_cas_t c_cas,
                                                 llcas_loaded_object_t c_obj) {
  auto &CAS = unwrap(c_cas)->DB->getGraphDB();
  ondisk::ObjectHandle Obj = ondisk::ObjectHandle(c_obj.opaque);
  auto Refs = CAS.getObjectRefs(Obj);
  return llcas_object_refs_t{Refs.begin().getOpaqueData(),
                             Refs.end().getOpaqueData()};
}

size_t llcas_object_refs_get_count(llcas_cas_t c_cas,
                                   llcas_object_refs_t c_refs) {
  auto B = object_refs_iterator::fromOpaqueData(c_refs.opaque_b);
  auto E = object_refs_iterator::fromOpaqueData(c_refs.opaque_e);
  return E - B;
}

llcas_objectid_t llcas_object_refs_get_id(llcas_cas_t c_cas,
                                          llcas_object_refs_t c_refs,
                                          size_t index) {
  auto RefsI = object_refs_iterator::fromOpaqueData(c_refs.opaque_b);
  ObjectID Ref = *(RefsI + index);
  return llcas_objectid_t{Ref.getOpaqueData()};
}

llcas_lookup_result_t
llcas_actioncache_get_for_digest(llcas_cas_t c_cas, llcas_digest_t c_key,
                                 llcas_objectid_t *p_value, bool globally,
                                 char **error) {
  auto &Wrap = *unwrap(c_cas);
  auto &DB = *Wrap.DB;
  ArrayRef Key(c_key.data, c_key.size);
  std::optional<ObjectID> Value;
  if (Error E = cacheGet(DB.getKeyValueDB(), Key).moveInto(Value))
    return reportError(std::move(E), error, LLCAS_LOOKUP_RESULT_ERROR);
  if (!Value) {
    if (!globally)
      return LLCAS_LOOKUP_RESULT_NOTFOUND;

    if (Error E = Wrap.downstreamKey(Key).moveInto(Value))
      return reportError(std::move(E), error, LLCAS_LOOKUP_RESULT_ERROR);
    if (!Value)
      return LLCAS_LOOKUP_RESULT_NOTFOUND;
  }
  *p_value = llcas_objectid_t{Value->getOpaqueData()};
  return LLCAS_LOOKUP_RESULT_SUCCESS;
}

void llcas_actioncache_get_for_digest_async(
    llcas_cas_t c_cas, llcas_digest_t c_key, bool globally, void *ctx_cb,
    llcas_actioncache_get_cb cb, llcas_cancellable_t *c_cancellable) {
  auto CancelState = std::make_shared<CancellableState>();
  if (c_cancellable) {
    *c_cancellable = wrap(new CancellableWrap{CancelState});
  }
  bool IsCancellable = c_cancellable != nullptr;

  ArrayRef Key(c_key.data, c_key.size);
  SmallVector<uint8_t, 32> KeyBuf(Key);

  unwrap(c_cas)->Pool.async([=] {
    if (IsCancellable) {
#if LLVM_ENABLE_THREADS
      // Wait a bit for the caller to have a chance to cancel.
      std::this_thread::sleep_for(std::chrono::milliseconds(50));
#endif
    }
    auto &Wrap = *unwrap(c_cas);
    if (CancelState->Cancelled) {
      Wrap.syncErrs([&](raw_ostream &OS) {
        OS << "actioncache_get_for_digest_async cancelled\n";
      });
      return cb(ctx_cb, LLCAS_LOOKUP_RESULT_NOTFOUND, llcas_objectid_t(),
                nullptr);
    }
    llcas_objectid_t c_value;
    char *c_err;
    llcas_lookup_result_t result = llcas_actioncache_get_for_digest(
        c_cas, llcas_digest_t{KeyBuf.data(), KeyBuf.size()}, &c_value, globally,
        &c_err);
    cb(ctx_cb, result, c_value, c_err);
  });
}

bool llcas_actioncache_put_for_digest(llcas_cas_t c_cas, llcas_digest_t c_key,
                                      llcas_objectid_t c_value, bool globally,
                                      char **error) {
  auto &Wrap = *unwrap(c_cas);
  auto &DB = *Wrap.DB;
  ObjectID Value = ObjectID::fromOpaqueData(c_value.opaque);
  ArrayRef Key(c_key.data, c_key.size);
  Expected<ObjectID> Ret = cachePut(DB.getKeyValueDB(), Key, Value);
  if (!Ret)
    return reportError(Ret.takeError(), error, true);
  if (*Ret != Value)
    return reportError(
        createStringError(errc::invalid_argument, "cache poisoned"), error,
        true);

  if (globally) {
    if (Error E = Wrap.upstreamKey(Key, Value))
      return reportError(std::move(E), error, true);
  }

  return false;
}

void llcas_actioncache_put_for_digest_async(
    llcas_cas_t c_cas, llcas_digest_t c_key, llcas_objectid_t c_value,
    bool globally, void *ctx_cb, llcas_actioncache_put_cb cb,
    llcas_cancellable_t *c_cancellable) {
  auto CancelState = std::make_shared<CancellableState>();
  if (c_cancellable) {
    *c_cancellable = wrap(new CancellableWrap{CancelState});
  }
  bool IsCancellable = c_cancellable != nullptr;

  ArrayRef Key(c_key.data, c_key.size);
  SmallVector<uint8_t, 32> KeyBuf(Key);

  unwrap(c_cas)->Pool.async([=] {
    if (IsCancellable) {
#if LLVM_ENABLE_THREADS
      // Wait a bit for the caller to have a chance to cancel.
      std::this_thread::sleep_for(std::chrono::milliseconds(50));
#endif
    }
    auto &Wrap = *unwrap(c_cas);
    if (CancelState->Cancelled) {
      Wrap.syncErrs([&](raw_ostream &OS) {
        OS << "actioncache_put_for_digest_async cancelled\n";
      });
      return cb(ctx_cb, false, nullptr);
    }
    char *c_err;
    bool failed = llcas_actioncache_put_for_digest(
        c_cas, llcas_digest_t{KeyBuf.data(), KeyBuf.size()}, c_value, globally,
        &c_err);
    cb(ctx_cb, failed, c_err);
  });
}

bool llcas_actioncache_validate(llcas_cas_t c_cas, char **error) {
  if (Error E = unwrap(c_cas)->DB->validateActionCache())
    return reportError(std::move(E), error, true);
  return false;
}