// 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 "service/http/action/show_hotspot_action.h"

#include <queue>

#include <string>

#include "cloud/cloud_tablet_mgr.h"

#include "service/http/http_channel.h"

#include "service/http/http_request.h"

namespace doris {

#include "common/compile_check_begin.h"

namespace {

enum class Metrics {

    READ_BLOCK = 0,

    WRITE = 1,

    COMPACTION = 2,

    NUM_ROWSETS = 3,

    NUM_BASE_ROWSETS = 4,

    NUM_CUMU_ROWSETS = 5,

    UNKNOWN = 100000,

};

Status check_param(HttpRequest* req, size_t& top_n, Metrics& metrics) {

    const std::string TOPN_PARAM = "topn";

    auto& topn_str = req->param(TOPN_PARAM);

    if (!topn_str.empty()) {

        try {

            top_n = std::stoul(topn_str);

        } catch (const std::exception& e) {

            return Status::InternalError("convert topn failed, {}", e.what());

        }

    }

    const std::string METRICS_PARAM = "metrics";

    auto& metrics_str = req->param(METRICS_PARAM);

    if (metrics_str.empty()) {

        return Status::InternalError("metrics must be specified");

    }

    if (metrics_str == "read_block") {

        metrics = Metrics::READ_BLOCK;

    } else if (metrics_str == "write") {

        metrics = Metrics::WRITE;

    } else if (metrics_str == "compaction") {

        metrics = Metrics::COMPACTION;

    } else if (metrics_str == "num_rowsets") {

        metrics = Metrics::NUM_ROWSETS;

    } else if (metrics_str == "num_cumu_rowsets") {

        metrics = Metrics::NUM_CUMU_ROWSETS;

    } else if (metrics_str == "num_base_rowsets") {

        metrics = Metrics::NUM_BASE_ROWSETS;

    } else {

        return Status::InternalError("unknown metrics: {}", metrics_str);

    }

    return Status::OK();

}

struct TabletCounter {

    int64_t tablet_id {0};

    int64_t count {0};

};

struct Comparator {

    constexpr bool operator()(const TabletCounter& lhs, const TabletCounter& rhs) const {

        return lhs.count > rhs.count;

    }

};

using MinHeap = std::priority_queue<TabletCounter, std::vector<TabletCounter>, Comparator>;

} // namespace

void ShowHotspotAction::handle(HttpRequest* req) {

    size_t topn = 0;

    Metrics metrics {Metrics::UNKNOWN};

    auto st = check_param(req, topn, metrics);

    if (!st.ok()) [[unlikely]] {

        HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, st.to_string());

        return;

    }

    std::function<int64_t(CloudTablet&)> count_fn;

    switch (metrics) {

    case Metrics::READ_BLOCK:

        count_fn = [](auto&& t) { return t.read_block_count.load(std::memory_order_relaxed); };

        break;

    case Metrics::WRITE:

        count_fn = [](auto&& t) { return t.write_count.load(std::memory_order_relaxed); };

        break;

    case Metrics::COMPACTION:

        count_fn = [](auto&& t) { return t.compaction_count.load(std::memory_order_relaxed); };

        break;

    case Metrics::NUM_ROWSETS:

        count_fn = [](auto&& t) { return t.fetch_add_approximate_num_rowsets(0); };

        break;

    case Metrics::NUM_BASE_ROWSETS:

        count_fn = [](auto&& t) {

            return t.fetch_add_approximate_num_rowsets(0) -

                   t.fetch_add_approximate_cumu_num_rowsets(0);

        };

        break;

    case Metrics::NUM_CUMU_ROWSETS:

        count_fn = [](auto&& t) { return t.fetch_add_approximate_cumu_num_rowsets(0); };

        break;

    default:

        break;

    }

    if (!count_fn) {

        HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "metrics not specified");

        return;

    }

    auto tablets = _storage_engine.tablet_mgr().get_weak_tablets();

    std::vector<TabletCounter> buffer;

    buffer.reserve(tablets.size());

    for (auto&& t : tablets) {

        if (auto tablet = t.lock(); tablet) {

            buffer.push_back({tablet->tablet_id(), count_fn(*tablet)});

        }

    }

    if (topn <= 0) {

        topn = tablets.size();

    }

    MinHeap min_heap;

    for (auto&& counter : buffer) {

        min_heap.push(counter);

        if (min_heap.size() > topn) {

            min_heap.pop();

        }

    }

    buffer.resize(0);

    while (!min_heap.empty()) {

        buffer.push_back(min_heap.top());

        min_heap.pop();

    }

    std::string res;

    res.reserve(buffer.size() * 20);

    // Descending order

    std::for_each(buffer.rbegin(), buffer.rend(), [&res](auto&& counter) {

        res += fmt::format("{} {}\n", counter.tablet_id, counter.count);

    });

    HttpChannel::send_reply(req, HttpStatus::OK, res);

#include "common/compile_check_end.h"

}

} // namespace doris