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

#include <event2/buffer.h>
#include <event2/http.h>
#include <event2/http_struct.h>
#include <event2/keyvalq_struct.h>

#include <memory>
#include <sstream>
#include <string>
#include <unordered_map>
#include <utility>

#include "http/http_handler.h"
#include "http/http_headers.h"
#include "runtime/stream_load/stream_load_context.h"
#include "util/stack_util.h"

namespace doris {

static std::string s_empty = "";

HttpRequest::HttpRequest(evhttp_request* evhttp_request) : _ev_req(evhttp_request) {}

HttpRequest::~HttpRequest() {
    if (_handler_ctx != nullptr) {
        DCHECK(_handler != nullptr);
        _handler->free_handler_ctx(_handler_ctx);
    }
}

int HttpRequest::init_from_evhttp() {
    _method = to_http_method(evhttp_request_get_command(_ev_req));
    if (_method == HttpMethod::UNKNOWN) {
        LOG(WARNING) << "unknown method of HTTP request, method="
                     << evhttp_request_get_command(_ev_req);
        return -1;
    }
    _uri = evhttp_request_get_uri(_ev_req);
    // conver header
    auto headers = evhttp_request_get_input_headers(_ev_req);
    for (auto header = headers->tqh_first; header != nullptr; header = header->next.tqe_next) {
        _headers.emplace(header->key, header->value);
    }
    // parse
    auto ev_uri = evhttp_request_get_evhttp_uri(_ev_req);
    _raw_path = evhttp_uri_get_path(ev_uri);
    auto query = evhttp_uri_get_query(ev_uri);
    if (query == nullptr || *query == '\0') {
        return 0;
    }
    struct evkeyvalq params;
    auto res = evhttp_parse_query_str(query, &params);
    if (res < 0) {
        LOG(WARNING) << "parse query str failed, query=" << query;
        return res;
    }
    for (auto param = params.tqh_first; param != nullptr; param = param->next.tqe_next) {
        _query_params.emplace(param->key, param->value);
    }
    _params.insert(_query_params.begin(), _query_params.end());
    evhttp_clear_headers(&params);
    return 0;
}

std::string HttpRequest::debug_string() const {
    std::stringstream ss;
    ss << "HttpRequest: \n"
       << "method:" << _method << "\n"
       << "uri:" << _uri << "\n"
       << "raw_path:" << _raw_path << "\n"
       << "headers: \n";
    for (auto& iter : _headers) {
        if (iequal(iter.first, HttpHeaders::AUTHORIZATION) ||
            iequal(iter.first, HttpHeaders::PROXY_AUTHORIZATION)) {
            ss << "key=" << iter.first << ", value=***MASKED***\n";
        } else {
            ss << "key=" << iter.first << ", value=" << iter.second << "\n";
        }
    }
    ss << "params: \n";
    for (auto& iter : _params) {
        ss << "key=" << iter.first << ", value=" << iter.second << "\n";
    }

    return ss.str();
}

const std::string& HttpRequest::header(const std::string& key) const {
    auto iter = _headers.find(key);
    if (iter == _headers.end()) {
        return s_empty;
    }
    return iter->second;
}

const std::string& HttpRequest::param(const std::string& key) const {
    auto iter = _params.find(key);
    if (iter == _params.end()) {
        return s_empty;
    }
    return iter->second;
}

std::string HttpRequest::get_all_headers() const {
    std::stringstream headers;
    for (const auto& header : _headers) {
        // Mask sensitive headers
        if (iequal(header.first, HttpHeaders::AUTHORIZATION) ||
            iequal(header.first, HttpHeaders::PROXY_AUTHORIZATION)) {
            headers << header.first << ":***MASKED***, ";
        } else {
            headers << header.first << ":" << header.second + ", ";
        }
    }
    return headers.str();
}

void HttpRequest::add_output_header(const char* key, const char* value) {
    evhttp_add_header(evhttp_request_get_output_headers(_ev_req), key, value);
}

std::string HttpRequest::get_request_body() {
    if (!_request_body.empty()) {
        return _request_body;
    }
    // read buf
    auto evbuf = evhttp_request_get_input_buffer(_ev_req);
    if (evbuf == nullptr) {
        return _request_body;
    }
    auto length = evbuffer_get_length(evbuf);
    _request_body.resize(length);
    evbuffer_remove(evbuf, (char*)_request_body.data(), length);
    return _request_body;
}

const char* HttpRequest::remote_host() const {
    return _ev_req->remote_host;
}

void HttpRequest::finish_send_reply() {
    if (_send_reply_type == REPLY_SYNC) {
        return;
    }

    std::string infos;
    if (_handler_ctx != nullptr) {
        infos = reinterpret_cast<StreamLoadContext*>(_handler_ctx.get())->brief();
    }
    _http_reply_promise.set_value(true);
}

void HttpRequest::wait_finish_send_reply() {
    if (_send_reply_type == REPLY_SYNC) {
        return;
    }

    std::string infos;
    StreamLoadContext* ctx = nullptr;
    if (_handler_ctx != nullptr) {
        ctx = reinterpret_cast<StreamLoadContext*>(_handler_ctx.get());
        infos = ctx->brief();
        _handler->free_handler_ctx(_handler_ctx);
    }

    VLOG_NOTICE << "start to wait send reply, infos=" << infos;
    auto status = _http_reply_future.wait_for(std::chrono::seconds(config::async_reply_timeout_s));
    // if request is timeout and can't cancel fragment in time, it will cause some new request block
    // so we will free cancelled request in time.
    if (status != std::future_status::ready) {
        LOG(WARNING) << "wait for send reply timeout, " << this->debug_string();
        std::unique_lock<std::mutex> lock1(ctx->_send_reply_lock);
        // do not send_reply after free current request
        ctx->_can_send_reply = false;
        ctx->_finish_send_reply = true;
        ctx->_can_send_reply_cv.notify_all();
    } else {
        VLOG_NOTICE << "wait send reply finished";
    }

    // delete _handler_ctx at the end, in case that finish_send_reply can't get detailed info
    _handler_ctx = nullptr;
}

} // namespace doris