// Scheduler against a FakeEnginePort and an in-memory store: admission, priority order,
// queue stop, restart reconciliation, and the engine-state -> store projection.

#include <string>

#include <nlohmann/json.hpp>

#include "check.hpp"
#include "rpc/event_hub.hpp"
#include "sched/fake_engine_port.hpp"
#include "sched/governor.hpp"
#include "sched/scheduler.hpp"
#include "store/migrations.hpp"
#include "store/settings.hpp"
#include "store/sqlite.hpp"
#include "store/tasks.hpp"

using namespace velox::daemon;
using sched::FakeEnginePort;
using sched::Governor;
using sched::GovernorConfig;
using sched::Scheduler;

namespace {

store::TaskRow task(std::string id, std::string state, std::string created,
                    std::optional<std::string> queue = std::nullopt, std::int64_t pos = 0) {
    store::TaskRow r;
    r.task_id = std::move(id);
    r.url = "https://cdn.example/" + r.task_id;
    r.save_dir = "/tmp";
    r.filename = r.task_id + ".bin";
    r.state = std::move(state);
    r.created_at = std::move(created);
    r.queue_id = std::move(queue);
    if (r.queue_id) r.queue_position = pos;
    return r;
}

std::string task_state(store::Db& db, const std::string& id) {
    store::Tasks t(db);
    auto g = t.get(id);
    return (g && *g) ? (*g)->state : std::string("<none>");
}

}  // namespace

void run() {
    auto db = store::Db::open(":memory:");
    CHECK(db.has_value());
    if (!db) return;
    CHECK(store::migrate_to_head(*db).has_value());
    store::Tasks tasks(*db);

    // --- admission: 4 queued, cap 2 -> start the 2 oldest, in order -----------------
    {
        FakeEnginePort engine;
        Scheduler sched(*db, engine,
                        Governor(GovernorConfig{.max_concurrent_downloads = 2,
                                                .max_active_segments = 32}));
        for (int i = 0; i < 4; ++i)
            CHECK(tasks.insert(task("a" + std::to_string(i), "queued",
                                    "2026-09-10T10:0" + std::to_string(i) + ":00Z"))
                      .has_value());

        CHECK(sched.tick().has_value());
        CHECK_EQ(engine.starts.size(), 2u);
        CHECK_EQ(engine.starts[0].url, std::string("https://cdn.example/a0"));
        CHECK_EQ(engine.starts[1].url, std::string("https://cdn.example/a1"));
        CHECK_EQ(engine.starts[0].save_path, std::string("/tmp/a0.bin"));
        CHECK_EQ(task_state(*db, "a0"), std::string("probing"));
        CHECK_EQ(task_state(*db, "a2"), std::string("queued"));  // not admitted

        // set_task_order carries exactly the started tasks, oldest first.
        CHECK_EQ(engine.last_order().size(), 2u);
        CHECK(engine.last_order()[0] == engine.starts[0].id);

        // A second tick with no free slots starts nothing new.
        CHECK(sched.tick().has_value());
        CHECK_EQ(engine.starts.size(), 2u);
    }

    // --- engine reports downloading, then one completes -> a slot frees ------------
    {
        for (const char* id : {"a0", "a1", "a2", "a3"}) tasks.remove(id);
        FakeEnginePort engine;
        Scheduler sched(*db, engine,
                        Governor(GovernorConfig{.max_concurrent_downloads = 1,
                                                .max_active_segments = 32}));
        CHECK(tasks.insert(task("b0", "queued", "2026-09-10T10:00:00Z")).has_value());
        CHECK(tasks.insert(task("b1", "queued", "2026-09-10T10:01:00Z")).has_value());

        CHECK(sched.tick().has_value());
        CHECK_EQ(engine.starts.size(), 1u);  // b0 only
        CHECK_EQ(task_state(*db, "b0"), std::string("probing"));

        sched.on_engine_state("b0", "probing", "downloading", std::nullopt);
        CHECK_EQ(task_state(*db, "b0"), std::string("downloading"));

        sched.on_engine_state("b0", "downloading", "complete", std::nullopt);
        CHECK_EQ(task_state(*db, "b0"), std::string("complete"));

        CHECK(sched.tick().has_value());  // b0 terminal -> b1 admitted
        CHECK_EQ(engine.starts.size(), 2u);
        CHECK_EQ(engine.starts[1].url, std::string("https://cdn.example/b1"));
    }

    // --- a stopped queue: running tasks get paused with reason queue_stopped -------
    {
        for (const char* id : {"b0", "b1"}) tasks.remove(id);
        CHECK(db->exec("UPDATE queues SET state='running' WHERE queue_id='main'").has_value());
        FakeEnginePort engine;
        Scheduler sched(*db, engine,
                        Governor(GovernorConfig{.max_concurrent_downloads = 10,
                                                .max_active_segments = 32}));
        CHECK(tasks.insert(task("q0", "queued", "2026-09-10T10:00:00Z", "main", 0)).has_value());
        CHECK(sched.tick().has_value());
        CHECK_EQ(engine.starts.size(), 1u);
        sched.on_engine_state("q0", "connecting", "downloading", std::nullopt);

        CHECK(db->exec("UPDATE queues SET state='stopped' WHERE queue_id='main'").has_value());
        CHECK(sched.tick().has_value());
        CHECK_EQ(engine.paused.size(), 1u);
        CHECK_EQ(task_state(*db, "q0"), std::string("paused"));
        store::Tasks t(*db);
        CHECK_EQ(t.get("q0").value().value().pause_reason.value_or(""), std::string("queue_stopped"));

        // Restart the queue -> the task resumes (not a fresh start).
        CHECK(db->exec("UPDATE queues SET state='running' WHERE queue_id='main'").has_value());
        CHECK(sched.tick().has_value());
        CHECK_EQ(engine.resumed.size(), 1u);
        CHECK_EQ(engine.starts.size(), 1u);  // no new start
    }

    // --- an engine auto-pause (error present) -> pause_reason 'auto', not touched --
    {
        for (const char* id : {"q0"}) tasks.remove(id);
        FakeEnginePort engine;
        Scheduler sched(*db, engine,
                        Governor(GovernorConfig{.max_concurrent_downloads = 10,
                                                .max_active_segments = 32}));
        CHECK(tasks.insert(task("auth", "queued", "2026-09-10T10:00:00Z")).has_value());
        CHECK(sched.tick().has_value());

        sched::TaskErrorFields ef;
        ef.code = "auth_required";
        ef.message = "401";
        ef.http_status = 401;
        sched.on_engine_state("auth", "connecting", "paused", ef);

        store::Tasks t(*db);
        auto row = t.get("auth").value().value();
        CHECK_EQ(row.state, std::string("paused"));
        CHECK_EQ(row.pause_reason.value_or(""), std::string("auto"));
        CHECK_EQ(row.error_code.value_or(""), std::string("auth_required"));

        // A tick must NOT resume an auto-paused task.
        engine.resumed.clear();
        CHECK(sched.tick().has_value());
        CHECK_EQ(engine.resumed.size(), 0u);
    }

    // --- reconcile_after_restart: CORE-owned states -> queued --------------------
    {
        for (const char* id : {"auth"}) tasks.remove(id);
        CHECK(tasks.insert(task("r0", "downloading", "2026-09-10T10:00:00Z")).has_value());
        CHECK(tasks.insert(task("r1", "verifying", "2026-09-10T10:01:00Z")).has_value());
        CHECK(tasks.insert(task("r2", "paused", "2026-09-10T10:02:00Z")).has_value());
        CHECK(db->exec("UPDATE tasks SET pause_reason='user' WHERE task_id='r2'").has_value());

        FakeEnginePort engine;
        Scheduler sched(*db, engine, Governor(GovernorConfig{}));
        CHECK(sched.reconcile_after_restart().has_value());
        CHECK_EQ(task_state(*db, "r0"), std::string("queued"));
        CHECK_EQ(task_state(*db, "r1"), std::string("queued"));
        CHECK_EQ(task_state(*db, "r2"), std::string("paused"));  // paused survives
        store::Tasks t(*db);
        CHECK_EQ(t.get("r2").value().value().pause_reason.value_or(""), std::string("user"));
    }

    // --- reload_config pushes the caps to the engine ---------------------------
    {
        store::Settings settings(*db);
        CHECK(settings.set_raw("connection.maxActiveSegments", "12").has_value());
        FakeEnginePort engine;
        Scheduler sched(*db, engine, Governor(GovernorConfig{}));
        CHECK(sched.reload_config().has_value());
        CHECK_EQ(engine.max_active_segments.size(), 1u);
        CHECK_EQ(engine.max_active_segments.back(), 12u);
        CHECK_EQ(sched.reload_config().has_value() ? 0 : 1, 0);
    }

    // --- event.task.state: published on admission, on a paused transition, and on an
    // engine-reported transition; previousState reflects the store's prior row ----------
    {
        for (const char* id : {"r0", "r1", "r2"}) tasks.remove(id);  // leftover from above
        FakeEnginePort engine;
        rpc::EventHub hub;
        Scheduler sched(*db, engine, Governor(GovernorConfig{.max_concurrent_downloads = 10,
                                                              .max_active_segments = 32}),
                        &hub);
        std::vector<nlohmann::json> received;
        const auto sub = hub.subscribe([&](const nlohmann::json& n) { received.push_back(n); });
        hub.set_filter(sub, {velox::proto::Event::TaskState}, std::nullopt);

        CHECK(tasks.insert(task("ev0", "queued", "2026-09-10T10:00:00Z")).has_value());
        CHECK(sched.tick().has_value());  // admits ev0: queued -> probing
        CHECK_EQ(received.size(), 1u);
        CHECK_EQ(received[0]["params"]["taskId"].get<std::string>(), std::string("ev0"));
        CHECK_EQ(received[0]["params"]["state"].get<std::string>(), std::string("probing"));
        CHECK_EQ(received[0]["params"]["previousState"].get<std::string>(), std::string("queued"));
        CHECK(received[0]["params"]["error"].is_null());
        CHECK_EQ(received[0]["params"]["summary"]["taskId"].get<std::string>(), std::string("ev0"));

        sched.on_engine_state("ev0", "probing", "downloading", std::nullopt);
        CHECK_EQ(received.size(), 2u);
        CHECK_EQ(received[1]["params"]["previousState"].get<std::string>(), std::string("probing"));
        CHECK_EQ(received[1]["params"]["state"].get<std::string>(), std::string("downloading"));

        sched::TaskErrorFields ef;
        ef.code = "connection_reset";
        ef.message = "reset";
        sched.on_engine_state("ev0", "downloading", "paused", ef);
        CHECK_EQ(received.size(), 3u);
        CHECK(!received[2]["params"]["error"].is_null());
        CHECK_EQ(received[2]["params"]["error"]["code"].get<std::string>(),
                 std::string("connection_reset"));

        tasks.remove("ev0");
    }

    // --- progress_snapshot: only started tasks, plus a store side-effect --------------
    {
        FakeEnginePort engine;
        Scheduler sched(*db, engine,
                        Governor(GovernorConfig{.max_concurrent_downloads = 10,
                                                .max_active_segments = 32}));
        CHECK(tasks.insert(task("pr0", "queued", "2026-09-10T10:00:00Z")).has_value());
        CHECK(sched.tick().has_value());
        CHECK_EQ(engine.starts.size(), 1u);

        vdm::task::Progress p;
        p.downloaded = 12345;
        p.speed_bps = 999;
        p.effective_segments = 4;
        p.effective_buffer_bytes = 65536;
        engine.fake_progress[engine.starts[0].id.value] = p;

        const auto snap = sched.progress_snapshot();
        CHECK_EQ(snap.size(), 1u);
        CHECK_EQ(snap[0].task_id, std::string("pr0"));
        CHECK_EQ(snap[0].downloaded_bytes, 12345u);
        CHECK_EQ(snap[0].speed_bps, 999u);

        auto row = tasks.get("pr0").value().value();
        CHECK_EQ(row.downloaded_bytes, 12345);
        CHECK_EQ(row.eff_segments, 4);
        CHECK((row.eff_buffer_bytes.has_value() && *row.eff_buffer_bytes == 65536));

        tasks.remove("pr0");
    }
}

TEST_MAIN()