// secs_bench — performance baseline harness.
//
// Spins up an in-process passive equipment (single io_context, single
// thread) plus an active host, runs a series of canned workloads, and
// emits a markdown table with throughput / latency / memory numbers.
// Intended use:
//   build/secs_bench --requests 50000 --concurrency 32 > bench.md
// then diff bench.md across commits to track regressions.
//
// Scenarios:
//   * S1F1/F2 — header-only round-trip; measures dispatch + framing
//   * S1F3/F4 — N-element SVID list; measures encode + decode throughput
//   * S6F11   — equipment-initiated event report (W=0); measures push
//   * Stores  — populate N PJ+CJ pairs, measure RSS delta
//
// Latency percentiles use a simple in-memory vector + std::nth_element;
// for the default 50k requests that's a 200 KB allocation, negligible
// next to the wire traffic itself.

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

#include "secsgem/endpoint.hpp"
#include "secsgem/gem/data_model.hpp"
#include "secsgem/gem/messages.hpp"
#include "secsgem/gem/router.hpp"
#include "secsgem/secs2/item.hpp"
#include "secsgem/secs2/message.hpp"

#ifdef __APPLE__
#include <mach/mach.h>
#elif defined(__linux__)
#include <fstream>
#endif

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

namespace {

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

// Resident set size in MiB.  Best-effort per platform; macOS uses
// mach_task_basic_info, Linux reads /proc/self/statm.
double rss_mib() {
#ifdef __APPLE__
  mach_task_basic_info info;
  mach_msg_type_number_t count = MACH_TASK_BASIC_INFO_COUNT;
  if (task_info(mach_task_self(), MACH_TASK_BASIC_INFO,
                reinterpret_cast<task_info_t>(&info), &count) == KERN_SUCCESS) {
    return static_cast<double>(info.resident_size) / (1024.0 * 1024.0);
  }
  return 0.0;
#elif defined(__linux__)
  std::ifstream in("/proc/self/statm");
  long pages = 0, rss_pages = 0;
  in >> pages >> rss_pages;
  const long page_kb = sysconf(_SC_PAGESIZE) / 1024;
  return static_cast<double>(rss_pages * page_kb) / 1024.0;
#else
  return 0.0;
#endif
}

struct LatencyStats {
  double p50_us = 0, p95_us = 0, p99_us = 0, max_us = 0;
};

LatencyStats summarize(std::vector<double>& samples_us) {
  LatencyStats s;
  if (samples_us.empty()) return s;
  auto at = [&](double pct) {
    auto n = static_cast<std::size_t>(samples_us.size() * pct);
    if (n >= samples_us.size()) n = samples_us.size() - 1;
    std::nth_element(samples_us.begin(), samples_us.begin() + n,
                     samples_us.end());
    return samples_us[n];
  };
  s.p50_us = at(0.50);
  s.p95_us = at(0.95);
  s.p99_us = at(0.99);
  s.max_us = *std::max_element(samples_us.begin(), samples_us.end());
  return s;
}

struct ScenarioResult {
  std::string name;
  std::size_t ops = 0;
  double seconds = 0;
  LatencyStats latency;
};

double tx_per_s(const ScenarioResult& r) {
  return r.seconds > 0 ? r.ops / r.seconds : 0;
}

// Boots an in-process server + client pair pinned to localhost on an
// OS-allocated port, drives the body of a scenario, then tears down.
struct Harness {
  asio::io_context io;
  std::shared_ptr<Server> server;
  std::shared_ptr<Client> client;
  std::shared_ptr<Connection> server_conn;
  std::shared_ptr<Connection> client_conn;
  bool server_selected = false;
  bool client_selected = false;

  void run_until(std::function<bool()> done,
                 std::chrono::seconds budget = 60s) {
    asio::steady_timer cap(io);
    cap.expires_after(budget);
    cap.async_wait([&](std::error_code ec) {
      if (!ec) {
        std::cerr << "bench: budget exceeded\n";
        io.stop();
      }
    });
    asio::steady_timer poll(io);
    std::function<void(std::error_code)> tick = [&](std::error_code ec) {
      if (ec) return;
      if (done()) { io.stop(); return; }
      poll.expires_after(1ms);
      poll.async_wait(tick);
    };
    poll.expires_after(1ms);
    poll.async_wait(tick);
    io.run();
    io.restart();
  }
};

void bring_up(Harness& h, gem::Router& router) {
  // OS-allocated port to avoid collisions with the demo server.
  asio::ip::tcp::acceptor probe(h.io, asio::ip::tcp::endpoint(
                                          asio::ip::address_v4::loopback(), 0));
  const auto port = probe.local_endpoint().port();
  probe.close();

  Server::Config sc;
  sc.port = port;
  sc.device_id = 0;
  h.server = std::make_shared<Server>(h.io, sc);
  h.server->on_connection([&h, &router](std::shared_ptr<Connection> conn) {
    h.server_conn = conn;
    conn->set_message_handler([&router](const s2::Message& msg) {
      return router.dispatch(msg);
    });
    conn->set_selected_handler([&h] { h.server_selected = true; });
  });
  h.server->start();

  Client::Config cc;
  cc.host = "127.0.0.1";
  cc.port = port;
  cc.timers.linktest = 0ms;
  h.client = std::make_shared<Client>(h.io, cc);
  h.client->on_connection([&h](std::shared_ptr<Connection> conn) {
    h.client_conn = conn;
    conn->set_selected_handler([&h] { h.client_selected = true; });
  });
  h.client->start();

  // Pump until both ends are SELECTED.  send_request before SELECT
  // queues the frame, but timing measurements assume an established
  // session.
  h.run_until([&] { return h.server_selected && h.client_selected; }, 10s);
  if (!h.server_selected || !h.client_selected)
    throw std::runtime_error("failed to bring up server/client");
}

// Bench: round-trip a message N times with a fixed in-flight window.
ScenarioResult bench_roundtrip(const std::string& name,
                                std::size_t requests,
                                std::size_t concurrency,
                                std::function<s2::Message()> build,
                                gem::Router& router) {
  Harness h;
  bring_up(h, router);

  std::vector<double> samples;
  samples.reserve(requests);
  std::size_t completed = 0, dispatched = 0;
  const auto start = std::chrono::steady_clock::now();

  std::function<void()> issue = [&]() {
    while (dispatched < requests &&
           (dispatched - completed) < concurrency) {
      const auto t0 = std::chrono::steady_clock::now();
      ++dispatched;
      h.client_conn->send_request(
          build(),
          [&samples, &completed, &issue, t0](std::error_code ec,
                                              const s2::Message&) {
            if (ec) {
              std::cerr << "bench: ec=" << ec.message() << "\n";
              return;
            }
            const auto t1 = std::chrono::steady_clock::now();
            samples.push_back(
                std::chrono::duration<double, std::micro>(t1 - t0).count());
            ++completed;
            issue();
          });
    }
  };
  asio::post(h.io, issue);
  h.run_until([&] { return completed >= requests; }, 120s);

  const auto elapsed = std::chrono::steady_clock::now() - start;
  ScenarioResult r;
  r.name = name;
  r.ops = completed;
  r.seconds = std::chrono::duration<double>(elapsed).count();
  r.latency = summarize(samples);
  return r;
}

// Bench: equipment pushes W=0 primaries to host as fast as possible.
ScenarioResult bench_push(const std::string& name, std::size_t requests) {
  gem::Router noop;
  Harness h;
  bring_up(h, noop);

  std::size_t received = 0;
  h.client_conn->set_message_handler(
      [&received](const s2::Message& msg) -> std::optional<s2::Message> {
        if (msg.stream == 6 && msg.function == 11) ++received;
        return std::nullopt;  // W=0 push, no reply
      });

  const auto start = std::chrono::steady_clock::now();
  for (std::size_t i = 0; i < requests; ++i) {
    auto msg = gem::s6f11_event_report(
        static_cast<uint32_t>(i), /*ceid=*/300, /*reports=*/{});
    msg.reply_expected = false;
    h.server_conn->send_data(std::move(msg));
  }
  h.run_until([&] { return received >= requests; }, 120s);
  const auto elapsed = std::chrono::steady_clock::now() - start;

  ScenarioResult r;
  r.name = name;
  r.ops = received;
  r.seconds = std::chrono::duration<double>(elapsed).count();
  return r;
}

// Bench: how much RSS does the model grow with N active PJ+CJ pairs?
double bench_store_memory(std::size_t pairs) {
  auto model = std::make_shared<gem::EquipmentDataModel>();
  model->process_jobs.set_table_factory([] {
    return gem::ProcessJobTransitionTable{};
  });
  model->control_jobs.set_table_factory([] {
    return gem::ControlJobTransitionTable{};
  });
  const double before = rss_mib();
  for (std::size_t i = 0; i < pairs; ++i) {
    const auto pj = "PJ-" + std::to_string(i);
    const auto cj = "CJ-" + std::to_string(i);
    model->process_jobs.create(pj, "RECIPE-A", {"W1", "W2"});
    model->control_jobs.create(cj, {pj},
                                [&](const std::string& id) {
                                  return model->process_jobs.has(id);
                                });
  }
  const double after = rss_mib();
  return after - before;
}

void emit_markdown_row(const ScenarioResult& r) {
  std::printf("| %-32s | %7zu | %7.2f | %10.0f | %7.1f | %7.1f | %7.1f | %7.1f |\n",
              r.name.c_str(), r.ops, r.seconds, tx_per_s(r),
              r.latency.p50_us, r.latency.p95_us, r.latency.p99_us,
              r.latency.max_us);
}

}  // namespace

int main(int argc, char** argv) {
  const auto requests =
      static_cast<std::size_t>(std::stoi(arg(argc, argv, "--requests", "20000")));
  const auto concurrency =
      static_cast<std::size_t>(std::stoi(arg(argc, argv, "--concurrency", "16")));
  const auto svid_count =
      static_cast<std::size_t>(std::stoi(arg(argc, argv, "--svid-count", "32")));
  const auto store_pairs =
      static_cast<std::size_t>(std::stoi(arg(argc, argv, "--store-pairs", "1000")));

  std::printf("# secs-gem performance baseline\n\n");
  std::printf("Single-threaded io_context, loopback TCP, MacBook-class machine.\n");
  std::printf("Re-run: `build/secs_bench --requests %zu --concurrency %zu`\n\n",
              requests, concurrency);

  std::printf("## Round-trip throughput / latency\n\n");
  std::printf("| Scenario                         | Ops     | Elapsed |   Ops/sec  |  p50 us |  p95 us |  p99 us |  max us |\n");
  std::printf("|----------------------------------|--------:|--------:|-----------:|--------:|--------:|--------:|--------:|\n");

  {
    gem::Router router;
    router.on(1, 1, [](const s2::Message&) {
      return gem::s1f2_on_line_data("BENCH", "1.0");
    });
    auto r = bench_roundtrip("S1F1/F2 (header-only)", requests, concurrency,
                              [] { return s2::Message(1, 1, true); }, router);
    emit_markdown_row(r);
  }

  {
    gem::Router router;
    router.on(1, 3, [svid_count](const s2::Message&) {
      std::vector<std::optional<s2::Item>> values;
      values.reserve(svid_count);
      for (std::size_t i = 0; i < svid_count; ++i)
        values.push_back(s2::Item::u4(static_cast<uint32_t>(i)));
      return gem::s1f4_selected_status_data(values);
    });
    std::vector<uint32_t> svids(svid_count);
    for (std::size_t i = 0; i < svid_count; ++i)
      svids[i] = static_cast<uint32_t>(i);
    auto r = bench_roundtrip(
        "S1F3/F4 (" + std::to_string(svid_count) + " SVIDs)",
        requests, concurrency,
        [&svids] { return gem::s1f3_selected_status_request(svids); },
        router);
    emit_markdown_row(r);
  }

  {
    auto r = bench_push("S6F11 push (W=0)", requests);
    emit_markdown_row(r);
  }

  std::printf("\n## Memory footprint\n\n");
  std::printf("| Scenario                              | Pairs   | RSS delta (MiB) | Bytes/pair |\n");
  std::printf("|---------------------------------------|--------:|----------------:|-----------:|\n");
  const auto delta = bench_store_memory(store_pairs);
  const double bytes_per = store_pairs > 0
                                ? (delta * 1024 * 1024) / store_pairs
                                : 0.0;
  std::printf("| %-37s | %7zu | %15.2f | %10.0f |\n",
              "PJ + CJ pair, no persistence",
              store_pairs, delta, bytes_per);

  std::printf("\n_Numbers are single-sample; variance can be ±20%% on the same\n"
              "hardware between runs.  For regression tracking, compare medians\n"
              "across N runs, not single values._\n");
  return 0;
}