Crafter.Network/implementations/Crafter.Network-ClientHTTP1.cpp

//SPDX-License-Identifier: LGPL-3.0-only
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

module;
#include <poll.h>
#include <sys/socket.h>
#include <cerrno>

module Crafter.Network:ClientHTTP1_impl;
import :ClientHTTP1;
import :ClientTCP;
import :HTTP;
import :HTTP1;
import Crafter.Thread;
import std;

using namespace Crafter;

namespace {
    // Read whatever is available, waiting at most `timeout`. Returns 0 on a
    // clean close by the peer. ClientTCP::RecieveSync() can't express a
    // timeout — and a server that accepts the connection and then says
    // nothing would hang Send() forever — so the read is done here.
    std::size_t ReadSome(int socketid, char* buffer, std::size_t size,
                         std::chrono::milliseconds timeout) {
        for (;;) {
            pollfd pfd{ .fd = socketid, .events = POLLIN, .revents = 0 };
            const int ready = poll(&pfd, 1, static_cast<int>(timeout.count()));
            if (ready < 0) {
                if (errno == EINTR) continue;
                throw std::runtime_error(std::string("poll failed: ") + std::strerror(errno));
            }
            if (ready == 0) throw std::runtime_error("timed out waiting for the response");

            const auto read = recv(socketid, buffer, size, 0);
            if (read < 0) {
                if (errno == EINTR) continue;
                throw std::runtime_error(std::string("recv failed: ") + std::strerror(errno));
            }
            return static_cast<std::size_t>(read);
        }
    }

    // Host header value. The port is elided when it is the scheme default,
    // which is what every other HTTP/1.1 client on the wire does and what
    // virtual-host matching on the far side tends to expect.
    std::string DefaultAuthority(const std::string& host, std::uint16_t port) {
        if (port == 80) return host;
        return host + ":" + std::to_string(port);
    }
}

struct ClientHTTP1::Impl {
    std::string host;
    std::uint16_t port;
    std::unique_ptr<ClientTCP> tcp;

    void Connect() {
        tcp = std::make_unique<ClientTCP>(host, port);
    }

    void Close() {
        tcp.reset();
    }

    // One request/response exchange on the current connection. `received`
    // is set as soon as any response byte arrives, which is what decides
    // whether a failure is safe to replay on a new connection.
    HTTPResponse Exchange(const std::string& wire, std::string_view method,
                          const HTTP1::MessageLimits& limits,
                          std::chrono::milliseconds timeout, bool& received) {
        tcp->Send(wire.data(), static_cast<std::uint32_t>(wire.size()));

        HTTP1::MessageParser parser(HTTP1::MessageKind::Response, limits);
        parser.SetRequestMethod(method);
        std::vector<char> chunk(16 * 1024);
        while (!parser.Complete()) {
            const std::size_t read = ReadSome(tcp->socketid, chunk.data(), chunk.size(), timeout);
            if (read == 0) {
                // Peer closed. Completes a response framed by close;
                // anything else throws out of Finish().
                parser.Finish();
                break;
            }
            received = true;
            parser.Feed(chunk.data(), read);
        }
        if (!parser.Complete()) {
            throw std::runtime_error("connection closed before a complete response arrived");
        }
        HTTPResponse response = parser.TakeResponse();
        if (!parser.KeepAlive()) Close();
        return response;
    }
};

ClientHTTP1::ClientHTTP1(const char* host, std::uint16_t port)
    : host(host), port(port), impl(std::make_unique<Impl>(std::string(host), port)) {}

ClientHTTP1::ClientHTTP1(std::string host, std::uint16_t port)
    : ClientHTTP1(host.c_str(), port) {}

ClientHTTP1::ClientHTTP1(ClientHTTP1&&) noexcept = default;
ClientHTTP1::~ClientHTTP1() = default;

bool ClientHTTP1::Connected() const noexcept {
    return impl && impl->tcp != nullptr;
}

void ClientHTTP1::Disconnect() {
    if (impl) impl->Close();
}

HTTPResponse ClientHTTP1::Send(const HTTPRequest& request) {
    HTTPRequest prepared = request;
    if (prepared.authority.empty()) prepared.authority = DefaultAuthority(host, port);
    const std::string wire = HTTP1::SerializeRequest(prepared);
    const std::string method = prepared.method.empty() ? std::string("GET") : prepared.method;

    // Two attempts at most, and the second one only for the keep-alive
    // race: the peer may have closed a pooled connection at the same moment
    // we wrote to it, which is indistinguishable from success until the
    // read fails. Nothing is replayed once a response byte has arrived.
    for (int attempt = 0;; ++attempt) {
        const bool reused = impl->tcp != nullptr;
        if (!reused) impl->Connect();

        bool received = false;
        try {
            return impl->Exchange(wire, method, limits, timeout, received);
        } catch (const std::exception&) {
            impl->Close();
            if (attempt == 0 && reused && !received) continue;
            throw;
        }
    }
}

void ClientHTTP1::SendAsync(const HTTPRequest& request,
                            std::function<void(HTTPResponse)> onSuccess,
                            std::function<void(std::string)> onError) {
    HTTPRequest copy = request;
    ThreadPool::Enqueue([this, copy = std::move(copy),
                         onSuccess = std::move(onSuccess),
                         onError = std::move(onError)]() mutable {
        try {
            HTTPResponse response = this->Send(copy);
            if (onSuccess) onSuccess(std::move(response));
        } catch (const std::exception& e) {
            if (onError) onError(e.what());
        } catch (...) {
            if (onError) onError("unknown error");
        }
    });
}