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;
};

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 when idle: Note() is lock-free atomics only. Flush() swaps
// the accumulator under a short mutex and writes one transaction.
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);

 private:
  std::atomic<bool> enabled_{true};
  std::mutex mu_;
  std::unique_ptr<StatsStore> store_;
  std::map<std::string, MethodStat> pending_;
  std::thread flush_thread_;
  std::atomic<bool> flush_run_{false};
};

}  // namespace molo::realtime::expose