package webapp import ( "context" "encoding/json" "fmt" "io" "log" "net/http" "sort" "strconv" "strings" "sync" "sync/atomic" "time" "github.com/runbear-io/beardrive/internal/journal" ) // Read telemetry: who consumes what, aggregated. Together with the write // history the journals already carry, this completes the read×write matrix — // heavily-read + long-unwritten is the danger zone an admin should fix first // (see docs/design/read-heatmap.md). // // What counts as a read: viewer file/render/download hits (human), share-link // hits (share), and agent tool reads reported by syncing devices (agent). // /store/* sync traffic is replication, not reading, and is never counted; // history /blob views are spelunking, not consumption, and aren't either. // // One exception to "never an event log": session reads (SessionRead), the // per-session detail behind a History run card. They are a separate table // with their own, shorter retention, and they never enter the ledger's // in-memory bucket map — see SessionReadRepo for why that separation is the // whole point. // // Privacy: rows are daily aggregation buckets, never an event log. The actor // column (account email / device id / share token) exists only to count // distinct readers and never appears in an API response — with exactly one // stated exception, ?by=device (handleHeat), which reports the ids of agent // devices. A device id is not a person and is already visible to every // project member through History's device join; nothing else in the column — // no email, no share token, no unowned device id — ever leaves the server. // The exception is only sound because the ingest path validates the id // (handleReadReport → ownsDevice): an actor recorded as an agent is shaped // like a device id and is not one another account is syncing — never an // arbitrary string, and never someone else's machine. This route also never // registers a device: registering the id it is about to judge is what turned // the round-2 check into a one-request speed bump. // // A session id is identity-adjacent and gets the same ruling, written down // before anything serves it: a session id appears ONLY in History responses, // on the op that carries it, and as a ?session= filter INPUT. It is never // enumerated — no "list sessions" response, no session column in /heat's // output, nothing new in ?by=device. That is sound because the id is already // visible to every project member inside Op.Note today, so serving it as its // own field discloses nothing new, while refusing to enumerate keeps /heat // identity-free exactly as documented above. // Read kinds. const ( ReadKindHuman = "human" ReadKindAgent = "agent" ReadKindShare = "share" ) // ReadStat is one aggregation bucket: reads of one path by one actor on one // day. Day == "" is the all-time fold that survives retention. type ReadStat struct { Project string `json:"project"` Path string `json:"path"` Day string `json:"day"` // "2006-01-02" UTC, or "" for all-time Kind string `json:"kind"` Actor string `json:"actor"` Count int64 `json:"count"` Last time.Time `json:"last"` } // ReadStatKey identifies one bucket. type ReadStatKey struct { Project, Path, Day, Kind, Actor string } func (s ReadStat) key() ReadStatKey { return ReadStatKey{s.Project, s.Path, s.Day, s.Kind, s.Actor} } // SessionRead records that one agent session read one path, from one device. // Not a count and not a bucket: the run card asks "did this session read this // file?", and one row per (session, device, path) answers it with no // aggregation. Device is always the hub-validated device the report arrived // from, never anything the client put in the body. type SessionRead struct { Project string `json:"project"` Session string `json:"session"` Device string `json:"device"` Path string `json:"path"` Last time.Time `json:"last"` } type sessionReadKey struct{ Project, Session, Device, Path string } func (s SessionRead) key() sessionReadKey { return sessionReadKey{s.Project, s.Session, s.Device, s.Path} } // HeatEntry is the per-path aggregate the heat API returns. Counts only — // never identities. type HeatEntry struct { Human int64 `json:"human,omitempty"` Agent int64 `json:"agent,omitempty"` Share int64 `json:"share,omitempty"` Readers int `json:"readers,omitempty"` // distinct human readers LastRead time.Time `json:"last_read,omitzero"` } const ( // readDebounce collapses request storms (reloads, render-then-raw double // fetches) into visits: repeat reads of a path by the same actor within // the window don't count again. readDebounce = 10 * time.Minute // readFlushEvery throttles persistence; dirty buckets ride in memory // between flushes, so a crash loses at most this much telemetry. readFlushEvery = 30 * time.Second // DefaultReadRetentionDays is how long daily buckets keep per-day // resolution before folding into the all-time row. DefaultReadRetentionDays = 400 // DefaultSessionRetentionDays is how long per-session read detail is // kept. Much shorter than the bucket retention: this is event-shaped // data whose only consumer is a History run card, and a month covers a // retro. Rows past it are deleted, not folded — the heat totals were // never derived from them, so nothing is lost from any count. DefaultSessionRetentionDays = 30 // sessionPruneEvery throttles the retention delete; it rides the same // flush the buckets use rather than owning a goroutine. sessionPruneEvery = time.Hour ) // ReadLedger is the in-memory read-telemetry service over a ReadRepo, in the // mold of DeviceRegistry: reads stay in memory, writes are throttled. There is // no background goroutine — flushes piggyback on Record calls, and telemetry // failures never surface to the request that triggered them. type ReadLedger struct { repo ReadRepo retention time.Duration // Session-read detail, optional (nil = off) and deliberately outside // byKey: these rows are never loaded into memory in bulk, so Heat's full // map scan and the boot load are unaffected by session cardinality. If a // future change ever moves them into byKey, Heat (below) is what pays. sessions SessionReadRepo sessionRetention time.Duration // scans counts ShareOpens passes over byKey. Tests assert one per // project per list render — the "never one scan per share" rule is // invisible in the response body, so this is the only thing that can // catch the regression. scans atomic.Int64 mu sync.Mutex byKey map[ReadStatKey]ReadStat dirty map[ReadStatKey]bool pendingDel []ReadStatKey // retention deletions awaiting a successful flush seen map[ReadStatKey]time.Time // debounce; Day field unused ("") pendingSess map[sessionReadKey]SessionRead lastFlush time.Time lastSessPrun time.Time warned bool sessWarned bool } // NewReadLedger loads the ledger and immediately folds buckets older than the // retention horizon into their all-time rows. retentionDays <= 0 means the // default. func NewReadLedger(repo ReadRepo, retentionDays int) (*ReadLedger, error) { if retentionDays <= 0 { retentionDays = DefaultReadRetentionDays } l := &ReadLedger{ repo: repo, retention: time.Duration(retentionDays) * 24 * time.Hour, byKey: map[ReadStatKey]ReadStat{}, dirty: map[ReadStatKey]bool{}, seen: map[ReadStatKey]time.Time{}, lastFlush: time.Now(), } stats, err := repo.Load() if err != nil { return nil, err } for _, st := range stats { l.byKey[st.key()] = st } // Fold anything past the retention horizon right away. A failed persist // is not a boot failure — the fold stays dirty and later flushes retry. l.mu.Lock() l.compactLocked() if err := l.persistLocked(); err != nil { log.Printf("beardrive: read telemetry compact failed (will retry): %v", err) } l.mu.Unlock() return l, nil } // OpenReadLedger loads the file-backed ledger at path. func OpenReadLedger(path string, retentionDays int) (*ReadLedger, error) { return NewReadLedger(newFileReadRepo(path), retentionDays) } // OpenSessionReadRepo is the file-backed session-read store, for hubs // running without a MetaStore (the historical JSON-files layout). func OpenSessionReadRepo(path string) SessionReadRepo { return newFileSessionReadRepo(path) } // WithSessions turns on per-session read detail (the data behind a History // run card). Separate from the constructor so every existing caller — and // every backend that has no session repo — keeps working with it off. // retentionDays <= 0 means the default. func (l *ReadLedger) WithSessions(repo SessionReadRepo, retentionDays int) *ReadLedger { if l == nil || repo == nil { return l } if retentionDays <= 0 { retentionDays = DefaultSessionRetentionDays } l.mu.Lock() defer l.mu.Unlock() l.sessions = repo l.sessionRetention = time.Duration(retentionDays) * 24 * time.Hour l.pendingSess = map[sessionReadKey]SessionRead{} return l } // Record counts one read. Nil-safe and never fails: telemetry must not break // the page view (or sync cycle) that triggered it. func (l *ReadLedger) Record(project, path, kind, actor string) { if l == nil || project == "" || path == "" { return } now := time.Now().UTC() l.mu.Lock() defer l.mu.Unlock() visit := ReadStatKey{Project: project, Path: path, Kind: kind, Actor: actor} if t, ok := l.seen[visit]; ok && now.Sub(t) < readDebounce { return } l.seen[visit] = now key := visit key.Day = now.Format("2006-01-02") st := l.byKey[key] st.Project, st.Path, st.Day, st.Kind, st.Actor = project, path, key.Day, kind, actor st.Count++ st.Last = now l.byKey[key] = st l.dirty[key] = true if now.Sub(l.lastFlush) >= readFlushEvery { l.flushLocked() } } // RecordSession notes that one agent session read one path from one device. // Nil-safe, off when no session repo is configured, and — like Record — // never fails: telemetry must not break the sync cycle that reported it. // Unlike Record it is NOT debounced: a row is a fact ("this session read this // file"), not a count, so repeats are the same row rewritten. func (l *ReadLedger) RecordSession(project, session, device, path string) { if l == nil || project == "" || session == "" || device == "" || path == "" { return } l.mu.Lock() defer l.mu.Unlock() if l.sessions == nil { return } now := time.Now() sr := SessionRead{Project: project, Session: session, Device: device, Path: path, Last: now.UTC()} l.pendingSess[sr.key()] = sr // Same throttle the buckets use. Record's own flush check sits behind its // debounce return, so a report whose buckets are all debounced would // otherwise leave these rows buffered indefinitely. if now.Sub(l.lastFlush) >= readFlushEvery { l.flushLocked() } } // SessionPaths returns the paths one session read from one device, for the // History run card. Both the session and the device are required by the // caller (handleHeat): a session-only lookup would return rows a member // reported under someone else's session id, which pinning to the validated // device is what makes harmless. func (l *ReadLedger) SessionPaths(project, session, device string) []string { if l == nil || project == "" || session == "" || device == "" { return nil } l.mu.Lock() repo := l.sessions // Flush first, so a card opened seconds after a sync sees that sync's // reads instead of an empty list. if repo != nil { l.flushSessionsLocked() } l.mu.Unlock() if repo == nil { return nil } rows, err := repo.ListBySession(project, session, device) if err != nil { log.Printf("beardrive: session reads lookup failed: %v", err) return nil } out := make([]string, 0, len(rows)) for _, r := range rows { out = append(out, r.Path) } return out } // flushSessionsLocked persists buffered session rows and, at most hourly, // deletes the ones past the session retention. Failures keep the buffer for // the next attempt and log once — a read_sessions failure must never affect // read_stats, so this is deliberately separate from persistLocked. Callers // hold mu. func (l *ReadLedger) flushSessionsLocked() { if l.sessions == nil { return } if len(l.pendingSess) > 0 { batch := make([]SessionRead, 0, len(l.pendingSess)) for _, sr := range l.pendingSess { batch = append(batch, sr) } if err := l.sessions.PutBatch(batch); err != nil { if !l.sessWarned { l.sessWarned = true log.Printf("beardrive: session read flush failed (will retry): %v", err) } } else { l.sessWarned = false l.pendingSess = map[sessionReadKey]SessionRead{} } } now := time.Now() if now.Sub(l.lastSessPrun) < sessionPruneEvery { return } l.lastSessPrun = now if err := l.sessions.PruneBefore(now.UTC().Add(-l.sessionRetention)); err != nil { log.Printf("beardrive: session read prune failed (will retry): %v", err) } } // Heat aggregates reads per path for one project. since bounds the window // (zero = all time, including retention folds); prefix "" means the whole // project, otherwise paths under "/". func (l *ReadLedger) Heat(project, prefix string, since time.Time) map[string]HeatEntry { if l == nil { return nil } sinceDay := "" if !since.IsZero() { sinceDay = since.UTC().Format("2006-01-02") } prefix = strings.TrimSuffix(prefix, "/") out := map[string]HeatEntry{} humans := map[string]map[string]bool{} // path → distinct human actors l.mu.Lock() defer l.mu.Unlock() for key, st := range l.byKey { if key.Project != project { continue } if prefix != "" && !strings.HasPrefix(key.Path, prefix+"/") { continue } if key.Day == "" { if sinceDay != "" { continue // all-time fold is older than any windowed query } } else if key.Day < sinceDay { continue } e := out[key.Path] switch key.Kind { case ReadKindHuman: e.Human += st.Count set := humans[key.Path] if set == nil { set = map[string]bool{} humans[key.Path] = set } set[key.Actor] = true case ReadKindAgent: e.Agent += st.Count case ReadKindShare: e.Share += st.Count } if st.Last.After(e.LastRead) { e.LastRead = st.Last } out[key.Path] = e } for p, set := range humans { e := out[p] e.Readers = len(set) out[p] = e } return out } // AgentHeat aggregates agent reads per device per top-level folder ("" for // root files) — the coverage-matrix data. Agent buckets only, by design: // agent actors are device ids, which history already exposes; human actors // (emails) must never leave the server, so human/share buckets are not // consulted at all. func (l *ReadLedger) AgentHeat(project string, since time.Time) map[string]map[string]int64 { if l == nil { return nil } sinceDay := "" if !since.IsZero() { sinceDay = since.UTC().Format("2006-01-02") } out := map[string]map[string]int64{} l.mu.Lock() defer l.mu.Unlock() for key, st := range l.byKey { if key.Project != project || key.Kind != ReadKindAgent { continue } if key.Day == "" { if sinceDay != "" { continue } } else if key.Day < sinceDay { continue } folder := "" if i := strings.IndexByte(key.Path, '/'); i >= 0 { folder = key.Path[:i] } m := out[key.Actor] if m == nil { m = map[string]int64{} out[key.Actor] = m } m[folder] += st.Count } return out } // ShareOpen is share-link consumption for one path: visits, and when. type ShareOpen struct { Count int64 Last time.Time } // ShareOpens aggregates share-kind reads per path for one project — the // receipt a person who shared something actually wants. All-time, because a // link's lifetime is the question a receipt answers. // // Share buckets only, and that is what makes Last mean *last opened*: // HeatEntry.LastRead is cross-kind, so a member viewing the file in the hub // would otherwise move the "opened through the link" date. // // Counts, never identities — the share actor is token+"/"+IP+"/"+UA hash, a // public credential joined to a network and a browser, and it must not leave // the ledger. There is deliberately no distinct-openers field. // // One byKey scan per project, never one per share: callers build this map // once and index it, because byKey is the full map and a project with 40 // links would otherwise pay 40 full scans per list render. func (l *ReadLedger) ShareOpens(project string) map[string]ShareOpen { if l == nil { return nil // reads disabled: absent, not zero } l.scans.Add(1) out := map[string]ShareOpen{} l.mu.Lock() defer l.mu.Unlock() for key, st := range l.byKey { if key.Project != project || key.Kind != ReadKindShare { continue } // No day filter: both the daily buckets and the folded Day == "" // all-time row count. e := out[key.Path] e.Count += st.Count if st.Last.After(e.Last) { e.Last = st.Last } out[key.Path] = e } return out } // Close flushes any pending buckets. func (l *ReadLedger) Close() error { if l == nil { return nil } l.mu.Lock() defer l.mu.Unlock() l.flushLocked() if n := len(l.dirty); n > 0 { return fmt.Errorf("read ledger: flush failed, %d buckets pending", n) } return nil } // flushLocked persists dirty buckets (and, once a day, retention folds), // pruning the debounce map along the way. Failures keep the buckets dirty for // the next attempt and log once — telemetry never breaks a request. func (l *ReadLedger) flushLocked() { now := time.Now() l.lastFlush = now for k, t := range l.seen { if now.Sub(t) >= readDebounce { delete(l.seen, k) } } l.compactLocked() if err := l.persistLocked(); err != nil { if !l.warned { l.warned = true log.Printf("beardrive: read telemetry flush failed (will retry): %v", err) } } else { l.warned = false } l.flushSessionsLocked() } // compactLocked folds daily buckets older than the retention horizon into // their all-time rows, queueing the daily rows for deletion. Callers hold mu. func (l *ReadLedger) compactLocked() { horizon := time.Now().UTC().Add(-l.retention).Format("2006-01-02") for key, st := range l.byKey { if key.Day == "" || key.Day >= horizon { continue } fold := key fold.Day = "" agg := l.byKey[fold] agg.Project, agg.Path, agg.Kind, agg.Actor = st.Project, st.Path, st.Kind, st.Actor agg.Day = "" agg.Count += st.Count if st.Last.After(agg.Last) { agg.Last = st.Last } l.byKey[fold] = agg l.dirty[fold] = true delete(l.byKey, key) delete(l.dirty, key) l.pendingDel = append(l.pendingDel, key) } } // persistLocked writes queued deletions and dirty buckets through the repo. // Callers hold mu. Both queues survive a failure so the next flush retries — // dropping a deletion would resurrect folded rows on the next load and // double-count them. func (l *ReadLedger) persistLocked() error { if len(l.pendingDel) > 0 { if err := l.repo.DeleteBatch(l.pendingDel); err != nil { // Same one-transaction problem as the put path below, worse for // being first: a key the store will never accept parks here // forever and PutBatch is then never reached at all, so the whole // hub's telemetry stops persisting. Retry one at a time; if some // land, the ones that did not are keys this store will never // accept, so drop them. If none land the store is down — // transient — and the queue stands for the next flush. if len(l.pendingDel) == 1 { return err } var stuck []ReadStatKey landed := 0 for _, key := range l.pendingDel { if l.repo.DeleteBatch([]ReadStatKey{key}) == nil { landed++ } else { stuck = append(stuck, key) } } if landed == 0 { return err } for _, key := range stuck { log.Printf("beardrive: read telemetry dropped an undeletable bucket (project %s, path %q): %v", key.Project, key.Path, err) } } l.pendingDel = nil } if len(l.dirty) == 0 { return nil } batch := make([]ReadStat, 0, len(l.dirty)) for key := range l.dirty { batch = append(batch, l.byKey[key]) } err := l.repo.PutBatch(batch) if err == nil { l.dirty = map[ReadStatKey]bool{} return nil } if len(batch) == 1 { return err } // One transaction, so one bucket the store refuses takes every other // bucket down with it — and, because they stay dirty, every bucket the hub // counts from then on. That is a hub-wide telemetry kill from the lowest // privilege there is (Postgres rejects a NUL byte in a path; sqlite and // the file backend store it happily). Retry one at a time: if some land, // the ones that did not are content this store will never accept, so drop // them rather than wedge the queue. If none land the store itself is // down — transient — and everything stays dirty for the next flush. var stuck []ReadStatKey landed := 0 for key := range l.dirty { if l.repo.PutBatch([]ReadStat{l.byKey[key]}) == nil { delete(l.dirty, key) landed++ } else { stuck = append(stuck, key) } } if landed == 0 { return err } for _, key := range stuck { log.Printf("beardrive: read telemetry dropped an unstorable bucket (project %s, path %q): %v", key.Project, key.Path, err) delete(l.dirty, key) } return nil } // hasControlChars reports whether s carries a C0/C7F control character — // never legitimate in a path, and fatal to a Postgres text column. func hasControlChars(s string) bool { return strings.ContainsFunc(s, func(r rune) bool { return r < 0x20 || r == 0x7f }) } // ---- server integration ---- // ctxProjectKey carries the resolved project id from the proj() route // resolver to handlers that record reads. type ctxProjectKey struct{} func withProjectID(r *http.Request, id string) *http.Request { return r.WithContext(context.WithValue(r.Context(), ctxProjectKey{}, id)) } func projectID(r *http.Request) string { id, _ := r.Context().Value(ctxProjectKey{}).(string) return id } // recordRead counts a human read of path for the request's project. No-op // outside hub mode (no project id) or when read tracking is off. func (s *Server) recordRead(r *http.Request, path string) { if s.Reads == nil { return } project := projectID(r) if project == "" { return } actor := s.requestUser(r).Email if actor == "" { actor = "anonymous" } s.Reads.Record(project, path, ReadKindHuman, actor) } // handleHeat serves per-path read aggregates: ?prefix= bounds to a folder, // ?days= bounds the window (default 30, 0 = all time). With ?by=device it // returns the agent-kind breakdown instead: per device (registry-joined), // reads per top-level folder — the one place an actor id is reported, and // only ever a device the reporting account owned (see the package comment). // Human and share actors never leave the server in any shape. func (s *Server) handleHeat(v *volume, w http.ResponseWriter, r *http.Request) { if s.Reads == nil { http.Error(w, "read tracking is not enabled on this server", http.StatusNotFound) return } _ = v q := r.URL.Query() // ?session=&device= is the run-card join: which paths that agent session // read. Both are required — a session-only query would also return rows a // member reported under someone else's session id, which pinning the row // to the reporting device (handleReadReport) is what makes harmless. This // is a filter INPUT only: nothing here or anywhere else enumerates // sessions, and the response carries paths, no identities and no counts. if session := q.Get("session"); session != "" || q.Get("device") != "" { device := q.Get("device") if session == "" || device == "" { http.Error(w, "session and device must be given together", http.StatusBadRequest) return } paths := s.Reads.SessionPaths(projectID(r), session, device) if paths == nil { paths = []string{} // an empty list, never a null the client must special-case } writeJSON(w, map[string]any{"paths": paths}) return } days := 30 if raw := q.Get("days"); raw != "" { var err error if days, err = strconv.Atoi(raw); err != nil || days < 0 { http.Error(w, "invalid days", http.StatusBadRequest) return } } var since time.Time if days > 0 { since = time.Now().UTC().AddDate(0, 0, -days) } switch q.Get("by") { case "": case "device": s.heatByDevice(w, projectID(r), since) return default: http.Error(w, "invalid by (use device)", http.StatusBadRequest) return } entries := s.Reads.Heat(projectID(r), q.Get("prefix"), since) out := map[string]any{"entries": entries} if !since.IsZero() { out["since"] = since.Format("2006-01-02") } writeJSON(w, out) } // deviceHeat is one row of the ?by=device response. type deviceHeat struct { ID string `json:"id"` Name string `json:"name,omitempty"` OS string `json:"os,omitempty"` Folders map[string]int64 `json:"folders"` Total int64 `json:"total"` } func (s *Server) heatByDevice(w http.ResponseWriter, project string, since time.Time) { byDevice := s.Reads.AgentHeat(project, since) visible := s.deviceVisibleIn(project) devices := make([]deviceHeat, 0, len(byDevice)) for id, folders := range byDevice { d := deviceHeat{ID: id, Folders: folders} // Scoped join: a device owned by an account outside this project's org // contributes no name or OS, so heat cannot become a window onto // another org's machines. if info, ok := s.Devices.LookupIn(id, visible); ok { d.Name, d.OS = info.Name, info.OS } for _, n := range folders { d.Total += n } devices = append(devices, d) } sort.Slice(devices, func(i, j int) bool { if devices[i].Total != devices[j].Total { return devices[i].Total > devices[j].Total } return devices[i].ID < devices[j].ID }) out := map[string]any{"devices": devices} if !since.IsZero() { out["since"] = since.Format("2006-01-02") } writeJSON(w, out) } // handleReadReport ingests agent reads from a syncing device: the client's // read spool, drained best-effort at sync time. Requires a device identity — // the device id is the actor, so reads count as agent traffic. func (s *Server) handleReadReport(v *volume, w http.ResponseWriter, r *http.Request) { if s.Reads == nil { http.Error(w, "read tracking is not enabled on this server", http.StatusNotFound) return } device := deviceID(r) if device == "" { http.Error(w, "agent read reports need a device identity", http.StatusBadRequest) return } var req struct { Reads []struct { Path string `json:"path"` // Session is the agent session the read happened in — a CLIENT // string, so it is only ever stored alongside the device the hub // validated below, never on its own. See the row write. Session string `json:"session,omitempty"` // Time is accepted for forward compatibility but buckets use // server time: client clocks are unreliable and late flushes are // telemetry noise, not data loss. Time time.Time `json:"time,omitzero"` } `json:"reads"` } if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil { http.Error(w, "bad request: "+err.Error(), http.StatusBadRequest) return } if len(req.Reads) > 4096 { http.Error(w, "too many reads in one report", http.StatusBadRequest) return } // The device id becomes the actor these buckets are keyed by, and /heat // reports agent actors — so an unvalidated header would let any member // plant any string (an id from another org, or an account email) and have // the hub serve it back to the whole project as a reader. This route // deliberately does NOT observe the device: registering the id it is about // to judge is what made the round-2 check a one-request speed bump. Only // /store/* traffic registers a device. mine := s.ownsDevice(r, device) // A reported path is a claim about a file, and the heat map is what the // Dashboard's reads-x-staleness quadrant is built from — the view an // operator reads to decide what is stale. Any member with PermRead could // report any string, so the quadrant was member-writable fiction: a // "compliance/soc2-evidence-2026.md" nobody ever wrote showed up as read. // The project's own replayed state is the only thing that can say a path is // real, and it is right here. A snapshot the store cannot produce records // nothing this cycle: the client's spool is drained best-effort and retried, // and telemetry must never fail a request (nor invent one). snap, err := v.snapshot(r.Context()) if err != nil { writeJSON(w, map[string]any{"accepted": 0}) return } project := projectID(r) n := 0 for _, e := range req.Reads { // A path is a bucket key that reaches the metadata store: a control // character (a NUL above all) is rejected outright by Postgres, and a // row the store will never accept has to be refused here rather than // discovered at flush time. // journal.SafePath is the rule, in the one place it is defined. This // was a fourth copy of it, and it disagreed in both directions: it // accepted "/etc/passwd", "a//b" and "./a", and refused "my..file". if !journal.SafePath(e.Path) { continue } if !mine { continue // not this account's device: counted for nobody } if _, real := snap.files[e.Path]; !real { continue // no such file in this project: a read of nothing is not a read } s.Reads.Record(project, e.Path, ReadKindAgent, device) // The session id is the one field here the hub cannot vouch for: it // arrives in the body, so any member could report reads naming a // teammate's session and paint files onto that teammate's run card. // The row is therefore pinned to `device` — the id ownsDevice just // validated — and the query side requires BOTH session and device, so // a forged row can only ever be found under the forger's own device, // which MayActAs guarantees is never someone else's. if sess := trimText(e.Session, 128); sess != "" && journal.SafeText(sess) { s.Reads.RecordSession(project, sess, device, e.Path) } n++ } writeJSON(w, map[string]any{"accepted": n}) }