secs-gem/apps/secs_server.cpp

// Passive SECS/GEM equipment. Capabilities (SVIDs, ECIDs, CEIDs, alarms,
// recipes, host commands) come from data/equipment.yaml; the E30 control
// state machine comes from data/control_state.yaml. Dispatch is a Router
// table.  No imperative if-ladder; no in-code device data dictionary.

#include <asio.hpp>
#include <chrono>
#include <cstdint>
#include <iostream>
#include <memory>
#include <optional>
#include <string>
#include <vector>

#include "secsgem/config/loader.hpp"
#include "secsgem/endpoint.hpp"
#include "secsgem/gem/control_state.hpp"
#include "secsgem/gem/data_model.hpp"
#include "secsgem/gem/messages.hpp"
#include "secsgem/gem/router.hpp"
#include "secsgem/secs2/message.hpp"

using namespace secsgem;
namespace s2 = secsgem::secs2;
namespace gem = secsgem::gem;

namespace {

std::string arg(int argc, char** argv, const std::string& key, const std::string& def) {
  for (int i = 1; i + 1 < argc; ++i)
    if (key == argv[i]) return argv[i + 1];
  return def;
}

constexpr uint32_t kSvidControlState = 1;
constexpr uint32_t kSvidClock        = 2;

void refresh(gem::EquipmentDataModel& m, const gem::ControlStateMachine& sm) {
  m.svids.set_value(kSvidControlState, s2::Item::ascii(gem::control_state_name(sm.state())));
  m.svids.set_value(kSvidClock, s2::Item::ascii(m.clock.current_time_string()));
}

}  // namespace

int main(int argc, char** argv) {
  const auto port = static_cast<uint16_t>(std::stoi(arg(argc, argv, "--port", "5000")));
  const auto equipment_yaml = arg(argc, argv, "--config", "/app/data/equipment.yaml");
  const auto state_yaml     = arg(argc, argv, "--state-table", "/app/data/control_state.yaml");

  auto logfn = [](const std::string& m) { std::cout << "[equip] " << m << std::endl; };

  auto model = std::make_shared<gem::EquipmentDataModel>();
  config::EquipmentDescriptor desc;
  config::ControlStateConfig sm_cfg;
  try {
    desc = config::load_equipment(equipment_yaml, *model);
    sm_cfg = config::load_control_state(state_yaml);
  } catch (const std::exception& e) {
    std::cerr << "[equip] config error: " << e.what() << std::endl;
    return 1;
  }
  logfn("loaded " + std::to_string(model->svids.size()) + " SVIDs, " +
        std::to_string(model->ecids.all().size()) + " ECIDs, " +
        std::to_string(model->events.all_events().size()) + " CEIDs, " +
        std::to_string(model->alarms.all().size()) + " alarms");

  auto sm = std::make_shared<gem::ControlStateMachine>(sm_cfg.table, sm_cfg.initial);

  asio::io_context io;
  Server::Config server_cfg{port, desc.device_id, {}};
  Server server(io, server_cfg);
  server.on_log(logfn);

  auto active_conn = std::make_shared<std::weak_ptr<Connection>>();

  // Deliver a primary message to the host, or spool it if we can't (either
  // there is no SELECTED session or the spool's force-flag is on).  Returns
  // true if the message was delivered or queued; false if dropped because
  // the stream isn't spoolable and there is no live session.
  auto deliver_or_spool = [active_conn, model, logfn](s2::Message msg,
                                                      const std::string& what) -> bool {
    auto conn = active_conn->lock();
    const bool spooling = model->spool.force_spool() || !conn;
    if (spooling) {
      auto r = model->spool.enqueue(msg);
      if (r == gem::SpoolStore::EnqueueResult::Queued) {
        logfn("spool: " + what + " queued (depth=" +
              std::to_string(model->spool.size()) + ")");
        return true;
      }
      logfn("spool: " + what + " dropped (stream not spoolable, no host)");
      return false;
    }
    conn->send_request(std::move(msg), [](std::error_code, const s2::Message&) {});
    return true;
  };

  auto emit_event = [&io, model, logfn, deliver_or_spool](uint32_t ceid) {
    asio::post(io, [model, logfn, deliver_or_spool, ceid]() {
      if (!model->is_event_enabled(ceid)) {
        logfn("CEID " + std::to_string(ceid) + " not enabled; suppressed");
        return;
      }
      auto reports = model->compose_reports_for(ceid);
      logfn("emit S6F11 CEID=" + std::to_string(ceid) + " (" +
            std::to_string(reports.size()) + " reports)");
      deliver_or_spool(gem::s6f11_event_report(0, ceid, reports),
                       "S6F11 CEID=" + std::to_string(ceid));
    });
  };

  auto emit_alarm_set = [&io, model, logfn, emit_event, deliver_or_spool](uint32_t alid) {
    asio::post(io, [model, logfn, emit_event, deliver_or_spool, 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)) {
        logfn("emit S5F1 alarm set ALID=" + std::to_string(alid));
        deliver_or_spool(gem::s5f1_alarm_report(*alcd, alid, alarm->text),
                         "S5F1 ALID=" + std::to_string(alid));
      } else {
        logfn("alarm " + std::to_string(alid) + " not enabled; suppressed");
      }
      // E30: an AlarmSetEvent CEID also fires (if linked + enabled).
    });
  };

  sm->set_state_change_handler(
      [logfn, emit_event, desc](gem::ControlState from, gem::ControlState to, gem::ControlEvent ev) {
        logfn(std::string("control: ") + gem::control_state_name(from) + " -> " +
              gem::control_state_name(to) + "  (" + gem::control_event_name(ev) + ")");
        if (desc.emit_on_control_change) emit_event(*desc.emit_on_control_change);
      });

  // ---- Build the SECS dispatch table once -------------------------------
  gem::Router router;

  router.on(1, 1, [desc, logfn](const s2::Message&) {
    logfn("S1F1 -> S1F2");
    return gem::s1f2_on_line_data(desc.model_name, desc.software_rev);
  });
  router.on(1, 3, [model, sm, logfn](const s2::Message& msg) -> std::optional<s2::Message> {
    refresh(*model, *sm);
    auto svids = gem::parse_s1f3(msg);
    if (!svids) return s2::Message(1, 0, false);
    std::vector<std::optional<s2::Item>> values;
    if (svids->empty()) {
      for (const auto& sv : model->svids.all()) values.push_back(sv.value);
    } else {
      for (auto id : *svids) {
        auto sv = model->svids.get(id);
        values.push_back(sv ? std::optional<s2::Item>(sv->value) : std::nullopt);
      }
    }
    logfn("S1F3 -> S1F4 (" + std::to_string(values.size()) + " values)");
    return gem::s1f4_selected_status_data(values);
  });
  router.on(1, 11, [model, logfn](const s2::Message&) {
    std::vector<gem::StatusName> rows;
    for (const auto& sv : model->svids.all())
      rows.push_back({sv.id, sv.name, sv.units});
    logfn("S1F11 -> S1F12 (namelist, " + std::to_string(rows.size()) + ")");
    return gem::s1f12_status_namelist_data(rows);
  });
  router.on(1, 13, [desc, logfn](const s2::Message&) {
    logfn("S1F13 -> S1F14");
    return gem::s1f14_establish_comms_ack(gem::CommAck::Accept,
                                          {desc.model_name, desc.software_rev});
  });
  router.on(1, 15, [sm, logfn](const s2::Message&) {
    auto ack = sm->on_host_request_offline();
    logfn("S1F15 -> S1F16 OFLACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s1f16_offline_ack(ack);
  });
  router.on(1, 17, [sm, logfn](const s2::Message&) {
    auto ack = sm->on_host_request_online();
    logfn("S1F17 -> S1F18 ONLACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s1f18_online_ack(ack);
  });
  router.on(1, 19, [desc, logfn](const s2::Message&) {
    std::vector<gem::CapabilityEntry> caps;
    for (const auto& c : desc.capabilities) caps.push_back({c.first, c.second});
    logfn("S1F19 -> S1F20 (" + std::to_string(caps.size()) + " capabilities)");
    return gem::s1f20_get_gem_compliance_data(desc.software_rev, desc.equipment_type, caps);
  });
  router.on(1, 21, [model, logfn](const s2::Message&) {
    std::vector<gem::StatusName> rows;
    for (const auto& dv : model->dvids.all())
      rows.push_back({dv.id, dv.name, dv.units});
    logfn("S1F21 -> S1F22 (" + std::to_string(rows.size()) + " DVIDs)");
    return gem::s1f22_data_variable_namelist_data(rows);
  });

  router.on(2, 13, [model, logfn](const s2::Message& msg) -> std::optional<s2::Message> {
    auto ids = gem::parse_u4_list_body(msg);
    if (!ids) return s2::Message(2, 0, false);
    std::vector<s2::Item> values;
    for (auto id : *ids) {
      auto ec = model->ecids.get(id);
      values.push_back(ec ? ec->value : s2::Item::list({}));
    }
    logfn("S2F13 -> S2F14 (" + std::to_string(values.size()) + " values)");
    return gem::s2f14_ec_data(values);
  });
  router.on(2, 15, [model, logfn](const s2::Message& msg) {
    auto sets = gem::parse_s2f15(msg);
    auto eac = gem::EquipmentAck::Accept;
    if (!sets) eac = gem::EquipmentAck::Denied_OutOfRange;
    else
      for (const auto& s : *sets) {
        auto r = model->ecids.set_value(s.ecid, s.value);
        if (r != gem::EquipmentAck::Accept) eac = r;
      }
    logfn("S2F15 -> S2F16 EAC=" + std::to_string(static_cast<int>(eac)));
    return gem::s2f16_ec_ack(eac);
  });
  router.on(2, 17, [model, logfn](const s2::Message&) {
    logfn("S2F17 -> S2F18 (clock)");
    return gem::s2f18_date_time_data(model->clock.current_time_string());
  });
  router.on(2, 29, [model, logfn](const s2::Message& msg) {
    auto ids = gem::parse_u4_list_body(msg);
    std::vector<gem::EquipmentConstant> ecs;
    if (ids && ids->empty()) ecs = model->ecids.all();
    else if (ids)
      for (auto id : *ids) {
        auto ec = model->ecids.get(id);
        if (ec) ecs.push_back(*ec);
      }
    std::vector<gem::EcNameRow> rows;
    for (const auto& ec : ecs)
      rows.push_back({ec.id, ec.name, ec.min_str, ec.max_str, "", ec.units});
    logfn("S2F29 -> S2F30 (" + std::to_string(rows.size()) + " ECs)");
    return gem::s2f30_ec_namelist_data(rows);
  });
  router.on(2, 31, [model, logfn](const s2::Message& msg) {
    auto t = gem::parse_s2f31(msg);
    auto ack = t ? model->clock.set_time_string(*t) : gem::TimeAck::Error;
    logfn("S2F31 -> S2F32 TIACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f32_date_time_ack(ack);
  });
  router.on(2, 33, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s2f33(msg);
    auto ack = gem::DefineReportAck::InvalidFormat;
    if (req) {
      std::vector<std::pair<uint32_t, std::vector<uint32_t>>> rows;
      rows.reserve(req->reports.size());
      for (const auto& r : req->reports) rows.emplace_back(r.rptid, r.vids);
      ack = model->define_reports(rows);
    }
    logfn("S2F33 -> S2F34 DRACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f34_define_report_ack(ack);
  });
  router.on(2, 35, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s2f35(msg);
    auto ack = gem::LinkEventAck::InvalidFormat;
    if (req) {
      std::vector<std::pair<uint32_t, std::vector<uint32_t>>> rows;
      rows.reserve(req->links.size());
      for (const auto& l : req->links) rows.emplace_back(l.ceid, l.rptids);
      ack = model->link_event_reports(rows);
    }
    logfn("S2F35 -> S2F36 LRACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f36_link_event_report_ack(ack);
  });
  router.on(2, 37, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s2f37(msg);
    auto ack = req ? model->enable_events(req->enable, req->ceids)
                   : gem::EnableEventAck::UnknownCeid;
    logfn(std::string("S2F37 ") + (req && req->enable ? "enable" : "disable") +
          " -> S2F38 ERACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f38_enable_event_ack(ack);
  });
  router.on(2, 41, [model, logfn, emit_event, emit_alarm_set](const s2::Message& msg) {
    auto cmd = gem::parse_s2f41(msg);
    if (!cmd) return gem::s2f42_host_command_ack(gem::HostCmdAck::ParameterInvalid, {});
    auto result = model->commands.dispatch(cmd->rcmd, cmd->params);
    logfn("S2F41 RCMD=" + cmd->rcmd + " -> S2F42 HCACK=" +
          std::to_string(static_cast<int>(result.ack)));
    if (result.ack == gem::HostCmdAck::Accept) {
      if (result.emit_ceid) emit_event(*result.emit_ceid);
      if (result.set_alarm) emit_alarm_set(*result.set_alarm);
      if (result.force_spool) {
        model->spool.set_force_spool(*result.force_spool);
        logfn(std::string("spool: force_spool=") + (*result.force_spool ? "true" : "false") +
              " (depth=" + std::to_string(model->spool.size()) + ")");
      }
    }
    return gem::s2f42_host_command_ack(result.ack, {});
  });

  router.on(2, 43, [model, logfn](const s2::Message& msg) {
    auto streams = gem::parse_s2f43(msg);
    auto ack = gem::ResetSpoolAck::Accept;
    std::vector<gem::ResetSpoolStreamAck> per;
    if (!streams) {
      ack = gem::ResetSpoolAck::Denied_NotAllowed;
    } else {
      model->spool.set_spoolable_streams(*streams);
      logfn("S2F43 spoolable=" + std::to_string(streams->size()) + " streams");
    }
    return gem::s2f44_reset_spooling_ack(ack, per);
  });

  router.on(6, 23, [&io, active_conn, model, logfn](const s2::Message& msg) {
    auto rsdc = gem::parse_s6f23(msg);
    if (!rsdc) return gem::s6f24_request_spool_data_ack(gem::SpoolRequestAck::Denied);
    if (*rsdc == gem::SpoolRequestCode::Purge) {
      const auto n = model->spool.size();
      model->spool.clear();
      logfn("S6F23 purge: dropped " + std::to_string(n) + " messages");
      return gem::s6f24_request_spool_data_ack(gem::SpoolRequestAck::Accept);
    }
    // Transmit: drain the queue, fire each as a fresh primary.  Defer to
    // the executor so the S6F24 ack flushes before the drained primaries
    // go out — the host should see ACK first, then the spooled traffic.
    auto drained = model->spool.drain();
    logfn("S6F23 transmit: draining " + std::to_string(drained.size()) +
          " messages");
    asio::post(io, [active_conn, drained = std::move(drained), logfn]() mutable {
      auto conn = active_conn->lock();
      if (!conn) return;
      for (auto& m : drained) {
        conn->send_request(std::move(m), [](std::error_code, const s2::Message&) {});
      }
    });
    return gem::s6f24_request_spool_data_ack(gem::SpoolRequestAck::Accept);
  });

  router.on(5, 3, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s5f3(msg);
    auto ack = req ? model->alarms.set_enabled(req->alid, (req->aled & 0x80) != 0)
                   : gem::AlarmAck::Error;
    logfn(std::string("S5F3 -> S5F4 ACKC5=") + std::to_string(static_cast<int>(ack)));
    return gem::s5f4_enable_alarm_ack(ack);
  });
  router.on(5, 7, [model, logfn](const s2::Message&) {
    std::vector<gem::AlarmListing> rows;
    for (const auto& a : model->alarms.all()) {
      if (!model->alarms.enabled(a.id)) continue;
      const uint8_t alcd = (a.severity_category & 0x7F) |
                           static_cast<uint8_t>(model->alarms.active(a.id) ? 0x80 : 0x00);
      rows.push_back({alcd, a.id, a.text});
    }
    logfn("S5F7 -> S5F8 (" + std::to_string(rows.size()) + " enabled)");
    return gem::s5f8_list_enabled_alarms_data(rows);
  });

  router.on(5, 5, [model, logfn](const s2::Message& msg) {
    auto ids = gem::parse_u4_list_body(msg);
    std::vector<gem::Alarm> alarms;
    if (ids && ids->empty()) alarms = model->alarms.all();
    else if (ids)
      for (auto id : *ids) {
        auto a = model->alarms.get(id);
        if (a) alarms.push_back(*a);
      }
    logfn("S5F5 -> S5F6 (" + std::to_string(alarms.size()) + " alarms)");
    return gem::s5f6_list_alarms_data(
        alarms, [model](uint32_t id) { return model->alarms.active(id); });
  });

  router.on(7, 3, [model, logfn](const s2::Message& msg) {
    auto pp = gem::parse_s7f3(msg);
    if (!pp) return gem::s7f4_process_program_ack(gem::ProcessProgramAck::LengthError);
    model->recipes.add(pp->ppid, pp->ppbody);
    logfn("S7F3 PPID=" + pp->ppid + " -> S7F4 (Accept)");
    return gem::s7f4_process_program_ack(gem::ProcessProgramAck::Accept);
  });
  router.on(7, 5, [model, logfn](const s2::Message& msg) {
    auto ppid = gem::parse_s7f5(msg);
    if (!ppid) return gem::s7f6_process_program_data("", "");
    auto body = model->recipes.get(*ppid);
    logfn("S7F5 PPID=" + *ppid + " -> S7F6");
    return gem::s7f6_process_program_data(*ppid, body ? *body : "");
  });
  router.on(7, 19, [model, logfn](const s2::Message&) {
    auto list = model->recipes.list();
    logfn("S7F19 -> S7F20 (" + std::to_string(list.size()) + " PPIDs)");
    return gem::s7f20_current_eppd_data(list);
  });

  router.on(10, 1, [logfn](const s2::Message& msg) {
    auto td = gem::parse_s10f1(msg);
    if (td) logfn("TERMINAL[" + std::to_string(td->tid) + "] " + td->text);
    return gem::s10f2_terminal_display_ack(gem::TerminalAck::Accepted);
  });

  logfn("registered " + std::to_string(router.size()) + " (stream,function) handlers");

  // ---- Wire the router into accepted connections -----------------------
  server.on_connection([sm, model, logfn, active_conn, &router, desc](
                           std::shared_ptr<Connection> conn) {
    *active_conn = conn;
    conn->set_closed_handler([active_conn](const std::string&) { active_conn->reset(); });

    conn->set_selected_handler([logfn, sm]() {
      logfn(std::string("host is online; control=") + gem::control_state_name(sm->state()));
    });

    // Wrap router.dispatch so we can emit S9F3 / S9F5 when an inbound
    // primary has no registered handler.  We use Connection's
    // current_header() accessor to capture the offending MHEAD; without
    // it we'd be left fabricating a synthetic header.
    std::weak_ptr<Connection> wconn = conn;
    conn->set_message_handler([&router, wconn, logfn](const s2::Message& msg)
                                  -> std::optional<s2::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;
          logfn("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);
    });
  });

  server.start();
  io.run();
  return 0;
}