// ACME-PVD-3000 — worked vendor application.
//
// This is what a real tool integrator's main() looks like.  Everything
// SECS-shaped (the data dictionary, the message catalogue, the state
// tables) comes from YAML; this file is the *vendor side* — the bits
// that connect the library to the actual tool:
//
//   §1.  Helpers and constants
//   §2.  Sensor simulator (4 sensors at 3 different cadences)
//   §3.  Recipe runner (drives PJ FSM through Processing → Complete)
//   §4.  Alarm threshold monitor (pressure-based)
//   §5.  EPT cycling (Standby ↔ Productive ↔ UnscheduledDowntime)
//   §6.  Router handlers (the minimum set; mostly copied from
//        apps/secs_server.cpp which has the full catalogue)
//   §7.  main() — wires everything together, including the
//        Prometheus metrics exporter on :9090 (§7.3)
//
// This file is deliberately self-contained — no pvd_internal/ helper
// headers, no factored-out "framework."  Customers should be able to
// fork this single file as a starting template.
//
// What's *not* here that production deployment needs:
//   - SECURITY.md: nftables, stunnel, minisign signing.
//   - The other router handlers (S5/S6/S7/S10/S14/S16) — apps/
//     secs_server.cpp has them all.  We register the ones a demo
//     host actually exercises here.

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

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

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

// =============================================================================
// §1.  Helpers and constants
// =============================================================================

namespace pvd {

constexpr uint32_t kSvidControlState   = 1;
constexpr uint32_t kSvidClock          = 2;
constexpr uint32_t kSvidChamberPressure = 10;
constexpr uint32_t kSvidChamberTemp     = 11;
constexpr uint32_t kSvidVacuumPumpRpm   = 13;
constexpr uint32_t kSvidSourcePower     = 20;
constexpr uint32_t kSvidTargetVoltage   = 21;
constexpr uint32_t kSvidSourceMaterial  = 23;
constexpr uint32_t kSvidArgonFlow       = 30;
constexpr uint32_t kSvidNitrogenFlow    = 31;
constexpr uint32_t kSvidCoolingWater    = 32;
constexpr uint32_t kSvidWaferTotal      = 50;
constexpr uint32_t kSvidWafersSinceClean = 51;
constexpr uint32_t kSvidActivePpid      = 53;
constexpr uint32_t kSvidStepName        = 54;
constexpr uint32_t kSvidStepElapsed     = 55;
constexpr uint32_t kSvidEptName         = 70;
constexpr uint32_t kSvidProductiveHours = 71;

constexpr uint32_t kEcChamberPressureSp = 100;
constexpr uint32_t kEcSourcePowerSp     = 102;
constexpr uint32_t kEcCleaningInterval  = 105;

constexpr uint32_t kCeidProcessStarted   = 300;
constexpr uint32_t kCeidProcessCompleted = 301;
constexpr uint32_t kCeidStepStarted      = 310;
constexpr uint32_t kCeidStepCompleted    = 311;
constexpr uint32_t kCeidCJExecuting      = 400;
constexpr uint32_t kCeidCJCompleted      = 401;

constexpr uint32_t kAlarmPressureHigh    = 1;
constexpr uint32_t kAlarmTempOutOfRange  = 3;
constexpr uint32_t kAlarmSourcePowerLoss = 4;
constexpr uint32_t kAlarmCoolingWaterLoss = 5;
constexpr uint32_t kAlarmTargetEol       = 8;
constexpr uint32_t kAlarmCleaningNeeded  = 10;

// =============================================================================
// §2.  Sensor simulator
// =============================================================================
//
// Real tool sensors come from PLCs / serial buses / DAQ cards.  Here
// we simulate them: a random walk around each sensor's setpoint,
// updated on its natural cadence:
//
//   - Chamber pressure: 10 Hz (fast — a real PLC ticks this rate)
//   - Chamber temperature, gas flows, water flow: 1 Hz
//   - Wafer counters, EPT hours: on-event, not polled
//
// All updates marshal onto the io_context via asio::post — that's
// the thread-safety contract documented in INTEGRATION.md §3.

struct Simulator {
  asio::io_context& io;
  std::shared_ptr<gem::EquipmentDataModel> model;
  std::mt19937 rng{std::random_device{}()};

  // Recipe state — when a recipe is running, sensors track the
  // step's target rather than the default setpoint.
  std::atomic<bool> processing{false};
  std::atomic<float> target_pressure_torr{1.0e-7f};
  std::atomic<float> target_temp_c{25.0f};
  std::atomic<float> source_power_setpoint{0.0f};
  std::atomic<float> argon_flow_setpoint{0.0f};

  std::shared_ptr<asio::steady_timer> fast_timer;
  std::shared_ptr<asio::steady_timer> slow_timer;

  Simulator(asio::io_context& io_,
             std::shared_ptr<gem::EquipmentDataModel> model_)
      : io(io_), model(std::move(model_)),
        fast_timer(std::make_shared<asio::steady_timer>(io)),
        slow_timer(std::make_shared<asio::steady_timer>(io)) {}

  // Random walk around `target` with normal-ish jitter.
  float jitter(float current, float target, float spread) {
    std::normal_distribution<float> n(0.0f, spread);
    const float drift = (target - current) * 0.05f;  // 5% of error toward target
    return current + drift + n(rng);
  }

  void tick_fast() {
    // Read current values, update toward target, write back.
    auto p_sv = model->svids.get(kSvidChamberPressure);
    if (p_sv) {
      const auto& cur = std::get<std::vector<float>>(p_sv->value.storage());
      const float new_p = jitter(cur.empty() ? 1.0e-7f : cur[0],
                                  target_pressure_torr.load(), 1.0e-8f);
      model->svids.set_value(kSvidChamberPressure, s2::Item::f4(new_p));
    }
    fast_timer->expires_after(100ms);
    fast_timer->async_wait([this](std::error_code ec) {
      if (!ec) tick_fast();
    });
  }

  void tick_slow() {
    auto upd_f4 = [&](uint32_t vid, float target, float spread, float fallback) {
      auto cur = model->svids.get(vid);
      const float c = (cur && std::holds_alternative<std::vector<float>>(cur->value.storage()))
                          ? std::get<std::vector<float>>(cur->value.storage())[0]
                          : fallback;
      model->svids.set_value(vid, s2::Item::f4(jitter(c, target, spread)));
    };
    upd_f4(kSvidChamberTemp,   target_temp_c.load(),         0.3f, 25.0f);
    upd_f4(kSvidSourcePower,   source_power_setpoint.load(), 5.0f, 0.0f);
    upd_f4(kSvidArgonFlow,     argon_flow_setpoint.load(),   0.5f, 0.0f);
    upd_f4(40 /*ChuckTemp*/,   target_temp_c.load() - 1.0f,  0.2f, 24.0f);
    upd_f4(41 /*ShieldTemp*/,  target_temp_c.load() - 2.0f,  0.2f, 24.0f);
    upd_f4(kSvidCoolingWater,  12.5f,                        0.1f, 12.5f);

    // Vacuum pump RPM — discrete state: ~0 when chamber pressure
    // > 1 Torr, ramping to ~80 000 when at vacuum.
    auto p_sv = model->svids.get(kSvidChamberPressure);
    if (p_sv) {
      const auto& v = std::get<std::vector<float>>(p_sv->value.storage());
      const float p = v.empty() ? 1.0f : v[0];
      const uint32_t rpm = p < 1.0e-3f ? 80000u : (p < 1.0f ? 40000u : 0u);
      model->svids.set_value(kSvidVacuumPumpRpm, s2::Item::u4(rpm));
    }

    // Update clock SVID for hosts that poll it via S1F3.
    model->svids.set_value(kSvidClock, s2::Item::ascii(
        model->clock.current_time_string()));

    slow_timer->expires_after(1s);
    slow_timer->async_wait([this](std::error_code ec) {
      if (!ec) tick_slow();
    });
  }

  void start() { tick_fast(); tick_slow(); }
};

// =============================================================================
// §3.  Recipe runner
// =============================================================================
//
// Drives a PJ through SettingUp → WaitingForStart → Processing →
// ProcessComplete by stepping through the recipe body line by line.
// Each STEP runs for its declared duration, with the sensor
// simulator's targets adjusted to match the step's parameters.

struct RecipeStep {
  std::string name;
  std::chrono::seconds duration{0};
  std::optional<float> target_pressure_torr;
  std::optional<float> target_temp_c;
  std::optional<float> source_power_w;
  std::optional<float> gas_flow_sccm;
  std::string source_material;
};

// Parse "STEP NAME duration=120s power=2500W gas=Argon flow=50sccm".
RecipeStep parse_step(const std::string& line) {
  RecipeStep s;
  std::istringstream is(line);
  std::string tok;
  is >> tok;  // STEP
  is >> s.name;
  while (is >> tok) {
    auto eq = tok.find('=');
    if (eq == std::string::npos) continue;
    const auto key = tok.substr(0, eq);
    auto val = tok.substr(eq + 1);
    auto strip_unit = [&](const char* unit) {
      const auto p = val.find(unit);
      if (p != std::string::npos) val = val.substr(0, p);
    };
    if (key == "duration") { strip_unit("s"); s.duration = std::chrono::seconds(std::stoi(val)); }
    else if (key == "target") {
      strip_unit("Torr"); strip_unit("C");
      try { s.target_pressure_torr = std::stof(val); } catch (...) {}
      try { if (!s.target_pressure_torr) s.target_temp_c = std::stof(val); } catch (...) {}
    }
    else if (key == "power")   { strip_unit("W");    s.source_power_w  = std::stof(val); }
    else if (key == "flow")    { strip_unit("sccm"); s.gas_flow_sccm   = std::stof(val); }
    else if (key == "gas")     { s.source_material = val; }
  }
  return s;
}

struct RecipeRunner {
  asio::io_context& io;
  std::shared_ptr<gem::EquipmentDataModel> model;
  Simulator& sim;
  std::function<void(uint32_t)> emit_event;

  RecipeRunner(asio::io_context& io_,
                std::shared_ptr<gem::EquipmentDataModel> model_,
                Simulator& sim_,
                std::function<void(uint32_t)> emit_)
      : io(io_), model(std::move(model_)), sim(sim_),
        emit_event(std::move(emit_)) {}

  void start(const std::string& prjobid) {
    auto* pj = model->process_jobs.get(prjobid);
    if (!pj) return;
    const auto body = model->recipes.get(pj->ppid);
    if (!body) return;

    // Parse steps from the recipe body.
    auto steps = std::make_shared<std::vector<RecipeStep>>();
    std::istringstream is(*body);
    std::string line;
    while (std::getline(is, line)) {
      if (line.rfind("STEP", 0) == 0) steps->push_back(parse_step(line));
    }
    model->svids.set_value(kSvidActivePpid, s2::Item::ascii(pj->ppid));
    sim.processing = true;

    run_step(prjobid, steps, 0);
  }

  void run_step(const std::string& prjobid,
                 std::shared_ptr<std::vector<RecipeStep>> steps,
                 std::size_t i) {
    if (i >= steps->size()) {
      // Recipe complete.
      model->process_jobs.fire_internal(prjobid, gem::ProcessJobEvent::ProcessComplete);
      model->svids.set_value(kSvidStepName, s2::Item::ascii(""));
      model->svids.set_value(kSvidStepElapsed, s2::Item::u4(0));
      sim.processing = false;
      sim.target_pressure_torr = 1.0e-7f;
      sim.target_temp_c = 25.0f;
      sim.source_power_setpoint = 0.0f;
      sim.argon_flow_setpoint = 0.0f;
      emit_event(kCeidProcessCompleted);
      return;
    }

    const auto& step = (*steps)[i];
    std::cout << "[recipe] step " << (i + 1) << "/" << steps->size()
              << ": " << step.name << " (" << step.duration.count() << "s)\n";

    model->svids.set_value(kSvidStepName, s2::Item::ascii(step.name));
    model->svids.set_value(kSvidStepElapsed, s2::Item::u4(0));
    if (step.target_pressure_torr) sim.target_pressure_torr = *step.target_pressure_torr;
    if (step.target_temp_c)        sim.target_temp_c        = *step.target_temp_c;
    if (step.source_power_w)       sim.source_power_setpoint = *step.source_power_w;
    if (step.gas_flow_sccm)        sim.argon_flow_setpoint   = *step.gas_flow_sccm;
    if (!step.source_material.empty())
      model->svids.set_value(kSvidSourceMaterial, s2::Item::ascii(step.source_material));

    emit_event(kCeidStepStarted);

    // Run the step for `duration` seconds (compressed to milliseconds
    // for demo runs — a real tool would actually wait the duration).
    // We use 100ms per declared second so a 120s step takes 12s.
    auto step_timer = std::make_shared<asio::steady_timer>(io);
    auto tick = std::make_shared<std::function<void(std::error_code)>>();
    auto elapsed = std::make_shared<std::atomic<uint32_t>>(0);
    *tick = [this, prjobid, steps, i, step_timer, tick, elapsed,
             duration_s = step.duration.count()](std::error_code ec) {
      if (ec) return;
      const uint32_t e = elapsed->fetch_add(1) + 1;
      model->svids.set_value(kSvidStepElapsed, s2::Item::u4(e));
      if (e >= static_cast<uint32_t>(duration_s)) {
        emit_event(kCeidStepCompleted);
        run_step(prjobid, steps, i + 1);
        return;
      }
      step_timer->expires_after(100ms);  // 1 demo-second
      step_timer->async_wait(*tick);
    };
    step_timer->expires_after(100ms);
    step_timer->async_wait(*tick);
  }
};

// =============================================================================
// §4.  Alarm threshold monitor
// =============================================================================

struct AlarmMonitor {
  asio::io_context& io;
  std::shared_ptr<gem::EquipmentDataModel> model;
  std::function<void(uint32_t)> emit_alarm_set;
  std::function<void(uint32_t)> emit_alarm_clear;
  std::shared_ptr<asio::steady_timer> timer;

  AlarmMonitor(asio::io_context& io_,
                std::shared_ptr<gem::EquipmentDataModel> model_,
                std::function<void(uint32_t)> set_,
                std::function<void(uint32_t)> clear_)
      : io(io_), model(std::move(model_)),
        emit_alarm_set(std::move(set_)), emit_alarm_clear(std::move(clear_)),
        timer(std::make_shared<asio::steady_timer>(io)) {}

  void start() { tick(); }

 private:
  void tick() {
    // Chamber pressure alarm: trips if > 2x setpoint.
    auto p_sv = model->svids.get(kSvidChamberPressure);
    auto sp = model->ecids.get(kEcChamberPressureSp);
    if (p_sv && sp) {
      const float p = std::get<std::vector<float>>(p_sv->value.storage())[0];
      const float setpoint = std::get<std::vector<float>>(sp->value.storage())[0];
      const bool over = p > setpoint * 100.0f;  // 2 orders of magnitude
      const bool was = model->alarms.active(kAlarmPressureHigh);
      if (over && !was) emit_alarm_set(kAlarmPressureHigh);
      else if (!over && was) emit_alarm_clear(kAlarmPressureHigh);
    }

    // Cleaning interval: alarm 10 fires when wafers-since-cleanup
    // exceeds the configured interval.
    auto wsc = model->svids.get(kSvidWafersSinceClean);
    auto ci  = model->ecids.get(kEcCleaningInterval);
    if (wsc && ci) {
      const uint32_t cur = std::get<std::vector<uint32_t>>(wsc->value.storage())[0];
      const uint32_t limit = std::get<std::vector<uint32_t>>(ci->value.storage())[0];
      const bool over = cur >= limit;
      const bool was = model->alarms.active(kAlarmCleaningNeeded);
      if (over && !was) emit_alarm_set(kAlarmCleaningNeeded);
      else if (!over && was) emit_alarm_clear(kAlarmCleaningNeeded);
    }

    timer->expires_after(500ms);
    timer->async_wait([this](std::error_code ec) { if (!ec) tick(); });
  }
};

// =============================================================================
// §5.  EPT cycling — E116 Equipment Performance Tracking
// =============================================================================
//
// Real tools transition EPT state based on operator actions, alarm
// activity, and process activity.  Here we follow a simple rule:
//   - Standby when nothing's running and no alarms active
//   - Productive when at least one PJ is in Processing
//   - UnscheduledDowntime when a safety alarm is active

struct EptCycler {
  std::shared_ptr<gem::EquipmentDataModel> model;
  asio::io_context& io;
  std::shared_ptr<asio::steady_timer> timer;

  EptCycler(asio::io_context& io_,
             std::shared_ptr<gem::EquipmentDataModel> m)
      : model(std::move(m)), io(io_),
        timer(std::make_shared<asio::steady_timer>(io)) {}

  void start() { tick(); }

 private:
  void tick() {
    bool any_processing = false;
    for (const auto& id : model->process_jobs.ids()) {
      auto* pj = model->process_jobs.get(id);
      if (pj && pj->fsm->state() == gem::ProcessJobState::Processing) {
        any_processing = true;
        break;
      }
    }
    const bool safety_alarm = model->alarms.active(kAlarmSourcePowerLoss) ||
                              model->alarms.active(kAlarmCoolingWaterLoss);
    const auto cur = model->ept.state();
    gem::EptState target = gem::EptState::Standby;
    if (safety_alarm) target = gem::EptState::UnscheduledDowntime;
    else if (any_processing) target = gem::EptState::Productive;

    if (cur != target) {
      auto ev = gem::EptEvent::EnterStandby;
      if (target == gem::EptState::Productive)            ev = gem::EptEvent::EnterProductive;
      if (target == gem::EptState::UnscheduledDowntime)   ev = gem::EptEvent::EnterUnscheduledDown;
      model->ept.on_event(ev);
    }
    model->svids.set_value(kSvidEptName,
        s2::Item::ascii(gem::ept_state_name(target)));
    timer->expires_after(2s);
    timer->async_wait([this](std::error_code ec) { if (!ec) tick(); });
  }
};

}  // namespace pvd

// =============================================================================
// §6.  Router handler registration
// =============================================================================
//
// This is the smallest set of handlers a host needs to talk to the
// tool and run a recipe.  apps/secs_server.cpp has the full
// catalogue (~30 more handlers) for terminal services, slot maps,
// E40/E94 jobs, etc.; in production you'd copy that here too.

void register_handlers(gem::Router& router,
                       std::shared_ptr<gem::EquipmentDataModel> model,
                       std::shared_ptr<gem::ControlStateMachine> sm,
                       const config::EquipmentDescriptor& desc,
                       std::function<void(uint32_t)> emit_event,
                       std::function<void(uint32_t)> emit_alarm_set,
                       std::shared_ptr<pvd::RecipeRunner> recipe) {

  // S1F1 → S1F2  Are You There
  router.on(1, 1, [desc](const s2::Message&) {
    return gem::s1f2_on_line_data(desc.model_name, desc.software_rev);
  });

  // S1F3 → S1F4  Selected Status Request
  router.on(1, 3, [model](const s2::Message& m) {
    auto svids = gem::parse_s1f3(m);
    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);
      }
    }
    return gem::s1f4_selected_status_data(values);
  });

  // S1F11 → S1F12  Status Variable Namelist Request
  router.on(1, 11, [model](const s2::Message&) {
    std::vector<gem::StatusName> rows;
    for (const auto& sv : model->svids.all())
      rows.push_back({sv.id, sv.name, sv.units});
    return gem::s1f12_status_namelist_data(rows);
  });

  // S1F13 → S1F14  Establish Communications
  router.on(1, 13, [desc](const s2::Message&) {
    return gem::s1f14_establish_comms_ack(gem::CommAck::Accept,
                                          {desc.model_name, desc.software_rev});
  });

  // S1F17 → S1F18  Request Online
  router.on(1, 17, [sm](const s2::Message&) {
    auto ack = sm->on_host_request_online();
    return gem::s1f18_online_ack(ack);
  });

  // S2F13 → S2F14  EC Values
  router.on(2, 13, [model](const s2::Message& m) {
    auto ids = gem::parse_u4_list_body(m);
    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({}));
    }
    return gem::s2f14_ec_data(values);
  });

  // S2F17 → S2F18  Clock
  router.on(2, 17, [model](const s2::Message&) {
    return gem::s2f18_date_time_data(model->clock.current_time_string());
  });

  // S2F41 → S2F42  Host Command
  router.on(2, 41, [model, emit_event, emit_alarm_set, recipe]
                    (const s2::Message& m) {
    auto cmd = gem::parse_s2f41(m);
    if (!cmd) return gem::s2f42_host_command_ack(gem::HostCmdAck::ParameterInvalid, {});
    auto result = model->commands.dispatch(cmd->rcmd, cmd->params);
    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);
      // Demo: RCMD=START with PJ in WaitingForStart triggers the
      // recipe runner.  Real tools would gate on richer state.
      if (cmd->rcmd == "START") {
        for (const auto& pjid : model->process_jobs.ids()) {
          auto* pj = model->process_jobs.get(pjid);
          if (pj && pj->fsm->state() == gem::ProcessJobState::WaitingForStart) {
            model->process_jobs.fire_internal(pjid, gem::ProcessJobEvent::Start);
            recipe->start(pjid);
            break;
          }
        }
      }
    }
    return gem::s2f42_host_command_ack(result.ack, {});
  });

  // S5F5 → S5F6  List Alarms
  router.on(5, 5, [model](const s2::Message& m) {
    auto ids = gem::parse_u4_list_body(m);
    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);
      }
    return gem::s5f6_list_alarms_data(
        alarms, [model](uint32_t id) { return model->alarms.active(id); });
  });

  // S7F5 → S7F6  Process Program Request
  router.on(7, 5, [model](const s2::Message& m) {
    auto ppid = gem::parse_s7f5(m);
    if (!ppid) return gem::s7f6_process_program_data("", "");
    auto body = model->recipes.get(*ppid);
    return gem::s7f6_process_program_data(*ppid, body ? *body : "");
  });

  // S7F19 → S7F20  Current PP List
  router.on(7, 19, [model](const s2::Message&) {
    return gem::s7f20_current_eppd_data(model->recipes.list());
  });

  // -------- Extended handlers (mirrors apps/secs_server.cpp) ----------
  // These follow the demo server's patterns one-for-one.  A real
  // vendor's main.cpp would either include them inline (as we do here)
  // or extract them to a shared helper.

  // S1F15 → S1F16  Request Offline
  router.on(1, 15, [sm](const s2::Message&) {
    return gem::s1f16_offline_ack(sm->on_host_request_offline());
  });
  // S1F19 → S1F20  GEM Compliance
  router.on(1, 19, [desc](const s2::Message&) {
    std::vector<gem::CapabilityEntry> caps;
    for (const auto& c : desc.capabilities) caps.push_back({c.first, c.second});
    return gem::s1f20_get_gem_compliance_data(
        desc.software_rev, desc.equipment_type, caps);
  });
  // S1F21 → S1F22  DVID Namelist
  router.on(1, 21, [model](const s2::Message&) {
    std::vector<gem::StatusName> rows;
    for (const auto& dv : model->dvids.all())
      rows.push_back({dv.id, dv.name, dv.units});
    return gem::s1f22_data_variable_namelist_data(rows);
  });
  // S1F23 → S1F24  CEID Namelist
  router.on(1, 23, [model](const s2::Message& m) {
    auto req = gem::parse_s1f23(m);
    std::vector<gem::CollectionEventName> rows;
    if (req && req->empty()) {
      for (const auto& e : model->events.all_events())
        rows.push_back({e.id, e.name, model->events.vids_for(e.id)});
    } else if (req) {
      for (auto id : *req) {
        auto info = model->events.event_info(id);
        rows.push_back({id, info ? info->name : "", model->events.vids_for(id)});
      }
    }
    return gem::s1f24_collection_event_namelist_data(rows);
  });
  // S2F15 → S2F16  EC Set
  router.on(2, 15, [model](const s2::Message& m) {
    auto sets = gem::parse_s2f15(m);
    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;
    }
    return gem::s2f16_ec_ack(eac);
  });
  // S2F29 → S2F30  EC Namelist
  router.on(2, 29, [model](const s2::Message& m) {
    auto ids = gem::parse_u4_list_body(m);
    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});
    return gem::s2f30_ec_namelist_data(rows);
  });
  // S2F31 → S2F32  Set Clock
  router.on(2, 31, [model](const s2::Message& m) {
    auto t = gem::parse_s2f31(m);
    return gem::s2f32_date_time_ack(
        t ? model->clock.set_time_string(*t) : gem::TimeAck::Error);
  });
  // S2F33/F35/F37  Dynamic event report config
  router.on(2, 33, [model](const s2::Message& m) {
    auto req = gem::parse_s2f33(m);
    auto ack = gem::DefineReportAck::InvalidFormat;
    if (req) {
      std::vector<std::pair<uint32_t, std::vector<uint32_t>>> rows;
      for (const auto& r : req->reports) rows.emplace_back(r.rptid, r.vids);
      ack = model->define_reports(rows);
    }
    return gem::s2f34_define_report_ack(ack);
  });
  router.on(2, 35, [model](const s2::Message& m) {
    auto req = gem::parse_s2f35(m);
    auto ack = gem::LinkEventAck::InvalidFormat;
    if (req) {
      std::vector<std::pair<uint32_t, std::vector<uint32_t>>> rows;
      for (const auto& l : req->links) rows.emplace_back(l.ceid, l.rptids);
      ack = model->link_event_reports(rows);
    }
    return gem::s2f36_link_event_report_ack(ack);
  });
  router.on(2, 37, [model](const s2::Message& m) {
    auto req = gem::parse_s2f37(m);
    auto ack = req ? model->enable_events(req->enable, req->ceids)
                   : gem::EnableEventAck::UnknownCeid;
    return gem::s2f38_enable_event_ack(ack);
  });
  // S2F21  Legacy remote command
  router.on(2, 21, [model, emit_event, emit_alarm_set](const s2::Message& m) {
    auto rcmd = gem::parse_s2f21(m);
    if (!rcmd) return gem::s2f22_remote_command_ack(gem::HostCmdAck::ParameterInvalid);
    auto result = model->commands.dispatch(*rcmd, {});
    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);
    }
    return gem::s2f22_remote_command_ack(result.ack);
  });
  // S2F23 trace, S2F43 spool reset, S2F45 limits, S2F47 limit attrs
  router.on(2, 23, [model](const s2::Message& m) {
    auto req = gem::parse_s2f23(m);
    auto ack = gem::TraceAck::Accept;
    if (!req) ack = gem::TraceAck::InvalidPeriod;
    else for (auto v : req->svids)
      if (!model->vid_exists(v)) { ack = gem::TraceAck::UnknownVid; break; }
    return gem::s2f24_trace_initialize_ack(ack);
  });
  router.on(2, 43, [model](const s2::Message& m) {
    auto streams = gem::parse_s2f43(m);
    if (streams) model->spool.set_spoolable_streams(*streams);
    return gem::s2f44_reset_spooling_ack(
        streams ? gem::ResetSpoolAck::Accept : gem::ResetSpoolAck::Denied_NotAllowed, {});
  });
  router.on(2, 47, [model](const s2::Message& m) {
    auto vids = gem::parse_s2f47(m);
    std::vector<gem::VidLimitsEntry> rows;
    if (vids) {
      const auto target = vids->empty() ? model->limits.all_vids() : *vids;
      for (auto v : target) rows.push_back({v, model->limits.get_for_vid(v)});
    }
    return gem::s2f48_variable_limit_attribute_data(rows);
  });
  // S2F49  Enhanced remote command
  router.on(2, 49, [model, emit_event](const s2::Message& m) {
    auto cmd = gem::parse_s2f49(m);
    if (!cmd) return gem::s2f50_enhanced_host_command_ack(
        gem::HostCmdAck::ParameterInvalid, {});
    auto result = model->commands.dispatch(cmd->rcmd, cmd->params);
    if (result.ack == gem::HostCmdAck::Accept && result.emit_ceid)
      emit_event(*result.emit_ceid);
    return gem::s2f50_enhanced_host_command_ack(result.ack, {});
  });
  // S5F3 enable alarm, S5F7 list enabled
  router.on(5, 3, [model](const s2::Message& m) {
    auto req = gem::parse_s5f3(m);
    return gem::s5f4_enable_alarm_ack(
        req ? model->alarms.set_enabled(req->alid, (req->aled & 0x80) != 0)
            : gem::AlarmAck::Error);
  });
  router.on(5, 7, [model](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) |
                           (model->alarms.active(a.id) ? 0x80 : 0x00);
      rows.push_back({alcd, a.id, a.text});
    }
    return gem::s5f8_list_enabled_alarms_data(rows);
  });
  // S5F13/F17 exception recover
  router.on(5, 13, [model](const s2::Message& m) {
    auto req = gem::parse_s5f13(m);
    return gem::s5f14_exception_recover_ack(
        req ? model->exceptions.on_recover(req->exid, req->exrecvra)
            : gem::AlarmAck::Error);
  });
  router.on(5, 17, [model](const s2::Message& m) {
    auto exid = gem::parse_s5f17(m);
    return gem::s5f18_exception_recover_abort_ack(
        exid ? model->exceptions.on_recover_abort(*exid) : gem::AlarmAck::Error);
  });
  // S6F15/F19/F21 host-initiated event/report queries
  router.on(6, 15, [model](const s2::Message& m) {
    auto ceid = gem::parse_s6f15(m);
    if (!ceid) return gem::s6f16_event_report_data({0, 0, {}});
    return gem::s6f16_event_report_data({0, *ceid, model->compose_reports_for(*ceid)});
  });
  router.on(6, 19, [model](const s2::Message& m) {
    auto rptid = gem::parse_s6f19(m);
    std::vector<s2::Item> values;
    if (rptid) for (const auto& r : model->events.all_reports()) {
      if (r.id != *rptid) continue;
      for (auto vid : r.vids) {
        auto v = model->vid_value(vid);
        values.push_back(v ? *v : s2::Item::list({}));
      }
      break;
    }
    return gem::s6f20_individual_report_data(values);
  });
  router.on(6, 21, [model](const s2::Message& m) {
    auto rptid = gem::parse_s6f21(m);
    std::vector<gem::AnnotatedValue> rows;
    if (rptid) for (const auto& r : model->events.all_reports()) {
      if (r.id != *rptid) continue;
      for (auto vid : r.vids) {
        auto v = model->vid_value(vid);
        rows.push_back({vid, v ? *v : s2::Item::list({})});
      }
      break;
    }
    return gem::s6f22_annotated_report_data(rows);
  });
  // S6F23 spool data request
  router.on(6, 23, [model](const s2::Message& m) {
    auto rsdc = gem::parse_s6f23(m);
    if (!rsdc) return gem::s6f24_request_spool_data_ack(gem::SpoolRequestAck::Denied);
    if (*rsdc == gem::SpoolRequestCode::Purge) model->spool.clear();
    else model->spool.drain();  // demo: drop the drained messages
    return gem::s6f24_request_spool_data_ack(gem::SpoolRequestAck::Accept);
  });
  // S7F1 PP load inquire, S7F3 PP send, S7F17 PP delete
  router.on(7, 1, [](const s2::Message& m) {
    auto req = gem::parse_s7f1(m);
    auto ack = gem::ProcessProgramAck::Accept;
    if (!req || req->ppid.empty()) ack = gem::ProcessProgramAck::PpidNotFound;
    return gem::s7f2_pp_load_grant(ack);
  });
  router.on(7, 3, [model](const s2::Message& m) {
    auto pp = gem::parse_s7f3(m);
    if (!pp) return gem::s7f4_process_program_ack(gem::ProcessProgramAck::LengthError);
    model->recipes.add(pp->ppid, pp->ppbody);
    return gem::s7f4_process_program_ack(gem::ProcessProgramAck::Accept);
  });
  router.on(7, 17, [model](const s2::Message& m) {
    auto req = gem::parse_s7f17(m);
    if (!req) return gem::s7f18_delete_pp_ack(gem::ProcessProgramAck::LengthError);
    if (req->empty()) for (const auto& id : model->recipes.list()) model->recipes.remove(id);
    else for (const auto& id : *req) model->recipes.remove(id);
    return gem::s7f18_delete_pp_ack(gem::ProcessProgramAck::Accept);
  });
  // S10F3 host→equipment terminal display, S10F5 multi-line
  router.on(10, 3, [](const s2::Message& m) {
    auto td = gem::parse_s10f3(m);
    if (td) std::cout << "[TERMINAL " << static_cast<int>(td->tid)
                      << "] " << td->text << "\n";
    return gem::s10f4_terminal_display_ack(gem::TerminalAck::Accepted);
  });
  router.on(10, 5, [](const s2::Message& m) {
    auto td = gem::parse_s10f5(m);
    if (td) for (const auto& line : td->lines)
      std::cout << "[TERMINAL " << static_cast<int>(td->tid) << "] " << line << "\n";
    return gem::s10f6_terminal_display_multi_ack(gem::TerminalAck::Accepted);
  });
  // S3 — E87 carriers (basic acceptance)
  router.on(3, 17, [model](const s2::Message& m) {
    auto req = gem::parse_s3f17(m);
    if (!req) return gem::s3f18_carrier_action_ack(gem::CarrierActionAck::ParameterInvalid);
    if (!model->carriers.has(req->carrierid))
      return gem::s3f18_carrier_action_ack(gem::CarrierActionAck::CarrierIDUnknown);
    return gem::s3f18_carrier_action_ack(gem::CarrierActionAck::Accept);
  });
  router.on(3, 19, [model](const s2::Message& m) {
    auto req = gem::parse_s3f19(m);
    if (!req) return gem::s3f20_slot_map_verify_ack(gem::SlotMapVerifyAck::Error);
    return gem::s3f20_slot_map_verify_ack(
        model->carriers.has(req->carrierid) ? gem::SlotMapVerifyAck::Accept
                                            : gem::SlotMapVerifyAck::CarrierUnknown);
  });
  router.on(3, 25, [](const s2::Message&) {
    return gem::s3f26_carrier_transfer_ack(gem::CarrierActionAck::Accept);
  });
  router.on(3, 27, [model](const s2::Message& m) {
    auto cid = gem::parse_s3f27(m);
    return gem::s3f28_cancel_carrier_ack(
        cid && model->carriers.has(*cid) ? gem::CarrierActionAck::Accept
                                          : gem::CarrierActionAck::CarrierIDUnknown);
  });
  // S14 — E39 GetAttr + E94 CJ create/delete
  router.on(14, 1, [model](const s2::Message& m) {
    auto req = gem::parse_s14f1(m);
    if (!req) return gem::s14f2_get_attr_data({}, gem::ObjectAck::Error);
    auto* obj = model->cem.get(req->objspec);
    if (!obj) return gem::s14f2_get_attr_data({}, gem::ObjectAck::Denied_UnknownObject);
    std::vector<gem::AttrValue> attrs;
    for (const auto& id : req->attrids) {
      auto v = model->cem.get_attr(req->objspec, id);
      attrs.push_back({id, v.value_or(s2::Item::ascii(""))});
    }
    return gem::s14f2_get_attr_data(attrs, gem::ObjectAck::Success);
  });
  router.on(14, 9, [model](const s2::Message& m) {
    auto req = gem::parse_s14f9(m);
    if (!req) return gem::s14f10_create_control_job_ack("", gem::ObjectAck::Error);
    auto r = model->control_jobs.create(req->ctljobid, req->prjobids,
        [model](const std::string& id) { return model->process_jobs.has(id); });
    auto ack = (r == gem::ControlJobStore::CreateResult::Created)
                   ? gem::ObjectAck::Success
                   : gem::ObjectAck::Denied_UnknownObject;
    return gem::s14f10_create_control_job_ack(req->ctljobid, ack);
  });
  router.on(14, 11, [model](const s2::Message& m) {
    auto id = gem::parse_s14f11(m);
    return gem::s14f12_delete_control_job_ack(
        id && model->control_jobs.remove(*id) ? gem::ObjectAck::Success
                                              : gem::ObjectAck::Denied_UnknownObject);
  });
  // S16 — E40 PJ create/command/dequeue/monitor, E94 CJ command
  router.on(16, 11, [model](const s2::Message& m) {
    auto req = gem::parse_s16f11(m);
    if (!req) return gem::s16f12_pr_job_create_ack(gem::HostCmdAck::ParameterInvalid);
    auto r = model->process_jobs.create(req->prjobid, req->rcpspec.ppid, req->mtrloutspec,
        [model](const std::string& ppid) { return model->recipes.get(ppid).has_value(); });
    auto ack = gem::HostCmdAck::Accept;
    if (r == gem::ProcessJobStore::CreateResult::Denied_AlreadyExists)
      ack = gem::HostCmdAck::Rejected;
    else if (r == gem::ProcessJobStore::CreateResult::Denied_InvalidPpid)
      ack = gem::HostCmdAck::ParameterInvalid;
    return gem::s16f12_pr_job_create_ack(ack);
  });
  router.on(16, 5, [model](const s2::Message& m) {
    auto req = gem::parse_s16f5(m);
    if (!req) return gem::s16f6_pr_job_command_ack(gem::HostCmdAck::ParameterInvalid);
    auto ev = gem::pr_cmd_to_event(req->prcmd);
    if (!ev) return gem::s16f6_pr_job_command_ack(gem::HostCmdAck::InvalidCommand);
    return gem::s16f6_pr_job_command_ack(model->process_jobs.on_host_command(req->prjobid, *ev));
  });
  router.on(16, 7, [model](const s2::Message& m) {
    auto req = gem::parse_s16f7(m);
    if (!req) return gem::s16f8_pr_job_monitor_ack(gem::HostCmdAck::ParameterInvalid);
    for (const auto& e : req->entries)
      model->process_jobs.set_alert(e.prjobid, (e.pralert & 0x80) != 0);
    return gem::s16f8_pr_job_monitor_ack(gem::HostCmdAck::Accept);
  });
  router.on(16, 13, [model](const s2::Message& m) {
    auto id = gem::parse_s16f13(m);
    return gem::s16f14_pr_job_dequeue_ack(
        id ? model->process_jobs.dequeue(*id) : gem::HostCmdAck::ParameterInvalid);
  });
  router.on(16, 27, [model](const s2::Message& m) {
    auto req = gem::parse_s16f27(m);
    if (!req) return gem::s16f28_cj_command_ack(gem::HostCmdAck::ParameterInvalid);
    auto ev = gem::ctl_cmd_to_event(req->ctljobcmd);
    if (!ev) return gem::s16f28_cj_command_ack(gem::HostCmdAck::InvalidCommand);
    return gem::s16f28_cj_command_ack(
        model->control_jobs.on_host_command(req->ctljobid, *ev));
  });
}

// =============================================================================
// §7.  main()
// =============================================================================

int main(int argc, char** argv) {
  const std::string config_path = (argc > 1) ? argv[1] :
      "/app/examples/pvd_tool/equipment.yaml";
  const std::string state_path  = (argc > 2) ? argv[2] :
      "/app/data/control_state.yaml";
  const uint16_t port           = (argc > 3) ?
      static_cast<uint16_t>(std::stoi(argv[3])) : 5000;
  const uint16_t metrics_port   = (argc > 4) ?
      static_cast<uint16_t>(std::stoi(argv[4])) : 9090;

  // ---- §7.1  Validate the YAML configs before binding the port ----------
  {
    config::ConfigValidator v;
    v.validate_equipment(config_path);
    v.validate_control_state(state_path);
    if (v.has_errors()) {
      config::format_issues_to(std::cerr, v.issues());
      std::cerr << v.error_count() << " config error(s); refusing to start\n";
      return 1;
    }
  }

  auto logfn = [](const std::string& m) {
    std::cout << "[pvd] " << m << "\n";
  };

  // ---- §7.2  Build the data model ---------------------------------------
  auto model = std::make_shared<gem::EquipmentDataModel>();
  config::EquipmentDescriptor desc;
  config::ControlStateConfig sm_cfg;
  try {
    desc = config::load_equipment(config_path, *model);
    sm_cfg = config::load_control_state(state_path);
  } catch (const std::exception& e) {
    std::cerr << "[pvd] config load failed: " << e.what() << "\n";
    return 1;
  }
  auto sm = std::make_shared<gem::ControlStateMachine>(
      sm_cfg.table, sm_cfg.initial);

  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, " +
        std::to_string(model->recipes.list().size()) + " recipes");

  asio::io_context io;

  // ---- §7.3  Metrics exporter on a second port --------------------------
  auto registry = std::make_shared<metrics::Registry>();
  registry->describe("pvd_messages_total", "SECS messages dispatched",
                      metrics::MetricType::Counter);
  registry->describe("pvd_chamber_pressure_torr", "Process chamber pressure",
                      metrics::MetricType::Gauge);
  registry->describe("pvd_spool_depth", "Queued spool messages",
                      metrics::MetricType::Gauge);
  auto exporter = std::make_shared<metrics::PrometheusServer>(io, metrics_port, registry);
  exporter->start();
  logfn("metrics exporter on :" + std::to_string(metrics_port) + "/metrics");

  // ---- §7.4  Sensor simulator, EPT cycler, alarm monitor ---------------
  auto sim = std::make_shared<pvd::Simulator>(io, model);
  sim->start();
  auto ept = std::make_shared<pvd::EptCycler>(io, model);
  ept->start();

  // ---- §7.5  Server + handler wiring ------------------------------------
  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>>();

  // Shared event-emission helper.
  auto deliver_or_spool = [active_conn, model, logfn](s2::Message msg) {
    auto conn = active_conn->lock();
    if (!conn) {
      model->spool.enqueue(msg);
      return;
    }
    if (msg.reply_expected) {
      conn->send_request(std::move(msg), [](std::error_code, const s2::Message&) {});
    } else {
      conn->send_data(std::move(msg));
    }
  };
  auto emit_event = [&io, model, deliver_or_spool, registry](uint32_t ceid) {
    asio::post(io, [model, deliver_or_spool, ceid]() {
      if (!model->is_event_enabled(ceid)) return;
      auto reports = model->compose_reports_for(ceid);
      deliver_or_spool(gem::s6f11_event_report(0, ceid, reports));
    });
    registry->inc("pvd_events_total", {{"ceid", std::to_string(ceid)}});
  };
  auto emit_alarm_set = [&io, model, deliver_or_spool, registry](uint32_t alid) {
    asio::post(io, [model, deliver_or_spool, alid]() {
      auto alarm = model->alarms.get(alid);
      auto alcd = model->alarms.set_active(alid);
      if (!alarm || !alcd || !model->alarms.enabled(alid)) return;
      deliver_or_spool(gem::s5f1_alarm_report(*alcd, alid, alarm->text));
    });
    registry->inc("pvd_alarm_set_total", {{"alid", std::to_string(alid)}});
  };
  auto emit_alarm_clear = [&io, model, deliver_or_spool](uint32_t alid) {
    asio::post(io, [model, deliver_or_spool, alid]() {
      auto alarm = model->alarms.get(alid);
      auto alcd = model->alarms.clear_active(alid);
      if (!alarm || !alcd || !model->alarms.enabled(alid)) return;
      deliver_or_spool(gem::s5f1_alarm_report(*alcd, alid, alarm->text));
    });
  };

  auto alarm_mon = std::make_shared<pvd::AlarmMonitor>(
      io, model, emit_alarm_set, emit_alarm_clear);
  alarm_mon->start();

  auto recipe_runner = std::make_shared<pvd::RecipeRunner>(
      io, model, *sim, emit_event);

  // Wire control-state-change → CEID emission.
  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);
      });

  // ---- §7.6  Periodic gauge updates for Prometheus ----------------------
  auto gauge_timer = std::make_shared<asio::steady_timer>(io);
  std::function<void(std::error_code)> gauge_tick =
      [&gauge_tick, gauge_timer, model, registry](std::error_code ec) {
        if (ec) return;
        auto p = model->svids.get(pvd::kSvidChamberPressure);
        if (p) {
          const float v = std::get<std::vector<float>>(p->value.storage())[0];
          registry->set_gauge("pvd_chamber_pressure_torr", v);
        }
        registry->set_gauge("pvd_spool_depth", model->spool.size());
        gauge_timer->expires_after(5s);
        gauge_timer->async_wait(gauge_tick);
      };
  gauge_timer->expires_after(5s);
  gauge_timer->async_wait(gauge_tick);

  // ---- §7.7  Router + per-connection handler wiring --------------------
  gem::Router router;
  register_handlers(router, model, sm, desc,
                     emit_event, emit_alarm_set, recipe_runner);
  logfn("registered " + std::to_string(router.size()) + " SECS-II handlers");

  server.on_connection([&io, sm, model, logfn, active_conn, &router, registry]
                        (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 SELECTED; control=") +
             gem::control_state_name(sm->state()));
    });
    conn->set_message_handler(
        [&router, model, conn, registry](const s2::Message& msg)
            -> std::optional<s2::Message> {
          registry->inc("pvd_messages_total",
                         {{"dir", "rx"},
                          {"stream", std::to_string(msg.stream)},
                          {"function", std::to_string(msg.function)}});
          return router.dispatch_with_s9(
              [&](uint8_t f, const std::array<uint8_t, 10>& mh) {
                conn->emit_s9(f, mh);
              },
              [&]() -> std::optional<std::array<uint8_t, 10>> {
                auto* h = conn->current_header();
                return h ? std::optional{h->encode()} : std::nullopt;
              }, msg);
        });
  });

  server.start();
  logfn("ACME-PVD-3000 ready, listening on :" + std::to_string(port));
  io.run();
  return 0;
}