#include "raid/data/findata.hpp"
#include "raid/core/error.hpp"
#include "raid/core/log.hpp"
#include "raid/core/hash.hpp"
#include "raid/core/manifest.hpp"
#include "raid/data/io.hpp"
#include <curl/curl.h>
#include <fstream>
#include <iostream>
#include <sstream>
#include <thread>
#include <cstdlib>
#include <filesystem>
#include <algorithm>
#include <cmath>

namespace raid::findata {

namespace fs = std::filesystem;
using json = nlohmann::json;

// ---------------------------------------------------------------- config / secrets
Config Config::from_env() {
  Config c;
  if (const char* u = std::getenv("FINDATA_URL")) { c.base_url = u; if (c.base_url.find("/api/v1") == std::string::npos) c.base_url += "/api/v1"; }
  if (const char* k = std::getenv("FINDATA_API_KEY")) c.api_key = k;
  else if (const char* f = std::getenv("FINDATA_API_KEY_FILE")) { std::ifstream in(f); std::getline(in, c.api_key); }
  if (const char* d = std::getenv("RAID_CACHE_DIR")) c.cache_dir = d;
  if (c.cache_dir.empty()) c.cache_dir = "datasets/cache";
  if (const char* o = std::getenv("RAID_OFFLINE")) c.offline = std::string(o) == "1";
  if (const char* r = std::getenv("RAID_FINDATA_RPS")) c.requests_per_second = std::atof(r);
  while (!c.api_key.empty() && (c.api_key.back() == '\n' || c.api_key.back() == '\r' || c.api_key.back() == ' ')) c.api_key.pop_back();
  return c;
}

double TokenBucket::acquire() {
  std::lock_guard<std::mutex> lk(m_);
  const auto now = std::chrono::steady_clock::now();
  tokens_ = std::min(capacity_, tokens_ + rate_ * std::chrono::duration<double>(now - last_).count()); last_ = now;
  double waited = 0;
  if (tokens_ < 1.0) { waited = (1.0 - tokens_) / rate_; std::this_thread::sleep_for(std::chrono::duration<double>(waited)); tokens_ = 1.0; last_ = std::chrono::steady_clock::now(); }
  tokens_ -= 1.0;
  return waited;
}

std::vector<std::pair<Date, Date>> chunk_dates(Date start, Date end, int max_days) {
  RAID_REQUIRE(max_days >= 1, invalid_argument, "max_days >= 1");
  RAID_REQUIRE(end >= start, invalid_argument, "end must be >= start");
  std::vector<std::pair<Date, Date>> out;
  for (Date a = start; a <= end;) { Date b = a + (max_days - 1); if (b > end) b = end; out.emplace_back(a, b); a = b + 1; }
  return out;
}
int max_days_per_call(Interval iv) { switch (iv) { case Interval::m1: return 31; case Interval::m5: return 186; case Interval::d1: return 3650; default: return 31; } }

// ---------------------------------------------------------------- HTTP
namespace {
size_t write_cb(char* p, size_t s, size_t n, void* ud) { static_cast<std::string*>(ud)->append(p, s * n); return s * n; }
size_t file_cb(char* p, size_t s, size_t n, void* ud) { static_cast<std::ofstream*>(ud)->write(p, static_cast<std::streamsize>(s * n)); return s * n; }
size_t header_cb(char* p, size_t s, size_t n, void* ud) {
  std::string line(p, s * n); auto* h = static_cast<std::map<std::string, std::string>*>(ud);
  const auto c = line.find(':'); if (c != std::string::npos) { std::string k = line.substr(0, c), v = line.substr(c + 1); std::transform(k.begin(), k.end(), k.begin(), ::tolower);
    while (!v.empty() && (v.back() == '\r' || v.back() == '\n' || v.back() == ' ')) v.pop_back(); while (!v.empty() && v.front() == ' ') v.erase(v.begin()); (*h)[k] = v; }
  return s * n;
}
struct CurlGlobal { CurlGlobal() { curl_global_init(CURL_GLOBAL_DEFAULT); } ~CurlGlobal() { curl_global_cleanup(); } };
void ensure_curl() { static CurlGlobal g; }
std::string url_encode(const std::string& s) { std::string o; char buf[4]; for (unsigned char c : s) { if (isalnum(c) || c == '-' || c == '_' || c == '.' || c == '~' || c == ',' ) o += static_cast<char>(c); else { std::snprintf(buf, sizeof buf, "%%%02X", c); o += buf; } } return o; }
}  // namespace

Client::Client(Config cfg) : cfg_(std::move(cfg)), bucket_(cfg_.requests_per_second, cfg_.burst) {
  ensure_curl(); key_ = cfg_.api_key; cfg_.api_key.clear();   // the key lives only in key_
  RAID_REQUIRE(cfg_.offline || !key_.empty(), api, "FinData API key missing: set FINDATA_API_KEY or FINDATA_API_KEY_FILE");
  fs::create_directories(cfg_.cache_dir);
}
Client::~Client() = default;
std::string Client::key_fingerprint() const { return key_.empty() ? "none" : Sha256::of(key_).substr(0, 12); }

Client::Response Client::request_once(const std::string& method, const std::string& url, const std::string& body, const std::string& content_type, const std::string& download_to) {
  Response r; CURL* c = curl_easy_init(); RAID_REQUIRE(c != nullptr, api, "curl init failed");
  struct curl_slist* hdrs = nullptr; hdrs = curl_slist_append(hdrs, ("Authorization: Bearer " + key_).c_str()); hdrs = curl_slist_append(hdrs, "User-Agent: raid/" RAID_VERSION " (UQO WP40; +simon-pierre.boucher@uqo.ca)");
  if (!body.empty()) hdrs = curl_slist_append(hdrs, ("Content-Type: " + content_type).c_str());
  std::ofstream out; if (!download_to.empty()) { out.open(download_to + ".part", std::ios::binary); RAID_REQUIRE(out.good(), io, "cannot write " + download_to); }
  curl_easy_setopt(c, CURLOPT_URL, url.c_str()); curl_easy_setopt(c, CURLOPT_HTTPHEADER, hdrs); curl_easy_setopt(c, CURLOPT_TIMEOUT, cfg_.timeout_seconds); curl_easy_setopt(c, CURLOPT_CONNECTTIMEOUT, 20L);
  curl_easy_setopt(c, CURLOPT_SSL_VERIFYPEER, cfg_.verify_tls ? 1L : 0L); curl_easy_setopt(c, CURLOPT_FOLLOWLOCATION, 0L); curl_easy_setopt(c, CURLOPT_ACCEPT_ENCODING, "");
  curl_easy_setopt(c, CURLOPT_HEADERFUNCTION, header_cb); curl_easy_setopt(c, CURLOPT_HEADERDATA, &r.headers);
  if (!download_to.empty()) { curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, file_cb); curl_easy_setopt(c, CURLOPT_WRITEDATA, &out); } else { curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, write_cb); curl_easy_setopt(c, CURLOPT_WRITEDATA, &r.body); }
  if (method == "POST") { curl_easy_setopt(c, CURLOPT_POST, 1L); curl_easy_setopt(c, CURLOPT_POSTFIELDS, body.c_str()); curl_easy_setopt(c, CURLOPT_POSTFIELDSIZE, static_cast<long>(body.size())); }
  else if (method != "GET") curl_easy_setopt(c, CURLOPT_CUSTOMREQUEST, method.c_str());
  const auto t0 = std::chrono::steady_clock::now(); const CURLcode rc = curl_easy_perform(c); r.seconds = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
  curl_easy_getinfo(c, CURLINFO_RESPONSE_CODE, &r.status); curl_slist_free_all(hdrs); curl_easy_cleanup(c);
  if (!download_to.empty()) { out.close(); if (rc == CURLE_OK && r.status == 200) fs::rename(download_to + ".part", download_to); else fs::remove(download_to + ".part"); }
  if (rc != CURLE_OK) { r.status = -static_cast<long>(rc); r.body = curl_easy_strerror(rc); }
  return r;
}

Client::Response Client::request(const std::string& method, const std::string& path, const std::string& body, const std::string& content_type, const std::string& download_to) {
  RAID_REQUIRE(!cfg_.offline, api, "offline mode: network call refused for " + method + " " + path);
  const std::string url = cfg_.base_url + path;
  Response r; double backoff = 1.0;
  for (int attempt = 1; attempt <= cfg_.max_retries + 1; ++attempt) {
    bucket_.acquire();
    r = request_once(method, url, body, content_type, download_to); r.attempts = attempt;
    Exchange ex; ex.method = method; ex.path = path; ex.status = r.status; ex.seconds = r.seconds; ex.bytes = r.body.size(); ex.attempts = attempt;
    if (auto it = r.headers.find("x-request-id"); it != r.headers.end()) ex.request_id = it->second;
    if (auto it = r.headers.find("x-dataset-id"); it != r.headers.end()) ex.dataset_id = it->second;
    if (auto it = r.headers.find("x-dataset-version"); it != r.headers.end()) ex.dataset_version = std::atoll(it->second.c_str());
    if (auto it = r.headers.find("x-next-cursor"); it != r.headers.end()) ex.next_cursor = it->second;
    { std::lock_guard<std::mutex> lk(log_m_); log_.push_back(ex); }
    const bool retry = r.status == 429 || r.status == 502 || r.status == 503 || r.status == 504 || r.status < 0;
    if (!retry) break;
    double wait = backoff; if (auto it = r.headers.find("retry-after"); it != r.headers.end()) wait = std::max(wait, std::atof(it->second.c_str()));
    log::warn("findata retry", {{"status", std::to_string(r.status)}, {"path", path}, {"attempt", std::to_string(attempt)}, {"wait_s", std::to_string(wait)}});
    if (attempt <= cfg_.max_retries) std::this_thread::sleep_for(std::chrono::duration<double>(wait));
    backoff = std::min(backoff * 2.0, 30.0);
  }
  if (r.status != 200 && r.status != 202) {
    std::string msg = r.body.substr(0, 400); try { auto j = json::parse(r.body); if (j.contains("error")) msg = j["error"].dump(); } catch (...) {}
    throw RaidError(ErrorCode::api, "FinData " + method + " " + path + " → HTTP " + std::to_string(r.status) + ": " + log::redact(msg));
  }
  return r;
}

std::string Client::cache_file(const std::string& kind, const std::string& key_material, const std::string& ext) const {
  return (fs::path(cfg_.cache_dir) / (kind + "_" + Sha256::of(key_material).substr(0, 24) + ext)).string();
}

// ---------------------------------------------------------------- simple JSON endpoints
nlohmann::json Client::health() { return json::parse(request("GET", "/health").body); }
nlohmann::json Client::openapi() { return json::parse(request("GET", "/openapi.json").body); }
nlohmann::json Client::catalog_datasets() { return json::parse(request("GET", "/catalog/datasets").body); }
nlohmann::json Client::dataset(const std::string& id) { return json::parse(request("GET", "/catalog/datasets/" + url_encode(id)).body); }
nlohmann::json Client::sessions(const std::string& symbol, Interval iv, Date start, Date end) {
  return json::parse(request("GET", "/market/sessions?symbol=" + url_encode(symbol) + "&interval=" + std::string(to_string(iv)) + "&start=" + start.iso() + "&end=" + end.iso()).body);
}
nlohmann::json Client::coverage_symbols(const std::string& dataset_id, int min_sessions, usize limit, const std::string& cursor) {
  std::string p = "/coverage/symbols?dataset_id=" + url_encode(dataset_id) + "&min_sessions=" + std::to_string(min_sessions) + "&limit=" + std::to_string(limit);
  if (!cursor.empty()) p += "&cursor=" + url_encode(cursor);
  return json::parse(request("GET", p).body);
}
nlohmann::json Client::coverage_symbols_all(const std::string& dataset_id, int min_sessions, usize max_symbols) {
  json out = json::array(); std::string cursor;
  for (;;) { auto page = coverage_symbols(dataset_id, min_sessions, 1000, cursor); for (auto& r : page["data"]) out.push_back(r); if (out.size() >= max_symbols || page["next_cursor"].is_null() || page["data"].empty()) break; cursor = page["next_cursor"].get<std::string>(); }
  return out;
}
nlohmann::json Client::fundamentals(const std::string& entity, const std::string& as_of_iso, const std::vector<std::string>& metrics, const std::string& period_type) {
  std::string p = "/fundamentals/" + url_encode(entity) + "?as_of=" + url_encode(as_of_iso);
  if (!metrics.empty()) { std::string m; for (auto& x : metrics) { if (!m.empty()) m += ","; m += x; } p += "&metrics=" + url_encode(m); }
  if (!period_type.empty()) p += "&period_type=" + period_type;
  return json::parse(request("GET", p).body);
}
nlohmann::json Client::query(const std::string& sql, usize max_rows) {
  RAID_REQUIRE(max_rows <= 100000, invalid_argument, "query: max_rows <= 100000");
  return json::parse(request("POST", "/query", json({{"sql", sql}, {"max_rows", max_rows}}).dump()).body);
}

// ---------------------------------------------------------------- bars (arrow pages)
BarsResult Client::bars(const BarsRequest& req) {
  BarsResult res; res.bars.symbol = req.symbol; res.bars.interval = req.interval; res.bars.adjustment = req.adjustment;
  const std::string key = "bars|" + req.symbol + "|" + std::string(to_string(req.interval)) + "|" + req.start.iso() + "|" + req.end.iso() + "|" + req.adjustment + "|v" + (req.version ? std::to_string(*req.version) : "latest");
  const std::string meta_path = cache_file("bars", key, ".meta.json"), data_path = cache_file("bars", key, ".csv");
  if (fs::exists(meta_path) && fs::exists(data_path)) {
    auto meta = json::parse(std::ifstream(meta_path)); if (Sha256::of_file(data_path) == meta["sha256"].get<std::string>()) {
      auto m = data::read_bars_csv(data_path, req.interval); if (!m.empty()) res.bars = m.begin()->second; res.bars.symbol = req.symbol; res.bars.instrument_id = meta.value("instrument_id", ""); res.bars.adjustment = req.adjustment;
      res.dataset_version = meta["dataset_version"].get<i64>(); res.dataset_id = meta["dataset_id"].get<std::string>(); res.from_cache = true; res.cache_path = data_path; return res; }
  }
#ifndef RAID_WITH_ARROW
  throw RaidError(ErrorCode::unsupported, "bars(): Arrow build required to decode pages (RAID_WITH_ARROW)");
#else
  const usize log_before = log_.size();
  for (auto [a, b] : chunk_dates(req.start, req.end, max_days_per_call(req.interval))) {
    std::string cursor;
    for (;;) {
      std::string p = "/market/bars?symbol=" + url_encode(req.symbol) + "&interval=" + std::string(to_string(req.interval)) + "&start=" + a.iso() + "&end=" + b.iso() + "&adjustment=" + req.adjustment + "&format=arrow&limit=" + std::to_string(req.page_limit);
      if (req.version) p += "&version=" + std::to_string(*req.version); if (!cursor.empty()) p += "&cursor=" + url_encode(cursor);
      auto r = request("GET", p);
      const std::string tmp = cache_file("page", key + p + std::to_string(log_.size()), ".arrow"); { std::ofstream f(tmp, std::ios::binary); f.write(r.body.data(), static_cast<std::streamsize>(r.body.size())); }
      auto pb = data::read_bars_arrow_ipc(tmp, req.interval); fs::remove(tmp);
      for (auto& [sym, bs] : pb.series) { if (res.bars.instrument_id.empty()) res.bars.instrument_id = bs.instrument_id; for (usize i = 0; i < bs.size(); ++i) res.bars.push_back(bs.t_open[i], bs.open[i], bs.high[i], bs.low[i], bs.close[i], bs.volume[i], bs.flags[i]); }
      if (auto it = r.headers.find("x-dataset-version"); it != r.headers.end()) res.dataset_version = std::atoll(it->second.c_str());
      if (auto it = r.headers.find("x-dataset-id"); it != r.headers.end()) res.dataset_id = it->second;
      auto nc = r.headers.find("x-next-cursor"); if (nc == r.headers.end() || nc->second.empty() || pb.rows_read == 0) break; cursor = nc->second;
    }
  }
  res.exchanges.assign(log_.begin() + static_cast<long>(log_before), log_.end());
  auto st = res.bars.normalize(); res.duplicates_dropped = st.duplicates_removed;
  data::write_bars_csv(data_path, res.bars);
  json meta = {{"key", key}, {"sha256", Sha256::of_file(data_path)}, {"dataset_version", res.dataset_version}, {"dataset_id", res.dataset_id}, {"instrument_id", res.bars.instrument_id}, {"rows", res.bars.size()}, {"duplicates_dropped", res.duplicates_dropped}, {"fetched_at", now_utc_iso()}, {"key_fingerprint", key_fingerprint()}, {"request_ids", json::array()}};
  for (auto& e : res.exchanges) meta["request_ids"].push_back(e.request_id);
  std::ofstream(meta_path) << meta.dump(1); res.cache_path = data_path;
  return res;
#endif
}

// ---------------------------------------------------------------- exports
ExportResult Client::export_bars(const ExportRequest& req, const std::string& out_path) {
  ExportResult res; const auto t0 = std::chrono::steady_clock::now();
  RAID_REQUIRE(!req.symbols.empty(), invalid_argument, "export: symbols required");
  RAID_REQUIRE(req.end.days - req.start.days <= 12 * 366, invalid_argument, "export: at most 12 years per request (chunk the range)");
  std::string syms; for (auto& s : req.symbols) { if (!syms.empty()) syms += ","; syms += s; }
  const std::string key = "export|" + req.dataset_id + "|" + syms + "|" + req.start.iso() + "|" + req.end.iso() + "|" + req.adjustment;
  const std::string meta_path = cache_file("export", key, ".meta.json"); std::string data_path = out_path.empty() ? cache_file("export", key, ".parquet") : out_path;
  if (fs::exists(meta_path) && fs::exists(data_path)) {
    auto meta = json::parse(std::ifstream(meta_path));
    if (Sha256::of_file(data_path) == meta["sha256"].get<std::string>()) { res.export_id = meta["export_id"]; res.path = data_path; res.sha256 = meta["sha256"]; res.dataset_version = meta["dataset_version"]; res.rows = meta["rows"]; res.bytes = meta["bytes"]; res.from_cache = true; return res; }
  }
  const usize log_before = log_.size();
  json body = {{"dataset_id", req.dataset_id}, {"symbols", req.symbols}, {"start", req.start.iso()}, {"end", req.end.iso()}, {"adjustment", req.adjustment}, {"format", "parquet"}};
  auto r = request("POST", "/exports", body.dump()); auto j = json::parse(r.body); res.export_id = j["id"].get<std::string>();
  json st; double wait = 1.0;
  for (int i = 0; i < 3600; ++i) {
    st = json::parse(request("GET", "/exports/" + res.export_id).body); const std::string s = st["status"].get<std::string>();
    if (s == "ready") break; RAID_REQUIRE(s != "failed" && s != "expired", api, "export " + s + ": " + st["error"].dump());
    std::this_thread::sleep_for(std::chrono::duration<double>(wait)); wait = std::min(wait * 1.5, 10.0);
  }
  RAID_REQUIRE(st["status"] == "ready", api, "export did not become ready");
  fs::create_directories(fs::path(data_path).parent_path().empty() ? "." : fs::path(data_path).parent_path());
  request("GET", "/exports/" + res.export_id + "/download", "", "application/json", data_path);
  res.path = data_path; res.sha256 = Sha256::of_file(data_path); res.rows = st.value("row_count", 0); res.bytes = st.value("bytes", 0);
  if (st.contains("snapshot") && st["snapshot"].contains("version")) res.dataset_version = st["snapshot"]["version"].get<i64>();
  res.exchanges.assign(log_.begin() + static_cast<long>(log_before), log_.end());
  RAID_REQUIRE(fs::file_size(data_path) == res.bytes || res.bytes == 0, io, "export size mismatch: " + std::to_string(fs::file_size(data_path)) + " vs " + std::to_string(res.bytes));
  json meta = {{"key", key}, {"export_id", res.export_id}, {"sha256", res.sha256}, {"dataset_version", res.dataset_version}, {"dataset_id", req.dataset_id}, {"rows", res.rows}, {"bytes", res.bytes}, {"request", body}, {"status", st}, {"fetched_at", now_utc_iso()}, {"key_fingerprint", key_fingerprint()}, {"request_ids", json::array()}};
  for (auto& e : res.exchanges) meta["request_ids"].push_back(e.request_id);
  std::ofstream(meta_path) << meta.dump(1);
  res.seconds_total = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
  return res;
}

std::map<std::string, BarSeries> Client::export_and_read(const ExportRequest& req) {
#ifdef RAID_WITH_ARROW
  auto e = export_bars(req); auto pb = data::read_bars_parquet(e.path); return std::move(pb.series);
#else
  (void)req; throw RaidError(ErrorCode::unsupported, "export_and_read requires the Arrow build");
#endif
}

Provenance Client::provenance(const std::string& dataset_id) const {
  Provenance p; p.dataset_id = dataset_id; p.source = "FirstRate Data via FinData OS (" + cfg_.base_url + ")"; p.code_version = RAID_VERSION;
  for (auto& e : log_) { if (!e.request_id.empty()) p.request_ids.push_back(e.request_id); if (e.dataset_version > 0 && (dataset_id.empty() || e.dataset_id == dataset_id)) p.dataset_version = e.dataset_version; }
  p.extra = {{"key_fingerprint", key_fingerprint()}, {"n_requests", log_.size()}};
  return p;
}

// ---------------------------------------------------------------- CLI
namespace {
std::string arg(const std::vector<std::string>& a, const std::string& k, const std::string& d = "") { for (usize i = 0; i + 1 < a.size(); ++i) if (a[i] == k) return a[i + 1]; return d; }
}
int cli_main(const std::vector<std::string>& a) {
  if (a.empty()) { std::fprintf(stderr, "raid data catalog|health|sessions SYM --start --end|coverage [--min-sessions N] [--dataset ID]|bars SYM --start --end [--interval 1m] [--adjustment raw]|export SYMS --start --end [--dataset ID] [--out file.parquet]|fund ENTITY --as-of ISO [--metrics a,b]|query \"SQL\" [--max-rows N]|log\n"); return 2; }
  Client c; const std::string cmd = a[0];
  auto out = [](const json& j) { std::cout << j.dump(1) << "\n"; };
  if (cmd == "catalog") { out(c.catalog_datasets()); return 0; }
  if (cmd == "health") { out(c.health()); return 0; }
  if (cmd == "sessions") { out(c.sessions(a.at(1), interval_from_string(arg(a, "--interval", "1m")), date_from_iso(arg(a, "--start")), date_from_iso(arg(a, "--end")))); return 0; }
  if (cmd == "coverage") { out(c.coverage_symbols_all(arg(a, "--dataset", "market_bars.stock.1m.raw"), std::stoi(arg(a, "--min-sessions", "250")), static_cast<usize>(std::stoul(arg(a, "--max", "100000"))))); return 0; }
  if (cmd == "bars") { BarsRequest r; r.symbol = a.at(1); r.interval = interval_from_string(arg(a, "--interval", "1m")); r.start = date_from_iso(arg(a, "--start")); r.end = date_from_iso(arg(a, "--end")); r.adjustment = arg(a, "--adjustment", "raw");
    auto res = c.bars(r); json j = {{"symbol", r.symbol}, {"rows", res.bars.size()}, {"sessions", res.bars.n_sessions()}, {"dataset_id", res.dataset_id}, {"dataset_version", res.dataset_version}, {"duplicates_dropped", res.duplicates_dropped}, {"from_cache", res.from_cache}, {"cache_path", res.cache_path}, {"requests", res.exchanges.size()}};
    if (!arg(a, "--out").empty()) { data::write_bars_csv(arg(a, "--out"), res.bars); j["out"] = arg(a, "--out"); } out(j); return 0; }
  if (cmd == "export") { ExportRequest r; std::string s = a.at(1); usize p = 0; for (usize i = 0; i <= s.size(); ++i) if (i == s.size() || s[i] == ',') { r.symbols.push_back(s.substr(p, i - p)); p = i + 1; }
    r.dataset_id = arg(a, "--dataset", "market_bars.stock.1m.raw"); r.start = date_from_iso(arg(a, "--start")); r.end = date_from_iso(arg(a, "--end")); r.adjustment = arg(a, "--adjustment", "raw");
    auto e = c.export_bars(r, arg(a, "--out")); out({{"export_id", e.export_id}, {"path", e.path}, {"sha256", e.sha256}, {"dataset_version", e.dataset_version}, {"rows", e.rows}, {"bytes", e.bytes}, {"from_cache", e.from_cache}, {"seconds", e.seconds_total}}); return 0; }
  if (cmd == "fund") { std::vector<std::string> m; std::string ms = arg(a, "--metrics"); usize p = 0; for (usize i = 0; i <= ms.size() && !ms.empty(); ++i) if (i == ms.size() || ms[i] == ',') { m.push_back(ms.substr(p, i - p)); p = i + 1; } out(c.fundamentals(a.at(1), arg(a, "--as-of", "2026-01-01T00:00:00Z"), m, arg(a, "--period", ""))); return 0; }
  if (cmd == "query") { out(c.query(a.at(1), static_cast<usize>(std::stoul(arg(a, "--max-rows", "1000"))))); return 0; }
  if (cmd == "log") { json j = json::array(); for (auto& e : c.log()) j.push_back({{"method", e.method}, {"path", e.path}, {"status", e.status}, {"request_id", e.request_id}}); out(j); return 0; }
  std::fprintf(stderr, "unknown data subcommand %s\n", cmd.c_str()); return 2;
}

}  // namespace raid::findata