// vdm/net/http_client.cpp — libcurl multi implementation.
//
// Threading model: one Worker == one std::jthread + one CURLM. An easy handle is created,
// used, paused, and destroyed only on its Worker's thread. Public calls (start / pause /
// resume / cancel) just enqueue a Command and curl_multi_wakeup() the worker.

#include "vdm/net/http_client.hpp"

#include <curl/curl.h>

#include <algorithm>
#include <atomic>
#include <charconv>
#include <deque>
#include <mutex>
#include <string_view>
#include <thread>
#include <vector>

#include "net/curl_error.hpp"
#include "vdm/util/log.hpp"

namespace vdm::net {
namespace {

struct CurlGlobal {
    CurlGlobal() {
        if (curl_global_init(CURL_GLOBAL_ALL) != CURLE_OK)
            VDM_LOG_ERROR("net", "curl_global_init failed");
    }
    ~CurlGlobal() { curl_global_cleanup(); }
};
void ensure_curl_global() {
    static CurlGlobal g;
    (void)g;
}

std::uint64_t next_id() {
    static std::atomic<std::uint64_t> counter{0};
    return ++counter;
}

bool parse_header_line(std::string_view line, std::string &name, std::string &value) {
    while (!line.empty() && (line.back() == '\r' || line.back() == '\n'))
        line.remove_suffix(1);
    if (line.empty())
        return false;
    auto colon = line.find(':');
    if (colon == std::string_view::npos)
        return false;
    name.assign(line.substr(0, colon));
    auto v = line.substr(colon + 1);
    while (!v.empty() && (v.front() == ' ' || v.front() == '\t'))
        v.remove_prefix(1);
    value.assign(v);
    return true;
}

// "HTTP/1.1 206 Partial Content" -> 206; 0 on parse failure.
long status_from_line(std::string_view line) {
    auto sp = line.find(' ');
    if (sp == std::string_view::npos)
        return 0;
    auto rest = line.substr(sp + 1);
    long code = 0;
    auto [p, ec] = std::from_chars(rest.data(), rest.data() + rest.size(), code);
    (void)p;
    return ec == std::errc{} ? code : 0;
}

}  // namespace

// --- Transfer::State -----------------------------------------------------------------
struct Transfer::State {
    enum class Stop { none, head_complete, aborted };

    std::uint64_t id = 0;
    struct HttpClient::Impl *client = nullptr;
    unsigned worker_index = 0;

    Request req;
    TransferCallbacks cbs;

    // Worker-thread-owned.
    CURL *easy = nullptr;
    curl_slist *header_slist = nullptr;
    std::string range_value;
    std::string cookie_value;

    ResponseHead head;
    long line_status = 0;  // status from the most recent HTTP/ line
    bool head_delivered = false;

    std::atomic<bool> pause_requested{false};
    std::atomic<Stop> stop{Stop::none};
    bool curl_paused = false;

    std::uint64_t bytes_received = 0;
    bool finished = false;
};

// --- Impl -------------------------------------------------------------------------
struct HttpClient::Impl {
    enum class CmdKind { add, pause, resume, cancel };
    struct Command {
        CmdKind kind;
        std::shared_ptr<Transfer::State> state;
    };

    struct Worker {
        CURLM *multi = nullptr;
        std::mutex mu;
        std::deque<Command> queue;
        std::vector<std::shared_ptr<Transfer::State>> live;
        std::jthread thread;
    };

    explicit Impl(Options o) : opts(o) {
        ensure_curl_global();

        unsigned n = opts.workers;
        if (n == 0) {
            unsigned hw = std::thread::hardware_concurrency();
            n = std::clamp<unsigned>(hw ? hw : 1, 1, 4);
        }

        if (opts.share_dns_and_tls) {
            share = curl_share_init();
            if (share) {
                curl_share_setopt(share, CURLSHOPT_LOCKFUNC, &Impl::share_lock);
                curl_share_setopt(share, CURLSHOPT_UNLOCKFUNC, &Impl::share_unlock);
                curl_share_setopt(share, CURLSHOPT_USERDATA, this);
                curl_share_setopt(share, CURLSHOPT_SHARE, CURL_LOCK_DATA_DNS);
                curl_share_setopt(share, CURLSHOPT_SHARE, CURL_LOCK_DATA_SSL_SESSION);
            }
        }

        workers.reserve(n);
        for (unsigned i = 0; i < n; ++i) {
            auto w = std::make_unique<Worker>();
            w->multi = curl_multi_init();
            if (opts.max_connections_per_worker > 0)
                curl_multi_setopt(w->multi, CURLMOPT_MAX_TOTAL_CONNECTIONS,
                                  opts.max_connections_per_worker);
            Worker *raw = w.get();
            w->thread = std::jthread([this, raw](std::stop_token st) { run(*raw, st); });
            workers.push_back(std::move(w));
        }
    }

    ~Impl() {
        stopping.store(true);
        for (auto &w : workers) {
            w->thread.request_stop();
            if (w->multi)
                curl_multi_wakeup(w->multi);
        }
        for (auto &w : workers)
            if (w->thread.joinable())
                w->thread.join();
        for (auto &w : workers)
            if (w->multi)
                curl_multi_cleanup(w->multi);
        if (share)
            curl_share_cleanup(share);
    }

    Options opts;
    std::vector<std::unique_ptr<Worker>> workers;
    CURLSH *share = nullptr;
    std::mutex share_mu[CURL_LOCK_DATA_LAST];
    std::atomic<unsigned> rr{0};
    std::atomic<bool> stopping{false};

    static void share_lock(CURL *, curl_lock_data data, curl_lock_access, void *userp) {
        static_cast<Impl *>(userp)->share_mu[data].lock();
    }
    static void share_unlock(CURL *, curl_lock_data data, void *userp) {
        static_cast<Impl *>(userp)->share_mu[data].unlock();
    }

    void enqueue(unsigned wi, Command cmd) {
        Worker &w = *workers[wi];
        {
            std::lock_guard lk(w.mu);
            w.queue.push_back(std::move(cmd));
        }
        curl_multi_wakeup(w.multi);
    }

    // ---- curl C callbacks ----
    static std::size_t header_cb(char *buf, std::size_t size, std::size_t n, void *userp) {
        auto *st = static_cast<Transfer::State *>(userp);
        const std::size_t total = size * n;
        std::string_view line(buf, total);

        if (line.starts_with("HTTP/")) {
            // A new status line after we already delivered a 401/407 means libcurl's
            // CURLAUTH_ANY handshake just resent with credentials: let the head of this
            // second response be delivered too, so callers see the real (2xx/4xx) status
            // rather than the challenge. Redirects never reach here delivered — their head
            // is suppressed below — so this only fires for the auth resend.
            if (st->head_delivered && (st->line_status == 401 || st->line_status == 407))
                st->head_delivered = false;
            st->line_status = status_from_line(line);
            st->head.headers.clear();  // keep only the final response's headers
            return total;
        }
        if (line == "\r\n" || line == "\n") {
            const bool redirect =
                st->req.follow_redirects && st->line_status >= 300 && st->line_status < 400;
            if (!redirect)
                deliver_head(st);
            return total;
        }
        std::string name, value;
        if (parse_header_line(line, name, value))
            st->head.headers.add(std::move(name), std::move(value));
        return total;
    }

    static void deliver_head(Transfer::State *st) {
        if (st->head_delivered)
            return;
        st->head_delivered = true;

        long code = 0;
        curl_easy_getinfo(st->easy, CURLINFO_RESPONSE_CODE, &code);
        st->head.status = code ? code : st->line_status;
        char *eff = nullptr;
        if (curl_easy_getinfo(st->easy, CURLINFO_EFFECTIVE_URL, &eff) == CURLE_OK && eff)
            st->head.effective_url = eff;
        curl_off_t clen = -1;
        if (curl_easy_getinfo(st->easy, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &clen) == CURLE_OK &&
            clen >= 0)
            st->head.content_length = static_cast<std::uint64_t>(clen);

        if (st->cbs.on_head) {
            DataAction a = st->cbs.on_head(st->head);
            if (a == DataAction::abort)
                st->stop.store(Transfer::State::Stop::head_complete);
            else if (a == DataAction::pause)
                st->pause_requested.store(true);
        }
    }

    static std::size_t write_cb(char *ptr, std::size_t size, std::size_t n, void *userp) {
        auto *st = static_cast<Transfer::State *>(userp);
        const std::size_t total = size * n;

        if (!st->head_delivered)
            deliver_head(st);
        if (st->stop.load() != Transfer::State::Stop::none)
            return 0;  // -> CURLE_WRITE_ERROR
        if (st->pause_requested.load()) {
            st->curl_paused = true;
            return CURL_WRITEFUNC_PAUSE;
        }

        if (total && st->cbs.on_data) {
            ConstByteSpan span(reinterpret_cast<const std::byte *>(ptr), total);
            DataAction a = st->cbs.on_data(span);
            if (a == DataAction::abort) {
                st->stop.store(Transfer::State::Stop::aborted);
                return 0;
            }
            if (a == DataAction::pause) {
                st->pause_requested.store(true);
                st->curl_paused = true;
                return CURL_WRITEFUNC_PAUSE;
            }
        }
        st->bytes_received += total;
        return total;
    }

    // ---- worker thread ----
    void run(Worker &w, std::stop_token stok) {
        while (!stok.stop_requested()) {
            drain_commands(w);
            int running = 0;
            curl_multi_perform(w.multi, &running);
            reap(w);
            if (stok.stop_requested())
                break;
            int numfds = 0;
            curl_multi_poll(w.multi, nullptr, 0, 1000, &numfds);
        }
        shutdown_worker(w);
    }

    void drain_commands(Worker &w) {
        // Called every worker-loop iteration (run(), below) -- once per curl_multi_poll
        // wake, so once per socket-readiness event on the transfer hot path -- but commands
        // (add/pause/resume/cancel) are rare next to that. Check empty under the lock
        // *before* touching `local`: libstdc++'s std::deque allocates its map array on
        // default construction even with nothing pushed to it, so constructing one every
        // iteration just to usually swap nothing into it was an allocation on every poll
        // wake, not just on an actual command -- exactly what the curl-write-callback path
        // must never do (AGENT-CORE.md; caught by tools/bench's alloc-check).
        std::unique_lock lk(w.mu);
        if (w.queue.empty())
            return;
        std::deque<Command> local;
        local.swap(w.queue);
        lk.unlock();
        for (auto &cmd : local) {
            auto &st = cmd.state;
            switch (cmd.kind) {
                case CmdKind::add:
                    attach(w, st);
                    break;
                case CmdKind::pause:
                    st->pause_requested.store(true);
                    if (st->easy && !st->curl_paused) {
                        curl_easy_pause(st->easy, CURLPAUSE_RECV);
                        st->curl_paused = true;
                    }
                    break;
                case CmdKind::resume:
                    st->pause_requested.store(false);
                    if (st->easy && st->curl_paused) {
                        st->curl_paused = false;
                        curl_easy_pause(st->easy, CURLPAUSE_CONT);
                    }
                    break;
                case CmdKind::cancel:
                    st->stop.store(Transfer::State::Stop::aborted);
                    if (st->easy && st->curl_paused) {
                        st->curl_paused = false;
                        curl_easy_pause(st->easy, CURLPAUSE_CONT);  // let write_cb return 0
                    }
                    break;
            }
        }
    }

    void attach(Worker &w, std::shared_ptr<Transfer::State> st) {
        if (stopping.load()) {
            complete(st, ErrorInfo(Error::canceled, "client shutting down"));
            return;
        }
        CURL *e = curl_easy_init();
        if (!e) {
            complete(st, ErrorInfo(Error::internal, "curl_easy_init"));
            return;
        }
        st->easy = e;
        const Request &r = st->req;

        curl_easy_setopt(e, CURLOPT_URL, r.url.c_str());
        curl_easy_setopt(e, CURLOPT_PRIVATE, st.get());
        curl_easy_setopt(e, CURLOPT_NOSIGNAL, 1L);
        curl_easy_setopt(e, CURLOPT_NOPROGRESS, 1L);
        curl_easy_setopt(e, CURLOPT_HEADERFUNCTION, &Impl::header_cb);
        curl_easy_setopt(e, CURLOPT_HEADERDATA, st.get());
        curl_easy_setopt(e, CURLOPT_WRITEFUNCTION, &Impl::write_cb);
        curl_easy_setopt(e, CURLOPT_WRITEDATA, st.get());
        curl_easy_setopt(e, CURLOPT_TCP_KEEPALIVE, 1L);
        if (share)
            curl_easy_setopt(e, CURLOPT_SHARE, share);

        if (r.method == Method::head)
            curl_easy_setopt(e, CURLOPT_NOBODY, 1L);

        if (r.follow_redirects) {
            curl_easy_setopt(e, CURLOPT_FOLLOWLOCATION, 1L);
            curl_easy_setopt(e, CURLOPT_MAXREDIRS, r.max_redirects);
        }
        curl_easy_setopt(e, CURLOPT_ACCEPT_ENCODING, r.accept_encoding ? "" : nullptr);

        if (r.range) {
            std::string v = r.range->to_header_value();  // "bytes=first-last"
            std::string_view sv = v;
            if (sv.starts_with("bytes="))
                sv.remove_prefix(6);
            st->range_value.assign(sv);
            curl_easy_setopt(e, CURLOPT_RANGE, st->range_value.c_str());
        }

        curl_easy_setopt(e, CURLOPT_CONNECTTIMEOUT_MS, static_cast<long>(r.connect_timeout_ms));
        if (r.overall_timeout_ms > 0)
            curl_easy_setopt(e, CURLOPT_TIMEOUT_MS, static_cast<long>(r.overall_timeout_ms));
        if (r.low_speed_bytes_per_sec > 0 && r.low_speed_secs > 0) {
            curl_easy_setopt(e, CURLOPT_LOW_SPEED_LIMIT, r.low_speed_bytes_per_sec);
            curl_easy_setopt(e, CURLOPT_LOW_SPEED_TIME, r.low_speed_secs);
        }
        if (r.max_recv_bytes_per_sec > 0)
            curl_easy_setopt(e, CURLOPT_MAX_RECV_SPEED_LARGE,
                             static_cast<curl_off_t>(r.max_recv_bytes_per_sec));

        if (!r.user_agent.empty())
            curl_easy_setopt(e, CURLOPT_USERAGENT, r.user_agent.c_str());
        if (!r.referrer.empty())
            curl_easy_setopt(e, CURLOPT_REFERER, r.referrer.c_str());

        if (r.proxy.kind != ProxyKind::none) {
            curl_easy_setopt(e, CURLOPT_PROXY, r.proxy.host.c_str());
            if (r.proxy.port)
                curl_easy_setopt(e, CURLOPT_PROXYPORT, long(r.proxy.port));
            long pt = CURLPROXY_HTTP;
            if (r.proxy.kind == ProxyKind::socks5)
                pt = CURLPROXY_SOCKS5;
            else if (r.proxy.kind == ProxyKind::socks5_hostname)
                pt = CURLPROXY_SOCKS5_HOSTNAME;
            curl_easy_setopt(e, CURLOPT_PROXYTYPE, pt);
            if (!r.proxy.username.empty()) {
                std::string up = r.proxy.username + ":" + r.proxy.password;
                curl_easy_setopt(e, CURLOPT_PROXYUSERPWD, up.c_str());
            }
        }

        if (r.auth.scheme != AuthScheme::none) {
            long m = CURLAUTH_ANY;
            if (r.auth.scheme == AuthScheme::basic)
                m = CURLAUTH_BASIC;
            else if (r.auth.scheme == AuthScheme::digest)
                m = CURLAUTH_DIGEST;
            curl_easy_setopt(e, CURLOPT_HTTPAUTH, m);
            std::string up = r.auth.username + ":" + r.auth.password;
            curl_easy_setopt(e, CURLOPT_USERPWD, up.c_str());
        }

        if (!r.cookies.empty()) {
            for (const auto &c : r.cookies) {
                if (!st->cookie_value.empty())
                    st->cookie_value += "; ";
                st->cookie_value += c.name + "=" + c.value;
            }
            curl_easy_setopt(e, CURLOPT_COOKIE, st->cookie_value.c_str());
        }

        for (const auto &h : r.headers) {
            std::string joined = h.name + ": " + h.value;
            st->header_slist = curl_slist_append(st->header_slist, joined.c_str());
        }
        if (st->header_slist)
            curl_easy_setopt(e, CURLOPT_HTTPHEADER, st->header_slist);

        CURLMcode mc = curl_multi_add_handle(w.multi, e);
        if (mc != CURLM_OK) {
            curl_easy_cleanup(e);
            st->easy = nullptr;
            complete(st, ErrorInfo(Error::internal, curl_multi_strerror(mc)));
            return;
        }
        w.live.push_back(std::move(st));
    }

    void reap(Worker &w) {
        CURLMsg *msg = nullptr;
        int inq = 0;
        while ((msg = curl_multi_info_read(w.multi, &inq)) != nullptr) {
            if (msg->msg != CURLMSG_DONE)
                continue;
            CURL *e = msg->easy_handle;
            const CURLcode res = msg->data.result;

            Transfer::State *raw = nullptr;
            curl_easy_getinfo(e, CURLINFO_PRIVATE, &raw);
            long code = 0;
            curl_easy_getinfo(e, CURLINFO_RESPONSE_CODE, &code);

            TransferStats stats;
            stats.http_status = code;
            gather_timings(e, stats);

            auto it = std::find_if(w.live.begin(), w.live.end(),
                                   [raw](const auto &s) { return s.get() == raw; });
            std::shared_ptr<Transfer::State> st = (it != w.live.end()) ? *it : nullptr;

            curl_multi_remove_handle(w.multi, e);
            curl_easy_cleanup(e);
            if (st) {
                st->easy = nullptr;
                if (st->header_slist) {
                    curl_slist_free_all(st->header_slist);
                    st->header_slist = nullptr;
                }
            }
            if (it != w.live.end())
                w.live.erase(it);
            if (!st)
                continue;

            using Stop = Transfer::State::Stop;
            const Stop stop = st->stop.load();

            if (stop == Stop::aborted) {
                complete(st, ErrorInfo(Error::canceled));
            } else if (stop == Stop::head_complete) {
                // A probe: on_head asked to stop. Headers were the goal -> success, even
                // though a ranged GET body-stop surfaces as CURLE_WRITE_ERROR.
                stats.bytes_received = st->bytes_received;
                stats.effective_url = st->head.effective_url;
                complete(st, std::move(stats));
            } else if (res == CURLE_OK && code < 400) {
                stats.bytes_received = st->bytes_received;
                stats.effective_url = st->head.effective_url;
                complete(st, std::move(stats));
            } else {
                complete(st, detail::make_error(res, code));
            }
        }
    }

    static void gather_timings(CURL *e, TransferStats &s) {
        auto us_to_ms = [](curl_off_t us) { return us > 0 ? static_cast<long>(us / 1000) : 0L; };
        curl_off_t t = 0;
        if (curl_easy_getinfo(e, CURLINFO_NAMELOOKUP_TIME_T, &t) == CURLE_OK)
            s.namelookup_ms = us_to_ms(t);
        if (curl_easy_getinfo(e, CURLINFO_CONNECT_TIME_T, &t) == CURLE_OK)
            s.connect_ms = us_to_ms(t);
        if (curl_easy_getinfo(e, CURLINFO_APPCONNECT_TIME_T, &t) == CURLE_OK)
            s.appconnect_ms = us_to_ms(t);
        if (curl_easy_getinfo(e, CURLINFO_STARTTRANSFER_TIME_T, &t) == CURLE_OK)
            s.starttransfer_ms = us_to_ms(t);
        if (curl_easy_getinfo(e, CURLINFO_TOTAL_TIME_T, &t) == CURLE_OK)
            s.total_ms = us_to_ms(t);
    }

    void complete(const std::shared_ptr<Transfer::State> &st, Result<TransferStats> r) {
        if (st->finished)
            return;
        st->finished = true;
        if (st->cbs.on_finished)
            st->cbs.on_finished(std::move(r));
    }

    void shutdown_worker(Worker &w) {
        for (auto &st : w.live) {
            if (st->easy) {
                curl_multi_remove_handle(w.multi, st->easy);
                curl_easy_cleanup(st->easy);
                st->easy = nullptr;
            }
            if (st->header_slist) {
                curl_slist_free_all(st->header_slist);
                st->header_slist = nullptr;
            }
            complete(st, ErrorInfo(Error::canceled, "client shutting down"));
        }
        w.live.clear();
    }
};

// --- Transfer -------------------------------------------------------------------
std::uint64_t Transfer::id() const noexcept {
    return state_ ? state_->id : 0;
}

void Transfer::pause() {
    if (state_ && state_->client)
        state_->client->enqueue(state_->worker_index, {HttpClient::Impl::CmdKind::pause, state_});
}
void Transfer::resume() {
    if (state_ && state_->client)
        state_->client->enqueue(state_->worker_index, {HttpClient::Impl::CmdKind::resume, state_});
}
void Transfer::cancel() {
    if (state_ && state_->client)
        state_->client->enqueue(state_->worker_index, {HttpClient::Impl::CmdKind::cancel, state_});
}

// --- HttpClient ---------------------------------------------------------------
HttpClient::HttpClient() : HttpClient(Options{}) {}
HttpClient::HttpClient(Options opts) : impl_(std::make_unique<Impl>(opts)) {}
HttpClient::~HttpClient() = default;

unsigned HttpClient::worker_count() const noexcept {
    return impl_ ? static_cast<unsigned>(impl_->workers.size()) : 0;
}

Transfer HttpClient::start(Request req, TransferCallbacks cbs) {
    auto st = std::make_shared<Transfer::State>();
    st->id = next_id();
    st->client = impl_.get();
    st->req = std::move(req);
    st->cbs = std::move(cbs);
    st->worker_index = impl_->workers.empty() ? 0 : impl_->rr.fetch_add(1) % impl_->workers.size();

    impl_->enqueue(st->worker_index, {Impl::CmdKind::add, st});
    return Transfer(st);
}

}  // namespace vdm::net