/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 **/

#define __STDC_FORMAT_MACROS

#include "storage/FileManagerPosix.hpp"

#include <fcntl.h>
#include <glob.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>

#include <cerrno>
#include <cinttypes>
#include <cstddef>
#include <cstdio>
#include <cstring>
#include <string>

#include "storage/StorageBlockInfo.hpp"
#include "storage/StorageConstants.hpp"
#include "storage/StorageErrors.hpp"
#include "utility/StringUtil.hpp"

#include "glog/logging.h"

using std::size_t;
using std::sscanf;
using std::strerror;
using std::string;

namespace quickstep {

block_id_counter FileManagerPosix::getMaxUsedBlockCounter(const block_id_domain block_domain) const {
  const string block_domain_str(ToZeroPaddedString(block_domain, kBlockIdDomainLengthInDigits));

  string glob_pattern(storage_path_);
  glob_pattern.append("qsblk_");
  glob_pattern.append(block_domain_str);
  glob_pattern.append("_*.qsb");

  glob_t glob_result;
  glob(glob_pattern.c_str(), 0, nullptr, &glob_result);

  string filename_pattern(storage_path_);
  filename_pattern.append("qsblk_");
  filename_pattern.append(block_domain_str);
  filename_pattern.append("_%");
  filename_pattern.append(SCNu64);
  filename_pattern.append(".qsb");

  block_id_counter counter_max = 0, counter;
  if (glob_result.gl_pathc > 0 &&
      sscanf(glob_result.gl_pathv[glob_result.gl_pathc - 1], filename_pattern.c_str(), &counter) == 1 &&
      counter > counter_max) {
    counter_max = counter;
  }

  globfree(&glob_result);
  return counter_max;
}

size_t FileManagerPosix::numSlots(const block_id block) const {
  const string filename(blockFilename(block));

  struct stat file_stat;
  if (stat(filename.c_str(), &file_stat) == -1) {
    if (errno != ENOENT) {
      LOG(ERROR) << "Failed to retrieve info about file " << filename << " with error: " << strerror(errno);
    }
    return 0;
  }

  if ((file_stat.st_size % kSlotSizeBytes) != 0) {
    throw CorruptPersistentStorage();
  }

  return static_cast<size_t>(file_stat.st_size) / kSlotSizeBytes;
}

bool FileManagerPosix::deleteBlockOrBlob(const block_id block) {
  const string filename(blockFilename(block));

  if ((unlink(filename.c_str()) == 0) || (errno == ENOENT)) {
    return true;
  } else {
    LOG(ERROR) << "Failed to delete file " << filename << " with error: " << strerror(errno);
    return false;
  }
}

bool FileManagerPosix::readBlockOrBlob(const block_id block,
                                       void *buffer,
                                       const std::size_t length) {
  DCHECK(buffer != nullptr);
  DCHECK_EQ(0u, length % kSlotSizeBytes);

  const string filename(blockFilename(block));

  const int fd = open(filename.c_str(), O_RDONLY);
  if (fd == -1) {
    LOG(ERROR) << "Failed to open file " << filename << " with error: " << strerror(errno);
    return false;
  }

  size_t bytes_total = 0;
  while (bytes_total < length) {
    const ssize_t bytes = read(fd, static_cast<char*>(buffer) + bytes_total, length - bytes_total);
    if (bytes > 0) {
      bytes_total += bytes;
    } else if (bytes == -1) {
      if (errno != EINTR) {
        LOG(ERROR) << "Failed to read file " << filename << " with error: " << strerror(errno);
        break;
      }
    } else {
      LOG(ERROR) << "Failed to read file " << filename << " since EOF was reached unexpectedly";
      break;
    }
  }

  if (close(fd) != 0) {
    LOG(ERROR) << "Failed to close file " << filename << " with error: " << strerror(errno);
  }

  return (bytes_total == length);
}

bool FileManagerPosix::writeBlockOrBlob(const block_id block,
                                        const void *buffer,
                                        const std::size_t length) {
  DCHECK(buffer != nullptr);
  DCHECK_EQ(0u, length % kSlotSizeBytes);

  const string filename(blockFilename(block));

  const int fd = open(filename.c_str(), O_WRONLY | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR);
  if (fd == -1) {
    LOG(ERROR) << "Failed to open file " << filename << " with error: " << strerror(errno);
    return false;
  }

  size_t bytes_total = 0;
  while (bytes_total < length) {
    const ssize_t bytes = write(fd, static_cast<const char*>(buffer) + bytes_total, length - bytes_total);
    if (bytes > 0) {
      bytes_total += bytes;
    } else if (bytes == -1 && errno != EINTR) {
      LOG(ERROR) << "Failed to write file " << filename << " with error: " << strerror(errno);
      break;
    }
  }

  if (fsync(fd) != 0) {
    LOG(ERROR) << "Failed to sync file " << filename << " with error: " << strerror(errno);
  }

  if (close(fd) != 0) {
    LOG(ERROR) << "Failed to close file " << filename << " with error: " << strerror(errno);
  }

  return (bytes_total == length);
}

}  // namespace quickstep