// Health monitor and switch controller.
//
// The monitor is scored arithmetic over an egress it does not own, so the whole
// module can be exercised against a fake egress with no tunnel, no sockets and
// no waiting: the probe outcome, the byte counters and the connect ledger are
// all set directly by the test. What is worth testing here is the scoring
// policy, not asio -- specifically the three things that are easy to get
// backwards:
//
//   * a signal that is *missing* (the direct egress keeps no byte counters)
//     must renormalise out of the score rather than count as a failure;
//   * a failed probe must not be averaged back into "fine" by healthy-looking
//     counters around it;
//   * a sustained-unhealthy report fires once per degradation, not once per
//     round, or every interval queues another switch request for a decision
//     already taken.
#include <asio.hpp>

#include <atomic>
#include <chrono>
#include <memory>
#include <string>
#include <vector>

#include "common/config.h"
#include "common/error.h"
#include "egress/egress.h"
#include "egress/egress_manager.h"
#include "harness.h"
#include "health/health_monitor.h"
#include "health/switch_controller.h"

using namespace ovg;
using namespace std::chrono_literals;

namespace {

bool run_until(asio::io_context &io, const std::function<bool()> &pred,
               std::chrono::milliseconds limit) {
  const auto deadline = std::chrono::steady_clock::now() + limit;
  while (std::chrono::steady_clock::now() < deadline) {
    if (pred()) return true;
    io.run_for(5ms);
    io.restart();
  }
  return pred();
}

// An egress whose every health input is a public member. `connect_result`
// decides what the probe sees; `stats_` is whatever the test wants reported.
class ProbeEgress final : public egress::Egress {
 public:
  explicit ProbeEgress(asio::io_context &io) : io_(io), label_("fake-node") {
    stats_.node_id = "fake-node";
    stats_.proto = "fake";
  }

  void async_connect_tcp(const asio::any_io_executor &ex, const Endpoint &ep,
                         Millis, ConnectHandler h) override {
    probes.fetch_add(1);
    last_target = ep.to_string();
    if (hang) return;  // handler dropped on purpose: exercises the guard timer
    auto ec = connect_ec;
    asio::post(ex, [h = std::move(h), ec]() mutable { h(ec, nullptr); });
  }
  void async_bind_udp(const asio::any_io_executor &ex,
                      UdpBindHandler h) override {
    asio::post(ex, [h = std::move(h)]() mutable {
      h(make_error_code(Error::NotSupported), nullptr);
    });
  }
  void async_resolve(const asio::any_io_executor &ex, const std::string &,
                     ResolveHandler h) override {
    asio::post(ex, [h = std::move(h)]() mutable {
      h(make_error_code(Error::NotSupported), {});
    });
  }

  egress::EgressState state() const override { return state_.load(); }
  egress::EgressStats stats() const override { return stats_; }
  std::string detail() const override { return detail_; }
  void begin_drain() override {}
  void shutdown(std::function<void()> on_done) override {
    state_.store(egress::EgressState::Down);
    if (on_done) asio::post(io_, std::move(on_done));
  }
  const std::string &label() const override { return label_; }

  std::error_code connect_ec{};
  bool hang = false;
  std::atomic<int> probes{0};
  std::string last_target;
  std::atomic<egress::EgressState> state_{egress::EgressState::Ready};
  egress::EgressStats stats_;
  std::string detail_;

 private:
  asio::io_context &io_;
  std::string label_;
};

// A monitor over one fake egress, with the timing wound down so a "sustained"
// verdict takes milliseconds instead of a minute.
//
// The config is copied into the monitor at construction, so `tweak` is where a
// test changes it -- there is no setter afterwards, by design.
struct Fixture {
  asio::io_context io;
  std::shared_ptr<ProbeEgress> eg = std::make_shared<ProbeEgress>(io);
  HealthConfig cfg;
  std::unique_ptr<health::HealthMonitor> mon;
  std::vector<std::string> unhealthy;
  bool provide = true;

  explicit Fixture(int windows = 2,
                   const std::function<void(HealthConfig &)> &tweak = {}) {
    cfg.interval = 20ms;
    cfg.probe_timeout = 200ms;
    cfg.unhealthy_windows = windows;
    cfg.probe_domain = "probe.invalid";
    cfg.probe_port = 8080;
    if (tweak) tweak(cfg);
    mon = std::make_unique<health::HealthMonitor>(
        io, cfg, [this]() -> egress::EgressPtr {
          return provide ? eg : nullptr;
        });
    mon->set_on_unhealthy(
        [this](const std::string &why) { unhealthy.push_back(why); });
  }
  ~Fixture() {
    mon->stop();
    io.run_for(50ms);
  }

  // Waits for `n` completed rounds rather than for wall-clock time.
  bool rounds(uint64_t n, std::chrono::milliseconds limit = 3s) {
    return run_until(io, [this, n] { return mon->rounds() >= n; }, limit);
  }
  // Waits for `n` more rounds than have already happened.
  bool more_rounds(uint64_t n, std::chrono::milliseconds limit = 3s) {
    return rounds(mon->rounds() + n, limit);
  }
};

}  // namespace

// ---------------------------------------------------------------------------
// Scoring
// ---------------------------------------------------------------------------

OVG_TEST(health_scores_a_working_egress_healthy) {
  Fixture f;
  f.mon->start();
  CHECK(f.rounds(1));

  const auto s = f.mon->last();
  CHECK(s.egress_present);
  CHECK(s.tunnel_up);
  CHECK(s.probe_ok);
  CHECK(s.healthy);
  // Three terms available (rtt, connect, loss), all near perfect against a fake
  // that answers instantly. Anything below 0.9 means a term scored a missing
  // input as a bad one.
  CHECK_GT(s.score, 0.9);
  CHECK(f.unhealthy.empty());
}

OVG_TEST(health_probes_the_configured_host_and_port) {
  Fixture f;
  f.mon->start();
  CHECK(f.rounds(1));
  // The probe target is a TCP handshake to probe_domain:probe_port. A DNS
  // lookup here would be answered from cache without crossing the tunnel.
  CHECK_EQ(f.eg->last_target, std::string("probe.invalid:8080"));
}

OVG_TEST(health_renormalises_around_a_missing_stall_signal) {
  Fixture f;
  f.mon->start();
  CHECK(f.rounds(1));

  // No byte counters at all -- the direct egress's situation.
  const auto without = f.mon->last();
  CHECK(!without.stall_known);

  // Now give it traffic, which makes the stall term available and *not* stalled.
  f.eg->stats_.tun_bytes_out = 4096;
  f.eg->stats_.tcp_active = 1;
  CHECK(f.more_rounds(2));

  const auto with = f.mon->last();
  CHECK(with.stall_known);
  // Adding a *satisfied* term must not move a healthy score materially. If the
  // missing term had been scoring zero, this would jump by ~0.2.
  CHECK_NEAR(with.score, without.score, 0.05);
}

OVG_TEST(health_caps_the_score_when_the_probe_fails) {
  Fixture f;
  // Everything else looks perfect: no drops, no failed connects, bytes moving.
  f.eg->stats_.tx_packets = 1000;
  f.eg->stats_.rx_packets = 1000;
  f.eg->stats_.tcp_opened = 100;
  f.eg->connect_ec = make_error_code(Error::Timeout);

  f.mon->start();
  CHECK(f.rounds(1));

  const auto s = f.mon->last();
  CHECK(!s.probe_ok);
  CHECK(!s.healthy);
  // The hard override, not the blend: healthy counters must not average a dead
  // tunnel back up to passing.
  CHECK(s.score <= f.cfg.min_score / 2.0);
  CHECK(s.verdict.find("probe failed") != std::string::npos);
}

OVG_TEST(health_caps_the_score_on_a_high_connect_failure_rate) {
  Fixture f;
  f.eg->stats_.tcp_opened = 10;
  f.eg->stats_.tcp_failed = 90;  // 90% failing, ceiling is 50%

  f.mon->start();
  CHECK(f.rounds(1));

  const auto s = f.mon->last();
  CHECK(s.probe_ok);  // the probe itself was fine
  CHECK(!s.healthy);
  CHECK(s.score <= f.cfg.min_score / 2.0);
  CHECK(s.verdict.find("connect failure rate") != std::string::npos);
}

OVG_TEST(health_reports_a_stall_only_with_streams_in_flight) {
  Fixture f(2, [](HealthConfig &c) { c.stall_threshold = 1ms; });

  // Bytes have moved once, so the signal exists, and now they stop.
  f.eg->stats_.tun_bytes_out = 1024;
  f.mon->start();
  CHECK(f.rounds(2));

  // Idle: no live streams, so a frozen counter is not a stall. A proxy with no
  // clients is the normal overnight state and must not switch nodes over it.
  CHECK(f.mon->last().stall_known);
  CHECK_EQ(f.mon->last().stalled_ms, int64_t(0));

  // Same frozen counter, but now something is waiting on it.
  f.eg->stats_.tcp_active = 3;
  CHECK(f.more_rounds(3));
  const auto s = f.mon->last();
  CHECK(s.stall_known);
  CHECK_GT(s.stalled_ms, int64_t(0));
  CHECK(s.verdict.find("no byte progress") != std::string::npos);
}

// ---------------------------------------------------------------------------
// Availability of the egress itself
// ---------------------------------------------------------------------------

OVG_TEST(health_records_a_sample_when_there_is_no_egress) {
  Fixture f;
  f.provide = false;
  f.mon->start();
  CHECK(f.rounds(1));

  const auto s = f.mon->last();
  CHECK(!s.egress_present);
  CHECK(!s.tunnel_up);
  CHECK(!s.healthy);
  CHECK_EQ(s.score, 0.0);
  CHECK_EQ(s.egress_label, std::string(""));
}

OVG_TEST(health_does_not_probe_an_egress_that_is_not_ready) {
  Fixture f;
  f.eg->state_.store(egress::EgressState::Connecting);
  f.eg->detail_ = "handshaking";
  f.mon->start();
  CHECK(f.rounds(2));

  // Probing a tunnel that is known to be down measures the timeout and nothing
  // else; the verdict is already known.
  CHECK_EQ(f.eg->probes.load(), 0);
  const auto s = f.mon->last();
  CHECK(s.egress_present);
  CHECK(!s.tunnel_up);
  CHECK(!s.healthy);
  CHECK(s.verdict.find("handshaking") != std::string::npos);
}

OVG_TEST(health_gives_up_on_a_probe_whose_handler_never_returns) {
  Fixture f(2, [](HealthConfig &c) { c.probe_timeout = 50ms; });
  f.eg->hang = true;  // handler simply dropped, as a wedged backend would

  f.mon->start();
  // The guard fires at probe_timeout + 2s. Without it the monitor wedges here
  // forever, having handed its only in-flight slot to a handler that is never
  // coming back -- and a monitor that has stopped sampling reports the last
  // thing it saw, which was healthy.
  CHECK(f.rounds(1, 5s));
  const auto s = f.mon->last();
  CHECK(!s.probe_ok);
  CHECK(s.verdict.find("abandoned") != std::string::npos);

  // And it recovers: the slot is released, so the next round runs.
  f.eg->hang = false;
  CHECK(f.rounds(2, 5s));
  CHECK(f.mon->last().probe_ok);
}

// ---------------------------------------------------------------------------
// Sustained-unhealthy reporting
// ---------------------------------------------------------------------------

OVG_TEST(health_reports_unhealthy_only_after_consecutive_windows) {
  Fixture f(3);
  f.eg->connect_ec = make_error_code(Error::Timeout);
  f.mon->start();

  CHECK(f.rounds(2));
  // Two bad windows out of three required: one bad sample on a volunteer tunnel
  // in another country is weather, not a failure.
  CHECK(f.unhealthy.empty());

  CHECK(f.rounds(3));
  CHECK(run_until(f.io, [&] { return !f.unhealthy.empty(); }, 2s));
  CHECK_EQ(f.unhealthy.size(), size_t(1));
  CHECK(f.unhealthy[0].find("fake-node") != std::string::npos);
}

OVG_TEST(health_reports_once_per_degradation_not_once_per_round) {
  Fixture f(2);
  f.eg->connect_ec = make_error_code(Error::Timeout);
  f.mon->start();

  CHECK(f.rounds(9));
  // Nine bad rounds at a 2-window threshold. The counter resets on each report,
  // so this is four or five reports -- not eight. If it did not reset, every
  // round past the second would queue another switch request.
  CHECK_GT(f.unhealthy.size(), size_t(1));
  CHECK(f.unhealthy.size() <= 5);
}

OVG_TEST(health_recovery_clears_the_consecutive_counter) {
  Fixture f(3);
  f.eg->connect_ec = make_error_code(Error::Timeout);
  f.mon->start();
  CHECK(f.rounds(2));
  CHECK_EQ(f.mon->consecutive_bad(), 2);

  f.eg->connect_ec = {};
  CHECK(f.more_rounds(2));
  CHECK(run_until(f.io, [&] { return f.mon->consecutive_bad() == 0; }, 2s));
  CHECK(f.unhealthy.empty());
}

OVG_TEST(health_keeps_a_bounded_history) {
  Fixture f;
  f.mon->start();
  CHECK(f.rounds(24, 5s));
  // kHistoryDepth is 20; ask for more and get what exists, never more.
  CHECK_EQ(f.mon->recent(50).size(), size_t(20));
  CHECK_EQ(f.mon->recent(5).size(), size_t(5));
  // recent() is oldest-first, so the last element is the newest sample.
  const auto r = f.mon->recent(5);
  CHECK(r.front().age_ms >= r.back().age_ms);
}

// ---------------------------------------------------------------------------
// Switch controller
// ---------------------------------------------------------------------------

OVG_TEST(switch_controller_reports_why_the_manager_declined) {
  asio::io_context io;
  Config cfg;
  cfg.egress_mode = "direct";
  cfg.health.interval = 1h;  // no rounds of its own during this test

  egress::EgressManager mgr(io, cfg, nullptr, nullptr, nullptr);
  health::HealthMonitor mon(io, cfg.health,
                            [&]() -> egress::EgressPtr { return nullptr; });
  health::SwitchController sw(io, cfg, mon, mgr);

  std::string detail;
  const bool ok = sw.force_switch("test", &detail);
  CHECK(!ok);
  // The specific reason, not a guess. A direct-mode build reporting "a switch
  // is already in progress" sends an operator looking for a switch that was
  // never attempted.
  CHECK(detail.find("direct") != std::string::npos);
  CHECK(detail.find("no tunnel") != std::string::npos);

  const auto st = sw.stats();
  CHECK_EQ(st.manual, uint64_t(1));
  CHECK_EQ(st.requested, uint64_t(1));
  CHECK_EQ(st.declined, uint64_t(1));
  CHECK(st.last_trigger.find("manual") != std::string::npos);

  bool done = false;
  mgr.shutdown([&] { done = true; });
  run_until(io, [&] { return done; }, 2s);
}

OVG_TEST(switch_controller_forwards_a_sustained_unhealthy_verdict) {
  asio::io_context io;
  Config cfg;
  cfg.egress_mode = "direct";  // every request is declined, but still counted
  cfg.health.interval = 20ms;
  cfg.health.unhealthy_windows = 2;
  cfg.health.probe_timeout = 100ms;

  auto eg = std::make_shared<ProbeEgress>(io);
  eg->connect_ec = make_error_code(Error::Timeout);

  egress::EgressManager mgr(io, cfg, nullptr, nullptr, nullptr);
  health::HealthMonitor mon(io, cfg.health,
                            [&]() -> egress::EgressPtr { return eg; });
  health::SwitchController sw(io, cfg, mon, mgr);

  sw.start();  // must subscribe before the monitor runs, or a verdict is lost
  mon.start();

  CHECK(run_until(io, [&] { return sw.stats().unhealthy > 0; }, 3s));
  const auto st = sw.stats();
  CHECK_GT(st.requested, uint64_t(0));
  CHECK_EQ(st.tunnel_down + st.opportunistic, uint64_t(0));
  CHECK(st.last_trigger.find("unhealthy") != std::string::npos);

  mon.stop();
  sw.stop();
  bool done = false;
  mgr.shutdown([&] { done = true; });
  run_until(io, [&] { return done; }, 2s);
}