// SPDX-License-Identifier: MPL-2.0
// Canal de rendu (spec Core - rendu 0.1) : poignee de main sequa-render, scene complete a
// render/ready, deltas apres chaque tick, un seul rendu a la fois. Sans reseau.

#include "render_hub.hpp"

#include "sequa/core/server/live_session.hpp"
#include "sequa/core/server/protocol.hpp"
#include "sequa/core/server/runtime.hpp"
#include "sequa/core/sim/input.hpp"

#include "sequa/3d/scene.hpp"
#include "sequa/3d/wire.hpp"

#include <doctest/doctest.h>
#include <nlohmann/json.hpp>

#include <cstdint>
#include <mutex>
#include <string>
#include <vector>

using namespace sequa::core::server;
using nlohmann::json;
namespace three_d = sequa::three_d;

namespace {

struct Outbox {
    std::mutex mutex;
    std::vector<json> messages;

    PushFn push()
    {
        return [this](std::string text) {
            const std::lock_guard lock(mutex);
            messages.push_back(json::parse(text));
        };
    }
    std::vector<json> take()
    {
        const std::lock_guard lock(mutex);
        std::vector<json> out;
        out.swap(messages);
        return out;
    }
};

CoreRuntimeOptions offline()
{
    CoreRuntimeOptions options;
    options.real_time = false; // aucun fil de simulation
    options.core_version = "test";
    return options;
}

const std::string render_hello = encode_hello(render_protocol_version, render_client_name);

RenderReady ready_payload()
{
    RenderReady ready;
    ready.capabilities.engine_name = "debug";
    ready.capabilities.engine_version = "0.1";
    ready.capabilities.buffers = three_d::BufferSet{three_d::FrameBuffer::color};
    ready.capabilities.headless = true;
    ready.width = 320;
    ready.height = 180;
    return ready;
}

std::string ready_message()
{
    const RenderReady ready = ready_payload();
    return json{{"v", 1},
                {"type", "render/ready"},
                {"payload",
                 {{"capabilities", three_d::wire::encode_capabilities(ready.capabilities)},
                  {"width", ready.width},
                  {"height", ready.height}}}}
        .dump();
}

three_d::wire::SceneDeltaMessage decode(const json& message)
{
    REQUIRE(message["type"] == "scene/delta");
    const auto delta = three_d::wire::decode_scene_delta(message["payload"]);
    REQUIRE(delta.has_value());
    return *delta;
}

} // namespace

TEST_CASE("Canal de rendu : poignee de main puis scene complete a render/ready")
{
    CoreRuntime runtime(offline());
    Outbox outbox;
    ProtocolSession session(runtime, 7, outbox.push());

    const Reply welcome = session.on_text(render_hello);
    REQUIRE(welcome.messages.size() == 1);
    const json w = json::parse(welcome.messages.front());
    CHECK(w["type"] == "welcome");
    CHECK(w["payload"]["protocolVersion"] == std::string(render_protocol_version));
    CHECK(w["payload"]["renderId"].get<std::string>().size() == 32);
    CHECK(outbox.take().empty()); // rien avant render/ready

    CHECK(session.on_text(ready_message()).messages.empty()); // notification, sans reponse
    const std::vector<json> pushed = outbox.take();
    REQUIRE(pushed.size() == 1);
    const auto full = decode(pushed[0]);
    CHECK(full.full);
    three_d::SceneState mirror;
    CHECK_FALSE(mirror.apply(full.delta).has_value());
    CHECK(mirror.objects().size() >= 2); // sol et joueur au moins

    const RenderHub::Status status = runtime.render_hub().status();
    CHECK(status.connected);
    CHECK(status.ready);
    CHECK(status.deltas_sent == 1);
}

TEST_CASE("Canal de rendu : un seul rendu a la fois, place liberee a la deconnexion")
{
    CoreRuntime runtime(offline());
    Outbox first_box;
    Outbox second_box;
    ProtocolSession first(runtime, 1, first_box.push());
    ProtocolSession second(runtime, 2, second_box.push());

    CHECK(json::parse(first.on_text(render_hello).messages.front())["type"] == "welcome");
    const Reply refused = second.on_text(render_hello);
    CHECK(refused.close);
    CHECK(json::parse(refused.messages.front())["payload"]["code"] == "render-exists");

    runtime.on_client_connected();
    runtime.on_client_disconnected(1);
    ProtocolSession third(runtime, 3, second_box.push());
    CHECK(json::parse(third.on_text(render_hello).messages.front())["type"] == "welcome");
}

TEST_CASE("Canal de rendu : un studio ne peut pas devenir rendu sans fonction d'envoi")
{
    CoreRuntime runtime(offline());
    ProtocolSession session(runtime, 1); // pas de push : connexion sans envoi spontane
    const Reply refused = session.on_text(render_hello);
    CHECK(refused.close);
}

namespace {

std::string input_message(const nlohmann::json& payload)
{
    return json{{"v", 1}, {"type", "render/input"}, {"payload", payload}}.dump();
}

std::string input_message(std::uint32_t buttons, int move_x, int move_y)
{
    three_d::wire::PlayerInputMessage message;
    message.input = three_d::PlayerInput{buttons, static_cast<std::int16_t>(move_x), static_cast<std::int16_t>(move_y)};
    message.captured_at_us = 1;
    return input_message(three_d::wire::encode_player_input(message));
}

// Session de reference sans rendu : memes entrees par session/input.
LiveSession reference_session()
{
    LiveSession::Options options;
    options.real_time = false;
    return LiveSession(options);
}

} // namespace

TEST_CASE("Canal de rendu : render/input s'applique comme session/input et s'enregistre")
{
    CoreRuntime runtime(offline());
    Outbox outbox;
    ProtocolSession render(runtime, 7, outbox.push());
    REQUIRE(json::parse(render.on_text(render_hello).messages.front())["type"] == "welcome");
    REQUIRE(render.on_text(ready_message()).messages.empty());

    CHECK(render.on_text(input_message(sequa::core::sim::button::jump, 32767, 0)).messages.empty());
    (void)runtime.live_session().advance(20);
    CHECK(render.on_text(input_message(0, -12000, 0)).messages.empty());
    (void)runtime.live_session().advance(20);

    LiveSession reference = reference_session();
    (void)reference.set_input(sequa::core::sim::button::jump, 32767, 0);
    (void)reference.advance(20);
    (void)reference.set_input(0, -12000, 0);
    (void)reference.advance(20);

    const SessionStateView got = runtime.session_status();
    CHECK(got.tick == 40);
    CHECK(got.input_count > 0);
    CHECK(got.state_hash == reference.status().state_hash);
    CHECK(runtime.render_hub().status().inputs_received == 2);

    // Le rejeu de l'enregistrement exporte retrouve le meme etat.
    const SessionExportData exported = runtime.session_export();
    LiveSession replayed = reference_session();
    const auto imported = replayed.import_session(exported.recording, false);
    REQUIRE(imported.has_value());
    CHECK(imported->state.state_hash == got.state_hash);
}

TEST_CASE("Canal de rendu : entree remise au neutre quand le rendu disparait")
{
    CoreRuntime runtime(offline());
    Outbox outbox;
    runtime.on_client_connected();
    ProtocolSession render(runtime, 7, outbox.push());
    REQUIRE(json::parse(render.on_text(render_hello).messages.front())["type"] == "welcome");
    REQUIRE(render.on_text(ready_message()).messages.empty());

    REQUIRE(render.on_text(input_message(0, 32767, 0)).messages.empty()); // le joueur court
    (void)runtime.live_session().advance(10);
    runtime.on_client_disconnected(7);
    (void)runtime.live_session().advance(30);

    LiveSession reference = reference_session();
    (void)reference.set_input(0, 32767, 0);
    (void)reference.advance(10);
    (void)reference.set_input(0, 0, 0);
    (void)reference.advance(30);
    CHECK(runtime.session_status().state_hash == reference.status().state_hash);
}

TEST_CASE("Canal de rendu : entree neutre du rendu, l'entree de Studio n'est pas ecrasee a la deconnexion")
{
    CoreRuntime runtime(offline());
    Outbox outbox;
    runtime.on_client_connected();
    ProtocolSession render(runtime, 7, outbox.push());
    REQUIRE(json::parse(render.on_text(render_hello).messages.front())["type"] == "welcome");
    REQUIRE(render.on_text(input_message(0, 0, 0)).messages.empty());
    (void)runtime.session_input(0, 20000, 0); // Studio
    runtime.on_client_disconnected(7);
    (void)runtime.live_session().advance(10);

    LiveSession reference = reference_session();
    (void)reference.set_input(0, 20000, 0);
    (void)reference.advance(10);
    CHECK(runtime.session_status().state_hash == reference.status().state_hash);
}

TEST_CASE("Canal de rendu : render/input invalide refuse, entree inchangee")
{
    CoreRuntime runtime(offline());
    Outbox outbox;
    ProtocolSession render(runtime, 7, outbox.push());
    REQUIRE(json::parse(render.on_text(render_hello).messages.front())["type"] == "welcome");

    const Reply refused = render.on_text(
        input_message(json{{"buttons", 0}, {"moveX", 40000}, {"moveY", 0}, {"capturedAtUs", "0"}}));
    REQUIRE(refused.messages.size() == 1);
    const json error = json::parse(refused.messages.front());
    CHECK(error["type"] == "error");
    CHECK(error["payload"]["code"] == "invalid-payload");
    CHECK_FALSE(refused.close);
    CHECK(runtime.render_hub().status().inputs_received == 0);
    (void)runtime.live_session().advance(5);
    CHECK(runtime.session_status().input_count == 0);
}

TEST_CASE("Canal de rendu : render/input refuse sur une connexion Studio")
{
    CoreRuntime runtime(offline());
    ProtocolSession studio(runtime, 1);
    REQUIRE(json::parse(studio.on_text(encode_hello(protocol_version, "sequa-studio")).messages.front())["type"] ==
            "welcome");
    const Reply refused = studio.on_text(input_message(0, 100, 0));
    REQUIRE(refused.messages.size() == 1);
    CHECK(json::parse(refused.messages.front())["type"] == "error");
    CHECK(runtime.render_hub().status().inputs_received == 0);
}

TEST_CASE("Canal de rendu : entree d'un ancien canal ignoree")
{
    RenderHub hub(sequa::core::sim::ArenaParams{});
    Outbox outbox;
    REQUIRE(hub.attach(1, outbox.push()).has_value());
    CHECK(hub.accept_input(1, three_d::PlayerInput{0, 5, 0}));
    CHECK(hub.disconnected(1)); // entree non neutre maintenue
    REQUIRE(hub.attach(2, outbox.push()).has_value());
    CHECK_FALSE(hub.accept_input(1, three_d::PlayerInput{0, 5, 0}));
    CHECK_FALSE(hub.disconnected(1));
    CHECK_FALSE(hub.disconnected(2)); // le nouveau canal n'a rien maintenu
}

TEST_CASE("Canal de rendu : un delta applicable par tick qui change la scene, versions croissantes")
{
    LiveSession::Options options;
    options.real_time = false;
    LiveSession session(options);
    RenderHub hub(sequa::core::sim::ArenaParams{});
    session.set_world_observer(
        [&hub](const sequa::core::sim::WorldState& world, std::uint64_t tick) { hub.on_world(world, tick); });

    Outbox outbox;
    REQUIRE(hub.attach(1, outbox.push()).has_value());
    (void)session.advance(5);
    CHECK(outbox.take().empty()); // pas encore pret : rien n'est pousse

    REQUIRE(hub.ready(1, ready_payload()));
    session.notify_world();
    std::vector<json> pushed = outbox.take();
    REQUIRE(pushed.size() == 1);
    three_d::SceneState mirror;
    const auto full = decode(pushed[0]);
    CHECK(full.full);
    REQUIRE_FALSE(mirror.apply(full.delta).has_value());

    (void)session.set_input(0, 32767, 0); // le joueur court vers la droite
    (void)session.advance(60);
    pushed = outbox.take();
    CHECK(pushed.size() > 10);
    std::uint64_t last_tick = full.tick;
    for (const json& message : pushed) {
        const auto delta = decode(message);
        CHECK_FALSE(delta.full);
        CHECK(delta.tick > last_tick);
        last_tick = delta.tick;
        REQUIRE_FALSE(mirror.apply(delta.delta).has_value());
    }
    CHECK(hub.status().world_version == mirror.world_version());

    CHECK_FALSE(hub.disconnected(1)); // aucune entree recue du rendu
    (void)session.advance(10);
    CHECK(outbox.take().empty()); // plus rien apres la deconnexion
}