// Scoring, history/backoff, and the two-phase selection.
//
// The property that matters most here is that scores are *absolute*: a node's
// score must not depend on which other nodes happen to be in the list. The
// switch controller's "beat the incumbent by 20%" rule is meaningless
// otherwise, because the incumbent's score would drift as the pool changed.
#include <asio.hpp>

#include <cstdio>
#include <filesystem>

#include "harness.h"
#include "selector/history.h"
#include "selector/prober.h"
#include "selector/scorer.h"
#include "selector/selector.h"

using namespace ovg;
using namespace ovg::selector;
using ovg::vpngate::Node;
using ovg::vpngate::Proto;
using ovg::vpngate::Remote;

namespace {

Node make_node(const std::string &name, const std::string &ip,
               const std::string &cc, int64_t score, int64_t speed,
               int sessions) {
  Node n;
  n.host_name = name;
  n.ip = ip;
  n.country_short = cc;
  n.country_long = cc;
  n.api.score = score;
  n.api.speed_bps = speed;
  n.api.num_sessions = sessions;
  n.api.uptime_ms = 3LL * 86400 * 1000;
  n.profile = "client\nremote " + ip + " 443 tcp\n";
  n.remotes.push_back(Remote{ip, 443, Proto::Tcp});
  return n;
}

std::string temp_path(const char *leaf) {
  auto p = std::filesystem::temp_directory_path() /
           ("ovg_test_" + std::string(leaf));
  std::error_code ec;
  std::filesystem::remove(p, ec);
  return p.string();
}

}  // namespace

// ---------------------------------------------------------------------------
// Term shapes

OVG_TEST(TermsAreBoundedAndMonotonic) {
  CHECK_EQ(terms::score_term(0), 0.0);
  CHECK_LT(terms::score_term(1000), terms::score_term(1000000));
  CHECK(terms::score_term(1000000000LL) <= 1.0);

  CHECK_EQ(terms::speed_term(0), 0.0);
  CHECK_LT(terms::speed_term(1000000), terms::speed_term(100000000));
  CHECK(terms::speed_term(1000000000000LL) <= 1.0);

  // Fewer sessions is better.
  CHECK_GT(terms::sessions_term(0), terms::sessions_term(20));
  CHECK_GT(terms::sessions_term(20), terms::sessions_term(200));
  CHECK_NEAR(terms::sessions_term(0), 1.0, 1e-9);
  CHECK_NEAR(terms::sessions_term(20), 0.5, 1e-9);

  // Lower RTT is better; unmeasured is mediocre, not zero.
  CHECK_GT(terms::rtt_term(10), terms::rtt_term(200));
  CHECK_NEAR(terms::rtt_term(0), 1.0, 1e-9);
  CHECK_NEAR(terms::rtt_term(100), 0.5, 1e-9);
  CHECK_GT(terms::rtt_term(-1), 0.0);
  CHECK_LT(terms::rtt_term(-1), terms::rtt_term(100));

  CHECK_EQ(terms::uptime_term(0), 0.0);
  CHECK_NEAR(terms::uptime_term(7LL * 86400 * 1000), 1.0, 1e-9);
  CHECK_NEAR(terms::uptime_term(70LL * 86400 * 1000), 1.0, 1e-9);  // clamped
}

OVG_TEST(ScoreIsAbsoluteNotSetRelative) {
  // The same node must score identically whether it is ranked alone or among
  // much better company. This is what makes the switch hysteresis meaningful.
  SelectorConfig cfg;
  Scorer scorer(cfg);
  HistoryStore history("", cfg);

  const Node target = make_node("target", "1.1.1.1", "JP", 500000, 50000000, 30);

  std::vector<Node> alone{target};
  std::vector<Node> crowded{
      target,
      make_node("giant", "2.2.2.2", "JP", 9000000, 900000000, 1),
      make_node("tiny", "3.3.3.3", "JP", 10, 100, 900),
  };

  auto a = scorer.rank_by_prior(alone, history);
  auto b = scorer.rank_by_prior(crowded, history);
  CHECK_EQ(a.size(), size_t(1));
  CHECK_EQ(b.size(), size_t(3));

  double crowded_target = -1;
  for (const auto &s : b)
    if (s.node->host_name == "target") crowded_target = s.score;
  CHECK_NEAR(a[0].score, crowded_target, 1e-12);
}

OVG_TEST(RankOrdersBetterNodesFirst) {
  SelectorConfig cfg;
  Scorer scorer(cfg);
  HistoryStore history("", cfg);

  std::vector<Node> nodes{
      make_node("weak", "1.1.1.1", "JP", 100, 1000000, 400),
      make_node("strong", "2.2.2.2", "JP", 5000000, 500000000, 5),
      make_node("middling", "3.3.3.3", "JP", 200000, 20000000, 60),
  };
  auto ranked = scorer.rank_by_prior(nodes, history);
  CHECK_EQ(ranked.size(), size_t(3));
  CHECK_EQ(ranked[0].node->host_name, std::string("strong"));
  CHECK_EQ(ranked[2].node->host_name, std::string("weak"));
}

OVG_TEST(CountryFiltersApply) {
  SelectorConfig cfg;
  cfg.country_allow = {"JP", "KR"};
  Scorer scorer(cfg);
  HistoryStore history("", cfg);

  std::vector<Node> nodes{
      make_node("jp", "1.1.1.1", "JP", 100, 1000, 1),
      make_node("kr", "2.2.2.2", "KR", 100, 1000, 1),
      make_node("ru", "3.3.3.3", "RU", 900000, 900000000, 1),
  };
  auto ranked = scorer.rank_by_prior(nodes, history);
  CHECK_EQ(ranked.size(), size_t(2));

  SelectorConfig deny;
  deny.country_deny = {"RU"};
  Scorer s2(deny);
  CHECK_EQ(s2.rank_by_prior(nodes, history).size(), size_t(2));
}

OVG_TEST(NodeWithNoRemotesIsFilteredOut) {
  SelectorConfig cfg;
  Scorer scorer(cfg);
  HistoryStore history("", cfg);

  auto broken = make_node("broken", "1.1.1.1", "JP", 900000, 900000000, 1);
  broken.remotes.clear();
  std::vector<Node> nodes{broken};
  CHECK_EQ(scorer.rank_by_prior(nodes, history).size(), size_t(0));
}

OVG_TEST(BackedOffNodesSinkToTheBottom) {
  SelectorConfig cfg;
  Scorer scorer(cfg);
  HistoryStore history("", cfg);

  std::vector<Node> nodes{
      make_node("good", "1.1.1.1", "JP", 5000000, 500000000, 1),
      make_node("meh", "2.2.2.2", "JP", 100, 1000, 300),
  };
  // Fail the strong node repeatedly; it must still be *returned* (as a last
  // resort) but ranked last.
  for (int i = 0; i < 3; ++i) history.record_failure(nodes[0].id());
  CHECK(history.is_backed_off(nodes[0].id()));

  auto ranked = scorer.rank_by_prior(nodes, history);
  CHECK_EQ(ranked.size(), size_t(2));
  CHECK_EQ(ranked[0].node->host_name, std::string("meh"));
  CHECK(ranked[1].backed_off);
}

// ---------------------------------------------------------------------------
// History

OVG_TEST(HistoryUnknownNodeIsNeutral) {
  SelectorConfig cfg;
  HistoryStore h("", cfg);
  const auto s = h.get("nobody@0.0.0.0");
  CHECK_NEAR(s.success_rate(), 0.5, 1e-12);
  CHECK(!h.is_backed_off("nobody@0.0.0.0"));
}

OVG_TEST(HistoryTracksSuccessAndFailure) {
  SelectorConfig cfg;
  HistoryStore h("", cfg);
  h.record_success("n@1", 50);
  h.record_success("n@1", 70);
  h.record_failure("n@1");
  const auto s = h.get("n@1");
  CHECK_EQ(s.successes, uint32_t(2));
  CHECK_EQ(s.failures, uint32_t(1));
  CHECK_EQ(s.consecutive_failures, uint32_t(1));
  CHECK_NEAR(s.success_rate(), 2.0 / 3.0, 1e-12);
  // EWMA sits between the two samples, nearer the recent one.
  CHECK_GT(s.ewma_rtt_ms, 50.0);
  CHECK_LT(s.ewma_rtt_ms, 70.0);
}

OVG_TEST(HistorySuccessClearsConsecutiveFailures) {
  SelectorConfig cfg;
  HistoryStore h("", cfg);
  h.record_failure("n@1");
  h.record_failure("n@1");
  CHECK(h.is_backed_off("n@1"));
  h.record_success("n@1", 20);
  CHECK_EQ(h.get("n@1").consecutive_failures, uint32_t(0));
  CHECK(!h.is_backed_off("n@1"));
}

OVG_TEST(HistoryBackoffGrowsAndIsCapped) {
  SelectorConfig cfg;
  cfg.failure_backoff_initial = Millis(1000);
  cfg.failure_backoff_max = Millis(8000);
  HistoryStore h("", cfg);

  h.record_failure("n@1");
  const auto one = h.backoff_remaining("n@1");
  CHECK_GT(one.count(), int64_t(0));
  CHECK(one.count() <= 1000);

  h.record_failure("n@1");
  CHECK_GT(h.backoff_remaining("n@1").count(), one.count());

  // Far past the cap: must not overflow or explode.
  for (int i = 0; i < 60; ++i) h.record_failure("n@1");
  CHECK(h.backoff_remaining("n@1").count() <= 8000);
  CHECK_GT(h.backoff_remaining("n@1").count(), int64_t(0));
}

OVG_TEST(HistoryRoundTripsThroughDisk) {
  const auto path = temp_path("history.tsv");
  SelectorConfig cfg;
  {
    HistoryStore h(path, cfg);
    h.record_success("alpha@1.1.1.1", 42);
    h.record_failure("beta@2.2.2.2");
    h.record_throughput("alpha@1.1.1.1", 1000000);
    h.save();
  }
  {
    HistoryStore h(path, cfg);
    h.load();
    CHECK_EQ(h.size(), size_t(2));
    const auto a = h.get("alpha@1.1.1.1");
    CHECK_EQ(a.successes, uint32_t(1));
    CHECK_NEAR(a.ewma_rtt_ms, 42.0, 1e-6);
    CHECK_NEAR(a.ewma_throughput_bps, 1000000.0, 1.0);
    CHECK_EQ(h.get("beta@2.2.2.2").failures, uint32_t(1));
  }
  std::filesystem::remove(path);
}

OVG_TEST(HistorySurvivesCorruptLines) {
  const auto path = temp_path("history_corrupt.tsv");
  {
    std::FILE *f = std::fopen(path.c_str(), "w");
    CHECK(f != nullptr);
    std::fputs("# header\n", f);
    std::fputs("good@1.1.1.1 5 1 0 0 1700000000000 33.5 1000\n", f);
    std::fputs("this line is nonsense\n", f);
    std::fputs("also@2.2.2.2 1 0 0 0 1700000000000 12.0 500\n", f);
    std::fclose(f);
  }
  SelectorConfig cfg;
  HistoryStore h(path, cfg);
  h.load();
  // One bad line costs one node, not the file.
  CHECK_EQ(h.size(), size_t(2));
  CHECK_EQ(h.get("good@1.1.1.1").successes, uint32_t(5));
  std::filesystem::remove(path);
}

// ---------------------------------------------------------------------------
// Prober

OVG_TEST(ProberMeasuresLocalListener) {
  asio::io_context io;
  asio::ip::tcp::acceptor acc(io, asio::ip::tcp::endpoint(
                                      asio::ip::make_address("127.0.0.1"), 0));
  acc.listen();
  const uint16_t port = acc.local_endpoint().port();

  // Accept and immediately drop; the prober only times the handshake.
  std::function<void()> accept_one = [&] {
    auto sock = std::make_shared<asio::ip::tcp::socket>(io);
    acc.async_accept(*sock, [sock, &accept_one](std::error_code ec) {
      if (!ec) accept_one();
    });
  };
  accept_one();

  SelectorConfig cfg;
  cfg.probe_samples = 2;
  cfg.probe_timeout = Millis(1000);
  Prober prober(io, cfg);

  std::vector<ProbeResult> got;
  prober.probe({ProbeTarget{"live", "127.0.0.1", port},
                // Port 1 on loopback: nothing listens, connect is refused fast.
                ProbeTarget{"dead", "127.0.0.1", 1}},
               [&](std::vector<ProbeResult> r) {
                 got = std::move(r);
                 acc.close();
               });
  io.run();

  CHECK_EQ(got.size(), size_t(2));
  CHECK_EQ(got[0].node_id, std::string("live"));
  CHECK(got[0].reachable);
  CHECK_EQ(got[0].samples_ok, 2);
  CHECK(got[0].rtt_ms >= 0.0);
  CHECK_EQ(got[1].node_id, std::string("dead"));
  CHECK(!got[1].reachable);
}

OVG_TEST(ProberHandlesEmptyBatch) {
  asio::io_context io;
  SelectorConfig cfg;
  Prober prober(io, cfg);
  bool called = false;
  prober.probe({}, [&](std::vector<ProbeResult> r) {
    called = true;
    CHECK(r.empty());
  });
  io.run();
  CHECK(called);
}

OVG_TEST(ProberTimesOutOnBlackhole) {
  // TEST-NET-1 (RFC 5737) is guaranteed not to be routable on a normal network,
  // so the connect hangs and the timeout path is what completes the probe.
  // Some sandboxes put a transparent proxy in front of all outbound TCP, which
  // accepts everything and makes the case untestable; detect that and skip
  // rather than assert something the environment cannot provide.
  {
    asio::io_context probe_io;
    asio::ip::tcp::socket s(probe_io);
    std::error_code ec = asio::error::would_block;
    s.async_connect(
        asio::ip::tcp::endpoint(asio::ip::make_address("192.0.2.1"), 443),
        [&](std::error_code e) { ec = e; });
    probe_io.run_for(std::chrono::milliseconds(300));
    std::error_code ig;
    s.close(ig);
    if (!ec) SKIP("outbound TCP is transparently proxied here");
  }

  asio::io_context io;
  SelectorConfig cfg;
  cfg.probe_samples = 1;
  cfg.probe_timeout = Millis(150);
  Prober prober(io, cfg);

  std::vector<ProbeResult> got;
  prober.probe({ProbeTarget{"blackhole", "192.0.2.1", 443}},
               [&](std::vector<ProbeResult> r) { got = std::move(r); });
  io.run();

  CHECK_EQ(got.size(), size_t(1));
  CHECK(!got[0].reachable);
}

// ---------------------------------------------------------------------------
// Selector, end to end over the real captured feed

namespace {

// A NodeStore primed from the sample CSV via its disk cache. The API URL points
// at a dead port so no network fetch can succeed, and we never run the
// io_context far enough for one to be attempted.
std::unique_ptr<vpngate::NodeStore> primed_store(asio::io_context &io) {
  VpnGateConfig vg;
  vg.api_urls = {"http://127.0.0.1:1/"};
  vg.cache_path = ovgtest::data_path("vpngate_sample.csv");
  vg.cache_max_age = std::chrono::hours(24 * 3650);
  auto store = std::make_unique<vpngate::NodeStore>(io, vg);
  store->start();
  return store;
}

}  // namespace

OVG_TEST(SelectorRanksTheRealFeedWithoutProbing) {
  asio::io_context io;
  auto store = primed_store(io);
  CHECK(store->has_nodes());

  SelectorConfig cfg;
  HistoryStore history("", cfg);
  Selector sel(io, cfg, *store, history);

  SelectRequest req;
  req.want = 5;
  req.probe = false;

  std::vector<Candidate> got;
  bool called = false;
  sel.select(req, [&](std::vector<Candidate> c) {
    called = true;
    got = std::move(c);
  });

  CHECK(called);  // the no-probe path must complete synchronously
  CHECK_EQ(got.size(), size_t(5));
  // Ordered best-first, and each carries the profile the tunnel will need.
  for (size_t i = 1; i < got.size(); ++i) CHECK(got[i - 1].score >= got[i].score);
  for (const auto &c : got) {
    CHECK(!c.node.profile.empty());
    CHECK(!c.node.remotes.empty());
    CHECK(!c.probed);
    CHECK(c.reachable);
  }
  store->stop();
}

OVG_TEST(SelectorHonoursExclusions) {
  asio::io_context io;
  auto store = primed_store(io);

  SelectorConfig cfg;
  HistoryStore history("", cfg);
  Selector sel(io, cfg, *store, history);

  SelectRequest first;
  first.want = 3;
  first.probe = false;
  std::vector<Candidate> a;
  sel.select(first, [&](std::vector<Candidate> c) { a = std::move(c); });
  CHECK_EQ(a.size(), size_t(3));

  // Exclude the winner -- this is what the switch controller does with the
  // incumbent and with anything already draining.
  SelectRequest second;
  second.want = 3;
  second.probe = false;
  second.exclude_ids = {a[0].node.id()};
  std::vector<Candidate> b;
  sel.select(second, [&](std::vector<Candidate> c) { b = std::move(c); });

  CHECK_EQ(b.size(), size_t(3));
  for (const auto &c : b) CHECK_NE(c.node.id(), a[0].node.id());
  CHECK_EQ(b[0].node.id(), a[1].node.id());

  store->stop();
}

OVG_TEST(SelectorRescoreMatchesRanking) {
  // The hysteresis check compares a rescored incumbent against fresh
  // candidates, so the two paths must agree for an unprobed node.
  asio::io_context io;
  auto store = primed_store(io);

  SelectorConfig cfg;
  HistoryStore history("", cfg);
  Selector sel(io, cfg, *store, history);

  SelectRequest req;
  req.want = 1;
  req.probe = false;
  std::vector<Candidate> got;
  sel.select(req, [&](std::vector<Candidate> c) { got = std::move(c); });
  CHECK_EQ(got.size(), size_t(1));

  const auto again = sel.rescore(got[0].node.id());
  CHECK(again.has_value());
  CHECK_NEAR(again->score, got[0].score, 1e-12);

  CHECK(!sel.rescore("no-such-node@0.0.0.0").has_value());
  store->stop();
}

OVG_TEST(SelectorReportsNothingWhenNodeListIsEmpty) {
  asio::io_context io;
  VpnGateConfig vg;
  vg.api_urls = {"http://127.0.0.1:1/"};
  vg.cache_path = "";  // no cache, no network => no nodes
  vpngate::NodeStore store(io, vg);

  SelectorConfig cfg;
  HistoryStore history("", cfg);
  Selector sel(io, cfg, store, history);

  bool called = false;
  sel.select({}, [&](std::vector<Candidate> c) {
    called = true;
    CHECK(c.empty());
  });
  CHECK(called);
}

OVG_TEST(ProberRejectsNonLiteralHost) {
  // Node hosts come from the .ovpn profile and are IP literals in practice.
  // A name must be reported unreachable, not silently resolved -- resolving it
  // here would be a DNS lookup outside the tunnel.
  asio::io_context io;
  SelectorConfig cfg;
  Prober prober(io, cfg);

  std::vector<ProbeResult> got;
  prober.probe({ProbeTarget{"named", "vpn.example.com", 443}},
               [&](std::vector<ProbeResult> r) { got = std::move(r); });
  io.run();

  CHECK_EQ(got.size(), size_t(1));
  CHECK(!got[0].reachable);
  CHECK_EQ(got[0].samples_ok, 0);
}