/**
 * Copyright 2014 (c) Baidu, Inc.
 *
 * Licensed 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 "bcesdk/bos/client_impl.h"
#include "bcesdk/util/util.h"
#include "bcesdk/http/http_client.h"

#ifndef _WIN32
#include "bcesdk/http/unix_curl_global.h"
#else
#include <Winsock2.h>
#pragma comment(lib, "Ws2_32.lib")
#include "bcesdk/http/win_curl_global.h"
#endif

BEGIN_CPPSDK_NAMESPACE

ClientImpl::ClientImpl(const Credential &credential, const ClientOptions &options)
    : _options(options) {
    DefaultSigner *signer = new DefaultSigner(options);
    signer->set_credential(credential);
    _signer = signer;
    if (_options.user_agent.empty()) {
        _options.user_agent = sdk_package_string();
    }
    if (_options.is_use_curl_pool) {
        _pool = new EasyCurlPool(_options.curl_pool_size < g_curl_pool_min_size ? g_curl_pool_min_size : _options.curl_pool_size);
        _pool->fillin();
    } else {
        _pool = nullptr;
    }
}

ClientImpl::~ClientImpl()
{
   if (_signer != NULL) {
       delete _signer;
   }
   if (_pool != NULL) {
       delete _pool;
   }
}

std::string ClientImpl::generate_url(BceRequest &request, int expire_seconds) {
    HttpRequest http_request;
    bool is_change_to_cname = false;
    is_change_to_cname = http_request.set_endpoint(_options.endpoint, _options, request.bucket_name());
    http_request.append_header("Host", http_request.host());
    int ret = request.build_http_request(&http_request, is_change_to_cname);
    if (ret != 0) {
        LOG(ERROR) << "generate url failed due to build http request failed, ret:" << ret;
        return "";
    }
    http_request.add_parameter("authorization",
        _signer->generate_auth(&http_request, expire_seconds));
    return http_request.generate_url();
}

int ClientImpl::build_http_request(BceRequest &request, HttpRequest *http_request, bool switch_endpoint) {
    bool is_change_to_cname = false;
    if (switch_endpoint) {
        is_change_to_cname = http_request->set_endpoint(_options.backup_endpoint, _options, request.bucket_name());
    }
    else {
        is_change_to_cname = http_request->set_endpoint(_options.endpoint, _options, request.bucket_name());
    }
    http_request->set_timeout(_options.timeout);
    http_request->set_connect_timeout_ms(_options.connect_timeout_ms);
    http_request->set_local_interface(_options.local_interface);
    http_request->set_local_port(_options.local_port);
    http_request->set_local_port_range(_options.local_port_range);
    http_request->append_header("Host", http_request->host());
    http_request->append_header("User-Agent", _options.user_agent);
    int ret = request.build_http_request(http_request, is_change_to_cname);
    if (ret != 0) {
        return ret;
    }
    http_request->append_header("x-bce-date", TimeUtil::now_utctime());
    if (!http_request->is_post_object_request()){
        _signer->sign(http_request);
    }
    http_request->set_is_cancelable(request.get_request_iscancelable());
    if (request.get_request_iscancelable() && _options.client_enable_cancel){
        http_request->set_father_bcerequest(&request);
    }
    else if (request.get_request_iscancelable() && !_options.client_enable_cancel){
        LOG(WARN) << "Client disabled cancel but request is setted to be cancelble\n";
    }
    return 0;
}


int ClientImpl::build_http_response(BceResponse *response, HttpResponse* http_response) {
    return response->fill_http_response(http_response);
}

int ClientImpl::send_request(BceRequest &request, BceResponse *response) {
    HttpResponse *http_response = response->mutable_http_response();
    int ret = 0;
    int try_num = 1;
    if (_options.retry > 0) {
        try_num += _options.retry;
    }
    HttpClient client;
    client.set_options(&_options);
    if (_options.is_use_curl_pool) {
        client.set_curl_pool(_pool);
    }

    bool switched_to_backup_endpoint = false;
    while (try_num-- > 0) {
        HttpRequest http_request;
        ret = build_http_request(request, &http_request, switched_to_backup_endpoint);
        build_http_response(response, http_response);
        if (RET_OK != ret) {
            LOG(ERROR) << "build http request failed, ret:" << ret;
            return ret;
        }
        ret = client.execute(http_request, http_response);
        if (ret == 0) {
            response->handle_response(*http_response);
        } else {
            std::string curl_error = stringfy_ret_code(ret);
            response->set_error(STATUS_RPC_FAIL, "HttpExecuteFailure", "", curl_error);
            LOG(WARN) << "http execute error: (" << ret << ')' << stringfy_ret_code(ret);
            // correct the ret code to keep compatibility
            if (ret != RET_INIT_CURL_FAIL){
                ret = RET_CLIENT_ERROR;
            }    
        }

        if (response->is_ok()) {
            return 0;
        }
        ret = response->status_code();
        
        // check if it is copy request and need retry
        bool is_copy_request = false;
        const struct curl_slist* current = http_request.header_list();
        while (current != NULL) {
            if (std::string(current->data).find("x-bce-copy-source") != std::string::npos) {
                is_copy_request = true;
                break;
            }
            current = current->next;
        }
        if (is_copy_request && ret == 500 && response->error().code() == "InternalError") {
            continue;
        }
        else if (is_copy_request && ret == 500) {
            break;
        }

        if (ret >= 500 && ret != 501) {
            if (try_num > 0) {
                InputStream *istream = http_request.get_input_stream();
                if (istream != NULL) {
                    istream->seek(0);
                }
                http_response->reset();
                LOG(WARN) << "send_request do retry. last ret status:" 
                          << ret << " retry count:" << try_num;
                continue;
            }
            else if (_options.enable_auto_switch_endpoint && !switched_to_backup_endpoint){
                InputStream *istream = http_request.get_input_stream();
                if (istream != NULL) {
                    istream->seek(0);
                }
                http_response->reset();
                LOG(WARN) << "send_request do switch endpoint retry. last ret status:" 
                          << ret << " retry count:" << try_num;
                switched_to_backup_endpoint = true;
                try_num = _options.retry + 1;
                continue;
            }
        }
        break;
    }
    return ret;
}

int ClientImpl::send_request(int n, BceRequestContext ctx[], int max_parallel, const std::atomic_bool * const flow_control) {
    int ret = 0;
    std::vector<HttpRequest> http_requests_buffer(n);
    std::vector<HttpRequestContext> http_ctx(n);
    if (max_parallel <= 0) {
        max_parallel = _options.max_parallel;
        if (max_parallel <= 0) {
            max_parallel = 10;
        }
    }
    max_parallel = n > max_parallel ? max_parallel : n;
    int i = 0;
    for (i = 0; i < n; ++i) {
        http_ctx[i].request = &http_requests_buffer[i];
        http_ctx[i].response = ctx[i].response->mutable_http_response();
        http_ctx[i].retry = _options.retry;
        http_ctx[i].data = &ctx[i];
    }
    HttpReactor reactor;
    reactor.set_options(&_options);
    for (i = 0; i < max_parallel; ++i) {
        ret = build_http_request(*ctx[i].request, http_ctx[i].request);
        build_http_response(ctx[i].response, http_ctx[i].response);
        if (ret != 0) {
            LOG(ERROR) << "build http request failed: " << ret;
            return ret;
        }
        ret = reactor.execute(&http_ctx[i]);
        if (ret != 0) {
            return ret;
        }
    }
    int finish_num = 0;
    while (finish_num < n) {
        if (flow_control != nullptr && *flow_control) {
            return RET_PARALLEL_COPY_EXIT;
        }
        HttpRequestContext *hctx = reactor.perform();
        if (hctx == NULL) {
            reactor.wait();
            continue;
        }
        BceRequestContext *bctx = (BceRequestContext *) hctx->data;
        if (hctx->rc == 0) {
            bctx->response->handle_response(*hctx->response);
        } else {
            bctx->response->set_error(STATUS_RPC_FAIL, "HttpExecuteFailure");
        }
        int status = bctx->response->status_code();
        if ((status >= 500 && status != 501) && hctx->retry > 0) {
            hctx->retry--;
            hctx->request->reset();
            hctx->response->reset();
            LOG(WARN) << "send_request do retry. last ret status:" 
                      << status << " retry count:" << hctx->retry;
        } else if (i < n) {
            hctx = &http_ctx[i++];
            bctx = (BceRequestContext *) hctx->data;
            ++finish_num;
        } else {
            ++finish_num;
            continue;
        }
        ret = build_http_request(*bctx->request, hctx->request);
        build_http_response(bctx->response, hctx->response);
        if (ret != 0) {
            LOG(ERROR) << "build http request failed: " << ret;
            return ret;
        }
        ret = reactor.execute(hctx);
        if (ret != 0) {
            return ret;
        }
    }
    return 0;
}

END_CPPSDK_NAMESPACE

