#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN
#include <doctest/doctest.h>

#include <grpcpp/grpcpp.h>

#include <chrono>
#include <thread>

#include "secsgem/daemon/equipment_service.hpp"
#include "secsgem/gem/default_handlers.hpp"
#include "secsgem/gem/messages.hpp"

using namespace secsgem;
namespace gem = secsgem::gem;
namespace s2  = secsgem::secs2;
namespace pb  = secsgem::v1;
namespace dmn = secsgem::daemon;

#ifndef SECSGEM_DATA_DIR
#error "SECSGEM_DATA_DIR not defined; see CMakeLists.txt"
#endif

static gem::EquipmentRuntime::Config test_config() {
  gem::EquipmentRuntime::Config c;
  c.equipment_yaml     = SECSGEM_DATA_DIR "/equipment.yaml";
  c.control_state_yaml = SECSGEM_DATA_DIR "/control_state.yaml";
  c.process_job_yaml   = SECSGEM_DATA_DIR "/process_job_state.yaml";
  c.control_job_yaml   = SECSGEM_DATA_DIR "/control_job_state.yaml";
  c.port = 0;  // ephemeral; the engine isn't run() here, only poll()ed
  return c;
}

// Exercises the real gRPC service over an in-process channel: client stub ->
// service -> EquipmentRuntime, proving the RPCs move data, not just compile.
TEST_CASE("Equipment gRPC service over an in-process channel") {
  gem::EquipmentRuntime rt(test_config());
  dmn::EquipmentService svc(rt);

  grpc::ServerBuilder builder;
  builder.RegisterService(&svc);
  std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
  REQUIRE(server);
  auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));

  SUBCASE("GetControlState returns the initial control state") {
    grpc::ClientContext ctx;
    pb::Empty req;
    pb::ControlState resp;
    auto st = stub->GetControlState(&ctx, req, &resp);
    CHECK(st.ok());
    CHECK(resp.state() == pb::ControlState::HOST_OFFLINE);
  }

  SUBCASE("SetVariables converts to the variable's declared wire format") {
    // ChamberPressure is declared F4 and WaferCounter U4 in equipment.yaml;
    // the daemon must honour those, not write F8/I8, or the host sees values
    // whose format contradicts the S1F11/S1F21 namelists.
    grpc::ClientContext ctx;
    pb::VariableUpdate req;
    pb::Ack resp;
    (*req.mutable_values())["ChamberPressure"].set_real(2.5);
    (*req.mutable_values())["WaferCounter"].set_integer(7);
    auto st = stub->SetVariables(&ctx, req, &resp);
    CHECK(st.ok());
    CHECK(resp.code() == pb::Ack::ACCEPT);
    rt.poll();  // drain the posted set_variable onto this thread
    CHECK(rt.model().dvids.value(101) == s2::Item::f4(2.5f));
    CHECK(rt.model().dvids.value(100) == s2::Item::u4(uint32_t{7}));
  }

  SUBCASE("SetVariables rejects an unknown variable name") {
    grpc::ClientContext ctx;
    pb::VariableUpdate req;
    pb::Ack resp;
    (*req.mutable_values())["definitely_not_a_var"].set_real(1.0);
    auto st = stub->SetVariables(&ctx, req, &resp);
    CHECK(st.ok());
    CHECK(resp.code() == pb::Ack::PARAMETER_INVALID);
  }

  SUBCASE("SetVariables rejects a Value with no kind set (silent-'' guard)") {
    grpc::ClientContext ctx;
    pb::VariableUpdate req;
    pb::Ack resp;
    (*req.mutable_values())["ChamberPressure"];  // map entry, kind never set
    auto st = stub->SetVariables(&ctx, req, &resp);
    CHECK(st.ok());
    CHECK(resp.code() == pb::Ack::PARAMETER_INVALID);
    // The previous value must be untouched (no silent ASCII "" write).
    rt.poll();
    CHECK(rt.model().dvids.value(101)->format() == s2::Format::F4);
  }

  SUBCASE("property: every configured variable keeps its declared format") {
    // Iterate ALL SVIDs+DVIDs from the live config: set each via gRPC with a
    // type-appropriate plain value and assert the stored Item keeps the
    // declared wire format. Catches any future variable whose format the
    // conversion table doesn't handle — not just the two pinned above.
    auto check_all = [&](const auto& entries) {
      for (const auto& v : entries) {
        const auto declared = v.value.format();
        grpc::ClientContext ctx;
        pb::VariableUpdate req;
        pb::Ack resp;
        pb::Value val;
        switch (declared) {
          case s2::Format::ASCII: case s2::Format::JIS8:
            val.set_text("x"); break;
          case s2::Format::Boolean:
            val.set_boolean(true); break;
          case s2::Format::Binary:
            val.set_binary("\x01"); break;
          case s2::Format::F4: case s2::Format::F8:
            val.set_real(1.5); break;
          default:  // all integer widths
            val.set_integer(1); break;
        }
        (*req.mutable_values())[v.name] = val;
        REQUIRE(stub->SetVariables(&ctx, req, &resp).ok());
        REQUIRE_MESSAGE(resp.code() == pb::Ack::ACCEPT, v.name);
        rt.poll();
        auto stored = rt.model().vid_value(v.id);
        REQUIRE_MESSAGE(stored.has_value(), v.name);
        CHECK_MESSAGE(stored->format() == declared,
                      v.name, ": declared ", static_cast<int>(declared),
                      " stored ", static_cast<int>(stored->format()));
      }
    };
    check_all(rt.model().svids.all());
    check_all(rt.model().dvids.all());
  }

  SUBCASE("SetAlarm / ClearAlarm by config name, by stringified id, unknown rejected") {
    auto call = [&](auto method, const std::string& name) {
      grpc::ClientContext ctx;
      pb::Alarm req;
      pb::Ack resp;
      req.set_name(name);
      REQUIRE((stub.get()->*method)(&ctx, req, &resp).ok());
      return resp.code();
    };
    // By config name.
    CHECK(call(&pb::Equipment::Stub::SetAlarm, "chiller_temp_high") == pb::Ack::ACCEPT);
    rt.poll();
    CHECK(rt.model().alarms.active(1));
    CHECK(call(&pb::Equipment::Stub::ClearAlarm, "chiller_temp_high") == pb::Ack::ACCEPT);
    rt.poll();
    CHECK_FALSE(rt.model().alarms.active(1));
    // By stringified ALID — always works, even for unnamed alarms.
    CHECK(call(&pb::Equipment::Stub::SetAlarm, "2") == pb::Ack::ACCEPT);
    rt.poll();
    CHECK(rt.model().alarms.active(2));
    // Unknown name.
    CHECK(call(&pb::Equipment::Stub::SetAlarm, "no_such_alarm") == pb::Ack::PARAMETER_INVALID);
  }

  SUBCASE("FireEvent accepts a known event and rejects an unknown one") {
    {
      grpc::ClientContext ctx;
      pb::Event req;
      pb::Ack resp;
      req.set_name("ProcessStarted");
      CHECK(stub->FireEvent(&ctx, req, &resp).ok());
      CHECK(resp.code() == pb::Ack::ACCEPT);
    }
    {
      grpc::ClientContext ctx;
      pb::Event req;
      pb::Ack resp;
      req.set_name("NoSuchEvent");
      CHECK(stub->FireEvent(&ctx, req, &resp).ok());
      CHECK(resp.code() == pb::Ack::PARAMETER_INVALID);
    }
  }

  server->Shutdown();
}

// GetVariables needs a live io thread (read_sync posts onto it), so this case
// runs the engine in run_async() — the daemon's PRODUCTION threading mode —
// with real concurrency between the gRPC handler thread and the io thread.
TEST_CASE("GetVariables round-trip under run_async (production threading mode)") {
  gem::EquipmentRuntime rt(test_config());
  dmn::EquipmentService svc(rt);   // snapshot BEFORE the io thread starts
  rt.run_async();

  grpc::ServerBuilder builder;
  builder.RegisterService(&svc);
  std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
  REQUIRE(server);
  auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));

  // Write through the API, then read back through the API: exercises BOTH
  // conversions (Value->Item with declared formats, Item->Value) end to end.
  {
    grpc::ClientContext ctx;
    pb::VariableUpdate req;
    pb::Ack resp;
    (*req.mutable_values())["ChamberPressure"].set_real(2.5);
    (*req.mutable_values())["WaferCounter"].set_integer(7);
    REQUIRE(stub->SetVariables(&ctx, req, &resp).ok());
    REQUIRE(resp.code() == pb::Ack::ACCEPT);
  }
  {
    grpc::ClientContext ctx;
    pb::VariableQuery req;
    pb::VariableSnapshot resp;
    req.add_names("ChamberPressure");
    req.add_names("WaferCounter");
    auto st = stub->GetVariables(&ctx, req, &resp);
    REQUIRE(st.ok());
    REQUIRE(resp.values().count("ChamberPressure") == 1);
    REQUIRE(resp.values().count("WaferCounter") == 1);
    CHECK(resp.values().at("ChamberPressure").real() == doctest::Approx(2.5));
    CHECK(resp.values().at("WaferCounter").integer() == 7);
  }

  SUBCASE("empty query returns every configured variable") {
    grpc::ClientContext ctx;
    pb::VariableQuery req;
    pb::VariableSnapshot resp;
    REQUIRE(stub->GetVariables(&ctx, req, &resp).ok());
    // The io thread is live here — model reads must go through read_sync
    // (the first violation of that contract was caught by the TSan lane in
    // exactly this line).
    auto expected = rt.read_sync([&rt] {
      return rt.model().svids.size() + rt.model().dvids.all().size();
    });
    REQUIRE(expected.has_value());
    CHECK(resp.values().size() == *expected);
  }

  SUBCASE("unknown name is INVALID_ARGUMENT, naming the offender") {
    grpc::ClientContext ctx;
    pb::VariableQuery req;
    pb::VariableSnapshot resp;
    req.add_names("definitely_not_a_var");
    auto st = stub->GetVariables(&ctx, req, &resp);
    CHECK(st.error_code() == grpc::StatusCode::INVALID_ARGUMENT);
    CHECK(st.error_message().find("definitely_not_a_var") != std::string::npos);
  }

  SUBCASE("RequestControlState walks the E30 table; WatchHealth pushes the change") {
    // Open the health stream first: the initial snapshot arrives immediately.
    grpc::ClientContext health_ctx;
    pb::Empty empty;
    auto reader = stub->WatchHealth(&health_ctx, empty);
    pb::Health h;
    REQUIRE(reader->Read(&h));
    CHECK(h.link() == pb::Health::DISCONNECTED);  // no HSMS host in this test
    CHECK(h.control_state() == pb::ControlState::HOST_OFFLINE);
    CHECK(h.spool_depth() == 0);

    auto request = [&](pb::ControlState::State desired) {
      grpc::ClientContext ctx;
      pb::ControlStateRequest req;
      pb::Ack resp;
      req.set_desired(desired);
      REQUIRE(stub->RequestControlState(&ctx, req, &resp).ok());
      return resp;
    };

    // HostOffline --operator_online--> (AttemptOnline) --> OnlineLocal.
    CHECK(request(pb::ControlState::ONLINE_LOCAL).code() == pb::Ack::ACCEPT);
    CHECK(rt.control_state() == gem::ControlState::OnlineLocal);

    // The stream pushes the change (well inside its 500ms poll interval).
    REQUIRE(reader->Read(&h));
    CHECK(h.control_state() == pb::ControlState::ONLINE_LOCAL);

    // OnlineLocal --operator_remote--> OnlineRemote.
    CHECK(request(pb::ControlState::ONLINE_REMOTE).code() == pb::Ack::ACCEPT);
    CHECK(rt.control_state() == gem::ControlState::OnlineRemote);

    // Transient state is not requestable.
    CHECK(request(pb::ControlState::ATTEMPT_ONLINE).code() == pb::Ack::PARAMETER_INVALID);

    // The shipped table has NO operator path to EquipmentOffline:
    // operator_offline lands HostOffline — the API must say so honestly.
    auto resp = request(pb::ControlState::EQUIPMENT_OFFLINE);
    CHECK(resp.code() == pb::Ack::CANNOT_DO_NOW);
    CHECK(resp.message().find("HostOffline") != std::string::npos);
    CHECK(rt.control_state() == gem::ControlState::HostOffline);

    // Requesting HOST_OFFLINE (the state operators actually get) succeeds —
    // idempotently, since we're already there.
    CHECK(request(pb::ControlState::HOST_OFFLINE).code() == pb::Ack::ACCEPT);

    health_ctx.TryCancel();
    pb::Health drain;
    while (reader->Read(&drain)) {}
    (void)reader->Finish();  // CANCELLED — expected
  }

  server->Shutdown();
  rt.stop();
}

// The Subscribe command stream, per the HCACK-4 contract: a real S2F41 is
// dispatched through the full default-handler router ON the io thread, while
// the gRPC tool client consumes the stream — the daemon's production shape.
TEST_CASE("Subscribe: S2F41 -> stream -> HCACK 4; declarative fallback without subscriber") {
  gem::EquipmentRuntime rt(test_config());
  gem::register_default_handlers(rt);   // the real S2F41/F21/F49 router path
  dmn::EquipmentService svc(rt);        // registers the forwarding handlers
  rt.run_async();

  grpc::ServerBuilder builder;
  builder.RegisterService(&svc);
  std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
  REQUIRE(server);
  auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));

  // Dispatch a wire-shaped S2F41 on the io thread (the model's owner) and
  // hand back the parsed S2F42 ack.
  auto send_s2f41 = [&](const std::string& rcmd,
                        std::vector<gem::CommandParameter> params) {
    auto reply = rt.read_sync([&rt, &rcmd, &params]() {
      return rt.router().dispatch(gem::s2f41_host_command(rcmd, params));
    });
    REQUIRE(reply.has_value());          // read_sync answered
    REQUIRE(reply->has_value());         // router produced an S2F42
    auto parsed = gem::parse_s2f42(**reply);
    REQUIRE(parsed.has_value());
    return parsed->hcack;
  };

  // --- no subscriber: declarative ack (START is Accept in equipment.yaml) ---
  CHECK(send_s2f41("START", {}) == gem::HostCmdAck::Accept);

  // --- subscriber connected: HCACK 4 + the command arrives on the stream ----
  grpc::ClientContext sub_ctx;
  pb::SubscribeRequest sreq;
  sreq.set_client("test-tool");
  auto reader = stub->Subscribe(&sub_ctx, sreq);

  // Subscription registration races the dispatch below only in this test
  // (the registry insert happens on the gRPC server thread); give it a beat.
  std::this_thread::sleep_for(std::chrono::milliseconds(100));

  CHECK(send_s2f41("START", {{"PPID", s2::Item::ascii("RECIPE-A")}}) ==
        gem::HostCmdAck::AcceptedWillFinishLater);

  pb::HostRequest hr;
  REQUIRE(reader->Read(&hr));
  REQUIRE(hr.has_command());
  CHECK(hr.command().name() == "START");
  CHECK_FALSE(hr.command().id().empty());
  REQUIRE(hr.command().params().count("PPID") == 1);
  CHECK(hr.command().params().at("PPID").text() == "RECIPE-A");

  // --- CompleteCommand: known id accepted, unknown rejected ------------------
  {
    grpc::ClientContext ctx;
    pb::CommandResult res;
    pb::Ack ack;
    res.set_id(hr.command().id());
    res.mutable_ack()->set_code(pb::Ack::ACCEPT);
    REQUIRE(stub->CompleteCommand(&ctx, res, &ack).ok());
    CHECK(ack.code() == pb::Ack::ACCEPT);
  }
  {
    grpc::ClientContext ctx;
    pb::CommandResult res;
    pb::Ack ack;
    res.set_id("no-such-id");
    REQUIRE(stub->CompleteCommand(&ctx, res, &ack).ok());
    CHECK(ack.code() == pb::Ack::PARAMETER_INVALID);
  }

  // --- unsubscribe: the fallback returns ------------------------------------
  sub_ctx.TryCancel();
  pb::HostRequest drain;
  while (reader->Read(&drain)) {}
  (void)reader->Finish();  // CANCELLED — expected
  // The server-side Subscribe loop notices the cancel within its 500ms poll.
  std::this_thread::sleep_for(std::chrono::milliseconds(700));
  CHECK(send_s2f41("START", {}) == gem::HostCmdAck::Accept);

  server->Shutdown();
  rt.stop();
}