file_journal.hpp

#pragma once
 
#include "./journal.hpp"
 
#include <algotradelib/core>
 
#include <boost/crc.hpp>
#include <boost/json.hpp>
 
#include <cerrno>
#include <chrono>
#include <cstdint>
#include <cstdio>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
 
#if defined(_WIN32)
#include <io.h>
#else
#include <fcntl.h>
#include <sys/file.h>
#include <unistd.h>
 
#include <cstring>
#endif
 
namespace algotradelib {
namespace runner {
 
namespace json = boost::json;
 
namespace journal_detail {

inline bool flush_to_disk(std::FILE* handle) noexcept {
  if (std::fflush(handle) != 0) {
    return false;
  }
#if defined(_WIN32)
  return ::_commit(::_fileno(handle)) == 0;
#else
  return ::fsync(::fileno(handle)) == 0;
#endif
}

inline bool plausible_instant(std::int64_t micros) noexcept {
  constexpr std::int64_t kYear2020 = 1577836800LL * 1000000LL;
  constexpr std::int64_t kCenturyOut = 4102444800LL * 1000000LL;
  return micros >= kYear2020 && micros <= kCenturyOut;
}
 
inline bool has_field(const json::object& row, std::string_view key) {
  return row.if_contains(key) != nullptr;
}

inline std::optional<std::string> readable_string(
  const json::object& row, std::string_view key
) {
  const auto value = row.if_contains(key);
  if (value == nullptr || !value->is_string() || value->as_string().empty()) {
    return std::nullopt;
  }
  return std::string(value->as_string());
}
 
inline std::string string_field(const json::object& row, std::string_view key) {
  const auto value = row.if_contains(key);
  if (value == nullptr || !value->is_string()) {
    return {};
  }
  return std::string(value->as_string());
}

inline constexpr std::string_view country_name(Country country) noexcept {
  switch (country) {
    case Country::KR:
      return "KR";
    case Country::US:
      return "US";
  }
  return "";
}
 
static_assert(
  lists_every_value(all_countries, [](Country country) { return !country_name(country).empty(); }),
  "all_countries must list every Country"
);

inline constexpr std::string_view venue_name(Venue venue) noexcept {
  switch (venue) {
    case Venue::UNSPECIFIED:
      return "UNSPECIFIED";
    case Venue::KRX:
      return "KRX";
    case Venue::NXT:
      return "NXT";
    case Venue::COMBINED:
      return "COMBINED";
  }
  return "";
}

inline std::optional<Venue> known_venue_from(std::string_view text) noexcept {
  return value_named(all_venues, venue_name, text);
}
 

inline constexpr std::string_view side_name(OrderSide side) noexcept {
  switch (side) {
    case OrderSide::BUY:
      return "BUY";
    case OrderSide::SELL:
      return "SELL";
  }
  return "";
}
 
static_assert(
  lists_every_value(all_order_sides, [](OrderSide side) { return !side_name(side).empty(); }),
  "all_order_sides must list every OrderSide"
);
 

inline constexpr std::string_view order_type_name(OrderType type) noexcept {
  switch (type) {
    case OrderType::LIMIT:
      return "LIMIT";
    case OrderType::MARKET:
      return "MARKET";
    case OrderType::CONDITIONAL_LIMIT:
      return "CONDITIONAL_LIMIT";
    case OrderType::BEST_LIMIT:
      return "BEST_LIMIT";
    case OrderType::PRIMARY_LIMIT:
      return "PRIMARY_LIMIT";
    case OrderType::MID_PRICE:
      return "MID_PRICE";
    case OrderType::STOP_LIMIT:
      return "STOP_LIMIT";
    case OrderType::CLOSING_PRICE:
      return "CLOSING_PRICE";
    case OrderType::UNKNOWN:
      return "UNKNOWN";
  }
  return "";
}
 
static_assert(
  lists_every_value(all_order_types, [](OrderType type) { return !order_type_name(type).empty(); }),
  "all_order_types must list every OrderType"
);
 

inline constexpr std::string_view time_in_force_name(TimeInForce condition) noexcept {
  switch (condition) {
    case TimeInForce::DAY:
      return "DAY";
    case TimeInForce::IMMEDIATE_OR_CANCEL:
      return "IOC";
    case TimeInForce::FILL_OR_KILL:
      return "FOK";
    case TimeInForce::UNKNOWN:
      return "UNKNOWN";
  }
  return "";
}
 
static_assert(
  lists_every_value(all_times_in_force, [](TimeInForce condition) { return !time_in_force_name(condition).empty(); }),
  "all_times_in_force must list every TimeInForce"
);

inline constexpr std::string_view order_session_name(OrderSession session) noexcept {
  switch (session) {
    case OrderSession::REGULAR:
      return "REGULAR";
    case OrderSession::OPENING_AUCTION:
      return "OPENING_AUCTION";
    case OrderSession::CLOSING_AUCTION:
      return "CLOSING_AUCTION";
    case OrderSession::PRE_MARKET:
      return "PRE_MARKET";
    case OrderSession::AFTER_HOURS:
      return "AFTER_HOURS";
    case OrderSession::AFTER_HOURS_AUCTION:
      return "AFTER_HOURS_AUCTION";
    case OrderSession::AFTER_MARKET:
      return "AFTER_MARKET";
    case OrderSession::GTP:
      return "GTP";
    case OrderSession::UNKNOWN:
      return "UNKNOWN";
  }
  return "";
}
 
static_assert(
  lists_every_value(all_order_sessions, [](OrderSession session) { return !order_session_name(session).empty(); }),
  "all_order_sessions must list every OrderSession"
);
 

inline std::optional<Decimal64> known_decimal_from(std::string_view text) noexcept {
  try {
    return Decimal64::parse(text);
  } catch (const std::exception&) {
    return std::nullopt;
  }
}
 
inline std::optional<Country> known_country_from(std::string_view text) noexcept {
  return value_named(all_countries, country_name, text);
}
 
inline std::optional<OrderSide> known_side_from(std::string_view text) noexcept {
  return value_named(all_order_sides, side_name, text);
}
 
inline std::optional<OrderType> known_order_type_from(std::string_view text) noexcept {
  return value_named(all_order_types, order_type_name, text);
}
 
inline std::optional<TimeInForce> known_time_in_force_from(std::string_view text) noexcept {
  return value_named(all_times_in_force, time_in_force_name, text);
}
 
inline std::optional<OrderSession> known_order_session_from(std::string_view text) noexcept {
  return value_named(all_order_sessions, order_session_name, text);
}
 

inline std::chrono::system_clock::time_point instant_from(std::int64_t micros) noexcept {
  return std::chrono::system_clock::time_point{} + std::chrono::microseconds(micros);
}

inline std::int64_t micros_of(std::chrono::system_clock::time_point value) noexcept {
  return std::chrono::duration_cast<std::chrono::microseconds>(value.time_since_epoch()).count();
}

inline std::optional<std::string> salvage_string(std::string_view line, std::string_view key) {
  std::string needle;
  needle.reserve(key.size() + 4);
  needle += '"';
  needle.append(key);
  needle += "\":\"";
  const auto at = line.find(needle);
  if (at == std::string_view::npos) {
    return std::nullopt;
  }
  const auto begin = at + needle.size();
  // **The closing quote is the first one not escaped**, not the first one. A value holding `\"`
  // was cut short at the escape, and a symbol cut short is another symbol's key.
  auto end = begin;
  while (end < line.size() && line[end] != '"') {
    end += line[end] == '\\' ? 2 : 1;
  }
  if (end >= line.size()) {
    // The value itself was cut off mid-write. Reporting what is there would invent a symbol.
    return std::nullopt;
  }
  // Decoded as the JSON string it is, so an escape reads as the character it stands for. A value
  // whose escapes do not decode is not salvaged, for the same reason a cut-off one is not.
  boost::system::error_code error;
  const auto quoted = json::parse(line.substr(begin - 1, end - begin + 2), error);
  if (error || !quoted.is_string()) {
    return std::nullopt;
  }
  return std::string(quoted.as_string());
}

inline std::uint32_t line_checksum(std::string_view bytes) {
  boost::crc_32_type crc;
  crc.process_bytes(bytes.data(), bytes.size());
  return crc.checksum();
}

inline std::string checksum_text(std::uint32_t value) {
  static constexpr char digits[] = "0123456789abcdef";
  std::string text(8, '0');
  for (int i = 7; i >= 0; --i) {
    text[static_cast<std::size_t>(i)] = digits[value & 0xFu];
    value >>= 4;
  }
  return text;
}

inline constexpr std::string_view kChecksumMember = ",\"crc\":\"";

inline constexpr std::size_t kChecksumSuffixSize = kChecksumMember.size() + 8 + 2;

inline bool ends_in_checksum(std::string_view line) {
  if (line.size() <= kChecksumSuffixSize) {
    return false;
  }
  const auto suffix = line.substr(line.size() - kChecksumSuffixSize);
  if (suffix.substr(0, kChecksumMember.size()) != kChecksumMember ||
      suffix.substr(kChecksumSuffixSize - 2) != "\"}") {
    return false;
  }
  for (const char digit : suffix.substr(kChecksumMember.size(), 8)) {
    if (!((digit >= '0' && digit <= '9') || (digit >= 'a' && digit <= 'f'))) {
      return false;
    }
  }
  return true;
}

inline bool checksum_matches(std::string_view line) {
  std::string body(line.substr(0, line.size() - kChecksumSuffixSize));
  body += '}';
  const auto stated = line.substr(line.size() - kChecksumSuffixSize + kChecksumMember.size(), 8);
  return checksum_text(line_checksum(body)) == stated;
}

inline bool known_entry_key(std::string_view key) {
  static constexpr std::string_view known[] = {
    "id", "action", "outcome", "symbol", "country", "venue", "side", "type", "time_in_force",
    "session", "route", "quantity", "price", "trigger", "at", "at_text", "order_id", "detail",
    "requires_operator", "scope", "crc", "baseline",
  };
  for (const auto candidate : known) {
    if (candidate == key) {
      return true;
    }
  }
  return false;
}
 
} // namespace journal_detail

inline constexpr std::string_view journal_format_name() noexcept {
  return "algotradelib-order-journal";
}

inline std::string journal_header_line(std::string_view scope) {
  json::object row;
  row["journal"] = std::string(journal_format_name());
  row["version"] = 1;
  row["scope"] = std::string(scope);
  // Declares that every entry after the header carries a checksum (older readers ignore the key).
  row["integrity"] = "crc32";
  return json::serialize(row);
}

inline bool journal_header_declares_checksums(std::string_view line) {
  boost::system::error_code error;
  const auto value = json::parse(line, error);
  if (error || !value.is_object()) {
    return false;
  }
  const auto* integrity = value.as_object().if_contains("integrity");
  return integrity != nullptr && integrity->is_string() &&
    std::string_view(integrity->as_string()) == "crc32";
}

inline std::optional<std::string> journal_header_scope(std::string_view line) {
  boost::system::error_code error;
  const auto value = json::parse(line, error);
  if (error || !value.is_object()) {
    return std::nullopt;
  }
  const auto& row = value.as_object();
  const auto marker = row.if_contains("journal");
  if (marker == nullptr || !marker->is_string() ||
      std::string_view(marker->as_string()) != journal_format_name()) {
    return std::nullopt;
  }
  return journal_detail::string_field(row, "scope");
}

struct JournalBytes {
  JournalReadStatus status_{JournalReadStatus::MISSING};
  std::string text_;
  std::string detail_;
};

inline std::string journal_entry_to_line(const JournalEntry& entry) {
  json::object row;
  // Ordered so that the fields a damaged line most needs to yield come first. A truncated write
  // keeps its prefix, so putting the instrument ahead of the detail text is what makes salvage
  // work in practice rather than only in principle.
  row["id"] = entry.id_;
  row["action"] = std::string(journal_action_name(entry.action_));
  row["outcome"] = std::string(journal_outcome_name(entry.outcome_));
  row["country"] = std::string(journal_detail::country_name(entry.instrument_.country_));
  row["venue"] = std::string(journal_detail::venue_name(entry.instrument_.venue_));
  row["symbol"] = entry.instrument_.symbol_;
  row["side"] = std::string(journal_detail::side_name(entry.side_));
  row["type"] = std::string(journal_detail::order_type_name(entry.type_));
  row["time_in_force"] = std::string(journal_detail::time_in_force_name(entry.time_in_force_));
  row["session"] = std::string(journal_detail::order_session_name(entry.session_));
  // Distinct from `venue`, which is the instrument's identity venue. This is
  // where the order actually goes -- the binding resolves a request that named
  // no venue into the one the broker's default reaches -- and a line that omits
  // the key entirely is
  // one written before the field existed -- which is not the same as one that
  // asked for no particular venue, and recovery treats the two differently.
  if (entry.venue_) {
    row["route"] = std::string(journal_detail::venue_name(*entry.venue_));
  }
  row["quantity"] = entry.quantity_.str();
  if (entry.price_) {
    row["price"] = entry.price_->str();
  }
  // Only when present, so a line that is not a stop carries no `trigger` key and every old journal
  // reads unchanged. (Every new line ends in a `crc` member, added below; the fields before it are
  // what an earlier writer produced.)
  if (entry.trigger_price_) {
    row["trigger"] = entry.trigger_price_->str();
  }
  // Only on a serialized runner's lines, so an overlapping runner's journal reads as before.
  if (entry.baseline_) {
    row["baseline"] = entry.baseline_->str();
  }
  // The authoritative instant, exact on every platform. `at_text` beside it is for the operator
  // reading the file with `tail`, is only second-resolution, and is never read back: one field is
  // the record and the other is the caption.
  row["at"] = journal_detail::micros_of(entry.recorded_at_);
  row["at_text"] = format_iso8601_utc(entry.recorded_at_);
  if (!entry.order_id_.empty()) {
    row["order_id"] = entry.order_id_;
  }
  if (!entry.detail_.empty()) {
    row["detail"] = entry.detail_;
  }
  if (entry.requires_operator_) {
    row["requires_operator"] = true;
  }
  if (!entry.scope_.empty()) {
    row["scope"] = entry.scope_;
  }
  // **Last, and over the line as it reads without it** -- the whole object, closing brace included,
  // which the member then replaces. See `journal_detail::line_checksum` for what it detects
  // and `journal_entry_from_line` for what a line that fails it becomes.
  auto line = json::serialize(row);
  const auto checksum = journal_detail::checksum_text(journal_detail::line_checksum(line));
  line.pop_back();
  line.append(journal_detail::kChecksumMember);
  line += checksum;
  line += "\"}";
  return line;
}

inline JournalEntry unattributable_journal_entry(std::string_view line) {
  JournalEntry entry;
  entry.id_ = "unknown~truncated";
  entry.outcome_ = JournalOutcome::PENDING;
  entry.requires_operator_ = true;
  entry.detail_ = "a truncated journal line named no instrument; something was in flight: " +
    std::string(line.substr(0, line.size() < 120 ? line.size() : 120));
  return entry;
}

inline std::optional<JournalEntry> journal_entry_from_line(std::string_view line) {
  boost::system::error_code error;
  const auto value = json::parse(line, error);
  if (error || !value.is_object()) {
    return std::nullopt;
  }
  const auto& row = value.as_object();
 
  const auto action = journal_action_from(journal_detail::string_field(row, "action"));
  const auto outcome = journal_outcome_from(journal_detail::string_field(row, "outcome"));
  // An operator-only line is exempt from the bound below, and deliberately: a
  // salvaged fragment records the epoch precisely *because* it has no
  // timestamp, and `requires_operator_` is what stops that epoch window
  // resolving it. Repairing such a line writes it back, so rejecting the epoch
  // here would make every repair unreadable on the next open. The bound is for
  // a line that claims to be ordinary and carries an instant it cannot have.
  const auto operator_only = row.if_contains("requires_operator");
  const bool marked = operator_only != nullptr && operator_only->is_bool() &&
    operator_only->as_bool();
 
  const auto at = row.if_contains("at");
  if (!action || !outcome || at == nullptr || !at->is_int64() ||
      (!marked && !journal_detail::plausible_instant(at->as_int64()))) {
    // A well-formed JSON object that is not a journal line. Treated as damage rather than as an
    // empty entry, because an entry with a default instrument would be a claim about an order.
    return std::nullopt;
  }
 
  // **A present value this reader cannot parse is damage, not a default.**
  // Each of these has a documented default for an *absent* key -- a line that
  // predates the field -- and applying that default to a token the line does
  // carry turns damage into evidence: a damaged `"SELL"` reads as a buy, a
  // damaged `"IOC"` as a day order, and a damaged `"COMBINED"` moves the entry
  // to an instrument the runner is not trading, so the ledger marks the wrong
  // key and the right one is never withheld at all.
  //
  // An unreadable value therefore produces an entry with **no instrument**,
  // which `unattributable_journal_entry` documents and recovery already turns
  // into a global halt. Marking a guessed instrument would not protect the real
  // one, because they are different keys.
  // Each field is asked two questions, because they have different answers.
  // *Is the key there?* -- if not, the line predates the field and its
  // documented default is a fact about such a line. *Is what it holds
  // readable?* -- if not, the line says something this release cannot use, and
  // guessing turns damage into evidence. `string_field` answers neither: it
  // flattens absent, non-string and empty into the same `""`.
  const auto read = [&row](std::string_view key) {
    return journal_detail::readable_string(row, key);
  };
  // Present but unusable, for any reason: not a string, an empty string, or a
  // token the reader could not parse. The first version of this asked only
  // whether the *string* was readable, which let `"IOC"` damaged to `"IO?"`
  // through -- that is a perfectly good string, and a perfectly bad condition.
  const auto unusable = [&row](std::string_view key, bool parsed) {
    return journal_detail::has_field(row, key) && !parsed;
  };
 
  // `std::optional::and_then` is C++23 and this library is C++17.
  const auto parsed = [&read](std::string_view key, auto reader) {
    using Result = decltype(reader(std::string_view{}));
    const auto text = read(key);
    return text ? reader(*text) : Result{};
  };
 
  // A `crc` member anywhere but last, or not eight lowercase hex digits, is not this writer's; one
  // that is must match the line without it -- the object with its own closing brace. See
  // `journal_detail::line_checksum`.
  const bool checksum_holds = !journal_detail::has_field(row, "crc") ||
    (journal_detail::ends_in_checksum(line) && journal_detail::checksum_matches(line));
  bool unknown_key = false;
  for (const auto& member : row) {
    if (!journal_detail::known_entry_key(member.key())) {
      unknown_key = true;
    }
  }
  // **A line whose bytes changed is not read at all**, rather than read as a damaged entry. A damaged
  // entry carries its own outcome and an id suffixed `~damaged`, which is right for a line whose
  // *value* no longer parses -- but a damaged *terminal* line with an intact pending partner then
  // pairs with nothing, is dropped as an orphan, and the pending line stays open as an ordinary one:
  // the change was detected and then discarded before recovery decided anything. An unreadable line
  // is counted by `parse_journal_bytes` and halts recovery outright, whatever it was.
  if (!checksum_holds || unknown_key) {
    return std::nullopt;
  }
 
  const auto symbol = read("symbol");
  const auto country = parsed("country", journal_detail::known_country_from);
  const auto venue = parsed("venue", journal_detail::known_venue_from);
  const auto side = parsed("side", journal_detail::known_side_from);
  const auto type = parsed("type", journal_detail::known_order_type_from);
  const auto condition = parsed("time_in_force", journal_detail::known_time_in_force_from);
  const auto session = parsed("session", journal_detail::known_order_session_from);
  const auto quantity = parsed("quantity", journal_detail::known_decimal_from);
  const auto price = parsed("price", journal_detail::known_decimal_from);
  const auto trigger = parsed("trigger", journal_detail::known_decimal_from);
  const auto baseline = parsed("baseline", journal_detail::known_decimal_from);
 
  const bool damaged =
    // Identity and the fields every complete line carries: required outright.
    // A line this reader cannot attribute to an instrument must not be
    // attributed to a guess, and `quantity` at a silent zero is a claim about
    // an order's size.
    !symbol || !country || !venue || !side || !type || !quantity ||
    // Legacy-defaulted fields: absent is fine, present and unreadable is not.
    unusable("time_in_force", condition.has_value()) ||
    unusable("session", session.has_value()) ||
    // A price is absent for a market order and required for a basis that
    // carries one, so its presence is judged against the basis rather than
    // assumed either way.
    unusable("price", price.has_value()) ||
    // The same rule for a stop's trigger: absent on every other order and on every line written
    // before the field existed, and a present value that does not parse is damage.
    unusable("trigger", trigger.has_value()) ||
    unusable("baseline", baseline.has_value()) ||
    // Checked above, where a changed line becomes unreadable rather than damaged; restated so the
    // condition reads whole.
    !checksum_holds ||
    unknown_key;
  // **A price missing from a priced basis, and a cancellation with no target
  // id, are deliberately *not* damage here.** Both are lines this library's own
  // writer emits: `JournalingExecutor` appends the pending line *before* the
  // executor runs, so an order the binding then refuses locally -- a limit with
  // no price, a cancellation of a reservation whose broker id is not known --
  // has already been written by the time anyone could object. Rejecting them
  // turned an ordinary local refusal into a global halt on the next restart,
  // which is worse than the hazard the rejection was meant to close.
  //
  // The hazard is real and it lives one step later: such a line must not be
  // *resolved* on evidence it does not have. `runner::match_against` refuses to
  // conclude anything from one, which leaves it withheld for an operator while
  // its own terminal line still closes it when the refusal was ordinary.
  if (damaged) {
    auto unattributable = unattributable_journal_entry(line);
    const auto id = journal_detail::string_field(row, "id");
    unattributable.id_ = (id.empty() ? std::string("unknown") : id) + "~damaged";
    unattributable.action_ = *action;
    // Its own outcome, not `PENDING`. Both lines of a pair carry the same id, so
    // a damaged *terminal* line still closes its damaged pending partner --
    // forcing `PENDING` left both open and halted a runner over an instruction
    // that had already resolved.
    unattributable.outcome_ = *outcome;
    unattributable.recorded_at_ = journal_detail::instant_from(at->as_int64());
    unattributable.detail_ =
      "a journal line carries a value this release cannot read, so the order it describes cannot "
      "be identified; nothing may be traded until an operator resolves it: " +
      std::string(line.substr(0, line.size() < 160 ? line.size() : 160));
    return unattributable;
  }
 
  JournalEntry entry;
  entry.id_ = journal_detail::string_field(row, "id");
  entry.action_ = *action;
  entry.outcome_ = *outcome;
  entry.instrument_ = InstrumentId{*country, *venue, *symbol};
  entry.side_ = *side;
  entry.type_ = *type;
  // These two keep their documented defaults for an *absent* key only; a
  // present value that failed to parse was caught above.
  entry.time_in_force_ = condition ? *condition : TimeInForce::DAY;
  entry.session_ = session ? *session : OrderSession::REGULAR;
  // Absent key -> absent value, which is "this line predates the field" rather
  // than "no route was asked for". Recovery will not attribute a row to an
  // instruction whose route nobody recorded. The field defaults to a recorded
  // `UNSPECIFIED`, so this clears it rather than leaving the default standing.
  // An absent key means the line predates the field. An unrecognized value means
  // the line says something this reader cannot use. Both are "unknown", and
  // neither may become the `UNSPECIFIED` that recovery treats as a recorded
  // request for no venue.
  entry.venue_ = journal_detail::known_venue_from(journal_detail::string_field(row, "route"));
  // Damaged numbers land in the same place as damaged enumerations, for the
  // same reason -- and they used to land somewhere far worse, throwing out of
  // recovery entirely. See `known_decimal_from`.
  entry.quantity_ = *quantity;
  if (price) {
    entry.price_ = *price;
  }
  if (trigger) {
    entry.trigger_price_ = *trigger;
  }
  entry.baseline_ = baseline;
  entry.recorded_at_ = journal_detail::instant_from(at->as_int64());
  entry.order_id_ = journal_detail::string_field(row, "order_id");
  entry.detail_ = journal_detail::string_field(row, "detail");
  entry.requires_operator_ = marked;
  entry.scope_ = journal_detail::string_field(row, "scope");
  return entry;
}

inline bool salvage_damaged(std::string_view line, std::string_view key) {
  const std::string needle = "\"" + std::string(key) + "\":";
  const auto at = line.find(needle);
  if (at == std::string_view::npos) {
    return false;
  }
  return line.find(',', at + needle.size()) != std::string_view::npos;
}

inline std::optional<JournalEntry> salvage_journal_line(std::string_view line) {
  const auto symbol = journal_detail::salvage_string(line, "symbol");
  if (!symbol || symbol->empty()) {
    return std::nullopt;
  }
  JournalEntry entry;
  // The flag that makes the promise in this comment block true. Without it the entry's missing
  // timestamp reads as the epoch, every window test passes immediately, no real row falls inside
  // the window, and the first sweep resolves it `NOT_PLACED`.
  entry.requires_operator_ = true;
  const auto id = journal_detail::salvage_string(line, "id");
  entry.id_ = (id ? *id : std::string("unknown")) + "~truncated";
  entry.action_ =
    journal_action_from(journal_detail::salvage_string(line, "action").value_or("place"))
      .value_or(JournalAction::PLACE);
  entry.outcome_ = JournalOutcome::PENDING;
  // **Absent is a default; present and unreadable is damage** -- the same rule
  // the strict reader follows, and for the same reason. A truncated line is
  // missing its *tail*, so an absent field is expected and its default is a fact
  // about what the line could have said. A field that is there and cannot be
  // read is corruption in the middle, and guessing it hands this entry a
  // different instrument than the order had: `requires_operator_` would then
  // withhold a key the runner is not trading under, while the real one stays
  // open. Such a line becomes unattributable instead, carrying no instrument at
  // all, which recovery turns into a global halt.
  // A key the truncation removed takes its default; a key that is *there* and
  // unreadable is damage, and `salvage_string` cannot tell those apart on its
  // own -- it recognizes `"key":"value"` and nothing else, so `null` and `7`
  // both read as absent.
  const auto salvaged = [&line](std::string_view key, const char* fallback) {
    const auto text = journal_detail::salvage_string(line, key);
    if (text) {
      return std::optional<std::string>(*text);
    }
    return salvage_damaged(line, key) ? std::optional<std::string>()
                                                       : std::optional<std::string>(fallback);
  };
  const auto parsed = [&salvaged](std::string_view key, const char* fallback, auto reader) {
    using Result = decltype(reader(std::string_view{}));
    const auto text = salvaged(key, fallback);
    return text ? reader(*text) : Result{};
  };
 
  const auto country = parsed("country", "KR", journal_detail::known_country_from);
  const auto venue = parsed("venue", "UNSPECIFIED", journal_detail::known_venue_from);
  const auto side = parsed("side", "BUY", journal_detail::known_side_from);
  const auto type = parsed("type", "UNKNOWN", journal_detail::known_order_type_from);
  const auto condition =
    parsed("time_in_force", "DAY", journal_detail::known_time_in_force_from);
  const auto session = parsed("session", "REGULAR", journal_detail::known_order_session_from);
  if (!country || !venue || !side || !type || !condition || !session) {
    auto unattributable = unattributable_journal_entry(line);
    unattributable.id_ = entry.id_;
    unattributable.detail_ =
      "a truncated journal line carries a value this release cannot read, so the order it "
      "describes cannot be identified; nothing may be traded until an operator resolves it: " +
      std::string(line.substr(0, line.size() < 160 ? line.size() : 160));
    return unattributable;
  }
  entry.instrument_ = InstrumentId{*country, *venue, *symbol};
  entry.side_ = *side;
  entry.type_ = *type;
  entry.time_in_force_ = *condition;
  entry.session_ = *session;
  entry.venue_ = journal_detail::known_venue_from(
    journal_detail::salvage_string(line, "route").value_or("")
  );
  if (const auto quantity = journal_detail::salvage_string(line, "quantity")) {
    // A truncated line's last field is routinely half-written, so an unparseable
    // number here is expected rather than exceptional -- and it must not throw
    // out of a salvage path that exists to survive exactly this.
    if (const auto value = journal_detail::known_decimal_from(*quantity)) {
      entry.quantity_ = *value;
    }
  }
  if (const auto price = journal_detail::salvage_string(line, "price")) {
    if (const auto value = journal_detail::known_decimal_from(*price)) {
      entry.price_ = *value;
    }
  }
  if (const auto trigger = journal_detail::salvage_string(line, "trigger")) {
    if (const auto value = journal_detail::known_decimal_from(*trigger)) {
      entry.trigger_price_ = *value;
    }
  }
  entry.detail_ = "recovered from a truncated journal line; only an operator can resolve it";
  return entry;
}
 

inline JournalBytes read_journal_bytes(const std::string& path) {
  JournalBytes bytes;
  errno = 0;
  std::unique_ptr<std::FILE, int (*)(std::FILE*)> handle(
    std::fopen(path.c_str(), "rb"), [](std::FILE* file) { return file ? std::fclose(file) : 0; }
  );
  if (handle == nullptr) {
    if (errno == ENOENT) {
      bytes.status_ = JournalReadStatus::MISSING;
      return bytes;
    }
    // Every other reason -- a permission the process does not have, a directory where a file was
    // expected, a device error -- means the file exists and its contents are unknown. Reporting
    // that as "no journal" is how a process claims a path holding its predecessor's in-flight
    // orders and then trades on an empty ledger.
    bytes.status_ = JournalReadStatus::UNREADABLE;
    bytes.detail_ = "the order journal at " + path + " exists and could not be opened for reading";
    return bytes;
  }
 
  char buffer[4096];
  while (const auto read = std::fread(buffer, 1, sizeof(buffer), handle.get())) {
    bytes.text_.append(buffer, read);
  }
  if (std::ferror(handle.get()) != 0) {
    // A read that stopped partway looks exactly like an ordinary end of file to the loop above, so
    // the truncated tail it produces would be a fiction about where the journal ends.
    bytes.status_ = JournalReadStatus::UNREADABLE;
    bytes.detail_ = "the order journal at " + path + " could not be read to the end";
    bytes.text_.clear();
    return bytes;
  }
  bytes.status_ = JournalReadStatus::READ;
  return bytes;
}

inline JournalReplay parse_journal_bytes(const JournalBytes& bytes) {
  JournalReplay replay;
  replay.status_ = bytes.status_;
  if (bytes.status_ != JournalReadStatus::READ) {
    replay.detail_ = bytes.detail_;
    return replay;
  }
 
  const auto& text = bytes.text_;
  std::size_t begin = 0;
  // Set by the first checksummed entry line; see the check below.
  bool seen_checksum = false;
  while (begin < text.size()) {
    const auto end = text.find('\n', begin);
    if (end == std::string::npos) {
      // Everything after the last newline is a write that never finished. This is the whole reason
      // the format is line-oriented, and dropping it silently would defeat the feature.
      replay.truncated_tail_ = true;
      const std::string_view tail(text.data() + begin, text.size() - begin);
      if (auto salvaged = salvage_journal_line(tail)) {
        replay.detail_ = "the final journal line was truncated; it named " +
          salvaged->instrument_.symbol_ + " and is treated as unresolved";
        replay.entries_.push_back(std::move(*salvaged));
      } else {
        replay.detail_ =
          "the final journal line was truncated and named no instrument; something was in flight";
        // The same entry a writer's repair appends for this tail, so recovery halts on it whether or
        // not the file was repaired, and whatever else is open beside it.
        replay.entries_.push_back(unattributable_journal_entry(tail));
      }
      break;
    }
    const std::string_view line(text.data() + begin, end - begin);
    begin = end + 1;
    // **An empty line, or a header anywhere but first, is damage.** This writer produces neither: every
    // append is one non-empty line and its newline, and the header is written once, before anything
    // else. Skipping them was how an entry overwritten by a bare newline, or by a copy of the header,
    // vanished from recovery with nothing counted -- a changed line removed rather than detected.
    const auto unreadable = [&replay](const char* why) {
      ++replay.unreadable_lines_;
      if (replay.detail_.empty()) {
        replay.detail_ = why;
      }
    };
    if (line.empty()) {
      unreadable("the journal contains an empty line, which this library never writes");
      continue;
    }
    if (auto scope = journal_header_scope(line)) {
      if (replay.has_header_ || !replay.entries_.empty() || replay.unreadable_lines_ > 0) {
        unreadable("the journal contains a header after its first line, which this library never "
                   "writes");
        continue;
      }
      // Not an instruction and not damage. Recorded so an opener can check ownership before it
      // writes anything at all.
      replay.has_header_ = true;
      replay.scope_ = std::move(*scope);
      // A header that declares checksums protects every entry after it, the first one included.
      seen_checksum = seen_checksum || journal_header_declares_checksums(line);
      continue;
    }
    const bool checksummed = journal_detail::ends_in_checksum(line);
    if (seen_checksum && !checksummed) {
      // **This writer never puts a checksum-less line after a checksummed one, or after a header that
      // declares checksums**, so this is a protected line that lost its checksum -- damage that removed
      // the whole member, which the line's own reader cannot tell from a line written before checksums
      // existed. Only the first checksummed line of a file begun by an earlier release is left without
      // this protection; see `journal_header_line`.
      ++replay.unreadable_lines_;
      if (replay.detail_.empty()) {
        replay.detail_ =
          "a journal line without a checksum follows a checksummed one, so it was changed after it "
          "was written";
      }
      continue;
    }
    seen_checksum = seen_checksum || checksummed;
    if (auto entry = journal_entry_from_line(line)) {
      replay.entries_.push_back(std::move(*entry));
    } else {
      // A complete line that is not a journal entry. Not expected from this writer, so it means
      // the file was edited or written by something else, or a repair was interrupted; counted
      // rather than ignored, and recovery halts on it.
      ++replay.unreadable_lines_;
      if (replay.detail_.empty()) {
        replay.detail_ = "the journal contains a line this library did not write";
      }
    }
  }
  return replay;
}

inline JournalReplay read_journal_file(const std::string& path) {
  return parse_journal_bytes(read_journal_bytes(path));
}

class FileJournal {
public:
  FileJournal(std::string path, std::string scope)
  : path_(std::move(path)), scope_(std::move(scope)), handle_(nullptr, &closer) {
    if (scope_.empty()) {
      // An unnamed journal could not tell another account's file from its own.
      rejection_ = "an order journal must name the broker and account it belongs to";
      return;
    }
#if defined(_WIN32)
    if (!claim_file()) {
      return;
    }
    repair_torn_tail();
    handle_.reset(std::fopen(path_.c_str(), "ab"));
#else
    // Locked before anything reads or repairs the file, and on the descriptor that becomes the
    // write handle: the lock lives exactly as long as `open()` is true, every refusal below closes
    // it, and moving the journal moves it with the handle.
    const int descriptor = ::open(path_.c_str(), O_WRONLY | O_APPEND | O_CREAT | O_CLOEXEC, 0666);
    if (descriptor < 0) {
      const int error = errno;
      rejection_ = "the journal at " + path_ + " could not be opened (" + std::strerror(error) + ")";
      return;
    }
    if (!lock(descriptor)) {
      ::close(descriptor);
      return;
    }
    try {
      if (!claim_file()) {
        ::close(descriptor);
        return;
      }
      repair_torn_tail();
    } catch (...) {
      // A held lock outliving a failed open would refuse this program's own retry.
      ::close(descriptor);
      throw;
    }
    handle_.reset(::fdopen(descriptor, "ab"));
    if (!open()) {
      const int error = errno;
      rejection_ =
        "the journal at " + path_ + " could not be opened for appending (" + std::strerror(error) + ")";
      ::close(descriptor);
    }
#endif
    if (!open()) {
      return;
    }
    if (needs_header_ && !write_header()) {
      return;
    }
    if (!pending_repair_.empty()) {
      finish_repair();
    }
  }

  bool open() const noexcept { return handle_ != nullptr; }

  const std::string& repaired() const noexcept { return repaired_; }

  const std::string& path() const noexcept { return path_; }

  const std::string& scope() const noexcept { return scope_; }

  const std::string& rejection() const noexcept { return rejection_; }

  bool append(const JournalEntry& entry) {
    if (!open()) {
      return false;
    }
    auto stamped = entry;
    // Overwritten: the journal owns the scope.
    stamped.scope_ = scope_;
    auto line = journal_entry_to_line(stamped);
    line += '\n';
    const auto written = std::fwrite(line.data(), 1, line.size(), handle_.get());
    if (written != line.size()) {
      return false;
    }
    return journal_detail::flush_to_disk(handle_.get());
  }

  JournalReplay replay() const { return read_journal_file(path_); }
 
private:
  // Returns whether this journal may write to the file, deciding ownership from the header alone.
  //
  // **The header, not the records, and that distinction is the fix.** Reading ownership from the
  // records leaves a hole exactly where it matters most: the scope is one field among many on a
  // record, so a process killed partway through writing the *first* instruction leaves a file whose
  // only line is torn and unscoped. Another account could then open the path, repair that line into
  // its own journal, and reconcile the first account's instruction against its own rows.
  //
  // A header is written before any instruction is contemplated, so it is complete or the file is
  // empty. Three cases, and the third is the strict one:
  //
  // - **Empty or absent.** Ours to claim; the header is written before the first append.
  // - **Header present.** The scopes must agree.
  // - **Non-empty with no header.** Refused. Something else wrote this file, or it predates the
  //   format, and either way its ownership cannot be established -- which is not a question to
  //   guess at while holding a file that may name another account's live orders.
  bool claim_file() {
    const auto replay = read_journal_file(path_);
    if (!replay.readable()) {
      // Opening for *append* can succeed on a file that cannot be read -- a write-only file is the
      // plain case -- so without this the journal would claim a path whose contents it has never
      // seen, write a header into it, and report nothing owed.
      rejection_ = replay.detail_.empty() ?
        "the journal at " + path_ + " exists and cannot be read" :
        replay.detail_;
      return false;
    }
    const bool empty = !replay.has_header_ && replay.entries_.empty() &&
      replay.unreadable_lines_ == 0 && !replay.truncated_tail_;
    if (empty) {
      needs_header_ = true;
      return true;
    }
    if (!replay.has_header_) {
      rejection_ = "the journal at " + path_ +
        " has no header, so the account it belongs to cannot be established";
      return false;
    }
    if (replay.scope_ != scope_) {
      rejection_ = "the journal at " + path_ + " belongs to " + replay.scope_ + ", not to " +
        scope_;
      return false;
    }
    return true;
  }
 
  // Writes and flushes the header, before any instruction can be recorded.
  bool write_header() {
    auto line = journal_header_line(scope_);
    line += '\n';
    const auto written = std::fwrite(line.data(), 1, line.size(), handle_.get());
    if (written != line.size() || !journal_detail::flush_to_disk(handle_.get())) {
      handle_.reset();
      rejection_ = "the journal header at " + path_ + " could not be written";
      return false;
    }
    needs_header_ = false;
    return true;
  }
 
  // Closes a stream, tolerating the null a failed open leaves behind.
  static void closer(std::FILE* handle) noexcept {
    if (handle != nullptr) {
      std::fclose(handle);
    }
  }
 
  // Reads the file and works out what a torn final line needs, without writing anything yet.
  //
  // Split from the writing half because the file must be inspected *before* it is opened for
  // appending: once a stream is in append mode, everything written goes to the end, which is
  // precisely what must not happen until the fragment is terminated.
  void repair_torn_tail() {
    const auto replay = read_journal_file(path_);
    if (!replay.truncated_tail_) {
      return;
    }
    const auto tail = read_torn_tail(path_);
    auto salvaged = salvage_journal_line(tail);
    pending_repair_ = salvaged ?
      journal_entry_to_line(*salvaged) :
      journal_entry_to_line(unattributable_journal_entry(tail));
    repaired_ = salvaged ?
      "a torn final line was terminated and salvaged as " + salvaged->instrument_.symbol_ :
      "a torn final line was terminated; it named no instrument";
  }
 
  // Terminates the fragment and appends the record that stands for it, durably.
  void finish_repair() {
    // The newline first and on its own: it is what turns the fragment into a complete -- if
    // unreadable -- line, so that the record written next is a line of its own rather than a
    // continuation of a half-written one.
    std::string repair = "\n";
    repair += pending_repair_;
    repair += '\n';
    const auto written = std::fwrite(repair.data(), 1, repair.size(), handle_.get());
    if (written != repair.size() || !journal_detail::flush_to_disk(handle_.get())) {
      // The repair itself failed, so the file is still torn. Refusing to be open is the only safe
      // answer: a caller that cannot journal must not trade, and `open()` is how it finds out.
      handle_.reset();
      repaired_ = "a torn final line could not be repaired";
      rejection_ = "the journal at " + path_ + " ends in a torn line that could not be repaired";
      return;
    }
    pending_repair_.clear();
  }
 
#if !defined(_WIN32)
  // Takes the writer's lock on `descriptor`, or says in `rejection_` why it could not.
  bool lock(int descriptor) {
    if (::flock(descriptor, LOCK_EX | LOCK_NB) == 0) {
      return true;
    }
    const int error = errno;
    rejection_ = error == EWOULDBLOCK ?
      "the journal at " + path_ + " is already open for writing, in another process or another "
      "FileJournal (one journal, one writer)" :
      "the journal at " + path_ + " could not be locked (" + std::strerror(error) +
      "), so a second writer could not be refused";
    return false;
  }
#endif
 
  // Returns the text after the file's last newline, which is the fragment a kill left behind.
  static std::string read_torn_tail(const std::string& path) {
    std::unique_ptr<std::FILE, void (*)(std::FILE*)> reader(
      std::fopen(path.c_str(), "rb"), &closer
    );
    if (reader == nullptr) {
      return {};
    }
    std::string text;
    char buffer[4096];
    while (const auto read = std::fread(buffer, 1, sizeof(buffer), reader.get())) {
      text.append(buffer, read);
    }
    const auto last = text.rfind('\n');
    return last == std::string::npos ? text : text.substr(last + 1);
  }
 
  std::string path_;
  std::string scope_;
  std::unique_ptr<std::FILE, void (*)(std::FILE*)> handle_;
  std::string pending_repair_;
  std::string repaired_;
  std::string rejection_;
  bool needs_header_{false};
};
 
} // namespace runner
} // namespace algotradelib