/*
 * The MIT License (MIT)
 *
 * Copyright (c) 2015 Microsoft Corporation
 *
 * -=- Robust Distributed System Nucleus (rDSN) -=-
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

/*
 * Description:
 *     What is this file about?
 *
 * Revision history:
 *     xxxx-xx-xx, author, first version
 *     xxxx-xx-xx, author, fix bug about xxx
 */
#include <cstdlib>
#include <sys/stat.h>
#include <dsn/utility/filesystem.h>
#include <dsn/tool-api/async_calls.h>

#include "nfs_server_impl.h"

namespace dsn {
namespace service {

DSN_DEFINE_int32("nfs", max_send_rate_megabytes, 100, "max rate of send to remote node(MB/s)");
DSN_TAG_VARIABLE(max_send_rate_megabytes, FT_MUTABLE);

DSN_DECLARE_int32(file_close_timer_interval_ms_on_server);
DSN_DECLARE_int32(file_close_expire_time_ms);

nfs_service_impl::nfs_service_impl() : ::dsn::serverlet<nfs_service_impl>("nfs")
{
    _file_close_timer = ::dsn::tasking::enqueue_timer(
        LPC_NFS_FILE_CLOSE_TIMER,
        &_tracker,
        [this] { close_file(); },
        std::chrono::milliseconds(FLAGS_file_close_timer_interval_ms_on_server));

    _recent_copy_data_size.init_app_counter("eon.nfs_server",
                                            "recent_copy_data_size",
                                            COUNTER_TYPE_VOLATILE_NUMBER,
                                            "nfs server copy data size in the recent period");
    _recent_copy_fail_count.init_app_counter(
        "eon.nfs_server",
        "recent_copy_fail_count",
        COUNTER_TYPE_VOLATILE_NUMBER,
        "nfs server copy fail count count in the recent period");

    _send_token_bucket = std::make_unique<folly::DynamicTokenBucket>();
    register_cli_commands();
}

void nfs_service_impl::on_copy(const ::dsn::service::copy_request &request,
                               ::dsn::rpc_replier<::dsn::service::copy_response> &reply)
{
    // dinfo(">>> on call RPC_COPY end, exec RPC_NFS_COPY");

    std::string file_path =
        dsn::utils::filesystem::path_combine(request.source_dir, request.file_name);
    disk_file *hfile;

    {
        zauto_lock l(_handles_map_lock);
        auto it = _handles_map.find(file_path); // find file handle cache first

        if (it == _handles_map.end()) // not found
        {
            hfile = file::open(file_path.c_str(), O_RDONLY | O_BINARY, 0);
            if (hfile) {

                auto fh = std::make_shared<file_handle_info_on_server>();
                fh->file_handle = hfile;
                fh->file_access_count = 1;
                fh->last_access_time = dsn_now_ms();
                _handles_map.insert(std::make_pair(file_path, std::move(fh)));
            }
        } else // found
        {
            hfile = it->second->file_handle;
            it->second->file_access_count++;
            it->second->last_access_time = dsn_now_ms();
        }
    }

    dinfo("nfs: copy file %s [%" PRId64 ", %" PRId64 ")",
          file_path.c_str(),
          request.offset,
          request.offset + request.size);

    if (hfile == 0) {
        derror("{nfs_service} open file %s failed", file_path.c_str());
        ::dsn::service::copy_response resp;
        resp.error = ERR_OBJECT_NOT_FOUND;
        reply(resp);
        return;
    }

    std::shared_ptr<callback_para> cp = std::make_shared<callback_para>(std::move(reply));
    cp->bb = blob(dsn::utils::make_shared_array<char>(request.size), request.size);
    cp->dst_dir = std::move(request.dst_dir);
    cp->file_path = std::move(file_path);
    cp->hfile = hfile;
    cp->offset = request.offset;
    cp->size = request.size;

    auto buffer_save = cp->bb.buffer().get();

    file::read(
        hfile,
        buffer_save,
        request.size,
        request.offset,
        LPC_NFS_READ,
        &_tracker,
        [this, cp](error_code err, size_t sz) mutable { internal_read_callback(err, sz, *cp); });
}

void nfs_service_impl::internal_read_callback(error_code err, size_t sz, callback_para &cp)
{
    _send_token_bucket->consumeWithBorrowAndWait(
        sz, FLAGS_max_send_rate_megabytes << 20, 1.5 * (FLAGS_max_send_rate_megabytes << 20));
    {
        zauto_lock l(_handles_map_lock);
        auto it = _handles_map.find(cp.file_path);

        if (it != _handles_map.end()) {
            it->second->file_access_count--;
        }
    }

    if (err != ERR_OK) {
        derror(
            "{nfs_service} read file %s failed, err = %s", cp.file_path.c_str(), err.to_string());
        _recent_copy_fail_count->increment();
    } else {
        _recent_copy_data_size->add(sz);
    }

    ::dsn::service::copy_response resp;
    resp.error = err;
    resp.file_content = std::move(cp.bb);
    resp.offset = cp.offset;
    resp.size = cp.size;

    cp.replier(resp);
}

// RPC_NFS_NEW_NFS_GET_FILE_SIZE
void nfs_service_impl::on_get_file_size(
    const ::dsn::service::get_file_size_request &request,
    ::dsn::rpc_replier<::dsn::service::get_file_size_response> &reply)
{
    // dinfo(">>> on call RPC_NFS_GET_FILE_SIZE end, exec RPC_NFS_GET_FILE_SIZE");

    get_file_size_response resp;
    error_code err = ERR_OK;
    std::vector<std::string> file_list;
    std::string folder = request.source_dir;
    if (request.file_list.size() == 0) // return all file size in the destination file folder
    {
        if (!dsn::utils::filesystem::directory_exists(folder)) {
            derror("{nfs_service} directory %s not exist", folder.c_str());
            err = ERR_OBJECT_NOT_FOUND;
        } else {
            if (!dsn::utils::filesystem::get_subfiles(folder, file_list, true)) {
                derror("{nfs_service} get subfiles of directory %s failed", folder.c_str());
                err = ERR_FILE_OPERATION_FAILED;
            } else {
                for (auto &fpath : file_list) {
                    // TODO: using uint64 instead as file ma
                    // Done
                    int64_t sz;
                    if (!dsn::utils::filesystem::file_size(fpath, sz)) {
                        derror("{nfs_service} get size of file %s failed", fpath.c_str());
                        err = ERR_FILE_OPERATION_FAILED;
                        break;
                    }

                    resp.size_list.push_back((uint64_t)sz);
                    resp.file_list.push_back(
                        fpath.substr(request.source_dir.length(), fpath.length() - 1));
                }
                file_list.clear();
            }
        }
    } else // return file size in the request file folder
    {
        for (size_t i = 0; i < request.file_list.size(); i++) {
            std::string file_path =
                dsn::utils::filesystem::path_combine(folder, request.file_list[i]);

            struct stat st;
            if (0 != ::stat(file_path.c_str(), &st)) {
                derror("{nfs_service} get stat of file %s failed, err = %s",
                       file_path.c_str(),
                       strerror(errno));
                err = ERR_OBJECT_NOT_FOUND;
                break;
            }

            // TODO: using int64 instead as file may exceed the size of 32bit
            // Done
            uint64_t size = st.st_size;

            resp.size_list.push_back(size);
            resp.file_list.push_back((folder + request.file_list[i])
                                         .substr(request.source_dir.length(),
                                                 (folder + request.file_list[i]).length() - 1));
        }
    }

    resp.error = err;
    reply(resp);
}

void nfs_service_impl::close_file() // release out-of-date file handle
{
    zauto_lock l(_handles_map_lock);

    for (auto it = _handles_map.begin(); it != _handles_map.end();) {
        auto fptr = it->second;

        // not used and expired
        if (fptr->file_access_count == 0 &&
            dsn_now_ms() - fptr->last_access_time > (uint64_t)FLAGS_file_close_expire_time_ms) {
            dinfo("nfs: close file handle %s", it->first.c_str());
            it = _handles_map.erase(it);
        } else
            it++;
    }
}

// TODO(jiashuo1): just for compatibility, ready to delete it later
void nfs_service_impl::register_cli_commands()
{
    static std::once_flag flag;
    std::call_once(flag, [&]() {
        _nfs_max_send_rate_megabytes_cmd = dsn::command_manager::instance().register_command(
            {"nfs.max_send_rate_megabytes"},
            "nfs.max_send_rate_megabytes [num]",
            "control the max rate(MB/s) to send file to remote node",
            [](const std::vector<std::string> &args) {
                std::string result("OK");

                if (args.empty()) {
                    return std::to_string(FLAGS_max_send_rate_megabytes);
                }

                int32_t max_send_rate_megabytes = 0;
                if (!dsn::buf2int32(args[0], max_send_rate_megabytes) ||
                    max_send_rate_megabytes <= 0) {
                    return std::string("ERR: invalid arguments");
                }

                FLAGS_max_send_rate_megabytes = max_send_rate_megabytes;
                return result;
            });
    });
}

} // namespace service
} // namespace dsn
