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

#include <glog/logging.h>

#include <unistd.h>

#include <memory>

#include <ostream>

#include "common/config.h"

#include "common/status.h"

#include "http/http_headers.h"

#include "runtime/exec_env.h"

#include "util/security.h"

#include "util/stack_util.h"

namespace doris {

class MultiFileSplitter {

public:

    MultiFileSplitter(std::string local_dir, std::unordered_set<std::string> expected_files)

            : _local_dir_path(std::move(local_dir)), _expected_files(std::move(expected_files)) {}

    ~MultiFileSplitter() {

        if (_fd >= 0) {

            close(_fd);

        }

        if (!_status.ok() && !downloaded_files.empty()) {

            LOG(WARNING) << "download files to " << _local_dir_path << " failed, try remove the "

                         << downloaded_files.size() << " downloaded files";

            for (const auto& file : downloaded_files) {

                remove(file.c_str());

            }

        }

    }

    bool append(const char* data, size_t length) {

        // Already failed.

        if (!_status.ok()) {

            return false;

        }

        std::string buf;

        if (!_buffer.empty()) {

            buf.swap(_buffer);

            buf.append(data, length);

            data = buf.data();

            length = buf.size();

        }

        return append_inner(data, length);

    }

    Status finish() {

        if (_status.ok()) {

            _status = finish_inner();

        }

        return _status;

    }

private:

    bool append_inner(const char* data, size_t length) {

        while (length > 0) {

            int consumed = 0;

            if (_is_reading_header) {

                consumed = parse_header(data, length);

            } else {

                consumed = append_file(data, length);

            }

            if (consumed < 0) {

                return false;

            }

            DCHECK(consumed <= length);

            data += consumed;

            length -= consumed;

        }

        return true;

    }

    int parse_header(const char* data, size_t length) {

        DCHECK(_fd < 0);

        std::string_view buf(data, length);

        size_t pos = buf.find("\r\n\r\n");

        if (pos == std::string::npos) {

            _buffer.append(data, length);

            return static_cast<int>(length);

        }

        // header already read.

        _is_reading_header = false;

        bool has_file_name = false;

        bool has_file_size = false;

        std::string_view header = buf.substr(0, pos);

        std::vector<std::string> headers =

                strings::Split(header, "\r\n", strings::SkipWhitespace());

        for (auto& s : headers) {

            size_t header_pos = s.find(':');

            if (header_pos == std::string::npos) {

                continue;

            }

            std::string_view header_view(s);

            std::string_view key = header_view.substr(0, header_pos);

            std::string_view value = header_view.substr(header_pos + 1);

            if (value.starts_with(' ')) {

                value.remove_prefix(std::min(value.find_first_not_of(' '), value.size()));

            }

            if (key == "File-Name") {

                _file_name = value;

                has_file_name = true;

            } else if (key == "Content-Length") {

                auto res = std::from_chars(value.data(), value.data() + value.size(), _file_size);

                if (res.ec != std::errc()) {

                    std::string error_msg = fmt::format("invalid content length: {}", value);

                    LOG(WARNING) << "download files to " << _local_dir_path

                                 << "failed, err=" << error_msg;

                    _status = Status::HttpError(std::move(error_msg));

                    return -1;

                }

                has_file_size = true;

            }

        }

        if (!has_file_name || !has_file_size) {

            std::string error_msg =

                    fmt::format("invalid multi part header, has file name: {}, has file size: {}",

                                has_file_name, has_file_size);

            LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg;

            _status = Status::HttpError(std::move(error_msg));

            return -1;

        }

        if (!_expected_files.contains(_file_name)) {

            std::string error_msg = fmt::format("unexpected file: {}", _file_name);

            LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg;

            _status = Status::HttpError(std::move(error_msg));

            return -1;

        }

        VLOG_DEBUG << "receive file " << _file_name << ", size " << _file_size;

        _written_size = 0;

        _local_file_path = fmt::format("{}/{}", _local_dir_path, _file_name);

        _fd = open(_local_file_path.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0644);

        if (_fd < 0) {

            std::string error_msg = "fail to open file to write: " + _local_file_path;

            LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg;

            _status = Status::IOError(std::move(error_msg));

            return -1;

        }

        downloaded_files.push_back(_local_file_path);

        return static_cast<int>(pos + 4);

    }

    int append_file(const char* data, size_t length) {

        DCHECK(_fd >= 0);

        DCHECK(_file_size >= _written_size);

        size_t write_size = std::min(length, _file_size - _written_size);

        if (write_size > 0 && write(_fd, data, write_size) < 0) {

            auto msg = fmt::format("write file failed, file={}, error={}", _local_file_path,

                                   strerror(errno));

            LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << msg;

            _status = Status::HttpError(std::move(msg));

            return -1;

        }

        _written_size += write_size;

        if (_written_size == _file_size) {

            // This file has been downloaded, switch to the next one.

            switch_to_next_file();

        }

        return write_size;

    }

    Status finish_inner() {

        if (!_is_reading_header && _written_size == _file_size) {

            switch_to_next_file();

        }

        if (_fd >= 0) {

            // This file is not completely downloaded.

            close(_fd);

            _fd = -1;

            auto error_msg = fmt::format("file {} is not completely downloaded", _local_file_path);

            LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg;

            return Status::HttpError(std::move(error_msg));

        }

        if (!_expected_files.empty()) {

            auto error_msg = fmt::format("not all files are downloaded, {} missing files",

                                         _expected_files.size());

            LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg;

            return Status::HttpError(std::move(error_msg));

        }

        downloaded_files.clear();

        return Status::OK();

    }

    void switch_to_next_file() {

        DCHECK(_fd >= 0);

        DCHECK(_written_size == _file_size);

        close(_fd);

        _fd = -1;

        _expected_files.erase(_file_name);

        _is_reading_header = true;

    }

    const std::string _local_dir_path;

    std::string _buffer;

    std::unordered_set<std::string> _expected_files;

    Status _status;

    bool _is_reading_header = true;

    int _fd = -1;

    std::string _local_file_path;

    std::string _file_name;

    size_t _file_size = 0;

    size_t _written_size = 0;

    std::vector<std::string> downloaded_files;

};

static const char* header_error_msg(CURLHcode code) {

    switch (code) {

    case CURLHE_OK:

        return "OK";

    case CURLHE_BADINDEX:

        return "header exists but not with this index ";

    case CURLHE_MISSING:

        return "no such header exists";

    case CURLHE_NOHEADERS:

        return "no headers at all exist (yet)";

    case CURLHE_NOREQUEST:

        return "no request with this number was used";

    case CURLHE_OUT_OF_MEMORY:

        return "out of memory while processing";

    case CURLHE_BAD_ARGUMENT:

        return "a function argument was not okay";

    case CURLHE_NOT_BUILT_IN:

        return "curl_easy_header() was disabled in the build";

    default:

        return "unknown";

    }

}

HttpClient::HttpClient() = default;

HttpClient::~HttpClient() {

    if (_curl != nullptr) {

        curl_easy_cleanup(_curl);

        _curl = nullptr;

    }

    if (_header_list != nullptr) {

        curl_slist_free_all(_header_list);

        _header_list = nullptr;

    }

}

Status HttpClient::init(const std::string& url, bool set_fail_on_error) {

    if (_curl == nullptr) {

        _curl = curl_easy_init();

        if (_curl == nullptr) {

            return Status::InternalError("fail to initialize curl");

        }

    } else {

        curl_easy_reset(_curl);

    }

    if (_header_list != nullptr) {

        curl_slist_free_all(_header_list);

        _header_list = nullptr;

    }

    // set error_buf

    _error_buf[0] = 0;

    auto code = curl_easy_setopt(_curl, CURLOPT_ERRORBUFFER, _error_buf);

    if (code != CURLE_OK) {

        LOG(WARNING) << "fail to set CURLOPT_ERRORBUFFER, msg=" << _to_errmsg(code);

        return Status::InternalError("fail to set error buffer");

    }

    // forbid signals

    code = curl_easy_setopt(_curl, CURLOPT_NOSIGNAL, 1L);

    if (code != CURLE_OK) {

        LOG(WARNING) << "fail to set CURLOPT_NOSIGNAL, msg=" << _to_errmsg(code);

        return Status::InternalError("fail to set CURLOPT_NOSIGNAL");

    }

    // set fail on error

    // When this option is set to `1L` (enabled), libcurl will return an error directly

    // when encountering HTTP error codes (>= 400), without reading the body of the error response.

    if (set_fail_on_error) {

        code = curl_easy_setopt(_curl, CURLOPT_FAILONERROR, 1L);

        if (code != CURLE_OK) {

            LOG(WARNING) << "fail to set CURLOPT_FAILONERROR, msg=" << _to_errmsg(code);

            return Status::InternalError("fail to set CURLOPT_FAILONERROR");

        }

    }

    // set redirect

    code = curl_easy_setopt(_curl, CURLOPT_FOLLOWLOCATION, 1L);

    if (code != CURLE_OK) {

        LOG(WARNING) << "fail to set CURLOPT_FOLLOWLOCATION, msg=" << _to_errmsg(code);

        return Status::InternalError("fail to set CURLOPT_FOLLOWLOCATION");

    }

    code = curl_easy_setopt(_curl, CURLOPT_MAXREDIRS, 20);

    if (code != CURLE_OK) {

        LOG(WARNING) << "fail to set CURLOPT_MAXREDIRS, msg=" << _to_errmsg(code);

        return Status::InternalError("fail to set CURLOPT_MAXREDIRS");

    }

    curl_write_callback callback = [](char* buffer, size_t size, size_t nmemb, void* param) {

        auto* client = (HttpClient*)param;

        return client->on_response_data(buffer, size * nmemb);

    };

    // set callback function

    code = curl_easy_setopt(_curl, CURLOPT_WRITEFUNCTION, callback);

    if (code != CURLE_OK) {

        LOG(WARNING) << "fail to set CURLOPT_WRITEFUNCTION, msg=" << _to_errmsg(code);

        return Status::InternalError("fail to set CURLOPT_WRITEFUNCTION");

    }

    code = curl_easy_setopt(_curl, CURLOPT_WRITEDATA, (void*)this);

    if (code != CURLE_OK) {

        LOG(WARNING) << "fail to set CURLOPT_WRITEDATA, msg=" << _to_errmsg(code);

        return Status::InternalError("fail to set CURLOPT_WRITEDATA");

    }

    std::string escaped_url;

    RETURN_IF_ERROR(_escape_url(url, &escaped_url));

    // set url

    code = curl_easy_setopt(_curl, CURLOPT_URL, escaped_url.c_str());

    if (code != CURLE_OK) {

        LOG(WARNING) << "failed to set CURLOPT_URL, errmsg=" << _to_errmsg(code);

        return Status::InternalError("fail to set CURLOPT_URL");

    }

#ifndef BE_TEST

    set_auth_token(ExecEnv::GetInstance()->cluster_info()->curr_auth_token);

#endif

    return Status::OK();

}

void HttpClient::set_method(HttpMethod method) {

    _method = method;

    switch (method) {

    case GET:

        curl_easy_setopt(_curl, CURLOPT_HTTPGET, 1L);

        return;

    case PUT:

        curl_easy_setopt(_curl, CURLOPT_UPLOAD, 1L);

        return;

    case POST:

        curl_easy_setopt(_curl, CURLOPT_POST, 1L);

        return;

    case DELETE:

        curl_easy_setopt(_curl, CURLOPT_CUSTOMREQUEST, "DELETE");

        return;

    case HEAD:

        curl_easy_setopt(_curl, CURLOPT_NOBODY, 1L);

        return;

    case OPTIONS:

        curl_easy_setopt(_curl, CURLOPT_CUSTOMREQUEST, "OPTIONS");

        return;

    default:

        return;

    }

}

void HttpClient::set_speed_limit() {

    curl_easy_setopt(_curl, CURLOPT_LOW_SPEED_LIMIT, config::download_low_speed_limit_kbps * 1024);

    curl_easy_setopt(_curl, CURLOPT_LOW_SPEED_TIME, config::download_low_speed_time);

    curl_easy_setopt(_curl, CURLOPT_MAX_RECV_SPEED_LARGE, config::max_download_speed_kbps * 1024);

}

size_t HttpClient::on_response_data(const void* data, size_t length) {

    if (*_callback != nullptr) {

        bool is_continue = (*_callback)(data, length);

        if (!is_continue) {

            return -1;

        }

    }

    return length;

}

Status HttpClient::execute_post_request(const std::string& payload, std::string* response) {

    set_method(POST);

    set_payload(payload);

    return execute(response);

}

Status HttpClient::execute_delete_request(const std::string& payload, std::string* response) {

    set_method(DELETE);

    set_payload(payload);

    return execute(response);

}

Status HttpClient::execute(const std::function<bool(const void* data, size_t length)>& callback) {

    if (VLOG_DEBUG_IS_ON) {

        VLOG_DEBUG << "execute http " << to_method_desc(_method) << " request, url " << _get_url();

    }

    _callback = &callback;

    auto code = curl_easy_perform(_curl);

    if (code != CURLE_OK) {

        std::string url = mask_token(_get_url());

        LOG(WARNING) << "fail to execute HTTP client, errmsg=" << _to_errmsg(code)

                     << ", trace=" << get_stack_trace() << ", url=" << url;

        std::string errmsg = fmt::format("{}, url={}", _to_errmsg(code), url);

        return Status::HttpError(std::move(errmsg));

    }

    if (VLOG_DEBUG_IS_ON) {

        VLOG_DEBUG << "execute http " << to_method_desc(_method) << " request, url " << _get_url()

                   << " done";

    }

    return Status::OK();

}

Status HttpClient::get_content_md5(std::string* md5) const {

    struct curl_header* header_ptr;

    auto code = curl_easy_header(_curl, HttpHeaders::CONTENT_MD5, 0, CURLH_HEADER, 0, &header_ptr);

    if (code == CURLHE_MISSING || code == CURLHE_NOHEADERS) {

        // no such headers exists

        md5->clear();

        return Status::OK();

    } else if (code != CURLHE_OK) {

        auto msg = fmt::format("failed to get http header {}: {} ({})", HttpHeaders::CONTENT_MD5,

                               header_error_msg(code), code);

        LOG(WARNING) << msg << ", trace=" << get_stack_trace();

        return Status::HttpError(std::move(msg));

    }

    *md5 = header_ptr->value;

    return Status::OK();

}

Status HttpClient::download(const std::string& local_path) {

    set_method(GET);

    set_speed_limit();

    auto fp_closer = [](FILE* fp) { fclose(fp); };

    std::unique_ptr<FILE, decltype(fp_closer)> fp(fopen(local_path.c_str(), "w"), fp_closer);

    if (fp == nullptr) {

        LOG(WARNING) << "open file failed, file=" << local_path;

        return Status::InternalError("open file failed");

    }

    Status status;

    auto callback = [&status, &fp, &local_path](const void* data, size_t length) {

        auto res = fwrite(data, length, 1, fp.get());

        if (res != 1) {

            LOG(WARNING) << "fail to write data to file, file=" << local_path

                         << ", error=" << ferror(fp.get());

            status = Status::InternalError("fail to write data when download");

            return false;

        }

        return true;

    };

    if (auto s = execute(callback); !s.ok()) {

        status = s;

    }

    if (!status.ok()) {

        remove(local_path.c_str());

    }

    return status;

}

Status HttpClient::download_multi_files(const std::string& local_dir,

                                        const std::unordered_set<std::string>& expected_files) {

    set_speed_limit();

    MultiFileSplitter splitter(local_dir, expected_files);

    auto callback = [&](const void* data, size_t length) {

        return splitter.append(reinterpret_cast<const char*>(data), length);

    };

    if (auto s = execute(callback); !s.ok()) {

        return s;

    }

    return splitter.finish();

}

Status HttpClient::execute(std::string* response) {

    auto callback = [response](const void* data, size_t length) {

        response->append((char*)data, length);

        return true;

    };

    return execute(callback);

}

const char* HttpClient::_to_errmsg(CURLcode code) const {

    if (_error_buf[0] == 0) {

        return curl_easy_strerror(code);

    }

    return _error_buf;

}

const char* HttpClient::_get_url() const {

    const char* url = nullptr;

    curl_easy_getinfo(_curl, CURLINFO_EFFECTIVE_URL, &url);

    if (!url) {

        url = "<unknown>";

    }

    return url;

}

// execute remote call action with retry

Status HttpClient::execute(int retry_times, int sleep_time,

                           const std::function<Status(HttpClient*)>& callback) {

    Status status;

    for (int i = 0; i < retry_times; ++i) {

        status = callback(this);

        if (status.ok()) {

            auto http_status = get_http_status();

            if (http_status == 200) {

                return status;

            } else {

                std::string url = mask_token(_get_url());

                auto error_msg = fmt::format("http status code is not 200, code={}, url={}",

                                             http_status, url);

                LOG(WARNING) << error_msg;

                return Status::HttpError(error_msg);

            }

        }

        sleep(sleep_time);

    }

    return status;

}

Status HttpClient::execute_with_retry(int retry_times, int sleep_time,

                                      const std::function<Status(HttpClient*)>& callback) {

    Status status;

    for (int i = 0; i < retry_times; ++i) {

        HttpClient client;

        status = callback(&client);

        if (status.ok()) {

            auto http_status = client.get_http_status();

            if (http_status == 200) {

                return status;

            } else {

                std::string url = mask_token(client._get_url());

                auto error_msg = fmt::format("http status code is not 200, code={}, url={}",

                                             http_status, url);

                LOG(WARNING) << error_msg;

                return Status::HttpError(error_msg);

            }

        }

        sleep(sleep_time);

    }

    return status;

}

// http://example.com/page?param1=value1&param2=value+with+spaces#section

Status HttpClient::_escape_url(const std::string& url, std::string* escaped_url) {

    size_t query_pos = url.find('?');

    if (query_pos == std::string::npos) {

        *escaped_url = url;

        return Status::OK();

    }

    size_t fragment_pos = url.find('#');

    std::string query;

    std::string fragment;

    if (fragment_pos == std::string::npos) {

        query = url.substr(query_pos + 1, url.length() - query_pos - 1);

    } else {

        query = url.substr(query_pos + 1, fragment_pos - query_pos - 1);

        fragment = url.substr(fragment_pos, url.length() - fragment_pos);

    }

    std::string encoded_query;

    size_t ampersand_pos = query.find('&');

    size_t equal_pos;

    if (ampersand_pos == std::string::npos) {

        ampersand_pos = query.length();

    }

    while (true) {

        equal_pos = query.find('=');

        if (equal_pos != std::string::npos) {

            std::string key = query.substr(0, equal_pos);

            std::string value = query.substr(equal_pos + 1, ampersand_pos - equal_pos - 1);

            auto encoded_value = std::unique_ptr<char, decltype(&curl_free)>(

                    curl_easy_escape(_curl, value.c_str(), value.length()), &curl_free);

            if (encoded_value) {

                encoded_query += key + "=" + std::string(encoded_value.get());

            } else {

                return Status::InternalError("escape url failed, url={}", url);

            }

        } else {

            encoded_query += query.substr(0, ampersand_pos);

        }

        if (ampersand_pos == query.length() || ampersand_pos == std::string::npos) {

            break;

        }

        encoded_query += "&";

        query = query.substr(ampersand_pos + 1);

        ampersand_pos = query.find('&');

    }

    *escaped_url = url.substr(0, query_pos + 1) + encoded_query + fragment;

    return Status::OK();

}

} // namespace doris