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

#pragma once

#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/control_job_state.hpp"

namespace secsgem::gem {

// One Control Job — owns an ordered list of PRJOBIDs (process jobs).
struct ControlJob {
  std::string ctljobid;
  std::vector<std::string> prjobids;
  std::unique_ptr<ControlJobStateMachine> fsm;
  std::filesystem::path journal_path;
};

class ControlJobStore {
 public:
  using TransitionTableFactory =
      std::function<ControlJobTransitionTable()>;
  using StateChangeHandler =
      std::function<void(const std::string& ctljobid,
                         ControlJobState from, ControlJobState to,
                         ControlJobEvent trigger)>;

  ControlJobStore()
      : factory_([] { return ControlJobTransitionTable::default_table(); }) {}

  void set_table_factory(TransitionTableFactory f) { factory_ = std::move(f); }
  void set_state_change_handler(StateChangeHandler h) { on_change_ = std::move(h); }

  enum class CreateResult {
    Created,
    Denied_AlreadyExists,
    Denied_UnknownPRJob,
    Denied_Empty,
  };

  CreateResult create(std::string ctljobid,
                      std::vector<std::string> prjobids,
                      const std::function<bool(const std::string&)>& pj_exists =
                          [](const std::string&) { return true; }) {
    if (jobs_.count(ctljobid)) return CreateResult::Denied_AlreadyExists;
    if (prjobids.empty())      return CreateResult::Denied_Empty;
    for (const auto& id : prjobids) {
      if (!pj_exists(id)) return CreateResult::Denied_UnknownPRJob;
    }
    install_(ctljobid, std::move(prjobids), ControlJobState::Queued);
    if (persistent_) write_record_(ctljobid);
    if (on_change_) {
      on_change_(ctljobid, ControlJobState::NoState,
                 ControlJobState::Queued, ControlJobEvent::Created);
    }
    return CreateResult::Created;
  }

  bool has(const std::string& ctljobid) const {
    return jobs_.count(ctljobid) > 0;
  }
  const ControlJob* get(const std::string& ctljobid) const {
    auto it = jobs_.find(ctljobid);
    return it == jobs_.end() ? nullptr : &it->second;
  }
  ControlJob* get(const std::string& ctljobid) {
    auto it = jobs_.find(ctljobid);
    return it == jobs_.end() ? nullptr : &it->second;
  }

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

  HostCmdAck on_host_command(const std::string& ctljobid, ControlJobEvent event) {
    auto* cj = get(ctljobid);
    if (!cj) return HostCmdAck::InvalidObject;
    HostCmdAck ack = cj->fsm->on_host_command(event);
    if (ack == HostCmdAck::Accept && persistent_) write_record_(ctljobid);
    return ack;
  }

  bool fire_internal(const std::string& ctljobid, ControlJobEvent event) {
    auto* cj = get(ctljobid);
    if (!cj) return false;
    bool ok = cj->fsm->on_internal(event);
    if (ok && persistent_) write_record_(ctljobid);
    return ok;
  }

  bool remove(const std::string& ctljobid) {
    auto it = jobs_.find(ctljobid);
    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);
    }
    jobs_.erase(it);
    return true;
  }

  std::size_t size() const { return jobs_.size(); }
  std::vector<std::string> ids() const {
    std::vector<std::string> out;
    out.reserve(jobs_.size());
    for (const auto& kv : jobs_) out.push_back(kv.first);
    return out;
  }

  // ---- Persistence ----------------------------------------------------
  // Per-CJ binary record (`<seq>.cj`).  CJ ordering across CJs is not
  // managed by E94 (each CJ is independent), so no index file is needed.
  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::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() != ".cj") 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->ctljobid, std::move(rec->prjobids), rec->state);
      auto* cj = get(rec->ctljobid);
      if (cj) cj->journal_path = path;
      if (seq >= next_seq_) next_seq_ = seq + 1;
    }
  }

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

 private:
  struct Record {
    std::string ctljobid;
    std::vector<std::string> prjobids;
    ControlJobState state;
  };

  // CJ record (v1):
  //   [u8 magic = 0xC8][u8 version][u8 state]
  //   [u16 ctljobid_len][ctljobid_bytes]
  //   [u16 pj_count][repeat: u16 len + bytes]
  //
  // Upgrade discipline: when adding fields, bump kVersion and append
  // the new fields at the end of the on-disk record.  The loader
  // accepts any version in [1, kVersion]; older records replay with
  // the trailing fields defaulted.  Future versions read by this
  // binary surface as a rejection (so a downgrade can't silently
  // corrupt data), but the same code that bumped kVersion to N must
  // also gate the new trailer behind `if (version >= N)`.
  static constexpr uint8_t kMagic   = 0xC8;
  static constexpr uint8_t kVersion = 0x01;

  void install_(const std::string& ctljobid,
                std::vector<std::string> prjobids,
                ControlJobState restored) {
    auto fsm = std::make_unique<ControlJobStateMachine>(factory_(), restored);
    const std::string id = ctljobid;
    if (on_change_) {
      auto cb = on_change_;
      fsm->set_state_change_handler(
          [cb, id](ControlJobState from, ControlJobState to,
                   ControlJobEvent trig) { cb(id, from, to, trig); });
    }
    jobs_.emplace(ctljobid,
                  ControlJob{ctljobid, std::move(prjobids),
                             std::move(fsm), {}});
  }

  void write_record_(const std::string& ctljobid) {
    namespace fs = std::filesystem;
    auto* cj = get(ctljobid);
    if (!cj) return;
    if (cj->journal_path.empty()) {
      char namebuf[32];
      std::snprintf(namebuf, sizeof(namebuf), "%010llu.cj",
                    static_cast<unsigned long long>(next_seq_++));
      cj->journal_path = journal_dir_ / namebuf;
    }
    fs::path tmp = cj->journal_path;
    tmp += ".tmp";
    std::ofstream out(tmp, std::ios::binary | std::ios::trunc);
    if (!out) return;
    uint8_t header[3] = {
        kMagic, kVersion, static_cast<uint8_t>(cj->fsm->state())};
    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(cj->ctljobid);
    uint16_t pjc = static_cast<uint16_t>(cj->prjobids.size());
    uint8_t pb[2] = {static_cast<uint8_t>(pjc >> 8), static_cast<uint8_t>(pjc)};
    out.write(reinterpret_cast<const char*>(pb), 2);
    for (const auto& id : cj->prjobids) write_str(id);
    out.flush();
    out.close();
    std::error_code ec;
    fs::rename(tmp, cj->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[3];
    in.read(reinterpret_cast<char*>(header), sizeof(header));
    if (!in || header[0] != kMagic ||
        header[1] < 1 || header[1] > kVersion) return std::nullopt;
    Record r;
    r.state = static_cast<ControlJobState>(header[2]);
    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;
    };
    if (!read_str(r.ctljobid)) return std::nullopt;
    uint8_t pb[2];
    in.read(reinterpret_cast<char*>(pb), 2);
    if (!in) return std::nullopt;
    uint16_t pjc = static_cast<uint16_t>((uint16_t(pb[0]) << 8) | pb[1]);
    r.prjobids.resize(pjc);
    for (auto& id : r.prjobids) {
      if (!read_str(id)) return std::nullopt;
    }
    return r;
  }

  std::map<std::string, ControlJob> jobs_;
  TransitionTableFactory factory_;
  StateChangeHandler on_change_;
  bool persistent_ = false;
  std::filesystem::path journal_dir_;
  uint64_t next_seq_ = 0;
};

}  // namespace secsgem::gem