#include "secsgem/gem/runtime.hpp"

#include "secsgem/gem/messages.hpp"

namespace secsgem::gem {

EquipmentRuntime::EquipmentRuntime(const Config& cfg)
    : model_(std::make_shared<EquipmentDataModel>()),
      log_(cfg.log),
      active_conn_(std::make_shared<std::weak_ptr<Connection>>()) {
  if (!cfg.spool_dir.empty()) {
    model_->spool.enable_persistence(cfg.spool_dir);
    log("spool: persisting to " + cfg.spool_dir + " (replayed " +
        std::to_string(model_->spool.size()) + " messages)");
  }

  descriptor_ = config::load_equipment(cfg.equipment_yaml, *model_);
  auto sm_cfg = config::load_control_state(cfg.control_state_yaml);
  sm_ = std::make_shared<ControlStateMachine>(sm_cfg.table, sm_cfg.initial);

  auto pj = config::load_process_job_state(cfg.process_job_yaml);
  auto cj = config::load_control_job_state(cfg.control_job_yaml);
  model_->process_jobs.set_table_factory([t = pj.table]() { return t; });
  model_->control_jobs.set_table_factory([t = cj.table]() { return t; });

  server_ = std::make_unique<Server>(
      io_, Server::Config{cfg.port, descriptor_.device_id, {}});
  server_->on_log([this](const std::string& m) { log(m); });
  wire_connection();
}

EquipmentRuntime::~EquipmentRuntime() { stop(); }

bool EquipmentRuntime::deliver_or_spool(secs2::Message msg,
                                        const std::string& what) {
  auto conn = active_conn_->lock();
  const bool spooling = model_->spool.force_spool() || !conn;
  if (spooling) {
    auto r = model_->spool.enqueue(msg);
    if (r == SpoolStore::EnqueueResult::Queued) {
      log("spool: " + what + " queued (depth=" +
          std::to_string(model_->spool.size()) + ")");
      return true;
    }
    log("spool: " + what + " dropped (stream not spoolable, no host)");
    return false;
  }
  if (msg.reply_expected) {
    conn->send_request(std::move(msg), [](std::error_code, const secs2::Message&) {});
  } else {
    conn->send_data(std::move(msg));
  }
  return true;
}

void EquipmentRuntime::set_variable(uint32_t vid, secs2::Item value) {
  asio::post(io_, [this, vid, value = std::move(value)]() mutable {
    if (model_->svids.has(vid)) model_->svids.set_value(vid, std::move(value));
    else if (model_->dvids.has(vid)) model_->dvids.set_value(vid, std::move(value));
  });
}

void EquipmentRuntime::emit_event(uint32_t ceid) {
  asio::post(io_, [this, ceid]() {
    if (!model_->is_event_enabled(ceid)) {
      log("CEID " + std::to_string(ceid) + " not enabled; suppressed");
      return;
    }
    auto reports = model_->compose_reports_for(ceid);
    log("emit S6F11 CEID=" + std::to_string(ceid) + " (" +
        std::to_string(reports.size()) + " reports)");
    deliver_or_spool(s6f11_event_report(0, ceid, reports),
                     "S6F11 CEID=" + std::to_string(ceid));
  });
}

void EquipmentRuntime::set_alarm(uint32_t alid) {
  asio::post(io_, [this, alid]() {
    auto alarm = model_->alarms.get(alid);
    if (!alarm) return;
    auto alcd = model_->alarms.set_active(alid);
    if (!alcd) return;
    if (model_->alarms.enabled(alid)) {
      log("emit S5F1 alarm set ALID=" + std::to_string(alid));
      deliver_or_spool(s5f1_alarm_report(*alcd, alid, alarm->text),
                       "S5F1 ALID=" + std::to_string(alid));
    } else {
      log("alarm " + std::to_string(alid) + " not enabled; suppressed");
    }
  });
}

void EquipmentRuntime::clear_alarm(uint32_t alid) {
  asio::post(io_, [this, alid]() {
    auto alarm = model_->alarms.get(alid);
    if (!alarm) return;
    auto alcd = model_->alarms.clear_active(alid);
    if (!alcd) return;
    if (model_->alarms.enabled(alid)) {
      log("emit S5F1 alarm clear ALID=" + std::to_string(alid));
      deliver_or_spool(s5f1_alarm_report(*alcd, alid, alarm->text),
                       "S5F1 clear ALID=" + std::to_string(alid));
    }
  });
}

void EquipmentRuntime::wire_connection() {
  server_->on_connection([this](std::shared_ptr<Connection> conn) {
    *active_conn_ = conn;
    conn->set_closed_handler([this](const std::string&) { active_conn_->reset(); });

    conn->set_selected_handler([this]() {
      log(std::string("host is online; control=") + control_state_name(sm_->state()));
      if (model_->spool.size() == 0) return;
      asio::post(io_, [this]() {
        auto c = active_conn_->lock();
        if (!c) return;
        const uint32_t n = static_cast<uint32_t>(model_->spool.size());
        log("spool: notifying host of " + std::to_string(n) + " queued messages");
        c->send_request(s6f25_spool_data_ready(n),
                        [](std::error_code, const secs2::Message&) {});
      });
    });

    std::weak_ptr<Connection> wconn = conn;
    conn->set_message_handler([this, wconn](const secs2::Message& msg)
                                  -> std::optional<secs2::Message> {
      if (!router_.has_handler(msg.stream, msg.function)) {
        auto c = wconn.lock();
        if (c && c->current_header()) {
          const uint8_t s9_function =
              router_.has_handler_for_stream(msg.stream) ? 5 : 3;
          log("unhandled S" + std::to_string(msg.stream) + "F" +
              std::to_string(msg.function) + "; emitting S9F" +
              std::to_string(s9_function));
          c->emit_s9(s9_function, c->current_header()->encode());
        }
      }
      return router_.dispatch(msg);
    });
  });
}

void EquipmentRuntime::run() {
  server_->start();
  io_.run();
}

void EquipmentRuntime::run_async() {
  work_.emplace(asio::make_work_guard(io_));
  server_->start();
  io_thread_ = std::thread([this]() { io_.run(); });
}

void EquipmentRuntime::poll() {
  // restart() clears the "stopped" state a prior drain leaves behind, so
  // repeated poll() calls each run their freshly-posted handlers.
  io_.restart();
  io_.poll();
}

void EquipmentRuntime::stop() {
  if (work_) {
    work_->reset();
    work_.reset();
  }
  io_.stop();
  if (io_thread_.joinable()) io_thread_.join();
}

}  // namespace secsgem::gem