// 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 "nacos_naming_service.h"

#include <gflags/gflags.h>

#include <set>

#include "brpc/http_status_code.h"
#include "brpc/log.h"
#include "butil/iobuf.h"
#include "butil/logging.h"
#include "butil/third_party/rapidjson/document.h"

namespace brpc {
namespace policy {

DEFINE_string(nacos_address, "",
              "The query string of request nacos for discovering service.");
DEFINE_string(nacos_service_discovery_path, "/nacos/v1/ns/instance/list",
              "The url path for discovering service.");
DEFINE_string(nacos_service_auth_path, "/nacos/v1/auth/login",
              "The url path for authentiction.");
DEFINE_int32(nacos_connect_timeout_ms, 200,
             "Timeout for creating connections to nacos in milliseconds");
DEFINE_string(nacos_username, "", "nacos username");
DEFINE_string(nacos_password, "", "nacos password");
DEFINE_string(nacos_load_balancer, "rr", "nacos load balancer name");

int NacosNamingService::Connect() {
    ChannelOptions opt;
    opt.protocol = PROTOCOL_HTTP;
    opt.connect_timeout_ms = FLAGS_nacos_connect_timeout_ms;
    const int ret = _channel.Init(FLAGS_nacos_address.c_str(),
                                  FLAGS_nacos_load_balancer.c_str(), &opt);
    if (ret != 0) {
        LOG(ERROR) << "Fail to init channel to nacos at "
                   << FLAGS_nacos_address;
    }
    return ret;
}

int NacosNamingService::RefreshAccessToken(const char *service_name) {
    Controller cntl;
    cntl.http_request().uri() = FLAGS_nacos_service_auth_path;
    cntl.http_request().set_method(brpc::HttpMethod::HTTP_METHOD_POST);
    cntl.http_request().set_content_type("application/x-www-form-urlencoded");

    auto &buf = cntl.request_attachment();
    buf.append("username=");
    buf.append(FLAGS_nacos_username);
    buf.append("&password=");
    buf.append(FLAGS_nacos_password);

    _channel.CallMethod(nullptr, &cntl, nullptr, nullptr, nullptr);
    if (cntl.Failed()) {
        LOG(ERROR) << "Fail to access " << FLAGS_nacos_service_auth_path << ": "
                   << cntl.ErrorText();
        return -1;
    }

    BUTIL_RAPIDJSON_NAMESPACE::Document doc;
    if (doc.Parse(cntl.response_attachment().to_string().c_str())
            .HasParseError()) {
        LOG(ERROR) << "Failed to parse nacos auth response";
        return -1;
    }
    if (!doc.IsObject()) {
        LOG(ERROR) << "The nacos's auth response for " << service_name
                   << " is not a json object";
        return -1;
    }

    auto iter = doc.FindMember("accessToken");
    if (iter != doc.MemberEnd() && iter->value.IsString()) {
        _access_token = iter->value.GetString();
    } else {
        LOG(ERROR) << "The nacos's auth response for " << service_name
                   << " has no accessToken field";
        return -1;
    }

    auto iter_ttl = doc.FindMember("tokenTtl");
    if (iter_ttl != doc.MemberEnd() && iter_ttl->value.IsInt()) {
        _token_expire_time = time(NULL) + iter_ttl->value.GetInt() - 10;
    } else {
        _token_expire_time = 0;
    }

    return 0;
}

int NacosNamingService::GetServerNodes(const char *service_name,
                                       bool token_changed,
                                       std::vector<ServerNode> *nodes) {
    if (_nacos_url.empty() || token_changed) {
        _nacos_url = FLAGS_nacos_service_discovery_path;
        _nacos_url += "?";
        if (!_access_token.empty()) {
            _nacos_url += "accessToken=" + _access_token;
            _nacos_url += "&";
        }
        _nacos_url += service_name;
    }

    Controller cntl;
    cntl.http_request().uri() = _nacos_url;
    _channel.CallMethod(nullptr, &cntl, nullptr, nullptr, nullptr);
    if (cntl.Failed()) {
        LOG(ERROR) << "Fail to access " << _nacos_url << ": "
                   << cntl.ErrorText();
        return -1;
    }
    if (cntl.http_response().status_code() != HTTP_STATUS_OK) {
        LOG(ERROR) << "Failed to request nacos, http status code: "
                   << cntl.http_response().status_code();
        return -1;
    }

    BUTIL_RAPIDJSON_NAMESPACE::Document doc;
    if (doc.Parse(cntl.response_attachment().to_string().c_str())
            .HasParseError()) {
        LOG(ERROR) << "Failed to parse nacos response";
        return -1;
    }
    if (!doc.IsObject()) {
        LOG(ERROR) << "The nacos's response for " << service_name
                   << " is not a json object";
        return -1;
    }

    auto it_hosts = doc.FindMember("hosts");
    if (it_hosts == doc.MemberEnd()) {
        LOG(ERROR) << "The nacos's response for " << service_name
                   << " has no hosts member";
        return -1;
    }
    auto &hosts = it_hosts->value;
    if (!hosts.IsArray()) {
        LOG(ERROR) << "hosts member in nacos response is not an array";
        return -1;
    }

    std::set<ServerNode> presence;
    for (auto it = hosts.Begin(); it != hosts.End(); ++it) {
        auto &host = *it;
        if (!host.IsObject()) {
            LOG(ERROR) << "host member in nacos response is not an object";
            continue;
        }

        auto it_ip = host.FindMember("ip");
        if (it_ip == host.MemberEnd() || !it_ip->value.IsString()) {
            LOG(ERROR) << "host in nacos response has not ip";
            continue;
        }
        auto &ip = it_ip->value;

        auto it_port = host.FindMember("port");
        if (it_port == host.MemberEnd() || !it_port->value.IsInt()) {
            LOG(ERROR) << "host in nacos response has not port";
            continue;
        }
        auto &port = it_port->value;

        auto it_enabled = host.FindMember("enabled");
        if (it_enabled == host.MemberEnd() || !(it_enabled->value.IsBool()) ||
            !(it_enabled->value.GetBool())) {
            LOG(INFO) << "nacos " << ip.GetString() << ":" << port.GetInt()
                      << " is not enabled";
            continue;
        }

        auto it_healthy = host.FindMember("healthy");
        if (it_healthy == host.MemberEnd() || !(it_healthy->value.IsBool()) ||
            !(it_healthy->value.GetBool())) {
            LOG(INFO) << "nacos " << ip.GetString() << ":" << port.GetInt()
                      << " is not healthy";
            continue;
        }

        butil::EndPoint end_point;
        if (str2endpoint(ip.GetString(), port.GetUint(), &end_point) != 0) {
            LOG(ERROR) << "ncos service with illegal address or port: "
                       << ip.GetString() << ":" << port.GetUint();
            continue;
        }

        ServerNode node(end_point);
        auto it_weight = host.FindMember("weight");
        if (it_weight != host.MemberEnd() && it_weight->value.IsNumber()) {
            node.tag =
                std::to_string(static_cast<long>(it_weight->value.GetDouble()));
        }

        presence.insert(node);
    }

    nodes->reserve(presence.size());
    nodes->assign(presence.begin(), presence.end());

    if (nodes->empty() && hosts.Size() != 0) {
        LOG(ERROR) << "All service about " << service_name
                   << " from nacos is invalid, refuse to update servers";
        return -1;
    }

    RPC_VLOG << "Got " << nodes->size()
             << (nodes->size() > 1 ? " servers" : " server") << " from "
             << service_name;

    auto it_cache = doc.FindMember("cacheMillis");
    if (it_cache != doc.MemberEnd() && it_cache->value.IsInt64()) {
        _cache_ms = it_cache->value.GetInt64();
    }

    return 0;
}

NacosNamingService::NacosNamingService()
    : _nacos_connected(false), _cache_ms(-1), _token_expire_time(0) {}

int NacosNamingService::GetNamingServiceAccessIntervalMs() const {
    if (0 < _cache_ms) {
        return _cache_ms;
    }
    return PeriodicNamingService::GetNamingServiceAccessIntervalMs();
}

int NacosNamingService::GetServers(const char *service_name,
                                   std::vector<ServerNode> *servers) {
    if (!_nacos_connected) {
        const int ret = Connect();
        if (0 == ret) {
            _nacos_connected = true;
        } else {
            return ret;
        }
    }

    const bool authentiction_enabled =
        !FLAGS_nacos_username.empty() && !FLAGS_nacos_password.empty();
    const bool has_invalid_access_token =
        _access_token.empty() ||
        (0 < _token_expire_time && _token_expire_time <= time(NULL));
    bool token_changed = false;

    if (authentiction_enabled && has_invalid_access_token) {
        const int ret = RefreshAccessToken(service_name);
        if (ret == 0) {
            token_changed = true;
        } else {
            return ret;
        }
    }

    servers->clear();
    return GetServerNodes(service_name, token_changed, servers);
}

void NacosNamingService::Describe(std::ostream &os,
                                  const DescribeOptions &) const {
    os << "nacos";
    return;
}

NamingService *NacosNamingService::New() const {
    return new NacosNamingService;
}

void NacosNamingService::Destroy() { delete this; }

}  // namespace policy
}  // namespace brpc
