// 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.

#include <chrono>
#include <cstdint>
#include <map>
#include <memory>
#include <string>
#include <utility>
#include <vector>

#include "backup_types.h"
#include "block_service/block_service.h"
#include "block_service/block_service_manager.h"
#include "common/backup_common.h"
#include "common/gpid.h"
#include "common/json_helper.h"
#include "common/replication.codes.h"
#include "dsn.layer2_types.h"
#include "meta/backup_engine.h"
#include "meta/meta_backup_service.h"
#include "meta/meta_data.h"
#include "meta/meta_service.h"
#include "runtime/api_layer1.h"
#include "runtime/rpc/rpc_address.h"
#include "runtime/rpc/rpc_holder.h"
#include "runtime/task/async_calls.h"
#include "runtime/task/task.h"
#include "runtime/task/task_code.h"
#include "runtime/task/task_tracker.h"
#include "server_state.h"
#include "utils/autoref_ptr.h"
#include "utils/blob.h"
#include "utils/chrono_literals.h"
#include "utils/error_code.h"
#include "utils/filesystem.h"
#include "utils/fmt_logging.h"
#include "utils/zlocks.h"

namespace dsn {
namespace replication {

backup_engine::backup_engine(backup_service *service)
    : _backup_service(service), _block_service(nullptr), _backup_path(""), _is_backup_failed(false)
{
}

backup_engine::~backup_engine() { _tracker.cancel_outstanding_tasks(); }

error_code backup_engine::init_backup(int32_t app_id)
{
    std::string app_name;
    int partition_count;
    {
        zauto_read_lock l;
        _backup_service->get_state()->lock_read(l);
        std::shared_ptr<app_state> app = _backup_service->get_state()->get_app(app_id);
        if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
            LOG_ERROR("app {} is not available, couldn't do backup now.", app_id);
            return ERR_INVALID_STATE;
        }
        app_name = app->app_name;
        partition_count = app->partition_count;
    }

    zauto_lock lock(_lock);
    _backup_status.clear();
    for (int i = 0; i < partition_count; ++i) {
        _backup_status.emplace(i, backup_status::UNALIVE);
    }
    _cur_backup.app_id = app_id;
    _cur_backup.app_name = app_name;
    _cur_backup.backup_id = static_cast<int64_t>(dsn_now_ms());
    _cur_backup.start_time_ms = _cur_backup.backup_id;
    return ERR_OK;
}

error_code backup_engine::set_block_service(const std::string &provider)
{
    _provider_type = provider;
    _block_service = _backup_service->get_meta_service()
                         ->get_block_service_manager()
                         .get_or_create_block_filesystem(provider);
    if (_block_service == nullptr) {
        return ERR_INVALID_PARAMETERS;
    }
    return ERR_OK;
}

error_code backup_engine::set_backup_path(const std::string &path)
{
    if (_block_service && _block_service->is_root_path_set()) {
        return ERR_INVALID_PARAMETERS;
    }
    LOG_INFO("backup path is set to {}.", path);
    _backup_path = path;
    return ERR_OK;
}

error_code backup_engine::write_backup_file(const std::string &file_name,
                                            const dsn::blob &write_buffer)
{
    dist::block_service::create_file_request create_file_req;
    create_file_req.ignore_metadata = true;
    create_file_req.file_name = file_name;

    dsn::error_code err;
    dist::block_service::block_file_ptr remote_file;
    _block_service
        ->create_file(create_file_req,
                      TASK_CODE_EXEC_INLINED,
                      [&err, &remote_file](const dist::block_service::create_file_response &resp) {
                          err = resp.err;
                          remote_file = resp.file_handle;
                      })
        ->wait();
    if (err != dsn::ERR_OK) {
        LOG_INFO("create file {} failed", file_name);
        return err;
    }
    CHECK_NOTNULL(
        remote_file, "create file {} succeed, but can't get handle", create_file_req.file_name);
    remote_file
        ->write(dist::block_service::write_request{write_buffer},
                TASK_CODE_EXEC_INLINED,
                [&err](const dist::block_service::write_response &resp) { err = resp.err; })
        ->wait();
    return err;
}

error_code backup_engine::backup_app_meta()
{
    dsn::blob app_info_buffer;
    {
        zauto_read_lock l;
        _backup_service->get_state()->lock_read(l);
        std::shared_ptr<app_state> app = _backup_service->get_state()->get_app(_cur_backup.app_id);
        if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
            LOG_ERROR("app {} is not available, couldn't do backup now.", _cur_backup.app_id);
            return ERR_INVALID_STATE;
        }
        app_state tmp = *app;
        // Because we don't restore app envs, so no need to write app envs to backup file.
        // TODO(zhangyifan): backup and restore app envs when needed.
        tmp.envs.clear();
        app_info_buffer = dsn::json::json_forwarder<app_info>::encode(tmp);
    }

    std::string backup_root =
        dsn::utils::filesystem::path_combine(_backup_path, _backup_service->backup_root());
    std::string file_name = cold_backup::get_app_metadata_file(
        backup_root, _cur_backup.app_name, _cur_backup.app_id, _cur_backup.backup_id);
    return write_backup_file(file_name, app_info_buffer);
}

void backup_engine::backup_app_partition(const gpid &pid)
{
    dsn::rpc_address partition_primary;
    {
        zauto_read_lock l;
        _backup_service->get_state()->lock_read(l);
        std::shared_ptr<app_state> app = _backup_service->get_state()->get_app(pid.get_app_id());
        if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
            LOG_ERROR("app {} is not available, couldn't do backup now.", pid.get_app_id());

            zauto_lock lock(_lock);
            _is_backup_failed = true;
            return;
        }
        partition_primary = app->partitions[pid.get_partition_index()].primary;
    }

    if (partition_primary.is_invalid()) {
        LOG_WARNING(
            "backup_id({}): partition {} doesn't have a primary now, retry to backup it later.",
            _cur_backup.backup_id,
            pid);
        tasking::enqueue(LPC_DEFAULT_CALLBACK,
                         &_tracker,
                         [this, pid]() { backup_app_partition(pid); },
                         0,
                         std::chrono::seconds(10));
        return;
    }

    auto req = std::make_unique<backup_request>();
    req->pid = pid;
    policy_info backup_policy_info;
    backup_policy_info.__set_backup_provider_type(_provider_type);
    backup_policy_info.__set_policy_name(get_policy_name());
    req->policy = backup_policy_info;
    req->backup_id = _cur_backup.backup_id;
    req->app_name = _cur_backup.app_name;
    if (!_backup_path.empty()) {
        req->__set_backup_path(_backup_path);
    }

    LOG_INFO("backup_id({}): send backup request to partition {}, target_addr = {}",
             _cur_backup.backup_id,
             pid,
             partition_primary);
    backup_rpc rpc(std::move(req), RPC_COLD_BACKUP, 10000_ms, 0, pid.thread_hash());
    rpc.call(
        partition_primary, &_tracker, [this, rpc, pid, partition_primary](error_code err) mutable {
            on_backup_reply(err, rpc.response(), pid, partition_primary);
        });

    zauto_lock l(_lock);
    _backup_status[pid.get_partition_index()] = backup_status::ALIVE;
}

inline void backup_engine::handle_replica_backup_failed(const backup_response &response,
                                                        const gpid pid)
{
    CHECK_EQ(response.pid, pid);
    CHECK_EQ(response.backup_id, _cur_backup.backup_id);

    LOG_ERROR("backup_id({}): backup for partition {} failed, response.err: {}",
              _cur_backup.backup_id,
              pid,
              response.err);
    zauto_lock l(_lock);
    // if one partition fail, the whole backup plan fail.
    _is_backup_failed = true;
    _backup_status[pid.get_partition_index()] = backup_status::FAILED;
}

inline void backup_engine::retry_backup(const dsn::gpid pid)
{
    tasking::enqueue(LPC_DEFAULT_CALLBACK,
                     &_tracker,
                     [this, pid]() { backup_app_partition(pid); },
                     0,
                     std::chrono::seconds(1));
}

void backup_engine::on_backup_reply(const error_code err,
                                    const backup_response &response,
                                    const gpid pid,
                                    const rpc_address &primary)
{
    {
        zauto_lock l(_lock);
        // if backup of some partition failed, we would not handle response from other partitions.
        if (_is_backup_failed) {
            return;
        }
    }

    // if backup completed, receive ERR_OK and
    // resp.progress=cold_backup_constant::PROGRESS_FINISHED;
    // if backup failed, receive ERR_LOCAL_APP_FAILURE;
    // backup not completed in other cases.
    // see replica::on_cold_backup() for details.

    auto rep_error = err == ERR_OK ? response.err : err;

    if (rep_error == ERR_LOCAL_APP_FAILURE) {
        handle_replica_backup_failed(response, pid);
        return;
    }

    if (rep_error != ERR_OK) {
        LOG_ERROR("backup_id({}): backup request to server {} failed, error: {}, retry to "
                  "send backup request.",
                  _cur_backup.backup_id,
                  primary,
                  rep_error);
        retry_backup(pid);
        return;
    };

    if (response.progress == cold_backup_constant::PROGRESS_FINISHED) {
        CHECK_EQ(response.pid, pid);
        CHECK_EQ(response.backup_id, _cur_backup.backup_id);
        LOG_INFO("backup_id({}): backup for partition {} completed.", _cur_backup.backup_id, pid);
        {
            zauto_lock l(_lock);
            _backup_status[pid.get_partition_index()] = backup_status::COMPLETED;
        }
        complete_current_backup();
        return;
    }

    // backup is not finished, meta polling to send request
    LOG_INFO("backup_id({}): receive backup response for partition {} from server {}, now "
             "progress {}, retry to send backup request.",
             _cur_backup.backup_id,
             pid,
             primary,
             response.progress);

    retry_backup(pid);
}

void backup_engine::write_backup_info()
{
    std::string backup_root =
        dsn::utils::filesystem::path_combine(_backup_path, _backup_service->backup_root());
    std::string file_name = cold_backup::get_backup_info_file(backup_root, _cur_backup.backup_id);
    blob buf = dsn::json::json_forwarder<app_backup_info>::encode(_cur_backup);
    error_code err = write_backup_file(file_name, buf);
    if (err == ERR_FS_INTERNAL) {
        LOG_ERROR(
            "backup_id({}): write backup info failed, error {}, do not try again for this error.",
            _cur_backup.backup_id,
            err);
        zauto_lock l(_lock);
        _is_backup_failed = true;
        return;
    }
    if (err != ERR_OK) {
        LOG_WARNING("backup_id({}): write backup info failed, retry it later.",
                    _cur_backup.backup_id);
        tasking::enqueue(LPC_DEFAULT_CALLBACK,
                         &_tracker,
                         [this]() { write_backup_info(); },
                         0,
                         std::chrono::seconds(1));
        return;
    }
    LOG_INFO("backup_id({}): successfully wrote backup info, backup for app {} completed.",
             _cur_backup.backup_id,
             _cur_backup.app_id);
    zauto_lock l(_lock);
    _cur_backup.end_time_ms = dsn_now_ms();
}

void backup_engine::complete_current_backup()
{
    {
        zauto_lock l(_lock);
        for (const auto &status : _backup_status) {
            if (status.second != backup_status::COMPLETED) {
                // backup for some partition was not finished.
                return;
            }
        }
    }
    // complete backup for all partitions.
    write_backup_info();
}

error_code backup_engine::start()
{
    error_code err = backup_app_meta();
    if (err != ERR_OK) {
        LOG_ERROR("backup_id({}): backup meta data for app {} failed, error {}",
                  _cur_backup.backup_id,
                  _cur_backup.app_id,
                  err);
        return err;
    }
    for (int i = 0; i < _backup_status.size(); ++i) {
        tasking::enqueue(LPC_DEFAULT_CALLBACK, &_tracker, [this, i]() {
            backup_app_partition(gpid(_cur_backup.app_id, i));
        });
    }
    return ERR_OK;
}

bool backup_engine::is_in_progress() const
{
    zauto_lock l(_lock);
    return _cur_backup.end_time_ms == 0 && !_is_backup_failed;
}

backup_item backup_engine::get_backup_item() const
{
    zauto_lock l(_lock);
    backup_item item;
    item.backup_id = _cur_backup.backup_id;
    item.app_name = _cur_backup.app_name;
    item.backup_path = _backup_path;
    item.backup_provider_type = _provider_type;
    item.start_time_ms = _cur_backup.start_time_ms;
    item.end_time_ms = _cur_backup.end_time_ms;
    item.is_backup_failed = _is_backup_failed;
    return item;
}

} // namespace replication
} // namespace dsn