From 6f0f474903e19fe4fff152a2d420882a548fe820 Mon Sep 17 00:00:00 2001 From: "Snow Lee (Sungwon)" Date: Thu, 13 Aug 2026 14:50:25 -0700 Subject: [PATCH] feat(hub): count file changes and headless users server-side (#164) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- architecture/webapp-server.md | 13 ++- internal/webapp/analytics.go | 122 +++++++++++++++++++++++ internal/webapp/analytics_test.go | 158 ++++++++++++++++++++++++++++++ internal/webapp/remove.go | 1 + internal/webapp/restore.go | 4 + internal/webapp/store.go | 33 +++++-- internal/webapp/upload.go | 2 + 7 files changed, 323 insertions(+), 10 deletions(-) create mode 100644 internal/webapp/analytics.go create mode 100644 internal/webapp/analytics_test.go diff --git a/architecture/webapp-server.md b/architecture/webapp-server.md index 216f840..e7c7e75 100644 --- a/architecture/webapp-server.md +++ b/architecture/webapp-server.md @@ -123,7 +123,7 @@ classDiagram ownJournal(key) whose journal is this journalOps(key, spooled) parse + validate opsNameTheirAuthor(ops) whose name - journalKeepsItsOps(ctx, be, key, ops) + journalKeepsItsOps(ctx, be, key, ops) ok + storedMax } note for journalDoor "store.go — the invariant "each device writes only its own journal" is now ENFORCED here, not assumed. The key must be journal/<canonical device id>.jsonl for the device in the request header, that device must already be owned by the caller (DeviceRegistry.OwnerOf) or the caller must be a project admin (the recovery arm) — the old first-writer-claims arm is gone. Every op must pass journal.SafePath + config.ReservedPath on its Path and journal.SafeText on Note/Author/UserName, must name its own owner's account, and the upload must keep every Seq the stored journal already had: append-only, 409 on truncation. Bodies are spooled first, and a blob PUT must hash to the key it claims. A Content-Encoding: gzip body is inflated ABOVE the spool — the sha, the op count and the billed size are all properties of the plaintext, so nothing below that line knows compression happened — and the inflate is bounded (maxInflatedPut, 256 MiB, only when an encoding was declared), because compression severs the one-wire-byte-one-disk-byte relationship that made spool safe unbounded" note for journalDoor "Delta sync grew the key space: validStoreKey also accepts chunks/<sha256> (content-addressed, PUT must hash to its key, presigned like blobs incl. refuse-existing) and manifests/<sha256> (keyed by the whole FILE's sha — not its own content hash — so it is never presigned and gets two ingest gates instead: every chunk it names must already EXIST in the store, and the key is WRITE-ONCE — an identical re-put is a 200 no-op so an interrupted push can retry, a different body 409s. Together these make "a manifest exists ⟹ its chunks exist" an invariant every consumer can lean on: the client's push skip-proof, reassemble, and bdrive import)" @@ -391,6 +391,14 @@ classDiagram } note for AnalyticsConfig "Third managed-deployment seam beside Quota and Billing, but a value rather than an interface — there is nothing to implement, only a project to name. Emitted as /api/config `analytics` when Key is set; empty means the frontend loads no tracker and contacts nobody, which is what a self-hosted hub gets. Endpoint() is exported because the cloud module renders its own loader from the same value." + class productAnalytics { + <> + capture(email, event, props) one POST, own goroutine + captureChange(r, source, puts, deletes) files_changed + countOps(ops, storedMax) new ops only + } + note for productAnalytics "The events the browser cannot see: a device syncing through /store/* never loads a page, so an agent editing files all day is invisible to the frontend's tracker. Every write door — sync, upload, remove, restore — funnels through captureChange so the file-change count is ONE event rather than a per-route set that silently misses whichever route someone forgets, and distinct_id is the same email analytics.ts identifies with so a person is not counted twice. No SDK: posthog-go would ship a tracker inside every self-hoster's binary, and capture is one JSON POST to Endpoint() + /i/v0/e/ that does nothing at all when Key is empty. Telemetry never fails a request — the POST is a goroutine and its error is a single log line. countOps needs journalDoor's storedMax because a device PUTs its WHOLE journal every cycle: counting the body would re-report the device's entire history every ten seconds. Blob PUTs are deliberately not change events, since content-addressed storage skips a blob it already holds." + Server o-- "0..1" Source : single-volume mode Server o-- "0..1" Backend : Root (hub mode) Server o-- ProjectDB @@ -420,6 +428,9 @@ classDiagram OrgDB ..> BuiltinAuth : seniorityLister, for the heir BuiltinAuth ..> DeviceRegistry : Bind at token issuance Server *-- AnalyticsConfig + Server *-- productAnalytics : every write door emits files_changed + productAnalytics ..> AnalyticsConfig : Key gates it, Endpoint() addresses it + journalDoor ..> productAnalytics : storedMax tells this cycle's ops from the whole history Server *-- volume : per project, cached volume o-- Source diff --git a/internal/webapp/analytics.go b/internal/webapp/analytics.go new file mode 100644 index 0000000..21aa5a5 --- /dev/null +++ b/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 +} diff --git a/internal/webapp/analytics_test.go b/internal/webapp/analytics_test.go new file mode 100644 index 0000000..128cbfe --- /dev/null +++ b/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) + } +} diff --git a/internal/webapp/remove.go b/internal/webapp/remove.go index cb84b2e..fe212e5 100644 --- a/internal/webapp/remove.go +++ b/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}) } diff --git a/internal/webapp/restore.go b/internal/webapp/restore.go index addae2f..f9ba481 100644 --- a/internal/webapp/restore.go +++ b/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}) } diff --git a/internal/webapp/store.go b/internal/webapp/store.go index 78f803c..71421ab 100644 --- a/internal/webapp/store.go +++ b/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}) } diff --git a/internal/webapp/upload.go b/internal/webapp/upload.go index b796b98..0125827 100644 --- a/internal/webapp/upload.go +++ b/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}) }