// 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 "meta-service/meta_server.h"

#include <brpc/server.h>
#include <butil/endpoint.h>
#include <bvar/window.h>
#include <gen_cpp/cloud.pb.h>
#include <gtest/gtest.h>

#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <cstdlib>
#include <fstream>
#include <memory>
#include <random>
#include <thread>

#include "common/config.h"
#include "common/logging.h"
#include "cpp/sync_point.h"
#include "meta-service/keys.h"
#include "meta-service/mem_txn_kv.h"
#include "meta-service/meta_service.h"
#include "meta-service/txn_kv.h"
#include "meta-service/txn_kv_error.h"
#include "mock_resource_manager.h"
#include "rate-limiter/rate_limiter.h"
#include "resource-manager/resource_manager.h"

using namespace doris;

int main(int argc, char** argv) {
    const std::string conf_file = "doris_cloud.conf";
    if (!cloud::config::init(conf_file.c_str(), true)) {
        std::cerr << "failed to init config file, conf=" << conf_file << std::endl;
        return -1;
    }

    if (!cloud::init_glog("meta_server_test")) {
        std::cerr << "failed to init glog" << std::endl;
        return -1;
    }
    ::testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}

namespace doris::cloud {
void notify_refresh_instance(std::shared_ptr<TxnKv> txn_kv, const std::string& instance_id);
} // namespace doris::cloud

TEST(MetaServerTest, FQDNRefreshInstance) {
    class MockMetaService : public cloud::MetaServiceImpl {
    public:
        MockMetaService(std::shared_ptr<TxnKv> txn_kv,
                        std::shared_ptr<ResourceManager> resource_mgr,
                        std::shared_ptr<RateLimiter> rate_controller)
                : MetaServiceImpl(txn_kv, resource_mgr, rate_controller) {}
        ~MockMetaService() override = default;

        void alter_instance(google::protobuf::RpcController* controller,
                            const ::doris::cloud::AlterInstanceRequest* request,
                            ::doris::cloud::AlterInstanceResponse* response,
                            ::google::protobuf::Closure* done) override {
            (void)controller;
            response->mutable_status()->set_code(cloud::MetaServiceCode::OK);
            LOG(INFO) << "alter instance " << request->DebugString();
            if (request->op() == cloud::AlterInstanceRequest::REFRESH) {
                std::unique_lock<std::mutex> lock(mu_);
                LOG(INFO) << "refresh instance " << request->instance_id();
                refreshed_instances_.insert(request->instance_id());
            }
            done->Run();
        }

        bool is_instance_refreshed(std::string instance_id) {
            std::unique_lock<std::mutex> lock(mu_);
            return refreshed_instances_.count(instance_id) > 0;
        }

        std::mutex mu_;
        std::unordered_set<std::string> refreshed_instances_;
    };

    std::shared_ptr<cloud::TxnKv> txn_kv = std::make_shared<cloud::MemTxnKv>();
    auto resource_mgr = std::make_shared<MockResourceManager>(txn_kv);
    auto rate_limiter = std::make_shared<cloud::RateLimiter>();
    auto mock_service = std::make_unique<MockMetaService>(txn_kv, resource_mgr, rate_limiter);
    MockMetaService* mock_service_ptr = mock_service.get();
    MetaServiceProxy meta_service(std::move(mock_service));

    brpc::ServerOptions options;
    options.num_threads = 1;
    brpc::Server server;
    ASSERT_EQ(server.AddService(&meta_service, brpc::ServiceOwnership::SERVER_DOESNT_OWN_SERVICE),
              0);
    ASSERT_EQ(server.Start(0, &options), 0);
    auto addr = server.listen_address();

    config::hostname = butil::my_hostname();
    config::brpc_listen_port = addr.port;
    config::meta_server_register_interval_ms = 1;

    // Register meta service.
    cloud::MetaServerRegister server_register(txn_kv);
    ASSERT_EQ(server_register.start(), 0);

    while (true) {
        std::unique_ptr<cloud::Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        auto system_key = cloud::system_meta_service_registry_key();
        std::string value;
        if (txn->get(system_key, &value) == TxnErrorCode::TXN_OK) {
            break;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }

    server_register.stop();

    // Refresh instance with FQDN endpoint.
    config::hostname = "";
    notify_refresh_instance(txn_kv, "fqdn_instance_id");

    bool refreshed = false;
    for (size_t i = 0; i < 100; ++i) {
        if (mock_service_ptr->is_instance_refreshed("fqdn_instance_id")) {
            refreshed = true;
            break;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }

    EXPECT_TRUE(refreshed);

    server.Stop(1);
    server.Join();
}

TEST(MetaServerTest, StartAndStop) {
    std::shared_ptr<cloud::TxnKv> txn_kv = std::make_shared<cloud::MemTxnKv>();
    auto server = std::make_unique<MetaServer>(txn_kv);
    auto resource_mgr = std::make_shared<MockResourceManager>(txn_kv);
    auto rate_limiter = std::make_shared<cloud::RateLimiter>();

    brpc::ServerOptions options;
    options.num_threads = 1;
    brpc::Server brpc_server;

    auto sp = SyncPoint::get_instance();

    std::array<std::string, 3> sps {"MetaServer::start:1", "MetaServer::start:2",
                                    "MetaServer::start:3"};
    // use structured binding for point alias (avoid multi lines of declaration)
    auto [meta_server_start_1, meta_server_start_2, meta_server_start_3] = sps;
    sp->enable_processing();
    std::unique_ptr<int, std::function<void(int*)>> defer((int*)0x01, [&](...) {
        for (auto& i : sps) {
            sp->clear_call_back(i);
        } // redundant
        sp->disable_processing();
    });

    auto foo = [](auto&& args) {
        auto* ret = try_any_cast<int*>(args[0]);
        *ret = 1;
    };

    // failed to init resource mgr
    sp->set_call_back(meta_server_start_1, foo);
    ASSERT_EQ(server->start(&brpc_server), 1);
    sp->clear_call_back(meta_server_start_1);

    // failed to start registry
    sp->set_call_back(meta_server_start_2, foo);
    ASSERT_EQ(server->start(&brpc_server), -1);
    sp->clear_call_back(meta_server_start_2);

    // failed to start fdb metrics exporter
    sp->set_call_back(meta_server_start_3, foo);
    ASSERT_EQ(server->start(&brpc_server), -2);
    sp->clear_call_back(meta_server_start_3);

    ASSERT_EQ(server->start(&brpc_server), 0);
    ASSERT_EQ(brpc_server.Start(0, &options), 0);
    auto addr = brpc_server.listen_address();

    config::hostname = butil::my_hostname();
    config::brpc_listen_port = addr.port;
    config::meta_server_register_interval_ms = 1;

    while (true) {
        std::unique_ptr<cloud::Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        auto system_key = cloud::system_meta_service_registry_key();
        std::string value;
        if (txn->get(system_key, &value) == TxnErrorCode::TXN_OK) {
            break;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }

    server->stop();
    brpc_server.Stop(1);
    brpc_server.Join();
}