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

#include <gen_cpp/cloud.pb.h>

#include <glog/logging.h>

#include "cloud/cloud_meta_mgr.h"

#include "cloud/cloud_storage_engine.h"

#include "cloud/cloud_tablet.h"

#include "cloud/config.h"

#include "runtime/exec_env.h"

#include "util/time.h"

namespace doris {

CloudClusterInfo::~CloudClusterInfo() {

    stop_bg_worker();

}

void CloudClusterInfo::start_bg_worker() {

    bool expected = true;

    if (!_bg_worker_stopped.compare_exchange_strong(expected, false)) {

        // Already running

        return;

    }

    _bg_worker = std::thread(&CloudClusterInfo::_bg_worker_func, this);

    LOG(INFO) << "CloudClusterInfo background worker started, "

              << "refresh_interval=" << config::cluster_status_cache_refresh_interval_sec << "s";

}

void CloudClusterInfo::stop_bg_worker() {

    bool expected = false;

    if (!_bg_worker_stopped.compare_exchange_strong(expected, true)) {

        // Already stopped

        return;

    }

    {

        std::lock_guard lock(_bg_worker_mutex);

        _bg_worker_cv.notify_all();

    }

    if (_bg_worker.joinable()) {

        _bg_worker.join();

    }

    LOG(INFO) << "CloudClusterInfo background worker stopped";

}

void CloudClusterInfo::_bg_worker_func() {

    LOG(INFO) << "CloudClusterInfo background worker thread running";

    while (!_bg_worker_stopped.load()) {

        _refresh_cluster_status();

        std::unique_lock lock(_bg_worker_mutex);

        _bg_worker_cv.wait_for(

                lock, std::chrono::seconds(config::cluster_status_cache_refresh_interval_sec),

                [this] { return _bg_worker_stopped.load(); });

    }

}

void CloudClusterInfo::_refresh_cluster_status() {

    if (!config::is_cloud_mode()) {

        return;

    }

    auto* cloud_engine =

            dynamic_cast<CloudStorageEngine*>(&ExecEnv::GetInstance()->storage_engine());

    if (!cloud_engine) {

        return;

    }

    std::unordered_map<std::string, std::pair<int32_t, int64_t>> cluster_status;

    std::string resolved_cluster_id;

    Status st = cloud_engine->meta_mgr().get_cluster_status(

            &cluster_status, my_cluster_id().empty() ? &resolved_cluster_id : nullptr);

    if (!st.ok()) {

        LOG(WARNING) << "Failed to refresh cluster status: " << st;

        return;

    }

    // Update cache

    {

        std::unique_lock lock(_mutex);

        _cluster_status_cache.clear();

        for (const auto& [cluster_id, status_pair] : cluster_status) {

            _cluster_status_cache[cluster_id] = {status_pair.first, status_pair.second};

        }

    }

    VLOG_DEBUG << "Refreshed cluster status cache, " << cluster_status.size() << " clusters";

    // Set our own cluster_id if resolved from the response

    if (my_cluster_id().empty() && !resolved_cluster_id.empty()) {

        set_my_cluster_id(resolved_cluster_id);

        LOG(INFO) << "Resolved my cluster_id: " << resolved_cluster_id;

    }

}

bool CloudClusterInfo::should_skip_compaction(CloudTablet* tablet) const {

    if (!config::enable_compaction_rw_separation) {

        return false;

    }

    std::string last_active_cluster = tablet->last_active_cluster_id();

    std::string my_cluster = my_cluster_id();

    int64_t tablet_id = tablet->tablet_id();

    // Case 1: No active cluster record, any cluster can compact

    if (last_active_cluster.empty()) {

        VLOG_DEBUG << "tablet " << tablet_id << " has no last_active_cluster record, "

                   << "my_cluster=" << my_cluster << ", allow compaction";

        return false;

    }

    // Case 2: This is the active cluster, allow compaction

    if (last_active_cluster == my_cluster) {

        VLOG_DEBUG << "tablet " << tablet_id << " last_active_cluster=" << last_active_cluster

                   << " equals my_cluster=" << my_cluster << ", allow compaction";

        return false;

    }

    // Case 3: Check if the last active cluster is available

    ClusterStatusCache cache;

    if (!get_cluster_status(last_active_cluster, &cache)) {

        // Cluster not found in cache, might be deleted, allow takeover

        LOG(INFO) << "compaction_rw_separation: tablet " << tablet_id

                  << " last_active_cluster=" << last_active_cluster

                  << " not found in cache (maybe deleted), my_cluster=" << my_cluster

                  << ", allow takeover";

        return false;

    }

    // Force compaction if tablet has too many rowsets (>80% of max_tablet_version_num),

    // even on read clusters, to prevent version count from growing unbounded

    // when the write cluster can't keep up or has compaction disabled.

    int64_t num_rowsets = tablet->fetch_add_approximate_num_rowsets(0);

    auto threshold = static_cast<int64_t>(tablet->max_version_config() *

                                          config::compaction_rw_separation_version_threshold_ratio);

    if (num_rowsets > threshold) {

        LOG(INFO) << "compaction_rw_separation: force compaction on tablet " << tablet_id

                  << ", num_rowsets=" << num_rowsets << " > threshold=" << threshold << " (80% of "

                  << tablet->max_version_config() << ")"

                  << ", my_cluster=" << my_cluster;

        return false;

    }

    auto status = static_cast<cloud::ClusterStatus>(cache.status);

    int64_t status_mtime = cache.mtime_ms;

    int64_t now = UnixMillis();

    int64_t elapsed = now - status_mtime;

    int64_t timeout = config::compaction_cluster_takeover_timeout_ms;

    // Case 4: Original cluster is NORMAL (still active), cannot takeover

    if (status == cloud::ClusterStatus::NORMAL) {

        LOG_EVERY_N(INFO, 100) << "compaction_rw_separation: skip tablet " << tablet_id

                               << ", last_active_cluster=" << last_active_cluster

                               << " is NORMAL (active), my_cluster=" << my_cluster;

        return true;

    }

    // Case 5: Original cluster is unavailable (SUSPENDED/MANUAL_SHUTDOWN/deleted)

    if (elapsed > timeout) {

        // Takeover successful

        LOG(INFO) << "compaction_rw_separation: takeover tablet " << tablet_id

                  << ", last_active_cluster=" << last_active_cluster

                  << " status=" << cloud::ClusterStatus_Name(status)

                  << " status_mtime=" << status_mtime << " elapsed=" << elapsed

                  << "ms > timeout=" << timeout << "ms"

                  << ", my_cluster=" << my_cluster;

        return false;

    } else {

        // Timeout not reached yet, waiting

        LOG_EVERY_N(INFO, 100) << "compaction_rw_separation: skip tablet " << tablet_id

                               << ", last_active_cluster=" << last_active_cluster

                               << " status=" << cloud::ClusterStatus_Name(status)

                               << " status_mtime=" << status_mtime << " elapsed=" << elapsed

                               << "ms <= timeout=" << timeout << "ms"

                               << ", my_cluster=" << my_cluster << ", waiting for takeover";

        return true;

    }

}

} // namespace doris