BergaBruh
137 lines
3.9 KiB
Go
137 lines
3.9 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sort"
|
|
)
|
|
|
|
// Agg holds median aggregates for a single node over an arbitrary time
|
|
// window. Counts and medians are computed Go-side (sort + middle pick)
|
|
// because SQLite has no native percentile aggregate.
|
|
type Agg struct {
|
|
Count int64 `json:"count"`
|
|
MedianDl float64 `json:"median_dl"`
|
|
MedianUl float64 `json:"median_ul"`
|
|
MedianPing float64 `json:"median_ping"`
|
|
}
|
|
|
|
// SpeedTest is a single client-side speedtest result reported via the
|
|
// RecordSpeedTest RPC. Timestamp is unix seconds. Node is 'fl' or 'pl1'
|
|
// (DB-level CHECK). Country is ISO-3166-1 alpha-2 (validated at the
|
|
// gRPC handler, not at the DB).
|
|
type SpeedTest struct {
|
|
ID int64
|
|
Timestamp int64
|
|
Node string
|
|
Country string
|
|
DlMbps float64
|
|
UlMbps float64
|
|
PingMs float64
|
|
JitterMs float64
|
|
}
|
|
|
|
// SaveSpeedTest inserts a speedtest row and returns the new id.
|
|
func (s *Store) SaveSpeedTest(ctx context.Context, st SpeedTest) (int64, error) {
|
|
res, err := s.db.ExecContext(ctx, `
|
|
INSERT INTO speedtests(timestamp, node, country, dl_mbps, ul_mbps, ping_ms, jitter_ms)
|
|
VALUES(?, ?, ?, ?, ?, ?, ?)`,
|
|
st.Timestamp, st.Node, st.Country, st.DlMbps, st.UlMbps, st.PingMs, st.JitterMs)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("insert speedtest: %w", err)
|
|
}
|
|
return res.LastInsertId()
|
|
}
|
|
|
|
// ListRecentSpeedTests returns up to limit speedtests ordered by id DESC
|
|
// (newest first).
|
|
func (s *Store) ListRecentSpeedTests(ctx context.Context, limit int) ([]SpeedTest, error) {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, timestamp, node, country, dl_mbps, ul_mbps, ping_ms, jitter_ms
|
|
FROM speedtests ORDER BY id DESC LIMIT ?`, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := []SpeedTest{}
|
|
for rows.Next() {
|
|
var st SpeedTest
|
|
if err := rows.Scan(&st.ID, &st.Timestamp, &st.Node, &st.Country,
|
|
&st.DlMbps, &st.UlMbps, &st.PingMs, &st.JitterMs); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, st)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// DeleteSpeedTestsBefore deletes rows with timestamp < ts and returns the
|
|
// number of rows removed. Used by the retention sweep (Task 14).
|
|
func (s *Store) DeleteSpeedTestsBefore(ctx context.Context, ts int64) (int64, error) {
|
|
res, err := s.db.ExecContext(ctx, `DELETE FROM speedtests WHERE timestamp < ?`, ts)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return res.RowsAffected()
|
|
}
|
|
|
|
// AggregateSpeedTestsSince returns per-node median aggregates of dl/ul/ping
|
|
// for rows with timestamp >= since. The result is keyed by node ("fl"/"pl1").
|
|
// Reads only metric columns — no privacy-sensitive data (country, jitter
|
|
// not used here).
|
|
func (s *Store) AggregateSpeedTestsSince(ctx context.Context, since int64) (map[string]Agg, error) {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT node, dl_mbps, ul_mbps, ping_ms FROM speedtests
|
|
WHERE timestamp >= ? ORDER BY node`, since)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
type bucket struct{ dl, ul, ping []float64 }
|
|
buckets := map[string]*bucket{}
|
|
for rows.Next() {
|
|
var node string
|
|
var dl, ul, ping float64
|
|
if err := rows.Scan(&node, &dl, &ul, &ping); err != nil {
|
|
return nil, err
|
|
}
|
|
b, ok := buckets[node]
|
|
if !ok {
|
|
b = &bucket{}
|
|
buckets[node] = b
|
|
}
|
|
b.dl = append(b.dl, dl)
|
|
b.ul = append(b.ul, ul)
|
|
b.ping = append(b.ping, ping)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
out := map[string]Agg{}
|
|
for node, b := range buckets {
|
|
out[node] = Agg{
|
|
Count: int64(len(b.dl)),
|
|
MedianDl: median(b.dl),
|
|
MedianUl: median(b.ul),
|
|
MedianPing: median(b.ping),
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// median sorts xs in place and returns the middle value (or mean of the
|
|
// two middle values if even-length). Returns 0 for empty input.
|
|
func median(xs []float64) float64 {
|
|
if len(xs) == 0 {
|
|
return 0
|
|
}
|
|
sort.Float64s(xs)
|
|
mid := len(xs) / 2
|
|
if len(xs)%2 == 0 {
|
|
return (xs[mid-1] + xs[mid]) / 2
|
|
}
|
|
return xs[mid]
|
|
}
|