// Integration tests for the GEM HostHandler.  Wires a real Active
// HostHandler vs a real Passive Connection over a TCP loopback pair,
// drives the host through its workflow + sender APIs, and asserts the
// equipment side observed the right primaries and the host received the
// right replies.

#include <doctest/doctest.h>

#include <asio.hpp>
#include <chrono>
#include <memory>
#include <string>
#include <thread>
#include <vector>

#include "secsgem/gem/host_handler.hpp"
#include "secsgem/gem/messages.hpp"
#include "secsgem/hsms/connection.hpp"
#include "secsgem/hsms/header.hpp"

using namespace secsgem;
namespace gem  = secsgem::gem;
namespace s2   = secsgem::secs2;

namespace {

struct Wired {
  asio::io_context io;
  std::shared_ptr<hsms::Connection> host;       // Active
  std::shared_ptr<hsms::Connection> equipment;  // Passive
  std::shared_ptr<gem::HostHandler> handler;

  Wired() {
    asio::ip::tcp::acceptor acc(io, asio::ip::tcp::endpoint(
                                        asio::ip::address_v4::loopback(), 0));
    const auto port = acc.local_endpoint().port();

    asio::ip::tcp::socket eq_sock(io);
    asio::ip::tcp::socket host_sock(io);
    bool a_done = false, c_done = false;
    std::error_code ec_a, ec_c;
    acc.async_accept(eq_sock,    [&](std::error_code e) { ec_a = e; a_done = true; });
    host_sock.async_connect({asio::ip::address_v4::loopback(), port},
                            [&](std::error_code e) { ec_c = e; c_done = true; });
    while (!(a_done && c_done)) {
      if (io.stopped()) io.restart();
      if (io.poll() == 0) std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
    REQUIRE_FALSE(ec_a);
    REQUIRE_FALSE(ec_c);

    hsms::Timers timers;
    timers.linktest = std::chrono::milliseconds(0);
    timers.t3 = std::chrono::milliseconds(1000);

    host = std::make_shared<hsms::Connection>(
        std::move(host_sock), hsms::Connection::Mode::Active, 0, timers);
    equipment = std::make_shared<hsms::Connection>(
        std::move(eq_sock), hsms::Connection::Mode::Passive, 0, timers);
    handler = std::make_shared<gem::HostHandler>(host);
  }
};

template <typename Pred>
void pump_until(asio::io_context& io, Pred pred,
                std::chrono::milliseconds budget = std::chrono::seconds(3)) {
  const auto deadline = std::chrono::steady_clock::now() + budget;
  while (!pred()) {
    if (std::chrono::steady_clock::now() > deadline) FAIL("pump_until budget exceeded");
    if (io.stopped()) io.restart();
    if (io.poll() == 0) std::this_thread::sleep_for(std::chrono::milliseconds(1));
  }
}

// Tracks primaries seen on the equipment side and the canned reply each
// gets.  Used by every test to capture wire-level behaviour without
// having to repeat the dispatch boilerplate.
struct Recorder {
  std::vector<std::pair<uint8_t, uint8_t>> primaries;  // (stream, function)
  std::vector<s2::Message> last;

  hsms::Connection::MessageHandler handler() {
    return [this](const s2::Message& msg) -> std::optional<s2::Message> {
      primaries.emplace_back(msg.stream, msg.function);
      last.push_back(msg);
      // Return a generic-ish reply for each known primary.
      if (msg.stream == 1 && msg.function == 13)
        return gem::s1f14_establish_comms_ack(gem::CommAck::Accept, {"EQUIP", "1.0"});
      if (msg.stream == 1 && msg.function == 17)
        return gem::s1f18_online_ack(gem::OnlineAck::Accept);
      if (msg.stream == 1 && msg.function == 15)
        return gem::s1f16_offline_ack(gem::OfflineAck::Accept);
      if (msg.stream == 2 && msg.function == 41)
        return gem::s2f42_host_command_ack(gem::HostCmdAck::Accept, {});
      if (msg.stream == 10 && msg.function == 1)
        return gem::s10f2_terminal_display_ack(gem::TerminalAck::Accepted);
      if (msg.stream == 16 && msg.function == 11)
        return gem::s16f12_pr_job_create_ack(gem::HostCmdAck::Accept);
      if (msg.stream == 16 && msg.function == 5)
        return gem::s16f6_pr_job_command_ack(gem::HostCmdAck::Accept);
      if (msg.stream == 14 && msg.function == 9)
        return gem::s14f10_create_control_job_ack("CJ-1", gem::ObjectAck::Success);
      // New senders (Task 18): provide canned replies so the host's
      // send_*_request transactions complete instead of timing out.
      if (msg.stream == 2 && msg.function == 21)
        return gem::s2f22_remote_command_ack(gem::HostCmdAck::Accept);
      if (msg.stream == 6 && msg.function == 15)
        return gem::s6f16_event_report_data({0, 300, {}});
      if (msg.stream == 6 && msg.function == 19)
        return gem::s6f20_individual_report_data({});
      if (msg.stream == 6 && msg.function == 21)
        return gem::s6f22_annotated_report_data({});
      if (msg.stream == 7 && msg.function == 1)
        return gem::s7f2_pp_load_grant(gem::ProcessProgramAck::Accept);
      if (msg.stream == 7 && msg.function == 17)
        return gem::s7f18_delete_pp_ack(gem::ProcessProgramAck::Accept);
      // Unhandled — leave the host to time out on T3.
      return std::nullopt;
    };
  }
};

}  // namespace

TEST_CASE("HostHandler: establish_communication + go_remote workflow") {
  Wired w;
  Recorder rec;
  w.equipment->set_message_handler(rec.handler());

  bool selected = false;
  w.host->set_selected_handler([&] { selected = true; });
  w.handler->install();
  w.host->start();
  w.equipment->start();
  pump_until(w.io, [&] { return selected; });

  bool s1f14_seen = false, s1f18_seen = false;
  w.handler->establish_communication([&](std::error_code ec, const s2::Message& m) {
    REQUIRE_FALSE(ec);
    CHECK(m.stream == 1);
    CHECK(m.function == 14);
    s1f14_seen = true;
  });
  pump_until(w.io, [&] { return s1f14_seen; });

  w.handler->go_remote([&](std::error_code ec, const s2::Message& m) {
    REQUIRE_FALSE(ec);
    CHECK(m.stream == 1);
    CHECK(m.function == 18);
    s1f18_seen = true;
  });
  pump_until(w.io, [&] { return s1f18_seen; });

  // Equipment saw both primaries in order.
  REQUIRE(rec.primaries.size() >= 2);
  CHECK(rec.primaries[0] == std::make_pair<uint8_t, uint8_t>(1, 13));
  CHECK(rec.primaries[1] == std::make_pair<uint8_t, uint8_t>(1, 17));
}

TEST_CASE("HostHandler: send_remote_command routes through S2F41") {
  Wired w;
  Recorder rec;
  w.equipment->set_message_handler(rec.handler());
  w.handler->install();
  w.host->start();
  w.equipment->start();
  bool selected = false;
  w.host->set_selected_handler([&] { selected = true; });
  pump_until(w.io, [&] { return selected; });

  bool acked = false;
  std::vector<gem::CommandParameter> params{
      {"LOTID", s2::Item::ascii("LOT-1")},
      {"PPID",  s2::Item::ascii("RECIPE-A")},
  };
  w.handler->send_remote_command("START", params,
                                 [&](std::error_code ec, const s2::Message& m) {
                                   REQUIRE_FALSE(ec);
                                   CHECK(m.stream == 2);
                                   CHECK(m.function == 42);
                                   acked = true;
                                 });
  pump_until(w.io, [&] { return acked; });

  // Equipment received the right primary with the right parameters.
  REQUIRE_FALSE(rec.last.empty());
  const auto& s2f41 = rec.last.back();
  auto parsed = gem::parse_s2f41(s2f41);
  REQUIRE(parsed.has_value());
  CHECK(parsed->rcmd == "START");
  REQUIRE(parsed->params.size() == 2);
  CHECK(parsed->params[0].name == "LOTID");
}

TEST_CASE("HostHandler: terminal display + S10F1 ack round-trip") {
  Wired w;
  Recorder rec;
  w.equipment->set_message_handler(rec.handler());
  w.handler->install();
  w.host->start();
  w.equipment->start();
  bool selected = false;
  w.host->set_selected_handler([&] { selected = true; });
  pump_until(w.io, [&] { return selected; });

  bool done = false;
  w.handler->send_terminal_display(7, "hello operator",
                                   [&](std::error_code ec, const s2::Message& m) {
                                     REQUIRE_FALSE(ec);
                                     CHECK(m.stream == 10);
                                     CHECK(m.function == 2);
                                     done = true;
                                   });
  pump_until(w.io, [&] { return done; });

  REQUIRE_FALSE(rec.last.empty());
  auto td = gem::parse_s10f1(rec.last.back());
  REQUIRE(td.has_value());
  CHECK(td->tid == 7);
  CHECK(td->text == "hello operator");
}

TEST_CASE("HostHandler: E40/E94 job creation senders produce correct wire") {
  Wired w;
  Recorder rec;
  w.equipment->set_message_handler(rec.handler());
  w.handler->install();
  w.host->start();
  w.equipment->start();
  bool selected = false;
  w.host->set_selected_handler([&] { selected = true; });
  pump_until(w.io, [&] { return selected; });

  // Create a PJ.
  bool pj_ack = false;
  w.handler->send_create_process_job(
      "PJ-1", "RECIPE-A", {"W1", "W2"},
      [&](std::error_code ec, const s2::Message& m) {
        REQUIRE_FALSE(ec);
        CHECK(m.stream == 16);
        CHECK(m.function == 12);
        pj_ack = true;
      });
  pump_until(w.io, [&] { return pj_ack; });

  // Create a CJ that references it.
  bool cj_ack = false;
  w.handler->send_create_control_job(
      "CJ-1", {"PJ-1"},
      [&](std::error_code ec, const s2::Message& m) {
        REQUIRE_FALSE(ec);
        CHECK(m.stream == 14);
        CHECK(m.function == 10);
        cj_ack = true;
      });
  pump_until(w.io, [&] { return cj_ack; });

  // Command the PJ.
  bool cmd_ack = false;
  w.handler->send_pr_job_command(
      "PJ-1", "PJSTART",
      [&](std::error_code ec, const s2::Message& m) {
        REQUIRE_FALSE(ec);
        CHECK(m.stream == 16);
        CHECK(m.function == 6);
        cmd_ack = true;
      });
  pump_until(w.io, [&] { return cmd_ack; });
}

TEST_CASE("HostHandler: new senders produce correct wire shapes") {
  // Wires up the host + equipment, installs the handler, fires each
  // new sender, and verifies the (stream, function) of the request
  // the equipment received.  The equipment auto-acks each via Recorder.
  Wired w;
  Recorder rec;
  w.equipment->set_message_handler(rec.handler());
  w.handler->install();
  w.host->start();
  w.equipment->start();
  bool selected = false;
  w.host->set_selected_handler([&] { selected = true; });
  pump_until(w.io, [&] { return selected; });

  auto run_one = [&](auto fire, uint8_t want_stream, uint8_t want_fn) {
    bool done = false;
    fire([&](std::error_code ec, const s2::Message&) {
      REQUIRE_FALSE(ec);
      done = true;
    });
    pump_until(w.io, [&] { return done; });
    REQUIRE_FALSE(rec.last.empty());
    CHECK(rec.last.back().stream == want_stream);
    CHECK(rec.last.back().function == want_fn);
  };

  run_one([&](auto cb) { w.handler->send_legacy_remote_command("PAUSE", cb); },
          2, 21);
  run_one([&](auto cb) { w.handler->send_event_report_request(300, cb); }, 6, 15);
  run_one([&](auto cb) { w.handler->send_individual_report_request(1, cb); },
          6, 19);
  run_one([&](auto cb) { w.handler->send_annotated_report_request(1, cb); },
          6, 21);
  run_one([&](auto cb) {
    w.handler->send_pp_load_inquire("RECIPE-X", 4096, cb);
  }, 7, 1);
  run_one([&](auto cb) {
    w.handler->send_delete_pp({"RECIPE-X", "RECIPE-Y"}, cb);
  }, 7, 17);
}

TEST_CASE("HostHandler: inbound S5F1 alarm auto-acks with S5F2") {
  Wired w;
  // Equipment side stays silent on inbound; we only need the host to
  // process and auto-ack an alarm primary the equipment pushes at it.
  w.equipment->set_message_handler(
      [](const s2::Message&) { return std::optional<s2::Message>{}; });

  // The host handler's alarm observer gets called.
  bool alarm_seen = false;
  uint32_t got_alid = 0;
  w.handler->set_alarm_handler([&](uint32_t alid, uint8_t, const std::string&) {
    got_alid = alid;
    alarm_seen = true;
  });
  w.handler->install();
  w.host->start();
  w.equipment->start();
  bool selected = false;
  w.host->set_selected_handler([&] { selected = true; });
  pump_until(w.io, [&] { return selected; });

  // Equipment side pushes S5F1 at the host.  We don't care about the
  // ack body — just that the handler observes the alarm.
  w.equipment->send_request(
      gem::s5f1_alarm_report(0x80, 42, "vacuum lost"),
      [](std::error_code, const s2::Message&) {});

  pump_until(w.io, [&] { return alarm_seen; });
  CHECK(got_alid == 42);
}