/*
 * Copyright 2019 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "Firestore/core/src/local/leveldb_persistence.h"

#include <limits>
#include <utility>

#include "Firestore/core/src/core/database_info.h"
#include "Firestore/core/src/credentials/user.h"
#include "Firestore/core/src/local/leveldb_key.h"
#include "Firestore/core/src/local/leveldb_lru_reference_delegate.h"
#include "Firestore/core/src/local/leveldb_migrations.h"
#include "Firestore/core/src/local/leveldb_opener.h"
#include "Firestore/core/src/local/leveldb_util.h"
#include "Firestore/core/src/local/listen_sequence.h"
#include "Firestore/core/src/local/lru_garbage_collector.h"
#include "Firestore/core/src/local/reference_delegate.h"
#include "Firestore/core/src/local/sizer.h"
#include "Firestore/core/src/util/filesystem.h"
#include "Firestore/core/src/util/hard_assert.h"
#include "Firestore/core/src/util/log.h"
#include "Firestore/core/src/util/string_util.h"
#include "absl/memory/memory.h"
#include "absl/strings/match.h"

namespace firebase {
namespace firestore {
namespace local {
namespace {

using credentials::User;
using leveldb::DB;
using model::ListenSequenceNumber;
using util::Filesystem;
using util::Path;
using util::Status;
using util::StatusOr;
using util::StringFormat;

/**
 * Finds all user ids in the database based on the existence of a mutation
 * queue.
 */
std::set<std::string> CollectUserSet(LevelDbTransaction* transaction) {
  std::set<std::string> result;

  std::string table_prefix = LevelDbMutationKey::KeyPrefix();
  auto it = transaction->NewIterator();
  it->Seek(table_prefix);

  LevelDbMutationKey row_key;
  while (it->Valid() && absl::StartsWith(it->key(), table_prefix) &&
         row_key.Decode(it->key())) {
    result.insert(row_key.user_id());

    auto user_end = LevelDbMutationKey::KeyPrefix(row_key.user_id());
    user_end = util::PrefixSuccessor(user_end);
    it->Seek(user_end);
  }
  return result;
}

}  // namespace

StatusOr<std::unique_ptr<LevelDbPersistence>> LevelDbPersistence::Create(
    util::Path dir,
    LevelDbMigrations::SchemaVersion version,
    LocalSerializer serializer,
    const LruParams& lru_params) {
  auto* fs = Filesystem::Default();
  Status status = EnsureDirectory(dir);
  if (!status.ok()) return status;

  status = fs->ExcludeFromBackups(dir);
  if (!status.ok()) return status;

  StatusOr<std::unique_ptr<DB>> created = OpenDb(dir);
  if (!created.ok()) return created.status();

  std::unique_ptr<DB> db = std::move(created).ValueOrDie();
  LevelDbMigrations::RunMigrations(db.get(), version, serializer);

  LevelDbTransaction transaction(db.get(), "Start LevelDB");
  std::set<std::string> users = CollectUserSet(&transaction);
  transaction.Commit();

  // Explicit conversion is required to allow the StatusOr to be created.
  std::unique_ptr<LevelDbPersistence> result(
      new LevelDbPersistence(std::move(db), std::move(dir), std::move(users),
                             std::move(serializer), lru_params));
  return {std::move(result)};
}

StatusOr<std::unique_ptr<LevelDbPersistence>> LevelDbPersistence::Create(
    util::Path dir, LocalSerializer serializer, const LruParams& lru_params) {
  return Create(std::move(dir), kSchemaVersion, std::move(serializer),
                lru_params);
}

LevelDbPersistence::LevelDbPersistence(std::unique_ptr<leveldb::DB> db,
                                       util::Path directory,
                                       std::set<std::string> users,
                                       LocalSerializer serializer,
                                       const LruParams& lru_params)
    : db_(std::move(db)),
      directory_(std::move(directory)),
      users_(std::move(users)),
      serializer_(std::move(serializer)) {
  target_cache_ = absl::make_unique<LevelDbTargetCache>(this, &serializer_);
  document_cache_ =
      absl::make_unique<LevelDbRemoteDocumentCache>(this, &serializer_);
  reference_delegate_ =
      absl::make_unique<LevelDbLruReferenceDelegate>(this, lru_params);
  bundle_cache_ = absl::make_unique<LevelDbBundleCache>(this, &serializer_);
  globals_cache_ = absl::make_unique<LevelDbGlobalsCache>(this);

  // TODO(gsoltis): set up a leveldb transaction for these operations.
  target_cache_->Start();
  reference_delegate_->Start();
  started_ = true;
}

// Handle unique_ptrs to forward declarations
LevelDbPersistence::~LevelDbPersistence() = default;

// MARK: - Startup

Status LevelDbPersistence::EnsureDirectory(const Path& dir) {
  auto* fs = Filesystem::Default();
  Status status = fs->RecursivelyCreateDir(dir);
  if (!status.ok()) {
    return Status{Error::kErrorInternal,
                  "Failed to create persistence directory"}
        .CausedBy(status);
  }

  return Status::OK();
}

StatusOr<std::unique_ptr<DB>> LevelDbPersistence::OpenDb(const Path& dir) {
  leveldb::Options options;
  options.create_if_missing = true;

  DB* database = nullptr;
  leveldb::Status status = DB::Open(options, dir.ToUtf8String(), &database);
  if (!status.ok()) {
    return Status{Error::kErrorInternal,
                  StringFormat("Failed to open LevelDB database at %s",
                               dir.ToUtf8String())}
        .CausedBy(ConvertStatus(status));
  }

  return std::unique_ptr<DB>(database);
}

// MARK: - LevelDB utilities

LevelDbTransaction* LevelDbPersistence::current_transaction() {
  HARD_ASSERT(transaction_ != nullptr,
              "Attempting to access transaction before one has started");
  return transaction_.get();
}

util::Status LevelDbPersistence::ClearPersistence(
    const core::DatabaseInfo& database_info) {
  LevelDbOpener opener(database_info);
  StatusOr<Path> maybe_data_dir = opener.LevelDbDataDir();
  HARD_ASSERT(maybe_data_dir.ok(), "Failed to find local LevelDB files: %s",
              maybe_data_dir.status().ToString());
  Path leveldb_dir = std::move(maybe_data_dir).ValueOrDie();

  LOG_DEBUG("Clearing persistence for path: %s", leveldb_dir.ToUtf8String());
  auto* fs = Filesystem::Default();
  return fs->RecursivelyRemove(leveldb_dir);
}

StatusOr<int64_t> LevelDbPersistence::CalculateByteSize() {
  auto* fs = Filesystem::Default();

  // Accumulate the total size in an unsigned integer to avoid undefined
  // behavior on overflow.
  uint64_t count = 0;
  auto iter = util::DirectoryIterator::Create(directory_);
  for (; iter->Valid(); iter->Next()) {
    StatusOr<int64_t> maybe_size = fs->FileSize(iter->file());
    if (!maybe_size.ok()) {
      return Status::FromCause("Failed to size LevelDB directory",
                               maybe_size.status());
    }

    uint64_t old_count = count;
    int64_t file_size = maybe_size.ValueOrDie();
    count += file_size;

    auto max_signed_value =
        static_cast<uint64_t>(std::numeric_limits<int64_t>::max());
    if (count < old_count || count > max_signed_value) {
      return Status(Error::kErrorOutOfRange,
                    "Failed to size LevelDB: count overflowed");
    }
  }

  if (!iter->status().ok()) {
    return Status::FromCause("Failed to iterate over LevelDB files",
                             iter->status());
  }
  return static_cast<int64_t>(count);
}

// MARK: - Persistence

model::ListenSequenceNumber LevelDbPersistence::current_sequence_number()
    const {
  return reference_delegate_->current_sequence_number();
}

void LevelDbPersistence::Shutdown() {
  HARD_ASSERT(started_, "LevelDbPersistence shutdown without start!");
  started_ = false;
  db_.reset();
}

LevelDbMutationQueue* LevelDbPersistence::GetMutationQueue(
    const credentials::User& user, IndexManager* manager) {
  users_.insert(user.uid());
  if (mutation_queues_.find(user.uid()) == mutation_queues_.end()) {
    mutation_queues_.insert(
        {user.uid(),
         absl::make_unique<LevelDbMutationQueue>(
             user, this, dynamic_cast<LevelDbIndexManager*>(manager),
             &serializer_)});
  }
  return mutation_queues_[user.uid()].get();
}

LevelDbTargetCache* LevelDbPersistence::target_cache() {
  return target_cache_.get();
}

LevelDbGlobalsCache* LevelDbPersistence::globals_cache() {
  return globals_cache_.get();
}

LevelDbRemoteDocumentCache* LevelDbPersistence::remote_document_cache() {
  return document_cache_.get();
}

LevelDbIndexManager* LevelDbPersistence::GetIndexManager(
    const credentials::User& user) {
  users_.insert(user.uid());
  if (index_managers_.find(user.uid()) == index_managers_.end()) {
    index_managers_.insert({user.uid(), absl::make_unique<LevelDbIndexManager>(
                                            user, this, &serializer_)});
  }
  return index_managers_[user.uid()].get();
}

LevelDbLruReferenceDelegate* LevelDbPersistence::reference_delegate() {
  return reference_delegate_.get();
}

LevelDbBundleCache* LevelDbPersistence::bundle_cache() {
  return bundle_cache_.get();
}

LevelDbDocumentOverlayCache* LevelDbPersistence::GetDocumentOverlayCache(
    const User& user) {
  users_.insert(user.uid());
  if (document_overlay_caches_.find(user.uid()) ==
      document_overlay_caches_.end()) {
    document_overlay_caches_.insert(
        {user.uid(), absl::make_unique<LevelDbDocumentOverlayCache>(
                         user, this, &serializer_)});
  }
  return document_overlay_caches_[user.uid()].get();
}

LevelDbOverlayMigrationManager* LevelDbPersistence::GetOverlayMigrationManager(
    const User& user) {
  if (overlay_migration_managers_.find(user.uid()) ==
      overlay_migration_managers_.end()) {
    overlay_migration_managers_.insert(
        {user.uid(),
         absl::make_unique<LevelDbOverlayMigrationManager>(this, user.uid())});
  }
  return overlay_migration_managers_[user.uid()].get();
}

void LevelDbPersistence::ReleaseOtherUserSpecificComponents(
    const std::string& target_uid) {
  for (const auto& uid : users_) {
    if (target_uid != uid) {
      document_overlay_caches_.erase(uid);
      mutation_queues_.erase(uid);
      index_managers_.erase(uid);
      overlay_migration_managers_.erase(uid);
    }
  }
}

void LevelDbPersistence::DeleteAllFieldIndexes() {
  DeleteEverythingWithPrefix("Delete All Index Configuration",
                             LevelDbIndexConfigurationKey::KeyPrefix());

  DeleteEverythingWithPrefix("Delete All Index States",
                             LevelDbIndexStateKey::KeyPrefix());

  DeleteEverythingWithPrefix("Delete All Index Entries",
                             LevelDbIndexEntryKey::KeyPrefix());
}

void LevelDbPersistence::RunInternal(absl::string_view label,
                                     std::function<void()> block) {
  HARD_ASSERT(transaction_ == nullptr,
              "Starting a transaction while one is already in progress");

  transaction_ = absl::make_unique<LevelDbTransaction>(db_.get(), label);
  reference_delegate_->OnTransactionStarted(label);

  block();

  reference_delegate_->OnTransactionCommitted();
  transaction_->Commit();
  transaction_.reset();
}

leveldb::ReadOptions StandardReadOptions() {
  // For now this is paranoid, but perhaps disable that in production builds.
  leveldb::ReadOptions options;
  options.verify_checksums = true;
  return options;
}

void LevelDbPersistence::DeleteEverythingWithPrefix(absl::string_view label,
                                                    const std::string& prefix) {
  bool more_deletes = true;

  auto fun = [&]() {
    more_deletes = false;

    auto it = transaction_->NewIterator();
    for (it->Seek(prefix); it->Valid() && absl::StartsWith(it->key(), prefix);
         it->Next()) {
      if (transaction_->changed_keys() >= kMaxOperationPerTransaction) {
        more_deletes = true;
        break;
      }
      transaction_->Delete(it->key());
    }
  };

  while (more_deletes) {
    RunInternal(label, fun);
  }
}

}  // namespace local
}  // namespace firestore
}  // namespace firebase
