Live page: the speed of every peer right now

A new Live page shows current download and upload per peer, updated
every 2 seconds, with the last 2 minutes as a chart and a small chart
per peer. The server reads the WireGuard counters every 2 seconds and
keeps 2 minutes in memory; GET /api/v1/live serves them, with since=
for only the newer steps. The dev simulator now adds traffic in
proportion to the time between samples.

speed.go

@@ -0,0 +1,107 @@
+package main
+
+import (
+	"log/slog"
+	"sync"
+	"time"
+)
+
+// Speeds keeps the last few minutes of each peer's speed in memory for the
+// Live page. It reads the kernel counters every speedStep, apart from the
+// traffic history in Stats, and never writes to disk.
+
+const (
+	speedStep   = 2 * time.Second
+	speedPoints = 60 // 2 minutes
+)
+
+// SpeedPoint is one step: per peer ID, download and upload in bits per
+// second, from the peer's point of view.
+type SpeedPoint struct {
+	T     int64               `json:"t"`
+	Peers map[string][2]int64 `json:"peers"`
+}
+
+type Speeds struct {
+	store  *Store
+	kernel Kernel
+
+	mu     sync.Mutex
+	last   map[string][2]int64 // raw rx, tx by public key
+	lastAt time.Time
+	points []SpeedPoint
+}
+
+func newSpeeds(store *Store, kernel Kernel) *Speeds {
+	return &Speeds{store: store, kernel: kernel, last: map[string][2]int64{}}
+}
+
+func (s *Speeds) sample(now time.Time) {
+	cfg := s.store.Get()
+	samples, err := s.kernel.Sample(cfg.Server.Interface)
+	if err != nil {
+		slog.Debug("speed sample failed", "err", err)
+		return
+	}
+	idByKey := map[string]string{}
+	for _, p := range cfg.Peers {
+		if p.hasKey() {
+			idByKey[p.PublicKey] = p.ID
+		}
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	secs := now.Sub(s.lastAt).Seconds()
+	first := s.lastAt.IsZero()
+	cur := map[string][2]int64{}
+	pt := SpeedPoint{T: now.Unix(), Peers: map[string][2]int64{}}
+	for _, smp := range samples {
+		cur[smp.PublicKey] = [2]int64{smp.RxBytes, smp.TxBytes}
+		id := idByKey[smp.PublicKey]
+		prev, ok := s.last[smp.PublicKey]
+		if id == "" || !ok || first {
+			continue
+		}
+		dRx, dTx := smp.RxBytes-prev[0], smp.TxBytes-prev[1]
+		if dRx < 0 || dTx < 0 { // counters were reset
+			continue
+		}
+		// Tx is what the server sent: the peer's download.
+		pt.Peers[id] = [2]int64{int64(float64(dTx*8) / secs), int64(float64(dRx*8) / secs)}
+	}
+	s.last, s.lastAt = cur, now
+	if first {
+		return
+	}
+	s.points = append(s.points, pt)
+	if len(s.points) > speedPoints {
+		s.points = s.points[len(s.points)-speedPoints:]
+	}
+}
+
+// Since returns the points newer than the unix time t, oldest first.
+func (s *Speeds) Since(t int64) []SpeedPoint {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := []SpeedPoint{}
+	for _, p := range s.points {
+		if p.T > t {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+func (s *Speeds) Run(stop <-chan struct{}) {
+	s.sample(time.Now())
+	t := time.NewTicker(speedStep)
+	defer t.Stop()
+	for {
+		select {
+		case <-stop:
+			return
+		case now := <-t.C:
+			s.sample(now)
+		}
+	}
+}