amadan

carlos / platform Public

analytics: the meter the edge feeds — pageview rule, visitors, top keys, requeue (spec §3.3)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev bbe75931d3c69e84cd6bcc95329ef344521eb0ab parent 3cfb076
2 files changed, +381 −0
  • internal/analytics/meter.go +219 −0
  • internal/analytics/meter_test.go +162 −0
diff --git a/internal/analytics/meter.go b/internal/analytics/meter.go
new file mode 100644
index 000000000..77889c762
--- /dev/null
+++ b/internal/analytics/meter.go
@@ -0,0 +1,219 @@
+package analytics
+
+import (
+ "crypto/sha256"
+ "encoding/binary"
+ "log"
+ "regexp"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/apperr"
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/usage"
+)
+
+// Obs is one served response as the edge saw it (spec §3.3). The edge
+// fills it from the observer defer; nothing here touches the request.
+type Obs struct {
+ Acct, App, Host string
+ Method string
+ Status int
+ ContentType string
+ Path string // raw; sanitised here
+ RefererHost string // host only, "" = direct
+ UA string
+ ClientIP string
+ Bytes int64
+ Prefetch bool // Sec-Purpose/Purpose: prefetch
+ FetchDest string // Sec-Fetch-Dest, "" when absent
+ When time.Time
+}
+
+// AppKey is the (account, app) the meter buckets by.
+type AppKey = applog.AppKey
+
+// IsPageview applies §3.3's rule: GET, 200, text/html, not prefetch, dest
+// document or absent, not under /__carlos/. A bot UA is reported
+// separately and never a pageview.
+func IsPageview(o Obs) (pageview, bot bool) {
+ if o.Method != "GET" || o.Status != 200 {
+ return false, false
+ }
+ mt := strings.ToLower(strings.TrimSpace(strings.SplitN(o.ContentType, ";", 2)[0]))
+ if mt != "text/html" {
+ return false, false
+ }
+ if o.Prefetch || (o.FetchDest != "" && o.FetchDest != "document") {
+ return false, false
+ }
+ if strings.HasPrefix(o.Path, "/__carlos/") {
+ return false, false
+ }
+ if IsBot(o.UA) {
+ return false, true
+ }
+ return true, false
+}
+
+// VisitorHash is sha256(salt ‖ host ‖ ip ‖ ua) folded to 64 bits for the
+// sketch; the platform never stores the inputs.
+func VisitorHash(salt []byte, host, ip, ua string) uint64 {
+ h := sha256.New()
+ h.Write(salt)
+ h.Write([]byte{0})
+ h.Write([]byte(host))
+ h.Write([]byte{0})
+ h.Write([]byte(ip))
+ h.Write([]byte{0})
+ h.Write([]byte(ua))
+ return binary.BigEndian.Uint64(h.Sum(nil)[:8])
+}
+
+var eventName = regexp.MustCompile(`^[a-z0-9_.-]{1,64}$`)
+
+// EventName normalises a named event: lowercased, ≤64 bytes of [a-z0-9_.-].
+func EventName(s string) (string, bool) {
+ s = strings.ToLower(strings.TrimSpace(s))
+ if !eventName.MatchString(s) {
+ return "", false
+ }
+ return s, true
+}
+
+// Meter accumulates per-app deltas between flushes. One per edge process;
+// Observe sits on the serving path (one mutex, no I/O except the first
+// salt read of a day).
+type Meter struct {
+ box string
+ saltDir string
+ now func() time.Time
+
+ mu sync.Mutex
+ cur map[AppKey]map[string]*Delta // app → date → delta being built
+ pending map[AppKey][]Delta // re-queued after a failed flush (one window)
+ seq int64
+ saltDay string
+ salt []byte
+ started string // date the process started: Restarted is set on that day's first delta
+}
+
+// NewMeter returns a meter whose salts live in saltDir (the box data dir).
+func NewMeter(saltDir, box string, now func() time.Time) *Meter {
+ return &Meter{box: box, saltDir: saltDir, now: now, cur: map[AppKey]map[string]*Delta{},
+ pending: map[AppKey][]Delta{}, seq: now().UnixMilli(), started: usage.DayOf(now())}
+}
+
+func (m *Meter) saltFor(date string) []byte {
+ if m.saltDay == date && m.salt != nil {
+ return m.salt
+ }
+ s, err := DaySalt(m.saltDir, date)
+ if err != nil {
+ log.Printf("analytics: day salt: %v (visitors not counted until it works)", err)
+ return nil
+ }
+ m.saltDay, m.salt = date, s
+ return s
+}
+
+func (m *Meter) delta(k AppKey, date string) *Delta {
+ days := m.cur[k]
+ if days == nil {
+ days = map[string]*Delta{}
+ m.cur[k] = days
+ }
+ d := days[date]
+ if d == nil {
+ d = &Delta{Date: date, Box: m.box, Hours: map[string]HourRow{}, Visitors: &HLL{},
+ Pages: NewTopK(), Referers: NewTopK(), Events: NewTopK(), Restarted: date == m.started}
+ days[date] = d
+ }
+ return d
+}
+
+// Observe counts one response. Bytes count for every response; pageview,
+// bot, visitor, page and referer only when the rule says so.
+func (m *Meter) Observe(o Obs) {
+ pv, bot := IsPageview(o)
+ date, hour := usage.DayOf(o.When), usage.HourLabel(o.When)
+ path := apperr.SanitizePath(o.Path)
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ d := m.delta(AppKey{Acct: o.Acct, App: o.App}, date)
+ row := d.Hours[hour]
+ row.Bytes += o.Bytes
+ if bot {
+ row.Bots++
+ }
+ if pv {
+ row.PV++
+ if salt := m.saltFor(date); salt != nil {
+ d.Visitors.Add(VisitorHash(salt, o.Host, o.ClientIP, o.UA))
+ }
+ d.Pages.Add(path, 1)
+ ref := o.RefererHost
+ if ref == "" {
+ ref = "direct"
+ }
+ d.Referers.Add(strings.ToLower(ref), 1)
+ }
+ d.Hours[hour] = row
+}
+
+// Track counts one named event (the doors call it; nothing does in step 2).
+func (m *Meter) Track(acct, app, name string, when time.Time) {
+ n, ok := EventName(name)
+ if !ok {
+ return
+ }
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.delta(AppKey{Acct: acct, App: app}, usage.DayOf(when)).Events.Add(n, 1)
+}
+
+// Drain returns everything buffered — re-queued deltas first, then the
+// current ones, each stamped with a fresh box-monotonic seq — and resets.
+func (m *Meter) Drain() map[AppKey][]Delta {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ out := map[AppKey][]Delta{}
+ for k, ds := range m.pending {
+ out[k] = append(out[k], ds...)
+ }
+ m.pending = map[AppKey][]Delta{}
+ for k, days := range m.cur {
+ for _, d := range days {
+ m.seq++
+ d.Seq = m.seq
+ out[k] = append(out[k], *d)
+ }
+ }
+ m.cur = map[AppKey]map[string]*Delta{}
+ if m.started != "" && usage.DayOf(m.now()) != m.started {
+ m.started = "" // the restart note is only for the day it happened
+ }
+ return out
+}
+
+// Requeue holds a failed flush for the next drain. One window survives: a
+// window already sitting in pending when Requeue is called again (i.e. no
+// successful Drain emptied it in between) is dropped and its delta count
+// is added to Dropped.Flush on the newest delta of the incoming batch.
+// Seq is left untouched — a re-queued delta keeps the seq it was drained
+// with, so a replay that actually landed is a no-op at the day file.
+func (m *Meter) Requeue(failed map[AppKey][]Delta) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ for k, ds := range failed {
+ if n := len(m.pending[k]); n > 0 {
+ dropped := int64(n)
+ m.pending[k] = nil
+ if len(ds) > 0 {
+ ds[len(ds)-1].Dropped.Flush += dropped
+ }
+ }
+ m.pending[k] = append(m.pending[k], ds...)
+ }
+}
diff --git a/internal/analytics/meter_test.go b/internal/analytics/meter_test.go
new file mode 100644
index 000000000..a6445bb6d
--- /dev/null
+++ b/internal/analytics/meter_test.go
@@ -0,0 +1,162 @@
+package analytics
+
+import (
+ "testing"
+ "time"
+)
+
+func obs(method string, status int, ctype string) Obs {
+ return Obs{Acct: "bab", App: "web", Host: "h1", Method: method, Status: status, ContentType: ctype,
+ Path: "/pricing?x=1", RefererHost: "news.ycombinator.com", UA: "Mozilla/5.0 Safari", ClientIP: "203.0.113.5",
+ Bytes: 1200, When: time.Date(2026, 9, 7, 9, 30, 0, 0, time.UTC)}
+}
+
+func TestIsPageviewRule(t *testing.T) {
+ cases := []struct {
+ name string
+ o Obs
+ pv bool
+ bot bool
+ }{
+ {"html 200", obs("GET", 200, "text/html; charset=utf-8"), true, false},
+ {"json", obs("GET", 200, "application/json"), false, false},
+ {"404", obs("GET", 404, "text/html"), false, false},
+ {"redirect", obs("GET", 302, "text/html"), false, false},
+ {"304", obs("GET", 304, ""), false, false},
+ {"head", obs("HEAD", 200, "text/html"), false, false},
+ {"post", obs("POST", 200, "text/html"), false, false},
+ {"prefetch", func() Obs { o := obs("GET", 200, "text/html"); o.Prefetch = true; return o }(), false, false},
+ {"fetch dest empty", func() Obs { o := obs("GET", 200, "text/html"); o.FetchDest = "empty"; return o }(), false, false},
+ {"fetch dest document", func() Obs { o := obs("GET", 200, "text/html"); o.FetchDest = "document"; return o }(), true, false},
+ {"carlos prefix", func() Obs { o := obs("GET", 200, "text/html"); o.Path = "/__carlos/a"; return o }(), false, false},
+ {"bot", func() Obs { o := obs("GET", 200, "text/html"); o.UA = "Googlebot/2.1"; return o }(), false, true},
+ }
+ for _, c := range cases {
+ pv, bot := IsPageview(c.o)
+ if pv != c.pv || bot != c.bot {
+ t.Errorf("%s: pv=%v bot=%v, want %v %v", c.name, pv, bot, c.pv, c.bot)
+ }
+ }
+}
+
+func TestMeterCountsAndDrains(t *testing.T) {
+ now := time.Date(2026, 9, 7, 9, 30, 0, 0, time.UTC)
+ m := NewMeter(t.TempDir(), "box1", func() time.Time { return now })
+ m.Observe(obs("GET", 200, "text/html"))
+ m.Observe(obs("GET", 200, "text/html"))
+ o := obs("GET", 200, "text/html")
+ o.ClientIP = "198.51.100.9"
+ m.Observe(o)
+ m.Observe(obs("GET", 200, "application/json")) // bytes only
+ b := obs("GET", 200, "text/html")
+ b.UA = "curl/8"
+ m.Observe(b)
+ m.Track("bab", "web", "Signup", now)
+ m.Track("bab", "web", "bad name!", now) // rejected
+ out := m.Drain()
+ ds := out[AppKey{Acct: "bab", App: "web"}]
+ if len(ds) != 1 {
+ t.Fatalf("deltas = %d", len(ds))
+ }
+ d := ds[0]
+ if d.Date != "2026-09-07" || d.Box != "box1" || d.Seq == 0 {
+ t.Fatalf("delta = %+v", d)
+ }
+ if h := d.Hours["09"]; h.PV != 3 || h.Bots != 1 || h.Bytes != 6000 {
+ t.Fatalf("hour = %+v", h)
+ }
+ if c := d.Visitors.Count(); c < 1.5 || c > 2.5 {
+ t.Fatalf("visitors ≈ 2, got %.2f", c)
+ }
+ if p := d.Pages.Fold(); p["/pricing"] != 3 {
+ t.Fatalf("pages = %v", p)
+ }
+ if r := d.Referers.Fold(); r["news.ycombinator.com"] != 3 {
+ t.Fatalf("referers = %v", r)
+ }
+ if e := d.Events.Fold(); e["signup"] != 1 || len(e) != 1 {
+ t.Fatalf("events = %v", e)
+ }
+ if len(m.Drain()) != 0 {
+ t.Fatal("second drain must be empty")
+ }
+ // Event-time bucketing: an observation at 09:59 flushed at 10:00 lands in 09.
+ m.Observe(func() Obs { o := obs("GET", 200, "text/html"); o.When = now.Add(29 * time.Minute); return o }())
+ now = now.Add(31 * time.Minute)
+ if d := m.Drain()[AppKey{Acct: "bab", App: "web"}][0]; d.Hours["09"].PV != 1 || d.Hours["10"].PV != 0 {
+ t.Fatalf("event-time bucket: %+v", d.Hours)
+ }
+}
+
+// TestMeterRequeueSurvivesOnce pins down the exact requeue semantics:
+//
+// 1. Requeue(first) after a failed flush: first's delta is held in
+// pending — nothing was pending before, so nothing is dropped.
+// 2. A normal Drain (no Requeue in between, so pending got emptied by
+// the Drain call itself) returns the held delta ahead of the
+// newly-current one, each keeping its own Seq — the re-queued delta
+// keeps the seq it was drained with (so a replay that actually landed
+// is a no-op at the day file, via Day.Merge's seq gate), while the
+// current delta gets a fresh, higher seq.
+// 3. Requeue(second) holds both of second's deltas in pending. Because
+// Drain had just cleared pending, this call finds nothing already
+// there and drops nothing yet.
+// 4. A second Requeue call — this time with a freshly built delta,
+// fired with NO Drain in between — finds second's two deltas still
+// sitting in pending from step 3. That whole older window is dropped;
+// its count (2) is added to Dropped.Flush on the newest delta in the
+// incoming batch, and only the incoming batch is kept in pending.
+// 5. A Drain() with nothing newly observed returns exactly that one
+// surviving, Dropped.Flush-stamped delta: the older window is gone.
+func TestMeterRequeueSurvivesOnce(t *testing.T) {
+ now := time.Date(2026, 9, 7, 9, 30, 0, 0, time.UTC)
+ m := NewMeter(t.TempDir(), "box1", func() time.Time { return now })
+ k := AppKey{Acct: "bab", App: "web"}
+
+ m.Observe(obs("GET", 200, "text/html"))
+ first := m.Drain()
+ m.Requeue(first) // flush failed
+
+ m.Observe(obs("GET", 200, "text/html"))
+ second := m.Drain()
+ ds := second[k]
+ if len(ds) != 2 || ds[0].Seq >= ds[1].Seq {
+ t.Fatalf("requeued delta must come back first with its own seq: %+v", ds)
+ }
+ if ds[0].Seq != first[k][0].Seq {
+ t.Fatalf("requeued delta must keep its original seq: got %d, want %d", ds[0].Seq, first[k][0].Seq)
+ }
+
+ m.Requeue(second) // failed again: pending was empty, so nothing drops yet
+
+ // A further failure before any successful Drain: pending still holds
+ // second's two deltas from the call above, so this Requeue drops that
+ // whole older window onto the newest delta of the incoming batch.
+ handBuilt := Delta{Date: "2026-09-07", Box: "box1", Seq: 999,
+ Hours: map[string]HourRow{}, Visitors: &HLL{}, Pages: NewTopK(), Referers: NewTopK(), Events: NewTopK()}
+ m.Requeue(map[AppKey][]Delta{k: {handBuilt}})
+
+ third := m.Drain()[k]
+ if len(third) != 1 || third[0].Dropped.Flush != 2 {
+ t.Fatalf("requeue-while-pending must drop the older window and count both its deltas: %+v", third)
+ }
+ if third[0].Seq != handBuilt.Seq {
+ t.Fatalf("surviving window must be the most recently re-queued one: %+v", third)
+ }
+}
+
+func TestVisitorHashAndEventName(t *testing.T) {
+ s1, s2 := []byte("salt-a"), []byte("salt-b")
+ if VisitorHash(s1, "h", "ip", "ua") != VisitorHash(s1, "h", "ip", "ua") {
+ t.Fatal("deterministic")
+ }
+ if VisitorHash(s1, "h", "ip", "ua") == VisitorHash(s2, "h", "ip", "ua") {
+ t.Fatal("salt must change the hash")
+ }
+ for in, want := range map[string]string{"Signup": "signup", "checkout.done": "checkout.done", "a b": "", "": "", "x/y": ""} {
+ got, ok := EventName(in)
+ if (want == "") == ok || got != want {
+ t.Errorf("EventName(%q) = %q,%v", in, got, ok)
+ }
+ }
+}