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

// IWYU pragma: no_include <bits/chrono.h>

#include <chrono> // IWYU pragma: keep

#include <exception>

#include <future>

#include <memory>

#include <mutex>

#include <sstream>

#include <string>

#include <thread>

#include <utility>

#include "absl/strings/substitute.h"

#include "cloud/cloud_base_compaction.h"

#include "cloud/cloud_compaction_action.h"

#include "cloud/cloud_cumulative_compaction.h"

#include "cloud/cloud_full_compaction.h"

#include "cloud/cloud_tablet.h"

#include "cloud/cloud_tablet_mgr.h"

#include "common/logging.h"

#include "common/metrics/doris_metrics.h"

#include "common/status.h"

#include "service/http/http_channel.h"

#include "service/http/http_headers.h"

#include "service/http/http_request.h"

#include "service/http/http_status.h"

#include "storage/compaction/base_compaction.h"

#include "storage/compaction/cumulative_compaction.h"

#include "storage/compaction/cumulative_compaction_policy.h"

#include "storage/compaction/cumulative_compaction_time_series_policy.h"

#include "storage/compaction/full_compaction.h"

#include "storage/compaction_task_tracker.h"

#include "storage/olap_define.h"

#include "storage/storage_engine.h"

#include "storage/tablet/tablet_manager.h"

#include "util/stopwatch.hpp"

namespace doris {

using namespace ErrorCode;

namespace {}

const static std::string HEADER_JSON = "application/json";

CloudCompactionAction::CloudCompactionAction(CompactionActionType ctype, ExecEnv* exec_env,

                                             CloudStorageEngine& engine, TPrivilegeHier::type hier,

                                             TPrivilegeType::type ptype)

        : HttpHandlerWithAuth(exec_env, hier, ptype), _engine(engine), _compaction_type(ctype) {}

/// check param and fetch tablet_id & table_id from req

static Status _check_param(HttpRequest* req, uint64_t* tablet_id, uint64_t* table_id) {

    // req tablet id and table id, we have to set only one of them.

    std::string req_tablet_id = req->param(TABLET_ID_KEY);

    std::string req_table_id = req->param(TABLE_ID_KEY);

    if (req_tablet_id == "") {

        if (req_table_id == "") {

            // both tablet id and table id are empty, return error.

            return Status::InternalError(

                    "tablet id and table id can not be empty at the same time!");

        } else {

            try {

                *table_id = std::stoull(req_table_id);

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

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

            }

            return Status::OK();

        }

    } else {

        if (req_table_id == "") {

            try {

                *tablet_id = std::stoull(req_tablet_id);

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

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

            }

            return Status::OK();

        } else {

            // both tablet id and table id are not empty, return err.

            return Status::InternalError("tablet id and table id can not be set at the same time!");

        }

    }

}

/// retrieve specific id from req

static Status _check_param(HttpRequest* req, uint64_t* id_param, const std::string param_name) {

    const auto& req_id_param = req->param(param_name);

    if (!req_id_param.empty()) {

        try {

            *id_param = std::stoull(req_id_param);

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

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

        }

    }

    return Status::OK();

}

// for viewing the compaction status

Status CloudCompactionAction::_handle_show_compaction(HttpRequest* req, std::string* json_result) {

    uint64_t tablet_id = 0;

    RETURN_NOT_OK_STATUS_WITH_WARN(_check_param(req, &tablet_id, TABLET_ID_KEY),

                                   "check param failed");

    if (tablet_id == 0) {

        return Status::InternalError("check param failed: missing tablet_id");

    }

    LOG(INFO) << "begin to handle show compaction, tablet id: " << tablet_id;

    //TabletSharedPtr tablet = _engine.tablet_manager()->get_tablet(tablet_id);

    CloudTabletSPtr tablet = DORIS_TRY(_engine.tablet_mgr().get_tablet(tablet_id));

    if (tablet == nullptr) {

        return Status::NotFound("Tablet not found. tablet_id={}", tablet_id);

    }

    tablet->get_compaction_status(json_result);

    LOG(INFO) << "finished to handle show compaction, tablet id: " << tablet_id;

    return Status::OK();

}

Status CloudCompactionAction::_handle_run_compaction(HttpRequest* req, std::string* json_result) {

    // 1. param check

    // check req_tablet_id or req_table_id is not empty and can not be set together.

    uint64_t tablet_id = 0;

    uint64_t table_id = 0;

    RETURN_NOT_OK_STATUS_WITH_WARN(_check_param(req, &tablet_id, &table_id), "check param failed");

    LOG(INFO) << "begin to handle run compaction, tablet id: " << tablet_id

              << " table id: " << table_id;

    // check compaction_type equals 'base' or 'cumulative'

    auto& compaction_type = req->param(PARAM_COMPACTION_TYPE);

    if (compaction_type == PARAM_COMPACTION_ROW_BINLOG_TTL) {

        if (tablet_id == 0 || table_id != 0) {

            return Status::InvalidArgument("row_binlog_ttl requires a tablet_id");

        }

        RETURN_IF_ERROR(_engine.submit_row_binlog_ttl(tablet_id, true));

        *json_result = R"({"status":"Success","msg":"ROW binlog TTL task queued"})";

        return Status::OK();

    }

    if (compaction_type != PARAM_COMPACTION_BASE &&

        compaction_type != PARAM_COMPACTION_CUMULATIVE &&

        compaction_type != PARAM_COMPACTION_FULL) {

        return Status::NotSupported("The compaction type '{}' is not supported", compaction_type);

    }

    bool sync_delete_bitmap = compaction_type != PARAM_COMPACTION_FULL;

    CloudTabletSPtr tablet =

            DORIS_TRY(_engine.tablet_mgr().get_tablet(tablet_id, false, sync_delete_bitmap));

    if (tablet == nullptr) {

        return Status::NotFound("Tablet not found. tablet_id={}", tablet_id);

    }

    if (compaction_type == PARAM_COMPACTION_BASE) {

        tablet->set_last_base_compaction_schedule_time(UnixMillis());

    } else if (compaction_type == PARAM_COMPACTION_CUMULATIVE) {

        tablet->set_last_cumu_compaction_schedule_time(UnixMillis());

    } else if (compaction_type == PARAM_COMPACTION_FULL) {

        tablet->set_last_full_compaction_schedule_time(UnixMillis());

    }

    LOG(INFO) << "manual submit compaction task, tablet id: " << tablet_id

              << " table id: " << table_id;

    // 3. submit compaction task (trigger_method=1 for MANUAL)

    RETURN_IF_ERROR(_engine.submit_compaction_task(

            tablet,

            compaction_type == PARAM_COMPACTION_BASE         ? CompactionType::BASE_COMPACTION

            : compaction_type == PARAM_COMPACTION_CUMULATIVE ? CompactionType::CUMULATIVE_COMPACTION

                                                             : CompactionType::FULL_COMPACTION,

            /*trigger_method=*/1));

    LOG(INFO) << "Manual compaction task is successfully triggered, tablet id: " << tablet_id

              << " table id: " << table_id;

    *json_result =

            R"({"status": "Success", "msg": "compaction task is successfully triggered. Table id: )" +

            std::to_string(table_id) + ". Tablet id: " + std::to_string(tablet_id) + "\"}";

    return Status::OK();

}

Status CloudCompactionAction::_handle_run_status_compaction(HttpRequest* req,

                                                            std::string* json_result) {

    uint64_t tablet_id = 0;

    RETURN_NOT_OK_STATUS_WITH_WARN(_check_param(req, &tablet_id, TABLET_ID_KEY),

                                   "check param failed");

    LOG(INFO) << "begin to handle run status compaction, tablet id: " << tablet_id;

    if (tablet_id == 0) {

        // overall compaction status

        RETURN_IF_ERROR(_engine.get_compaction_status_json(json_result));

    } else {

        std::string json_template = R"({

            "status" : "Success",

            "run_status" : $0,

            "msg" : "$1",

            "tablet_id" : $2,

            "compact_type" : "$3"

        })";

        std::string msg = "compaction task for this tablet is not running";

        std::string compaction_type;

        bool run_status = false;

        if (_engine.has_cumu_compaction(tablet_id)) {

            msg = "compaction task for this tablet is running";

            compaction_type = "cumulative";

            run_status = true;

            *json_result =

                    absl::Substitute(json_template, run_status, msg, tablet_id, compaction_type);

            return Status::OK();

        }

        if (_engine.has_base_compaction(tablet_id)) {

            msg = "compaction task for this tablet is running";

            compaction_type = "base";

            run_status = true;

            *json_result =

                    absl::Substitute(json_template, run_status, msg, tablet_id, compaction_type);

            return Status::OK();

        }

        if (_engine.has_full_compaction(tablet_id)) {

            msg = "compaction task for this tablet is running";

            compaction_type = "full";

            run_status = true;

            *json_result =

                    absl::Substitute(json_template, run_status, msg, tablet_id, compaction_type);

            return Status::OK();

        }

        // not running any compaction

        *json_result = absl::Substitute(json_template, run_status, msg, tablet_id, compaction_type);

    }

    LOG(INFO) << "finished to handle run status compaction, tablet id: " << tablet_id;

    return Status::OK();

}

void CloudCompactionAction::handle(HttpRequest* req) {

    req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.c_str());

    if (_compaction_type == CompactionActionType::SHOW_INFO) {

        std::string json_result;

        Status st = _handle_show_compaction(req, &json_result);

        if (!st.ok()) {

            HttpChannel::send_reply(req, HttpStatus::OK, st.to_json());

        } else {

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

        }

    } else if (_compaction_type == CompactionActionType::RUN_COMPACTION) {

        std::string json_result;

        Status st = _handle_run_compaction(req, &json_result);

        if (!st.ok()) {

            HttpChannel::send_reply(req, HttpStatus::OK, st.to_json());

        } else {

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

        }

    } else {

        std::string json_result;

        Status st = _handle_run_status_compaction(req, &json_result);

        if (!st.ok()) {

            HttpChannel::send_reply(req, HttpStatus::OK, st.to_json());

        } else {

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

        }

    }

}

} // end namespace doris