// Real-socket tests for S9F7 (illegal data) emission.
//
// S9F3 / F5 are covered by test_s9_fallback.cpp; S9F9 / F11 by
// test_hsms_timers.cpp.  S9F7 sits in between: it fires when the peer
// sends a primary (or a reply) whose body fails secs2::decode.

#include <doctest/doctest.h>

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

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

using namespace secsgem::hsms;
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();
    std::error_code ea, eb;
    bool da = false, db = false;
    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);
  }
}

void send_bytes(SocketPair& sp, std::vector<uint8_t> bytes) {
  auto buf = std::make_shared<std::vector<uint8_t>>(std::move(bytes));
  bool done = false;
  asio::async_write(sp.b, asio::buffer(*buf),
                    [buf, &done](std::error_code ec, std::size_t) {
                      REQUIRE_FALSE(ec);
                      done = true;
                    });
  pump_until(sp.io, [&] { return done; });
}

std::optional<Frame> try_recv_frame(SocketPair& sp,
                                     std::chrono::milliseconds budget = 2s) {
  auto lenbuf = std::make_shared<std::array<uint8_t, 4>>();
  bool len_done = false;
  std::error_code rec_ec;
  asio::async_read(sp.b, asio::buffer(*lenbuf),
                   [lenbuf, &len_done, &rec_ec](std::error_code ec, std::size_t) {
                     rec_ec = ec; len_done = true;
                   });
  const auto deadline = std::chrono::steady_clock::now() + budget;
  while (!len_done) {
    if (std::chrono::steady_clock::now() > deadline) return std::nullopt;
    if (sp.io.stopped()) sp.io.restart();
    if (sp.io.poll() == 0) std::this_thread::sleep_for(1ms);
  }
  if (rec_ec) return std::nullopt;
  const uint32_t len = (uint32_t((*lenbuf)[0]) << 24) | (uint32_t((*lenbuf)[1]) << 16) |
                       (uint32_t((*lenbuf)[2]) << 8) | uint32_t((*lenbuf)[3]);
  auto payload = std::make_shared<std::vector<uint8_t>>(len);
  bool payload_done = false;
  asio::async_read(sp.b, asio::buffer(*payload),
                   [payload, &payload_done](std::error_code, std::size_t) {
                     payload_done = true;
                   });
  pump_until(sp.io, [&] { return payload_done; });
  return Frame::decode(payload->data(), payload->size());
}

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

// Build a SELECTED passive connection, ready to be poked.
std::pair<std::shared_ptr<Connection>, std::unique_ptr<SocketPair>>
make_selected_passive() {
  auto sp = std::make_unique<SocketPair>();
  auto conn = std::make_shared<Connection>(std::move(sp->a),
                                            Connection::Mode::Passive,
                                            /*device_id=*/0, permissive_timers());
  bool selected = false;
  conn->set_selected_handler([&] { selected = true; });
  conn->start();
  send_bytes(*sp, Frame(Header::control(SType::SelectReq, 1)).encode());
  auto rsp = try_recv_frame(*sp);
  REQUIRE(rsp.has_value());
  REQUIRE(rsp->header.stype == SType::SelectRsp);
  pump_until(sp->io, [&] { return selected; });
  return {conn, std::move(sp)};
}

}  // namespace

TEST_CASE("S9F7: primary with malformed body emits S9F7 echoing original MHEAD") {
  auto [conn, sp] = make_selected_passive();

  // Build a primary S1F3 frame with a deliberately broken body — the
  // formatcode prefix 0x21 says "binary, 1 length byte" but we don't
  // supply the length byte.  Body of just {0x21}.
  Frame primary(Header::data_message(/*session=*/0, /*stream=*/1,
                                       /*function=*/3, /*reply_expected=*/true,
                                       /*sys=*/777),
                std::vector<uint8_t>{0x21});
  send_bytes(*sp, primary.encode());

  auto s9 = try_recv_frame(*sp);
  REQUIRE(s9.has_value());
  CHECK(s9->header.stream() == 9);
  CHECK(s9->header.function() == 7);
  // S9F7 body: <B 10> MHEAD echo.
  REQUIRE(s9->body.size() >= 12);
  CHECK(s9->body[0] == 0x21);
  CHECK(s9->body[1] == 10);
  // MHEAD bytes 6..9 are the original sys=777 = 0x00000309.
  const uint32_t echoed_sys =
      (uint32_t(s9->body[2 + 6]) << 24) | (uint32_t(s9->body[2 + 7]) << 16) |
      (uint32_t(s9->body[2 + 8]) << 8) | uint32_t(s9->body[2 + 9]);
  CHECK(echoed_sys == 777);
  // MHEAD byte2's lower 7 bits should be stream 1; bit7 carries the W-bit.
  CHECK((s9->body[2 + 2] & 0x7F) == 1);
  CHECK(s9->body[2 + 3] == 3);  // function

  conn->close("test done");
}

TEST_CASE("S9F7: emission keeps the connection up (next frame still routes)") {
  auto [conn, sp] = make_selected_passive();

  bool got_good = false;
  conn->set_message_handler([&](const secsgem::secs2::Message& m)
                              -> std::optional<secsgem::secs2::Message> {
    if (m.stream == 1 && m.function == 13) {
      got_good = true;
      return secsgem::secs2::Message(1, 14, false);
    }
    return std::nullopt;
  });

  // First send a broken frame to provoke S9F7.
  Frame broken(Header::data_message(0, 1, 3, true, 100), {0x21});
  send_bytes(*sp, broken.encode());
  auto s9 = try_recv_frame(*sp);
  REQUIRE(s9.has_value());
  CHECK(s9->header.function() == 7);

  // Now send a well-formed S1F13 — it should still route to the handler.
  Frame good(Header::data_message(0, 1, 13, true, 101), {});
  send_bytes(*sp, good.encode());

  auto reply = try_recv_frame(*sp);
  REQUIRE(reply.has_value());
  CHECK(reply->header.stream() == 1);
  CHECK(reply->header.function() == 14);
  CHECK(got_good);

  conn->close("test done");
}

TEST_CASE("Data frame while NOT_SELECTED still echoes Reject (sanity vs S9 paths)") {
  // This case lives in test_hsms_connection.cpp already; including it
  // here too because the family of "what does the equipment say when
  // peer sends garbage" tests reads better as one place.
  SocketPair sp;
  auto conn = std::make_shared<Connection>(std::move(sp.a),
                                            Connection::Mode::Passive,
                                            /*device_id=*/0, permissive_timers());
  conn->start();

  Frame data(Header::data_message(0, 1, 1, true, 42));
  send_bytes(sp, data.encode());

  auto rej = try_recv_frame(sp);
  REQUIRE(rej.has_value());
  CHECK(rej->header.stype == SType::RejectReq);
  CHECK(rej->header.byte3 == static_cast<uint8_t>(RejectReason::EntityNotSelected));

  conn->close("test done");
}