// Live-HSMS GEM 300 scenario.  Mirrors the in-memory test in
// test_gem300_scenario.cpp but drives every step through real HSMS
// frames on a loopback socket pair, validating both the dispatch
// path (host → equipment) and the CEID emission path
// (equipment → host) for the full lifecycle.
//
// Equipment-side emulator wires:
//   S1F13 → S1F14   establish comm
//   S3F17 → S3F18   carrier action (proceed)
//   S16F11 → S16F12 PJ create
//   S14F9 → S14F10  CJ create
//   S16F27 → S16F28 CJ command
// and auto-emits S6F11 on CJ Executing transition + S16F9 on each
// PJ state change.

#include <doctest/doctest.h>

#include <asio.hpp>
#include <chrono>
#include <cstdint>
#include <memory>
#include <optional>
#include <thread>
#include <utility>
#include <vector>

#include "secsgem/gem/data_model.hpp"
#include "secsgem/gem/messages.hpp"
#include "secsgem/hsms/connection.hpp"
#include "secsgem/hsms/header.hpp"
#include "secsgem/secs2/message.hpp"

using namespace secsgem;
using namespace std::chrono_literals;

namespace {

struct SocketPair {
  asio::io_context io;
  asio::ip::tcp::socket a{io};
  asio::ip::tcp::socket b{io};

  SocketPair() {
    asio::ip::tcp::acceptor acc(io, asio::ip::tcp::endpoint(
                                        asio::ip::address_v4::loopback(), 0));
    const auto port = acc.local_endpoint().port();
    bool da = false, db = false;
    std::error_code ea, eb;
    acc.async_accept(a, [&](std::error_code ec) { ea = ec; da = true; });
    b.async_connect({asio::ip::address_v4::loopback(), port},
                    [&](std::error_code ec) { eb = ec; db = true; });
    while (!(da && db)) {
      if (io.stopped()) io.restart();
      if (io.poll() == 0) std::this_thread::sleep_for(1ms);
    }
    REQUIRE_FALSE(ea);
    REQUIRE_FALSE(eb);
  }
};

template <typename Pred>
void pump_until(asio::io_context& io, Pred pred,
                std::chrono::milliseconds budget = 3s) {
  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(1ms);
  }
}

hsms::Timers permissive_timers() {
  hsms::Timers t;
  t.t3 = 5s; t.t6 = 5s; t.t7 = 5s; t.t8 = 5s; t.linktest = 0ms;
  return t;
}

static constexpr uint32_t kCeidControlJobExecuting = 400;

// EquipmentEmulator: model + connection + per-stream handlers.
struct EquipmentEmulator {
  gem::EquipmentDataModel m;
  std::shared_ptr<hsms::Connection> conn;
  uint32_t dataid = 1;

  explicit EquipmentEmulator(asio::ip::tcp::socket socket) {
    conn = std::make_shared<hsms::Connection>(
        std::move(socket), hsms::Connection::Mode::Passive, /*device_id=*/0,
        permissive_timers());

    m.recipes.add("RECIPE-A", "step1");
    m.svids.add({1, "STATUS", "", secs2::Item::ascii("READY")});
    m.events.register_event({kCeidControlJobExecuting, "ControlJobExecuting"});

    // Wire CJ → emit S6F11 when entering Executing.
    auto* conn_raw = conn.get();
    m.control_jobs.set_state_change_handler(
        [this, conn_raw](const std::string&, gem::ControlJobState,
                         gem::ControlJobState to, gem::ControlJobEvent) {
          if (to != gem::ControlJobState::Executing) return;
          if (!m.is_event_enabled(kCeidControlJobExecuting)) return;
          conn_raw->send_data(gem::s6f11_event_report(
              dataid++, kCeidControlJobExecuting,
              m.compose_reports_for(kCeidControlJobExecuting)));
        });

    // Wire PJ → emit S16F9 on every (non-Created) transition.
    m.process_jobs.set_state_change_handler(
        [conn_raw](const std::string& prjobid, gem::ProcessJobState,
                   gem::ProcessJobState to, gem::ProcessJobEvent trigger) {
          if (trigger == gem::ProcessJobEvent::Created) return;
          conn_raw->send_data(gem::s16f9_pr_job_alert(prjobid, to));
        });

    conn->set_message_handler(
        [this](const secs2::Message& msg) -> std::optional<secs2::Message> {
          return dispatch(msg);
        });
  }

  std::optional<secs2::Message> dispatch(const secs2::Message& msg) {
    if (msg.stream == 1 && msg.function == 13) {
      return gem::s1f14_establish_comms_ack(gem::CommAck::Accept, gem::McAck{"E", "1.0"});
    }
    if (msg.stream == 3 && msg.function == 17) {
      auto req = gem::parse_s3f17(msg);
      if (!req) return gem::s3f18_carrier_action_ack(gem::CarrierActionAck::CarrierActionInvalid);
      if (!m.carriers.has(req->carrierid)) m.carriers.create(req->carrierid, 1, 4);
      if (req->carrieraction == "ProceedWithCarrier") {
        m.carriers.fire_id_event(req->carrierid, gem::CarrierIDEvent::ProceedWithCarrier);
      }
      return gem::s3f18_carrier_action_ack(gem::CarrierActionAck::Accept);
    }
    if (msg.stream == 16 && msg.function == 11) {
      auto req = gem::parse_s16f11(msg);
      if (!req) return gem::s16f12_pr_job_create_ack(gem::HostCmdAck::InvalidCommand);
      auto rc = m.process_jobs.create(
          req->prjobid, req->rcpspec.ppid, req->mtrloutspec,
          [&](const std::string& ppid) { return m.recipes.get(ppid).has_value(); });
      return gem::s16f12_pr_job_create_ack(
          rc == gem::ProcessJobStore::CreateResult::Created
              ? gem::HostCmdAck::Accept
              : gem::HostCmdAck::CannotDoNow);
    }
    if (msg.stream == 14 && msg.function == 9) {
      auto req = gem::parse_s14f9(msg);
      if (!req) {
        return gem::s14f10_create_control_job_ack("",
                                                   gem::ObjectAck::Denied_InvalidAttribute);
      }
      auto rc = m.control_jobs.create(req->ctljobid, req->prjobids,
                                       [&](const std::string& id) {
                                         return m.process_jobs.has(id);
                                       });
      return gem::s14f10_create_control_job_ack(
          req->ctljobid,
          rc == gem::ControlJobStore::CreateResult::Created
              ? gem::ObjectAck::Success
              : gem::ObjectAck::Denied_AlreadyExists);
    }
    if (msg.stream == 16 && msg.function == 27) {
      auto req = gem::parse_s16f27(msg);
      if (!req) return gem::s16f28_cj_command_ack(gem::HostCmdAck::InvalidCommand);
      auto event = gem::ctl_cmd_to_event(req->ctljobcmd);
      if (!event) return gem::s16f28_cj_command_ack(gem::HostCmdAck::InvalidCommand);
      // Walk Queued → Selected → WaitingForStart → Executing on CJSTART.
      if (*event == gem::ControlJobEvent::Start) {
        m.control_jobs.fire_internal(req->ctljobid, gem::ControlJobEvent::Select);
        m.control_jobs.fire_internal(req->ctljobid, gem::ControlJobEvent::SetupComplete);
      }
      auto ack = m.control_jobs.on_host_command(req->ctljobid, *event);
      return gem::s16f28_cj_command_ack(ack);
    }
    return std::nullopt;
  }
};

struct HostEmulator {
  std::shared_ptr<hsms::Connection> conn;
  // Captured equipment → host primaries.
  std::vector<secs2::Message> primaries;

  explicit HostEmulator(asio::ip::tcp::socket socket) {
    conn = std::make_shared<hsms::Connection>(
        std::move(socket), hsms::Connection::Mode::Active, 0, permissive_timers());
    conn->set_message_handler(
        [this](const secs2::Message& m) -> std::optional<secs2::Message> {
          primaries.push_back(m);
          return std::nullopt;
        });
  }

  bool saw(uint8_t stream, uint8_t function) const {
    for (const auto& m : primaries)
      if (m.stream == stream && m.function == function) return true;
    return false;
  }
};

}  // namespace

TEST_CASE("Live HSMS GEM300 lifecycle: comm → carrier → PJ → CJ → execute") {
  SocketPair sp;
  EquipmentEmulator eq(std::move(sp.a));
  HostEmulator      host(std::move(sp.b));

  // Subscribe to ControlJobExecuting so the equipment will emit S6F11.
  eq.m.events.enable_events(true, {kCeidControlJobExecuting});

  bool eq_sel = false, host_sel = false;
  eq.conn->set_selected_handler([&] { eq_sel = true; });
  host.conn->set_selected_handler([&] { host_sel = true; });
  eq.conn->start();
  host.conn->start();
  pump_until(sp.io, [&] { return eq_sel && host_sel; });

  auto send_request = [&](secs2::Message req, uint8_t expected_function) {
    std::optional<secs2::Message> reply;
    host.conn->send_request(std::move(req),
                             [&](std::error_code ec, const secs2::Message& m) {
                               REQUIRE_FALSE(ec);
                               reply = m;
                             });
    pump_until(sp.io, [&] { return reply.has_value(); });
    CHECK(reply->function == expected_function);
    return *reply;
  };

  // 1. Establish comms.
  auto comm_ack = send_request(
      gem::s1f13_establish_comms("HOST", "1.0"), /*expected=*/14);
  CHECK(comm_ack.stream == 1);
  CHECK(comm_ack.function == 14);

  // 2. Carrier action: ProceedWithCarrier "CAR-1".
  auto car_ack = send_request(
      gem::s3f17_carrier_action(1, "ProceedWithCarrier", "CAR-1", {}),
      /*expected=*/18);
  CHECK(car_ack.stream == 3);
  CHECK(car_ack.function == 18);
  CHECK(eq.m.carriers.has("CAR-1"));
  CHECK(eq.m.carriers.get("CAR-1")->fsm->id_status() ==
        gem::CarrierIDStatus::Confirmed);

  // 3. PJ create.
  gem::PRJobCreateRequest pj_req;
  pj_req.prjobid = "PJ-1";
  pj_req.mf = gem::MaterialFlag::Substrate;
  pj_req.prrecipemethod = gem::ProcessRecipeMethod::RecipeOnly;
  pj_req.rcpspec.ppid = "RECIPE-A";
  pj_req.mtrloutspec = {"W-1"};
  auto pj_ack = send_request(gem::s16f11_pr_job_create(pj_req), 12);
  CHECK(eq.m.process_jobs.has("PJ-1"));

  // 4. CJ create.
  auto cj_ack = send_request(gem::s14f9_create_control_job("CJ-1", {"PJ-1"}), 10);
  auto parsed_cj = gem::parse_s14f10(cj_ack);
  REQUIRE(parsed_cj.has_value());
  CHECK(parsed_cj->ack == gem::ObjectAck::Success);
  CHECK(eq.m.control_jobs.has("CJ-1"));

  // 5. CJ Start.
  auto cmd_ack = send_request(gem::s16f27_cj_command("CJ-1", "CJSTART"), 28);
  CHECK(eq.m.control_jobs.get("CJ-1")->fsm->state() ==
        gem::ControlJobState::Executing);

  // 6. Equipment auto-emitted S6F11(ControlJobExecuting) — wait for it.
  pump_until(sp.io, [&] { return host.saw(6, 11); });
  // Decode and assert.
  for (const auto& m : host.primaries) {
    if (m.stream == 6 && m.function == 11) {
      auto parsed = gem::parse_s6f11(m);
      REQUIRE(parsed.has_value());
      CHECK(parsed->ceid == kCeidControlJobExecuting);
    }
  }

  // 7. CJ Complete (internal) — observed on equipment side, no wire frame yet.
  eq.m.control_jobs.fire_internal("CJ-1", gem::ControlJobEvent::AllJobsComplete);
  CHECK(eq.m.control_jobs.get("CJ-1")->fsm->state() ==
        gem::ControlJobState::Completed);
}