// E87 slot-map mismatch + multi-port concurrent-carrier wire tests.
//
// At the FSM level, CSMS NotRead → Read → Mismatched is covered by
// test_carrier_state.cpp.  This test exercises the same transitions
// across an HSMS wire (S3F19 verify → S3F20 ack, with the equipment
// comparing the host-supplied slot vector against its stored one) and
// adds a multi-LP scenario that asserts independence between ports.

#include <doctest/doctest.h>

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

#include "secsgem/gem/data_model.hpp"
#include "secsgem/gem/messages.hpp"
#include "secsgem/hsms/connection.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}, 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;
}

// Slim equipment emulator that knows S3F19 only (slot-map verify).
struct E87Emulator {
  gem::EquipmentDataModel m;
  std::shared_ptr<hsms::Connection> conn;

  explicit E87Emulator(asio::ip::tcp::socket s) {
    conn = std::make_shared<hsms::Connection>(
        std::move(s), hsms::Connection::Mode::Passive, 0, permissive_timers());
    conn->set_message_handler(
        [this](const secs2::Message& msg) -> std::optional<secs2::Message> {
          if (msg.stream == 3 && msg.function == 19) {
            auto req = gem::parse_s3f19(msg);
            if (!req) return gem::s3f20_slot_map_verify_ack(
                gem::SlotMapVerifyAck::CarrierUnknown);
            auto* c = m.carriers.get(req->carrierid);
            if (!c) return gem::s3f20_slot_map_verify_ack(
                gem::SlotMapVerifyAck::CarrierUnknown);

            // Compare host-supplied slot vector vs equipment-stored one.
            bool match = (req->slots.size() == c->slots.size());
            for (std::size_t i = 0; match && i < req->slots.size(); ++i) {
              if (req->slots[i] != c->slots[i].state) match = false;
            }
            if (match) {
              m.carriers.fire_slot_map_event(req->carrierid,
                                              gem::SlotMapEvent::Read);
              m.carriers.fire_slot_map_event(req->carrierid,
                                              gem::SlotMapEvent::Confirm);
              return gem::s3f20_slot_map_verify_ack(
                  gem::SlotMapVerifyAck::Accept);
            }
            m.carriers.fire_slot_map_event(req->carrierid,
                                            gem::SlotMapEvent::Read);
            m.carriers.fire_slot_map_event(req->carrierid,
                                            gem::SlotMapEvent::Mismatch);
            return gem::s3f20_slot_map_verify_ack(
                gem::SlotMapVerifyAck::Mismatch);
          }
          return std::nullopt;
        });
  }
};

struct HostSide {
  std::shared_ptr<hsms::Connection> conn;
  explicit HostSide(asio::ip::tcp::socket s) {
    conn = std::make_shared<hsms::Connection>(
        std::move(s), hsms::Connection::Mode::Active, 0, permissive_timers());
  }
};

secs2::Message await_reply(asio::io_context& io,
                            const std::shared_ptr<hsms::Connection>& c,
                            secs2::Message req) {
  std::optional<secs2::Message> got;
  c->send_request(std::move(req),
                   [&](std::error_code ec, const secs2::Message& m) {
                     REQUIRE_FALSE(ec);
                     got = m;
                   });
  pump_until(io, [&] { return got.has_value(); });
  return *got;
}

}  // namespace

TEST_CASE("E87 slot-map verify on the wire: matching map → Accept, CSMS Read") {
  SocketPair sp;
  E87Emulator eq(std::move(sp.a));
  HostSide    host(std::move(sp.b));

  // Equipment pre-loads CAR-1 with slots [1,1,0,0] (occupied, occupied, empty, empty).
  eq.m.carriers.create("CAR-1", 1, 4);
  eq.m.carriers.get("CAR-1")->slots[0].state = 1;
  eq.m.carriers.get("CAR-1")->slots[1].state = 1;
  eq.m.carriers.get("CAR-1")->slots[2].state = 0;
  eq.m.carriers.get("CAR-1")->slots[3].state = 0;

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

  // Host sends the same map → equipment should ack Accept and land in Read.
  auto reply = await_reply(sp.io, host.conn,
                            gem::s3f19_slot_map_verify("CAR-1", {1, 1, 0, 0}));
  CHECK(reply.function == 20);
  CHECK(eq.m.carriers.get("CAR-1")->fsm->slot_map_status() ==
        gem::SlotMapStatus::Read);
}

TEST_CASE("E87 slot-map verify on the wire: mismatched map → CSMS Mismatched") {
  SocketPair sp;
  E87Emulator eq(std::move(sp.a));
  HostSide    host(std::move(sp.b));

  eq.m.carriers.create("CAR-2", 1, 4);
  eq.m.carriers.get("CAR-2")->slots[0].state = 1;
  eq.m.carriers.get("CAR-2")->slots[1].state = 0;
  eq.m.carriers.get("CAR-2")->slots[2].state = 0;
  eq.m.carriers.get("CAR-2")->slots[3].state = 0;

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

  // Host claims [1,1,0,0] but equipment has [1,0,0,0].
  auto reply = await_reply(sp.io, host.conn,
                            gem::s3f19_slot_map_verify("CAR-2", {1, 1, 0, 0}));
  CHECK(reply.function == 20);
  CHECK(eq.m.carriers.get("CAR-2")->fsm->slot_map_status() ==
        gem::SlotMapStatus::Mismatched);
}

TEST_CASE("Multi-LP: 4 ports are independent — events on one don't affect another") {
  SocketPair sp;
  E87Emulator eq(std::move(sp.a));
  HostSide    host(std::move(sp.b));

  // Create 4 LPs + 4 carriers, one per port.
  for (uint8_t pid : {1, 2, 3, 4}) {
    eq.m.load_ports.create(pid);
    std::string cid = "CAR-" + std::to_string(int(pid));
    eq.m.carriers.create(cid, pid, 4);
    auto* c = eq.m.carriers.get(cid);
    for (auto& s : c->slots) s.state = 1;
  }

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

  // Verify CAR-1 with a *mismatched* map (slots all 0 vs. all 1).
  auto r1 = await_reply(sp.io, host.conn,
                         gem::s3f19_slot_map_verify("CAR-1", {0, 0, 0, 0}));
  CHECK(r1.function == 20);

  // Verify CAR-3 with the matching map.
  auto r3 = await_reply(sp.io, host.conn,
                         gem::s3f19_slot_map_verify("CAR-3", {1, 1, 1, 1}));
  CHECK(r3.function == 20);

  // CAR-1 should be Mismatched; CAR-3 Read; CAR-2 + CAR-4 untouched.
  CHECK(eq.m.carriers.get("CAR-1")->fsm->slot_map_status() ==
        gem::SlotMapStatus::Mismatched);
  CHECK(eq.m.carriers.get("CAR-3")->fsm->slot_map_status() ==
        gem::SlotMapStatus::Read);
  CHECK(eq.m.carriers.get("CAR-2")->fsm->slot_map_status() ==
        gem::SlotMapStatus::NotRead);
  CHECK(eq.m.carriers.get("CAR-4")->fsm->slot_map_status() ==
        gem::SlotMapStatus::NotRead);
}

TEST_CASE("Multi-LP: per-port E84 FSMs run truly in parallel") {
  // E84 is now per-port (E84PortStore).  Two ports can be in completely
  // different states at the same time without sequencing.
  gem::EquipmentDataModel m;
  m.load_ports.create(1);
  m.load_ports.create(2);

  // LP1: interleave load-handshake bring-up with LP2: interleave
  // unload-handshake bring-up.  Both reach their respective Ready
  // states with no cross-contamination.
  m.e84_ports.on_signal_change(1, gem::E84Signal::CS_0, true);
  m.e84_ports.on_signal_change(2, gem::E84Signal::CS_0, true);
  m.e84_ports.on_signal_change(1, gem::E84Signal::VALID, true);
  m.e84_ports.on_signal_change(2, gem::E84Signal::VALID, true);
  m.e84_ports.on_signal_change(1, gem::E84Signal::L_REQ, true);
  m.e84_ports.on_signal_change(2, gem::E84Signal::U_REQ, true);

  CHECK(m.e84_ports.get(1)->state() == gem::E84State::LoadReady);
  CHECK(m.e84_ports.get(2)->state() == gem::E84State::UnloadReady);

  // ES on LP1 must not affect LP2.
  m.e84_ports.on_signal_change(1, gem::E84Signal::ES, true);
  CHECK(m.e84_ports.get(1)->state() == gem::E84State::EmergencyStop);
  CHECK(m.e84_ports.get(2)->state() == gem::E84State::UnloadReady);
}