/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *   Copyright 2020-Present Couchbase, Inc.
 *
 *   Licensed 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 "dns_srv_tracker.hxx"

#include "core/logger/logger.hxx"
#include "core/utils/join_strings.hxx"

#include <couchbase/error_codes.hxx>

#include <asio/bind_executor.hpp>
#include <asio/io_context.hpp>
#include <asio/post.hpp>

#include <memory>

namespace couchbase::core::impl
{
dns_srv_tracker::dns_srv_tracker(asio::io_context& ctx, std::string address, const io::dns::dns_config& config, bool use_tls)
  : ctx_{ ctx }
  , dns_client_{ ctx_ }
  , address_{ std::move(address) }
  , config_{ config }
  , use_tls_{ use_tls }
  , service_{ use_tls_ ? "_couchbases" : "_couchbase" }
{
}

void
dns_srv_tracker::get_srv_nodes(utils::movable_function<void(origin::node_list, std::error_code)> callback)
{
    CB_LOG_DEBUG(
      "Query DNS-SRV: address=\"{}\", service=\"{}\", nameserver=\"{}:{}\"", address_, service_, config_.nameserver(), config_.port());
    dns_client_.query_srv(
      address_,
      service_,
      config_,
      [self = shared_from_this(), callback = std::move(callback)](couchbase::core::io::dns::dns_srv_response&& resp) mutable {
          origin::node_list nodes;
          if (resp.ec) {
              CB_LOG_WARNING("failed to fetch DNS SRV records for \"{}\" ({}), assuming that cluster is listening this address",
                             self->address_,
                             resp.ec.message());
          } else if (resp.targets.empty() && self->address_ != "localhost") {
              CB_LOG_WARNING("DNS SRV query returned 0 records for \"{}\", assuming that cluster is listening this address",
                             self->address_);
          } else {
              nodes.reserve(resp.targets.size());
              for (const auto& address : resp.targets) {
                  origin::node_entry node;
                  node.first = address.hostname;
                  node.second = std::to_string(address.port);
                  nodes.emplace_back(node);
              }
          }
          return callback(nodes, resp.ec);
      });
}

void
dns_srv_tracker::do_dns_refresh()
{
    get_srv_nodes([self = shared_from_this()](origin::node_list nodes, std::error_code dns_ec) mutable {
        bool expected_state{ true };
        if (dns_ec || nodes.empty()) {
            if (dns_ec) {
                CB_LOG_WARNING("unable to perform DNS-SRV refresh: {}", dns_ec.message());
            }
            self->refresh_in_progress_.compare_exchange_strong(expected_state, false);
            return;
        }
        std::set<std::shared_ptr<config_listener>> listeners;
        {
            std::scoped_lock lock(self->config_listeners_mutex_);
            listeners = self->config_listeners_;
        }

        if (!listeners.empty()) {
            auto config = topology::make_blank_configuration(nodes, self->use_tls_, true);
            std::vector<std::string> endpoints;
            endpoints.reserve(nodes.size());
            for (const auto& [host, port] : nodes) {
                endpoints.emplace_back(fmt::format("\"{}:{}\"", host, port));
            }
            CB_LOG_DEBUG(
              "generated configuration from DNS-SRV response \"{}\": [{}]", self->address_, utils::join_strings(endpoints, ", "));
            for (const auto& listener : listeners) {
                listener->update_config(config);
            }
        }
        self->refresh_in_progress_.compare_exchange_strong(expected_state, false);
    });
}

void
dns_srv_tracker::report_bootstrap_error(const std::string& endpoint, std::error_code ec)
{
    bool trigger_dns_srv_refresh = false;

    if (ec && ec != errc::common::request_canceled) {
        std::scoped_lock lock(known_endpoints_mutex_);
        known_endpoints_.erase(endpoint);
        if (known_endpoints_.empty()) {
            trigger_dns_srv_refresh = true;
        }
    } else {
        return;
    }

    bool expected_state{ false };
    if (trigger_dns_srv_refresh && refresh_in_progress_.compare_exchange_strong(expected_state, true)) {
        CB_LOG_DEBUG("all nodes failed to bootstrap, triggering DNS-SRV refresh, ec={}, last endpoint=\"{}\"", ec.message(), endpoint);
        return asio::post(asio::bind_executor(ctx_, [self = shared_from_this()]() mutable { self->do_dns_refresh(); }));
    }
}

void
dns_srv_tracker::report_bootstrap_success(const std::vector<std::string>& endpoints)
{
    std::set<std::string, std::less<>> known_endpoints{ endpoints.begin(), endpoints.end() };
    std::scoped_lock lock(known_endpoints_mutex_);
    std::swap(known_endpoints_, known_endpoints);
}

void
dns_srv_tracker::register_config_listener(std::shared_ptr<config_listener> listener)
{
    std::scoped_lock lock(config_listeners_mutex_);
    config_listeners_.insert(listener);
}

void
dns_srv_tracker::unregister_config_listener(std::shared_ptr<config_listener> listener)
{
    std::scoped_lock lock(config_listeners_mutex_);
    config_listeners_.erase(listener);
}
} // namespace couchbase::core::impl