#include "tile/net/internal/http_engine.h"

#include "tile/base/internal/lazy_init.h"
#include "tile/base/internal/thread_pool.h"
#include "tile/base/object_pool.h"
#include "tile/base/thread/mutex.h"
#include "tile/base/thread/scoped_lock.h"

#include "gflags/gflags.h"

#include <fcntl.h>
#include <unistd.h>

DEFINE_int32(tile_http_engine_workers_per_scheduling_group, 1, "http engine background workers per scheduling group");
DEFINE_int32(tile_http_engine_max_connections_per_host_per_worker, 50, "max connections per host per worker");
DEFINE_int32(tile_http_engine_max_total_connections_per_worker, 200, "max total connections per worker");
DEFINE_bool(tile_http_engine_use_epoll, false, "http client use epoll or poll");
DEFINE_bool(tile_http_engine_enable_debug, false, "If set, debugging output from libcurl is logged.");
DEFINE_bool(tile_http_engine_enable_debug_body, false, "If set, HTTP body is also logged.");

namespace tile {
namespace internal {

class Notifier {
public:
    Notifier() {}

    ~Notifier()
    {
        ::close(fds_[0]);
        ::close(fds_[1]);
    }

    bool Init()
    {
        TILE_CHECK_EQ(::pipe(fds_), 0);
        int old_fl = fcntl(fds_[0], F_GETFL, 0);
        if (old_fl < 0) { return false; }
        int f = fcntl(fds_[0], F_SETFL, old_fl | O_NONBLOCK);
        if (f == -1) { return false; }

        return true;
    }

    void Read()
    {
        char c;
        while (::read(fds_[0], &c, 1) == 0) {}
    }

    bool Notify()
    {
        TILE_DCHECK_GE(fds_[0], 0);
        return write(fds_[1], "a", 1) == 1;
    }

    int fd() const { return fds_[0]; }

private:
    int fds_[2];
};

class CallContextQueue {
public:
    void Push(PooledPtr<HttpTaskCallContext> ctx)
    {
        ScopedLock _(pending_call_ctx_mutex_);
        pending_call_ctxs_.push(std::move(ctx));
    }

    // Out should ensure that the size is at least max_ctxs.
    // Returns the actual pop number.
    std::size_t Pop(PooledPtr<HttpTaskCallContext> *out, std::size_t max_ctxs)
    {
        if (max_ctxs == 0) return 0;
        ScopedLock _(pending_call_ctx_mutex_);
        std::size_t i = 0;
        while (max_ctxs-- && !pending_call_ctxs_.empty()) {
            out[i++] = std::move(pending_call_ctxs_.front());
            pending_call_ctxs_.pop();
        }
        return i;
    }

private:
    Mutex pending_call_ctx_mutex_;
    std::queue<PooledPtr<HttpTaskCallContext>> pending_call_ctxs_;
};

int SockCallback(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp);

// int MultiTimerCallback(CURLM *multi, long timeout_ms, int *tfd) { // NOLINT
//   struct itimerspec its;
//   if (timeout_ms > 0) {
//     its.it_interval.tv_sec = 0;
//     its.it_interval.tv_nsec = 0;
//     its.it_value.tv_sec = timeout_ms / 1000;
//     its.it_value.tv_nsec = (timeout_ms % 1000) * 1000 * 1000;
//   } else if (timeout_ms == 0) {
//     /* libcurl wants us to timeout now, however setting both fields of
//      * new_value.it_value to zero disarms the timer. The closest we can
//      * do is to schedule the timer to fire in 1 ns. */
//     its.it_interval.tv_sec = 0;
//     its.it_interval.tv_nsec = 0;
//     its.it_value.tv_sec = 0;
//     its.it_value.tv_nsec = 1;
//   } else {
//     memset(&its, 0, sizeof(struct itimerspec));
//   }
//
//   timerfd_settime(*tfd, /*flags=*/0, &its, nullptr);
//   return 0;
// }

class CurlClient {
public:
    CurlClient(std::size_t group, CallContextQueue *call_context_queue, Notifier *notifier) : thread_pool_(1, {}, 0)
    {
        multi_handle_ = curl_multi_init();
        TILE_CHECK(multi_handle_, "Curl multi init failed.");
        curl_multi_setopt(multi_handle_, CURLMOPT_MAXCONNECTS, FLAGS_tile_http_engine_max_total_connections_per_worker);
        curl_multi_setopt(multi_handle_, CURLMOPT_MAX_HOST_CONNECTIONS,
                          FLAGS_tile_http_engine_max_connections_per_host_per_worker);
        notifier_           = notifier;
        group_              = group;
        call_context_queue_ = call_context_queue;
        worker_             = std::thread([this, group] { LoopPoll(); });
    }

    void Stop()
    {
        exiting_.store(true, std::memory_order_relaxed);
        thread_pool_.Stop();
    }

    ~CurlClient()
    {
        worker_.join();
        thread_pool_.Join();
        curl_multi_cleanup(multi_handle_);
    }

    void CheckMultiInfo()
    {
        CURLMsg *msg;
        int msgs_left;
        while ((msg = curl_multi_info_read(multi_handle_, &msgs_left))) {
            if (msg->msg == CURLMSG_DONE) {
                CURL *e = msg->easy_handle;
                curl_multi_remove_handle(multi_handle_, e);

                auto task = [this, e, msg]() mutable {
                    void *pointer;
                    curl_easy_getinfo(e, CURLINFO_PRIVATE, &pointer);
                    EasyHandleDone(e, msg->data.result, reinterpret_cast<HttpTaskCallContext *>(pointer));
                };
                thread_pool_.QueueJob(std::move(task));

                // fiber::internal::StartFiberDetached(
                //     Fiber::Attributes{.scheduling_group = group_},
                //     [this, e, msg]() mutable {
                //       void *pointer;
                //       curl_easy_getinfo(e, CURLINFO_PRIVATE, &pointer);
                //       EasyHandlerDone(e, msg->data.result,
                //                       reinterpret_cast<HttpTaskCallContext
                //                       *>(pointer));
                //     });
            }
        }
    }

    void AddHandlers()
    {
        static constexpr std::size_t kMaxStilRunning    = 50;
        static constexpr std::size_t kLocalPopArraySize = 30;
        PooledPtr<HttpTaskCallContext> local_pop_array[kLocalPopArraySize];
        std::size_t max_to_pop = (still_running_ >= kMaxStilRunning) ? 1 : kMaxStilRunning - still_running_;
        PooledPtr<HttpTaskCallContext> *out;
        std::vector<PooledPtr<HttpTaskCallContext>> large_pop_array;
        if (max_to_pop <= kLocalPopArraySize) {
            out = local_pop_array;
        } else {
            large_pop_array.resize(max_to_pop);
            out = large_pop_array.data();
        }
        auto n = call_context_queue_->Pop(out, max_to_pop);
        for (std::size_t i = 0; i < n; ++i) {
            HttpTaskCallContext *ctx = out[i].Leak();
            TILE_CHECK_EQ(CURLE_OK, curl_easy_setopt(ctx->curl_handler, CURLOPT_PRIVATE, ctx));
            curl_multi_add_handle(multi_handle_, ctx->curl_handler);
        }
    }

    void LoopPoll()
    {
        int numfds;

        while (!exiting_.load(std::memory_order_relaxed)) {
            CURLMcode mc = curl_multi_perform(multi_handle_, &still_running_);
            if (mc == CURLM_OK) {
                struct curl_waitfd extra_fds[1];
                extra_fds[0].fd      = notifier_->fd();
                extra_fds[0].events  = CURL_WAIT_POLLIN;
                extra_fds[0].revents = 0;

                CheckMultiInfo();
                AddHandlers();

                long curl_timeo = -1;
                curl_multi_timeout(multi_handle_, &curl_timeo);
                if (curl_timeo < 0) {
                    curl_timeo = 5;// ms
                }
                mc = curl_multi_poll(multi_handle_, extra_fds, 1, curl_timeo, &numfds);
                // TILE_LOG_INFO("CurlClient LoopPoll");
                notifier_->Read();
            }
        }
    }

    void EasyHandleDone(CURL *easy_handler, CURLcode result_code, HttpTaskCallContext *ctx)
    {
        auto done = std::move(ctx->done);
        if (result_code != CURLE_OK) {
            done(std::make_unexpected(Status(result_code, curl_easy_strerror(result_code))));
            object_pool::Put<HttpTaskCallContext>(ctx);
        } else {
            done(HttpTaskCompletion(ctx));
        }
    }

    CURLM *multi_handle_;

private:
    std::atomic<bool> exiting_{false};
    std::thread worker_;
    ThreadPool thread_pool_;

    Notifier *notifier_;
    std::size_t group_;
    CallContextQueue *call_context_queue_;
    int still_running_ = 0;
};

class CurlClientGroup {
public:
    CurlClientGroup(std::size_t sz, std::size_t group) : clients_(), queue_(), notifier_()
    {
        TILE_PCHECK(notifier_.Init(), "Failed to init notifier");
        for (std::size_t i = 0; i < sz; ++i) {
            clients_.push_back(make_unique<CurlClient>(group, &queue_, &notifier_));
        }
    }

    void PushContext(PooledPtr<HttpTaskCallContext> ctx)
    {
        queue_.Push(std::move(ctx));
        notifier_.Notify();
    }

    void Stop()
    {
        for (auto &&c : clients_) { c->Stop(); }

        for (auto &&c : clients_) { c.reset(); }
    }

private:
    std::vector<std::unique_ptr<CurlClient>> clients_;
    CallContextQueue queue_;
    Notifier notifier_;
    // Notifier notifier_;
};

size_t
HttpWriteCallback(char *ptr, size_t size, size_t nmemb, void *pstr)
{
    auto bytes = size * nmemb;
    static_cast<tile::NoncontiguousBufferBuilder *>(pstr)->Append(ptr, bytes);
    return bytes;
}

// The header callback will be called once for each header and
// only complete header lines are passed on to the callback.
// Parsing headers is very easy using this.
size_t
HttpHeaderCallback(char *ptr, size_t size, size_t nmemb, void *pstr)
{
    std::size_t bytes = size * nmemb;
    Slice s(ptr, bytes);
    if (s.find(':') == Slice::npos) {// Status-Line.
        return bytes;
    }
    if (s.size() > 2 && s[bytes - 2] == '\r' && s[bytes - 1] == '\n') { s.RemoveSuffix(2); }
    static_cast<std::vector<std::string> *>(pstr)->emplace_back(s);
    return bytes;
}

int
HttpDebugCallback(CURL *handle, curl_infotype type, char *data, size_t size, void *userptr)
{
    Slice data_view(data, size);
    if (type == CURLINFO_TEXT || type == CURLINFO_HEADER_IN || type == CURLINFO_HEADER_OUT) {
        TILE_LOG_INFO("[{}] {}", type, data_view);
    } else if (type == CURLINFO_DATA_IN || type == CURLINFO_DATA_OUT) {
        if (FLAGS_tile_http_engine_enable_debug_body) {
            TILE_LOG_INFO("[{}] {}", type, data_view);
        }// Ignored otherwise.
    }// Everything else is ignored.
    return 0;
}

std::vector<std::unique_ptr<CurlClientGroup>> curl_client_groups;

HttpEngine *
HttpEngine::Instance()
{
    static NeverDestroyedSingleton<HttpEngine> engine;
    return engine.Get();
}

void
HttpEngine ::StartTask(HttpTask task, std::function<void(std::expected<HttpTaskCompletion, Status>)> done)
{
    auto ctx  = task.ctx_.Get();
    ctx->hdrs = std::move(task.hdrs_);
    if (FLAGS_tile_http_engine_enable_debug) {
        TILE_CHECK_EQ(CURLE_OK, curl_easy_setopt(ctx->curl_handler, CURLOPT_DEBUGFUNCTION, HttpDebugCallback));
        TILE_CHECK_EQ(CURLE_OK, curl_easy_setopt(ctx->curl_handler, CURLOPT_VERBOSE, 1));
    }
    TILE_CHECK_EQ(CURLE_OK, curl_easy_setopt(ctx->curl_handler, CURLOPT_NOSIGNAL, 1));
    TILE_CHECK_EQ(CURLE_OK, curl_easy_setopt(ctx->curl_handler, CURLOPT_WRITEFUNCTION, HttpWriteCallback));
    ctx->body = make_unique<NoncontiguousBufferBuilder>();
    TILE_CHECK_EQ(CURLE_OK, curl_easy_setopt(ctx->curl_handler, CURLOPT_WRITEDATA, ctx->body.get()));
    TILE_CHECK_EQ(CURLE_OK, curl_easy_setopt(ctx->curl_handler, CURLOPT_HEADERFUNCTION, HttpHeaderCallback));
    TILE_CHECK_EQ(CURLE_OK, curl_easy_setopt(ctx->curl_handler, CURLOPT_HEADERDATA, &ctx->headers));
    TILE_CHECK_EQ(CURLE_OK, curl_easy_setopt(ctx->curl_handler, CURLOPT_HTTPHEADER, ctx->hdrs.get()));
    ctx->done = std::move(done);
    // auto c = curl_client_groups[fiber::GetCurrentSchedulingGroupIndex()].get();
    // c->PushContext(std::move(task.ctx_));
    auto c = curl_client_groups[0].get();
    c->PushContext(std::move(task.ctx_));
}

void
HttpEngine::Stop()
{
    for (auto &&c : curl_client_groups) { c->Stop(); }
}

void
HttpEngine::Join()
{
    for (auto &&c : curl_client_groups) { c.reset(); }
    curl_global_cleanup();
}

HttpEngine::HttpEngine()
{
    auto ret = curl_global_init(CURL_GLOBAL_DEFAULT);
    TILE_CHECK(!ret, "Curl Init failed {}", ret);
    for (auto i = 0; i < 5; ++i) {
        curl_client_groups.push_back(
            make_unique<CurlClientGroup>(FLAGS_tile_http_engine_workers_per_scheduling_group, i));
    }
}

}// namespace internal
}// namespace tile
