moloapp/src/realtime/expose/stats.h

#pragma once
// Spec: reverse.md section 9 (observability), zero-cost stats for
// View/stream: kB up/down, per-method counts, avg latency, batched to
// SQLite (default) or Postgres (DATABASE_URL). OO: StatsCollector owns an
// atomic in-memory accumulator + a pluggable Store backend; flush is
// explicit (timer-driven by caller) inside one transaction.
#include <atomic>
#include <chrono>
#include <cstdint>
#include <thread>
#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <vector>

namespace molo::realtime::expose {

struct MethodStat {
  std::uint64_t count = 0;
  std::uint64_t bytes_in = 0;
  std::uint64_t bytes_out = 0;
  std::uint64_t latency_us_sum = 0;
  double kb_s_in = 0.0;   // windowed throughput at last flush
  double kb_s_out = 0.0;
};

struct StatsSnapshot {
  std::map<std::string, MethodStat> by_method;  // + "_total" roll-up
  std::uint64_t streams = 0;
};

// Storage backend interface (SQLite default, Postgres optional).
class StatsStore {
 public:
  virtual ~StatsStore() = default;
  virtual bool Open(std::string& err) = 0;
  // One transaction per Flush call.
  virtual bool Flush(const StatsSnapshot& snap, std::string& err) = 0;
  virtual bool Query(std::string& text_out, bool as_json, std::string& err) = 0;
};

std::unique_ptr<StatsStore> MakeStatsStore(const std::string& database_url);

// Zero-cost fast path: Note() is lock-free atomics after first-seen
// method registration (thread-local index cache; mutex only on new-method
// insert and on Flush swap). Disabled Note() is a single atomic load.
// Windowed kb/s up/down computed per flush interval from byte deltas.
struct StatCell {
  std::atomic<std::uint64_t> count{0}, bytes_in{0}, bytes_out{0}, latency_us_sum{0};
};
class StatsCollector {
 public:
  static StatsCollector& Global();
  void Configure(std::unique_ptr<StatsStore> store) {
    std::lock_guard<std::mutex> l(mu_);
    store_ = std::move(store);
  }
  bool enabled() const { return enabled_.load(std::memory_order_relaxed); }
  void set_enabled(bool v) { enabled_.store(v, std::memory_order_relaxed); }
  ~StatsCollector() { StopAutoFlush(); }
  // Time-based persisted flush: background thread calls Flush() every
  // interval (default 5s; STATS_FLUSH_SECS in .env.json/env overrides).
  // Idempotent; safe to call twice.
  void StartAutoFlush(std::chrono::seconds interval = std::chrono::seconds(5));
  void StopAutoFlush();

  void Note(const std::string& method, std::uint64_t bytes_in,
            std::uint64_t bytes_out, std::uint64_t latency_us);
  // Swap accumulator and persist in one transaction. No-op without a store.
  bool Flush(std::string& err);
  bool Query(std::string& out, bool as_json, std::string& err);

  // Windowed throughput from last flush (updated under mu_ on Flush).
  double RateIn() const { return kb_s_in_; }
  double RateOut() const { return kb_s_out_; }

 private:
  friend class StatsCollectorTestHook;
  size_t CellFor(const std::string& method);  // mutex only on new method
  std::atomic<bool> enabled_{true};
  std::mutex mu_;
  std::unique_ptr<StatsStore> store_;
  std::map<std::string, MethodStat> pending_;  // drained via cells on Flush
  std::map<std::string, size_t> method_idx_;
  std::vector<std::unique_ptr<StatCell>> cells_;
  std::vector<std::string> cell_names_;
  std::mutex idx_mu_;
  std::uint64_t flush_bytes_in_ = 0, flush_bytes_out_ = 0;
  std::chrono::steady_clock::time_point last_flush_tp_ = std::chrono::steady_clock::now();
  double kb_s_in_ = 0.0, kb_s_out_ = 0.0;
  std::thread flush_thread_;
  std::atomic<bool> flush_run_{false};
};

}  // namespace molo::realtime::expose