/*
 * 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:
 *     meta server service for EON (rDSN layer 2)
 *
 * Revision history:
 *     2015-03-09, @imzhenyu (Zhenyu Guo), first version
 *     xxxx-xx-xx, author, fix bug about xxx
 */

#pragma once

#include <memory>

#include <dsn/cpp/serverlet.h>
#include <dsn/dist/meta_state_service.h>
#include <dsn/perf_counter/perf_counter_wrapper.h>

#include "common/replication_common.h"
#include "meta_rpc_types.h"
#include "meta_options.h"
#include "meta_backup_service.h"
#include "meta_state_service_utils.h"
#include "block_service/block_service_manager.h"

namespace dsn {
namespace security {
class access_controller;
} // namespace security
namespace replication {

class server_state;
class meta_server_failure_detector;
class server_load_balancer;
class meta_duplication_service;
class meta_split_service;
class bulk_load_service;
namespace test {
class test_checker;
}

DEFINE_TASK_CODE(LPC_DEFAULT_CALLBACK, TASK_PRIORITY_COMMON, dsn::THREAD_POOL_DEFAULT)

class meta_service : public serverlet<meta_service>
{
public:
    meta_service();
    virtual ~meta_service();

    error_code start();

    const replication_options &get_options() const { return _opts; }
    const meta_options &get_meta_options() const { return _meta_opts; }

    /// NOTE: prefer using mss::meta_storage instead.
    dist::meta_state_service *get_remote_storage() const { return _storage.get(); }
    mss::meta_storage *get_meta_storage() const { return _meta_storage.get(); }

    server_state *get_server_state() { return _state.get(); }
    server_load_balancer *get_balancer() { return _balancer.get(); }
    dist::block_service::block_service_manager &get_block_service_manager()
    {
        return _block_service_manager;
    }
    bulk_load_service *get_bulk_load_service() { return _bulk_load_svc.get(); }

    meta_function_level::type get_function_level()
    {
        meta_function_level::type level = _function_level.load();
        if (level > meta_function_level::fl_freezed && check_freeze()) {
            level = meta_function_level::fl_freezed;
        }
        return level;
    }
    void set_function_level(meta_function_level::type level) { _function_level.store(level); }

    template <typename TResponse>
    void reply_data(dsn::message_ex *request, const TResponse &data)
    {
        dsn::message_ex *response = request->create_response();
        dsn::marshall(response, data);
        reply_message(request, response);
    }

    virtual void reply_message(dsn::message_ex *, dsn::message_ex *response)
    {
        dsn_rpc_reply(response);
    }
    virtual void send_message(const rpc_address &target, dsn::message_ex *request)
    {
        dsn_rpc_call_one_way(target, request);
    }
    virtual void send_request(dsn::message_ex * /*req*/,
                              const rpc_address &target,
                              const rpc_response_task_ptr &callback)
    {
        dsn_rpc_call(target, callback);
    }

    // these two callbacks are running in fd's thread_pool, and in fd's lock
    void set_node_state(const std::vector<rpc_address> &nodes_list, bool is_alive);
    void get_node_state(/*out*/ std::map<rpc_address, bool> &all_nodes);

    void start_service();
    void balancer_run();

    dsn::task_tracker *tracker() { return &_tracker; }

private:
    void register_rpc_handlers();
    void register_ctrl_commands();
    void unregister_ctrl_commands();

    // client => meta server
    void on_query_configuration_by_index(configuration_query_by_index_rpc rpc);

    // partition server => meta server
    void on_config_sync(configuration_query_by_node_rpc rpc);

    // update configuration
    void on_propose_balancer(configuration_balancer_rpc rpc);
    void on_update_configuration(dsn::message_ex *req);

    // app operations
    void on_create_app(dsn::message_ex *req);
    void on_drop_app(dsn::message_ex *req);
    void on_recall_app(dsn::message_ex *req);
    void on_list_apps(configuration_list_apps_rpc rpc);
    void on_list_nodes(configuration_list_nodes_rpc rpc);

    // app env operations
    void update_app_env(app_env_rpc env_rpc);

    // ddd diagnose
    void ddd_diagnose(ddd_diagnose_rpc rpc);

    // cluster info
    void on_query_cluster_info(configuration_cluster_info_rpc rpc);

    // meta control
    void on_control_meta_level(configuration_meta_control_rpc rpc);
    void on_start_recovery(configuration_recovery_rpc rpc);
    void on_start_restore(dsn::message_ex *req);
    void on_add_backup_policy(dsn::message_ex *req);
    void on_query_backup_policy(query_backup_policy_rpc policy_rpc);
    void on_modify_backup_policy(configuration_modify_backup_policy_rpc rpc);
    void on_report_restore_status(configuration_report_restore_status_rpc rpc);
    void on_query_restore_status(configuration_query_restore_rpc rpc);

    // duplication
    void on_add_duplication(duplication_add_rpc rpc);
    void on_modify_duplication(duplication_modify_rpc rpc);
    void on_query_duplication_info(duplication_query_rpc rpc);
    void on_duplication_sync(duplication_sync_rpc rpc);
    void register_duplication_rpc_handlers();
    void recover_duplication_from_meta_state();
    void initialize_duplication_service();

    // split
    void on_start_partition_split(start_split_rpc rpc);
    void on_control_partition_split(control_split_rpc rpc);
    void on_query_partition_split(query_split_rpc rpc);
    void on_register_child_on_meta(register_child_rpc rpc);
    void on_notify_stop_split(notify_stop_split_rpc rpc);

    // bulk load
    void on_start_bulk_load(start_bulk_load_rpc rpc);
    void on_control_bulk_load(control_bulk_load_rpc rpc);
    void on_query_bulk_load_status(query_bulk_load_rpc rpc);

    // common routines
    // ret:
    //   1. the meta is leader
    //   0. meta isn't leader, and rpc-msg can forward to others
    //  -1. meta isn't leader, and rpc-msg can't forward to others
    // if return -1 and `forward_address' != nullptr, then return leader by `forward_address'.
    int check_leader(dsn::message_ex *req, dsn::rpc_address *forward_address);
    template <typename TRpcHolder>
    int check_leader(TRpcHolder rpc, /*out*/ rpc_address *forward_address);
    // ret:
    //    false: check failed
    //    true:  check succeed
    template <typename TRpcHolder>
    bool check_status(TRpcHolder rpc, /*out*/ rpc_address *forward_address = nullptr);
    template <typename TRespType>
    bool check_status_with_msg(message_ex *req, TRespType &response_struct);

    error_code remote_storage_initialize();
    bool check_freeze() const;

private:
    friend class bulk_load_service_test;
    friend class meta_backup_service_test;
    friend class meta_backup_test_base;
    friend class meta_duplication_service;
    friend class meta_http_service;
    friend class meta_http_service_test;
    friend class meta_load_balance_test;
    friend class meta_service_test;
    friend class meta_service_test_app;
    friend class meta_test_base;
    friend class policy_context_test;
    friend class test::test_checker;

    replication_options _opts;
    meta_options _meta_opts;
    uint64_t _node_live_percentage_threshold_for_update;
    dsn_handle_t _ctrl_node_live_percentage_threshold_for_update = nullptr;

    std::shared_ptr<server_state> _state;
    std::shared_ptr<meta_server_failure_detector> _failure_detector;

    std::shared_ptr<dist::meta_state_service> _storage;
    std::unique_ptr<mss::meta_storage> _meta_storage;

    std::shared_ptr<server_load_balancer> _balancer;
    std::shared_ptr<backup_service> _backup_handler;

    std::unique_ptr<meta_duplication_service> _dup_svc;

    std::unique_ptr<meta_split_service> _split_svc;

    std::unique_ptr<bulk_load_service> _bulk_load_svc;

    // handle all the block filesystems for current meta service
    // (in other words, current service node)
    dist::block_service::block_service_manager _block_service_manager;

    // [
    // this is protected by failure_detector::_lock
    std::set<rpc_address> _alive_set;
    std::set<rpc_address> _dead_set;
    // ]
    mutable zrwlock_nr _meta_lock;

    std::atomic_bool _started;
    std::atomic_bool _recovering;
    // reference replication.thrift for what the meta_function_level means
    std::atomic<meta_function_level::type> _function_level;

    std::string _cluster_root;

    perf_counter_wrapper _recent_disconnect_count;
    perf_counter_wrapper _unalive_nodes_count;

    dsn::task_tracker _tracker;

    std::unique_ptr<security::access_controller> _access_controller;
};

} // namespace replication
} // namespace dsn
