Crafter.Review/implementations/Crafter.Review-Memory.cpp

// SPDX-License-Identifier: AGPL-3.0-only
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

module Crafter.Review:Memory_impl;
import :Memory;
import :Json;
import :Text;
import :Platform;
import std;

namespace fs = std::filesystem;

namespace Crafter::Review::Memory {
    std::optional<std::int64_t> Available(const fs::path& path) {
        const std::map<std::string, std::int64_t> info = Platform::MemInfo(path);
        const auto found = info.find("MemAvailable");
        if (found == info.end()) return std::nullopt;
        return found->second;
    }

    std::int64_t Reserve(const std::optional<std::string>& configured, const fs::path& path) {
        if (configured && !configured->empty()) {
            const std::optional<double> value = Dyn::ParseFloat(*configured);
            if (!value) throw std::invalid_argument(Dyn::FloatError(*configured));
            return static_cast<std::int64_t>(*value);
        }
        const std::map<std::string, std::int64_t> info = Platform::MemInfo(path);
        const auto total = info.find("MemTotal");
        return std::max<std::int64_t>(2048, (total == info.end() ? 0 : total->second) / 10);
    }

    std::int64_t Reserve() {
        return Reserve(Platform::GetEnv("REVIEW_GATES_MEM_RESERVE_MB"));
    }

    struct Sampler::State {
        double interval = 3.0;
        Measure measure;
        Scan scan;
        std::mutex lock;
        std::condition_variable wake;
        std::map<std::string, std::int64_t> watched;
        std::map<std::string, double> peaks;
        bool started = false;
        bool stopping = false;
        std::thread thread;

        double Sample() {
            const auto start = std::chrono::steady_clock::now();
            std::vector<std::pair<std::string, std::int64_t>> items;
            {
                const std::lock_guard guard(lock);
                items.assign(watched.begin(), watched.end());
            }
            if (!items.empty()) {
                const std::map<std::int64_t, std::vector<std::int64_t>> kids = scan();
                for (const auto& [key, pid] : items) {
                    const double value = measure(pid, kids);
                    const std::lock_guard guard(lock);
                    const auto peak = peaks.find(key);
                    if (peak != peaks.end()) peak->second = std::max(peak->second, value);
                }
            }
            return std::chrono::duration<double>(std::chrono::steady_clock::now() - start).count();
        }

        void Loop() {
            double sweep = 0.0;
            std::unique_lock guard(lock);
            while (!stopping) {
                wake.wait_for(guard, std::chrono::duration<double>(Pause(interval, sweep)), [this] { return stopping; });
                if (stopping) break;
                guard.unlock();
                try {
                    sweep = Sample();
                } catch (const std::exception&) {
                    // A measurement never stops the run.
                    sweep = interval;
                }
                guard.lock();
            }
        }
    };

    Sampler::Sampler(double interval, Measure measure, Scan scan) : state(std::make_shared<State>()) {
        state->interval = interval;
        state->measure = measure ? std::move(measure) : Measure([](std::int64_t pid, const std::map<std::int64_t, std::vector<std::int64_t>>& kids) {
            return Platform::PssMb(Platform::Tree(pid, kids));
        });
        state->scan = scan ? std::move(scan) : Scan([] { return Platform::PpidMap(); });
    }

    Sampler::~Sampler() {
        {
            const std::lock_guard guard(state->lock);
            state->stopping = true;
        }
        state->wake.notify_all();
        if (state->thread.joinable()) state->thread.join();
    }

    void Sampler::Watch(const std::string& key, std::int64_t pid) {
        const std::lock_guard guard(state->lock);
        state->watched[key] = pid;
        state->peaks[key] = 0.0;
        if (!state->started) {
            state->started = true;
            state->thread = std::thread([owner = state.get()] { owner->Loop(); });
        }
    }

    double Sampler::Sample() {
        return state->Sample();
    }

    std::optional<double> Sampler::Done(const std::string& key) {
        const std::lock_guard guard(state->lock);
        state->watched.erase(key);
        const auto peak = state->peaks.find(key);
        if (peak == state->peaks.end()) return std::nullopt;
        const double value = peak->second;
        state->peaks.erase(peak);
        if (value == 0.0) return std::nullopt;
        return Dyn::Round(value, 1);
    }

    double Sampler::Pause(double interval, double sweep) {
        return std::max(interval, 3 * sweep);
    }

    namespace {
        // A peak worth keeping: a number, no bool, above zero.
        bool Positive(const Json& value) {
            if (!value.is_number()) return false;
            return value.is_number_float() ? value.get<double>() > 0.0 : (value.is_number_unsigned() ? value.get<std::uint64_t>() > 0 : value.get<std::int64_t>() > 0);
        }

        double AsDouble(const Json& value) {
            return value.is_number_float() ? value.get<double>() : (value.is_number_unsigned() ? static_cast<double>(value.get<std::uint64_t>()) : static_cast<double>(value.get<std::int64_t>()));
        }

        Json Clean(const Json& value) {
            Json out = Json::array();
            if (!value.is_array()) return out;
            for (const Json& item : value) {
                if (Positive(item)) out.push_back(item);
            }
            return out;
        }
    }

    Calibration::Calibration(fs::path path, const fs::path& repo, std::map<std::string, std::int64_t> defaults)
        : path(std::move(path)), defaults(std::move(defaults)) {
        std::error_code error;
        this->repo = fs::weakly_canonical(fs::absolute(repo), error).string();
    }

    Json Calibration::Load() const {
        try {
            Json document = Dyn::LoadJsonFile(path);
            return document.is_object() ? document : Json::object();
        } catch (const std::exception&) {
            return Json::object();
        }
    }

    std::vector<Json> Calibration::Peaks(std::string_view environment) const {
        const Json document = Load();
        const Json& mine = Dyn::Get(document, repo);
        const Json peaks = Clean(mine.is_object() ? Dyn::Get(mine, environment) : Json());
        return {peaks.begin(), peaks.end()};
    }

    Calibration::Cost Calibration::CostOf(std::string_view environment) const {
        const std::vector<Json> peaks = Peaks(environment);
        if (!peaks.empty()) {
            // max(): the first of equal maxima.
            const Json* best = &peaks.front();
            for (const Json& peak : peaks) {
                if (AsDouble(peak) > AsDouble(*best)) best = &peak;
            }
            return {*best, std::format("the highest of its last {} measured peaks", peaks.size())};
        }
        const auto found = defaults.find(std::string(environment));
        std::int64_t fallback = 1024;
        if (found != defaults.end()) {
            fallback = found->second;
        } else if (!defaults.empty()) {
            fallback = std::ranges::max(std::views::values(defaults));
        }
        return {Json(fallback), "the default measured 2026-10-22; none measured in this repository yet"};
    }

    bool Calibration::Record(std::string_view environment, double megabytes) {
        if (!(megabytes > 0.0) || !std::isfinite(megabytes)) return false;
        try {
            const std::lock_guard guard(lock);
            // Without the lock the write is still made.
            const Platform::FileLock fileLock(std::format("{}.lock", path.string()));
            Json document = Load();
            Json mine = Dyn::Get(document, repo);
            if (!mine.is_object()) mine = Json::object();
            Json peaks = Clean(Dyn::Get(mine, environment));
            peaks.push_back(Dyn::Round(megabytes, 1));
            if (peaks.size() > Keep) peaks.erase(peaks.begin(), peaks.begin() + static_cast<std::ptrdiff_t>(peaks.size() - Keep));
            mine[std::string(environment)] = std::move(peaks);
            document[repo] = std::move(mine);
            Platform::WriteAtomic(path, DumpFile(document), 0666);
            return true;
        } catch (const std::exception&) {
            return false;
        }
    }
}