Crafter.Debug/implementations/Crafter.Debug.Cli-Quic.cpp

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

module Crafter.Debug.Cli:Quic_impl;
import std;
import Crafter.Network;
import Crafter.Debug.Core;
import :Client;
import :Quic;

using namespace Crafter::Debug;
using namespace Crafter::Debug::Cli;
using Crafter::ClientQUIC;
using Crafter::QUICClientCredentials;
using Crafter::QUICStream;
using Clock = std::chrono::steady_clock;

namespace {
    QUICClientCredentials Credentials() {
        QUICClientCredentials credentials;
        // The app serves a per-user development certificate. What
        // authenticates is the token in each request, which only this user
        // can read.
        credentials.insecureNoServerValidation = true;
        // A port nobody answers on fails in seconds, not at msquic's 10.
        credentials.settings.handshakeIdleTimeoutMs = 3'000;
        // Keep-alives under the idle timeout hold a quiet `log --follow` or
        // a long `wait` open; an app that died is noticed within the timeout.
        credentials.settings.idleTimeoutMs = 10'000;
        credentials.settings.keepAliveIntervalMs = 2'000;
        return credentials;
    }

    // What a request's callbacks, on msquic's worker, hand the thread that
    // waits for its reply.
    struct Inbox {
        std::mutex mutex;
        std::condition_variable changed;
        std::string partial;
        std::deque<std::string> lines;
        bool ended = false;
        bool clean = false;

        void Add(std::span<const char> data) {
            {
                std::lock_guard lock(mutex);
                partial.append(data.data(), data.size());
                std::size_t start = 0;
                for (std::size_t newline = partial.find('\n'); newline != std::string::npos; newline = partial.find('\n', start)) {
                    lines.emplace_back(partial, start, newline - start);
                    start = newline + 1;
                }
                partial.erase(0, start);
            }
            changed.notify_all();
        }

        void End(bool wasClean) {
            {
                std::lock_guard lock(mutex);
                ended = true;
                clean = wasClean;
            }
            changed.notify_all();
        }
    };

    std::string Abbreviated(std::string_view text) {
        return text.size() <= 200 ? std::string(text) : std::format("{}...", text.substr(0, 200));
    }

    class QuicConnection final : public Connection {
    public:
        explicit QuicConnection(std::uint16_t port) : port(port) {}

        std::expected<Value, CallError> Call(const Value& request, std::chrono::milliseconds timeout,
                                             const std::function<bool(const Value& line)>& onLine) override {
            const Clock::time_point deadline = Clock::now() + timeout;
            if (!client) {
                try {
                    client = std::make_unique<ClientQUIC>(std::string("127.0.0.1"), port, std::string(alpn), Credentials());
                } catch (const std::exception& error) {
                    return std::unexpected(CallError{ CallError::Kind::Unreachable, error.what() });
                }
            }
            auto result = Send(request, deadline, onLine);
            // A lost connection is not reused: the next call dials again.
            // After Send, whose stream must go before its connection.
            if (!result && result.error().kind == CallError::Kind::Unreachable) client.reset();
            return result;
        }

    private:
        std::uint16_t port;
        std::unique_ptr<ClientQUIC> client;

        std::expected<Value, CallError> Send(const Value& request, Clock::time_point deadline,
                                                 const std::function<bool(const Value& line)>& onLine) {
            const auto inbox = std::make_shared<Inbox>();
            QUICStream stream;
            try {
                stream = client->OpenStream();
            } catch (const std::exception& error) {
                return std::unexpected(CallError{ CallError::Kind::Unreachable, error.what() });
            }
            stream.OnRecieve([inbox](std::span<const char> data) { inbox->Add(data); },
                             [inbox](bool clean) { inbox->End(clean); });
            std::string text = request.Dump();
            text += '\n';
            try {
                stream.SendSync(text.data(), static_cast<std::uint32_t>(text.size()), true);
            } catch (const std::exception& error) {
                return std::unexpected(CallError{ CallError::Kind::Unreachable, error.what() });
            }

            std::unique_lock lock(inbox->mutex);
            for (;;) {
                if (!inbox->lines.empty()) {
                    const std::string line = std::move(inbox->lines.front());
                    inbox->lines.pop_front();
                    // Not under the lock: onLine prints, and the callbacks
                    // that fill the inbox must not wait for a terminal.
                    lock.unlock();
                    Value value;
                    try {
                        value = Value::Parse(line);
                    } catch (const ParseError& error) {
                        stream.Abort();
                        return std::unexpected(CallError{ CallError::Kind::Broken, std::format("{} in {}", error.what(), Abbreviated(line)) });
                    }
                    if (!value.IsObject()) {
                        stream.Abort();
                        return std::unexpected(CallError{ CallError::Kind::Broken, std::format("not an object: {}", Abbreviated(line)) });
                    }
                    // The reply is the object with a result or an error;
                    // anything before it is a stream line.
                    if (value.Find("result") != nullptr || value.Find("error") != nullptr) return value;
                    if (onLine && !onLine(value)) {
                        stream.Abort();
                        Value stopped = Value::MakeObject();
                        stopped.Set("ended", Value("stopped by the client"));
                        return MakeReply(std::move(stopped));
                    }
                    lock.lock();
                    continue;
                }
                if (inbox->ended) {
                    if (inbox->clean) return std::unexpected(CallError{ CallError::Kind::Broken, "the app ended the request without a reply" });
                    return std::unexpected(CallError{ CallError::Kind::Unreachable, "the connection to the app was lost" });
                }
                if (inbox->changed.wait_until(lock, deadline) == std::cv_status::timeout && inbox->lines.empty() && !inbox->ended) {
                    lock.unlock();
                    stream.Abort();
                    return std::unexpected(CallError{ CallError::Kind::TimedOut, "no reply in time" });
                }
            }
        }
    };
}

std::unique_ptr<Connection> Crafter::Debug::Cli::ConnectQuic(const InstanceRecord& instance) {
    return std::make_unique<QuicConnection>(instance.port);
}