#include "Memory_Monitor.h"
#include <algorithm>
#include <chrono>
#include <cmath>
#include <fstream>
#include <iterator>
#include <mimalloc.h>
#include <sstream>
#include "Memory_Resource.h"
namespace {
constexpr std::uint64_t kibibyte = 1024;
constexpr std::uint64_t mebibyte = 1024 * kibibyte;
constexpr std::size_t history_capacity = 600;
constexpr std::size_t response_history_capacity = 120;
constexpr std::uint64_t growth_warmup_ms = 10 * 60 * 1000;
constexpr std::uint64_t growth_window_ms = 15 * 60 * 1000;
constexpr std::uint64_t recent_growth_window_ms = 5 * 60 * 1000;
constexpr std::uint64_t minimum_growth_window_ms = 10 * 60 * 1000;
constexpr std::uint64_t minimum_recent_growth_window_ms = 4 * 60 * 1000;
std::uint64_t now_milliseconds() {
    return static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::milliseconds>(
    std::chrono::system_clock::now().time_since_epoch()).count());
}
std::uint64_t parse_kibibytes(std::string_view line) {
    const auto separator = line.find(':');
    if (separator == std::string_view::npos)
        return 0;
    std::uint64_t value{};
    std::istringstream stream(std::string(line.substr(separator + 1)));
    stream >> value;
    return value * kibibyte;
}
#if defined(ECAP_TARGET_LINUX)
void read_linux_process_memory(Process_Memory_Snapshot& result) {
    std::ifstream stream("/proc/self/status");
    std::string line;
    while (std::getline(stream, line)) {
        if (line.starts_with("VmRSS:"))
            result.rss_bytes = parse_kibibytes(line);
        else if (line.starts_with("VmHWM:"))
            result.peak_rss_bytes = parse_kibibytes(line);
        else if (line.starts_with("RssAnon:"))
            result.anonymous_rss_bytes = parse_kibibytes(line);
        else if (line.starts_with("RssFile:"))
            result.file_rss_bytes = parse_kibibytes(line);
        else if (line.starts_with("RssShmem:"))
            result.shared_rss_bytes = parse_kibibytes(line);
        else if (line.starts_with("VmSize:"))
            result.virtual_bytes = parse_kibibytes(line);
        else if (line.starts_with("VmSwap:"))
            result.swap_bytes = parse_kibibytes(line);
    }
}
void read_linux_system_memory(Process_Memory_Snapshot& result) {
    std::ifstream stream("/proc/meminfo");
    std::string line;
    while (std::getline(stream, line)) {
        if (line.starts_with("MemTotal:"))
            result.system_total_bytes = parse_kibibytes(line);
        else if (line.starts_with("MemAvailable:"))
            result.system_available_bytes = parse_kibibytes(line);
    }
}
#endif
struct Growth_Trend {
    std::int64_t bytes_per_hour{};
    std::uint64_t window_ms{};
};
template <typename Getter>
Growth_Trend growth_trend(const std::pmr::deque<Process_Memory_Snapshot>& history, Getter getter,
                          std::uint64_t maximum_window_ms) {
    if (history.size() < 2)
        return {};
    const auto latest_timestamp = history.back().timestamp_ms;
    const auto warmup_end = history.front().timestamp_ms + growth_warmup_ms;
    if (latest_timestamp <= warmup_end)
        return {};
    const auto tail_start = latest_timestamp > maximum_window_ms ? latest_timestamp - maximum_window_ms : 0;
    const auto analysis_start = std::max(warmup_end, tail_start);
    const auto first = std::find_if(history.begin(), history.end(), [analysis_start](const auto& sample) {
        return sample.timestamp_ms >= analysis_start;
    });
    if (first == history.end() || std::next(first) == history.end())
        return {};
    const auto origin = first->timestamp_ms;
    long double sum_x = 0;
    long double sum_y = 0;
    for (auto iter = first; iter != history.end(); ++iter) {
        const auto& sample = *iter;
        sum_x += static_cast<long double>(sample.timestamp_ms - origin) / 1000.0L;
        sum_y += static_cast<long double>(getter(sample));
    }
    const auto count = static_cast<long double>(std::distance(first, history.end()));
    const auto mean_x = sum_x / count;
    const auto mean_y = sum_y / count;
    long double numerator = 0;
    long double denominator = 0;
    for (auto iter = first; iter != history.end(); ++iter) {
        const auto& sample = *iter;
        const auto x = static_cast<long double>(sample.timestamp_ms - origin) / 1000.0L - mean_x;
        const auto y = static_cast<long double>(getter(sample)) - mean_y;
        numerator += x * y;
        denominator += x * x;
    }
    if (denominator == 0)
        return {};
    return {static_cast<std::int64_t>(numerator / denominator * 3600.0L), latest_timestamp - origin};
}
Psc::JSON snapshot_json(const Process_Memory_Snapshot& value) {
    Psc::JSON result = Psc::JSON::object();
    result.append({"timestamp_ms", value.timestamp_ms});
    result.append({"rss_bytes", value.rss_bytes});
    result.append({"peak_rss_bytes", value.peak_rss_bytes});
    result.append({"anonymous_rss_bytes", value.anonymous_rss_bytes});
    result.append({"file_rss_bytes", value.file_rss_bytes});
    result.append({"shared_rss_bytes", value.shared_rss_bytes});
    result.append({"virtual_bytes", value.virtual_bytes});
    result.append({"swap_bytes", value.swap_bytes});
    result.append({"commit_bytes", value.commit_bytes});
    result.append({"peak_commit_bytes", value.peak_commit_bytes});
    result.append({"system_total_bytes", value.system_total_bytes});
    result.append({"system_available_bytes", value.system_available_bytes});
    result.append({"page_faults", value.page_faults});
    result.append({"allocator_live_bytes", value.allocator_live_bytes});
    result.append({"allocator_peak_live_bytes", value.allocator_peak_live_bytes});
    result.append({"pmr_live_bytes", value.pmr_live_bytes});
    result.append({"pmr_peak_live_bytes", value.pmr_peak_live_bytes});
    return result;
}
}
Memory_Monitor::Memory_Monitor() : history_(server_memory_resource()) {
}
Memory_Monitor& Memory_Monitor::instance() {
    static Memory_Monitor monitor;
    return monitor;
}
void Memory_Monitor::sample() {
    Process_Memory_Snapshot result;
    result.timestamp_ms = now_milliseconds();
    std::size_t elapsed{};
    std::size_t user{};
    std::size_t system{};
    std::size_t rss{};
    std::size_t peak_rss{};
    std::size_t commit{};
    std::size_t peak_commit{};
    std::size_t page_faults{};
    mi_process_info(&elapsed, &user, &system, &rss, &peak_rss, &commit, &peak_commit, &page_faults);
    result.rss_bytes = rss;
    result.peak_rss_bytes = peak_rss;
    result.commit_bytes = commit;
    result.peak_commit_bytes = peak_commit;
    result.page_faults = page_faults;
    result.system_total_bytes = Psc::get_system_memory();
#if defined(ECAP_TARGET_LINUX)
    read_linux_process_memory(result);
    read_linux_system_memory(result);
#endif
    const auto allocator = server_allocator_snapshot();
    result.allocator_live_bytes = allocator.live_bytes;
    result.allocator_peak_live_bytes = allocator.peak_live_bytes;
    result.pmr_live_bytes = allocator.pmr_live_bytes;
    result.pmr_peak_live_bytes = allocator.pmr_peak_live_bytes;
    std::lock_guard lock(mutex_);
    history_.push_back(result);
    if (history_.size() > history_capacity)
        history_.pop_front();
}
Psc::JSON Memory_Monitor::state_json() const {
    std::lock_guard lock(mutex_);
    Psc::JSON result = Psc::JSON::object();
    if (history_.empty())
        return result;
    const auto& latest = history_.back();
    const auto window_ms = latest.timestamp_ms - history_.front().timestamp_ms;
    const auto rss_growth = growth_trend(history_, [](const auto& sample) { return sample.rss_bytes; }, growth_window_ms);
    const auto anonymous_growth = growth_trend(history_, [](const auto& sample) { return sample.anonymous_rss_bytes; },
                                               growth_window_ms);
    const auto recent_anonymous_growth = growth_trend(history_, [](const auto& sample) {
        return sample.anonymous_rss_bytes;
    }, recent_growth_window_ms);
    const auto allocator_growth = growth_trend(history_, [](const auto& sample) {
        return sample.allocator_live_bytes;
    }, growth_window_ms);
    const auto process_percent = latest.system_total_bytes == 0 ? 0.0 :
    static_cast<double>(latest.rss_bytes) * 100.0 / static_cast<double>(latest.system_total_bytes);
    const auto available_percent = latest.system_total_bytes == 0 ? 0.0 :
    static_cast<double>(latest.system_available_bytes) * 100.0 / static_cast<double>(latest.system_total_bytes);
    std::string pressure = "normal";
    if (process_percent >= 75.0 || (available_percent > 0.0 && available_percent <= 5.0))
        pressure = "critical";
    else if (process_percent >= 50.0 || (available_percent > 0.0 && available_percent <= 15.0))
        pressure = "warning";
    const bool sustained_growth = anonymous_growth.window_ms >= minimum_growth_window_ms &&
                                  recent_anonymous_growth.window_ms >= minimum_recent_growth_window_ms &&
                                  anonymous_growth.bytes_per_hour >= static_cast<std::int64_t>(16 * mebibyte) &&
                                  recent_anonymous_growth.bytes_per_hour >= static_cast<std::int64_t>(16 * mebibyte);
    const auto allocator = server_allocator_snapshot();
    result.append({"allocator", "mimalloc"});
    result.append({"allocator_version", allocator.allocator_version});
    result.append({"current", snapshot_json(latest)});
    result.append({"process_memory_percent", process_percent});
    result.append({"system_available_percent", available_percent});
    result.append({"pressure", pressure});
    result.append({"sample_count", history_.size()});
    result.append({"window_seconds", window_ms / 1000});
    result.append({"growth_window_seconds", anonymous_growth.window_ms / 1000});
    result.append({"rss_growth_bytes_per_hour", rss_growth.bytes_per_hour});
    result.append({"anonymous_growth_bytes_per_hour", anonymous_growth.bytes_per_hour});
    result.append({"recent_anonymous_growth_bytes_per_hour", recent_anonymous_growth.bytes_per_hour});
    result.append({"allocator_growth_bytes_per_hour", allocator_growth.bytes_per_hour});
    result.append({"sustained_growth", sustained_growth});
    result.append({"allocated_bytes_total", allocator.allocated_bytes_total});
    result.append({"allocation_count", allocator.allocation_count});
    result.append({"deallocation_count", allocator.deallocation_count});
    result.append({"pmr_allocated_bytes_total", allocator.pmr_allocated_bytes_total});
    result.append({"pmr_allocation_count", allocator.pmr_allocation_count});
    result.append({"pmr_deallocation_count", allocator.pmr_deallocation_count});
    Psc::JSON history = Psc::JSON::array();
    const auto stride = std::max<std::size_t>(1, (history_.size() + response_history_capacity - 1) / response_history_capacity);
    for (std::size_t index = 0; index < history_.size(); index += stride)
        history.append(snapshot_json(history_[index]));
    if ((history_.size() - 1) % stride != 0)
        history.append(snapshot_json(history_.back()));
    result.append({"history", history});
    return result;
}
