secs-gem/include/secsgem/gem/store/process_jobs.hpp

#pragma once

#include <algorithm>
#include <cstdint>
#include <cstdio>
#include <filesystem>
#include <fstream>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <system_error>
#include <utility>
#include <vector>

#include "secsgem/gem/handler_slot.hpp"
#include "secsgem/gem/e40_constants.hpp"
#include "secsgem/gem/process_job_state.hpp"
#include "secsgem/secs2/codec.hpp"
#include "secsgem/secs2/item.hpp"

namespace secsgem::gem {

// One Process Job record. The FSM is heap-allocated through unique_ptr so
// the per-PJ state-change handler can capture a stable pointer to
// `this`-style state without invalidation on map rehash.
//
// The MF / recipe-method / rcp-vars / process-params fields are the
// optional E40-0705 §10.2 trailers on S16F11.  Simple callers leave
// them defaulted; tools that actually need recipe-variable tuning or
// per-job process parameters populate them.
struct RcpVar {
  std::string name;
  secs2::Item value;
};

struct ProcessParam {
  std::string name;
  secs2::Item value;
};

struct ProcessJob {
  std::string prjobid;
  std::string ppid;                          // recipe identifier
  std::vector<std::string> mtrloutspec;      // material identifiers
  bool alert_enabled = true;                  // S16F9 alerts on/off
  MaterialFlag mf = MaterialFlag::Substrate;
  ProcessRecipeMethod prrecipemethod = ProcessRecipeMethod::RecipeOnly;
  std::vector<RcpVar> rcpvars;
  std::vector<ProcessParam> prprocessparams;
  std::unique_ptr<ProcessJobStateMachine> fsm;
  std::filesystem::path journal_path;
};

class ProcessJobStore {
 public:
  using TransitionTableFactory =
      std::function<ProcessJobTransitionTable()>;
  using StateChangeHandler =
      std::function<void(const std::string& prjobid,
                         ProcessJobState from, ProcessJobState to,
                         ProcessJobEvent trigger)>;

  ProcessJobStore()
      : factory_([] { return ProcessJobTransitionTable::default_table(); }) {}

  // The per-PJ FSM closes over `this`, so the store must keep a stable
  // address. unique_ptr makes copying impossible anyway; moves would
  // silently dangle the per-PJ lambdas — disallow them explicitly.
  ProcessJobStore(const ProcessJobStore&) = delete;
  ProcessJobStore& operator=(const ProcessJobStore&) = delete;
  ProcessJobStore(ProcessJobStore&&) = delete;
  ProcessJobStore& operator=(ProcessJobStore&&) = delete;

  void set_table_factory(TransitionTableFactory f) { factory_ = std::move(f); }
  // set_ replaces the primary handler (legacy semantics); add_ appends an
  // observer that survives set_ calls (see handler_slot.hpp).
  void set_state_change_handler(StateChangeHandler h) { on_change_ = std::move(h); }
  void add_state_change_handler(StateChangeHandler h) { on_change_.add(std::move(h)); }

  enum class CreateResult {
    Created,
    Denied_AlreadyExists,
    Denied_InvalidPpid,
  };

  // Validate `ppid` against the optional callback; if accepted, create
  // the PJ in Queued and fire the Created -> Queued change handler.
  CreateResult create(std::string prjobid, std::string ppid,
                      std::vector<std::string> materials,
                      const std::function<bool(const std::string&)>& ppid_exists =
                          [](const std::string&) { return true; }) {
    if (jobs_.count(prjobid)) return CreateResult::Denied_AlreadyExists;
    if (!ppid_exists(ppid))   return CreateResult::Denied_InvalidPpid;
    install_(prjobid, std::move(ppid), std::move(materials),
             /*alert=*/true, MaterialFlag::Substrate,
             ProcessRecipeMethod::RecipeOnly, ProcessJobState::Queued);
    order_.push_back(prjobid);
    if (persistent_) {
      write_record_(prjobid);
      write_order_();
    }
    // Synthetic NoState -> Queued so subscribers observe creation.
    if (on_change_) {
      on_change_(jobs_.find(prjobid)->first, ProcessJobState::NoState,
                 ProcessJobState::Queued, ProcessJobEvent::Created);
    }
    return CreateResult::Created;
  }

  // After `create`, populate the optional E40-0705 fields on the new PJ.
  // Returns true if the PJ exists.  These fields don't influence the FSM;
  // they're carried so the tool's recipe engine can read them later.
  bool set_e40_extras(const std::string& prjobid,
                      MaterialFlag mf,
                      ProcessRecipeMethod prrecipemethod,
                      std::vector<RcpVar> rcpvars,
                      std::vector<ProcessParam> params) {
    auto it = jobs_.find(prjobid);
    if (it == jobs_.end()) return false;
    it->second.mf = mf;
    it->second.prrecipemethod = prrecipemethod;
    it->second.rcpvars = std::move(rcpvars);
    it->second.prprocessparams = std::move(params);
    if (persistent_) write_record_(prjobid);
    return true;
  }

  bool has(const std::string& prjobid) const {
    return jobs_.count(prjobid) > 0;
  }

  const ProcessJob* get(const std::string& prjobid) const {
    auto it = jobs_.find(prjobid);
    return it == jobs_.end() ? nullptr : &it->second;
  }
  ProcessJob* get(const std::string& prjobid) {
    auto it = jobs_.find(prjobid);
    return it == jobs_.end() ? nullptr : &it->second;
  }

  ProcessJobState state(const std::string& prjobid) const {
    auto it = jobs_.find(prjobid);
    return it == jobs_.end() ? ProcessJobState::NoState
                              : it->second.fsm->state();
  }

  // Host-initiated S16F5 PRJobCommand. Returns InvalidObject for unknown
  // PJs; CannotDoNow when the current PJ state has no row for the event.
  // For HOQ the FSM gates legality (Queued only) and the store performs
  // the actual reorder so the next CJ Select picks this PJ first.
  HostCmdAck on_host_command(const std::string& prjobid, ProcessJobEvent event) {
    auto* pj = get(prjobid);
    if (!pj) return HostCmdAck::InvalidObject;
    HostCmdAck ack = pj->fsm->on_host_command(event);
    if (ack == HostCmdAck::Accept) {
      if (event == ProcessJobEvent::HeadOfQueue) {
        move_to_head(prjobid);
        if (persistent_) write_order_();
      } else if (persistent_) {
        write_record_(prjobid);
      }
    }
    return ack;
  }

  // Position of `prjobid` in the queue order (insertion order; HOQ rewrites
  // it). Returns -1 if unknown. Exposed for tests and CJ promotion logic.
  int position(const std::string& prjobid) const {
    auto it = std::find(order_.begin(), order_.end(), prjobid);
    return it == order_.end() ? -1 : static_cast<int>(it - order_.begin());
  }

  // Internal events from the application (recipe runner etc.).
  bool fire_internal(const std::string& prjobid, ProcessJobEvent event) {
    auto* pj = get(prjobid);
    if (!pj) return false;
    bool ok = pj->fsm->on_internal(event);
    if (ok && persistent_) write_record_(prjobid);
    return ok;
  }

  // Dequeue S16F13: only legal while QUEUED.
  HostCmdAck dequeue(const std::string& prjobid) {
    auto it = jobs_.find(prjobid);
    if (it == jobs_.end()) return HostCmdAck::InvalidObject;
    if (it->second.fsm->state() != ProcessJobState::Queued)
      return HostCmdAck::CannotDoNow;
    if (persistent_ && !it->second.journal_path.empty()) {
      std::error_code ec;
      std::filesystem::remove(it->second.journal_path, ec);
    }
    erase_from_order(prjobid);
    jobs_.erase(it);
    if (persistent_) write_order_();
    return HostCmdAck::Accept;
  }

  // Remove a terminal job from the store (caller responsibility to ensure
  // ProcessComplete before deleting). Used after CJ Completed cleanup.
  bool remove(const std::string& prjobid) {
    auto it = jobs_.find(prjobid);
    if (it == jobs_.end()) return false;
    if (persistent_ && !it->second.journal_path.empty()) {
      std::error_code ec;
      std::filesystem::remove(it->second.journal_path, ec);
    }
    erase_from_order(prjobid);
    jobs_.erase(it);
    if (persistent_) write_order_();
    return true;
  }

  // Per-PJ S16F9 alert gate. E40 §10.3 leaves PRALERT control to the host
  // (S16F1/F2 in the full multi-create form, which we don't model yet);
  // exposed here so application code can toggle alerts directly.
  bool set_alert(const std::string& prjobid, bool enabled) {
    auto* pj = get(prjobid);
    if (!pj) return false;
    pj->alert_enabled = enabled;
    if (persistent_) write_record_(prjobid);
    return true;
  }

  std::size_t size() const { return jobs_.size(); }

  // Insertion-order list of PRJOBIDs, mutated by HOQ. Used by CJ promotion
  // logic and by tests that need to assert ordering.
  const std::vector<std::string>& ids() const { return order_; }

  // ---- Persistence ----------------------------------------------------
  // One file per PJ (`<seq>.pj`); a single `order.idx` index records the
  // queue ordering (insertion-order modulo HOQ).  Replay reads the index
  // first to preserve ordering across restarts.  Version 2 of the record
  // (current) journals the E40 extras (rcpvars + prprocessparams) by
  // round-tripping each value through secs2::encode/decode.  Version 1
  // records are accepted on replay; their extras come back empty.
  void enable_persistence(std::filesystem::path dir) {
    namespace fs = std::filesystem;
    journal_dir_ = std::move(dir);
    persistent_ = true;
    std::error_code ec;
    fs::create_directories(journal_dir_, ec);

    std::map<std::string, ProcessJob> staging;
    std::map<std::string, fs::path> paths;
    std::vector<std::pair<uint64_t, fs::path>> existing;
    for (auto& entry : fs::directory_iterator(journal_dir_, ec)) {
      if (!entry.is_regular_file()) continue;
      const auto& p = entry.path();
      if (p.extension() != ".pj") continue;
      uint64_t seq = 0;
      try { seq = std::stoull(p.stem().string()); }
      catch (...) { continue; }
      existing.emplace_back(seq, p);
    }
    std::sort(existing.begin(), existing.end(),
              [](const auto& a, const auto& b) { return a.first < b.first; });

    for (auto& [seq, path] : existing) {
      auto rec = load_record_(path);
      if (!rec) { fs::remove(path, ec); continue; }
      install_(rec->prjobid, rec->ppid, std::move(rec->materials),
               rec->alert, rec->mf, rec->prrecipemethod, rec->state);
      auto* pj = get(rec->prjobid);
      if (pj) {
        pj->journal_path = path;
        pj->rcpvars = std::move(rec->rcpvars);
        pj->prprocessparams = std::move(rec->prprocessparams);
      }
      if (seq >= next_seq_) next_seq_ = seq + 1;
    }

    // Rebuild order_ from the index file when present; otherwise fall
    // back to creation order (already insertion-sorted by seq above).
    std::vector<std::string> idx = read_order_();
    order_.clear();
    if (!idx.empty()) {
      for (auto& id : idx) if (jobs_.count(id)) order_.push_back(id);
      // Append any PJs not in the index (defensive: shouldn't happen).
      for (auto& kv : jobs_) {
        if (std::find(order_.begin(), order_.end(), kv.first) == order_.end())
          order_.push_back(kv.first);
      }
    } else {
      for (auto& kv : jobs_) order_.push_back(kv.first);
    }
  }

  bool persistence_enabled() const { return persistent_; }
  std::filesystem::path journal_dir() const { return journal_dir_; }

 private:
  struct Record {
    std::string prjobid;
    std::string ppid;
    std::vector<std::string> materials;
    bool alert;
    MaterialFlag mf;
    ProcessRecipeMethod prrecipemethod;
    ProcessJobState state;
    std::vector<RcpVar> rcpvars;
    std::vector<ProcessParam> prprocessparams;
  };

  // PJ record (version 2):
  //   [u8 magic = 0xC7][u8 version]
  //   [u8 state][u8 alert][u8 mf][u8 prrecipemethod]
  //   [u16 prjobid_len][prjobid_bytes]
  //   [u16 ppid_len][ppid_bytes]
  //   [u16 material_count][repeat: u16 len + bytes]
  //   --- v2 additions ---
  //   [u16 rcpvar_count][repeat: u16 name_len, name, u32 enc_len, secs2::encode(value)]
  //   [u16 ppparam_count][repeat: u16 name_len, name, u32 enc_len, secs2::encode(value)]
  // v1 records (no extras) are accepted on replay; we just skip the
  // trailers and the in-memory rcpvars/prprocessparams stay empty.
  static constexpr uint8_t kMagic   = 0xC7;
  static constexpr uint8_t kVersion = 0x02;

  void install_(const std::string& prjobid, std::string ppid,
                std::vector<std::string> materials, bool alert,
                MaterialFlag mf, ProcessRecipeMethod prrecipemethod,
                ProcessJobState restored) {
    auto fsm = std::make_unique<ProcessJobStateMachine>(factory_(), restored);
    fsm->set_state_change_handler(
        [this, id = prjobid](ProcessJobState from, ProcessJobState to,
                             ProcessJobEvent trig) {
          if (on_change_) on_change_(id, from, to, trig);
        });
    ProcessJob pj;
    pj.prjobid = prjobid;
    pj.ppid = std::move(ppid);
    pj.mtrloutspec = std::move(materials);
    pj.alert_enabled = alert;
    pj.mf = mf;
    pj.prrecipemethod = prrecipemethod;
    pj.fsm = std::move(fsm);
    jobs_.emplace(prjobid, std::move(pj));
  }

  void write_record_(const std::string& prjobid) {
    namespace fs = std::filesystem;
    auto* pj = get(prjobid);
    if (!pj) return;
    if (pj->journal_path.empty()) {
      char namebuf[32];
      std::snprintf(namebuf, sizeof(namebuf), "%010llu.pj",
                    static_cast<unsigned long long>(next_seq_++));
      pj->journal_path = journal_dir_ / namebuf;
    }
    fs::path tmp = pj->journal_path;
    tmp += ".tmp";
    std::ofstream out(tmp, std::ios::binary | std::ios::trunc);
    if (!out) return;
    uint8_t header[6] = {
        kMagic, kVersion,
        static_cast<uint8_t>(pj->fsm->state()),
        static_cast<uint8_t>(pj->alert_enabled ? 1 : 0),
        static_cast<uint8_t>(pj->mf),
        static_cast<uint8_t>(pj->prrecipemethod)};
    out.write(reinterpret_cast<const char*>(header), sizeof(header));
    auto write_str = [&](const std::string& str) {
      uint16_t n = static_cast<uint16_t>(str.size());
      uint8_t lb[2] = {static_cast<uint8_t>(n >> 8), static_cast<uint8_t>(n)};
      out.write(reinterpret_cast<const char*>(lb), 2);
      out.write(str.data(), static_cast<std::streamsize>(n));
    };
    write_str(pj->prjobid);
    write_str(pj->ppid);
    auto write_be16 = [&](uint16_t v) {
      uint8_t b[2] = {static_cast<uint8_t>(v >> 8), static_cast<uint8_t>(v)};
      out.write(reinterpret_cast<const char*>(b), 2);
    };
    auto write_be32 = [&](uint32_t v) {
      uint8_t b[4] = {static_cast<uint8_t>((v >> 24) & 0xFF),
                      static_cast<uint8_t>((v >> 16) & 0xFF),
                      static_cast<uint8_t>((v >> 8) & 0xFF),
                      static_cast<uint8_t>(v & 0xFF)};
      out.write(reinterpret_cast<const char*>(b), 4);
    };
    write_be16(static_cast<uint16_t>(pj->mtrloutspec.size()));
    for (const auto& m : pj->mtrloutspec) write_str(m);
    auto write_named = [&](const std::string& name, const secs2::Item& value) {
      write_str(name);
      auto enc = secs2::encode(value);
      write_be32(static_cast<uint32_t>(enc.size()));
      out.write(reinterpret_cast<const char*>(enc.data()),
                static_cast<std::streamsize>(enc.size()));
    };
    write_be16(static_cast<uint16_t>(pj->rcpvars.size()));
    for (const auto& v : pj->rcpvars) write_named(v.name, v.value);
    write_be16(static_cast<uint16_t>(pj->prprocessparams.size()));
    for (const auto& v : pj->prprocessparams) write_named(v.name, v.value);
    out.flush();
    out.close();
    std::error_code ec;
    fs::rename(tmp, pj->journal_path, ec);
    if (ec) fs::remove(tmp, ec);
  }

  static std::optional<Record> load_record_(const std::filesystem::path& p) {
    std::ifstream in(p, std::ios::binary);
    if (!in) return std::nullopt;
    uint8_t header[6];
    in.read(reinterpret_cast<char*>(header), sizeof(header));
    if (!in || header[0] != kMagic) return std::nullopt;
    // v1 and v2 share the same prefix; v2 adds rcpvars+prprocessparams.
    if (header[1] != 0x01 && header[1] != 0x02) return std::nullopt;
    const uint8_t version = header[1];
    Record r;
    r.state          = static_cast<ProcessJobState>(header[2]);
    r.alert          = header[3] != 0;
    r.mf             = static_cast<MaterialFlag>(header[4]);
    r.prrecipemethod = static_cast<ProcessRecipeMethod>(header[5]);
    auto read_str = [&](std::string& out) {
      uint8_t lb[2];
      in.read(reinterpret_cast<char*>(lb), 2);
      if (!in) return false;
      uint16_t n = static_cast<uint16_t>((uint16_t(lb[0]) << 8) | lb[1]);
      out.resize(n);
      in.read(out.data(), n);
      return static_cast<bool>(in) || n == 0;
    };
    auto read_be16 = [&](uint16_t& v) {
      uint8_t b[2];
      in.read(reinterpret_cast<char*>(b), 2);
      if (!in) return false;
      v = static_cast<uint16_t>((uint16_t(b[0]) << 8) | b[1]);
      return true;
    };
    auto read_be32 = [&](uint32_t& v) {
      uint8_t b[4];
      in.read(reinterpret_cast<char*>(b), 4);
      if (!in) return false;
      v = (uint32_t(b[0]) << 24) | (uint32_t(b[1]) << 16) |
          (uint32_t(b[2]) << 8) | uint32_t(b[3]);
      return true;
    };
    if (!read_str(r.prjobid)) return std::nullopt;
    if (!read_str(r.ppid))    return std::nullopt;
    uint16_t mc = 0;
    if (!read_be16(mc)) return std::nullopt;
    r.materials.resize(mc);
    for (auto& m : r.materials) {
      if (!read_str(m)) return std::nullopt;
    }
    if (version >= 2) {
      auto read_named_list = [&](std::vector<RcpVar>& dst) -> bool {
        uint16_t n = 0;
        if (!read_be16(n)) return false;
        dst.resize(n);
        for (auto& kv : dst) {
          if (!read_str(kv.name)) return false;
          uint32_t enc_len = 0;
          if (!read_be32(enc_len)) return false;
          std::vector<uint8_t> bytes(enc_len);
          if (enc_len) {
            in.read(reinterpret_cast<char*>(bytes.data()), enc_len);
            if (!in) return false;
          }
          try { kv.value = secs2::decode(bytes); }
          catch (...) { return false; }
        }
        return true;
      };
      if (!read_named_list(r.rcpvars)) return std::nullopt;
      // ProcessParam has same shape as RcpVar; reuse via reinterpret-style:
      std::vector<RcpVar> pp_tmp;
      if (!read_named_list(pp_tmp)) return std::nullopt;
      r.prprocessparams.reserve(pp_tmp.size());
      for (auto& kv : pp_tmp) {
        r.prprocessparams.push_back({std::move(kv.name), std::move(kv.value)});
      }
    }
    return r;
  }

  void write_order_() {
    namespace fs = std::filesystem;
    fs::path target = journal_dir_ / "order.idx";
    fs::path tmp    = target;
    tmp += ".tmp";
    std::ofstream out(tmp, std::ios::binary | std::ios::trunc);
    if (!out) return;
    for (const auto& id : order_) {
      out << id << '\n';
    }
    out.flush();
    out.close();
    std::error_code ec;
    fs::rename(tmp, target, ec);
    if (ec) fs::remove(tmp, ec);
  }

  std::vector<std::string> read_order_() const {
    std::vector<std::string> out;
    std::ifstream in(journal_dir_ / "order.idx");
    if (!in) return out;
    std::string line;
    while (std::getline(in, line)) {
      if (!line.empty()) out.push_back(line);
    }
    return out;
  }

  void erase_from_order(const std::string& prjobid) {
    auto it = std::find(order_.begin(), order_.end(), prjobid);
    if (it != order_.end()) order_.erase(it);
  }
  void move_to_head(const std::string& prjobid) {
    auto it = std::find(order_.begin(), order_.end(), prjobid);
    if (it == order_.end() || it == order_.begin()) return;
    std::rotate(order_.begin(), it, it + 1);
  }

  std::map<std::string, ProcessJob> jobs_;
  std::vector<std::string> order_;        // queue position (E40 HOQ-aware)
  TransitionTableFactory factory_;
  HandlerSlot<const std::string&, ProcessJobState, ProcessJobState,
              ProcessJobEvent> on_change_;
  bool persistent_ = false;
  std::filesystem::path journal_dir_;
  uint64_t next_seq_ = 0;
};

}  // namespace secsgem::gem