#pragma once

// The gRPC Equipment service: translates proto/secsgem/v1 RPCs onto an
// EquipmentRuntime. Header-only so both secs_gemd and the daemon tests share
// one definition.
//
// Threading: gRPC handlers run on gRPC's own threads while the engine's
// io thread owns the data model, so handlers must not read the live stores.
// The constructor therefore snapshots the name->id/format maps (immutable
// after config load) — construct the service BEFORE run_async(). All writes
// go through the runtime's posting API.

#include <grpcpp/grpcpp.h>

#include <cstdint>
#include <map>
#include <string>
#include <utility>
#include <vector>

#include "secsgem/gem/runtime.hpp"
#include "secsgem/secs2/item.hpp"
#include "secsgem/v1/equipment.grpc.pb.h"
#include "secsgem/v1/equipment.pb.h"

namespace secsgem::daemon {

namespace gem = secsgem::gem;
namespace s2  = secsgem::secs2;
namespace pb  = secsgem::v1;

// proto Value -> SECS-II Item, honouring the variable's declared wire format
// (from equipment.yaml) so the host sees the same format in S1F11 namelists
// and in reported values. E.g. real 2.5 -> F4 if the variable is F4.
inline s2::Item to_item(const pb::Value& v, s2::Format want) {
  using F = s2::Format;
  switch (v.kind_case()) {
    case pb::Value::kText:
      return s2::Item::ascii(v.text());
    case pb::Value::kBoolean:
      return s2::Item::boolean(v.boolean());
    case pb::Value::kBinary:
      return s2::Item::binary({v.binary().begin(), v.binary().end()});
    case pb::Value::kReal:
      return want == F::F4 ? s2::Item::f4(static_cast<float>(v.real()))
                           : s2::Item::f8(v.real());
    case pb::Value::kInteger: {
      const int64_t n = v.integer();
      switch (want) {
        case F::U1: return s2::Item::u1(static_cast<uint8_t>(n));
        case F::U2: return s2::Item::u2(static_cast<uint16_t>(n));
        case F::U4: return s2::Item::u4(static_cast<uint32_t>(n));
        case F::U8: return s2::Item::u8(static_cast<uint64_t>(n));
        case F::I1: return s2::Item::i1(static_cast<int8_t>(n));
        case F::I2: return s2::Item::i2(static_cast<int16_t>(n));
        case F::I4: return s2::Item::i4(static_cast<int32_t>(n));
        case F::F4: return s2::Item::f4(static_cast<float>(n));
        case F::F8: return s2::Item::f8(static_cast<double>(n));
        case F::Boolean: return s2::Item::boolean(n != 0);
        default:    return s2::Item::i8(n);
      }
    }
    case pb::Value::kList: {
      // TODO(daemon): list elements inherit the variable's scalar format;
      // honouring per-element formats needs the declared Item's nested shape.
      s2::Item::List items;
      for (const auto& e : v.list().items()) items.push_back(to_item(e, want));
      return s2::Item::list(std::move(items));
    }
    default:
      // Unreachable: callers reject KIND_NOT_SET before converting (see
      // value_is_set). Kept as a safe fallback for future oneof additions.
      return s2::Item::ascii("");
  }
}

// Validate a client-supplied Value before conversion. An unset oneof would
// otherwise silently become ASCII "" — reject it at the API edge instead.
inline bool value_is_set(const pb::Value& v) {
  return v.kind_case() != pb::Value::KIND_NOT_SET;
}

// SECS-II Item -> proto Value (the read direction). Single-element numeric
// arrays — the overwhelmingly common case — become scalars; multi-element
// arrays become a List so nothing is lost.
// TODO(daemon): C2 (2-byte Unicode) currently surfaces as integer code
// points (it shares storage with U2); convert to text when a tool needs it.
// TODO(daemon): U8 values above 2^63-1 wrap through the sint64 integer
// field; switch Value to a dedicated uint64 arm if that ever bites.
inline pb::Value from_item(const s2::Item& item) {
  pb::Value out;
  switch (item.format()) {
    case s2::Format::List: {
      auto* list = out.mutable_list();
      for (const auto& child : item.as_list()) *list->add_items() = from_item(child);
      return out;
    }
    case s2::Format::ASCII:
    case s2::Format::JIS8:
      out.set_text(item.as_ascii());
      return out;
    case s2::Format::Binary: {
      const auto& b = item.as_bytes();
      out.set_binary(std::string(b.begin(), b.end()));
      return out;
    }
    case s2::Format::Boolean: {
      const auto& b = item.as_bytes();
      if (b.size() == 1) {
        out.set_boolean(b[0] != 0);
      } else {
        auto* list = out.mutable_list();
        for (auto v : b) list->add_items()->set_boolean(v != 0);
      }
      return out;
    }
    default:
      break;  // numeric formats: handled generically below
  }
  return std::visit(
      [&](const auto& vec) -> pb::Value {
        using V = std::decay_t<decltype(vec)>;
        if constexpr (std::is_same_v<V, s2::Item::List> ||
                      std::is_same_v<V, std::string>) {
          return out;  // unreachable: handled above
        } else {
          auto set_one = [](pb::Value& dst, auto v) {
            if constexpr (std::is_floating_point_v<decltype(v)>)
              dst.set_real(static_cast<double>(v));
            else
              dst.set_integer(static_cast<int64_t>(v));
          };
          if (vec.size() == 1) {
            set_one(out, vec[0]);
          } else {
            auto* list = out.mutable_list();
            for (auto v : vec) set_one(*list->add_items(), v);
          }
          return out;
        }
      },
      item.storage());
}

inline pb::ControlState::State to_proto_state(gem::ControlState s) {
  switch (s) {
    case gem::ControlState::EquipmentOffline: return pb::ControlState::EQUIPMENT_OFFLINE;
    case gem::ControlState::AttemptOnline:    return pb::ControlState::ATTEMPT_ONLINE;
    case gem::ControlState::HostOffline:      return pb::ControlState::HOST_OFFLINE;
    case gem::ControlState::OnlineLocal:      return pb::ControlState::ONLINE_LOCAL;
    case gem::ControlState::OnlineRemote:     return pb::ControlState::ONLINE_REMOTE;
  }
  return pb::ControlState::EQUIPMENT_OFFLINE;
}

class EquipmentService final : public pb::Equipment::Service {
 public:
  // Snapshots the (immutable) name->id/format dictionaries. Construct before
  // run_async() so the model is read while the io thread isn't running yet.
  explicit EquipmentService(gem::EquipmentRuntime& rt) : rt_(rt) {
    for (const auto& sv : rt.model().svids.all())
      vars_.insert({sv.name, {sv.id, sv.value.format()}});
    for (const auto& dv : rt.model().dvids.all())
      vars_.insert({dv.name, {dv.id, dv.value.format()}});
    for (const auto& ev : rt.model().events.all_events())
      events_.insert({ev.name, ev.id});
  }

  grpc::Status SetVariables(grpc::ServerContext*, const pb::VariableUpdate* req,
                            pb::Ack* resp) override {
    for (const auto& kv : req->values()) {
      auto it = vars_.find(kv.first);
      if (it == vars_.end()) {
        resp->set_code(pb::Ack::PARAMETER_INVALID);
        resp->set_message("no variable named '" + kv.first + "'");
        return grpc::Status::OK;
      }
      if (!value_is_set(kv.second)) {
        resp->set_code(pb::Ack::PARAMETER_INVALID);
        resp->set_message("value for '" + kv.first + "' has no kind set");
        return grpc::Status::OK;
      }
      rt_.set_variable(it->second.vid, to_item(kv.second, it->second.format));
    }
    resp->set_code(pb::Ack::ACCEPT);
    return grpc::Status::OK;
  }

  grpc::Status FireEvent(grpc::ServerContext*, const pb::Event* req,
                         pb::Ack* resp) override {
    // Optional per-fire variable values, then trigger the collection event.
    for (const auto& kv : req->data()) {
      auto it = vars_.find(kv.first);
      if (it == vars_.end()) {
        resp->set_code(pb::Ack::PARAMETER_INVALID);
        resp->set_message("no variable named '" + kv.first + "'");
        return grpc::Status::OK;
      }
      if (!value_is_set(kv.second)) {
        resp->set_code(pb::Ack::PARAMETER_INVALID);
        resp->set_message("value for '" + kv.first + "' has no kind set");
        return grpc::Status::OK;
      }
      rt_.set_variable(it->second.vid, to_item(kv.second, it->second.format));
    }
    auto ev = events_.find(req->name());
    if (ev == events_.end()) {
      resp->set_code(pb::Ack::PARAMETER_INVALID);
      resp->set_message("no event named '" + req->name() + "'");
      return grpc::Status::OK;
    }
    rt_.emit_event(ev->second);
    resp->set_code(pb::Ack::ACCEPT);
    return grpc::Status::OK;
  }

  grpc::Status GetVariables(grpc::ServerContext*, const pb::VariableQuery* req,
                            pb::VariableSnapshot* resp) override {
    // Resolve names against the snapshot maps (empty query = everything).
    std::vector<std::pair<std::string, uint32_t>> wanted;
    if (req->names().empty()) {
      wanted.reserve(vars_.size());
      for (const auto& [name, ref] : vars_) wanted.emplace_back(name, ref.vid);
    } else {
      for (const auto& name : req->names()) {
        auto it = vars_.find(name);
        if (it == vars_.end())
          return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
                              "no variable named '" + name + "'");
        wanted.emplace_back(name, it->second.vid);
      }
    }
    // Mutable values live on the io thread — read them there (read_sync is
    // the standard pattern; see runtime.hpp).
    std::vector<uint32_t> vids;
    vids.reserve(wanted.size());
    for (const auto& w : wanted) vids.push_back(w.second);
    auto values = rt_.read_sync([this, vids]() {
      std::vector<std::optional<s2::Item>> out;
      out.reserve(vids.size());
      for (auto vid : vids) out.push_back(rt_.model().vid_value(vid));
      return out;
    });
    if (!values)
      return grpc::Status(grpc::StatusCode::UNAVAILABLE,
                          "engine io thread did not answer (not running?)");
    for (std::size_t i = 0; i < wanted.size(); ++i) {
      if (!(*values)[i]) continue;  // vid vanished — defensive, can't happen
      (*resp->mutable_values())[wanted[i].first] = from_item(*(*values)[i]);
    }
    return grpc::Status::OK;
  }

  grpc::Status GetControlState(grpc::ServerContext*, const pb::Empty*,
                               pb::ControlState* resp) override {
    // Thread-safe: control_state() reads the runtime's atomic mirror.
    resp->set_state(to_proto_state(rt_.control_state()));
    return grpc::Status::OK;
  }

 private:
  struct VarRef {
    uint32_t vid;
    s2::Format format;  // declared wire format from equipment.yaml
  };

  gem::EquipmentRuntime& rt_;
  std::map<std::string, VarRef> vars_;     // SVIDs + DVIDs (SVIDs win on clash)
  std::map<std::string, uint32_t> events_; // CEID by name
};

}  // namespace secsgem::daemon