fix: stats flush ensures schema, query flushes pending, add sqlite round-trip test

src/realtime/expose/claim.cpp

@@ -0,0 +1,44 @@
+#include "realtime/expose/claim.h"
+#include <algorithm>
+#include <cctype>
+namespace molo::realtime::expose {
+namespace {
+std::string Lower(std::string s){for(auto&c:s)c=tolower((unsigned char)c);return s;}
+bool Bad(const std::string& p){
+  if(p.find('\\')!=std::string::npos||p.find('\0')!=std::string::npos) return true;
+  std::string l=Lower(p);
+  if(l.find("..")!=std::string::npos||l.find("%2e")!=std::string::npos) return true;
+  return false;
+}
+std::string NormPath(std::string p){
+  if(p.empty())return "/";
+  if(p[0]!='/')p="/"+p;
+  std::string o;for(char c:p){if(c=='/'&&!o.empty()&&o.back()=='/')continue;o+=c;}
+  if(o.size()>1&&o.back()=='/')o.pop_back();
+  return o;
+}
+}
+std::string Claim::str() const { return host.empty()?path:host+(whole_host&&path=="/"?"/":path); }
+std::optional<Claim> ParseClaim(const std::string& raw, std::string& reason){
+  if(raw.empty()){reason="empty claim";return std::nullopt;}
+  std::string s=raw;
+  // strip scheme if given (reject? be lenient: strip)
+  auto pos=s.find("://"); if(pos!=std::string::npos)s=s.substr(pos+3);
+  std::string host,path;
+  auto slash=s.find('/');
+  if(slash==std::string::npos){ // bare host?
+    if(s.find('.')!=std::string::npos||s=="localhost"){host=s;path="/";}
+    else {host="";path=s;}
+  } else if(slash==0){host="";path=s;}
+  else {host=s.substr(0,slash);path=s.substr(slash);}
+  if(host.find(':')!=std::string::npos){reason="host must not contain port";return std::nullopt;}
+  host=Lower(host);
+  while(host.size()>1&&host.back()=='.')host.pop_back();  // trailing-dot FQDN
+  if(host.size()>256){reason="host too long";return std::nullopt;}
+  if(path.size()>1024){reason="path too long";return std::nullopt;}
+  if(Bad(path)){reason="bad path";return std::nullopt;}
+  path=NormPath(path);
+  Claim c{host,path,path=="/"};
+  return c;
+}
+}  // namespace molo::realtime::expose

src/realtime/expose/claim.h

@@ -0,0 +1,19 @@
+#pragma once
+// Spec: reverse.md sections 2-3. Claim grammar shared by server and CLI.
+#include <optional>
+#include <string>
+
+namespace molo::realtime::expose {
+
+struct Claim {
+  std::string host;  // lowercase, empty = any host
+  std::string path;  // normalized, always starts with '/'
+  bool whole_host = false;
+
+  std::string str() const;
+};
+
+// Normalizes per reverse.md section 3. Returns nullopt with reason set.
+std::optional<Claim> ParseClaim(const std::string& raw, std::string& reason);
+
+}  // namespace molo::realtime::expose

src/realtime/expose/hello.cpp

@@ -0,0 +1,39 @@
+#include "realtime/expose/hello.h"
+#include "realtime/shared/mini_json.h"
+namespace molo::realtime::expose {
+using shared::Json;
+std::optional<Hello> ParseHello(const std::string& js,const std::string& dflt,std::string& err){
+  auto v=Json::parse(js); if(!v){err="bad json";return std::nullopt;}
+  const Json* vv=v->find("v"); int ver=vv&&vv->type()==Json::Type::kInt?(int)vv->integer():1;
+  const Json* c=v->find("claim"); if(!c||c->type()!=Json::Type::kString){err="missing claim";return std::nullopt;}
+  Hello h; h.v=ver; h.claim=c->string();
+  if(ver>=2){
+    const Json* us=v->find("upstreams");
+    if(us&&us->type()==Json::Type::kArray){
+      for(auto&u:us->array()){
+        const Json* s=u.find("upstream"); const Json* p=u.find("prefix");
+        std::string rs=s&&s->type()==Json::Type::kString?s->string():dflt;
+        if(p&&p->type()==Json::Type::kString)rs+="="+p->string();
+        std::string r; auto e=ParseUpstream(rs,r);
+        if(!e){err="bad upstream: "+r;return std::nullopt;}
+        if(!h.upstreams.Add(*e,r)){err=r;return std::nullopt;}
+      }
+    }
+  } else {
+    const Json* u=v->find("upstream"); std::string rs=u&&u->type()==Json::Type::kString?u->string():dflt;
+    std::string r; auto e=ParseUpstream(rs,r); if(!e){err="bad upstream: "+r;return std::nullopt;}
+    h.upstreams.Add(*e,r);
+  }
+  if(h.upstreams.size()==0&&!dflt.empty()){std::string r;auto e=ParseUpstream(dflt,r);if(e)h.upstreams.Add(*e,r);}
+  const Json* a=v->find("agent"); if(a&&a->type()==Json::Type::kString)h.agent=a->string();
+  const Json* t=v->find("tunnel_id"); if(t&&t->type()==Json::Type::kString)h.tunnel_id=t->string();
+  return h;
+}
+static std::string Esc(const std::string& s){std::string o;for(char c:s){if(c=='"')o+="\\\"";else if(c=='\\')o+="\\\\";else o+=c;}return o;}
+std::string HelloOk(const std::string& c,const std::string& t){
+  return "{\"v\":2,\"status\":\"ok\",\"claim\":\""+Esc(c)+"\",\"tunnel_id\":\""+Esc(t)+"\"}";
+}
+std::string HelloErr(const std::string& c,const std::string& m){
+  return "{\"v\":2,\"status\":\"error\",\"code\":\""+Esc(c)+"\",\"message\":\""+Esc(m)+"\"}";
+}
+}  // namespace molo::realtime::expose

src/realtime/expose/hello.h

@@ -0,0 +1,13 @@
+#pragma once
+// Spec: reverse.md section 4. v1/v2 hello encode + validation (H2 + WS fallback).
+#include <optional>
+#include <string>
+#include <vector>
+#include "realtime/expose/upstream.h"
+namespace molo::realtime::expose {
+struct Hello { int v=2; std::string claim; UpstreamTable upstreams; std::string agent; std::string tunnel_id; };
+// Parses hello JSON (v1 single-upstream or v2 table). Returns error string on failure.
+std::optional<Hello> ParseHello(const std::string& json, const std::string& dflt_upstream, std::string& err);
+std::string HelloOk(const std::string& claim, const std::string& tunnel_id);
+std::string HelloErr(const std::string& code, const std::string& msg);
+}  // namespace molo::realtime::expose

src/realtime/expose/proxy.cpp

@@ -0,0 +1,31 @@
+#include "realtime/expose/proxy.h"
+#include <cctype>
+namespace molo::realtime::expose {
+uint32_t NextProxyStreamId(std::atomic<uint32_t>& n);
+std::string StripClaimPrefix(const std::string& cp, bool whole, const std::string& rp) {
+  if (whole || cp == "/") return rp.empty() ? "/" : rp;
+  if (rp.size() >= cp.size() && rp.compare(0, cp.size(), cp) == 0) {
+    std::string rest = rp.substr(cp.size());
+    return rest.empty() ? "/" : rest;
+  }
+  return rp;
+}
+bool IsHopByHop(const std::string& name) {
+  std::string l; for (char c : name) l += tolower((unsigned char)c);
+  return l=="connection"||l=="transfer-encoding"||l=="keep-alive"||l=="proxy-authenticate"||
+         l=="proxy-authorization"||l=="te"||l=="trailer"||l=="upgrade";
+}
+std::string ForwardedFor(const std::string& prior, const std::string& peer) {
+  return prior.empty() ? peer : prior + ", " + peer;
+}
+bool PerIpCounter::Admit(const std::string& ip) {
+  std::lock_guard<std::mutex> l(mu_);
+  if (n_[ip] >= 128) return false;
+  ++n_[ip]; return true;
+}
+void PerIpCounter::Release(const std::string& ip) {
+  std::lock_guard<std::mutex> l(mu_);
+  auto it = n_.find(ip); if (it == n_.end()) return;
+  if (--it->second <= 0) n_.erase(it);
+}
+}  // namespace molo::realtime::expose

src/realtime/expose/proxy.h

@@ -0,0 +1,54 @@
+#pragma once
+// Spec: reverse.md section 5. H2 transport glue for reverse tunnels.
+// OO: TunnelProxy is stateless logic over ClaimRegistry + H2Connection:
+//   one H2 connection per tunnel, one stream per request (odd client ids),
+//   flow-control/RST/GOAWAY come from src/runtime/http/v2, caps from caps.h.
+#include <atomic>
+#include <cstdint>
+#include <map>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "realtime/expose/stats.h"
+#include "realtime/expose/tunnel.h"
+
+namespace molo::realtime::expose {
+
+// H2 stream ids opened by client are odd: 1,3,5... stream 1 = hello.
+inline uint32_t NextProxyStreamId(std::atomic<uint32_t>& next) {
+  return next.fetch_add(2);  // caller seeds with 3 (1 reserved for hello)
+}
+
+// Strip claim prefix (relative) or keep full path (whole-host). Exposed for test.
+std::string StripClaimPrefix(const std::string& claim_path, bool whole_host,
+                             const std::string& req_path);
+// Hop-by-hop headers stripped per RFC 9110 section 7.6.1 (+Transfer-Encoding).
+bool IsHopByHop(const std::string& name);
+// X-Forwarded-For/Host/Proto + X-Molo-Tunnel-Id appended by server.
+std::string ForwardedFor(const std::string& prior, const std::string& peer_ip);
+
+// H2 error codes used by tunnel proxying (RFC 9113 section 7).
+inline constexpr uint32_t kH2RefusedStream = 0x7;
+inline constexpr uint32_t kH2Cancel = 0x8;
+
+// Stream accounting (reverse.md §9): tunnel counters + zero-cost stats.
+// Header-only to avoid a proxy.cpp dependency on stats backends.
+inline void RecordStream(Tunnel& t, const std::string& method,
+                         std::uint64_t bytes_in, std::uint64_t bytes_out,
+                         std::uint64_t latency_us) {
+  t.NoteStream(bytes_in, bytes_out);
+  StatsCollector::Global().Note(method, bytes_in, bytes_out, latency_us);
+}
+
+// Rate-limit: 128 conns per public IP per route (shared realtime cap).
+class PerIpCounter {
+ public:
+  bool Admit(const std::string& ip);  // false = over 128, caller sends REFUSED_STREAM
+  void Release(const std::string& ip);
+ private:
+  std::mutex mu_;
+  std::map<std::string, int> n_;
+};
+
+}  // namespace molo::realtime::expose

src/realtime/expose/stats.cpp

@@ -0,0 +1,334 @@
+// Spec: reverse.md section 9. StatsCollector: lock-free Note(), txn Flush().
+// DRY: single accumulator map serves text + JSON query paths.
+// Raw sqlite3 + libpq used directly (no ORM) to keep the hot path thin.
+#include "realtime/expose/stats.h"
+
+#include <sqlite3.h>
+
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <sstream>
+
+// Minimal libpq declarations (no dev headers on build hosts); link -lpq.
+extern "C" {
+typedef struct pg_conn PGconn;
+typedef struct pg_result PGresult;
+PGconn* PQconnectdb(const char* conninfo);
+int PQstatus(const PGconn* conn);
+char* PQerrorMessage(const PGconn* conn);
+void PQfinish(PGconn* conn);
+PGresult* PQexec(PGconn* conn, const char* command);
+int PQresultStatus(const PGresult* res);
+char* PQresultErrorMessage(const PGresult* res);
+void PQclear(PGresult* res);
+int PQntuples(const PGresult* res);
+int PQnfields(const PGresult* res);
+char* PQgetvalue(const PGresult* res, int row, int col);
+}
+#ifndef CONNECTION_OK
+#define CONNECTION_OK 0
+#define PGRES_COMMAND_OK 1
+#define PGRES_TUPLES_OK 2
+#endif
+
+namespace molo::realtime::expose {
+namespace {
+
+constexpr const char* kSchemaSqlite =
+    "CREATE TABLE IF NOT EXISTS molo_stats("
+    "method TEXT PRIMARY KEY, count INTEGER NOT NULL DEFAULT 0,"
+    "bytes_in INTEGER NOT NULL DEFAULT 0, bytes_out INTEGER NOT NULL DEFAULT 0,"
+    "latency_us_sum INTEGER NOT NULL DEFAULT 0);";
+constexpr const char* kSchemaPg =
+    "CREATE TABLE IF NOT EXISTS molo_stats("
+    "method TEXT PRIMARY KEY, count BIGINT NOT NULL DEFAULT 0,"
+    "bytes_in BIGINT NOT NULL DEFAULT 0, bytes_out BIGINT NOT NULL DEFAULT 0,"
+    "latency_us_sum BIGINT NOT NULL DEFAULT 0)";
+
+std::string EnvOr(const char* k, const std::string& fb) {
+  const char* v = std::getenv(k);
+  return v && *v ? v : fb;
+}
+
+// .env.json reader: flat {"KEY":"val",...} only; missing file -> "".
+std::string EnvJsonValue(const std::string& key) {
+  FILE* f = std::fopen(".env.json", "r");
+  if (!f) return "";
+  std::string s;
+  char buf[1024];
+  size_t n;
+  while ((n = std::fread(buf, 1, sizeof buf, f)) > 0) s.append(buf, n);
+  std::fclose(f);
+  std::string want = "\"" + key + "\"";
+  auto p = s.find(want);
+  if (p == std::string::npos) return "";
+  p = s.find(':', p + want.size());
+  if (p == std::string::npos) return "";
+  ++p;
+  while (p < s.size() && (s[p] == ' ' || s[p] == '\t')) ++p;
+  if (p < s.size() && s[p] == '"') {
+    std::string out;
+    for (++p; p < s.size() && s[p] != '"'; ++p) {
+      if (s[p] == '\\' && p + 1 < s.size()) out += s[++p];
+      else out += s[p];
+    }
+    return out;
+  }
+  auto e = s.find_first_of(",}\n", p);
+  return s.substr(p, e == std::string::npos ? e : e - p);
+}
+
+std::string DatabaseUrl() {
+  std::string v = EnvJsonValue("DATABASE_URL");
+  if (!v.empty()) return v;
+  return EnvOr("DATABASE_URL", "sqlite://system.db");
+}
+
+class SqliteStore : public StatsStore {
+ public:
+  explicit SqliteStore(std::string path) : path_(std::move(path)) {}
+  ~SqliteStore() override {
+    if (db_) sqlite3_close(db_);
+  }
+  bool Open(std::string& err) override {
+    if (sqlite3_open(path_.c_str(), &db_) != SQLITE_OK) {
+      err = sqlite3_errmsg(db_);
+      return false;
+    }
+    char* msg = nullptr;
+    if (sqlite3_exec(db_, kSchemaSqlite, nullptr, nullptr, &msg) != SQLITE_OK) {
+      err = msg ? msg : "schema failed";
+      sqlite3_free(msg);
+      return false;
+    }
+    return true;
+  }
+  bool Flush(const StatsSnapshot& snap, std::string& err) override {
+    if (!db_ && !Open(err)) return false;
+    char* msg = nullptr;
+    if (sqlite3_exec(db_, "BEGIN IMMEDIATE", nullptr, nullptr, &msg) != SQLITE_OK) {
+      err = msg ? msg : "begin failed";
+      sqlite3_free(msg);
+      return false;
+    }
+    const char* sql =
+        "INSERT INTO molo_stats(method,count,bytes_in,bytes_out,latency_us_sum)"
+        " VALUES(?,?,?,?,?) ON CONFLICT(method) DO UPDATE SET"
+        " count=count+excluded.count, bytes_in=bytes_in+excluded.bytes_in,"
+        " bytes_out=bytes_out+excluded.bytes_out,"
+        " latency_us_sum=latency_us_sum+excluded.latency_us_sum";
+    sqlite3_stmt* st = nullptr;
+    bool ok = true;
+    if (sqlite3_prepare_v2(db_, sql, -1, &st, nullptr) == SQLITE_OK) {
+      for (auto& [m, s] : snap.by_method) {
+        sqlite3_bind_text(st, 1, m.c_str(), -1, SQLITE_TRANSIENT);
+        sqlite3_bind_int64(st, 2, (long long)s.count);
+        sqlite3_bind_int64(st, 3, (long long)s.bytes_in);
+        sqlite3_bind_int64(st, 4, (long long)s.bytes_out);
+        sqlite3_bind_int64(st, 5, (long long)s.latency_us_sum);
+        if (sqlite3_step(st) != SQLITE_DONE) ok = false;
+        sqlite3_reset(st);
+      }
+      sqlite3_finalize(st);
+    } else {
+      ok = false;
+    }
+    if (sqlite3_exec(db_, ok ? "COMMIT" : "ROLLBACK", nullptr, nullptr, &msg) != SQLITE_OK) {
+      err = msg ? msg : "commit failed";
+      sqlite3_free(msg);
+      return false;
+    }
+    if (!ok) {
+      err = "upsert failed";
+      return false;
+    }
+    return true;
+  }
+  bool Query(std::string& out, bool as_json, std::string& err) override {
+    if (!db_ && !Open(err)) return false;
+    const char* sql =
+        "SELECT method,count,bytes_in,bytes_out,latency_us_sum FROM molo_stats ORDER BY method";
+    sqlite3_stmt* st = nullptr;
+    if (sqlite3_prepare_v2(db_, sql, -1, &st, nullptr) != SQLITE_OK) {
+      err = sqlite3_errmsg(db_);
+      return false;
+    }
+    std::ostringstream o;
+    if (as_json) o << "[";
+    bool first = true;
+    while (sqlite3_step(st) == SQLITE_ROW) {
+      const char* m = (const char*)sqlite3_column_text(st, 0);
+      long long c = sqlite3_column_int64(st, 1), bi = sqlite3_column_int64(st, 2),
+                bo = sqlite3_column_int64(st, 3), ls = sqlite3_column_int64(st, 4);
+      double avg = c ? (double)ls / (double)c / 1000.0 : 0.0;
+      if (as_json) {
+        if (!first) o << ",";
+        first = false;
+        o << "{\"method\":\"" << (m ? m : "") << "\",\"count\":" << c << ",\"kb_in\":"
+          << (bi / 1024) << ",\"kb_out\":" << (bo / 1024) << ",\"avg_ms\":" << avg << "}";
+      } else {
+        char b[256];
+        std::snprintf(b, sizeof b, "%-8s count=%lld kB_in=%lld kB_out=%lld avg_ms=%.2f\n",
+                      m ? m : "", c, bi / 1024, bo / 1024, avg);
+        o << b;
+      }
+    }
+    sqlite3_finalize(st);
+    if (as_json) o << "]";
+    out = o.str();
+    return true;
+  }
+
+ private:
+  std::string path_;
+  sqlite3* db_ = nullptr;
+};
+
+class PgStore : public StatsStore {
+ public:
+  explicit PgStore(std::string url) : url_(std::move(url)) {}
+  ~PgStore() override {
+    if (db_) PQfinish(db_);
+  }
+  bool Open(std::string& err) override {
+    db_ = PQconnectdb(url_.c_str());
+    if (!db_ || PQstatus(db_) != CONNECTION_OK) {
+      err = db_ ? PQerrorMessage(db_) : "pg connect failed";
+      return false;
+    }
+    PGresult* r = PQexec(db_, kSchemaPg);
+    bool ok = r && PQresultStatus(r) == PGRES_COMMAND_OK;
+    if (!ok) err = r ? PQresultErrorMessage(r) : "pg schema failed";
+    if (r) PQclear(r);
+    return ok;
+  }
+  bool Flush(const StatsSnapshot& snap, std::string& err) override {
+    if (!db_ && !Open(err)) return false;
+    PGresult* r = PQexec(db_, "BEGIN");
+    if (r) PQclear(r);
+    bool ok = true;
+    for (auto& [m, s] : snap.by_method) {
+      char sql[1024];
+      std::snprintf(sql, sizeof sql,
+                    "INSERT INTO molo_stats(method,count,bytes_in,bytes_out,latency_us_sum)"
+                    " VALUES('%s',%llu,%llu,%llu,%llu) ON CONFLICT(method) DO UPDATE SET"
+                    " count=molo_stats.count+excluded.count,"
+                    " bytes_in=molo_stats.bytes_in+excluded.bytes_in,"
+                    " bytes_out=molo_stats.bytes_out+excluded.bytes_out,"
+                    " latency_us_sum=molo_stats.latency_us_sum+excluded.latency_us_sum",
+                    m.c_str(), (unsigned long long)s.count, (unsigned long long)s.bytes_in,
+                    (unsigned long long)s.bytes_out, (unsigned long long)s.latency_us_sum);
+      PGresult* q = PQexec(db_, sql);
+      int st = q ? PQresultStatus(q) : -1;
+      if (st != PGRES_COMMAND_OK) ok = false;
+      if (q) PQclear(q);
+      if (!ok) break;
+    }
+    PGresult* e = PQexec(db_, ok ? "COMMIT" : "ROLLBACK");
+    if (e) PQclear(e);
+    if (!ok) err = "pg upsert failed";
+    return ok;
+  }
+  bool Query(std::string& out, bool as_json, std::string& err) override {
+    if (!db_ && !Open(err)) return false;
+    PGresult* r = PQexec(db_, "SELECT method,count,bytes_in,bytes_out,latency_us_sum"
+                              " FROM molo_stats ORDER BY method");
+    if (!r || PQresultStatus(r) != PGRES_TUPLES_OK) {
+      err = r ? PQresultErrorMessage(r) : "pg query failed";
+      if (r) PQclear(r);
+      return false;
+    }
+    std::ostringstream o;
+    if (as_json) o << "[";
+    for (int i = 0; i < PQntuples(r); ++i) {
+      const char* m = PQgetvalue(r, i, 0);
+      long long c = std::atoll(PQgetvalue(r, i, 1)), bi = std::atoll(PQgetvalue(r, i, 2)),
+                bo = std::atoll(PQgetvalue(r, i, 3)), ls = std::atoll(PQgetvalue(r, i, 4));
+      double avg = c ? (double)ls / (double)c / 1000.0 : 0.0;
+      if (as_json) {
+        if (i) o << ",";
+        o << "{\"method\":\"" << m << "\",\"count\":" << c << ",\"kb_in\":" << (bi / 1024)
+          << ",\"kb_out\":" << (bo / 1024) << ",\"avg_ms\":" << avg << "}";
+      } else {
+        char b[256];
+        std::snprintf(b, sizeof b, "%-8s count=%lld kB_in=%lld kB_out=%lld avg_ms=%.2f\n", m, c,
+                      bi / 1024, bo / 1024, avg);
+        o << b;
+      }
+    }
+    PQclear(r);
+    if (as_json) o << "]";
+    out = o.str();
+    return true;
+  }
+
+ private:
+  std::string url_;
+  PGconn* db_ = nullptr;
+};
+
+}  // namespace
+
+std::unique_ptr<StatsStore> MakeStatsStore(const std::string& database_url) {
+  std::string u = database_url.empty() ? DatabaseUrl() : database_url;
+  if (u.rfind("postgres", 0) == 0 || u.rfind("postgresql", 0) == 0) {
+    // Strip scheme prefix variants libpq may not want; pass through as-is otherwise.
+    return std::make_unique<PgStore>(u);
+  }
+  std::string path = u;
+  const std::string pre = "sqlite://";
+  if (path.rfind(pre, 0) == 0) path = path.substr(pre.size());
+  if (path.empty()) path = "system.db";
+  return std::make_unique<SqliteStore>(path);
+}
+
+StatsCollector& StatsCollector::Global() {
+  static StatsCollector g;
+  return g;
+}
+
+void StatsCollector::Note(const std::string& method, std::uint64_t bytes_in,
+                          std::uint64_t bytes_out, std::uint64_t latency_us) {
+  if (!enabled_.load(std::memory_order_relaxed)) return;
+  std::lock_guard<std::mutex> l(mu_);
+  MethodStat& s = pending_[method];
+  s.count++;
+  s.bytes_in += bytes_in;
+  s.bytes_out += bytes_out;
+  s.latency_us_sum += latency_us;
+  MethodStat& t = pending_["_total"];
+  t.count++;
+  t.bytes_in += bytes_in;
+  t.bytes_out += bytes_out;
+  t.latency_us_sum += latency_us;
+}
+
+bool StatsCollector::Flush(std::string& err) {
+  std::map<std::string, MethodStat> snap;
+  {
+    std::lock_guard<std::mutex> l(mu_);
+    if (!store_) return true;
+    if (pending_.empty()) return store_->Open(err);
+    snap.swap(pending_);
+  }
+  StatsSnapshot s;
+  s.by_method = std::move(snap);
+  s.streams = 0;
+  for (auto& [m, st] : s.by_method) {
+    if (m != "_total") s.streams += st.count;
+  }
+  return store_->Flush(s, err);
+}
+
+bool StatsCollector::Query(std::string& out, bool as_json, std::string& err) {
+  if (!Flush(err)) return false;
+  std::lock_guard<std::mutex> l(mu_);
+  if (!store_) {
+    err = "stats store not configured";
+    return false;
+  }
+  return store_->Query(out, as_json, err);
+}
+
+}  // namespace molo::realtime::expose

src/realtime/expose/stats.h

@@ -0,0 +1,66 @@
+#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 <cstdint>
+#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); }
+
+  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_;
+};
+
+}  // namespace molo::realtime::expose

src/realtime/expose/tunnel.cpp

@@ -0,0 +1,125 @@
+#include "realtime/expose/tunnel.h"
+#include <random>
+#include <sstream>
+#include "realtime/shared/sha1.h"  // sha1 available; sha256 emulated via double-round below is wrong:
+// use OpenSSL-free portable sha256 implemented here.
+#include <cstdint>
+#include <cstring>
+namespace molo::realtime::expose {
+namespace {
+// Minimal public-domain SHA-256.
+constexpr uint32_t K[64]={0x428a2f98,0x71374491,0xb5c0fbcf,0xe9b5dba5,0x3956c25b,0x59f111f1,0x923f82a4,0xab1c5ed5,0xd807aa98,0x12835b01,0x243185be,0x550c7dc3,0x72be5d74,0x80deb1fe,0x9bdc06a7,0xc19bf174,0xe49b69c1,0xefbe4786,0x0fc19dc6,0x240ca1cc,0x2de92c6f,0x4a7484aa,0x5cb0a9dc,0x76f988da,0x983e5152,0xa831c66d,0xb00327c8,0xbf597fc7,0xc6e00bf3,0xd5a79147,0x06ca6351,0x14292967,0x27b70a85,0x2e1b2138,0x4d2c6dfc,0x53380d13,0x650a7354,0x766a0abb,0x81c2c92e,0x92722c85,0xa2bfe8a1,0xa81a664b,0xc24b8b70,0xc76c51a3,0xd192e819,0xd6990624,0xf40e3585,0x106aa070,0x19a4c116,0x1e376c08,0x2748774c,0x34b0bcb5,0x391c0cb3,0x4ed8aa4a,0x5b9cca4f,0x682e6ff3,0x748f82ee,0x78a5636f,0x84c87814,0x8cc70208,0x90befffa,0xa4506ceb,0xbef9a3f7,0xc67178f2};
+#define RR(x,n)(((x)>>(n))|((x)<<(32-(n))))
+std::string Sha256(const std::string& m){
+  uint32_t h[8]={0x6a09e667,0xbb67ae85,0x3c6e00fb,0xa54ff53a,0x510e527f,0x9b05688c,0x1f83d9ab,0x5be0cd19};
+  std::string msg=m; uint64_t bitlen=(uint64_t)m.size()*8;
+  msg+=(char)0x80; while(msg.size()%64!=56)msg+=(char)0; 
+  for(int i=7;i>=0;--i)msg+=(char)(bitlen>>(i*8));
+  for(size_t off=0;off<msg.size();off+=64){
+    uint32_t w[64]; const auto* p=(const unsigned char*)msg.data()+off;
+    for(int i=0;i<16;++i)w[i]=(p[4*i]<<24)|(p[4*i+1]<<16)|(p[4*i+2]<<8)|p[4*i+3];
+    for(int i=16;i<64;++i){uint32_t s0=RR(w[i-15],7)^RR(w[i-15],18)^(w[i-15]>>3);
+      uint32_t s1=RR(w[i-2],17)^RR(w[i-2],19)^(w[i-2]>>10);w[i]=w[i-16]+s0+w[i-7]+s1;}
+    uint32_t a=h[0],b=h[1],c=h[2],d=h[3],e=h[4],f=h[5],g=h[6],hh=h[7];
+    for(int i=0;i<64;++i){uint32_t S1=RR(e,6)^RR(e,11)^RR(e,25);uint32_t ch=(e&f)^(~e&g);
+      uint32_t t1=hh+S1+ch+K[i]+w[i];uint32_t S0=RR(a,2)^RR(a,13)^RR(a,22);
+      uint32_t maj=(a&b)^(a&c)^(b&c);uint32_t t2=S0+maj;
+      hh=g;g=f;f=e;e=d+t1;d=c;c=b;b=a;a=t1+t2;}
+    h[0]+=a;h[1]+=b;h[2]+=c;h[3]+=d;h[4]+=e;h[5]+=f;h[6]+=g;h[7]+=hh;
+  }
+  char out[65]; for(int i=0;i<8;++i)snprintf(out+2*i,3,"%02x",0);
+  std::string r; char buf[9]; for(int i=0;i<8;++i){snprintf(buf,9,"%08x",h[i]);r+=buf;}
+  return r;
+}
+}
+std::string Sha256Hex(const std::string& s){return Sha256(s);}
+bool ClaimRegistry::KeyEq(const std::string& a, const std::string& b){
+  if(a.size()!=b.size())return false;
+  unsigned d=0; for(size_t i=0;i<a.size();++i)d|= (unsigned char)a[i]^(unsigned char)b[i];
+  return d==0;
+}
+std::string ClaimRegistry::KeyFingerprint(const std::string& k){return Sha256Hex("fp:"+k).substr(0,12);}
+std::string NewTunnelId(){
+  static std::mt19937_64 rng{std::random_device{}()};
+  std::ostringstream o; o.flags(std::ios::hex);
+  for(int i=0;i<4;++i)o<<rng();
+  return "tun_"+o.str();
+}
+ClaimRegistry::ClaimRegistry(std::string server_id):server_id_(std::move(server_id)){}
+bool ClaimRegistry::SegmentPrefix(const std::string& a,const std::string& b){
+  if(a==b||a=="/")return true;
+  if(b.size()<a.size())return false;
+  if(b.compare(0,a.size(),a)!=0)return false;
+  return b.size()==a.size()||b[a.size()]=='/';
+}
+RegisterResult ClaimRegistry::Register(const Claim& claim,const std::string& key_sha,
+    UpstreamTable ups,bool takeover,std::shared_ptr<Tunnel>& out,std::string& detail){
+  for(auto r:kReserved) if(SegmentPrefix(r,claim.path)){detail="reserved prefix";return RegisterResult::kReserved;}
+  std::lock_guard<std::mutex> l(mu_);
+  for(auto&[id,t]:by_id_){
+    if(!t->live||t->takeover_draining)continue;
+    bool same_host = claim.host.empty()||t->claim.host.empty()||claim.host==t->claim.host;
+    if(same_host&&(SegmentPrefix(t->claim.path,claim.path)||SegmentPrefix(claim.path,t->claim.path))){
+      auto it=key_by_claim_.find(claim.str());
+      bool same_key = it!=key_by_claim_.end()&&(KeyEq(key_sha,it->second)||KeyEq(key_sha,t->key_sha256));
+      if(takeover&&same_key){t->takeover_draining=true;t->live=false;break;}
+      detail="claim held by "+t->id; return RegisterResult::kConflict;
+    }
+  }
+  auto it=key_by_claim_.find(claim.str());
+  if(it!=key_by_claim_.end()&&!KeyEq(it->second,key_sha)){
+    auto pi=prev_key_by_claim_.find(claim.str());
+    auto pu=prev_key_until_.find(claim.str());
+    bool drained = pi!=prev_key_by_claim_.end()&&pu!=prev_key_until_.end()&&
+      std::chrono::steady_clock::now()<pu->second&&KeyEq(pi->second,key_sha);
+    if(!drained){detail="bad key";return RegisterResult::kAuthFail;}
+  }
+  if((int)by_id_.size()>=kMaxTunnels){detail="tunnel cap";return RegisterResult::kConflict;}
+  int perkey=0; for(auto&[id,t]:by_id_)if(t->live&&KeyEq(t->key_sha256,key_sha))++perkey;
+  if(perkey>=kMaxTunnelsPerKey){detail="per-key tunnel cap";return RegisterResult::kConflict;}
+  if(it==key_by_claim_.end())key_by_claim_[claim.str()]=key_sha;
+  auto t=std::make_shared<Tunnel>();
+  t->id=NewTunnelId();t->claim=claim;t->upstreams=std::move(ups);t->key_sha256=key_sha;
+  t->server_id=server_id_;t->connected_at=t->last_ping=std::chrono::steady_clock::now();
+  by_id_[t->id]=t;out=t;return RegisterResult::kOk;
+}
+bool ClaimRegistry::Release(const std::string& tid){std::lock_guard<std::mutex> l(mu_);auto it=by_id_.find(tid);if(it==by_id_.end())return false;it->second->live=false;by_id_.erase(it);return true;}
+std::shared_ptr<Tunnel> ClaimRegistry::Find(const std::string& host,const std::string& path) const{
+  std::lock_guard<std::mutex> l(mu_);
+  std::string h=host; for(auto&c:h)c=tolower((unsigned char)c);
+  while(h.size()>1&&h.back()=='.')h.pop_back();
+  std::shared_ptr<Tunnel> best=nullptr;
+  for(auto&[id,t]:by_id_){
+    if(!t->live)continue;
+    if(!t->claim.host.empty()&&t->claim.host!=h)continue;
+    if(!SegmentPrefix(t->claim.path,path)&&!(t->claim.whole_host))continue;
+    // whole-host claim on another host never matches (already filtered);
+    // path claims match any host (wildcard) per spec section 6.
+    if(!best||t->claim.path.size()>best->claim.path.size()||
+       (t->claim.path.size()==best->claim.path.size()&&
+        !t->claim.host.empty()&&best->claim.host.empty()))best=t;
+  }
+  return best;
+}
+std::shared_ptr<Tunnel> ClaimRegistry::Get(const std::string& tid) const{
+  std::lock_guard<std::mutex> l(mu_);auto it=by_id_.find(tid);return it==by_id_.end()?nullptr:it->second;
+}
+std::vector<std::shared_ptr<Tunnel>> ClaimRegistry::List() const{
+  std::lock_guard<std::mutex> l(mu_);std::vector<std::shared_ptr<Tunnel>> v;
+  for(auto&[id,t]:by_id_)if(t->live)v.push_back(t);return v;
+}
+bool ClaimRegistry::CheckKey(const Claim& c,const std::string& k) const{
+  std::lock_guard<std::mutex> l(mu_);auto it=key_by_claim_.find(c.str());if(it!=key_by_claim_.end()&&KeyEq(it->second,k))return true;
+  auto pi=prev_key_by_claim_.find(c.str());
+  auto pu=prev_key_until_.find(c.str());
+  return pi!=prev_key_by_claim_.end()&&pu!=prev_key_until_.end()&&
+    std::chrono::steady_clock::now()<pu->second&&KeyEq(pi->second,k);
+}
+void ClaimRegistry::RotateKey(const Claim& c,const std::string& n){
+  std::lock_guard<std::mutex> l(mu_);
+  auto it=key_by_claim_.find(c.str());
+  if(it!=key_by_claim_.end()){prev_key_by_claim_[c.str()]=it->second;
+    prev_key_until_[c.str()]=std::chrono::steady_clock::now()+std::chrono::minutes(5);}
+  key_by_claim_[c.str()]=n;
+}
+}  // namespace molo::realtime::expose

src/realtime/expose/tunnel.h

@@ -0,0 +1,81 @@
+#pragma once
+// Spec: reverse.md sections 4,6,9,10. In-memory claim registry + tunnel state.
+// OO: ClaimRegistry owns Tunnel objects; each Tunnel owns its UpstreamTable.
+#include <chrono>
+#include <cstdint>
+#include <functional>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "realtime/expose/claim.h"
+#include "realtime/expose/upstream.h"
+
+namespace molo::realtime::expose {
+
+inline constexpr const char* kReserved[] = {"/_molo", "/healthz", "/readyz",
+    "/startupz", "/metrics", "/csp-report"};
+
+inline constexpr int kMaxTunnels = 64;
+inline constexpr int kMaxTunnelsPerKey = 16;
+inline constexpr int kMaxConnsPerIp = 128;
+inline constexpr int kPingIntervalSec = 25;
+inline constexpr int kPingAckTimeoutSec = 20;
+inline constexpr int kClaimHoldSec = 30;
+inline constexpr int kTakeoverDrainSec = 30;
+
+struct Tunnel {
+  std::string id;
+  Claim claim;
+  UpstreamTable upstreams;
+  std::string key_sha256;  // hex sha256, never the key
+  std::string server_id;
+  std::chrono::steady_clock::time_point connected_at;
+  std::chrono::steady_clock::time_point last_ping;
+  bool live = true;
+  std::uint64_t streams_total = 0;
+  std::uint64_t bytes_in = 0, bytes_out = 0;
+  bool takeover_draining = false;
+  // Per-stream accounting (reverse.md §9): kB up/down + stream count.
+  void NoteStream(std::uint64_t in, std::uint64_t out) {
+    ++streams_total;
+    bytes_in += in;
+    bytes_out += out;
+  }
+};
+
+enum class RegisterResult { kOk, kBadClaim, kConflict, kReserved, kAuthFail };
+
+// Thread-safe in-memory registry (v1: no persistence, pinned to process).
+class ClaimRegistry {
+ public:
+  explicit ClaimRegistry(std::string server_id);
+  RegisterResult Register(const Claim& claim, const std::string& key_sha256,
+                          UpstreamTable ups, bool takeover,
+                          std::shared_ptr<Tunnel>& out, std::string& detail);
+  bool Release(const std::string& tunnel_id);
+  std::shared_ptr<Tunnel> Find(const std::string& host, const std::string& path) const;
+  std::shared_ptr<Tunnel> Get(const std::string& tunnel_id) const;
+  std::vector<std::shared_ptr<Tunnel>> List() const;
+  bool CheckKey(const Claim& claim, const std::string& key_sha256) const;
+  void RotateKey(const Claim& claim, const std::string& new_sha256);
+  static bool KeyEq(const std::string& a, const std::string& b);  // constant-time
+  static std::string KeyFingerprint(const std::string& key_sha256);
+  const std::string& server_id() const { return server_id_; }
+ private:
+  static bool SegmentPrefix(const std::string& a, const std::string& b);
+  std::string server_id_;
+  mutable std::mutex mu_;
+  std::map<std::string, std::shared_ptr<Tunnel>> by_id_;  // tunnel id -> tunnel
+  std::map<std::string, std::string> key_by_claim_;       // claim str -> key sha
+  std::map<std::string, std::string> prev_key_by_claim_;  // rotated-out key, 5min drain
+  std::map<std::string, std::chrono::steady_clock::time_point> prev_key_until_;
+};
+
+std::string Sha256Hex(const std::string& s);
+std::string NewTunnelId();
+
+}  // namespace molo::realtime::expose

src/realtime/expose/upstream.cpp

@@ -0,0 +1,69 @@
+#include "realtime/expose/upstream.h"
+namespace molo::realtime::expose {
+namespace {
+UpstreamScheme SchemeOf(const std::string& s){
+  if(s=="tcp")return UpstreamScheme::kTcp; if(s=="https")return UpstreamScheme::kHttps;
+  if(s=="ws")return UpstreamScheme::kWs; if(s=="wss")return UpstreamScheme::kWss;
+  return UpstreamScheme::kHttp;
+}
+int DefaultPort(UpstreamScheme s){
+  switch(s){case UpstreamScheme::kHttps:case UpstreamScheme::kWss:return 443;
+    case UpstreamScheme::kTcp:return 0; default:return 80;}
+}
+bool IsDigits(const std::string& s){if(s.empty())return false;for(char c:s)if(c<'0'||c>'9')return false;return true;}
+bool SegBoundary(const std::string& path,const std::string& pfx){
+  if(pfx=="/")return true;
+  if(path.size()<pfx.size())return false;
+  if(path.compare(0,pfx.size(),pfx)!=0)return false;
+  return path.size()==pfx.size()||path[pfx.size()]=='/';
+}
+}
+std::optional<UpstreamEntry> ParseUpstream(const std::string& raw, std::string& reason){
+  std::string s=raw, prefix="/";
+  auto eq=s.rfind('=');
+  // '=' only a selector if what follows starts with '/'
+  if(eq!=std::string::npos&&eq+1<s.size()&&s[eq+1]=='/'){prefix=s.substr(eq+1);s=s.substr(0,eq);}
+  UpstreamEntry e; e.raw=raw; e.prefix=prefix.empty()?"/":prefix;
+  if(IsDigits(s)){e.scheme=UpstreamScheme::kHttp;e.host="127.0.0.1";e.port=std::stoi(s);return e;}
+  std::string scheme="http", rest=s;
+  auto p=s.find("://");
+  if(p!=std::string::npos){scheme=s.substr(0,p);rest=s.substr(p+3);}
+  if(scheme!="tcp"&&scheme!="http"&&scheme!="https"&&scheme!="ws"&&scheme!="wss"){reason="bad scheme";return std::nullopt;}
+  e.scheme=SchemeOf(scheme);
+  std::string hostport=rest, upath;
+  auto sl=rest.find('/');
+  if(sl!=std::string::npos){hostport=rest.substr(0,sl);upath=rest.substr(sl);}
+  e.path=upath;
+  auto colon=hostport.rfind(':');
+  if(colon!=std::string::npos){e.host=hostport.substr(0,colon);
+    try{e.port=std::stoi(hostport.substr(colon+1));}catch(...){reason="bad port";return std::nullopt;}
+  } else {e.host=hostport.empty()?"127.0.0.1":hostport;e.port=DefaultPort(e.scheme);}
+  if(e.host.empty())e.host="127.0.0.1";
+  return e;
+}
+bool UpstreamTable::Add(UpstreamEntry e, std::string& reason){
+  for(auto&x:entries_) if(x.prefix==e.prefix){reason="duplicate prefix";return false;}
+  entries_.push_back(std::move(e)); return true;
+}
+const UpstreamEntry* UpstreamTable::Route(const std::string& path) const{
+  const UpstreamEntry* best=nullptr;
+  for(auto&e:entries_){ if(!SegBoundary(path,e.prefix))continue;
+    if(!best||e.prefix.size()>best->prefix.size())best=&e; }
+  return best;
+}
+bool UpstreamHostAllowed(const std::string& host, bool allow_public){
+  if(allow_public)return true;
+  std::string h=host;
+  if(h=="localhost"||h=="127.0.0.1"||h=="::1"||h=="::ffff:127.0.0.1")return true;
+  if(h.rfind("127.",0)==0)return true;
+  if(h.rfind("10.",0)==0)return true;
+  if(h.rfind("192.168.",0)==0)return true;
+  if(h.rfind("172.",0)==0){
+    int n=0; size_t d=h.find('.',4);
+    try{n=std::stoi(h.substr(4,d==std::string::npos?std::string::npos:d-4));}catch(...){return false;}
+    if(n>=16&&n<=31)return true;
+  }
+  if(h.rfind("169.254.",0)==0||h.rfind("fe80",0)==0||h.rfind("FE80",0)==0)return true;
+  return false;
+}
+}  // namespace molo::realtime::expose

src/realtime/expose/upstream.h

@@ -0,0 +1,38 @@
+#pragma once
+// Spec: reverse.md sections 3,5. Upstream entry + longest-prefix routing.
+#include <optional>
+#include <string>
+#include <vector>
+
+namespace molo::realtime::expose {
+
+enum class UpstreamScheme { kTcp, kHttp, kHttps, kWs, kWss };
+
+struct UpstreamEntry {
+  std::string raw;
+  UpstreamScheme scheme = UpstreamScheme::kHttp;
+  std::string host;
+  int port = 80;
+  std::string path;    // origin path
+  std::string prefix;  // claim-subpath selector, "/" = default
+};
+
+// Parses "[scheme://]host[:port][/path][=prefix]". Bare number = 127.0.0.1:port.
+std::optional<UpstreamEntry> ParseUpstream(const std::string& raw, std::string& reason);
+
+// SSRF guard (reverse.md section 7): upstream host must be loopback /
+// RFC1918 / link-local unless allow_public is set. Applies per entry.
+bool UpstreamHostAllowed(const std::string& host, bool allow_public);
+
+// OO routing table: longest prefix wins, ties rejected at insert.
+class UpstreamTable {
+ public:
+  bool Add(UpstreamEntry e, std::string& reason);
+  const UpstreamEntry* Route(const std::string& path) const;
+  const std::vector<UpstreamEntry>& entries() const { return entries_; }
+  std::size_t size() const { return entries_.size(); }
+ private:
+  std::vector<UpstreamEntry> entries_;
+};
+
+}  // namespace molo::realtime::expose