/// @file dpf/run_config.hpp
/// @brief One explicit configuration for harness runs, sweeps, and nodes.
/// @details Everything the harness decides is a field here: transport, host,
///          lanes, framing, instances, wire policy (window, chunking,
///          coalescing, socket options), pipelining, wait budgets, compute
///          threads, warmup and trials, CPU pins, and setup deadlines.
///          `from_env()` and `apply_args()` both go through `set(key, value)`,
///          so `DPF_LANES=4` and `--lanes=4` mean the same thing, and an
///          unknown key or value throws. The run log is configured the same
///          way (`log_level`, `log`, `log_seeds`) and applied by
///          `app::start_logging`. So is link security: `encryption`,
///          `identity` (this party's key file), `peer.N` / `dealer_key` (keys
///          this party checks), and the `client_*` / `server_*` keys for client
///          links. `config=FILE` (or `DPF_CONFIG`) applies a file of
///          `key = value` lines through the same parser.
#ifndef LIBDPF_INCLUDE_DPF_RUN_CONFIG_HPP__
#define LIBDPF_INCLUDE_DPF_RUN_CONFIG_HPP__

#include <chrono>
#include <cstddef>
#include <cstdlib>
#include <cstring>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>

#include <fstream>
#include <map>
#include <memory>

#include "dpf/compose.hpp"
#include "dpf/log.hpp"
#include "dpf/net/identity.hpp"
#include "dpf/net/policy.hpp"
#include "dpf/net/round_lane.hpp"
#include "dpf/net/security.hpp"

namespace dpf
{
namespace app
{

struct run_config
{
    net::transport kind = net::transport::async_memory;
    /// Address the socket transports connect to (in-process runs bind all
    /// interfaces and connect here, so a NIC address exercises the NIC path).
    std::string host = "127.0.0.1";
    std::size_t n_lanes = 0;
    net::framing_mode framing = net::framing_mode::automatic;
    /// RoundSink batch width: protocol instances carried per round.
    std::size_t instances = 1;
    net::wire_policy policy{};
    /// Rounds a session may submit ahead of its current one, while that edge's
    /// window has room. Only rounds whose bytes do not depend on the peer
    /// qualify; compose plans are fully dependent, so overlap them with
    /// `instances` instead.
    std::size_t pipeline_credit = 0;
    std::chrono::milliseconds wait_timeout{30000};
    /// Kernels run on a compute pool of this size (0 = inline on the party
    /// thread).
    std::size_t compute_threads = 0;
    std::size_t warmup = 0;
    std::size_t trials = 1;
    /// Pin party threads to CPUs (-1 = no pin).
    int cpu[3] = {-1, -1, -1};
    net::deadlines limits{};
    /// Run-log threshold, sinks (`stderr`, `syslog`, `file:PATH`, `none`,
    /// comma-separated), and how seeds are printed (`full`, `hash`, `off`).
    log::level log_level = log::level::info;
    std::string log_sinks = "stderr";
    log::seed_policy log_seeds = log::seed_policy::full;
    /// Party links: TLS 1.3 unless `encryption=off`; this party's key
    /// (`identity`) and the keys it checks (`peer.N`, `dealer_key`).
    net::peer_security security{};
    /// Client links: what a client accepts (`client_*`) and what a server
    /// presents (`server_*`).
    net::client_security client{};
    net::server_security server{};

    protocol::drive_options drive() const
    {
        protocol::drive_options o;
        o.n_lanes = n_lanes;
        o.framing = framing;
        o.pipeline_credit = pipeline_credit;
        o.wait_timeout = wait_timeout;
        return o;
    }

    void set(const std::string & key, const std::string & v)
    {
        if (key == "transport")
            kind = net::parse_transport(v);
        else if (key == "host")
            host = v;
        else if (key == "lanes")
            n_lanes = v == "per_round" ? net::lanes_one_per_round : num(key, v);
        else if (key == "framing")
            framing = net::parse_framing(v);
        else if (key == "instances")
            instances = positive(key, v);
        else if (key == "window")
            policy.window_bytes = num(key, v);
        else if (key == "max_frame")
            policy.max_frame = positive(key, v);
        else if (key == "chunk")
            policy.chunk_bytes = num(key, v);
        else if (key == "coalesce_frames")
            policy.coalesce_frames = positive(key, v);
        else if (key == "coalesce_bytes")
            policy.coalesce_bytes = num(key, v);
        else if (key == "compact")
            policy.compact_bytes = num(key, v);
        else if (key == "nodelay")
            policy.socket.no_delay = flag(key, v);
        else if (key == "quickack")
            policy.socket.quickack = flag(key, v);
        else if (key == "keepalive")
            policy.socket.keepalive = flag(key, v);
        else if (key == "keepalive_idle")
            policy.socket.keepalive_idle_s = static_cast<int>(num(key, v));
        else if (key == "keepalive_interval")
            policy.socket.keepalive_interval_s = static_cast<int>(num(key, v));
        else if (key == "keepalive_count")
            policy.socket.keepalive_count = static_cast<int>(num(key, v));
        else if (key == "sndbuf")
            policy.socket.send_buffer = static_cast<int>(num(key, v));
        else if (key == "rcvbuf")
            policy.socket.recv_buffer = static_cast<int>(num(key, v));
        else if (key == "pipeline")
            pipeline_credit = num(key, v);
        else if (key == "wait_ms")
            wait_timeout = std::chrono::milliseconds(num(key, v));
        else if (key == "compute_threads")
            compute_threads = num(key, v);
        else if (key == "warmup")
            warmup = num(key, v);
        else if (key == "trials")
            trials = positive(key, v);
        else if (key == "cpu0" || key == "cpu1" || key == "cpu2")
            cpu[key[3] - '0'] = v == "-1" ? -1 : static_cast<int>(num(key, v));
        else if (key == "connect_ms")
            limits.connect = std::chrono::milliseconds(num(key, v));
        else if (key == "accept_ms")
            limits.accept = std::chrono::milliseconds(num(key, v));
        else if (key == "join_ms")
            limits.join = std::chrono::milliseconds(num(key, v));
        else if (key == "handshake_ms")
            limits.handshake = std::chrono::milliseconds(num(key, v));
        else if (key == "log_level")
            log_level = log::parse_level(v);
        else if (key == "log")
            log_sinks = v;
        else if (key == "log_seeds")
            log_seeds = log::parse_seed_policy(v);
        else if (key == "config")
            load_file(v);
        else if (key == "encryption")
            security.encrypt = flag(key, v);
        else if (key == "identity")
            security.self = load_identity(key, v);
        else if (key.rfind("peer.", 0) == 0)
            security.trusted[party_index(key)] = parse_key(key, v);
        else if (key == "dealer_key")
            security.trusted[net::dealer_id] = parse_key(key, v);
        else if (key == "client_verify")
            client.verify = flag(key, v);
        else if (key == "client_pin")
            client.pins = parse_keys(key, v);
        else if (key == "client_ca")
            client.ca_file = v;
        else if (key == "client_server_name")
            client.server_name = v;
        else if (key == "client_identity")
            client.self = load_identity(key, v);
        else if (key == "server_cert")
            server.cert_file = v;
        else if (key == "server_key")
            server.key_file = v;
        else if (key == "server_identity")
            server.self = load_identity(key, v);
        else if (key == "server_client_pin")
            server.client_pins = parse_keys(key, v);
        else
            throw std::invalid_argument("run_config: unknown key '" + key + "'");
        policy.validate();
    }

    /// @brief `DPF_<KEY>` for every key (`DPF_WIRE_WINDOW` for `window`;
    ///        `DPF_LANES=0` keeps its old meaning, one stream per round).
    static run_config from_env()
    {
        run_config c;
        c.merge_env();
        return c;
    }

    void merge_env()
    {
        static const char * const keys[] = {"config", "encryption", "identity",
            "transport", "host", "lanes",
            "framing", "instances", "window", "max_frame", "chunk",
            "coalesce_frames", "coalesce_bytes", "compact", "nodelay", "quickack",
            "keepalive", "keepalive_idle", "keepalive_interval", "keepalive_count",
            "sndbuf", "rcvbuf", "pipeline", "wait_ms", "compute_threads",
            "warmup", "trials", "cpu0", "cpu1", "cpu2", "connect_ms", "accept_ms",
            "join_ms", "handshake_ms", "log_level", "log", "log_seeds"};
        for (const char * k : keys)
        {
            std::string env = "DPF_";
            for (const char * p = k; *p != '\0'; ++p)
                env += static_cast<char>(*p >= 'a' && *p <= 'z' ? *p - 32 : *p);
            if (std::string(k) == "window")
                env = "DPF_WIRE_WINDOW";
            const char * v = std::getenv(env.c_str());
            if (v == nullptr || v[0] == '\0')
                continue;
            if (std::string(k) == "lanes" && std::string(v) == "0")
            {
                n_lanes = net::lanes_one_per_round;
                continue;
            }
            set(k, v);
        }
    }

    /// @brief Apply `--key=value` arguments; returns the arguments it did not
    ///        recognise as `--key=value` (e.g. positional ones).
    std::vector<std::string> apply_args(int argc, char ** argv)
    {
        std::vector<std::string> rest;
        for (int i = 1; i < argc; ++i)
        {
            const std::string a = argv[i];
            const auto eq = a.find('=');
            if (a.rfind("--", 0) != 0 || eq == std::string::npos)
            {
                rest.push_back(a);
                continue;
            }
            const std::string key = a.substr(2, eq - 2);
            if (key == "party" || key == "peers" || key == "dealer")
            {
                rest.push_back(a);
                continue;
            }
            set(key, a.substr(eq + 1));
        }
        return rest;
    }

    /// @brief Every setting as text, for `config.csv` and bench output.
    std::vector<std::pair<std::string, std::string>> describe() const
    {
        std::vector<std::pair<std::string, std::string>> kv;
        kv.emplace_back("transport", net::transport_name(kind));
        kv.emplace_back("host", host);
        kv.emplace_back("lanes", n_lanes == net::lanes_one_per_round
                ? std::string("per_round")
                : std::to_string(n_lanes == 0 ? net::default_lanes : n_lanes));
        kv.emplace_back("framing", net::framing_name(framing));
        kv.emplace_back("instances", std::to_string(instances));
        kv.emplace_back("window", std::to_string(policy.window_bytes));
        kv.emplace_back("max_frame", std::to_string(policy.max_frame));
        kv.emplace_back("chunk", std::to_string(policy.chunk_bytes));
        kv.emplace_back("coalesce_frames", std::to_string(policy.coalesce_frames));
        kv.emplace_back("coalesce_bytes", std::to_string(policy.coalesce_bytes));
        kv.emplace_back("compact", std::to_string(policy.compact_bytes));
        kv.emplace_back("nodelay", policy.socket.no_delay ? "1" : "0");
        kv.emplace_back("quickack", policy.socket.quickack ? "1" : "0");
        kv.emplace_back("keepalive", policy.socket.keepalive ? "1" : "0");
        kv.emplace_back("keepalive_idle", std::to_string(policy.socket.keepalive_idle_s));
        kv.emplace_back("keepalive_interval",
            std::to_string(policy.socket.keepalive_interval_s));
        kv.emplace_back("keepalive_count", std::to_string(policy.socket.keepalive_count));
        kv.emplace_back("sndbuf", std::to_string(policy.socket.send_buffer));
        kv.emplace_back("rcvbuf", std::to_string(policy.socket.recv_buffer));
        kv.emplace_back("pipeline", std::to_string(pipeline_credit));
        kv.emplace_back("wait_ms", std::to_string(wait_timeout.count()));
        kv.emplace_back("compute_threads", std::to_string(compute_threads));
        kv.emplace_back("warmup", std::to_string(warmup));
        kv.emplace_back("trials", std::to_string(trials));
        kv.emplace_back("cpu0", std::to_string(cpu[0]));
        kv.emplace_back("cpu1", std::to_string(cpu[1]));
        kv.emplace_back("cpu2", std::to_string(cpu[2]));
        kv.emplace_back("connect_ms", std::to_string(limits.connect.count()));
        kv.emplace_back("accept_ms", std::to_string(limits.accept.count()));
        kv.emplace_back("join_ms", std::to_string(limits.join.count()));
        kv.emplace_back("handshake_ms", std::to_string(limits.handshake.count()));
        kv.emplace_back("log_level", log::level_name(log_level));
        kv.emplace_back("log", log_sinks);
        kv.emplace_back("log_seeds", log::seed_policy_name(log_seeds));
        kv.emplace_back("encryption", security.encrypt ? "on" : "off");
        kv.emplace_back("identity",
            security.self ? security.self->key().base64() : std::string("ephemeral"));
        std::string trusted;
        for (const auto & t : security.trusted)
            trusted += (trusted.empty() ? "" : ",")
                + (t.first == net::dealer_id ? std::string("dealer")
                                             : std::to_string(t.first));
        kv.emplace_back("trusted", trusted.empty() ? std::string("none") : trusted);
        kv.emplace_back("client_verify", client.verify ? "on" : "off");
        kv.emplace_back("client_pins", std::to_string(client.pins.size()));
        kv.emplace_back("client_ca", client.ca_file.empty() ? std::string("none")
                                                            : client.ca_file);
        kv.emplace_back("server_cert", !server.cert_file.empty() ? server.cert_file
                : server.self                                   ? std::string("identity")
                                                                : std::string("development"));
        return kv;
    }

    /// @brief One line: `key=value key=value ...`.
    std::string summary() const
    {
        std::string out;
        for (const auto & kv : describe())
        {
            if (!out.empty())
                out += ' ';
            out += kv.first + "=" + kv.second;
        }
        return out;
    }

    /// @brief Apply `key = value` lines from `path` through `set`. `#` starts
    ///        a comment; relative paths in the file resolve against its
    ///        directory. Errors name the file and line.
    void load_file(const std::string & path)
    {
        std::ifstream in(path);
        if (!in)
            throw std::invalid_argument("run_config: cannot read config '" + path + "'");
        const auto slash = path.find_last_of('/');
        const std::string dir = slash == std::string::npos ? std::string()
                                                           : path.substr(0, slash);
        std::string line;
        std::size_t lineno = 0;
        while (std::getline(in, line))
        {
            ++lineno;
            const auto hash = line.find('#');
            if (hash != std::string::npos)
                line.erase(hash);
            line = trim(line);
            if (line.empty())
                continue;
            const auto eq = line.find('=');
            const std::string where = path + ":" + std::to_string(lineno) + ": ";
            if (eq == std::string::npos)
                throw std::invalid_argument("run_config: " + where
                    + "expected key = value, got '" + line + "'");
            const std::string key = trim(line.substr(0, eq));
            const std::string value = trim(line.substr(eq + 1));
            if (key == "config")
                throw std::invalid_argument("run_config: " + where
                    + "a config file cannot include another");
            try
            {
                set(key, resolve(dir, key, value));
            }
            catch (const std::exception & e)
            {
                throw std::invalid_argument(where + e.what());
            }
        }
    }

private:
    static std::string trim(const std::string & s)
    {
        std::size_t b = 0;
        std::size_t e = s.size();
        while (b < e && (s[b] == ' ' || s[b] == '\t' || s[b] == '\r'))
            ++b;
        while (e > b && (s[e - 1] == ' ' || s[e - 1] == '\t' || s[e - 1] == '\r'))
            --e;
        return s.substr(b, e - b);
    }

    /// @brief Make paths in a config file relative to the file.
    static std::string resolve(const std::string & dir, const std::string & key,
        const std::string & value)
    {
        if (dir.empty() || value.empty())
            return value;
        auto rel = [&](const std::string & p) {
            return p.empty() || p[0] == '/' ? p : dir + "/" + p;
        };
        const bool path_key = key == "identity" || key == "client_identity"
            || key == "server_identity" || key == "server_cert" || key == "server_key"
            || (key == "client_ca" && value != "system");
        if (path_key)
            return rel(value);
        const bool key_list = key.rfind("peer.", 0) == 0 || key == "dealer_key"
            || key == "client_pin" || key == "server_client_pin";
        if (!key_list)
            return value;
        std::string out;
        std::size_t start = 0;
        while (start <= value.size())
        {
            const auto comma = value.find(',', start);
            std::string item = trim(value.substr(start,
                comma == std::string::npos ? std::string::npos : comma - start));
            if (item.rfind("file:", 0) == 0)
                item = "file:" + rel(item.substr(5));
            out += (out.empty() ? "" : ",") + item;
            if (comma == std::string::npos)
                break;
            start = comma + 1;
        }
        return out;
    }

    static std::uint32_t party_index(const std::string & key)
    {
        return static_cast<std::uint32_t>(num(key, key.substr(5)));
    }

    static net::public_key parse_key(const std::string & key, const std::string & v)
    {
        try
        {
            return net::public_key::parse(v);
        }
        catch (const std::exception & e)
        {
            throw std::invalid_argument("run_config: " + key + ": " + e.what());
        }
    }

    static std::vector<net::public_key> parse_keys(const std::string & key,
        const std::string & v)
    {
        std::vector<net::public_key> out;
        std::size_t start = 0;
        while (start <= v.size())
        {
            const auto comma = v.find(',', start);
            const std::string item = trim(v.substr(start,
                comma == std::string::npos ? std::string::npos : comma - start));
            if (!item.empty())
                out.push_back(parse_key(key, item));
            if (comma == std::string::npos)
                break;
            start = comma + 1;
        }
        return out;
    }

    static std::shared_ptr<const net::identity> load_identity(const std::string & key,
        const std::string & path)
    {
        try
        {
            return std::make_shared<net::identity>(net::identity::load(path));
        }
        catch (const std::exception & e)
        {
            throw std::invalid_argument("run_config: " + key + ": " + e.what());
        }
    }

    static std::size_t num(const std::string & key, const std::string & v)
    {
        std::size_t pos = 0;
        unsigned long long x = 0;
        try
        {
            x = std::stoull(v, &pos, 10);
        }
        catch (const std::exception &)
        {
            pos = 0;
        }
        if (pos != v.size() || v.empty())
            throw std::invalid_argument("run_config: " + key
                + " needs a number, got '" + v + "'");
        return static_cast<std::size_t>(x);
    }

    static std::size_t positive(const std::string & key, const std::string & v)
    {
        const auto x = num(key, v);
        if (x == 0)
            throw std::invalid_argument("run_config: " + key + " must be > 0");
        return x;
    }

    static bool flag(const std::string & key, const std::string & v)
    {
        if (v == "1" || v == "on" || v == "true")
            return true;
        if (v == "0" || v == "off" || v == "false")
            return false;
        throw std::invalid_argument("run_config: " + key + " needs 0/1, got '"
            + v + "'");
    }
};

}  // namespace app
}  // namespace dpf

#endif  // LIBDPF_INCLUDE_DPF_RUN_CONFIG_HPP__