// CEID-on-the-wire emission tests.
//
// The FSM unit tests in test_ept / test_process_jobs / test_substrates /
// test_control_jobs / test_modules verify that state transitions fire
// the change handler — but they don't prove the S6F11 (or S16F9) frame
// is what actually leaves the socket with the right CEID + linked
// report payload.
//
// Each test here wires a small "emitter" between an EquipmentDataModel
// and a passive HSMS Connection, drives a transition, and asserts on
// the exact frame the peer receives.

#include <doctest/doctest.h>

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

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

#include "secsgem/gem/messages.hpp"  // s6f11_event_report, s16f9_pr_job_alert

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;
}

// Wire a SELECTED pair of HSMS Connections: equipment (passive) and
// host (active). Returns both ready-to-use.
struct WirePair {
  SocketPair sp;
  std::shared_ptr<hsms::Connection> equipment;
  std::shared_ptr<hsms::Connection> host;
  std::vector<std::pair<uint8_t, uint8_t>> host_seen;     // (stream, function)
  std::vector<secs2::Message>            host_messages;  // full bodies

  WirePair() {
    equipment = std::make_shared<hsms::Connection>(
        std::move(sp.a), hsms::Connection::Mode::Passive, 0, permissive_timers());
    host = std::make_shared<hsms::Connection>(
        std::move(sp.b), hsms::Connection::Mode::Active, 0, permissive_timers());

    host->set_message_handler([this](const secs2::Message& m)
                                  -> std::optional<secs2::Message> {
      host_seen.emplace_back(m.stream, m.function);
      host_messages.push_back(m);
      return std::nullopt;
    });

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

  bool host_saw(uint8_t stream, uint8_t function) const {
    for (auto& p : host_seen)
      if (p.first == stream && p.second == function) return true;
    return false;
  }
  const secs2::Message* host_last(uint8_t stream, uint8_t function) const {
    for (auto it = host_messages.rbegin(); it != host_messages.rend(); ++it) {
      if (it->stream == stream && it->function == function) return &*it;
    }
    return nullptr;
  }
};

}  // namespace

// ---------------- EPT / E116 -------------------------------------------------

TEST_CASE("EPT → Productive fires S6F11(kCeidProductive) carrying linked report") {
  WirePair wp;

  // Equipment-side model + wire wiring.
  gem::EquipmentDataModel m;
  m.svids.add({1001, "SVID-A", "", secs2::Item::u4(uint32_t{42})});
  m.events.register_event({gem::e116::kCeidProductive, "EptProductive"});
  REQUIRE(m.define_reports({{500, {1001}}}) == gem::DefineReportAck::Accept);
  REQUIRE(m.link_event_reports({{gem::e116::kCeidProductive, {500}}}) ==
          gem::LinkEventAck::Accept);
  REQUIRE(m.enable_events(true, {gem::e116::kCeidProductive}) ==
          gem::EnableEventAck::Accept);

  // Hook the EPT change handler: emit S6F11 on every transition.
  uint32_t dataid_counter = 1;
  m.ept.set_state_change_handler([&](gem::EptState, gem::EptState to,
                                gem::EptEvent, std::chrono::milliseconds) {
    uint32_t ceid = 0;
    switch (to) {
      case gem::EptState::Productive: ceid = gem::e116::kCeidProductive; break;
      case gem::EptState::Standby:    ceid = gem::e116::kCeidStandby;    break;
      default: break;
    }
    if (!ceid || !m.is_event_enabled(ceid)) return;
    auto reports = m.compose_reports_for(ceid);
    wp.equipment->send_data(
        gem::s6f11_event_report(dataid_counter++, ceid, reports));
  });

  // Drive: NonScheduledTime → Standby → Productive.
  m.ept.on_event(gem::EptEvent::EnterStandby);
  m.ept.on_event(gem::EptEvent::EnterProductive);

  pump_until(wp.sp.io, [&] { return wp.host_saw(6, 11); });
  const auto* msg = wp.host_last(6, 11);
  REQUIRE(msg);
  auto parsed = gem::parse_s6f11(*msg);
  REQUIRE(parsed.has_value());
  CHECK(parsed->ceid == gem::e116::kCeidProductive);
  REQUIRE(parsed->reports.size() == 1);
  CHECK(parsed->reports[0].rptid == 500);
}

TEST_CASE("EPT: disabled CEID does NOT emit S6F11") {
  WirePair wp;
  gem::EquipmentDataModel m;
  m.events.register_event({gem::e116::kCeidProductive, "EptProductive"});
  // No define_reports / link_event_reports / enable_events.

  bool emitted = false;
  m.ept.set_state_change_handler([&](gem::EptState, gem::EptState to,
                                      gem::EptEvent, std::chrono::milliseconds) {
    if (to != gem::EptState::Productive) return;
    if (!m.is_event_enabled(gem::e116::kCeidProductive)) return;
    emitted = true;
    wp.equipment->send_data(gem::s6f11_event_report(
        1, gem::e116::kCeidProductive, m.compose_reports_for(gem::e116::kCeidProductive)));
  });
  m.ept.on_event(gem::EptEvent::EnterStandby);
  m.ept.on_event(gem::EptEvent::EnterProductive);
  CHECK_FALSE(emitted);

  // Pump briefly to make sure no S6F11 sneaks through on a later tick.
  const auto deadline = std::chrono::steady_clock::now() + 100ms;
  while (std::chrono::steady_clock::now() < deadline) {
    if (wp.sp.io.stopped()) wp.sp.io.restart();
    wp.sp.io.poll();
    std::this_thread::sleep_for(5ms);
  }
  CHECK_FALSE(wp.host_saw(6, 11));
}

// ---------------- E40 PRJobAlert (S16F9) -------------------------------------

TEST_CASE("PJ Queued→SettingUp fires S16F9 PRJobAlert on the wire") {
  WirePair wp;
  gem::EquipmentDataModel m;
  m.recipes.add("RCP-1", "step1");

  // Hook PJ state-change handler to fire S16F9 when alerts are enabled.
  m.process_jobs.set_state_change_handler(
      [&](const std::string& prjobid, gem::ProcessJobState,
          gem::ProcessJobState to, gem::ProcessJobEvent trigger) {
        if (trigger == gem::ProcessJobEvent::Created) return;
        auto* pj = m.process_jobs.get(prjobid);
        if (!pj || !pj->alert_enabled) return;
        wp.equipment->send_data(
            gem::s16f9_pr_job_alert(prjobid, to));
      });

  REQUIRE(m.process_jobs.create(
              "PJ-A", "RCP-1", {"W1"},
              [&](const std::string& ppid) { return m.recipes.get(ppid).has_value(); }) ==
          gem::ProcessJobStore::CreateResult::Created);
  REQUIRE(m.process_jobs.fire_internal("PJ-A", gem::ProcessJobEvent::Select));

  pump_until(wp.sp.io, [&] { return wp.host_saw(16, 9); });
  const auto* msg = wp.host_last(16, 9);
  REQUIRE(msg);
  REQUIRE(msg->body.has_value());
  // Body: <L,2 PRJOBID PRJOBSTATE>
  const auto& outer = msg->body->as_list();
  REQUIRE(outer.size() == 2);
  CHECK(outer[0].as_ascii() == "PJ-A");
  CHECK(outer[1].as_bytes()[0] ==
        static_cast<uint8_t>(gem::ProcessJobState::SettingUp));
}

TEST_CASE("PJ alert_enabled=false suppresses S16F9 emission") {
  WirePair wp;
  gem::EquipmentDataModel m;
  m.recipes.add("RCP-1", "");

  m.process_jobs.set_state_change_handler(
      [&](const std::string& prjobid, gem::ProcessJobState,
          gem::ProcessJobState to, gem::ProcessJobEvent trigger) {
        if (trigger == gem::ProcessJobEvent::Created) return;
        auto* pj = m.process_jobs.get(prjobid);
        if (!pj || !pj->alert_enabled) return;
        wp.equipment->send_data(
            gem::s16f9_pr_job_alert(prjobid, to));
      });

  REQUIRE(m.process_jobs.create(
              "PJ-Quiet", "RCP-1", {},
              [&](const std::string& ppid) { return m.recipes.get(ppid).has_value(); }) ==
          gem::ProcessJobStore::CreateResult::Created);
  REQUIRE(m.process_jobs.set_alert("PJ-Quiet", false));

  REQUIRE(m.process_jobs.fire_internal("PJ-Quiet", gem::ProcessJobEvent::Select));

  const auto deadline = std::chrono::steady_clock::now() + 100ms;
  while (std::chrono::steady_clock::now() < deadline) {
    if (wp.sp.io.stopped()) wp.sp.io.restart();
    wp.sp.io.poll();
    std::this_thread::sleep_for(5ms);
  }
  CHECK_FALSE(wp.host_saw(16, 9));
}

// ---------------- E94 Control Job CEIDs --------------------------------------

TEST_CASE("CJ Executing transition fires the configured ControlJobExecuting CEID") {
  static constexpr uint32_t kControlJobExecuting = 400;
  WirePair wp;
  gem::EquipmentDataModel m;
  // events.define_reports treats an empty VID list as "delete this
  // RPTID", so we register at least one SVID and reference it.
  m.svids.add({2001, "STATUS", "", secs2::Item::ascii("OK")});
  m.events.register_event({kControlJobExecuting, "ControlJobExecuting"});
  REQUIRE(m.define_reports({{700, {2001}}}) == gem::DefineReportAck::Accept);
  REQUIRE(m.link_event_reports({{kControlJobExecuting, {700}}}) ==
          gem::LinkEventAck::Accept);
  REQUIRE(m.enable_events(true, {kControlJobExecuting}) ==
          gem::EnableEventAck::Accept);

  m.control_jobs.set_state_change_handler(
      [&](const std::string&, gem::ControlJobState,
          gem::ControlJobState to, gem::ControlJobEvent) {
        if (to != gem::ControlJobState::Executing) return;
        if (!m.is_event_enabled(kControlJobExecuting)) return;
        wp.equipment->send_data(gem::s6f11_event_report(
            1, kControlJobExecuting, m.compose_reports_for(kControlJobExecuting)));
      });

  REQUIRE(m.control_jobs.create("CJ-X", {"PJ-1"},
                                [](const std::string&) { return true; }) ==
          gem::ControlJobStore::CreateResult::Created);
  REQUIRE(m.control_jobs.fire_internal("CJ-X", gem::ControlJobEvent::Select));
  REQUIRE(m.control_jobs.fire_internal("CJ-X", gem::ControlJobEvent::SetupComplete));
  REQUIRE(m.control_jobs.on_host_command("CJ-X", gem::ControlJobEvent::Start) ==
          gem::HostCmdAck::Accept);

  pump_until(wp.sp.io, [&] { return wp.host_saw(6, 11); });
  const auto* msg = wp.host_last(6, 11);
  REQUIRE(msg);
  auto parsed = gem::parse_s6f11(*msg);
  REQUIRE(parsed.has_value());
  CHECK(parsed->ceid == kControlJobExecuting);
}

// ---------------- E90 Substrate processing -----------------------------------

TEST_CASE("Substrate StartProcessing fires the configured SubstrateInProcess CEID") {
  static constexpr uint32_t kSubstrateInProcess = 500;
  WirePair wp;
  gem::EquipmentDataModel m;
  m.svids.add({3001, "WAFER", "", secs2::Item::ascii("W-1")});
  m.events.register_event({kSubstrateInProcess, "SubstrateInProcess"});
  REQUIRE(m.define_reports({{800, {3001}}}) == gem::DefineReportAck::Accept);
  REQUIRE(m.link_event_reports({{kSubstrateInProcess, {800}}}) ==
          gem::LinkEventAck::Accept);
  REQUIRE(m.enable_events(true, {kSubstrateInProcess}) ==
          gem::EnableEventAck::Accept);

  m.substrates.set_processing_handler(
      [&](const std::string&, gem::SubstrateProcessingState,
          gem::SubstrateProcessingState to, gem::SubstrateProcessingEvent) {
        if (to != gem::SubstrateProcessingState::InProcess) return;
        if (!m.is_event_enabled(kSubstrateInProcess)) return;
        wp.equipment->send_data(gem::s6f11_event_report(
            1, kSubstrateInProcess, m.compose_reports_for(kSubstrateInProcess)));
      });

  REQUIRE(m.substrates.create("W-1") ==
          gem::SubstrateStore::CreateResult::Created);
  REQUIRE(m.substrates.fire_processing_event(
      "W-1", gem::SubstrateProcessingEvent::StartProcessing));

  pump_until(wp.sp.io, [&] { return wp.host_saw(6, 11); });
  const auto* msg = wp.host_last(6, 11);
  REQUIRE(msg);
  auto parsed = gem::parse_s6f11(*msg);
  REQUIRE(parsed.has_value());
  CHECK(parsed->ceid == kSubstrateInProcess);
}