feat(hub): count file changes and headless users server-side (#164)

The frontend's PostHog tracker sees everything a person clicks, but a
device syncing through /store/* never loads a page — so an agent editing
files all day was invisible, and "number of file changes" and "daily
active users" both undercounted by however much of the product runs
headless.

One event, files_changed, from every write door: sync, upload (relay and
direct commit), remove, restore. Its distinct_id is the same email
analytics.ts identifies with, so a person on a laptop and a browser is
one user, and its puts/deletes properties sum to the change count.

The count comes from ops the hub has not stored before, not from the
request body: a device PUTs its WHOLE journal every cycle, so counting
the body would re-report the device's entire history every ten seconds
and the metric would climb while nobody edited anything.
journalKeepsItsOps already parsed the stored journal for the append-only
check and threw the sequence away; it returns storedMax now, so this
costs no extra read. Blob PUTs are deliberately not change events —
content-addressed storage skips a blob it already holds, so blob writes
undercount edits while ops are exact.

No SDK: posthog-go would ship a tracker inside every self-hoster's
binary, which is the exact thing the frontend avoids by loading
posthog-js from a CDN only when a key is configured. Capture is one JSON
POST, on its own goroutine, that does nothing when Analytics.Key is
empty — an OSS hub still contacts nobody.

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

internal/webapp/analytics.go

@@ -0,0 +1,122 @@
+package webapp
+
+import (
+	"bytes"
+	"encoding/json"
+	"log"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/runbear-io/beardrive/internal/journal"
+)
+
+// Server-side product analytics: the events the browser cannot see.
+//
+// The frontend (frontend/src/analytics.ts) covers everything a person clicks,
+// but a device syncing through /store/* never loads a page — an agent editing
+// files all day is invisible to it. That is both metrics this exists for:
+// how many file changes land, and how many accounts are active at all once
+// the headless ones are counted.
+//
+// No SDK. posthog-go would sit in go.mod and ship inside every self-hoster's
+// binary, which is the exact thing the frontend avoids by loading posthog-js
+// from a CDN only when a key is configured: an OSS install must not carry a
+// tracker it never runs. Capture is one JSON POST, and AnalyticsConfig.Key is
+// already a public write-only project token (server.go), so this needs no new
+// credential and no new config seam.
+//
+// Telemetry never fails a request: the POST runs in its own goroutine, its
+// error is dropped, and a broken analytics host is at worst a log line.
+
+// analyticsClient bounds a hung ingestion host. Without the timeout a stalled
+// POST would pin its goroutine for as long as the process lives.
+var analyticsClient = &http.Client{Timeout: 10 * time.Second}
+
+// capture sends one event to PostHog for the given account, or does nothing
+// when analytics is unconfigured — which is every self-hosted hub.
+//
+// email is the distinct id, and it must be the same one the frontend
+// identifies with (analytics.ts calls identify(cfg.me.email)), or the same
+// person counts twice: once for their browser and once for their laptop.
+func (s *Server) capture(email, event string, props map[string]any) {
+	if s.Analytics.Key == "" || email == "" {
+		return
+	}
+	body, err := json.Marshal(map[string]any{
+		"api_key":     s.Analytics.Key,
+		"event":       event,
+		"distinct_id": email,
+		"properties":  props,
+	})
+	if err != nil {
+		return
+	}
+	endpoint := s.Analytics.Endpoint() + "/i/v0/e/"
+	go func() {
+		resp, err := analyticsClient.Post(endpoint, "application/json", bytes.NewReader(body))
+		if err != nil {
+			analyticsFailed(err)
+			return
+		}
+		resp.Body.Close()
+	}()
+}
+
+// captureChange records file changes for the account behind r.
+//
+// Every write path on the hub calls this — sync, upload, remove, restore — so
+// "how many files changed" stays ONE number in PostHog instead of a per-route
+// event set that has to be summed by hand and silently misses whichever route
+// someone forgets. `source` splits agent traffic from browser traffic when
+// that is the question; nothing here carries a path or a file name.
+func (s *Server) captureChange(r *http.Request, source string, puts, deletes int) {
+	if puts+deletes == 0 {
+		return
+	}
+	// The account that made the request, not Op.User: an op authored offline
+	// carries no signed-in account, and the pusher is who was active either way.
+	s.capture(s.requestUser(r).Email, "files_changed", map[string]any{
+		"puts":    puts,
+		"deletes": deletes,
+		"source":  source,
+		"project": r.PathValue("project"),
+	})
+}
+
+// analyticsFailed logs the first delivery failure and nothing after it. A hub
+// that cannot reach PostHog would otherwise write a line per sync cycle per
+// device, forever, about a thing no operator can act on. Once, not a bool:
+// these run on their own goroutines.
+var analyticsWarnOnce sync.Once
+
+func analyticsFailed(err error) {
+	analyticsWarnOnce.Do(func() {
+		log.Printf("beardrive: product analytics delivery failed (further failures silent): %v", err)
+	})
+}
+
+// countOps splits ops the hub has not seen before into put and delete totals.
+//
+// storedMax is the highest Seq already on the hub for this journal, because a
+// device PUTs its WHOLE journal every cycle (journalKeepsItsOps depends on
+// exactly that). Counting len(ops) would re-count the device's entire history
+// every ten seconds. Seq is the device's own monotone counter and a journal
+// object belongs to one device, so it is the honest discriminator.
+//
+// Blob PUTs are deliberately not a change event: content-addressed storage
+// skips a blob it already holds, so blob writes undercount edits while ops
+// are exact.
+func countOps(ops []journal.Op, storedMax int64) (puts, deletes int) {
+	for _, op := range ops {
+		if op.Seq <= storedMax {
+			continue
+		}
+		if op.Kind == journal.KindDelete {
+			deletes++
+		} else {
+			puts++
+		}
+	}
+	return puts, deletes
+}

internal/webapp/analytics_test.go

@@ -0,0 +1,158 @@
+package webapp
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// phSpy stands in for PostHog's ingestion host.
+type phSpy struct {
+	*httptest.Server
+	mu   sync.Mutex
+	got  []map[string]any
+	seen chan struct{}
+}
+
+func newPHSpy(t *testing.T) *phSpy {
+	t.Helper()
+	spy := &phSpy{seen: make(chan struct{}, 64)}
+	spy.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		var ev map[string]any
+		if err := json.Unmarshal(body, &ev); err != nil {
+			t.Errorf("posthog got unparseable body %q: %v", body, err)
+		}
+		if r.URL.Path != "/i/v0/e/" {
+			t.Errorf("capture posted to %q, want /i/v0/e/", r.URL.Path)
+		}
+		spy.mu.Lock()
+		spy.got = append(spy.got, ev)
+		spy.mu.Unlock()
+		w.WriteHeader(200)
+		spy.seen <- struct{}{}
+	}))
+	t.Cleanup(spy.Close)
+	return spy
+}
+
+// events waits for n deliveries. capture is fire-and-forget, so the send
+// outlives the request that triggered it.
+func (spy *phSpy) events(t *testing.T, n int) []map[string]any {
+	t.Helper()
+	for i := 0; i < n; i++ {
+		select {
+		case <-spy.seen:
+		case <-time.After(3 * time.Second):
+			spy.mu.Lock()
+			defer spy.mu.Unlock()
+			t.Fatalf("waited for %d events, got %d: %v", n, len(spy.got), spy.got)
+		}
+	}
+	spy.mu.Lock()
+	defer spy.mu.Unlock()
+	return append([]map[string]any(nil), spy.got...)
+}
+
+func (spy *phSpy) count(t *testing.T) int {
+	t.Helper()
+	// Nothing more should arrive; give a stray goroutine a moment to land.
+	time.Sleep(200 * time.Millisecond)
+	spy.mu.Lock()
+	defer spy.mu.Unlock()
+	return len(spy.got)
+}
+
+func evProp(t *testing.T, ev map[string]any, key string) any {
+	t.Helper()
+	props, ok := ev["properties"].(map[string]any)
+	if !ok {
+		t.Fatalf("event has no properties: %v", ev)
+	}
+	return props[key]
+}
+
+// A device PUTs its WHOLE journal every cycle, so the count has to come from
+// the ops the hub has not already stored. Counting the body would report the
+// device's entire history again every ten seconds, and "number of file
+// changes" would climb on its own while nobody edited anything.
+func TestAnalytics_SyncCountsOnlyNewOps(t *testing.T) {
+	spy := newPHSpy(t)
+	h, srv, c, p := permHub(t)
+	srv.Analytics = AnalyticsConfig{Key: "phc_test", Host: spy.URL}
+
+	const dev = "alice-laptop-6f2a"
+	// Alice syncs once so the device id is hers.
+	if rec := secfx4Store(t, h, "GET", "/api/p/"+p.ID+"/store/list", "", c["alice"], dev); rec.Code != 200 {
+		t.Fatalf("control: alice's own sync: %d %s", rec.Code, rec.Body)
+	}
+
+	first := secaudOpLine(1, dev, "put", "plan.md", strings.Repeat("a", 64)) +
+		secaudOpLine(2, dev, "put", "notes.md", strings.Repeat("b", 64))
+	if rec := secfx4PushJournal(t, h, p.ID, dev, first, c["alice"]); rec.Code != 200 {
+		t.Fatalf("first push: %d %s", rec.Code, rec.Body)
+	}
+	ev := spy.events(t, 1)[0]
+	if got := ev["event"]; got != "files_changed" {
+		t.Errorf("event = %v, want files_changed", got)
+	}
+	if got := ev["distinct_id"]; got != "alice@x.io" {
+		t.Errorf("distinct_id = %v, want alice@x.io — it must match the id the frontend "+
+			"identifies with (analytics.ts), or one person counts as two users", got)
+	}
+	if got := evProp(t, ev, "puts"); got != float64(2) {
+		t.Errorf("puts = %v, want 2", got)
+	}
+	if got := evProp(t, ev, "source"); got != "sync" {
+		t.Errorf("source = %v, want sync", got)
+	}
+
+	// The second cycle repeats both ops and appends one delete, exactly as a
+	// real client does. Only the delete is new.
+	second := first + secaudOpLine(3, dev, "delete", "plan.md", "")
+	if rec := secfx4PushJournal(t, h, p.ID, dev, second, c["alice"]); rec.Code != 200 {
+		t.Fatalf("second push: %d %s", rec.Code, rec.Body)
+	}
+	ev = spy.events(t, 1)[1]
+	if got, want := evProp(t, ev, "deletes"), float64(1); got != want {
+		t.Errorf("deletes = %v, want %v", got, want)
+	}
+	if got := evProp(t, ev, "puts"); got != float64(0) {
+		t.Errorf("puts = %v on a re-push of the same journal, want 0 — the whole history "+
+			"is being counted again every cycle", got)
+	}
+
+	// A cycle that adds nothing (the daemon re-pushing an unchanged journal)
+	// is not a file change and must not land as one.
+	if rec := secfx4PushJournal(t, h, p.ID, dev, second, c["alice"]); rec.Code != 200 {
+		t.Fatalf("idempotent push: %d %s", rec.Code, rec.Body)
+	}
+	if n := spy.count(t); n != 2 {
+		t.Errorf("%d events after a no-op re-push, want 2", n)
+	}
+}
+
+// The OSS default: no key, no third-party request. A self-hosted hub must not
+// phone home, which is the same rule the frontend follows.
+func TestAnalytics_UnconfiguredHubSendsNothing(t *testing.T) {
+	spy := newPHSpy(t)
+	h, srv, c, p := permHub(t)
+	srv.Analytics = AnalyticsConfig{Host: spy.URL} // host set, key empty
+
+	const dev = "alice-laptop-6f2a"
+	if rec := secfx4Store(t, h, "GET", "/api/p/"+p.ID+"/store/list", "", c["alice"], dev); rec.Code != 200 {
+		t.Fatalf("control: alice's own sync: %d %s", rec.Code, rec.Body)
+	}
+	body := secaudOpLine(1, dev, "put", "plan.md", strings.Repeat("a", 64))
+	if rec := secfx4PushJournal(t, h, p.ID, dev, body, c["alice"]); rec.Code != 200 {
+		t.Fatalf("push: %d %s", rec.Code, rec.Body)
+	}
+	if n := spy.count(t); n != 0 {
+		t.Errorf("a hub with no analytics key sent %d events", n)
+	}
+}

internal/webapp/remove.go

@@ -80,5 +80,6 @@ func (s *Server) handleRemove(v *volume, w http.ResponseWriter, r *http.Request)
 	}
 	s.quota().RecordUsage(org, 0)
 	v.invalidate()
+	s.captureChange(r, "browser", 0, 1)
 	writeJSON(w, map[string]any{"ok": true, "path": p})
 }

internal/webapp/restore.go

@@ -98,5 +98,9 @@ func (s *Server) handleRestore(v *volume, w http.ResponseWriter, r *http.Request
 	}
 	s.quota().RecordUsage(org, 0)
 	v.invalidate()
+	// A restore writes a put op like any other edit, so it belongs in the same
+	// count. The frontend's `file_restored` says which BUTTON was pressed; this
+	// says a file changed.
+	s.captureChange(r, "browser", 1, 0)
 	writeJSON(w, map[string]any{"ok": true, "blob": req.SHA, "size": found.Size})
 }

internal/webapp/store.go

@@ -547,26 +547,33 @@ func (s *Server) opsNameTheirAuthor(w http.ResponseWriter, r *http.Request, ops
 // refused unparseable bodies since round 6. A backend that cannot answer fails
 // the push closed: the client degrades to Offline and retries next cycle, which
 // is the posture everywhere else on this path.
-func journalKeepsItsOps(ctx context.Context, be remote.Backend, key string, ops []journal.Op) (bool, error) {
+//
+// storedMax is the highest Seq the hub already holds, returned because this is
+// the only place that parses the stored journal and analytics needs it to tell
+// this cycle's new ops from the whole history the body repeats (countOps). It
+// is 0 for a first push and for a stored journal that would not parse — the
+// latter over-counts one push, which is the right price for not reading the
+// object twice.
+func journalKeepsItsOps(ctx context.Context, be remote.Backend, key string, ops []journal.Op) (ok bool, storedMax int64, err error) {
 	switch have, err := be.Exists(ctx, key); {
 	case err != nil:
-		return false, err
+		return false, 0, err
 	case !have:
-		return true, nil
+		return true, 0, nil
 	}
 	rc, err := be.Get(ctx, key)
 	if err != nil {
-		return false, err
+		return false, 0, err
 	}
 	defer rc.Close()
 	data, err := io.ReadAll(rc)
 	if err != nil {
-		return false, err
+		return false, 0, err
 	}
 	stored, err := journal.Parse(data)
 	if err != nil {
 		log.Printf("beardrive: %s is not parseable, its ops cannot be protected from a rewrite: %v", key, err)
-		return true, nil
+		return true, 0, nil
 	}
 	seen := make(map[int64]bool, len(ops))
 	for _, op := range ops {
@@ -574,10 +581,13 @@ func journalKeepsItsOps(ctx context.Context, be remote.Backend, key string, ops
 	}
 	for _, op := range stored {
 		if !seen[op.Seq] {
-			return false, nil
+			return false, 0, nil
+		}
+		if op.Seq > storedMax {
+			storedMax = op.Seq
 		}
 	}
-	return true, nil
+	return true, storedMax, nil
 }
 
 // storePutBody hands back the plaintext of a PUT body, inflating it when the
@@ -698,8 +708,11 @@ func (s *Server) handleStorePut(v *volume, w http.ResponseWriter, r *http.Reques
 		http.Error(w, err.Error(), http.StatusForbidden)
 		return
 	}
+	var storedMax int64
 	if strings.HasPrefix(key, "journal/") {
-		switch ok, err := journalKeepsItsOps(r.Context(), rs.Backend, key, ops); {
+		var ok bool
+		var err error
+		switch ok, storedMax, err = journalKeepsItsOps(r.Context(), rs.Backend, key, ops); {
 		case err != nil:
 			storageErr(w, http.StatusBadGateway, "could not read the stored journal", err)
 			return
@@ -795,6 +808,8 @@ func (s *Server) handleStorePut(v *volume, w http.ResponseWriter, r *http.Reques
 	s.quota().RecordUsage(org, size)
 	if strings.HasPrefix(key, "journal/") {
 		v.invalidate() // new ops should show in the viewer immediately
+		puts, deletes := countOps(ops, storedMax)
+		s.captureChange(r, "sync", puts, deletes)
 	}
 	writeJSON(w, map[string]any{"ok": true})
 }

internal/webapp/upload.go

@@ -473,6 +473,7 @@ func (s *Server) handleUploadContent(v *volume, w http.ResponseWriter, r *http.R
 	}
 	s.quota().RecordUsage(org, size)
 	v.invalidate()
+	s.captureChange(r, "browser", 1, 0)
 	writeJSON(w, map[string]any{"ok": true, "path": p})
 }
 
@@ -528,5 +529,6 @@ func (s *Server) handleUploadCommit(v *volume, w http.ResponseWriter, r *http.Re
 		s.quota().RecordUsage(org, size)
 	}
 	v.invalidate()
+	s.captureChange(r, "browser", 1, 0)
 	writeJSON(w, map[string]any{"ok": true, "path": req.Path})
 }