// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include <ctime>
#include <regex>
#include <utility>

#include "./logging.h"
#include "encoder/file_observation_store.h"
#include "encoder/file_observation_store_internal.pb.h"
#include "third_party/protobuf/src/google/protobuf/util/delimited_message_util.h"

namespace cobalt {
namespace encoder {

using tensorflow_statusor::StatusOr;
using util::FileSystem;
using util::Status;
using util::StatusCode;

const char kActiveFileName[] = "in_progress.data";
// The first part of the filename is 13 digits representing the milliseconds
// since the unix epoch. The millisecond timestamp became 13 digits in
// September 2001, and won't be 14 digits until 2286.
//
// The second part of the filename is 7 digits, which is a random number in the
// range 1000000-9999999.
const std::regex kFinalizedFileRegex(R"(\d{13}-\d{7}.data)");

FileObservationStore::FileObservationStore(size_t max_bytes_per_observation,
                                           size_t max_bytes_per_envelope,
                                           size_t max_bytes_total,
                                           std::unique_ptr<FileSystem> fs,
                                           std::string root_directory,
                                           std::string name)
    : ObservationStore(max_bytes_per_observation, max_bytes_per_envelope,
                       max_bytes_total),
      fs_(std::move(fs)),
      root_directory_(std::move(root_directory)),
      active_file_name_(FullPath(kActiveFileName)),
      random_int_(1000000, 9999999),
      name_(std::move(name)),
      num_observations_added_(0) {
  CHECK(fs_);

  // Check if root_directory_ already exists.
  if (!fs_->ListFiles(root_directory_).ok()) {
    // If it doesn't exist, create it here.
    // TODO(zmbush): If MakeDirectory doesn't work, we should fail over to
    // MemoryObservationStore.
    CHECK(fs_->MakeDirectory(root_directory_));
  }

  {
    auto fields = protected_fields_.lock();
    fields->finalized_bytes = 0;

    for (auto file : ListFinalizedFiles()) {
      fields->finalized_bytes +=
          fs_->FileSize(FullPath(file)).ConsumeValueOr(0);
    }

    // If there exists an active file, it likely means that the process
    // terminated unexpectedly last time. In this case, the file should be
    // finalized in order to be Taken from the store.
    //
    // For simplicity's sake, we attempt to the active_file_name_ while ignoring
    // the result. If the operation succeeds, then we rescued the active file.
    // Otherwise, there probably was no active file in the first place.
    FinalizeActiveFile(&fields);
  }
}

ObservationStore::StoreStatus FileObservationStore::AddEncryptedObservation(
    std::unique_ptr<EncryptedMessage> message,
    std::unique_ptr<ObservationMetadata> metadata) {
  auto fields = protected_fields_.lock();

  auto active_file = GetActiveFile(&fields);
  auto metadata_str = metadata->SerializeAsString();

  // "+1" below is for the |scheme| field of EncryptedMessage.
  size_t obs_size = message->ciphertext().size() +
                    message->public_key_fingerprint().size() + 1;
  if (obs_size > max_bytes_per_observation_) {
    LOG(WARNING) << "An observation that was too big was passed in to "
                    "FileObservationStore::AddEncryptedObservation(): "
                 << obs_size;
    return kObservationTooBig;
  }

  VLOG(6) << name_ << ": AddEncryptedObservation() metric=("
          << metadata->customer_id() << "," << metadata->project_id() << ","
          << metadata->metric_id() << "), size=" << obs_size << ".";

  size_t estimated_new_byte_count = active_file->ByteCount() + obs_size;
  if (estimated_new_byte_count > max_bytes_total_) {
    VLOG(4) << name_
            << ": The observation store is full. estimated_new_byte_count="
            << estimated_new_byte_count << " > " << max_bytes_total_ << ".";
    return kStoreFull;
  }

  if (!fields->metadata_written ||
      metadata_str != fields->last_written_metadata) {
    VLOG(5) << name_ << ": Writing observation metadata.";
    ObservationStoreRecord stored_metadata;
    stored_metadata.mutable_meta_data()->Swap(metadata.get());
    if (!google::protobuf::util::SerializeDelimitedToZeroCopyStream(
            stored_metadata, active_file)) {
      LOG(WARNING) << name_ << ": Unable to write metadata to `"
                   << active_file_name_ << "`";
      return kWriteFailed;
    }
    fields->metadata_written = true;
    fields->last_written_metadata = metadata_str;
  }

  ObservationStoreRecord stored_message;
  stored_message.mutable_encrypted_observation()->Swap(message.get());
  if (!google::protobuf::util::SerializeDelimitedToZeroCopyStream(
          stored_message, active_file)) {
    LOG(WARNING) << "Unable to write encrypted_observation to `"
                 << active_file_name_ << "`";
    return kWriteFailed;
  }

  if (active_file->ByteCount() >= (int64_t)max_bytes_per_envelope_) {
    VLOG(4) << name_ << ": In-progress file contains "
            << active_file->ByteCount()
            << " bytes (>= " << max_bytes_per_envelope_ << "). Finalizing it.";

    if (!FinalizeActiveFile(&fields)) {
      LOG(WARNING) << "Unable to finalize `" << active_file_name_;
      return kWriteFailed;
    }
  }

  num_observations_added_++;
  return kOk;
}

bool FileObservationStore::FinalizeActiveFile(
    util::ProtectedFields<Fields>::LockedFieldsPtr *fields) {
  VLOG(6) << name_ << ": FinalizeActiveFile()";
  auto &f = *fields;

  // Close the current file (if it is open).
  f->active_file = nullptr;
  if (f->active_fstream.is_open()) {
    f->active_fstream.close();
  }
  f->metadata_written = false;

  auto filesize = fs_->FileSize(active_file_name_);
  if (filesize.ok()) {
    if (filesize.ConsumeValueOrDie() == 0) {
      // File exists, but is empty. Let's just delete it instead of renaming.
      fs_->Delete(active_file_name_);
      return false;
    }
  } else {
    // if !filesize.ok(), the file likely doesn't even exist.
    return false;
  }

  auto new_name = FullPath(GenerateFinalizedName());
  if (!fs_->Rename(active_file_name_, new_name)) {
    return false;
  }

  f->finalized_bytes += fs_->FileSize(new_name).ConsumeValueOr(0);
  return true;
}

std::string FileObservationStore::GenerateFinalizedName() const {
  auto now = std::chrono::duration_cast<std::chrono::milliseconds>(
                 std::chrono::system_clock::now().time_since_epoch())
                 .count();
  std::stringstream ss;
  ss << now << "-" << random_int_(random_dev_) << ".data";
  return ss.str();
}

std::string FileObservationStore::FullPath(const std::string &filename) const {
  return root_directory_ + "/" + filename;
}

std::string FileObservationStore::FileEnvelopeHolder::FullPath(
    const std::string &filename) const {
  return root_directory_ + "/" + filename;
}

google::protobuf::io::OstreamOutputStream *FileObservationStore::GetActiveFile(
    util::ProtectedFields<Fields>::LockedFieldsPtr *fields) {
  auto &f = *fields;

  if (f->active_file == nullptr) {
    f->active_fstream.open(active_file_name_);
    CHECK(f->active_fstream.is_open()) << "Failed to open " << active_file_name_
                                       << ": " << std::strerror(errno);
    f->active_file.reset(
        new google::protobuf::io::OstreamOutputStream(&f->active_fstream));
  }
  return f->active_file.get();
}

std::vector<std::string> FileObservationStore::ListFinalizedFiles() const {
  auto files = fs_->ListFiles(root_directory_).ConsumeValueOr({});
  std::vector<std::string> retval;
  for (auto file : files) {
    if (std::regex_match(file, kFinalizedFileRegex)) {
      retval.push_back(file);
    }
  }
  return retval;
}

StatusOr<std::string> FileObservationStore::GetOldestFinalizedFile(
    util::ProtectedFields<Fields>::LockedFieldsPtr *fields) {
  auto &f = *fields;

  std::string found_file_name;
  for (auto file : ListFinalizedFiles()) {
    if (f->files_taken.find(file) == f->files_taken.end()) {
      if (found_file_name == "") {
        found_file_name = file;
      } else {
        // We compare the file names to try to find the oldest one. This works
        // because file names are prefixed with the timestamp when they were
        // finalized and that timestamp is always 13 digits long.
        //
        // A lexigraphic order of fixed length number strings is identical to
        // ordering their numerical values.
        auto result = strcmp(file.c_str(), found_file_name.c_str());
        if (result < 0) {
          found_file_name = file;
        }
      }
    }
  }
  if (found_file_name == "") {
    return Status(StatusCode::NOT_FOUND, "No finalized file");
  }
  return found_file_name;
}

std::unique_ptr<ObservationStore::EnvelopeHolder>
FileObservationStore::TakeNextEnvelopeHolder() {
  auto fields = protected_fields_.lock();

  auto oldest_file_name_or = GetOldestFinalizedFile(&fields);
  if (!oldest_file_name_or.ok()) {
    if (!fields->active_file || fields->active_file->ByteCount() == 0) {
      // Active file isn't open or is empty. Return nullptr.
      return nullptr;
    }
    if (!FinalizeActiveFile(&fields)) {
      // Finalizing the active file failed, no envelope to return.
      return nullptr;
    }
    oldest_file_name_or = GetOldestFinalizedFile(&fields);
    if (!oldest_file_name_or.ok()) {
      return nullptr;
    }
  }

  auto oldest_file_name = oldest_file_name_or.ConsumeValueOrDie();
  fields->files_taken.insert(oldest_file_name);
  fields->finalized_bytes -=
      fs_->FileSize(FullPath(oldest_file_name)).ConsumeValueOr(0);
  return std::make_unique<FileEnvelopeHolder>(fs_.get(), root_directory_,
                                              oldest_file_name);
}

void FileObservationStore::ReturnEnvelopeHolder(
    std::unique_ptr<ObservationStore::EnvelopeHolder> envelope) {
  std::unique_ptr<FileObservationStore::FileEnvelopeHolder> env(
      static_cast<FileObservationStore::FileEnvelopeHolder *>(
          envelope.release()));

  auto fields = protected_fields_.lock();
  for (auto file_name : env->file_names()) {
    fields->files_taken.erase(file_name);
    fields->finalized_bytes +=
        fs_->FileSize(FullPath(file_name)).ConsumeValueOr(0);
  }
  env->clear();
}

size_t FileObservationStore::Size() const {
  auto fields = protected_fields_.const_lock();
  auto bytes = fields->finalized_bytes;
  VLOG(4) << name_ << "::Size(): finalized_bytes=" << bytes;
  if (fields->active_file) {
    bytes += fields->active_file->ByteCount();
  } else {
    VLOG(4) << name_ << "::Size(): there is no active file.";
  }
  VLOG(4) << name_ << "::Size(): total_bytes=" << bytes;
  return bytes;
}

bool FileObservationStore::Empty() const { return Size() == 0; }

size_t FileObservationStore::num_observations_added() {
  return num_observations_added_;
}

void FileObservationStore::ResetObservationCounter() {
  num_observations_added_ = 0;
}

void FileObservationStore::Delete() {
  auto files = fs_->ListFiles(root_directory_).ConsumeValueOr({});
  for (auto file : files) {
    fs_->Delete(FullPath(file));
  }
  fs_->Delete(root_directory_);
}

FileObservationStore::FileEnvelopeHolder::~FileEnvelopeHolder() {
  for (auto file_name : file_names_) {
    fs_->Delete(FullPath(file_name));
  }
}

void FileObservationStore::FileEnvelopeHolder::MergeWith(
    std::unique_ptr<EnvelopeHolder> container) {
  std::unique_ptr<FileEnvelopeHolder> file_container(
      static_cast<FileEnvelopeHolder *>(container.release()));

  file_names_.insert(file_container->file_names_.begin(),
                     file_container->file_names_.end());

  file_container->file_names_.clear();

  envelope_.Clear();
  cached_file_size_ = 0;
  envelope_read_ = false;
}

const Envelope &FileObservationStore::FileEnvelopeHolder::GetEnvelope() {
  if (envelope_read_) {
    return envelope_;
  }

  std::string serialized_metadata;
  std::unordered_map<std::string, ObservationBatch *> batch_map;
  ObservationBatch *current_batch;

  ObservationStoreRecord stored;

  for (auto file_name : file_names_) {
    std::ifstream ifs(FullPath(file_name), std::ifstream::in);
    google::protobuf::io::IstreamInputStream iis(&ifs);

    bool clean_eof;
    while (google::protobuf::util::ParseDelimitedFromZeroCopyStream(
        &stored, &iis, &clean_eof)) {
      if (stored.has_meta_data()) {
        std::unique_ptr<ObservationMetadata> current_metadata(
            stored.release_meta_data());
        current_metadata->SerializeToString(&serialized_metadata);

        auto iter = batch_map.find(serialized_metadata);
        if (iter != batch_map.end()) {
          current_batch = iter->second;
        } else {
          current_batch = envelope_.add_batch();
          current_batch->set_allocated_meta_data(current_metadata.release());
          batch_map[serialized_metadata] = current_batch;
        }
      } else if (stored.has_encrypted_observation()) {
        current_batch->add_encrypted_observation()->Swap(
            stored.mutable_encrypted_observation());
      } else {
        clean_eof = false;
        break;
      }
    }

    if (!clean_eof) {
      VLOG(1)
          << "WARNING: Trying to read from `" << file_name
          << "` encountered a corrupted message. Returning the envelope that "
             "has been read so far.";
      break;
    }
  }

  envelope_read_ = true;
  return envelope_;
}

size_t FileObservationStore::FileEnvelopeHolder::Size() {
  if (cached_file_size_ != 0) {
    return cached_file_size_;
  }

  cached_file_size_ = 0;
  for (auto file_name : file_names_) {
    cached_file_size_ += fs_->FileSize(FullPath(file_name)).ConsumeValueOr(0);
  }
  return cached_file_size_;
}

}  // namespace encoder
}  // namespace cobalt