#include <iostream>
#include <filesystem>
#include <fstream>

#include <catch2/catch.hpp>
#include <pistache/client.h>

#include "daggy/Server.hpp"
#include "daggy/executors/task/ForkingTaskExecutor.hpp"
#include "daggy/loggers/dag_run/OStreamLogger.hpp"

Pistache::Http::Response
REQUEST(Pistache::Http::Experimental::Client &client, std::string url, std::string payload = "") {
    Pistache::Http::Response response;
    auto reqSpec = (payload.empty() ? client.get(url) : client.post(url));
    reqSpec.timeout(std::chrono::seconds(2));
    if (!payload.empty()) {
        reqSpec.body(payload);
    }
    auto request = reqSpec.send();
    bool ok = false, error = false;
    std::string msg;
    request.then(
            [&](Pistache::Http::Response rsp) {
                ok = true;
                response = rsp;
            },
            [&](std::exception_ptr ptr) {
                error = true;
                try {
                    std::rethrow_exception(ptr);
                } catch (std::exception &e) {
                    msg = e.what();
                }
            }
    );

    Pistache::Async::Barrier<Pistache::Http::Response> barrier(request);
    barrier.wait_for(std::chrono::seconds(2));
    if (error) {
        throw std::runtime_error(msg);
    }
    return response;
}

TEST_CASE("Server Basic Endpoints", "[server_basic]") {
    std::stringstream ss;
    daggy::executors::task::ForkingTaskExecutor executor(10);
    daggy::loggers::dag_run::OStreamLogger logger(ss);
    Pistache::Address listenSpec("localhost", Pistache::Port(0));

    const size_t nDAGRunners = 10,
            nWebThreads = 10;

    daggy::Server server(listenSpec, logger, executor, 10);
    server.init(nWebThreads);
    server.start();

    Pistache::Http::Experimental::Client client;
    const std::string host = "localhost:";
    const std::string baseURL = host + std::to_string(server.getPort());
    client.init();

    SECTION("/ready endpoint") {
        auto response = REQUEST(client, baseURL + "/ready");
        REQUIRE(response.code() == Pistache::Http::Code::Ok);
    }

    SECTION("dag submission") {
        // submit a DAGRun
        std::string dagRun = R"({
    "name": "unit_server",
    "parameters": { "DIRS": [ "A", "B" ] },
    "tasks": [
       { "name": "touch",
         "command": [ "/usr/bin/touch", "/tmp/{{DIRS}}" ]
       },
       {
         "name": "cat",
         "command": [ "/usr/bin/cat", "/tmp/A", "/tmp/B" ]
         "parents": [ "touch" ]
       }
       ]
    }
        )";

        auto response = REQUEST(client, baseURL + "/v1/dagrun/", dagRun);
        REQUIRE(response.code() == Pistache::Http::Code::Ok);
    }

    server.shutdown();
    client.shutdown();
}